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