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 dynamic stub import API.""" 15 16import contextlib 17import functools 18import logging 19import multiprocessing 20import os 21import sys 22import unittest 23 24from tests.unit import test_common 25 26_DATA_DIR = os.path.join("tests", "unit", "data") 27 28 29@contextlib.contextmanager 30def _grpc_tools_unimportable(): 31 original_sys_path = sys.path 32 sys.path = [path for path in sys.path if "grpcio_tools" not in str(path)] 33 try: 34 import grpc_tools 35 except ImportError: 36 pass 37 else: 38 del grpc_tools 39 sys.path = original_sys_path 40 raise unittest.SkipTest("Failed to make grpc_tools unimportable.") 41 try: 42 yield 43 finally: 44 sys.path = original_sys_path 45 46 47def _collect_errors(fn): 48 @functools.wraps(fn) 49 def _wrapped(error_queue): 50 try: 51 fn() 52 except Exception as e: 53 error_queue.put(e) 54 raise 55 56 return _wrapped 57 58 59def _python3_check(fn): 60 @functools.wraps(fn) 61 def _wrapped(): 62 if sys.version_info[0] == 3: 63 fn() 64 else: 65 _assert_unimplemented("Python 3") 66 67 return _wrapped 68 69 70def _run_in_subprocess(test_case): 71 sys.path.insert( 72 0, os.path.join(os.path.realpath(os.path.dirname(__file__)), "..") 73 ) 74 error_queue = multiprocessing.Queue() 75 proc = multiprocessing.Process(target=test_case, args=(error_queue,)) 76 proc.start() 77 proc.join() 78 sys.path.pop(0) 79 if not error_queue.empty(): 80 raise error_queue.get() 81 assert proc.exitcode == 0, "Process exited with code {}".format( 82 proc.exitcode 83 ) 84 85 86def _assert_unimplemented(msg_substr): 87 import grpc 88 89 try: 90 protos, services = grpc.protos_and_services( 91 "tests/unit/data/foo/bar.proto" 92 ) 93 except NotImplementedError as e: 94 assert msg_substr in str(e), "{} was not in '{}'".format( 95 msg_substr, str(e) 96 ) 97 else: 98 assert False, "Did not raise NotImplementedError" 99 100 101@_collect_errors 102@_python3_check 103def _test_sunny_day(): 104 import grpc 105 106 protos, services = grpc.protos_and_services( 107 os.path.join(_DATA_DIR, "foo", "bar.proto") 108 ) 109 assert protos.BarMessage is not None 110 assert services.BarStub is not None 111 112 113@_collect_errors 114@_python3_check 115def _test_well_known_types(): 116 import grpc 117 118 protos, services = grpc.protos_and_services( 119 os.path.join(_DATA_DIR, "foo", "bar_with_wkt.proto") 120 ) 121 assert protos.BarMessage is not None 122 assert services.BarStub is not None 123 124 125@_collect_errors 126@_python3_check 127def _test_grpc_tools_unimportable(): 128 with _grpc_tools_unimportable(): 129 _assert_unimplemented("grpcio-tools") 130 131 132# NOTE(rbellevi): multiprocessing.Process fails to pickle function objects 133# when they do not come from the "__main__" module, so this test passes 134# if run directly on Windows or MacOS, but not if started by the test runner. 135@unittest.skipIf( 136 os.name == "nt" or "darwin" in sys.platform, 137 "Windows and MacOS multiprocessing unsupported", 138) 139class DynamicStubTest(unittest.TestCase): 140 def test_sunny_day(self): 141 _run_in_subprocess(_test_sunny_day) 142 143 def test_well_known_types(self): 144 _run_in_subprocess(_test_well_known_types) 145 146 def test_grpc_tools_unimportable(self): 147 _run_in_subprocess(_test_grpc_tools_unimportable) 148 149 150if __name__ == "__main__": 151 logging.basicConfig() 152 unittest.main(verbosity=2) 153