Test: Implement packet capture fixture using netsniff-ng tool. (#1283)

Remove tcpdump implementation as tcpdump does not provide correct
results for compliance checks.
Add EBU list pcap upload and compliance check support fixture. 
Update results by compliance result.
Add test config options for packet capture and ebu list server info.
Implement capturing packets in RxTxApp engine. 
Update test cases for RxTxApp with compliance check (with multicast
destination address only) - ST20 and ST30 test cases

Author: andremiszcz

tests/validation/compliance/compliance_client.py

@@ -0,0 +1,138 @@
+import logging
+import os
+import time
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+
+class PcapComplianceClient:
+    def __init__(
+        self,
+        ebu_ip,
+        user,
+        password,
+        pcap_file=None,
+        pcap_id=None,
+        proxies={"http": "", "https": "", "ftp": ""},
+    ):
+        """
+        Initialize the client.
+        """
+        self.ebu_ip = ebu_ip
+        self.user = user
+        self.password = password
+        self.pcap_file = pcap_file
+        self.proxies = proxies
+        self.pcap_id = pcap_id
+        self.token = None
+        self.session = requests.Session()
+        self.session.trust_env = False  # Do not use system proxy settings
+        self.authenticate()
+
+    def authenticate(self):
+        """
+        Authenticate with the EBU server and store the access token.
+        """
+        url = f"http://{self.ebu_ip}/auth/login"
+        headers = {"Content-Type": "application/json"}
+        data = {"username": self.user, "password": self.password}
+        response = self.session.post(
+            url, headers=headers, json=data, verify=False, proxies=self.proxies
+        )
+        response.raise_for_status()
+        self.token = response.json().get("content", {}).get("token")
+        if not self.token:
+            raise Exception("Authentication failed: No token received.")
+
+    def upload_pcap(self):
+        """
+        Upload the PCAP file to the EBU server and store the returned UUID.
+        Returns the UUID of the uploaded PCAP.
+        """
+        url = f"http://{self.ebu_ip}/api/pcap"
+        headers = {"Authorization": f"Bearer {self.token}"}
+        if self.pcap_file:
+            with open(self.pcap_file, "rb") as f:
+                files = {
+                    "pcap": (
+                        os.path.basename(self.pcap_file),
+                        f,
+                        "application/vnd.tcpdump.pcap",
+                    )
+                }
+                response = self.session.put(
+                    url,
+                    headers=headers,
+                    files=files,
+                    verify=False,
+                    proxies=self.proxies,
+                )
+            response.raise_for_status()
+            self.pcap_id = response.json().get("uuid")
+        if not self.pcap_id:
+            raise Exception("Upload failed: No UUID received.")
+        return self.pcap_id
+
+    def download_report(self, retries=3):
+        """
+        Download the compliance report for the uploaded PCAP file.
+        returns the report as a JSON object.
+        """
+        url = f"http://{self.ebu_ip}/api/pcap/{self.pcap_id}/report?type=json"
+        headers = {"Authorization": f"Bearer {self.token}"}
+        while retries > 0:
+            response = self.session.get(
+                url, headers=headers, verify=False, proxies=self.proxies
+            )
+            response.raise_for_status()
+            report = response.json()
+            if report.get("analyzed", False):
+                return report
+            time.sleep(1)
+            retries -= 1
+        logger.error(
+            f"Report is not ready after {retries} seconds, skipping compliance check"
+        )
+        return False
+
+    def check_compliance(self, report=None):
+        """
+        Check the compliance result from the downloaded report.
+        Returns True if compliant, False otherwise.
+        """
+        if report is None:
+            report = self.download_report()
+        is_compliant = report.get("not_compliant_streams", 1) == 0
+        if is_compliant:
+            is_compliant = not any(
+                [
+                    1 if stream.get("media_type") == "unknown" else 0
+                    for stream in report.get("streams", [])
+                ]
+            )
+        return is_compliant, report
+
+    def delete_pcap(self, pcap_id=None):
+        """
+        Delete the PCAP file and its report from the EBU server.
+        If pcap_id is not provided, uses self.pcap_id.
+        """
+        if pcap_id is None:
+            pcap_id = self.pcap_id
+        if not pcap_id:
+            raise ValueError("No PCAP ID provided for deletion.")
+        url = f"http://{self.ebu_ip}/api/pcap/{pcap_id}"
+        headers = {"Authorization": f"Bearer {self.token}"}
+        response = self.session.delete(
+            url, headers=headers, verify=False, proxies=self.proxies
+        )
+        if response.status_code == 200:
+            logger.info(f"PCAP {pcap_id} deleted successfully from EBU server.")
+            return True
+        else:
+            logger.error(
+                f"Failed to delete PCAP {pcap_id}: {response.status_code} {response.text}"
+            )
+            return False

tests/validation/compliance/upload_pcap.py

@@ -0,0 +1,69 @@
+import argparse
+import os
+
+from compliance_client import PcapComplianceClient
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description="Upload a PCAP file to the EBU LIST server."
+    )
+    parser.add_argument(
+        "--pcap",
+        type=str,
+        required=True,
+        help="Path to the PCAP file to upload.",
+    )
+    parser.add_argument(
+        "--ip",
+        type=str,
+        required=True,
+        help="EBU LIST server IP address.",
+    )
+    parser.add_argument(
+        "--user",
+        type=str,
+        required=True,
+        help="Username for EBU LIST service.",
+    )
+    parser.add_argument(
+        "--password",
+        type=str,
+        required=True,
+        help="Password for EBU LIST service.",
+    )
+    parser.add_argument(
+        "--proxy",
+        type=str,
+        required=False,
+        help="Proxy for uploading to EBU LIST service.",
+    )
+    return parser.parse_args()
+
+
+def upload_pcap(file_path, ip, login, password, proxies):
+    # Check for login and password
+    if not ip or not login or not password:
+        raise Exception("IP address, login and password are required.")
+
+    # Check if the file exists before proceeding
+    if not os.path.isfile(file_path):
+        raise Exception(f"File not found: {file_path}")
+
+    # Create the uploader object and upload the PCAP file
+    uploader = PcapComplianceClient(
+        ebu_ip=ip, user=login, password=password, pcap_file=file_path, proxies=proxies
+    )
+    uuid = uploader.upload_pcap()  # Upload the PCAP file and get the UUID
+    return uuid
+
+
+if __name__ == "__main__":
+    args = parse_args()
+    # TODO: Handle proxies better, so each type can have a proper value
+    proxies = None
+    if args.proxy is not None:
+        proxies = {"http": args.proxy, "https": args.proxy, "ftp": args.proxy}
+    uuid = upload_pcap(args.pcap, args.ip, args.user, args.password, proxies)
+    # Print extractable UUID
+    print(f">>>UUID: {uuid}")

tests/validation/configs/test_config.yaml

@@ -1,16 +1,19 @@
 build: MTL_PATH_PLACEHOLDER
 mtl_path: MTL_PATH_PLACEHOLDER
 media_path: /mnt/media
-capture_cfg:
-  enable: false
-  test_name: test_name
-  pcap_dir: /mnt/ramdisk/pcap
-  capture_time: 5
-  interface: null
 ramdisk:
   media: 
     mountpoint: /mnt/ramdisk/media
     size_gib: 32
   pcap:
     mountpoint: /mnt/ramdisk/pcap
     size_gib: 768
+compliance: false
+capture_cfg:
+  enable: false
+  pcap_dir: /mnt/ramdisk/pcap
+ebu_server:
+  ebu_ip: EBU_IP_PLACEHOLDER
+  user: EBU_LOGIN_PLACEHOLDER
+  password: EBU_PASS_PLACEHOLDER
+  proxy: false

tests/validation/conftest.py

@@ -12,15 +12,19 @@
 import pytest
 from common.mtl_manager.mtlManager import MtlManager
 from common.nicctl import Nicctl
+from compliance.compliance_client import PcapComplianceClient
+from create_pcap_file.netsniff import NetsniffRecorder, calculate_packets_per_frame
 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 LOG_FOLDER, TESTCMD_LVL
+from mtl_engine.const import FRAMES_CAPTURE, LOG_FOLDER, TESTCMD_LVL
 from mtl_engine.csv_report import (
     csv_add_test,
     csv_write_report,
+    get_compliance_result,
     update_compliance_result,
 )
+from mtl_engine.execute import log_fail
 from mtl_engine.ramdisk import Ramdisk
 from mtl_engine.stash import (
     clear_issue,
@@ -249,23 +253,75 @@ def log_session():
     csv_write_report(f"{LOG_FOLDER}/latest/report.csv")
 
 
-@pytest.fixture(scope="session", autouse=True)
-def compliance_report(request, log_session, test_config):
-    """
-    This function is used for compliance check and report.
-    """
-    # TODO: Implement compliance check logic. When tcpdump pcap is enabled, at the end of the test session all pcaps
-    # shall be send into EBU list.
-    # Pcaps shall be stored in the ramdisk, and then moved to the compliance
-    # folder or send into EBU list after each test finished and remove it from the ramdisk.
-    # Compliance report generation logic goes here after yield. Or in another class / function but triggered here.
-    # AFAIK names of pcaps contains test name so it can be matched with result of each test like in code below.
-    yield
-    if test_config.get("compliance", False):
-        logging.info("Compliance mode enabled, updating compliance results")
-        for item in request.session.items:
-            test_case = item.nodeid
-            update_compliance_result(test_case, "Fail")
+@pytest.fixture(scope="function")
+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)
+            )
+            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[
+                "frames_number"
+            ] * calculate_packets_per_frame(media_file_info)
+            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),
+        )
+    yield 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")
+            return
+        ebu_ip = ebu_server.get("ebu_ip", None)
+        ebu_login = ebu_server.get("user", None)
+        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 = capturer.host.connection.execute_command(
+            "python3 ./tests/validation/compliance/upload_pcap.py"
+            f" --ip {ebu_ip}"
+            f" --user {ebu_login}"
+            f" --password {ebu_passwd}"
+            f" --pcap {capturer.pcap_file}{proxy_cmd}",
+            cwd=f"{str(mtl_path)}",
+        )
+        if compliance_upl.return_code != 0:
+            logger.error(f"PCAP upload failed: {compliance_upl.stderr}")
+            return
+        uuid = compliance_upl.stdout.split(">>>UUID: ")[1].strip()
+        logger.debug(f"PCAP successfully uploaded to EBU LIST with UUID: {uuid}")
+        uploader = PcapComplianceClient(
+            ebu_ip=ebu_ip,
+            user=ebu_login,
+            password=ebu_passwd,
+            pcap_id=uuid,
+            proxies={"http": ebu_proxy, "https": ebu_proxy},
+        )
+        result, report = uploader.check_compliance()
+        update_compliance_result(request.node.nodeid, "Pass" if result else "Fail")
+        if result:
+            logger.info("PCAP compliance check passed")
+        else:
+            log_fail("PCAP compliance check failed")
+            logger.info(f"Compliance report: {report}")
 
 
 @pytest.fixture(scope="function", autouse=True)
@@ -287,20 +343,26 @@ def log_case(request, caplog: pytest.LogCaptureFixture):
     clear_issue()
     yield
     report = request.node.stash[phase_report_key]
-    if report["setup"].failed:
-        logging.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:
-        logging.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:
-        logging.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:
-        logging.log(level=TEST_INFO, msg=f"Test skipped for {case_id}")
+        logger.log(level=TEST_INFO, msg=f"Test skipped for {case_id}")
         result = "Skip"
 
     logger.removeHandler(fh)