• 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"""Conducts interop tests locally."""
15
16import logging
17import unittest
18
19import grpc
20from grpc.experimental import aio
21
22from src.proto.grpc.testing import test_pb2_grpc
23from tests.interop import resources
24from tests_aio.interop import methods
25from tests_aio.unit._test_base import AioTestBase
26from tests_aio.unit._test_server import start_test_server
27
28_SERVER_HOST_OVERRIDE = "foo.test.google.fr"
29
30
31class InteropTestCaseMixin:
32    """Unit test methods.
33
34    This class must be mixed in with unittest.TestCase and a class that defines
35    setUp and tearDown methods that manage a stub attribute.
36    """
37
38    _stub: test_pb2_grpc.TestServiceStub
39
40    async def test_empty_unary(self):
41        await methods.test_interoperability(
42            methods.TestCase.EMPTY_UNARY, self._stub, None
43        )
44
45    async def test_large_unary(self):
46        await methods.test_interoperability(
47            methods.TestCase.LARGE_UNARY, self._stub, None
48        )
49
50    async def test_server_streaming(self):
51        await methods.test_interoperability(
52            methods.TestCase.SERVER_STREAMING, self._stub, None
53        )
54
55    async def test_client_streaming(self):
56        await methods.test_interoperability(
57            methods.TestCase.CLIENT_STREAMING, self._stub, None
58        )
59
60    async def test_ping_pong(self):
61        await methods.test_interoperability(
62            methods.TestCase.PING_PONG, self._stub, None
63        )
64
65    async def test_cancel_after_begin(self):
66        await methods.test_interoperability(
67            methods.TestCase.CANCEL_AFTER_BEGIN, self._stub, None
68        )
69
70    async def test_cancel_after_first_response(self):
71        await methods.test_interoperability(
72            methods.TestCase.CANCEL_AFTER_FIRST_RESPONSE, self._stub, None
73        )
74
75    async def test_timeout_on_sleeping_server(self):
76        await methods.test_interoperability(
77            methods.TestCase.TIMEOUT_ON_SLEEPING_SERVER, self._stub, None
78        )
79
80    async def test_empty_stream(self):
81        await methods.test_interoperability(
82            methods.TestCase.EMPTY_STREAM, self._stub, None
83        )
84
85    async def test_status_code_and_message(self):
86        await methods.test_interoperability(
87            methods.TestCase.STATUS_CODE_AND_MESSAGE, self._stub, None
88        )
89
90    async def test_unimplemented_method(self):
91        await methods.test_interoperability(
92            methods.TestCase.UNIMPLEMENTED_METHOD, self._stub, None
93        )
94
95    async def test_unimplemented_service(self):
96        await methods.test_interoperability(
97            methods.TestCase.UNIMPLEMENTED_SERVICE, self._stub, None
98        )
99
100    async def test_custom_metadata(self):
101        await methods.test_interoperability(
102            methods.TestCase.CUSTOM_METADATA, self._stub, None
103        )
104
105    async def test_special_status_message(self):
106        await methods.test_interoperability(
107            methods.TestCase.SPECIAL_STATUS_MESSAGE, self._stub, None
108        )
109
110
111class InsecureLocalInteropTest(InteropTestCaseMixin, AioTestBase):
112    async def setUp(self):
113        address, self._server = await start_test_server()
114        self._channel = aio.insecure_channel(address)
115        self._stub = test_pb2_grpc.TestServiceStub(self._channel)
116
117    async def tearDown(self):
118        await self._channel.close()
119        await self._server.stop(None)
120
121
122class SecureLocalInteropTest(InteropTestCaseMixin, AioTestBase):
123    async def setUp(self):
124        server_credentials = grpc.ssl_server_credentials(
125            [(resources.private_key(), resources.certificate_chain())]
126        )
127        channel_credentials = grpc.ssl_channel_credentials(
128            resources.test_root_certificates()
129        )
130        channel_options = (
131            (
132                "grpc.ssl_target_name_override",
133                _SERVER_HOST_OVERRIDE,
134            ),
135        )
136
137        address, self._server = await start_test_server(
138            secure=True, server_credentials=server_credentials
139        )
140        self._channel = aio.secure_channel(
141            address, channel_credentials, channel_options
142        )
143        self._stub = test_pb2_grpc.TestServiceStub(self._channel)
144
145    async def tearDown(self):
146        await self._channel.close()
147        await self._server.stop(None)
148
149
150if __name__ == "__main__":
151    logging.basicConfig(level=logging.INFO)
152    unittest.main(verbosity=2)
153