• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2021 The Pigweed Authors
2#
3# Licensed under the Apache License, Version 2.0 (the "License"); you may not
4# use this file except in compliance with the License. You may obtain a copy of
5# the License at
6#
7#     https://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, WITHOUT
11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12# License for the specific language governing permissions and limitations under
13# the License.
14"""Utilities for testing pw_rpc."""
15
16import argparse
17import subprocess
18import sys
19import tempfile
20import time
21from typing import Optional, Sequence
22
23TEMP_DIR_MARKER = '(pw_rpc:CREATE_TEMP_DIR)'
24
25
26def parse_test_server_args(
27        parser: argparse.ArgumentParser = None) -> argparse.Namespace:
28    """Parses arguments for running a Python-based integration test."""
29    if parser is None:
30        parser = argparse.ArgumentParser(
31            description=sys.modules['__main__'].__doc__)
32
33    parser.add_argument('--test-server-command',
34                        nargs='+',
35                        required=True,
36                        help='Command that starts the test server.')
37    parser.add_argument(
38        '--port',
39        type=int,
40        required=True,
41        help=('The port to use to connect to the test server. This value is '
42              'passed to the test server as the last argument.'))
43    parser.add_argument('unittest_args',
44                        nargs=argparse.REMAINDER,
45                        help='Arguments after "--" are passed to unittest.')
46
47    args = parser.parse_args()
48
49    # Append the port number to the test server command.
50    args.test_server_command.append(str(args.port))
51
52    # Make the script name argv[0] and drop the "--".
53    args.unittest_args = sys.argv[:1] + args.unittest_args[1:]
54
55    return args
56
57
58def _parse_subprocess_integration_test_args() -> argparse.Namespace:
59    parser = argparse.ArgumentParser(
60        description='Executes a test between two subprocesses')
61    parser.add_argument('--client', required=True, help='Client binary to run')
62    parser.add_argument('--server', required=True, help='Server binary to run')
63    parser.add_argument(
64        'common_args',
65        metavar='-- ...',
66        nargs=argparse.REMAINDER,
67        help=('Arguments to pass to both the server and client; '
68              f'pass {TEMP_DIR_MARKER} to generate a temporary directory'))
69
70    args = parser.parse_args()
71
72    if not args.common_args or args.common_args[0] != '--':
73        parser.error('The common arguments must start with "--"')
74
75    args.common_args.pop(0)
76
77    return args
78
79
80def execute_integration_test(server: str,
81                             client: str,
82                             common_args: Sequence[str],
83                             setup_time_s: float = 0.2) -> int:
84    temp_dir: Optional[tempfile.TemporaryDirectory] = None
85
86    if TEMP_DIR_MARKER in common_args:
87        temp_dir = tempfile.TemporaryDirectory(prefix='pw_rpc_test_')
88        common_args = [
89            temp_dir.name if a == TEMP_DIR_MARKER else a for a in common_args
90        ]
91
92    try:
93        server_process = subprocess.Popen([server, *common_args])
94        # TODO(pwbug/508): Replace this delay with some sort of IPC.
95        time.sleep(setup_time_s)
96
97        result = subprocess.run([client, *common_args]).returncode
98
99        server_process.terminate()
100        server_process.communicate()
101    finally:
102        if temp_dir:
103            temp_dir.cleanup()
104
105    return result
106
107
108if __name__ == '__main__':
109    sys.exit(
110        execute_integration_test(
111            **vars(_parse_subprocess_integration_test_args())))
112