Add cpu policy helper and test

Add a CPU policy tooling to be able to check for certain CPU feature.
The test_cpu_policy also logs hosts CPU policies for debugging purposes.

Signed-off-by: Teddy Astie <teddy.astie@vates.tech>

Author: TSnake41

jobs.py

@@ -469,6 +469,12 @@ class JobData(TypedDict):
             "--vm[]": "multi/limits",
         },
         "paths": ["tests/limits"],
+    },
+    "cpu-policy": {
+        "description": "Tests displaying CPU policies",
+        "nb_pools": 1,
+        "params": {},
+        "paths": ["tests/misc/test_cpu_policy.py"]
     }
 }
 

lib/cpu_policy.py

@@ -0,0 +1,83 @@
+from lib.host import Host
+
+NO_SUBLEAF = 0xffffffff
+
+class CpuPolicy:
+    """A specific CPU policy (PV, HVM, ...)"""
+    # (leaf, subleaf) -> (eax, ebx, ecx, edx)
+    cpuid: dict[(int, int), (int, int, int, int)]
+    # msr -> val
+    msr: dict[int, int]
+
+    def __init__(self, cpuid: dict[(int, int), int], msr: dict[int, int]):
+        self.cpuid = cpuid
+        self.msr = msr
+
+class HostCpuPolicy:
+    """All CPU policies of a host (CPUID, specific MSRs)"""
+    policies: dict[str, CpuPolicy]
+
+    def __init__(self, host: Host):
+        self.policies = dict()
+        text = host.ssh("xen-cpuid --policy")
+        current_policy: CpuPolicy = None
+        mode = None  # None | "cpuid" | "msr"
+
+        lines = text.splitlines()
+
+        for line in lines:
+            line = line.rstrip()
+
+            # ---- Policy header ----
+            if "policy:" in line:
+                # Example: "Raw policy: 32 leaves, 2 MSRs"
+                name, rest = line.split("policy:", 1)
+                name = name.strip()
+
+                self.policies[name] = CpuPolicy(dict(), dict())
+                current_policy = self.policies[name]
+                mode = None
+                continue
+
+            if current_policy is None:
+                continue
+
+            # ---- Section switches ----
+            if line.strip() == "CPUID:":
+                mode = "cpuid"
+                continue
+
+            if line.strip() == "MSRs:":
+                mode = "msr"
+                continue
+
+            # Skip table headers
+            if "leaf" in line or "index" in line or not line.strip():
+                continue
+
+            # ---- CPUID parsing ----
+            if mode == "cpuid":
+                # Example:
+                # 00000004:00000003 -> 1c03c163:02c0003f:00001fff:00000006
+                left, right = line.split("->")
+                leaf_hex, subleaf_hex = left.strip().split(":")
+                eax, ebx, ecx, edx = right.strip().split(":")
+
+                key = (int(leaf_hex, 16), int(subleaf_hex, 16))
+                current_policy.cpuid[key] = (
+                    int(eax, 16),
+                    int(ebx, 16),
+                    int(ecx, 16),
+                    int(edx, 16)
+                )
+                continue
+
+            # ---- MSR parsing ----
+            if mode == "msr":
+                # Example:
+                # 0000010a -> 400000000c000000
+                idx_hex, val_hex = line.split("->")
+                idx = int(idx_hex.strip(), 16)
+                val = int(val_hex.strip(), 16)
+                current_policy.msr[idx] = val
+                continue

tests/misc/test_cpu_policy.py

@@ -0,0 +1,16 @@
+import pytest
+
+import logging
+
+from lib.cpu_policy import NO_SUBLEAF, HostCpuPolicy
+
+class TestCpuPolicy:
+    def test_cpu_policy(self, host):
+        cpu_policy = HostCpuPolicy(host)
+
+        for name, policy in cpu_policy.policies.items():
+            for ((leaf, subleaf), (eax, ebx, ecx, edx)) in policy.cpuid.items():
+                logging.info(f"CPUID[{name}]: {leaf:08x}:{subleaf:08x} -> {eax:08x}:{ebx:08x}:{ecx:08x}:{edx:08x}")
+
+            for (msr, val) in policy.msr.items():
+                logging.info(f"MSR  [{name}]: {msr:08x} -> {val:08x}")