utils/pci: Introduce functions for attaching driver and interrupt capability checks

This patch introduces a set of PCI utility functions to simplify device
management and capability checks:

* add_vendor_id() – retrieves the vendor ID of a PCI device and associate
  it with the specified driver
* attach_driver() – unbinds a device from its current driver and rebinds
  it to a given driver
* check_msix_capability() – verifies whether the device supports msix
  interrupts
* device_supports_irqs() – checks if the device supports at least a
  specified number of IRQs

These helpers provide a cleaner and reusable interface for driver binding
and interrupt capability validation.

Signed-off-by: Dheeraj Kumar Srivastava <dheerajkumar.srivastava@amd.com>

Author: dhsrivas

avocado/utils/pci.py

@@ -21,6 +21,7 @@
 """
 
 
+import errno
 import os
 import re
 
@@ -376,6 +377,94 @@ def get_vendor_id(full_pci_address):
     return out.split(" ")[2].strip()
 
 
+def add_vendor_id(full_pci_address, driver):
+    """
+    Retrieve and add the vendor ID of a PCI device to the specified driver.
+
+    :param full_pci_address: Full PCI device address, including domain
+                             (e.g., 0000:03:00.0).
+    :param driver: Driver to associate with the vendor ID of the PCI device.
+    """
+    # Get vendor id of pci device
+    output = get_vendor_id(full_pci_address)
+    vid = output.replace(":", " ")
+
+    # Add device vendor id to the driver
+    try:
+        genio.write_file(f"/sys/bus/pci/drivers/{driver}/new_id", f"{vid}\n")
+    except OSError as e:
+        if e.errno != errno.EEXIST:
+            raise ValueError(
+                f"Failed to add vendor ID '{vid}' to driver '{driver}': {e}"
+            ) from e
+
+
+def attach_driver(full_pci_address, driver):
+    """
+    Unbind the device from its existing driver and bind it to the given driver.
+
+    :param full_pci_address: Full PCI device address (e.g., 0000:03:00.0)
+    :param driver: Driver to be attached to the specified PCI device
+    :raises ValueError: If `driver` or `full_pci_address` is None, or if the driver
+        could not be attached successfully due to an OS error.
+    :warning: This function may unbind the device from its current driver, which can
+        temporarily make the device unavailable until it is reattached.
+    """
+
+    try:
+        if not driver or not full_pci_address:
+            raise ValueError("'driver' or 'full_pci_address' inputs are None")
+
+        # add vendor id of device to driver
+        add_vendor_id(full_pci_address, driver)
+
+        # unbind the device from its initial driver
+        cur_driver = get_driver(full_pci_address)
+        if cur_driver is not None:
+            unbind(cur_driver, full_pci_address)
+
+        # Bind device to driver
+        bind(driver, full_pci_address)
+
+    except (OSError, ValueError, genio.GenIOError) as e:
+        raise ValueError(
+            f"Not able to attach {driver} to {full_pci_address}. Reason: {e}"
+        ) from e
+
+
+def check_msix_capability(full_pci_address):
+    """
+    Check whether the PCI device supports Extended Message Signaled Interrupts.
+
+    :param full_pci_address: Full PCI address including domain (0000:03:00.0)
+    :return: True if supported, False otherwise
+    """
+    out = process.run(f"lspci -vvs {full_pci_address}", ignore_status=True, shell=True)
+    if out.exit_status:
+        raise ValueError(f"lspci failed for {full_pci_address}: {out.stderr_text}")
+    return any("MSI-X:" in line for line in out.stdout_text.splitlines())
+
+
+def device_supports_irqs(full_pci_address, count):
+    """
+    Check if the device supports at least the specified number of interrupts.
+
+    :param full_pci_address: Full PCI device address including domain (e.g., 0000:03:00.0)
+    :param count: Number of IRQs the device should support
+    :return: True if supported, False otherwise
+    """
+    out = process.run(
+        f"lspci -vv -s {full_pci_address}", ignore_status=True, shell=True
+    )
+    if out.exit_status:
+        raise ValueError(f"lspci failed for {full_pci_address}: {out.stderr_text}")
+    match = re.search(r"MSI-X:.*?Count=(\d+)", out.stdout_text, re.S)
+    if match:
+        nirq = int(match.group(1))
+        return nirq >= count
+    return False
+
+
 def reset_check(full_pci_address):
     """
     Check if reset for "full_pci_address" is successful

selftests/check.py

@@ -27,7 +27,7 @@
     "job-api-check-tmp-directory-exists": 1,
     "nrunner-interface": 90,
     "nrunner-requirement": 28,
-    "unit": 957,
+    "unit": 965,
     "jobs": 11,
     "functional-parallel": 366,
     "functional-serial": 7,

selftests/unit/utils/pci.py

@@ -1,3 +1,4 @@
+import errno
 import unittest.mock
 
 from avocado.utils import pci
@@ -42,6 +43,69 @@ def test_get_slot_from_sysfs_negative(self):
             ):
                 self.assertRaises(ValueError, pci.get_slot_from_sysfs, "0002:01:00.1")
 
+    @unittest.mock.patch("avocado.utils.pci.get_vendor_id")
+    @unittest.mock.patch("avocado.utils.pci.genio.write_file")
+    def test_add_vendor_id_success(self, mock_write, mock_get_vid):
+        mock_get_vid.return_value = "1234:abcd"
+        pci.add_vendor_id("0000:03:00.0", "driver")
+        mock_get_vid.assert_called_once_with("0000:03:00.0")
+        mock_write.assert_called_once_with(
+            "/sys/bus/pci/drivers/driver/new_id", "1234 abcd\n"
+        )
+
+    @unittest.mock.patch("avocado.utils.pci.get_vendor_id", return_value="1234:abcd")
+    @unittest.mock.patch(
+        "avocado.utils.pci.genio.write_file",
+        side_effect=OSError(errno.EEXIST, "File exists"),
+    )
+    def test_add_vendor_id_already_exists(self, mock_write, mock_get_vid):
+        pci.add_vendor_id("0000:03:00.0", "driver")
+        mock_get_vid.assert_called_once_with("0000:03:00.0")
+        mock_write.assert_called_once_with(
+            "/sys/bus/pci/drivers/driver/new_id", "1234 abcd\n"
+        )
+
+    @unittest.mock.patch("avocado.utils.pci.bind")
+    @unittest.mock.patch("avocado.utils.pci.unbind")
+    @unittest.mock.patch("avocado.utils.pci.get_driver")
+    @unittest.mock.patch("avocado.utils.pci.add_vendor_id")
+    def test_attach_driver(self, mock_add_vid, mock_get_driver, mock_unbind, mock_bind):
+        mock_get_driver.return_value = "old_driver"
+        pci.attach_driver("0000:03:00.0", "new_driver")
+        mock_add_vid.assert_called_once()
+        mock_unbind.assert_called_once_with("old_driver", "0000:03:00.0")
+        mock_bind.assert_called_once_with("new_driver", "0000:03:00.0")
+
+    @unittest.mock.patch("avocado.utils.pci.process.run")
+    def test_check_msix_capability_supported(self, mock_run):
+        mock_run.return_value.exit_status = 0
+        mock_run.return_value.stdout_text = "Capabilities: [90] MSI-X: Enable+ Count=16"
+        self.assertTrue(pci.check_msix_capability("0000:03:00.0"))
+
+    @unittest.mock.patch("avocado.utils.pci.process.run")
+    def test_check_msix_capability_unsupported(self, mock_run):
+        mock_run.return_value.exit_status = 0
+        mock_run.return_value.stdout_text = "Capabilities: [90] MSI: Enable+ Count=16"
+        self.assertFalse(pci.check_msix_capability("0000:03:00.0"))
+
+    @unittest.mock.patch("avocado.utils.pci.process.run")
+    def test_device_supports_irqs_enough(self, mock_run):
+        mock_run.return_value.exit_status = 0
+        mock_run.return_value.stdout_text = "MSI-X: Enable+ Count=64"
+        self.assertTrue(pci.device_supports_irqs("0000:03:00.0", 32))
+
+    @unittest.mock.patch("avocado.utils.pci.process.run")
+    def test_device_supports_irqs_insufficient(self, mock_run):
+        mock_run.return_value.exit_status = 0
+        mock_run.return_value.stdout_text = "MSI-X: Enable+ Count=4"
+        self.assertFalse(pci.device_supports_irqs("0000:03:00.0", 16))
+
+    @unittest.mock.patch("avocado.utils.pci.process.run")
+    def test_device_supports_irqs_no_msix(self, mock_run):
+        mock_run.return_value.exit_status = 0
+        mock_run.return_value.stdout_text = "Capabilities: [90] MSI: Enable+ Count=16"
+        self.assertFalse(pci.device_supports_irqs("0000:03:00.0", 8))
+
 
 if __name__ == "__main__":
     unittest.main()