[VIRT] Add PCI topology stability tests

tests/virt/node/general/test_pci_topology_stability.py

@@ -0,0 +1,176 @@
+"""
+PCI Topology Stability Tests
+
+STP: https://github.com/RedHatQE/openshift-virtualization-tests-design-docs/blob/main/stps/sig-virt/pci-topology-stability.md
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+import pytest
+from ocp_resources.template import Template
+from ocp_resources.virtual_machine_restore import VirtualMachineRestore
+from ocp_resources.virtual_machine_snapshot import VirtualMachineSnapshot
+
+from tests.os_params import RHEL_LATEST, RHEL_LATEST_LABELS
+from tests.virt.utils import get_pci_addresses
+from utilities.virt import (
+    VirtualMachineForTestsFromTemplate,
+    migrate_vm_and_verify,
+    restart_vm_wait_for_running_vm,
+    running_vm,
+)
+
+if TYPE_CHECKING:
+    from utilities.virt import VirtualMachineForTests
+
+LOGGER = logging.getLogger(__name__)
+
+pytestmark = [pytest.mark.rwx_default_storage, pytest.mark.data_collector_scope(scope="module")]
+
+
+@pytest.fixture(scope="class")
+def pci_topology_vm(
+    request,
+    namespace,
+    unprivileged_client,
+    golden_image_data_volume_template_for_test_scope_class,
+    modern_cpu_for_migration,
+):
+    with VirtualMachineForTestsFromTemplate(
+        name=request.param["vm_name"],
+        labels=Template.generate_template_labels(**request.param["template_labels"]),
+        namespace=namespace.name,
+        client=unprivileged_client,
+        data_volume_template=golden_image_data_volume_template_for_test_scope_class,
+        cpu_model=modern_cpu_for_migration,
+    ) as vm:
+        running_vm(vm=vm)
+        yield vm
+
+
+@pytest.fixture()
+def initial_pci_addresses(pci_topology_vm):
+    return get_pci_addresses(vm=pci_topology_vm)
+
+
+@pytest.fixture()
+def restarted_pci_topology_vm(pci_topology_vm, initial_pci_addresses):
+    restart_vm_wait_for_running_vm(vm=pci_topology_vm)
+    return pci_topology_vm
+
+
+@pytest.fixture()
+def migrated_pci_topology_vm(admin_client, pci_topology_vm, initial_pci_addresses):
+    migrate_vm_and_verify(vm=pci_topology_vm, client=admin_client, check_ssh_connectivity=True)
+    return pci_topology_vm
+
+
+@pytest.fixture()
+def snapshot_restored_pci_topology_vm(admin_client, pci_topology_vm, initial_pci_addresses):
+    with VirtualMachineSnapshot(
+        name=f"{pci_topology_vm.name}-snapshot",
+        namespace=pci_topology_vm.namespace,
+        vm_name=pci_topology_vm.name,
+    ) as snapshot:
+        snapshot.wait_snapshot_done()
+        pci_topology_vm.stop(wait=True)
+        LOGGER.info(f"Restoring VM {pci_topology_vm.name} from snapshot {snapshot.name}")
+        with VirtualMachineRestore(
+            client=admin_client,
+            name=f"{pci_topology_vm.name}-restore",
+            namespace=pci_topology_vm.namespace,
+            vm_name=pci_topology_vm.name,
+            snapshot_name=snapshot.name,
+        ) as restore:
+            restore.wait_restore_done()
+    running_vm(vm=pci_topology_vm)
+    return pci_topology_vm
+
+
+@pytest.mark.parametrize(
+    "golden_image_data_source_for_test_scope_class, pci_topology_vm",
+    [
+        pytest.param(
+            {"os_dict": RHEL_LATEST},
+            {"template_labels": RHEL_LATEST_LABELS, "vm_name": "pci-topology-vm"},
+        ),
+    ],
+    indirect=True,
+)
+@pytest.mark.arm64
+class TestPCITopologyStability:
+    """
+    Verify PCI device addresses remain stable across VM lifecycle operations.
+
+    Preconditions:
+        - RHEL VM created from latest template with a modern CPU model for migration,
+          started and running with SSH access
+        - Sorted PCI BDF address list captured before each operation
+    """
+
+    @pytest.mark.polarion("CNV-16326")
+    def test_pci_address_stability_on_restart(
+        self,
+        initial_pci_addresses: list[str],
+        restarted_pci_topology_vm: VirtualMachineForTests,
+    ):
+        """
+        Steps:
+            1. Restart the VM and wait until it is running with SSH access
+            2. Capture PCI addresses from the guest
+
+        Expected:
+            - Address lists match (PCI topology unchanged)
+        """
+        addresses_after = get_pci_addresses(vm=restarted_pci_topology_vm)
+        assert addresses_after == initial_pci_addresses, (
+            f"PCI topology changed after restart:\n  before: {initial_pci_addresses}\n  after:  {addresses_after}"
+        )
+
+    @pytest.mark.polarion("CNV-16327")
+    def test_pci_address_stability_on_migration(
+        self,
+        initial_pci_addresses: list[str],
+        migrated_pci_topology_vm: VirtualMachineForTests,
+    ):
+        """
+        Steps:
+            1. Live-migrate the VM and verify SSH connectivity
+            2. Capture PCI addresses from the guest
+
+        Expected:
+            - Address lists match (PCI topology unchanged)
+        """
+        addresses_after = get_pci_addresses(vm=migrated_pci_topology_vm)
+        assert addresses_after == initial_pci_addresses, (
+            f"PCI topology changed after migration:\n  before: {initial_pci_addresses}\n  after:  {addresses_after}"
+        )
+
+    @pytest.mark.polarion("CNV-16328")
+    @pytest.mark.usefixtures("skip_if_no_storage_class_for_snapshot")
+    def test_pci_address_stability_on_snapshot_restore(
+        self,
+        initial_pci_addresses: list[str],
+        snapshot_restored_pci_topology_vm: VirtualMachineForTests,
+    ):
+        """
+        Preconditions:
+            - Storage class supports snapshots
+
+        Steps:
+            1. Create a VM snapshot and wait until it completes
+            2. Stop the VM
+            3. Restore the VM from the snapshot
+            4. Start the VM and wait until it is running with SSH access
+            5. Capture PCI addresses from the guest
+
+        Expected:
+            - Address lists match (PCI topology unchanged)
+        """
+        addresses_after = get_pci_addresses(vm=snapshot_restored_pci_topology_vm)
+        assert addresses_after == initial_pci_addresses, (
+            f"PCI topology changed after snapshot restore:\n  before: {initial_pci_addresses}\n  after:  {addresses_after}"
+        )

tests/virt/upgrade/conftest.py

@@ -19,7 +19,7 @@
     vm_from_template,
     wait_for_automatic_vm_migrations,
 )
-from tests.virt.utils import get_boot_time_for_multiple_vms
+from tests.virt.utils import get_boot_time_for_multiple_vms, get_pci_addresses
 from utilities.artifactory import get_test_artifact_server_url
 from utilities.constants import Images
 from utilities.constants.images import OS_FLAVOR_RHEL
@@ -321,6 +321,11 @@ def virt_migratable_vms_names(virt_migratable_vms):
     return vm_names
 
 
+@pytest.fixture(scope="session")
+def pci_addresses_before_upgrade(vms_for_upgrade):
+    return {vm.name: get_pci_addresses(vm=vm) for vm in vms_for_upgrade}
+
+
 @pytest.fixture(scope="session")
 def linux_boot_time_before_upgrade(vms_for_upgrade):
     return get_boot_time_for_multiple_vms(vm_list=vms_for_upgrade)

tests/virt/upgrade/test_upgrade_virt.py

@@ -21,7 +21,7 @@
     verify_run_strategy_vmi_status,
     verify_vms_ssh_connectivity,
 )
-from tests.virt.utils import assert_migration_post_copy_mode, verify_guest_boot_time
+from tests.virt.utils import assert_migration_post_copy_mode, get_pci_addresses, verify_guest_boot_time
 from utilities.constants.hco import DATA_SOURCE_NAME
 from utilities.constants.pytest import DEPENDENCY_SCOPE_SESSION
 from utilities.exceptions import ResourceValueError
@@ -78,7 +78,8 @@ class TestUpgradeVirt:
     @pytest.mark.polarion("CNV-2974")
     @pytest.mark.order("first")
     @pytest.mark.dependency(name=VMS_RUNNING_BEFORE_UPGRADE_TEST_NODE_ID)
-    def test_is_vm_running_before_upgrade(self, vms_for_upgrade, linux_boot_time_before_upgrade):
+    @pytest.mark.usefixtures("linux_boot_time_before_upgrade", "pci_addresses_before_upgrade")
+    def test_is_vm_running_before_upgrade(self, vms_for_upgrade):
         for vm in vms_for_upgrade:
             assert vm.vmi.status == VirtualMachineInstance.Status.RUNNING
 
@@ -298,6 +299,29 @@ def test_vms_boot_time_after_upgrade(
         migratable_vms = [vm for vm in vms_for_upgrade if vm.name in virt_migratable_vms_names]
         verify_guest_boot_time(vm_list=migratable_vms, initial_boot_time=linux_boot_time_before_upgrade)
 
+    @pytest.mark.ocp_upgrade
+    @pytest.mark.sno
+    @pytest.mark.polarion("CNV-16329")
+    @pytest.mark.order(
+        after=[IMAGE_UPDATE_AFTER_UPGRADE_NODE_ID, VIRT_VMS_RUNNING_AFTER_UPGRADE_TEST_NODE_ID],
+        before=AFTER_UPGRADE_STORAGE_ORDERING,
+    )
+    @pytest.mark.dependency(
+        depends=[
+            IUO_UPGRADE_TEST_DEPENDENCY_NODE_ID,
+            VIRT_VMS_RUNNING_AFTER_UPGRADE_TEST_NODE_ID,
+        ],
+        scope=DEPENDENCY_SCOPE_SESSION,
+    )
+    def test_pci_topology_after_upgrade(self, vms_for_upgrade, pci_addresses_before_upgrade):
+        """STP: https://github.com/RedHatQE/openshift-virtualization-tests-design-docs/blob/main/stps/sig-virt/pci-topology-stability.md"""
+        failed_vms = {}
+        for vm in vms_for_upgrade:
+            current_addresses = get_pci_addresses(vm=vm)
+            if current_addresses != pci_addresses_before_upgrade[vm.name]:
+                failed_vms[vm.name] = {"before": pci_addresses_before_upgrade[vm.name], "after": current_addresses}
+        assert not failed_vms, f"PCI topology changed after upgrade: {failed_vms}"
+
     @pytest.mark.ocp_upgrade
     @pytest.mark.sno
     @pytest.mark.polarion("CNV-3682")

tests/virt/utils.py

@@ -399,6 +399,27 @@ def get_non_terminated_pods(client, node):
     )
 
 
+def get_pci_addresses(vm: VirtualMachineForTests) -> list[str]:
+    """Get sorted PCI device lines visible to the guest.
+
+    Each line pairs a BDF address with its device description, enabling
+    detection of both address shifts and device swaps on failure.
+
+    Args:
+        vm: Running VM with SSH access.
+
+    Returns:
+        Sorted list of lspci lines (e.g. ["00:01.0 Display controller: ..."]).
+    """
+    output = run_ssh_commands(
+        host=vm.ssh_exec,
+        commands=["lspci"],
+    )[0].strip()
+    addresses = output.splitlines()
+    LOGGER.info(f"PCI addresses for VM {vm.name}: {addresses}")
+    return addresses
+
+
 def get_boot_time_for_multiple_vms(vm_list):
     return {vm.name: get_vm_boot_time(vm=vm) for vm in vm_list}