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