Add device-passthrough testcase for CH platform

Add test case to verify if passthrough-devices
are visible to guest for CH platform

Signed-off-by: Smit Gardhariya <[email protected]>

Author: smit-gardhariya

microsoft/testsuites/device_passthrough/functional_tests.py

@@ -0,0 +1,87 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+from pathlib import Path
+from typing import Any, Dict, cast
+
+from lisa import (
+    Environment,
+    Logger,
+    Node,
+    TestCaseMetadata,
+    TestSuite,
+    TestSuiteMetadata,
+    schema,
+)
+from lisa.sut_orchestrator import CLOUD_HYPERVISOR
+from lisa.sut_orchestrator.libvirt.ch_platform import CloudHypervisorPlatform
+from lisa.sut_orchestrator.libvirt.context import get_node_context as get_ch_context
+from lisa.sut_orchestrator.libvirt.schema import BaseLibvirtNodeSchema
+from lisa.testsuite import TestResult, simple_requirement
+from lisa.tools import Lspci
+from lisa.util import LisaException, SkippedException
+
+
+@TestSuiteMetadata(
+    area="device_passthrough",
+    category="functional",
+    description="""
+    This test suite is for testing device passthrough functional tests.
+    """,
+)
+class DevicePassthroughFunctionalTests(TestSuite):
+    @TestCaseMetadata(
+        description="""
+            Check if passthrough device is visible to guest.
+            This testcase support only on CLOUD_HYPERVISOR
+            platform of LISA.
+        """,
+        priority=4,
+        requirement=simple_requirement(
+            supported_platform_type=[CLOUD_HYPERVISOR],
+        ),
+    )
+    def verify_device_passthrough_on_guest(
+        self,
+        log: Logger,
+        node: Node,
+        environment: Environment,
+        log_path: Path,
+        result: TestResult,
+        variables: Dict[str, Any],
+    ) -> None:
+        ctx = get_ch_context(node)
+        if not ctx.passthrough_devices:
+            raise SkippedException("No device-passthrough is set for node")
+
+        lspci = node.tools[Lspci]
+        platform = cast(CloudHypervisorPlatform, environment.platform)
+        pool_vendor_device_map = {}
+        for pool in platform.platform_runbook.device_pools:
+            pool_type = str(pool.type.value)
+            vendor_device_id = {
+                "vendord_id": pool.devices[0].vendor_id,
+                "device_id": pool.devices[0].device_id,
+            }
+            pool_vendor_device_map[pool_type] = vendor_device_id
+
+        for node_space in environment.runbook.nodes_requirement:
+            node_runbook: BaseLibvirtNodeSchema = node_space.get_extended_runbook(
+                BaseLibvirtNodeSchema, CLOUD_HYPERVISOR
+            )
+            for req in node_runbook.device_passthrough:
+                pool_type = str(req.pool_type.value)
+                ven_dev_id_of_pool = pool_vendor_device_map[pool_type]
+                ven_id = ven_dev_id_of_pool["vendor_id"]
+                dev_id = ven_dev_id_of_pool["device_id"]
+                devices = lspci.get_devices_by_vendor_device_id(
+                    vendor_id=ven_id,
+                    device_id=dev_id,
+                    force_run=True,
+                )
+                if len(devices) != req.count:
+                    raise LisaException(
+                        f"Device count don't match, got total: {len(devices)} "
+                        f"As per runbook, Required device count: {req.count}, "
+                        f"for pool_type: {pool_type} having Vendor/Device ID as "
+                        f"vendor_id = {ven_id}, device_id={dev_id}"
+                    )