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,85 @@
+# 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,
+)
+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="cvm",
+    category="functional",
+    description="""
+    This test suite is for generating CVM attestation report only for azure cvms.
+    """,
+)
+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)
+        rb = environment.platform.runbook.get_extended_runbook(BaseLibvirtNodeSchema)
+        libvirt_runbook = cast(BaseLibvirtNodeSchema, rb)
+
+        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 req in libvirt_runbook.device_passthrough:
+            pool_type = str(req.pool_type.value)
+            ven_dev_of_pool = pool_vendor_device_map[pool_type]
+            ven_id = ven_dev_of_pool["vendor_id"]
+            dev_id = ven_dev_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)} devices on "
+                    f"guest vm. 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}"
+                )