import os
from pathlib import Path
import signal
-from subprocess import Popen, PIPE, STDOUT
+from subprocess import Popen, PIPE, STDOUT, call
import sys
import time
import venv
+import datetime
+import re
# Required Std. Path Variables
test_dir = os.path.dirname(os.path.realpath(__file__))
ws_root = os.path.dirname(test_dir)
build_root = os.path.join(ws_root, "build-root")
-venv_dir = os.path.join(test_dir, "venv")
+venv_dir = os.path.join(build_root, "test", "venv")
venv_bin_dir = os.path.join(venv_dir, "bin")
venv_lib_dir = os.path.join(venv_dir, "lib")
venv_run_dir = os.path.join(venv_dir, "run")
pip_version = "22.0.4"
pip_tools_version = "6.6.0"
-# Test requirement files
-test_requirements_file = os.path.join(test_dir, "requirements.txt")
-# Auto-generated requirement file
+# Compiled pip requirements file
pip_compiled_requirements_file = os.path.join(test_dir, "requirements-3.txt")
signal.signal(signal.SIGTERM, handler)
-def show_progress(stream):
+def show_progress(stream, exclude_pattern=None):
"""
Read lines from a subprocess stdout/stderr streams and write
to sys.stdout & the logfile
+
+ arguments:
+ stream - subprocess stdout or stderr data stream
+ exclude_pattern - lines matching this reg-ex will be excluded
+ from stdout.
"""
while True:
s = stream.readline()
# Filter the annoying SIGTERM signal from the output when VPP is
# terminated after a test run
if "SIGTERM" not in data:
- sys.stdout.write(data)
+ if exclude_pattern is not None:
+ if bool(re.search(exclude_pattern, data)) is False:
+ sys.stdout.write(data)
+ else:
+ sys.stdout.write(data)
logging.debug(data)
sys.stdout.flush()
stream.close()
os.environ[
"CUSTOM_COMPILE_COMMAND"
] = "make test-refresh-deps (or update requirements.txt)"
- # Cleanup previously auto-generated pip req. file
- try:
- os.unlink(pip_compiled_requirements_file)
- except OSError:
- pass
# Set the venv python executable & binary install path
env_exe = context.env_exe
bin_path = context.bin_path
test_req = [
["pip", "install", "pip===%s" % pip_version],
["pip", "install", "pip-tools===%s" % pip_tools_version],
- [
- "piptools",
- "compile",
- "-q",
- "--generate-hashes",
- test_requirements_file,
- "--output-file",
- pip_compiled_requirements_file,
- ],
["piptools", "sync", pip_compiled_requirements_file],
["pip", "install", "-e", papi_python_src_dir],
]
# Runs a test inside a spawned QEMU VM
# If a kernel image is not provided, a linux-image-kvm image is
# downloaded to the test_data_dir
-def vm_test_runner(test_name, kernel_image, test_data_dir, cpu_mask, mem):
+def vm_test_runner(test_name, kernel_image, test_data_dir, cpu_mask, mem, jobs="auto"):
script = os.path.join(test_dir, "scripts", "run_vpp_in_vm.sh")
+ os.environ["TEST_JOBS"] = str(jobs)
p = Popen(
[script, test_name, kernel_image, test_data_dir, cpu_mask, mem],
stdout=PIPE,
- stderr=STDOUT,
cwd=ws_root,
)
- show_progress(p.stdout)
+ # Show only the test result without clobbering the stdout.
+ # The VM console displays VPP stderr & Linux IPv6 netdev change
+ # messages, which is logged by default and can be excluded.
+ exclude_pattern = r"vpp\[\d+\]:|ADDRCONF\(NETDEV_CHANGE\):"
+ show_progress(p.stdout, exclude_pattern)
post_vm_test_run()
logging.basicConfig(filename=filename, level=logging.DEBUG)
+def run_tests_in_venv(
+ test,
+ jobs,
+ log_dir,
+ socket_dir="",
+ running_vpp=False,
+ extended=False,
+):
+ """Runs tests in the virtual environment set by venv_dir.
+
+ Arguments:
+ test: Name of the test to run
+ jobs: Maximum concurrent test jobs
+ log_dir: Directory location for storing log files
+ socket_dir: Use running VPP's socket files
+ running_vpp: True if tests are run against a running VPP
+ extended: Run extended tests
+ """
+ script = os.path.join(test_dir, "scripts", "run.sh")
+ args = [
+ f"--venv-dir={venv_dir}",
+ f"--vpp-ws-dir={ws_root}",
+ f"--socket-dir={socket_dir}",
+ f"--filter={test}",
+ f"--jobs={jobs}",
+ f"--log-dir={log_dir}",
+ f"--tmp-dir={log_dir}",
+ f"--cache-vpp-output",
+ ]
+ if running_vpp:
+ args = args + [f"--use-running-vpp"]
+ if extended:
+ args = args + [f"--extended"]
+ print(f"Running script: {script} " f"{' '.join(args)}")
+ process_args = [script] + args
+ call(process_args)
+
+
if __name__ == "__main__":
# Build a Virtual Environment for running tests on host & QEMU
+ # (TODO): Create a single config object by merging the below args with
+ # config.py after gathering dev use-cases.
parser = argparse.ArgumentParser(
description="Run VPP Unit Tests", formatter_class=argparse.RawTextHelpFormatter
)
parser.add_argument(
"--vm",
dest="vm",
- required=True,
+ required=False,
action="store_true",
help="Run Test Inside a QEMU VM",
)
parser.add_argument(
- "-d",
"--debug",
dest="debug",
required=False,
help="Run Tests on Debug Build",
)
parser.add_argument(
- "-r",
"--release",
dest="release",
required=False,
help="Run Tests on release Build",
)
parser.add_argument(
+ "-t",
"--test",
dest="test_name",
required=False,
action="store",
default="",
- help="Tests to Run",
+ help="Test Name or Test filter",
)
parser.add_argument(
"--vm-kernel-image",
default="2",
help="Guest Memory in Gibibytes\n" "E.g. 4 (Default: 2)",
)
+ parser.add_argument(
+ "--log-dir",
+ action="store",
+ default=os.path.abspath(f"./test-run-{datetime.date.today()}"),
+ help="directory where to store directories "
+ "containing log files (default: ./test-run-YYYY-MM-DD)",
+ )
+ parser.add_argument(
+ "--jobs",
+ action="store",
+ default="auto",
+ help="maximum concurrent test jobs",
+ )
+ parser.add_argument(
+ "-r",
+ "--use-running-vpp",
+ dest="running_vpp",
+ required=False,
+ action="store_true",
+ default=False,
+ help="Runs tests against a running VPP.",
+ )
+ parser.add_argument(
+ "-d",
+ "--socket-dir",
+ dest="socket_dir",
+ required=False,
+ action="store",
+ default="",
+ help="Relative or absolute path of running VPP's socket directory "
+ "containing api.sock & stats.sock files.\n"
+ "Default: /var/run/vpp if VPP is started as the root user, else "
+ "/var/run/user/${uid}/vpp.",
+ )
+ parser.add_argument(
+ "-e",
+ "--extended",
+ dest="extended",
+ required=False,
+ action="store_true",
+ default=False,
+ help="Run extended tests.",
+ )
args = parser.parse_args()
+ vm_tests = False
# Enable VM tests
if args.vm and args.test_name:
test_data_dir = "/tmp/vpp-vm-tests"
debug = False if args.release else True
build_vpp(debug, args.release)
set_environ()
- if vm_tests:
+ if args.running_vpp:
+ print("Tests will be run against a running VPP..")
+ elif not vm_tests:
+ print("Tests will be run by spawning a new VPP instance..")
+ # Run tests against a running VPP or a new instance of VPP
+ if not vm_tests:
+ run_tests_in_venv(
+ test=args.test_name,
+ jobs=args.jobs,
+ log_dir=args.log_dir,
+ socket_dir=args.socket_dir,
+ running_vpp=args.running_vpp,
+ extended=args.extended,
+ )
+ # Run tests against a VPP inside a VM
+ else:
print("Running VPP unit test(s):{0} inside a QEMU VM".format(args.test_name))
# Check Available CPUs & Usable Memory
cpus = expand_mix_string(args.vm_cpu_list)
print(f"Error: Mem Size:{args.vm_mem}G > Avail Mem:{avail_mem}G")
sys.exit(1)
vm_test_runner(
- args.test_name, args.kernel_image, test_data_dir, cpus, f"{args.vm_mem}G"
+ args.test_name,
+ args.kernel_image,
+ test_data_dir,
+ cpus,
+ f"{args.vm_mem}G",
+ args.jobs,
)
Code Review / vpp.git / blobdiff