Clean up BMv2's run-stf-test script and integrate it with testutils (#4981)

* Modernize run-stf-test script.

Signed-off-by: fruffy <[email protected]>

* Add ruff support and fix linters.

Signed-off-by: fruffy <[email protected]>

---------

Signed-off-by: fruffy <[email protected]>

Author: fruffy

backends/bmv2/CMakeLists.txt

@@ -253,4 +253,4 @@ set (GTEST_BMV2_SOURCES
   gtest/load_ir_from_json.cpp
 )
 set (GTEST_SOURCES ${GTEST_SOURCES} ${GTEST_BMV2_SOURCES} PARENT_SCOPE)
-set (GTEST_LDADD ${GTEST_LDADD} bmv2backend PARENT_SCOPE)
+set (GTEST_LDADD ${GTEST_LDADD} bmv2backend PARENT_SCOPE)

backends/bmv2/bmv2stf.py

@@ -27,6 +27,7 @@
 from collections import OrderedDict
 from glob import glob
 from pathlib import Path
+from typing import Dict, List, Optional, Tuple
 
 try:
     import scapy.utils as scapy_util
@@ -35,10 +36,10 @@
     pass
 
 FILE_DIR = Path(__file__).resolve().parent
-# Append tools to the import path.
-sys.path.append(str(FILE_DIR.joinpath("../../tools")))
-import testutils
-from stf.stf_parser import STFLexer, STFParser
+# Append the root to the import path.
+sys.path.append(str(FILE_DIR.joinpath("../..")))
+import tools.testutils as testutils
+from tools.stf.stf_parser import STFLexer, STFParser
 
 
 class TimeoutException(Exception):
@@ -52,13 +53,28 @@ def signal_handler(signum, frame):
 signal.signal(signal.SIGALRM, signal_handler)
 
 
-class Options(object):
-    def __init__(self):
-        self.binary = None
-        self.verbose = False
-        self.preserveTmp = False
-        self.observationLog = None
-        self.usePsa = False
+class Options:
+    # 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_options: List[str] = []
+    switch_target_options: List[str] = []
+    has_bmv2: bool = False  # Is the behavioral model installed?
+    use_psa: bool = False  # Use the psa switch behavioral model?
+    run_debugger: str = ""
+    # Log packets produced by the BMV2 model if path to log is supplied
+    observation_log: Optional[Path] = None
+    init_commands: List[str] = []
 
 
 def ByteToHex(byteStr):
@@ -488,8 +504,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: Options, jsonfile: Path) -> None:
+        self.clifile = folder.joinpath("cli.txt")
         self.jsonfile = jsonfile
         self.stffile = None
         self.folder = folder
@@ -503,10 +519,8 @@ def __init__(self, folder, options, jsonfile):
         self.actions = []
         self.switchLogFile = "switch.log"  # .txt is added by BMv2
         self.readJson()
-        self.cmd_line_args = getattr(options, "switchOptions", ())
-        self.target_specific_cmd_line_args = getattr(options, "switchTargetSpecificOptions", ())
 
-    def readJson(self):
+    def readJson(self) -> None:
         with open(self.jsonfile) as jf:
             self.json = json.load(jf)
         for a in self.json["actions"]:
@@ -516,10 +530,10 @@ 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):
+    def interface_of_filename(self, f) -> int:
         return int(os.path.basename(f).rstrip(".pcap").lstrip(self.pcapPrefix).rsplit("_", 1)[0])
 
     def do_cli_command(self, cmd):
@@ -529,7 +543,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 == "":
@@ -679,14 +693,14 @@ def tableByName(self, tableName):
             return candidate
         raise Exception("Could not find table " + tableName)
 
-    def interfaceArgs(self):
+    def interfaceArgs(self) -> List[str]:
         # return list of interface names suitable for bmv2
         result = []
         for interface in sorted(self.interfaces):
             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: Dict[str, str]) -> int:
         for entry in stf_map:
             cmd = entry[0]
             if cmd in ["packet", "expect"]:
@@ -711,11 +725,11 @@ def check_switch_server_ready(self, proc, thriftPort):
             if result == 0:
                 return True
 
-    def run(self, stf_map):
+    def run(self, stf_map: Dict[str, str]) -> int:
         testutils.log.info("Running model")
         wait = 0  # Time to wait before model starts running
 
-        if self.options.usePsa:
+        if self.options.use_psa:
             switch = "psa_switch"
             switch_cli = "psa_switch_CLI"
         else:
@@ -750,17 +764,15 @@ def run(self, stf_map):
                 + self.interfaceArgs()
                 + [self.jsonfile]
             )
-            if self.cmd_line_args:
-                runswitch += self.cmd_line_args
-            if self.target_specific_cmd_line_args:
+            runswitch += self.options.switch_options
+            if self.options.switch_target_options:
                 runswitch += [
                     "--",
-                ] + self.target_specific_cmd_line_args
-                testutils.log.info("Running %s", " ".join(runswitch))
+                ] + self.options.switch_target_options
             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 +857,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 +921,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)