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
31    @functools.wraps(fn)
32    def _wrapped():
33        try:
34            fn()
35        except Exception as e:
36            error_queue.put(e)
37            raise
38
39    return _wrapped
40
41
42def _run_in_subprocess(test_case):
43    error_queue = multiprocessing.Queue()
44    wrapped_case = _wrap_in_subprocess(error_queue, test_case)
45    proc = multiprocessing.Process(target=wrapped_case)
46    proc.start()
47    proc.join()
48    if not error_queue.empty():
49        raise error_queue.get()
50    assert proc.exitcode == 0, "Process exited with code {}".format(
51        proc.exitcode)
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    with _augmented_syspath(
68        ("tools/distrib/python/grpcio_tools/grpc_tools/test/",)):
69        protos = protoc._protos("simple.proto")
70        assert protos.SimpleMessage is not None
71
72
73def _test_import_services():
74    from grpc_tools import protoc
75    with _augmented_syspath(
76        ("tools/distrib/python/grpcio_tools/grpc_tools/test/",)):
77        protos = protoc._protos("simple.proto")
78        services = protoc._services("simple.proto")
79        assert services.SimpleMessageServiceStub is not None
80
81
82def _test_import_services_without_protos():
83    from grpc_tools import protoc
84    with _augmented_syspath(
85        ("tools/distrib/python/grpcio_tools/grpc_tools/test/",)):
86        services = protoc._services("simple.proto")
87        assert services.SimpleMessageServiceStub is not None
88
89
90def _test_proto_module_imported_once():
91    from grpc_tools import protoc
92    with _augmented_syspath(
93        ("tools/distrib/python/grpcio_tools/grpc_tools/test/",)):
94        protos = protoc._protos("simple.proto")
95        services = protoc._services("simple.proto")
96        complicated_protos = protoc._protos("complicated.proto")
97        simple_message = protos.SimpleMessage()
98        complicated_message = complicated_protos.ComplicatedMessage()
99        assert (simple_message.simpler_message.simplest_message.__class__ is
100                complicated_message.simplest_message.__class__)
101
102
103def _test_static_dynamic_combo():
104    with _augmented_syspath(
105        ("tools/distrib/python/grpcio_tools/grpc_tools/test/",)):
106        from grpc_tools import protoc  # isort:skip
107        import complicated_pb2
108        protos = protoc._protos("simple.proto")
109        static_message = complicated_pb2.ComplicatedMessage()
110        dynamic_message = protos.SimpleMessage()
111        assert (dynamic_message.simpler_message.simplest_message.__class__ is
112                static_message.simplest_message.__class__)
113
114
115def _test_combined_import():
116    from grpc_tools import protoc
117    protos, services = protoc._protos_and_services("simple.proto")
118    assert protos.SimpleMessage is not None
119    assert services.SimpleMessageServiceStub is not None
120
121
122def _test_syntax_errors():
123    from grpc_tools import protoc
124    try:
125        protos = protoc._protos("flawed.proto")
126    except Exception as e:
127        error_str = str(e)
128        assert "flawed.proto" in error_str
129        assert "17:23" in error_str
130        assert "21:23" in error_str
131    else:
132        assert False, "Compile error expected. None occurred."
133
134
135class ProtocTest(unittest.TestCase):
136
137    def test_import_protos(self):
138        _run_in_subprocess(_test_import_protos)
139
140    def test_import_services(self):
141        _run_in_subprocess(_test_import_services)
142
143    def test_import_services_without_protos(self):
144        _run_in_subprocess(_test_import_services_without_protos)
145
146    def test_proto_module_imported_once(self):
147        _run_in_subprocess(_test_proto_module_imported_once)
148
149    def test_static_dynamic_combo(self):
150        _run_in_subprocess(_test_static_dynamic_combo)
151
152    def test_combined_import(self):
153        _run_in_subprocess(_test_combined_import)
154
155    def test_syntax_errors(self):
156        _run_in_subprocess(_test_syntax_errors)
157
158
159if __name__ == '__main__':
160    unittest.main()
161