1# Copyright 2018 The gRPC Authors. 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14 15from concurrent import futures 16import unittest 17 18import grpc 19 20 21class _ActualGenericRpcHandler(grpc.GenericRpcHandler): 22 23 def service(self, handler_call_details): 24 return None 25 26 27class ServerTest(unittest.TestCase): 28 29 def test_not_a_generic_rpc_handler_at_construction(self): 30 with self.assertRaises(AttributeError) as exception_context: 31 grpc.server( 32 futures.ThreadPoolExecutor(max_workers=5), 33 handlers=[ 34 _ActualGenericRpcHandler(), 35 object(), 36 ]) 37 self.assertIn('grpc.GenericRpcHandler', 38 str(exception_context.exception)) 39 40 def test_not_a_generic_rpc_handler_after_construction(self): 41 server = grpc.server(futures.ThreadPoolExecutor(max_workers=5)) 42 with self.assertRaises(AttributeError) as exception_context: 43 server.add_generic_rpc_handlers([ 44 _ActualGenericRpcHandler(), 45 object(), 46 ]) 47 self.assertIn('grpc.GenericRpcHandler', 48 str(exception_context.exception)) 49 50 51if __name__ == '__main__': 52 unittest.main(verbosity=2) 53