Fix Cloud Hypervisor live migration nextest filtering

Run Cloud Hypervisor live migration tests through the CLI path and construct the nextest filter as a shell-safe expression. Preserve the base test filter while excluding the serial-only live migration test from the parallel run.

Author: vyadavmsft

lisa/microsoft/testsuites/cloud_hypervisor/ch_tests.py

@@ -126,14 +126,17 @@ def verify_cloud_hypervisor_live_migration_tests(
             "ch_live_migration_tests_excluded",
         )
         ch_tests: CloudHypervisorTests = node.tools[CloudHypervisorTests]
+        # Cloud Hypervisor exposes live migration as integration subtests.
         ch_tests.run_tests(
             result,
             "integration-live-migration",
             hypervisor,
             log_path,
             ref,
-            include_list,
-            exclude_list,
+            only=include_list,
+            skip=exclude_list,
+            cli_test_type="integration",
+            cli_test_filter="live_migration",
         )
 
     @TestCaseMetadata(

lisa/microsoft/testsuites/cloud_hypervisor/ch_tests_tool.py

@@ -144,18 +144,73 @@ def _sanitize_name(self, s: str) -> str:
         """Sanitize names for filenames: keep alphanumeric, dot, dash, underscore."""
         return re.sub(r"[^A-Za-z0-9_.-]", "_", s)
 
+    @staticmethod
+    def _escape_nextest_filter_matcher(matcher: str) -> str:
+        return matcher.replace("\\", "\\\\").replace(")", "\\)")
+
+    @staticmethod
+    def _quote_bash_arg(arg: str) -> str:
+        escaped = (
+            arg.replace("\\", "\\\\")
+            .replace('"', '\\"')
+            .replace("$", "\\$")
+            .replace("`", "\\`")
+        )
+        return f'"{escaped}"'
+
+    def _build_nextest_filterset(
+        self,
+        base_filter: str,
+        only: Optional[List[str]],
+        skip: Optional[List[str]],
+    ) -> str:
+        filterset = f"test(~{self._escape_nextest_filter_matcher(base_filter)})"
+
+        if only:
+            included_tests = "|".join(
+                f"test(={self._escape_nextest_filter_matcher(test_name)})"
+                for test_name in only
+            )
+            filterset = f"({filterset}&({included_tests}))"
+
+        if skip:
+            for skipped_test in skip:
+                skipped_filter = self._escape_nextest_filter_matcher(skipped_test)
+                filterset = f"{filterset}-test(={skipped_filter})"
+
+        return f"--filterset={filterset}"
+
     def _prepare_subtests(
         self,
         test_type: str,
         hypervisor: str,
         only: Optional[List[str]],
         skip: Optional[List[str]],
+        cli_test_type: Optional[str] = None,
+        only_test_prefixes: Optional[List[str]] = None,
     ) -> Dict[str, Any]:
         """Prepare subtests and skip arguments."""
-        subtests = self._list_subtests(hypervisor, test_type)
+        subtests = self._list_subtests(hypervisor, cli_test_type or test_type)
         # Store the ordered list for diagnostic purposes
         self._ordered_subtests = subtests.copy()
 
+        if only_test_prefixes is not None:
+            prefixed_subtests = [
+                subtest
+                for subtest in subtests
+                if any(subtest.startswith(prefix) for prefix in only_test_prefixes)
+            ]
+            if not prefixed_subtests:
+                fail(
+                    f"No Cloud Hypervisor {test_type} subtests matched prefixes "
+                    f"{only_test_prefixes}. Verify the selected Cloud Hypervisor "
+                    "ref contains those integration tests."
+                )
+            if only is None:
+                only = prefixed_subtests
+            else:
+                only = [subtest for subtest in only if subtest in prefixed_subtests]
+
         if only is not None:
             if not skip:
                 skip = []
@@ -409,16 +464,46 @@ def run_tests(
         ref: str = "",
         only: Optional[List[str]] = None,
         skip: Optional[List[str]] = None,
+        cli_test_type: Optional[str] = None,
+        only_test_prefixes: Optional[List[str]] = None,
+        cli_test_filter: Optional[str] = None,
     ) -> None:
+        cli_test_type = cli_test_type or test_type
+
         if ref:
             self.node.tools[Git].checkout(ref, self.repo_root)
 
-        subtests = self._prepare_subtests(test_type, hypervisor, only, skip)
+        if cli_test_filter:
+            subtests: Dict[str, Any] = {"subtest_set": set(), "skip_args": ""}
+        else:
+            subtests = self._prepare_subtests(
+                test_type,
+                hypervisor,
+                only,
+                skip,
+                cli_test_type,
+                only_test_prefixes,
+            )
         self._configure_environment_if_needed(hypervisor)
 
         # Use enhanced diagnostics for better debugging and monitoring
         skip_args = subtests["skip_args"]
-        cmd_args = f"tests --hypervisor {hypervisor} --{test_type} -- -- {skip_args}"
+        if cli_test_filter:
+            nextest_filter = self._build_nextest_filterset(
+                cli_test_filter,
+                only,
+                skip,
+            )
+            test_script_args = f"--test-filter {self._quote_bash_arg(nextest_filter)}"
+            cmd_args = (
+                f"tests --hypervisor {hypervisor} --{cli_test_type} -- "
+                f"{test_script_args}"
+            )
+        else:
+            cmd_args = (
+                f"tests --hypervisor {hypervisor} --{cli_test_type} -- -- "
+                f"{skip_args}"
+            )
         # normalize name so artifacts are predictable (no spaces/colons/slashes)
         safe_test_type = self._sanitize_name(test_type.replace("-", "_"))
         test_name = self._sanitize_name(f"ch_{safe_test_type}_{hypervisor}")
@@ -1379,8 +1464,8 @@ def _cache_vmm_version(self) -> None:
         except Exception as e:
             self._log.debug(f"Could not cache VMM version: {e}")
 
-    def _list_subtests(self, hypervisor: str, test_type: str) -> List[str]:
-        cmd_args = f"tests --hypervisor {hypervisor} --{test_type} -- -- --list"
+    def _list_subtests(self, hypervisor: str, cli_test_type: str) -> List[str]:
+        cmd_args = f"tests --hypervisor {hypervisor} --{cli_test_type} -- -- --list"
         # Use enhanced environment variables for consistency
         enhanced_env_vars = self.env_vars.copy()
         enhanced_env_vars.update(