blob: a9ab4d373e48ce44f9fea241355bd361cf9536e6 [file] [log] [blame]
# Copyright 2021 The Pigweed Authors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
"""Utilities for testing pw_rpc."""
import argparse
import subprocess
import sys
import tempfile
import time
from typing import Optional, Sequence
TEMP_DIR_MARKER = '(pw_rpc:CREATE_TEMP_DIR)'
def parse_test_server_args(
parser: argparse.ArgumentParser = None) -> argparse.Namespace:
"""Parses arguments for running a Python-based integration test."""
if parser is None:
parser = argparse.ArgumentParser(
description=sys.modules['__main__'].__doc__)
parser.add_argument('--test-server-command',
nargs='+',
required=True,
help='Command that starts the test server.')
parser.add_argument(
'--port',
type=int,
required=True,
help=('The port to use to connect to the test server. This value is '
'passed to the test server as the last argument.'))
parser.add_argument('unittest_args',
nargs=argparse.REMAINDER,
help='Arguments after "--" are passed to unittest.')
args = parser.parse_args()
# Append the port number to the test server command.
args.test_server_command.append(str(args.port))
# Make the script name argv[0] and drop the "--".
args.unittest_args = sys.argv[:1] + args.unittest_args[1:]
return args
def _parse_subprocess_integration_test_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description='Executes a test between two subprocesses')
parser.add_argument('--client', required=True, help='Client binary to run')
parser.add_argument('--server', required=True, help='Server binary to run')
parser.add_argument(
'common_args',
metavar='-- ...',
nargs=argparse.REMAINDER,
help=('Arguments to pass to both the server and client; '
f'pass {TEMP_DIR_MARKER} to generate a temporary directory'))
args = parser.parse_args()
if not args.common_args or args.common_args[0] != '--':
parser.error('The common arguments must start with "--"')
args.common_args.pop(0)
return args
def execute_integration_test(server: str,
client: str,
common_args: Sequence[str],
setup_time_s: float = 0.2) -> int:
temp_dir: Optional[tempfile.TemporaryDirectory] = None
if TEMP_DIR_MARKER in common_args:
temp_dir = tempfile.TemporaryDirectory(prefix='pw_rpc_test_')
common_args = [
temp_dir.name if a == TEMP_DIR_MARKER else a for a in common_args
]
try:
server_process = subprocess.Popen([server, *common_args])
# TODO(pwbug/508): Replace this delay with some sort of IPC.
time.sleep(setup_time_s)
result = subprocess.run([client, *common_args]).returncode
server_process.terminate()
server_process.communicate()
finally:
if temp_dir:
temp_dir.cleanup()
return result
if __name__ == '__main__':
sys.exit(
execute_integration_test(
**vars(_parse_subprocess_integration_test_args())))