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 21from tests.unit import resources 22 23 24class _ActualGenericRpcHandler(grpc.GenericRpcHandler): 25 26 def service(self, handler_call_details): 27 return None 28 29 30class ServerTest(unittest.TestCase): 31 32 def test_not_a_generic_rpc_handler_at_construction(self): 33 with self.assertRaises(AttributeError) as exception_context: 34 grpc.server(futures.ThreadPoolExecutor(max_workers=5), 35 handlers=[ 36 _ActualGenericRpcHandler(), 37 object(), 38 ]) 39 self.assertIn('grpc.GenericRpcHandler', 40 str(exception_context.exception)) 41 42 def test_not_a_generic_rpc_handler_after_construction(self): 43 server = grpc.server(futures.ThreadPoolExecutor(max_workers=5)) 44 with self.assertRaises(AttributeError) as exception_context: 45 server.add_generic_rpc_handlers([ 46 _ActualGenericRpcHandler(), 47 object(), 48 ]) 49 self.assertIn('grpc.GenericRpcHandler', 50 str(exception_context.exception)) 51 52 def test_failed_port_binding_exception(self): 53 server = grpc.server(None, options=(('grpc.so_reuseport', 0),)) 54 port = server.add_insecure_port('localhost:0') 55 bind_address = "localhost:%d" % port 56 57 with self.assertRaises(RuntimeError): 58 server.add_insecure_port(bind_address) 59 60 server_credentials = grpc.ssl_server_credentials([ 61 (resources.private_key(), resources.certificate_chain()) 62 ]) 63 with self.assertRaises(RuntimeError): 64 server.add_secure_port(bind_address, server_credentials) 65 66 67if __name__ == '__main__': 68 logging.basicConfig() 69 unittest.main(verbosity=2) 70