1# Copyright 2019 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"""Test of RPCs made using local credentials.""" 15 16from concurrent.futures import ThreadPoolExecutor 17import os 18import unittest 19 20import grpc 21 22 23class _GenericHandler(grpc.GenericRpcHandler): 24 def service(self, handler_call_details): 25 return grpc.unary_unary_rpc_method_handler( 26 lambda request, unused_context: request 27 ) 28 29 30class LocalCredentialsTest(unittest.TestCase): 31 def _create_server(self): 32 server = grpc.server(ThreadPoolExecutor()) 33 server.add_generic_rpc_handlers((_GenericHandler(),)) 34 return server 35 36 @unittest.skipIf( 37 os.name == "nt", "TODO(https://github.com/grpc/grpc/issues/20078)" 38 ) 39 def test_local_tcp(self): 40 server_addr = "localhost:{}" 41 channel_creds = grpc.local_channel_credentials( 42 grpc.LocalConnectionType.LOCAL_TCP 43 ) 44 server_creds = grpc.local_server_credentials( 45 grpc.LocalConnectionType.LOCAL_TCP 46 ) 47 48 server = self._create_server() 49 port = server.add_secure_port(server_addr.format(0), server_creds) 50 server.start() 51 with grpc.secure_channel( 52 server_addr.format(port), channel_creds 53 ) as channel: 54 self.assertEqual( 55 b"abc", 56 channel.unary_unary( 57 "/test/method", 58 _registered_method=True, 59 )(b"abc", wait_for_ready=True), 60 ) 61 server.stop(None) 62 63 @unittest.skipIf( 64 os.name == "nt", "Unix Domain Socket is not supported on Windows" 65 ) 66 def test_uds(self): 67 server_addr = "unix:/tmp/grpc_fullstack_test" 68 channel_creds = grpc.local_channel_credentials( 69 grpc.LocalConnectionType.UDS 70 ) 71 server_creds = grpc.local_server_credentials( 72 grpc.LocalConnectionType.UDS 73 ) 74 75 server = self._create_server() 76 server.add_secure_port(server_addr, server_creds) 77 server.start() 78 with grpc.secure_channel(server_addr, channel_creds) as channel: 79 self.assertEqual( 80 b"abc", 81 channel.unary_unary( 82 "/test/method", 83 _registered_method=True, 84 )(b"abc", wait_for_ready=True), 85 ) 86 server.stop(None) 87 88 89if __name__ == "__main__": 90 unittest.main() 91