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,73 @@ 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, hosts, interface_type="VF", count=2):
+        selected_interfaces = {k: [] for k in hosts.keys()}
+        for host in hosts.values():
+            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 interface_type == "VFxPF":
+                    for i in range(count):
+                        vfs = self.nicctl_objs[host.name].create_vfs(host.network_interfaces[i].pci_address.lspci, 1)
+                        selected_interfaces[host.name].extend(vfs)
+                        self.register_cleanup(self.nicctl_objs[host.name], host.network_interfaces[i].pci_address.lspci, "VF")
+                else:
+                    raise Exception(f"Unknown interface type {interface_type}")
+        return selected_interfaces
+
+    def get_test_interfaces_list(self, hosts, interface_type="VF", count=2):
+        selected_interfaces = self.get_test_interfaces(hosts, interface_type, count)
+        interface_list = []
+        for host in hosts.values():
+            interface_list.extend(selected_interfaces[host.name])
+        return interface_list
+
+    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
@@ -124,6 +124,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/tests/single/st30p/st30p_channel/test_st30p_channel.py

@@ -2,6 +2,7 @@
 # Copyright(c) 2024-2025 Intel Corporation
 import logging
 
+from common.nicctl import InterfaceSetup
 import mtl_engine.RxTxApp as rxtxapp
 import pytest
 from common.integrity.integrity_runner import FileAudioIntegrityRunner
@@ -35,7 +36,7 @@ def test_st30p_channel(
     hosts,
     build,
     media,
-    nic_port_list,
+    setup_interfaces: InterfaceSetup,
     test_time,
     audio_channel,
     request,
@@ -49,12 +50,13 @@ def test_st30p_channel(
     SDBQ1001_audio_channel_check(audio_channel, media_file_info["format"], request)
 
     host = list(hosts.values())[0]
+    interfaces_list = setup_interfaces.get_test_interfaces_list(hosts, test_config.get("interface_type", "VF"))
     out_file_url = host.connection.path(media_file_path).parent / "out.pcm"
 
     config = rxtxapp.create_empty_config()
     config = rxtxapp.add_st30p_sessions(
         config=config,
-        nic_port_list=host.vfs,
+        nic_port_list=interfaces_list,
         test_mode="multicast",
         audio_format=media_file_info["format"],
         audio_channel=[audio_channel],