• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2020 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"""Tests for protoc."""
15
16from __future__ import absolute_import
17from __future__ import division
18from __future__ import print_function
19
20import contextlib
21import functools
22import multiprocessing
23import sys
24import unittest
25
26
27# TODO(https://github.com/grpc/grpc/issues/23847): Deduplicate this mechanism with
28# the grpcio_tests module.
29def _wrap_in_subprocess(error_queue, fn):
30    @functools.wraps(fn)
31    def _wrapped():
32        try:
33            fn()
34        except Exception as e:
35            error_queue.put(e)
36            raise
37
38    return _wrapped
39
40
41def _run_in_subprocess(test_case):
42    error_queue = multiprocessing.Queue()
43    wrapped_case = _wrap_in_subprocess(error_queue, test_case)
44    proc = multiprocessing.Process(target=wrapped_case)
45    proc.start()
46    proc.join()
47    if not error_queue.empty():
48        raise error_queue.get()
49    assert proc.exitcode == 0, "Process exited with code {}".format(
50        proc.exitcode
51    )
52
53
54@contextlib.contextmanager
55def _augmented_syspath(new_paths):
56    original_sys_path = sys.path
57    if new_paths is not None:
58        sys.path = list(new_paths) + sys.path
59    try:
60        yield
61    finally:
62        sys.path = original_sys_path
63
64
65def _test_import_protos():
66    from grpc_tools import protoc
67
68    with _augmented_syspath(
69        ("tools/distrib/python/grpcio_tools/grpc_tools/test/",)
70    ):
71        protos = protoc._protos("simple.proto")
72        assert protos.SimpleMessage is not None
73
74
75def _test_import_services():
76    from grpc_tools import protoc
77
78    with _augmented_syspath(
79        ("tools/distrib/python/grpcio_tools/grpc_tools/test/",)
80    ):
81        protos = protoc._protos("simple.proto")
82        services = protoc._services("simple.proto")
83        assert services.SimpleMessageServiceStub is not None
84
85
86def _test_import_services_without_protos():
87    from grpc_tools import protoc
88
89    with _augmented_syspath(
90        ("tools/distrib/python/grpcio_tools/grpc_tools/test/",)
91    ):
92        services = protoc._services("simple.proto")
93        assert services.SimpleMessageServiceStub is not None
94
95
96def _test_proto_module_imported_once():
97    from grpc_tools import protoc
98
99    with _augmented_syspath(
100        ("tools/distrib/python/grpcio_tools/grpc_tools/test/",)
101    ):
102        protos = protoc._protos("simple.proto")
103        services = protoc._services("simple.proto")
104        complicated_protos = protoc._protos("complicated.proto")
105        simple_message = protos.SimpleMessage()
106        complicated_message = complicated_protos.ComplicatedMessage()
107        assert (
108            simple_message.simpler_message.simplest_message.__class__
109            is complicated_message.simplest_message.__class__
110        )
111
112
113def _test_static_dynamic_combo():
114    with _augmented_syspath(
115        ("tools/distrib/python/grpcio_tools/grpc_tools/test/",)
116    ):
117        from grpc_tools import protoc  # isort:skip
118        import complicated_pb2
119
120        protos = protoc._protos("simple.proto")
121        static_message = complicated_pb2.ComplicatedMessage()
122        dynamic_message = protos.SimpleMessage()
123        assert (
124            dynamic_message.simpler_message.simplest_message.__class__
125            is static_message.simplest_message.__class__
126        )
127
128
129def _test_combined_import():
130    from grpc_tools import protoc
131
132    protos, services = protoc._protos_and_services("simple.proto")
133    assert protos.SimpleMessage is not None
134    assert services.SimpleMessageServiceStub is not None
135
136
137def _test_syntax_errors():
138    from grpc_tools import protoc
139
140    try:
141        protos = protoc._protos("flawed.proto")
142    except Exception as e:
143        error_str = str(e)
144        assert "flawed.proto" in error_str
145        assert "17:23" in error_str
146        assert "21:23" in error_str
147    else:
148        assert False, "Compile error expected. None occurred."
149
150
151class ProtocTest(unittest.TestCase):
152    def test_import_protos(self):
153        _run_in_subprocess(_test_import_protos)
154
155    def test_import_services(self):
156        _run_in_subprocess(_test_import_services)
157
158    def test_import_services_without_protos(self):
159        _run_in_subprocess(_test_import_services_without_protos)
160
161    def test_proto_module_imported_once(self):
162        _run_in_subprocess(_test_proto_module_imported_once)
163
164    def test_static_dynamic_combo(self):
165        _run_in_subprocess(_test_static_dynamic_combo)
166
167    def test_combined_import(self):
168        _run_in_subprocess(_test_combined_import)
169
170    def test_syntax_errors(self):
171        _run_in_subprocess(_test_syntax_errors)
172
173
174if __name__ == "__main__":
175    unittest.main()
176