Added an actor to install a specific RPM driver package for HP Gen8 NIC

repos/system_upgrade/cloudlinux/actors/pcidevicedrivers/actor.py

@@ -0,0 +1,20 @@
+from leapp.actors import Actor
+from leapp.libraries.actor import pcidevicedrivers
+from leapp.models import PCIDevices, RpmTransactionTasks
+from leapp.tags import IPUWorkflowTag, FactsPhaseTag
+
+
+class PCIDeviceDrivers(Actor):
+    """
+    Handles driver installation for different PCI devices.
+
+    Consumes data from lspci to detect if any bundled RPM driver packages need to be installed.
+    """
+
+    name = 'pci_device_drivers'
+    consumes = (PCIDevices,)
+    produces = (RpmTransactionTasks,)
+    tags = (IPUWorkflowTag, FactsPhaseTag)
+
+    def process(self):
+        pcidevicedrivers.process()

repos/system_upgrade/cloudlinux/actors/pcidevicedrivers/libraries/pcidevicedrivers.py

@@ -0,0 +1,56 @@
+import os
+
+from leapp.libraries.stdlib import api
+from leapp.models import PCIDevices, RpmTransactionTasks
+
+
+def process():
+    """
+    Main processing function to handle drivers for PCI devices.
+    """
+    pci_devices = next(api.consume(PCIDevices), None)
+    if pci_devices is None:
+        api.current_logger().warning('No PCI devices data available.')
+        return
+
+    for device in pci_devices.devices:
+        api.current_logger().info(
+            'Processing device: class {}, vendor {}, name {}'.format(device.dev_cls,
+                                                                     device.vendor,
+                                                                     device.name))
+        if _is_hp_gen8_device(device):
+            _process_hp_gen8_nic_driver(device)
+
+
+def _is_hp_gen8_device(device):
+    """
+    Check if the device is HP Gen8 NIC.
+    """
+    return (device.dev_cls == 'Ethernet controller'
+            and 'Emulex' in device.vendor
+            and 'OneConnect 10Gb NIC' in device.name)
+
+
+def _process_hp_gen8_nic_driver(device):
+    """
+    Handle driver installation for HP Gen8 NIC.
+    """
+    api.current_logger().info('Processing driver update for device: {}'.format(device.name))
+    driver_rpm_name = 'kmod-be2net-12.0.0.0-12.el8_8.elrepo.x86_64.rpm'
+    _add_local_rpm(driver_rpm_name)
+
+
+def _add_local_rpm(rpm_name):
+    """
+    Add a local RPM to the transaction tasks if it exists.
+    """
+    location = api.get_folder_path('drivers')
+    local_rpms = [name for name in os.listdir(location) if name.endswith('.rpm')]
+
+    if rpm_name in local_rpms:
+        rpm_real_path = os.path.realpath(os.path.join(location, rpm_name))
+        api.produce(RpmTransactionTasks(local_rpms=[rpm_real_path]))
+        api.current_logger().info(
+            'Produced RpmTransactionTasks for RPM: {}'.format(rpm_real_path))
+    else:
+        api.current_logger().warning('RPM {} not found in {}'.format(rpm_name, location))

repos/system_upgrade/cloudlinux/actors/pcidevicedrivers/tests/test_pci_device_drivers.py

@@ -0,0 +1,55 @@
+import os
+
+import pytest
+
+from leapp.libraries.actor import pcidevicedrivers as pdd
+from leapp.libraries.common.testutils import logger_mocked
+from leapp.models import PCIDevice, PCIDevices, RpmTransactionTasks
+
+FAKE_FOLDER = '/FAKE/FOLDER/PATH'
+HP_GEN8_DRIVER_RPM = 'kmod-be2net-12.0.0.0-12.el8_8.elrepo.x86_64.rpm'
+
+DEVICES = [
+    PCIDevice(
+        slot='00:02.0',
+        dev_cls='Ethernet controller',
+        vendor='Red Hat, Inc.',
+        name='Virtio network device',
+        pci_id='8086:7020:1af4:1100',
+    ),
+    PCIDevice(
+        slot='00:02.1',
+        dev_cls='Ethernet controller',
+        vendor='Emulex Corporation',
+        name='OneConnect 10Gb NIC (be3)',
+        pci_id='1af4:1002:1af4:0005',
+    ),
+]
+
+TEST_CASES = [
+    (['abc.rpm', 'random.file', HP_GEN8_DRIVER_RPM], HP_GEN8_DRIVER_RPM),
+]
+
+
+@pytest.mark.parametrize('rpms,driver_rpm', TEST_CASES)
+def test_pci_device_drivers(monkeypatch, rpms, driver_rpm):
+    result = []
+
+    def consume_pci_devices_mocked(*models):
+        yield PCIDevices(devices=DEVICES)
+
+    monkeypatch.setattr(pdd.api, 'consume', consume_pci_devices_mocked)
+    monkeypatch.setattr(pdd.api, 'current_logger', logger_mocked())
+    monkeypatch.setattr(pdd.api, 'get_folder_path', lambda _: FAKE_FOLDER)
+    monkeypatch.setattr(pdd.os, 'listdir', lambda _: rpms)
+    monkeypatch.setattr(pdd.api, 'produce', lambda *models: result.extend(models))
+
+    pdd.process()
+
+    assert result
+    assert isinstance(result[0], RpmTransactionTasks)
+    assert result[0].local_rpms
+    assert result[0].local_rpms == [os.path.join(FAKE_FOLDER, driver_rpm)]
+    assert not result[0].to_install
+    assert not result[0].to_remove
+    assert not result[0].to_keep