Updating Kself Test Case to support custom collections and tests to be skipped

microsoft/testsuites/kselftest/kselftest-suite.py

@@ -33,6 +33,15 @@ class KselftestTestsuite(TestSuite):
 
         For both cases, verify that the kselftest tool extracts the tar, runs the script
         run_kselftest.sh and redirects test results to a file kselftest-results.txt.
+
+        Customization:
+        Users can customize the test by specifying the
+        `kselftest_include_test_collections` and `kselftest_skip_tests` variables
+        in the runbook. For example:
+        - `kselftest_include_test_collections`: A comma-separated list of collections
+        to run (e.g., "bpf,net,timers").
+        - `kselftest_skip_tests`: A comma-separated list of tests to skip
+        (e.g., "net:test_tcp,test_udp").
         """,
         priority=3,
         timeout=_CASE_TIME_OUT,
@@ -50,12 +59,30 @@ def verify_kselftest(
         file_path = variables.get("kselftest_file_path", "")
         working_path = variables.get("kselftest_working_path", "")
         run_as_root = variables.get("kselftest_run_as_root", False)
+        test_collection_list = (
+            variables.get("kselftest_include_test_collections", "").split(",")
+            if variables.get("kselftest_include_test_collections", "")
+            else []
+        )
+        skip_tests_list = (
+            variables.get("kselftest_skip_tests", "").split(",")
+            if variables.get("kselftest_skip_tests", "")
+            else []
+        )
+
         try:
             kselftest: Kselftest = node.tools.get(
                 Kselftest,
                 working_path=working_path,
                 file_path=file_path,
             )
-            kselftest.run_all(result, log_path, self._KSELF_TIMEOUT, run_as_root)
+            kselftest.run_all(
+                test_result=result,
+                log_path=log_path,
+                timeout=self._KSELF_TIMEOUT,
+                run_test_as_root=run_as_root,
+                run_collections=test_collection_list,
+                skip_tests=skip_tests_list,
+            )
         except UnsupportedDistroException as identifier:
             raise SkippedException(identifier)

microsoft/testsuites/kselftest/kselftest.py

@@ -214,6 +214,7 @@ def run_all(
         log_path: str,
         timeout: int = 5000,
         run_test_as_root: bool = False,
+        **kwargs: Any,
     ) -> List[KselftestResult]:
         # Executing kselftest as root may cause
         # VM to hang
@@ -229,13 +230,62 @@ def run_all(
 
         result_file_name = "kselftest-results.txt"
         result_file = f"{result_directory}/{result_file_name}"
-        self.run(
-            f" 2>&1 | tee {result_file}",
-            sudo=run_test_as_root,
-            force_run=True,
-            shell=True,
-            timeout=timeout,
-        )
+        # Initialize run_tests and skip_tests from kwargs if provided
+        run_collections = kwargs.get("run_collections", [])
+        skip_tests = kwargs.get("skip_tests", [])
+
+        if run_collections or skip_tests:
+            # List all available tests
+            list_result = self.run(" -l", shell=True)
+            list_result.assert_exit_code(
+                message="failed to retrieve the list of available kself tests"
+            )
+            all_tests = list_result.stdout.splitlines()
+
+            # Filter tests based on run_collections if it exists
+            # Example: if run_collections = ['uevent']
+            # all_tests will already have all tests in the format:
+            #   ['core:close_range_test', 'core:unshare_test',
+            #    'tty:tty_tstamp_update', 'uevent:uevent_filtering']
+            # The filtered_tests will then have the value:
+            #   ['uevent:uevent_filtering']
+            # This means all the tests that belong to the 'uevent'
+            #   collection are selected.
+            if run_collections:
+                filtered_tests = [
+                    test
+                    for test in all_tests
+                    if any(
+                        (match := re.match(r"^[^:/]+", test))
+                        and collection == match.group(0)
+                        for collection in run_collections
+                    )
+                ]
+            else:
+                filtered_tests = all_tests
+
+            # Exclude tests based on skip_tests
+            tests_to_run = [test for test in filtered_tests if test not in skip_tests]
+
+            if tests_to_run:
+                tests_to_run_str = " ".join(f"-t {test}" for test in tests_to_run)
+                self._log.debug(f"Running tests: {tests_to_run}")
+                self.run(
+                    f" {tests_to_run_str} 2>&1 | tee -a {result_file}",
+                    sudo=run_test_as_root,
+                    force_run=True,
+                    shell=True,
+                    timeout=timeout,
+                )
+        else:
+            # run all tests
+            self.run(
+                f" 2>&1 | tee {result_file}",
+                sudo=run_test_as_root,
+                force_run=True,
+                shell=True,
+                timeout=timeout,
+            )
 
         # Allow read permissions for "others" to remote copy the file
         # kselftest-results.txt