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