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 17import logging 18 19import grpc 20 21 22class _ActualGenericRpcHandler(grpc.GenericRpcHandler): 23 24 def service(self, handler_call_details): 25 return None 26 27 28class ServerTest(unittest.TestCase): 29 30 def test_not_a_generic_rpc_handler_at_construction(self): 31 with self.assertRaises(AttributeError) as exception_context: 32 grpc.server(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 logging.basicConfig() 53 unittest.main(verbosity=2) 54