Adjust ST20 and ST30 tests with netsniff

Author: andremiszcz

tests/validation/conftest.py

@@ -17,10 +17,16 @@
 from mfd_common_libs.custom_logger import add_logging_level
 from mfd_common_libs.log_levels import TEST_FAIL, TEST_INFO, TEST_PASS
 from mfd_connect.exceptions import ConnectionCalledProcessError
-from mtl_engine.const import DOWNLOAD_REPORT_TRIES, FRAMES_CAPTURE, LOG_FOLDER, TESTCMD_LVL
+from mtl_engine.const import (
+    DOWNLOAD_REPORT_TRIES,
+    FRAMES_CAPTURE,
+    LOG_FOLDER,
+    TESTCMD_LVL,
+)
 from mtl_engine.csv_report import (
     csv_add_test,
     csv_write_report,
+    get_compliance_result,
     update_compliance_result,
 )
 
@@ -39,7 +45,6 @@
 )
 from pytest_mfd_logging.amber_log_formatter import AmberLogFormatter
 
-
 logger = logging.getLogger(__name__)
 phase_report_key = pytest.StashKey[Dict[str, pytest.CollectReport]]()
 
@@ -280,37 +285,38 @@ def log_session():
 
 
 @pytest.fixture(scope="function")
-def pcap_capture(request, media_file,   test_config, hosts, mtl_path
-):
+def pcap_capture(request, media_file, test_config, hosts, mtl_path):
     capture_cfg = test_config.get("capture_cfg", {})
     capturer = None
     if capture_cfg and capture_cfg.get("enable"):
         host = hosts["client"] if "client" in hosts else list(hosts.values())[0]
         media_file_info, _ = media_file
         test_name = request.node.name
         if "frames_number" not in capture_cfg and "capture_time" not in capture_cfg:
-            capture_cfg["packets_number"] = FRAMES_CAPTURE * calculate_packets_per_frame(
-                media_file_info["width"],
-                media_file_info["height"]
+            capture_cfg["packets_number"] = (
+                FRAMES_CAPTURE * calculate_packets_per_frame(media_file_info)
+            )
+            logger.info(
+                f"Capture {capture_cfg['packets_number']} packets for {FRAMES_CAPTURE} frames"
             )
-            logger.info(f"Capture {capture_cfg['packets_number']} packets for {FRAMES_CAPTURE} frames")
         elif "frames_number" in capture_cfg:
-            capture_cfg["packets_number"] = capture_cfg.pop("frames_number") * calculate_packets_per_frame(
-                media_file_info["width"],
-                media_file_info["height"]
+            capture_cfg["packets_number"] = capture_cfg[
+                "frames_number"
+            ] * calculate_packets_per_frame(media_file_info)
+            logger.info(
+                f"Capture {capture_cfg['packets_number']} packets for {capture_cfg['frames_number']} frames"
             )
-            logger.info(f"Capture {capture_cfg['packets_number']} packets for {capture_cfg['frames_number']} frames")
         capturer = NetsniffRecorder(
             host=host,
             test_name=test_name,
             pcap_dir=capture_cfg.get("pcap_dir", "/tmp"),
             interface=host.network_interfaces[0].name,
             silent=capture_cfg.get("silent", True),
             packets_capture=capture_cfg.get("packets_number", None),
-            capture_time=capture_cfg.get("capture_time", None)
+            capture_time=capture_cfg.get("capture_time", None),
         )
     yield capturer
-    if capturer:
+    if capturer and capturer.netsniff_process:
         ebu_server = test_config.get("ebu_server", {})
         if not ebu_server:
             logger.error("EBU server configuration not found in test_config.yaml")
@@ -320,7 +326,7 @@ def pcap_capture(request, media_file,   test_config, hosts, mtl_path
         ebu_passwd = ebu_server.get("password", None)
         ebu_proxy = ebu_server.get("proxy", None)
         proxy_cmd = f" --proxy {ebu_proxy}" if ebu_proxy else ""
-        compliance_upl = host.connection.execute_command(
+        compliance_upl = capturer.host.connection.execute_command(
             "python3 ./tests/validation/compliance/upload_pcap.py"
             f" --ip {ebu_ip}"
             f" --login {ebu_login}"
@@ -332,13 +338,9 @@ def pcap_capture(request, media_file,   test_config, hosts, mtl_path
             logger.error(f"PCAP upload failed: {compliance_upl.stderr}")
             return
         uuid = (
-            compliance_upl.stdout.split(">>>UUID>>>")[1]
-            .split("<<<UUID<<<")[0]
-            .strip()
-        )
-        logger.debug(
-            f"PCAP successfully uploaded to EBU LIST with UUID: {uuid}"
+            compliance_upl.stdout.split(">>>UUID>>>")[1].split("<<<UUID<<<")[0].strip()
         )
+        logger.debug(f"PCAP successfully uploaded to EBU LIST with UUID: {uuid}")
         uploader = PcapComplianceClient(config_path="")
         uploader.ebu_ip = ebu_ip
         uploader.user = ebu_login
@@ -353,9 +355,13 @@ def pcap_capture(request, media_file,   test_config, hosts, mtl_path
             tries -= 1
             report = uploader.download_report()
         if not report.get("analyzed", False):
-            logger.error(f"Report is not ready after {DOWNLOAD_REPORT_TRIES} seconds, skipping compliance check")
+            logger.error(
+                f"Report is not ready after {DOWNLOAD_REPORT_TRIES} seconds, skipping compliance check"
+            )
             return
-        if report.get("not_compliant_streams", 1) == 0:
+        result = report.get("not_compliant_streams", 1)
+        update_compliance_result(request.node.nodeid, "Pass" if result == 0 else "Fail")
+        if result == 0:
             logger.info("PCAP compliance check passed")
         else:
             log_fail("PCAP compliance check failed")
@@ -380,18 +386,24 @@ def log_case(request, caplog: pytest.LogCaptureFixture):
     clear_issue()
     yield
     report = request.node.stash[phase_report_key]
-    if report["setup"].failed:
-        logger.log(level=TEST_FAIL, msg=f"Setup failed for {case_id}")
+
+    def fail_test(stage):
+        logger.log(level=TEST_FAIL, msg=f"{stage} failed for {case_id}")
         os.chmod(logfile, 0o4755)
-        result = "Fail"
+        return "Fail"
+
+    if report["setup"].failed:
+        result = fail_test("Setup")
     elif ("call" not in report) or report["call"].failed:
-        logger.log(level=TEST_FAIL, msg=f"Test failed for {case_id}")
-        os.chmod(logfile, 0o4755)
-        result = "Fail"
+        result = fail_test("Test")
     elif report["call"].passed:
-        logger.log(level=TEST_PASS, msg=f"Test passed for {case_id}")
-        os.chmod(logfile, 0o755)
-        result = "Pass"
+        compliance = get_compliance_result(case_id)
+        if compliance is not None and compliance == "Fail":
+            result = fail_test("Compliance")
+        else:
+            logger.log(level=TEST_PASS, msg=f"Test passed for {case_id}")
+            os.chmod(logfile, 0o755)
+            result = "Pass"
     else:
         logger.log(level=TEST_INFO, msg=f"Test skipped for {case_id}")
         result = "Skip"

tests/validation/create_pcap_file/netsniff.py

@@ -2,24 +2,31 @@
 # Copyright 2025 Intel Corporation
 import logging
 import os
-from datetime import datetime
 from time import sleep
-from wsgiref import headers
 
-from mfd_connect.exceptions import ConnectionCalledProcessError, RemoteProcessInvalidState, RemoteProcessTimeoutExpired, SSHRemoteProcessEndException
+from mfd_connect.exceptions import (
+    ConnectionCalledProcessError,
+    RemoteProcessInvalidState,
+    RemoteProcessTimeoutExpired,
+    SSHRemoteProcessEndException,
+)
 
 logger = logging.getLogger(__name__)
-
 STARTUP_WAIT = 2  # Default wait time after starting the process
 
 
-def calculate_packets_per_frame(width: int, height: int, mtu: int = 1500) -> int:
+def calculate_packets_per_frame(media_file_info, mtu: int = 1500) -> int:
     # Simplified calculation for the number of packets per frame
-    # Supported only 4:2:2 format
-    pgroupsize = 5
-    pgroupcoverage = 2
-    headersize = 74  # Ethernet + IP + UDP + RTP headers
-    packets = 1 + int((width * height) / (int((mtu - headersize) / pgroupsize) * pgroupcoverage))
+    # Supported only 4:2:2 format or audio formats assuming 1 packet per 1ms
+    packets = 1000
+    if "width" in media_file_info and "height" in media_file_info:
+        pgroupsize = 5
+        pgroupcoverage = 2
+        headersize = 74  # Ethernet + IP + UDP + RTP headers
+        packets = 1 + int(
+            (media_file_info["width"] * media_file_info["height"])
+            / (int((mtu - headersize) / pgroupsize) * pgroupcoverage)
+        )
     return packets
 
 
@@ -77,7 +84,11 @@ def start(self):
                     str(self.interface),
                     "--out",
                     self.pcap_file,
-                    f"--num {self.packets_capture}" if self.packets_capture is not None else "",
+                    (
+                        f"--num {self.packets_capture}"
+                        if self.packets_capture is not None
+                        else ""
+                    ),
                     f'-f "{self.capture_filter}"' if self.capture_filter else "",
                 ]
                 logger.info(f"Running command: {' '.join(cmd)}")
@@ -114,12 +125,17 @@ def capture(self, startup_wait=2):
                 logger.info("Capture complete.")
             else:
                 try:
-                    logger.info(f"Capturing traffic for {self.packets_capture} packets...")
-                    self.netsniff_process.wait(timeout=startup_wait+10)
+                    logger.info(
+                        f"Capturing traffic for {self.packets_capture} packets..."
+                    )
+                    self.netsniff_process.wait(timeout=startup_wait + 10)
                     logger.info("Capture complete.")
                     logger.debug(self.netsniff_process.stdout_text)
                 except RemoteProcessTimeoutExpired:
-                    logger.error("Capture timed out.")
+                    logger.warning(
+                        "Capture timed out. Probably not enough packets were sent. "
+                        "Please adjust packets_capture or capture_time to the test case."
+                    )
                     self.stop()
         else:
             logger.error("netsniff-ng did not start; skipping capture.")
@@ -128,7 +144,7 @@ def stop(self):
         """
         Stops all netsniff-ng processes on the host using pkill.
         """
-        
+
         try:
             logger.info("Stopping netsniff-ng using pkill netsniff-ng...")
             self.netsniff_process.stop(wait=5)
@@ -155,4 +171,4 @@ def update_filter(self, src_ip=None, dst_ip=None):
             self.capture_filter = " and ".join(filters)
         elif len(filters) == 1:
             self.capture_filter = filters[0]
-        logger.info(f"Updated capture filter to: {self.capture_filter}")
+        logger.info(f"Updated capture filter to: {self.capture_filter}")

tests/validation/mtl_engine/RxTxApp.py

@@ -126,6 +126,7 @@ def prepare_netsniff(
 
     returns: NetsniffRecorder instance or None.
     """
+    # TODO: remove me as this is now in udp_app.py but it's not needed anymore
     if (
         capture_cfg
         and capture_cfg.get("enable")
@@ -617,14 +618,14 @@ def execute_test(
         timeout=timeout,
         testcmd=True,
         host=remote_host,
-        background=True
+        background=True,
     )
-    
+
     if netsniff:
         netsniff.update_filter(dst_ip=config["tx_sessions"][0]["dip"][0])
         netsniff.capture()
         logger.info(f"Finished netsniff-ng capture on host {host.name}")
-    
+
     cp.wait(timeout=timeout)
     # Capture stdout output for logging
     capture_stdout(cp, "RxTxApp")
@@ -746,7 +747,7 @@ def execute_perf_test(
     build: str,
     test_time: int,
     fail_on_error: bool = True,
-    capture_cfg=None,
+    netsniff=None,
     host=None,
 ) -> bool:
     # Only initialize logging if it hasn't been initialized already
@@ -774,11 +775,6 @@ def execute_perf_test(
 
     logger.info(f"Performance RxTxApp Command: {command}")
 
-    # Prepare tcpdump recorder if capture is enabled in the test configuration.
-    tcpdump = prepare_tcpdump(capture_cfg, host)
-    netsniff = prepare_netsniff(capture_cfg, host)
-    background = (tcpdump is not None) or (netsniff is not None)
-
     # For 4TX and 8k streams more timeout is needed
     # Also scale timeout with replica count for performance tests
     total_replicas = 0
@@ -816,19 +812,15 @@ def execute_perf_test(
         timeout=timeout,
         testcmd=True,
         host=host,
-        background=background,
+        background=True,
     )
 
-    try:
-        # Start tcpdump capture (blocking, so it captures during traffic)
-        if tcpdump:
-            tcpdump.capture(capture_time=capture_cfg.get("capture_time", 0.5))
-            logger.info("Started performance test tcpdump capture")
-        if netsniff:
-            netsniff.start()
-            logger.info("Started performance test netsniff-ng capture")
-    finally:
-        cp.wait()
+    if netsniff:
+        netsniff.update_filter(dst_ip=config["tx_sessions"][0]["dip"][0])
+        netsniff.capture()
+        logger.info(f"Finished netsniff-ng capture on host {host.name}")
+
+    cp.wait()
 
     # Capture stdout output for logging
     capture_stdout(cp, "RxTxApp Performance")
@@ -1435,7 +1427,7 @@ def execute_dual_test(
     virtio_user: bool = False,
     rx_timing_parser: bool = False,
     ptp: bool = False,
-    capture_cfg=None,
+    netsniff=None,
 ) -> bool:
     case_id = os.environ["PYTEST_CURRENT_TEST"]
     case_id = case_id[: case_id.rfind("(") - 1]
@@ -1510,11 +1502,10 @@ def execute_dual_test(
         host=tx_host,
     )
 
-    # Start tcpdump capture if enabled
-    tcpdump = prepare_tcpdump(capture_cfg, rx_host)
-    if tcpdump:
-        tcpdump.capture(capture_time=capture_cfg.get("capture_time", 0.5))
-        logger.info("Started dual test tcpdump capture")
+    if netsniff:
+        netsniff.update_filter(dst_ip=tx_config["tx_sessions"][0]["dip"][0])
+        netsniff.capture()
+        logger.info(f"Finished netsniff-ng capture on host {netsniff.host.name}")
 
     # Wait for both processes
     tx_cp.wait()
@@ -1524,10 +1515,6 @@ def execute_dual_test(
     capture_stdout(tx_cp, "TX RxTxApp")
     capture_stdout(rx_cp, "RX RxTxApp")
 
-    # Stop tcpdump if it was started
-    if tcpdump:
-        tcpdump.stop()
-
     # Get output from both hosts
     tx_output = tx_cp.stdout_text.splitlines()
     rx_output = rx_cp.stdout_text.splitlines()

tests/validation/mtl_engine/const.py

@@ -4,4 +4,4 @@
 LOG_FOLDER = "logs"
 TESTCMD_LVL = 24  # Custom logging level for test commands
 FRAMES_CAPTURE = 4  # Number of frames to capture for compliance tests
-DOWNLOAD_REPORT_TRIES = 5  # Number of tries to download EBU report
+DOWNLOAD_REPORT_TRIES = 5  # Number of tries to download EBU report

tests/validation/mtl_engine/csv_report.py

@@ -12,13 +12,17 @@ def csv_add_test(
     issue: str,
     result_note: str | None = None,
 ):
-    report[test_case] = {
+    res_dict = {
         "Test case": test_case,
         "Commands": commands,
         "Result": result,
         "Issue": issue,
         "Result note": result_note,
     }
+    if test_case in report:
+        report[test_case].update(res_dict)
+    else:
+        report[test_case] = res_dict
 
 
 def csv_write_report(filename: str):
@@ -29,6 +33,7 @@ def csv_write_report(filename: str):
             "Commands",
             "Status",
             "Result",
+            "Compliance",
             "Issue",
             "Result note",
         ]
@@ -44,3 +49,10 @@ def csv_write_report(filename: str):
 def update_compliance_result(test_case: str, result: str):
     if test_case in report:
         report[test_case]["Compliance"] = result
+    else:
+        report[test_case] = {"Compliance": result}
+
+
+def get_compliance_result(test_case: str) -> str:
+    if test_case in report and "Compliance" in report[test_case]:
+        return report[test_case]["Compliance"]