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