diff --git a/backends/bmv2/bmv2stf.py b/backends/bmv2/bmv2stf.py index acdf2d194b..e88df581c6 100755 --- a/backends/bmv2/bmv2stf.py +++ b/backends/bmv2/bmv2stf.py @@ -488,8 +488,8 @@ def p_meter_rate_one(self, p): # able to invoke the BMV2 simulator, create a CLI file # and test packets in pcap files. class RunBMV2(object): - def __init__(self, folder, options, jsonfile): - self.clifile = folder + "/cli.txt" + def __init__(self, folder: Path, options, jsonfile: Path) -> None: + self.clifile = folder.joinpath("cli.txt") self.jsonfile = jsonfile self.stffile = None self.folder = folder @@ -516,8 +516,8 @@ def readJson(self): for t in self.json["pipelines"][1]["tables"]: self.tables.append(BMV2Table(t)) - def filename(self, interface, direction): - return self.folder + "/" + self.pcapPrefix + str(interface) + "_" + direction + ".pcap" + def filename(self, interface :str , direction : str) -> Path: + return self.folder.joinpath(f"{self.pcapPrefix}{interface}_{direction}.pcap") def interface_of_filename(self, f): return int(os.path.basename(f).rstrip(".pcap").lstrip(self.pcapPrefix).rsplit("_", 1)[0]) @@ -529,7 +529,7 @@ def do_cli_command(self, cmd): self.packetDelay = 1 def execute_stf_command(self, stf_entry): - if self.options.verbose and stf_entry: + if stf_entry: testutils.log.info("STF Command: %s", stf_entry) cmd = stf_entry[0] if cmd == "": @@ -686,7 +686,7 @@ def interfaceArgs(self): result.append("-i " + str(interface) + "@" + self.pcapPrefix + str(interface)) return result - def generate_model_inputs(self, stf_map): + def generate_model_inputs(self, stf_map: testutils.Dict[str, str]) -> int: for entry in stf_map: cmd = entry[0] if cmd in ["packet", "expect"]: @@ -711,7 +711,7 @@ def check_switch_server_ready(self, proc, thriftPort): if result == 0: return True - def run(self, stf_map): + def run(self, stf_map) -> int: testutils.log.info("Running model") wait = 0 # Time to wait before model starts running @@ -760,7 +760,7 @@ def run(self, stf_map): sw = subprocess.Popen(runswitch, cwd=self.folder) def openInterface(ifname): - fp = self.interfaces[interface] = scapy_util.RawPcapWriter(ifname, linktype=0) + fp = self.interfaces[interface] = scapy_util.RawPcapWriter(str(ifname), linktype=0) fp._write_header(None) # Try to open input interfaces. Each time, we set a 2 second @@ -845,16 +845,18 @@ def openInterface(ifname): testutils.log.info("Execution completed") return rv - def showLog(self): - with open(self.folder + "/" + self.switchLogFile + ".txt") as a: + def showLog(self) -> None: + """Show the log file""" + folder = self.folder.joinpath(self.switchLogFile).with_suffix(".txt") + with folder.open() as a: log = a.read() testutils.log.info("Log file:\n%s", log) - def checkOutputs(self): + def checkOutputs(self) -> int: """Checks if the output of the filter matches expectations""" testutils.log.info("Comparing outputs") direction = "out" - for file in glob(self.filename("*", direction)): + for file in glob(str(self.filename("*", direction))): testutils.log.info("Checking file %s", file) interface = self.interface_of_filename(file) if os.stat(file).st_size == 0: @@ -907,8 +909,8 @@ def checkOutputs(self): testutils.log.info("All went well.") return testutils.SUCCESS - def parse_stf_file(self, testfile): - with open(testfile) as raw_stf: + def parse_stf_file(self, testfile: Path) -> tuple[dict[str, str], int]: + with testfile.open() as raw_stf: parser = Bmv2StfParser() stf_str = raw_stf.read() return parser.parse(stf_str) diff --git a/backends/bmv2/run-bmv2-test.py b/backends/bmv2/run-bmv2-test.py index 94ec54d5da..cdac77be4e 100755 --- a/backends/bmv2/run-bmv2-test.py +++ b/backends/bmv2/run-bmv2-test.py @@ -19,110 +19,151 @@ import argparse import logging import os -import shutil +import random +import subprocess import sys import tempfile +import time +import traceback +import uuid +from datetime import datetime from pathlib import Path -from subprocess import Popen -from threading import Thread +from typing import Any, Dict, List, Optional + +PARSER = argparse.ArgumentParser() +PARSER.add_argument( + "rootdir", + help="The root directory of the compiler source tree." + "This is used to import P4C's Python libraries", +) +PARSER.add_argument("p4_file", help="the p4 file to process") +PARSER.add_argument( + "-tf", + "--test_file", + dest="test_file", + help=( + "Provide the path for the stf file for this test. " + "If no path is provided, the script will search for an" + " stf file in the same folder." + ), +) +PARSER.add_argument( + "-td", + "--testdir", + dest="testdir", + help="The location of the test directory.", +) +PARSER.add_argument( + "-b", + "--nocleanup", + action="store_true", + dest="nocleanup", + help="Do not remove temporary results for failing tests.", +) +PARSER.add_argument( + "-n", + "--num-ifaces", + default=8, + dest="num_ifaces", + help="How many virtual interfaces to create.", +) +PARSER.add_argument( + "-nn", + "--use-nanomsg", + action="store_true", + dest="use_nn", + help="Use nanomsg for packet sending instead of virtual interfaces.", +) +PARSER.add_argument( + "-ll", + "--log_level", + dest="log_level", + default="WARNING", + choices=["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG", "NOTSET"], + help="The log level to choose.", +) +PARSER.add_argument( + "-f", + "--replace", + action="store_true", + help="replace reference outputs with newly generated ones", +) +PARSER.add_argument("-p", "--usePsa", dest="use_psa", action="store_true", help="Use psa switch") +PARSER.add_argument( + "-bd", + "--buildir", + dest="builddir", + help="The path to the compiler build directory, default is current directory.", +) +PARSER.add_argument("-pp", dest="pp", help="pass this option to the compiler") +PARSER.add_argument("-gdb", "--gdb", action="store_true", help="Run the compiler under gdb.") +PARSER.add_argument("-lldb", "--lldb", action="store_true", help="Run the compiler under lldb.") +PARSER.add_argument( + "-a", + dest="compiler_options", + default=[], + action="append", + nargs="?", + help="Pass this option string to the compiler", +) +PARSER.add_argument( + "--target-specific-switch-arg", + dest="switch_options", + default=[], + action="append", + nargs="?", + help="Pass this target-specific option to the switch", +) +PARSER.add_argument( + "--init", + dest="init_cmds", + default=[], + action="append", + nargs="?", + help="Run before the start of the test", +) +PARSER.add_argument("--observation-log", dest="obs_log", help="save packet output to ") + + +# Parse options and process argv +ARGS, ARGV = PARSER.parse_known_args() + +# Append the root directory to the import path. +FILE_DIR = Path(__file__).resolve().parent +ROOT_DIR = Path(ARGS.rootdir).absolute() +sys.path.append(str(ROOT_DIR)) from bmv2stf import RunBMV2 from scapy.layers.all import * from scapy.utils import * -sys.path.insert(0, os.path.dirname(os.path.realpath(__file__)) + "/../../tools") -from testutils import FAILURE, SUCCESS, check_if_dir, check_if_file - - -def parse_args(): - parser = argparse.ArgumentParser() - parser.add_argument("rootdir", help="the root directory of the compiler source tree") - parser.add_argument("p4filename", help="the p4 file to process") - parser.add_argument( - "-b", - "--nocleanup", - action="store_false", - help="do not remove temporary results for failing tests", - ) - parser.add_argument( - "-bd", - "--buildir", - dest="builddir", - help="The path to the compiler build directory, default is current directory.", - ) - parser.add_argument("-v", "--verbose", action="store_true", help="verbose operation") - parser.add_argument( - "-f", - "--replace", - action="store_true", - help="replace reference outputs with newly generated ones", - ) - parser.add_argument( - "-p", "--usePsa", dest="use_psa", action="store_true", help="Use psa switch" - ) - parser.add_argument("-pp", dest="pp", help="pass this option to the compiler") - parser.add_argument("-gdb", "--gdb", action="store_true", help="Run the compiler under gdb.") - parser.add_argument("-lldb", "--lldb", action="store_true", help="Run the compiler under lldb.") - parser.add_argument( - "-a", - dest="compiler_options", - default=[], - action="append", - nargs="?", - help="Pass this option string to the compiler", - ) - parser.add_argument( - "--target-specific-switch-arg", - dest="switch_options", - default=[], - action="append", - nargs="?", - help="Pass this target-specific option to the switch", - ) - parser.add_argument( - "--init", - dest="init_cmds", - default=[], - action="append", - nargs="?", - help="Run before the start of the test", - ) - parser.add_argument("--observation-log", dest="obs_log", help="save packet output to ") - parser.add_argument( - "-tf", - "--testFile", - dest="testFile", - help=( - "Provide the path for the stf file for this test. " - "If no path is provided, the script will search for an" - " stf file in the same folder." - ), - ) - return parser.parse_known_args() +from tools import testutils # pylint: disable=wrong-import-position class Options: - def __init__(self): - self.binary = "" # this program's name - self.cleanupTmp = True # if false do not remote tmp folder created - self.p4filename = "" # file that is being compiled - self.compilerSrcDir = "" # path to compiler source tree - self.compilerBuildDir = "" # path to compiler build directory - self.testFile = "" # path to stf test file that is used - self.testName = None # Name of the test - self.verbose = False - self.replace = False # replace previous outputs - self.compilerOptions = [] - self.switchTargetSpecificOptions = [] - self.hasBMv2 = False # Is the behavioral model installed? - self.usePsa = False # Use the psa switch behavioral model? - self.runDebugger = False - # Log packets produced by the BMV2 model if path to log is supplied - self.observationLog = None - self.initCommands = [] - - -def nextWord(text, sep=" "): + # File that is being compiled. + p4_file: Path = Path(".") + # Path to stf test file that is used. + test_file: Optional[Path] = None + # Actual location of the test framework. + testdir: Path = Path(".") + # The base directory of the compiler.. + rootdir: Path = Path(".") + cleanupTmp = True # if false do not remote tmp folder created + compiler_build_dir: Path = Path(".") # path to compiler build directory + testName: str = "" # Name of the test + replace: bool = False # replace previous outputs + compiler_options: List[str] = [] + switch_target_specific_options: List[str] = [] + hasBMv2: bool = False # Is the behavioral model installed? + usePsa: bool = False # Use the psa switch behavioral model? + runDebugger: str = "" + # Log packets produced by the BMV2 model if path to log is supplied + observationLog: Optional[Path] = None + init_commands: List[str] = [] + + +def nextWord(text: str, sep: str = " ") -> tuple[str, str]: # Split a text at the indicated separator. # Note that the separator can be a string. # Separator is discarded. @@ -130,28 +171,28 @@ def nextWord(text, sep=" "): if pos < 0: return text, "" l, r = text[0:pos].strip(), text[pos + len(sep) : len(text)].strip() - # print(text, "/", sep, "->", l, "#", r) + # testutils.log.info(text, "/", sep, "->", l, "#", r) return l, r class ConfigH: # Represents an autoconf config.h file # fortunately the structure of these files is very constrained - def __init__(self, file): + def __init__(self, file: Path) -> None: self.file = file - self.vars = {} + self.vars: Dict[str, str] = {} with open(file, "r", encoding="utf-8") as a: self.text = a.read() self.ok = False self.parse() - def parse(self): + def parse(self) -> None: while self.text != "": self.text = self.text.strip() if self.text.startswith("/*"): end = self.text.find("*/") if end < 1: - reportError("Unterminated comment in config file") + testutils.log.error("Unterminated comment in config file") return self.text = self.text[end + 2 : len(self.text)] elif self.text.startswith("#define"): @@ -162,246 +203,197 @@ def parse(self): elif self.text.startswith("#ifndef"): _, self.text = nextWord(self.text, "#endif") else: - reportError("Unexpected text:", self.text) + testutils.log.error("Unexpected text:", self.text) return self.ok = True - def __str__(self): + def __str__(self) -> str: return str(self.vars) -def isError(p4filename): +def isError(p4_file: Path) -> bool: # True if the filename represents a p4 program that should fail - return "_errors" in p4filename - - -def reportError(*message): - print("***", *message) - - -class Local: - # object to hold local vars accessible to nested functions - process = None - - -def run_timeout(options, args, timeout, stderr): - if options.verbose: - print("Executing ", " ".join(args)) - local = Local() - - def target(): - procstderr = None - if stderr is not None: - procstderr = open(stderr, "w", encoding="utf-8") - local.process = Popen(args, stderr=procstderr) - local.process.wait() - - thread = Thread(target=target) - thread.start() - thread.join(timeout) - if thread.is_alive(): - print("Timeout ", " ".join(args), file=sys.stderr) - local.process.terminate() - thread.join() - if local.process is None: - # never even started - reportError("Process failed to start") - return -1 - if options.verbose: - print("Exit code ", local.process.returncode) - return local.process.returncode - - -timeout = 10 * 60 - - -def run_model(options, tmpdir, jsonfile): - # We can do this if an *.stf file is present - basename = os.path.basename(options.p4filename) - base, _ = os.path.splitext(basename) - dirname = os.path.dirname(options.p4filename) - - testFile = options.testFile - # If no test file is provided, try to find it in the folder. - if not testFile: - testFile = dirname + "/" + base + ".stf" - print("Check for ", testFile) - if not os.path.isfile(testFile): - # If no stf file is present just use the empty file - testFile = dirname + "/empty.stf" - if not os.path.isfile(testFile): - # If no empty.stf present, don't try to run the model at all - return SUCCESS + return "_errors" in str(p4_file) + + +def run_model(options: Options, tmpdir: Path, jsonfile: Path) -> int: + if not options.test_file: + return testutils.SUCCESS bmv2 = RunBMV2(tmpdir, options, jsonfile) - stf_map, result = bmv2.parse_stf_file(testFile) - if result != SUCCESS: + stf_map, result = bmv2.parse_stf_file(options.test_file) + if result != testutils.SUCCESS: return result result = bmv2.generate_model_inputs(stf_map) - if result != SUCCESS: + if result != testutils.SUCCESS: return result if not options.hasBMv2: - reportError("config.h indicates that BMv2 is not installed. Will skip running BMv2 tests") - return SUCCESS + testutils.log.error( + "config.h indicates that BMv2 is not installed. Will skip running BMv2 tests" + ) + return testutils.SUCCESS result = bmv2.run(stf_map) - if result != SUCCESS: + if result != testutils.SUCCESS: return result result = bmv2.checkOutputs() return result -def run_init_commands(options): - if not options.initCommands: - return SUCCESS - for cmd in options.initCommands: - args = cmd.split() - result = run_timeout(options, args, timeout, None) - if result != SUCCESS: - return FAILURE - return SUCCESS +def run_init_commands(options: Options) -> int: + if not options.init_commands: + return testutils.SUCCESS + for cmd in options.init_commands: + result = testutils.exec_process(cmd, timeout=30) + if result != testutils.SUCCESS: + return testutils.FAILURE + return testutils.SUCCESS -def process_file(options, argv): +def process_file(options: Options) -> int: assert isinstance(options, Options) - if run_init_commands(options) != SUCCESS: - return FAILURE - # ensure that tempfile.mkdtemp returns an absolute path, regardless of the py3 version - tmpdir = tempfile.mkdtemp(dir=Path(".").absolute()) - basename = os.path.basename(options.p4filename) - base, _ = os.path.splitext(basename) + if run_init_commands(options) != testutils.SUCCESS: + return testutils.FAILURE + # Ensure that tempfile.mkdtemp returns an absolute path, regardless of the py3 version. + tmpdir = Path(tempfile.mkdtemp(dir=options.testdir.absolute())) + base = options.p4_file.stem - if options.verbose: - print("Writing temporary files into ", tmpdir) + testutils.log.debug("Writing temporary files into ", tmpdir) if options.testName: - jsonfile = options.testName + ".json" + jsonfile = Path(options.testName).with_suffix(".json") else: - jsonfile = tmpdir + "/" + base + ".json" - stderr = tmpdir + "/" + basename + "-stderr" + jsonfile = tmpdir.joinpath(base).with_suffix(".json") - if not os.path.isfile(options.p4filename): - raise Exception("No such file " + options.p4filename) + if not options.p4_file.is_file(): + raise Exception(f"No such file {options.p4_file}") if options.usePsa: - binary = options.compilerBuildDir + "/p4c-bm2-psa" + binary = options.compiler_build_dir.joinpath("p4c-bm2-psa") else: - binary = options.compilerBuildDir + "/p4c-bm2-ss" - - args = [binary, "-o", jsonfile] + options.compilerOptions - if "p4_14" in options.p4filename or "v1_samples" in options.p4filename: - args.extend(["--std", "p4-14"]) - args.append(options.p4filename) - args.extend(argv) + binary = options.compiler_build_dir.joinpath("p4c-bm2-ss") + + cmd: str = f"{binary} -o {jsonfile}" + for opt in options.compiler_options: + cmd += f" {opt}" + for opt in options.switch_target_specific_options: + cmd += f" {opt}" + if "p4_14" in str(options.p4_file) or "v1_samples" in str(options.p4_file): + cmd += " --std p4-14" + cmd += f" {options.p4_file}" if options.runDebugger: - args[0:0] = options.runDebugger.split() - os.execvp(args[0], args) - result = run_timeout(options, args, timeout, stderr) - - if result != SUCCESS: - print("Error compiling") - with open(stderr, mode="r", encoding="utf-8") as stderr_file: - print(stderr_file.read()) - # If the compiler crashed fail the test - if "Compiler Bug" in stderr_file.read(): - return FAILURE - - expected_error = isError(options.p4filename) + # TODO: Is this still working? + cmd = options.runDebugger + cmd + os.execvp(cmd[0], cmd) + result = testutils.exec_process(cmd, timeout=30) + returnvalue = result.returncode + if returnvalue != testutils.SUCCESS: + testutils.log.info("Error compiling") + # If the compiler crashed fail the test + if "Compiler Bug" in result.output: + return testutils.FAILURE + + expected_error = isError(options.p4_file) if expected_error: # invert result - if result == SUCCESS: - result = FAILURE + if returnvalue == testutils.SUCCESS: + returnvalue = testutils.FAILURE else: - result = SUCCESS + returnvalue = testutils.SUCCESS - if result == SUCCESS and not expected_error: - result = run_model(options, tmpdir, jsonfile) + if returnvalue == testutils.SUCCESS and not expected_error: + return run_model(options, tmpdir, jsonfile) - if options.cleanupTmp: - if options.verbose: - print("Removing", tmpdir) - shutil.rmtree(tmpdir) - return result + return returnvalue -def main(argv): +def run_test(options: Options) -> int: try: - result = process_file(options, argv) + result = process_file(options) except Exception as e: - print(f"There was a problem executing the STF test {options.testFile}:") - raise e + testutils.log.error(f"There was a problem executing the STF test {options.test_file}:") + testutils.log.error(traceback.format_exc()) + return testutils.FAILURE - if result != SUCCESS: - reportError("Test failed") - sys.exit(result) + if result != testutils.SUCCESS: + testutils.log.error("Test failed") + return result -if __name__ == "__main__": - # Parse options and process argv - args, argv = parse_args() +def create_options(test_args: Any) -> Optional[Options]: + """Parse the input arguments and create a processed options object.""" options = Options() - options.binary = "" - # TODO: Convert these paths to pathlib's Path. - options.p4filename = check_if_file(args.p4filename).as_posix() - options.compilerSrcDir = check_if_dir(args.rootdir).as_posix() + testdir = test_args.testdir + if not testdir: + print("No test directory provided. Generating temporary folder.") + testdir = tempfile.mkdtemp(dir=Path(".").absolute()) + # Generous permissions because the program is usually edited by sudo. + os.chmod(testdir, 0o755) + options.testdir = Path(testdir) + # Configure logging. + logging.basicConfig( + filename=options.testdir.joinpath("test.log"), + format="%(levelname)s: %(message)s", + level=getattr(logging, test_args.log_level), + filemode="w", + ) + stderr_log = logging.StreamHandler() + stderr_log.setFormatter(logging.Formatter("%(levelname)s: %(message)s")) + logging.getLogger().addHandler(stderr_log) + + result = testutils.check_if_file(test_args.p4_file) + if not result: + return None + options.p4_file = result + test_file = test_args.test_file + if not test_file: + testutils.log.info("No test file provided. Checking for file in folder.") + test_file = options.p4_file.with_suffix(".stf") + result = testutils.check_if_file(test_file) + if testutils.check_if_file(test_file): + testutils.log.info(f"Using test file {test_file}") + options.test_file = Path(test_file) + else: + options.test_file = None + options.rootdir = Path(test_args.rootdir) # If no build directory is provided, use current working directory - if args.builddir: - options.compilerBuildDir = args.builddir + if test_args.builddir: + options.compiler_build_dir = test_args.builddir else: - options.compilerBuildDir = "." - options.verbose = args.verbose - options.replace = args.replace - options.cleanupTmp = args.nocleanup - options.testFile = args.testFile + options.compiler_build_dir = Path.cwd() + options.replace = test_args.replace + options.cleanupTmp = test_args.nocleanup # We have to be careful here, passing compiler options is ambiguous in # argparse because the parser is not positional. # https://stackoverflow.com/a/21894384/3215972 # For each list option, such as compiler_options, # we use -a="--compiler-arg" instead. - for compiler_option in args.compiler_options: - options.compilerOptions.extend(compiler_option.split()) - for switch_option in args.switch_options: - options.switchTargetSpecificOptions.extend(switch_option.split()) - for init_cmd in args.init_cmds: - options.initCommands.append(init_cmd) - options.usePsa = args.use_psa - if args.pp: - options.compilerOptions.append(args.pp) - if args.gdb: + for compiler_option in test_args.compiler_options: + options.compiler_options.extend(compiler_option.split()) + for switch_option in test_args.switch_options: + options.switch_target_specific_options.extend(switch_option.split()) + for init_cmd in test_args.init_cmds: + options.init_commands.append(init_cmd) + options.usePsa = test_args.use_psa + if test_args.pp: + options.compiler_options.append(test_args.pp) + if test_args.gdb: options.runDebugger = "gdb --args" - if args.lldb: + if test_args.lldb: options.runDebugger = "lldb --" - options.observationLog = args.obs_log - residual_argv = [] - for arg in argv: - if arg in ("-D", "-I", "-T"): - options.compilerOptions.append(arg) - else: - residual_argv.append(arg) - - config = ConfigH(options.compilerBuildDir + "/config.h") + options.observationLog = test_args.obs_log + config = ConfigH(options.compiler_build_dir.joinpath("config.h")) if not config.ok: - print("Error parsing config.h") - sys.exit(FAILURE) - - # Configure logging. - logging.basicConfig( - filename="test.log", - format="%(levelname)s: %(message)s", - level=getattr(logging, "INFO"), - filemode="w", - ) - stderr_log = logging.StreamHandler() - stderr_log.setFormatter(logging.Formatter("%(levelname)s: %(message)s")) - logging.getLogger().addHandler(stderr_log) + testutils.log.info("Error parsing config.h") + return None options.hasBMv2 = "HAVE_SIMPLE_SWITCH" in config.vars - if options.p4filename.startswith(options.compilerBuildDir): - options.testName = options.p4filename[len(options.compilerBuildDir) :] + if options.compiler_build_dir in options.p4_file.parents: + options.testName = str(options.p4_file.relative_to(options.compiler_build_dir)) if options.testName.startswith("/"): options.testName = options.testName[1:] if options.testName.endswith(".p4"): @@ -409,12 +401,23 @@ def main(argv): options.testName = "bmv2/" + options.testName if not options.observationLog: if options.testName: - options.observationLog = os.path.join(f"{options.testName}.p4.obs") + options.observationLog = Path(options.testName).with_suffix(".p4.obs") else: - basename = os.path.basename(options.p4filename) + basename = os.path.basename(options.p4_file) base, _ = os.path.splitext(basename) - dirname = os.path.dirname(options.p4filename) - options.observationLog = os.path.join(dirname, f"{base}.p4.obs") + dirname = options.p4_file.parent + options.observationLog = dirname.joinpath(f"{base}.p4.obs") + return options - # All args after '--' are intended for the p4 compiler - main(residual_argv) + +if __name__ == "__main__": + test_options = create_options(ARGS) + if not test_options: + sys.exit(testutils.FAILURE) + + # Run the test with the extracted options + test_result = run_test(test_options) + if not (ARGS.nocleanup or test_result != testutils.SUCCESS): + testutils.log.info("Removing temporary test directory.") + testutils.del_dir(test_options.testdir) + sys.exit(test_result) diff --git a/tools/testutils.py b/tools/testutils.py index 9946544853..82d11f5d60 100644 --- a/tools/testutils.py +++ b/tools/testutils.py @@ -25,7 +25,7 @@ import subprocess import threading from pathlib import Path -from typing import Any, List, NamedTuple, Optional, Union +from typing import Any, Dict, List, NamedTuple, Optional, Union import scapy.packet