Adding a CLI timeout flag and unit tests

Author: antonio-amjr

tests/test_run_tests.py

@@ -689,6 +689,103 @@ def test_run_tests_config_data_processing(
         assert "dut_config" in result.output
         assert "network" in result.output
 
+    def test_run_tests_prompt_timeout_merges_into_execution_config(
+        self,
+        cli_runner: CliRunner,
+        mock_async_apis: Mock,
+        sample_test_collections: api_models.TestCollections,
+        sample_test_run_execution: api_models.TestRunExecutionWithChildren,
+        sample_default_config_dict: dict,
+    ) -> None:
+        """--prompt-timeout should merge th_config.prompt_timeout_seconds into the
+        execution config submitted for the run, without requiring --config."""
+        # Arrange
+        project_api = mock_async_apis.projects_api.default_config_api_v1_projects_default_config_get
+        test_collection_api = mock_async_apis.test_collections_api.read_test_collections_api_v1_test_collections__get
+        test_run_executions_api = mock_async_apis.test_run_executions_api
+        cli_api = test_run_executions_api.create_cli_test_run_execution_api_v1_test_run_executions_cli_post
+        id_start = test_run_executions_api.start_test_run_execution_api_v1_test_run_executions_id_start_post
+
+        project_api.return_value = sample_default_config_dict
+        test_collection_api.return_value = sample_test_collections
+        cli_api.return_value = sample_test_run_execution
+        id_start.return_value = sample_test_run_execution
+        with patch("th_cli.commands.run_tests.AsyncApis", return_value=mock_async_apis):
+            with patch(
+                "th_cli.commands.run_tests.test_logging.configure_logger_for_run", return_value="./test_logs/test.log"
+            ):
+                with patch("th_cli.commands.run_tests.TestRunSocket") as mock_socket_class:
+                    with patch(
+                        "th_cli.commands.run_tests.convert_nested_to_dict", return_value=sample_default_config_dict
+                    ):
+                        mock_socket = Mock()
+                        mock_socket.connect_websocket = AsyncMock()
+                        mock_socket_class.return_value = mock_socket
+
+                        # Act
+                        result = cli_runner.invoke(
+                            run_tests,
+                            ["--tests-list", "TC-ACE-1.1", "--prompt-timeout", "300"],
+                        )
+
+        # Assert
+        assert result.exit_code == 0
+        assert "Prompt Timeout Used (Execution Only)" in result.output
+        submitted_body = cli_api.call_args[0][0]
+        assert submitted_body.execution_config["th_config"]["prompt_timeout_seconds"] == 300
+
+    def test_run_tests_prompt_timeout_wins_over_config_file(
+        self,
+        cli_runner: CliRunner,
+        mock_async_apis: Mock,
+        sample_test_collections: api_models.TestCollections,
+        sample_test_run_execution: api_models.TestRunExecutionWithChildren,
+        sample_default_config_dict: dict,
+        mock_json_config_file: Path,
+    ) -> None:
+        """--prompt-timeout takes precedence over any prompt_timeout_seconds already
+        present in a --config file, since it's applied after the config merge."""
+        # Arrange
+        project_api = mock_async_apis.projects_api.default_config_api_v1_projects_default_config_get
+        test_collection_api = mock_async_apis.test_collections_api.read_test_collections_api_v1_test_collections__get
+        test_run_executions_api = mock_async_apis.test_run_executions_api
+        cli_api = test_run_executions_api.create_cli_test_run_execution_api_v1_test_run_executions_cli_post
+        id_start = test_run_executions_api.start_test_run_execution_api_v1_test_run_executions_id_start_post
+
+        project_api.return_value = sample_default_config_dict
+        test_collection_api.return_value = sample_test_collections
+        cli_api.return_value = sample_test_run_execution
+        id_start.return_value = sample_test_run_execution
+        with patch("th_cli.commands.run_tests.AsyncApis", return_value=mock_async_apis):
+            with patch(
+                "th_cli.commands.run_tests.test_logging.configure_logger_for_run", return_value="./test_logs/test.log"
+            ):
+                with patch("th_cli.commands.run_tests.TestRunSocket") as mock_socket_class:
+                    with patch(
+                        "th_cli.commands.run_tests.convert_nested_to_dict", return_value=sample_default_config_dict
+                    ):
+                        mock_socket = Mock()
+                        mock_socket.connect_websocket = AsyncMock()
+                        mock_socket_class.return_value = mock_socket
+
+                        # Act
+                        result = cli_runner.invoke(
+                            run_tests,
+                            [
+                                "--tests-list",
+                                "TC-ACE-1.1",
+                                "--config",
+                                str(mock_json_config_file),
+                                "--prompt-timeout",
+                                "45",
+                            ],
+                        )
+
+        # Assert
+        assert result.exit_code == 0
+        submitted_body = cli_api.call_args[0][0]
+        assert submitted_body.execution_config["th_config"]["prompt_timeout_seconds"] == 45
+
     @pytest.mark.parametrize(
         "invalid_test_id",
         [

tests/test_utils.py

@@ -573,6 +573,23 @@ def test_merge_configs_nested(self) -> None:
         assert result["network"]["thread"]["channel"] == 15  # Preserved
         assert result["dut_config"]["pairing_mode"] == "onnetwork"  # Preserved
 
+    def test_merge_configs_th_config_override(self) -> None:
+        """Test merging a partial th_config override, as used by --prompt-timeout."""
+        # Arrange
+        base = {
+            "network": {"wifi": {"ssid": "default", "password": "default"}},
+            "th_config": {"prompt_timeout_seconds": 60, "enable_realtime_python_test_logs": None},
+        }
+        override = {"th_config": {"prompt_timeout_seconds": 300}}
+
+        # Act
+        result = merge_configs(base, override)
+
+        # Assert
+        assert result["th_config"]["prompt_timeout_seconds"] == 300
+        assert result["th_config"]["enable_realtime_python_test_logs"] is None  # Preserved
+        assert result["network"]["wifi"]["ssid"] == "default"  # Preserved
+
     def test_merge_configs_deep_nesting(self) -> None:
         """Test deeply nested configuration merging."""
         # Arrange

th_cli/commands/run_tests.py

@@ -127,6 +127,14 @@
     is_flag=True,
     help=colorize_help("Disable real-time log streaming via web browser (enabled by default)."),
 )
+@click.option(
+    "--prompt-timeout",
+    type=int,
+    help=colorize_help(
+        "Override the user-prompt response timeout in seconds for this run only "
+        "(th_config.prompt_timeout_seconds)."
+    ),
+)
 @async_cmd
 @click.pass_context
 async def run_tests(
@@ -139,6 +147,7 @@ async def run_tests(
     project_id: int | None = None,
     no_color: bool = False,
     no_streaming: bool = False,
+    prompt_timeout: int | None = None,
 ) -> None:
     """Execute a CLI test run from selected test cases.
 
@@ -151,6 +160,7 @@ async def run_tests(
         tc_params_file: Optional path to TC parameters mapping JSON file
         project_id: Optional project ID for the test run
         no_color: Flag to disable colored output
+        prompt_timeout: Optional override for the user-prompt response timeout (seconds)
 
     Raises:
         CLIError: If there are validation or execution errors
@@ -271,6 +281,13 @@ async def run_tests(
                 test_run_config["test_parameters"] = {}
             test_run_config["test_parameters"].update(extra_test_params)
 
+        # Override the user-prompt timeout if provided (execution-only, not persisted)
+        if prompt_timeout is not None:
+            click.echo(colorize_key_value("Prompt Timeout Used (Execution Only)", f"{prompt_timeout}s"))
+            test_run_config = merge_configs(
+                test_run_config, {"th_config": {"prompt_timeout_seconds": prompt_timeout}}
+            )
+
         # Retrieve available test collections to build test selection
         test_collections = await test_collections_api.read_test_collections_api_v1_test_collections__get()
         selected_tests_dict = build_test_selection(test_collections, validated_test_ids)