Merge pull request avocado-framework#6225 from dhsrivas/master

utils: Add PCI device, drivers, modules, dmesg related avocado utilities

Author: richtja

avocado/utils/dmesg.py

@@ -182,3 +182,22 @@ def filter_strings(line):
         return not any(string in line for string in skip_messages)
 
     return "\n".join(filter(None, filter(filter_strings, dmesg_stdout.splitlines())))
+
+
+def check_kernel_logs(pattern):
+    """
+    Check if "pattern" is present in kernel logs.
+
+    :param pattern: string to check in kernel logs
+    :return: True if pattern found, False otherwise
+    """
+    out = process.run("dmesg", ignore_status=True, verbose=False, sudo=True)
+    if pattern in out.stdout_text:
+        return True
+
+    cmd = "journalctl -k -b"
+    out = process.run(cmd, ignore_status=True, verbose=False, sudo=True)
+    if pattern in out.stdout_text:
+        return True
+
+    return False

avocado/utils/linux_modules.py

@@ -264,6 +264,31 @@ def parse_kernel_config(kernel_config):
             return parse_kernel_config(kernel_config)
 
 
+def configure_module(module, config):
+    """
+    Check if 'config' is not set, builtin or module.
+    Load 'module' if 'config' is set as module.
+
+    :param module: kernel module to configure
+    :param config: kernel config to check and validate
+    :return: True if module is builtin or loaded successfully. Else False.
+    :rtype: boolean
+    """
+    config_status = check_kernel_config(config)
+    if config_status == ModuleConfig.BUILTIN:
+        LOG.info("%s is built-in. %s already loaded.", config, module)
+        return True
+
+    if config_status == ModuleConfig.MODULE:
+        if load_module(module) and module_is_loaded(module):
+            LOG.info("Module %s loaded successfully", module)
+            return True
+        LOG.error("Failed to load module %s", module)
+
+    LOG.error("%s is not set. %s cannot be loaded.", config, module)
+    return False
+
+
 def get_modules_dir():
     """
     Return the modules dir for the running kernel version

avocado/utils/pci.py

@@ -21,6 +21,7 @@
 """
 
 
+import errno
 import os
 import re
 
@@ -377,6 +378,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": 950,
+    "unit": 965,
     "jobs": 11,
     "functional-parallel": 366,
     "functional-serial": 7,

selftests/unit/utils/dmesg.py

@@ -0,0 +1,53 @@
+import unittest
+from unittest import mock
+
+from avocado import Test
+from avocado.utils import dmesg
+
+
+class TestCheckKernelLogs(Test):
+    @mock.patch("avocado.utils.dmesg.process.run")
+    def test_pattern_found_in_dmesg(self, mock_run):
+        # Simulate `dmesg` output containing the pattern
+        mock_run.return_value.stdout_text = "kernel: test_pattern detected"
+        mock_run.return_value.exit_status = 0
+
+        result = dmesg.check_kernel_logs("test_pattern")
+        self.assertTrue(result)
+        mock_run.assert_called_with(
+            "dmesg", ignore_status=True, verbose=False, sudo=True
+        )
+
+    @mock.patch("avocado.utils.dmesg.process.run")
+    def test_pattern_found_in_journalctl(self, mock_run):
+        dmesg_out = mock.Mock()
+        dmesg_out.stdout_text = "kernel: something else"
+        dmesg_out.exit_status = 0
+
+        journal_out = mock.Mock()
+        journal_out.stdout_text = "kernel: test_pattern here"
+        journal_out.exit_status = 0
+
+        mock_run.side_effect = [dmesg_out, journal_out]
+
+        result = dmesg.check_kernel_logs("test_pattern")
+        self.assertTrue(result)
+
+    @mock.patch("avocado.utils.dmesg.process.run")
+    def test_pattern_not_found(self, mock_run):
+        dmesg_out = mock.Mock()
+        dmesg_out.stdout_text = "kernel: no match"
+        dmesg_out.exit_status = 0
+
+        journal_out = mock.Mock()
+        journal_out.stdout_text = "kernel: still no match"
+        journal_out.exit_status = 0
+
+        mock_run.side_effect = [dmesg_out, journal_out]
+
+        result = dmesg.check_kernel_logs("test_pattern")
+        self.assertFalse(result)
+
+
+if __name__ == "__main__":
+    unittest.main()

selftests/unit/utils/linux_modules.py

@@ -63,6 +63,38 @@ def test_is_module_loaded(self):
             self.assertTrue(linux_modules.module_is_loaded("rfcomm"))
             self.assertFalse(linux_modules.module_is_loaded("unknown_module"))
 
+    @unittest.mock.patch("avocado.utils.linux_modules.check_kernel_config")
+    def test_configure_module_1(self, mock_check_kernel_config):
+        mock_check_kernel_config.return_value = linux_modules.ModuleConfig.NOT_SET
+        self.assertFalse(linux_modules.configure_module("mod", "CONFIG_MOD"))
+
+    @unittest.mock.patch("avocado.utils.linux_modules.module_is_loaded")
+    @unittest.mock.patch("avocado.utils.linux_modules.load_module")
+    @unittest.mock.patch("avocado.utils.linux_modules.check_kernel_config")
+    def test_configure_module_2(self, mock_check, mock_load, mock_loaded):
+        mock_check.return_value = linux_modules.ModuleConfig.BUILTIN
+        self.assertTrue(linux_modules.configure_module("mod", "CONFIG_MOD"))
+        mock_load.assert_not_called()
+        mock_loaded.assert_not_called()
+
+    @unittest.mock.patch("avocado.utils.linux_modules.module_is_loaded")
+    @unittest.mock.patch("avocado.utils.linux_modules.load_module")
+    @unittest.mock.patch("avocado.utils.linux_modules.check_kernel_config")
+    def test_configure_module_3(self, mock_check, mock_load, mock_loaded):
+        mock_check.return_value = linux_modules.ModuleConfig.MODULE
+        mock_load.return_value = True
+        mock_loaded.return_value = True
+        self.assertTrue(linux_modules.configure_module("mod", "CONFIG_MOD"))
+
+    @unittest.mock.patch("avocado.utils.linux_modules.module_is_loaded")
+    @unittest.mock.patch("avocado.utils.linux_modules.load_module")
+    @unittest.mock.patch("avocado.utils.linux_modules.check_kernel_config")
+    def test_configure_module_4(self, mock_check, mock_load, mock_loaded):
+        mock_check.return_value = linux_modules.ModuleConfig.MODULE
+        mock_load.return_value = False
+        mock_loaded.return_value = False
+        self.assertFalse(linux_modules.configure_module("mod", "CONFIG_MOD"))
+
 
 if __name__ == "__main__":
     unittest.main()

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()