Support configured test skips

classes/commands/RunTestsCommand.py

@@ -41,6 +41,12 @@ def register_command(cls, subparsers: argparse._SubParsersAction):
         parser.add_argument('-v', '--verbose', action='store_true', help="Enables verbose output")
         # TestFramework arguments
         parser.add_argument('-rn', '--run-name', default='xUnit', help='The name to be given to the test run')
+        parser.add_argument(
+            '--skip-tests',
+            action='append',
+            default=[],
+            help='Exact Suite@Test path to skip; may be specified more than once',
+        )
         parser.set_defaults(command_class=cls)
 
     async def execute(self) -> None:
@@ -82,7 +88,11 @@ async def execute(self) -> None:
         # Now we can start the remote control server to run tests.
         # This two-step approach prevents rebuilding the project on restarts (test timeout, crash, ...)
         args = self._build_gmrt_arguments(run_job)
-        remote = RemoteControlServer(ExecutionMode.AUTOMATIC, run_name=run_name)
+        remote = RemoteControlServer(
+            ExecutionMode.AUTOMATIC,
+            run_name=run_name,
+            skip_tests=self.get_argument('skip_tests'),
+        )
         await manage_server(
             lambda: remote.serve_or_wait_for_space(gmrt_exe, args, port=TCP_PORT), 
             port=HTTP_PORT

classes/server/RemoteControlServer.py

@@ -29,16 +29,23 @@ class RemoteCommand(Enum):
 
 class RemoteControlServer:
 
-    def __init__(self, mode: ExecutionMode, timeout: int = 1, run_name = 'xUnit'):
+    def __init__(
+        self,
+        mode: ExecutionMode,
+        timeout: int = 1,
+        run_name='xUnit',
+        skip_tests: Optional[list[str]] = None,
+    ):
         """
         Initialize the RemoteControlServer with the given mode.
 
         Args:
             mode (Mode): The mode of operation, either AUTOMATIC or MANUAL.
         """
         self.mode = mode
-        self.timeout = timeout
-        self.run_name = run_name
+        self.timeout = timeout
+        self.run_name = run_name
+        self.skip_tests = set(skip_tests or [])
 
         # Parse platform metadata from run_name (format: name:platform:config)
         self.properties = {}
@@ -53,10 +60,33 @@ def __init__(self, mode: ExecutionMode, timeout: int = 1, run_name = 'xUnit'):
         self.state = State.WAITING
         self.stop_event = asyncio.Event()
         self.reboot_event = asyncio.Event()
-        self.strategy = self._select_strategy()
+        self.strategy = self._select_strategy()
 
         self.framework_result: TestFrameworkResult = None
-        self.suite_results: dict[str, TestSuiteResult] = {}
+        self.suite_results: dict[str, TestSuiteResult] = {}
+
+    def _configure_tests(self, available_tests: list[str]) -> None:
+        """Select runnable tests and record configured exclusions as skipped."""
+        available_test_set = set(available_tests)
+        for test_path in sorted(self.skip_tests - available_test_set):
+            LOGGER.warning("Configured skipped test was not found: %s", test_path)
+
+        self.tests = []
+        for test_path in available_tests:
+            if test_path not in self.skip_tests:
+                self.tests.append(test_path)
+                continue
+
+            suite_name, test_name = test_path.split('@', 1)
+            LOGGER.info("Skipping configured test: %s", test_path)
+            self._add_test_result(
+                {
+                    'name': test_name,
+                    'result': 'Skipped',
+                },
+                suite_name,
+                time.time(),
+            )
 
     def _select_strategy(self) -> Coroutine[Any,Any,None]:
         """
@@ -272,8 +302,8 @@ async def _handle_automatic_mode(self, reader: asyncio.StreamReader, writer: asy
             if not received_data:
                 return
 
-            # Update test list
-            self.tests = received_data.splitlines()
+            # Update test list and record configured exclusions as skipped.
+            self._configure_tests(received_data.splitlines())
 
         # Transition to RUNNING state
         self.state = State.RUNNING

launcher.py

@@ -90,8 +90,15 @@ def load_config_and_merge_with_cli_args():
     # Merge config args with command args, allowing CLI args to take precedence
     merged_args = merge_config_and_cli_args(config_args, command_args)
 
-    # Reconstruct remaining_argv from merged_args
-    remaining_argv = [f'--{k}' if v is None else f'--{k}={v}' for k, v in merged_args.items()]
+    # Reconstruct remaining_argv from merged_args. List-valued JSON entries are
+    # represented as repeated command-line options for argparse ``append``
+    # arguments.
+    remaining_argv = []
+    for key, value in merged_args.items():
+        if isinstance(value, list):
+            remaining_argv.extend(f'--{key}={item}' for item in value)
+        else:
+            remaining_argv.append(f'--{key}' if value is None else f'--{key}={value}')
 
     # Add the original command and non-flag CLI arguments back
     if command: