xref: /aosp_15_r20/external/grpc-grpc/examples/python/compression/test/compression_example_test.py (revision cc02d7e222339f7a4f6ba5f422e6413f4bd931f2) 
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 for compression example."""
15
16import contextlib
17import os
18import socket
19import subprocess
20import unittest
21
22_BINARY_DIR = os.path.realpath(
23    os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
24)
25_SERVER_PATH = os.path.join(_BINARY_DIR, "server")
26_CLIENT_PATH = os.path.join(_BINARY_DIR, "client")
27
28
29@contextlib.contextmanager
30def _get_port():
31    sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
32    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
33    if sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT) == 0:
34        raise RuntimeError("Failed to set SO_REUSEPORT.")
35    sock.bind(("", 0))
36    try:
37        yield sock.getsockname()[1]
38    finally:
39        sock.close()
40
41
42class CompressionExampleTest(unittest.TestCase):
43    def test_compression_example(self):
44        with _get_port() as test_port:
45            server_process = subprocess.Popen(
46                (
47                    _SERVER_PATH,
48                    "--port",
49                    str(test_port),
50                    "--server_compression",
51                    "gzip",
52                    "--no_compress_every_n",
53                    "3",
54                )
55            )
56            try:
57                server_target = "localhost:{}".format(test_port)
58                client_process = subprocess.Popen(
59                    (
60                        _CLIENT_PATH,
61                        "--server",
62                        server_target,
63                        "--channel_compression",
64                        "gzip",
65                    )
66                )
67                client_return_code = client_process.wait()
68                self.assertEqual(0, client_return_code)
69                self.assertIsNone(server_process.poll())
70            finally:
71                server_process.kill()
72                server_process.wait()
73
74
75if __name__ == "__main__":
76    unittest.main(verbosity=2)
77