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

@@ -2,7 +2,7 @@
 import re
 from dataclasses import dataclass
 from pathlib import PurePath, PurePosixPath
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 from assertpy import assert_that
 
@@ -212,6 +212,8 @@ def run_all(
         log_path: str,
         timeout: int = 5000,
         run_test_as_root: bool = False,
+        run_collections: Optional[List[str]] = None,
+        skip_tests: Optional[List[str]] = None,
     ) -> List[KselftestResult]:
         # Executing kselftest as root may cause VM to hang
 
@@ -232,14 +234,62 @@ def run_all(
         else:
             work_dir = None
 
-        self.run(
-            f" -s 2>&1 | tee {result_file}",
-            cwd=work_dir,
-            sudo=run_test_as_root,
-            force_run=True,
-            shell=True,
-            timeout=timeout,
-        )
+        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
+
+            # Ensure skip_tests is not None
+            skip_tests = skip_tests or []
+            # 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}",
+                    cwd=work_dir,
+                    sudo=run_test_as_root,
+                    force_run=True,
+                    shell=True,
+                    timeout=timeout,
+                )
+        else:
+            # run all tests
+            self.run(
+                f" 2>&1 | tee {result_file}",
+                cwd=work_dir,
+                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