Add PF support fixture for tests

Author: andremiszcz

tests/validation/common/nicctl.py

@@ -24,18 +24,18 @@ def _nicctl_path(self) -> str:
             return str(self.connection.path(self.mtl_path, "script", self.tool_name))
         return str(self.connection.path(nicctl_path, self.tool_name))
 
-    def _parse_vf_list(self, output: str, all: bool = True) -> list:
+    def _parse_vf_list(self, output: str) -> list:
         if "No VFs found" in output:
             return []
-        vf_info_regex = r"\d{1,3}\s+(.+)\s+vfio" if all else r"(\d{4}:\S+)"
+        vf_info_regex = r"(\d{4}[0-9a-fA-F:.]+)\(?\S*\)?\s+\S*\s*vfio"
         return re.findall(vf_info_regex, output)
 
     def vfio_list(self, pci_addr: str = "all") -> list:
         """Returns list of VFs created on host."""
         resp = self.connection.execute_command(
             f"{self.nicctl} list {pci_addr}", shell=True
         )
-        return self._parse_vf_list(resp.stdout, "all" in pci_addr)
+        return self._parse_vf_list(resp.stdout)
 
     def create_vfs(self, pci_id: str, num_of_vfs: int = 6) -> list:
         """Create VFs on NIC.
@@ -68,3 +68,121 @@ def prepare_vfs_for_test(self, nic: NetworkInterface) -> list:
                 self.create_vfs(nic_pci)
                 self.host.vfs = self.vfio_list(nic_pci)
         return self.host.vfs
+
+    def bind_pmd(self, pci_id: str) -> None:
+        """Bind VF to DPDK PMD driver."""
+        self.connection.execute_command(self.nicctl + " bind_pmd " + pci_id, shell=True)
+
+    def bind_kernel(self, pci_id: str) -> None:
+        """Bind VF to kernel driver."""
+        self.connection.execute_command(
+            self.nicctl + " bind_kernel " + pci_id, shell=True
+        )
+
+
+class InterfaceSetup:
+    def __init__(self, hosts, mtl_path):
+        self.hosts = hosts
+        self.mtl_path = mtl_path
+        self.nicctl_objs = {
+            host.name: Nicctl(mtl_path, host) for host in hosts.values()
+        }
+        self.customs = []
+        self.cleanups = []
+
+    def get_test_interfaces(self, interface_type="VF", count=2, host=None) -> dict:
+        """
+        Creates VFs and binds them into dpdk or bind PF into dpdk.
+
+        :param interface_type: VF - create X VFs on firt available test adapter, PF - prepare list of PFs PCI addresses for test,
+        xxVFxPF - create xx number VFs per each PF. E.G if you need 3 VFs on every PF and you have 2 PF then pass 3VFxPF param with count=6
+        :param count: total number of VFs or PFs needed for test.
+        :param host: You can specify host if you need to test only on this host
+        :return: Returns dictionary with list of PCI addresses of VFs or PFs per host name.
+        """
+        if host:
+            hosts = [host]
+        else:
+            hosts = list(self.hosts.values())
+        selected_interfaces = {k.name: [] for k in hosts}
+        for host in hosts:
+            if getattr(host.topology.extra_info, "custom_interface", None):
+                selected_interfaces[host.name] = [
+                    host.topology.extra_info["custom_interface"]
+                ]
+                self.customs.append(host.name)
+                if len(selected_interfaces[host.name]) < count:
+                    raise Exception(
+                        f"Not enough interfaces for test on host {host.name} in extra_info.custom_interface"
+                    )
+            else:
+                if interface_type == "VF":
+                    vfs = self.nicctl_objs[host.name].create_vfs(
+                        host.network_interfaces[0].pci_address.lspci, count
+                    )
+                    selected_interfaces[host.name] = vfs
+                    self.register_cleanup(
+                        self.nicctl_objs[host.name],
+                        host.network_interfaces[0].pci_address.lspci,
+                        interface_type,
+                    )
+                elif interface_type == "PF":
+                    try:
+                        selected_interfaces[host.name] = []
+                        for i in range(count):
+                            self.nicctl_objs[host.name].bind_pmd(
+                                host.network_interfaces[i].pci_address.lspci
+                            )
+                            selected_interfaces[host.name].append(
+                                str(host.network_interfaces[i])
+                            )
+                            self.register_cleanup(
+                                self.nicctl_objs[host.name],
+                                host.network_interfaces[i].pci_address.lspci,
+                                interface_type,
+                            )
+                    except IndexError:
+                        raise Exception(
+                            f"Not enough interfaces for test on host {host.name} in topology config."
+                        )
+                elif "VFxPF" in interface_type:
+                    vfs_count = interface_type.split("VFxPF")[0]
+                    vfs_count = int(vfs_count) if vfs_count else 1
+                    for i in range(count // vfs_count):
+                        try:
+                            vfs = self.nicctl_objs[host.name].create_vfs(
+                                host.network_interfaces[i].pci_address.lspci, vfs_count
+                            )
+                            selected_interfaces[host.name].extend(vfs)
+                            self.register_cleanup(
+                                self.nicctl_objs[host.name],
+                                host.network_interfaces[i].pci_address.lspci,
+                                "VF",
+                            )
+                        except IndexError:
+                            raise Exception(
+                                f"Not enough interfaces for test on host {host.name} in topology config. "
+                                f"Expected {count // vfs_count} adapters "
+                                f"to be able to create total {count} VFs - {vfs_count} per adapter"
+                            )
+                else:
+                    raise Exception(f"Unknown interface type {interface_type}")
+        return selected_interfaces
+
+    def get_interfaces_list_single(self, interface_type="VF", count=2) -> list:
+        """
+        Wrapper for get_test_interfaces method if you use only single node tests and need only list of interfaces
+        """
+        host = list(self.hosts.values())[0]
+        selected_interfaces = self.get_test_interfaces(interface_type, count, host=host)
+        return selected_interfaces[host.name]
+
+    def register_cleanup(self, nicctl, interface, if_type):
+        self.cleanups.append((nicctl, interface, if_type))
+
+    def cleanup(self):
+        for nicctl, interface, if_type in self.cleanups:
+            if if_type == "VF":
+                nicctl.disable_vf(interface)
+            elif if_type == "PF":
+                nicctl.bind_kernel(interface)

tests/validation/conftest.py

@@ -11,7 +11,7 @@
 
 import pytest
 from common.mtl_manager.mtlManager import MtlManager
-from common.nicctl import Nicctl
+from common.nicctl import InterfaceSetup, 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
@@ -125,6 +125,13 @@ def nic_port_list(hosts: dict, mtl_path) -> None:
         host.vfs = vfs
 
 
+@pytest.fixture(scope="function")
+def setup_interfaces(hosts, test_config, mtl_path):
+    interface_setup = InterfaceSetup(hosts, mtl_path)
+    yield interface_setup
+    interface_setup.cleanup()
+
+
 @pytest.fixture(scope="session")
 def test_time(test_config: dict) -> int:
     test_time = test_config.get("test_time", 30)

tests/validation/mtl_engine/ffmpeg_app.py

@@ -57,6 +57,7 @@ def execute_test(
     test_time: int,
     build: str,
     host,
+    nic_port_list,
     type_: str,
     video_format: str,
     pg_format: str,
@@ -65,13 +66,9 @@ def execute_test(
     multiple_sessions: bool = False,
     tx_is_ffmpeg: bool = True,
 ):
-    # Initialize logging for this test
-    init_test_logging()
-
     case_id = os.environ.get("PYTEST_CURRENT_TEST", "ffmpeg_test")
     case_id = case_id[: case_id.rfind("(") - 1] if "(" in case_id else case_id
 
-    nic_port_list = host.vfs
     video_size, fps = decode_video_format_16_9(video_format)
     match output_format:
         case "yuv":
@@ -234,14 +231,14 @@ def execute_test_rgb24(
     test_time: int,
     build: str,
     host,
+    nic_port_list,
     type_: str,
     video_format: str,
     pg_format: str,
     video_url: str,
 ):
     # Initialize logging for this test
     init_test_logging()
-    nic_port_list = host.vfs
     video_size, fps = decode_video_format_16_9(video_format)
     logger.info(f"Creating RX config for RGB24 test with video_format: {video_format}")
     try:

tests/validation/tests/single/ffmpeg/test_rx_ffmpeg_tx_ffmpeg.py

@@ -4,6 +4,7 @@
 import os
 
 import pytest
+from common.nicctl import InterfaceSetup
 from mtl_engine import ffmpeg_app
 from mtl_engine.media_files import yuv_files
 
@@ -23,21 +24,25 @@ def test_rx_ffmpeg_tx_ffmpeg(
     test_time,
     build,
     media,
-    nic_port_list,
+    setup_interfaces: InterfaceSetup,
     video_format,
     test_time_multipler,
     output_format,
     test_config,
     prepare_ramdisk,
 ):
     host = list(hosts.values())[0]
+    interfaces_list = setup_interfaces.get_interfaces_list_single(
+        test_config.get("interface_type", "VF")
+    )
 
     video_file = yuv_files[video_format]
 
     ffmpeg_app.execute_test(
         test_time=test_time * test_time_multipler,
         build=build,
         host=host,
+        nic_port_list=interfaces_list,
         type_="frame",
         video_format=video_format,
         pg_format=video_file["format"],

tests/validation/tests/single/ffmpeg/test_rx_ffmpeg_tx_ffmpeg_rgb24.py

@@ -4,6 +4,7 @@
 import os
 
 import pytest
+from common.nicctl import InterfaceSetup
 from mtl_engine import ffmpeg_app
 from mtl_engine.media_files import yuv_files
 
@@ -23,13 +24,16 @@ def test_rx_ffmpeg_tx_ffmpeg_rgb24(
     test_time,
     build,
     media,
-    nic_port_list,
+    setup_interfaces: InterfaceSetup,
     video_format,
     test_time_mutlipler,
     test_config,
     prepare_ramdisk,
 ):
     host = list(hosts.values())[0]
+    interfaces_list = setup_interfaces.get_interfaces_list_single(
+        test_config.get("interface_type", "VF")
+    )
 
     video_file = yuv_files[video_format]
 
@@ -41,4 +45,5 @@ def test_rx_ffmpeg_tx_ffmpeg_rgb24(
         video_format=video_format,
         pg_format=video_file["format"],
         video_url=os.path.join(media, video_file["filename"]),
+        nic_port_list=interfaces_list,
     )

tests/validation/tests/single/ffmpeg/test_rx_ffmpeg_tx_ffmpeg_rgb24_multiple.py

@@ -4,6 +4,7 @@
 import os
 
 import pytest
+from common.nicctl import InterfaceSetup
 from mtl_engine import ffmpeg_app
 from mtl_engine.media_files import yuv_files
 
@@ -25,22 +26,25 @@ def test_rx_ffmpeg_tx_ffmpeg_rgb24_multiple(
     test_time,
     build,
     media,
-    nic_port_list,
+    setup_interfaces: InterfaceSetup,
     video_format_1,
     video_format_2,
     test_time_mutlipler,
     test_config,
     prepare_ramdisk,
 ):
     host = list(hosts.values())[0]
+    interfaces_list = setup_interfaces.get_interfaces_list_single(
+        test_config.get("interface_type", "VF"), count=4
+    )
 
     video_file_1 = yuv_files[video_format_1]
     video_file_2 = yuv_files[video_format_2]
 
     ffmpeg_app.execute_test_rgb24_multiple(
         test_time=test_time * test_time_mutlipler,
         build=build,
-        nic_port_list=host.vfs,
+        nic_port_list=interfaces_list,
         type_="frame",
         video_format_list=[video_format_1, video_format_2],
         pg_format=video_file_1["format"],

tests/validation/tests/single/ffmpeg/test_rx_ffmpeg_tx_rxtxapp.py

@@ -4,6 +4,7 @@
 import os
 
 import pytest
+from common.nicctl import InterfaceSetup
 from mtl_engine import ffmpeg_app
 from mtl_engine.media_files import yuv_files
 
@@ -27,7 +28,7 @@ def test_rx_ffmpeg_tx_rxtxapp(
     test_time,
     build,
     media,
-    nic_port_list,
+    setup_interfaces: InterfaceSetup,
     video_format,
     multiple_sessions,
     test_time_multipler,
@@ -36,12 +37,16 @@ def test_rx_ffmpeg_tx_rxtxapp(
     prepare_ramdisk,
 ):
     host = list(hosts.values())[0]
+    interfaces_list = setup_interfaces.get_interfaces_list_single(
+        test_config.get("interface_type", "VF")
+    )
     video_file = yuv_files[video_format]
 
     ffmpeg_app.execute_test(
         test_time=test_time * test_time_multipler,
         build=build,
         host=host,
+        nic_port_list=interfaces_list,
         type_="frame",
         video_format=video_format,
         pg_format=video_file["format"],