xref: /aosp_15_r20/external/grpc-grpc/src/python/grpcio_tests/tests/interop/server.py (revision cc02d7e222339f7a4f6ba5f422e6413f4bd931f2) 
1# Copyright 2015 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"""The Python implementation of the GRPC interoperability test server."""
15
16from concurrent import futures
17import logging
18
19from absl import app
20from absl.flags import argparse_flags
21import grpc
22
23from src.proto.grpc.testing import test_pb2_grpc
24from tests.interop import resources
25from tests.interop import service
26from tests.unit import test_common
27
28logging.basicConfig()
29_LOGGER = logging.getLogger(__name__)
30
31
32def parse_interop_server_arguments(argv):
33    parser = argparse_flags.ArgumentParser()
34    parser.add_argument(
35        "--port", type=int, required=True, help="the port on which to serve"
36    )
37    parser.add_argument(
38        "--use_tls",
39        default=False,
40        type=resources.parse_bool,
41        help="require a secure connection",
42    )
43    parser.add_argument(
44        "--use_alts",
45        default=False,
46        type=resources.parse_bool,
47        help="require an ALTS connection",
48    )
49    return parser.parse_args(argv[1:])
50
51
52def get_server_credentials(use_tls):
53    if use_tls:
54        private_key = resources.private_key()
55        certificate_chain = resources.certificate_chain()
56        return grpc.ssl_server_credentials(((private_key, certificate_chain),))
57    else:
58        return grpc.alts_server_credentials()
59
60
61def serve(args):
62    server = test_common.test_server()
63    test_pb2_grpc.add_TestServiceServicer_to_server(
64        service.TestService(), server
65    )
66    if args.use_tls or args.use_alts:
67        credentials = get_server_credentials(args.use_tls)
68        server.add_secure_port("[::]:{}".format(args.port), credentials)
69    else:
70        server.add_insecure_port("[::]:{}".format(args.port))
71
72    server.start()
73    _LOGGER.info("Server serving.")
74    server.wait_for_termination()
75    _LOGGER.info("Server stopped; exiting.")
76
77
78if __name__ == "__main__":
79    app.run(serve, flags_parser=parse_interop_server_arguments)
80