[Feature] Adding THconfig schema as a new TH config field

app/schemas/test_environment_config.py

@@ -48,11 +48,21 @@ class TestEnvironmentConfigError(Exception):
     """
 
 
+class THConfig(BaseModel):
+    """Test-Harness-specific settings, independent of any particular DUT/program."""
+
+    prompt_timeout_seconds: int = 60
+    # None means "not set at project level", deferring to the
+    # ENABLE_REALTIME_PYTHON_TEST_LOGS environment variable.
+    enable_realtime_python_test_logs: Optional[bool] = None
+
+
 class TestEnvironmentConfig(BaseModel):
     __test__ = False  # Needed to indicate to PyTest that this is not a "test"
 
     # TODO(#490): Need to be refactored to support real PIXIT format
     test_parameters: Optional[dict[str, Any]]
+    th_config: Optional[THConfig] = None
 
     def __init__(self, **kwargs: Any):
         try:

app/tests/user_prompts/test_prompt_exchange.py

@@ -21,11 +21,34 @@
 
 from app.user_prompt_support import user_prompt_manager
 from app.user_prompt_support.constants import UserResponseStatusEnum
-from app.user_prompt_support.prompt_request import PromptRequest
+from app.user_prompt_support.prompt_request import PromptRequest, default_timeout_s
 from app.user_prompt_support.prompt_response import PromptResponse
 from app.user_prompt_support.user_prompt_manager import PromptExchange
 
 
+def test_prompt_request_default_timeout_is_unresolved() -> None:
+    """
+    PromptRequest.timeout defaults to None (unresolved), not a concrete value.
+
+    UserPromptSupport.send_prompt_request() is responsible for resolving it from
+    th_config before dispatch; PromptExchange applies a defensive fallback to
+    default_timeout_s if a request ever reaches it unresolved (see below).
+    """
+    assert PromptRequest(prompt="Test string").timeout is None
+
+
+def test_prompt_exchange_resolves_unset_timeout_to_default() -> None:
+    """
+    PromptExchange defensively resolves an unset (None) timeout to
+    default_timeout_s, so a request bypassing UserPromptSupport's funnel never
+    reaches wait_for() with an unresolved timeout.
+    """
+    request: PromptRequest = PromptRequest(prompt="Test string")
+    exchange: PromptExchange = PromptExchange(prompt=request, message_id=0)
+
+    assert exchange.prompt.timeout == default_timeout_s
+
+
 def test_prompt_exchange_handle_empty_response() -> None:
     """
     This tests the handle_response() by passing an empty message dictionary.

app/tests/user_prompts/test_user_prompt_support.py

@@ -18,6 +18,9 @@
 import pytest
 
 from app.user_prompt_support import user_prompt_manager
+from app.user_prompt_support.constants import UserResponseStatusEnum
+from app.user_prompt_support.prompt_request import PromptRequest, default_timeout_s
+from app.user_prompt_support.prompt_response import PromptResponse
 from app.user_prompt_support.user_prompt_support import (
     UserPromptError,
     UserPromptSupport,
@@ -39,3 +42,108 @@
         with pytest.raises(UserPromptError):
             await prompt_support.send_prompt_request(prompt_request=mock.MagicMock())
             send_prompt_request.assert_called_once()
+
+
+async def _send_and_capture_timeout(
+    prompt_support: UserPromptSupport, prompt_request: PromptRequest
+) -> int:
+    """Send prompt_request and return the timeout that reached user_prompt_manager."""
+    captured: dict = {}
+
+    async def _capture(prompt_request: PromptRequest) -> PromptResponse:
+        captured["timeout"] = prompt_request.timeout
+        return PromptResponse(status_code=UserResponseStatusEnum.OKAY, response="ok")
+
+    with mock.patch.object(
+        user_prompt_manager.user_prompt_manager,
+        "send_prompt_request",
+        side_effect=_capture,
+    ):
+        await prompt_support.send_prompt_request(prompt_request=prompt_request)
+
+    return captured["timeout"]
+
+
+@pytest.mark.asyncio
+async def test_send_prompt_request_resolves_timeout_from_config() -> None:
+    """An unspecified timeout is resolved from th_config.prompt_timeout_seconds."""
+    prompt_support = UserPromptSupport()
+    prompt_support.config = {"th_config": {"prompt_timeout_seconds": 300}}
+
+    timeout = await _send_and_capture_timeout(
+        prompt_support, PromptRequest(prompt="hi")
+    )
+
+    assert timeout == 300
+
+
+@pytest.mark.asyncio
+async def test_send_prompt_request_falls_back_without_config_attribute() -> None:
+    """No `.config` attribute at all (e.g. TestStep-based mixins) falls back safely."""
+    prompt_support = UserPromptSupport()
+
+    timeout = await _send_and_capture_timeout(
+        prompt_support, PromptRequest(prompt="hi")
+    )
+
+    assert timeout == default_timeout_s
+
+
+@pytest.mark.asyncio
+async def test_send_prompt_request_falls_back_with_empty_config() -> None:
+    """An empty/absent th_config key falls back to the default timeout."""
+    prompt_support = UserPromptSupport()
+    prompt_support.config = {}
+
+    timeout = await _send_and_capture_timeout(
+        prompt_support, PromptRequest(prompt="hi")
+    )
+
+    assert timeout == default_timeout_s
+
+
+@pytest.mark.asyncio
+async def test_send_prompt_request_falls_back_with_null_th_config() -> None:
+    """A hand-edited "th_config": null must not raise AttributeError."""
+    prompt_support = UserPromptSupport()
+    prompt_support.config = {"th_config": None}
+
+    timeout = await _send_and_capture_timeout(
+        prompt_support, PromptRequest(prompt="hi")
+    )
+
+    assert timeout == default_timeout_s
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "prompt_timeout_seconds",
+    ["not-a-number", 0, -5, True],
+)
+async def test_send_prompt_request_falls_back_with_malformed_value(
+    prompt_timeout_seconds: object,
+) -> None:
+    """A malformed/invalid configured value falls back safely instead of crashing."""
+    prompt_support = UserPromptSupport()
+    prompt_support.config = {
+        "th_config": {"prompt_timeout_seconds": prompt_timeout_seconds}
+    }
+
+    timeout = await _send_and_capture_timeout(
+        prompt_support, PromptRequest(prompt="hi")
+    )
+
+    assert timeout == default_timeout_s
+
+
+@pytest.mark.asyncio
+async def test_send_prompt_request_explicit_timeout_wins_over_config() -> None:
+    """A caller-supplied timeout is never overridden by th_config."""
+    prompt_support = UserPromptSupport()
+    prompt_support.config = {"th_config": {"prompt_timeout_seconds": 300}}
+
+    timeout = await _send_and_capture_timeout(
+        prompt_support, PromptRequest(prompt="hi", timeout=5)
+    )
+
+    assert timeout == 5

app/user_prompt_support/prompt_request.py

@@ -24,7 +24,10 @@
 
 class PromptRequest(BaseModel):
     prompt: Optional[str]
-    timeout: int = default_timeout_s
+    # None means "caller did not specify a timeout"; UserPromptSupport.send_prompt_request
+    # resolves it (from the project's th_config, falling back to default_timeout_s) before
+    # the request is dispatched.
+    timeout: Optional[int] = None
 
     @property
     def messageType(self) -> MessageTypeEnum:

app/user_prompt_support/user_prompt_manager.py

@@ -30,7 +30,7 @@
     UserResponseStatusEnum,
 )
 
-from .prompt_request import PromptRequest
+from .prompt_request import PromptRequest, default_timeout_s
 from .prompt_response import PromptResponse
 
 
@@ -40,6 +40,11 @@
 class PromptExchange(object):
     def __init__(self, prompt: PromptRequest, message_id: int) -> None:
         self.message_id = message_id
+        # Callers are expected to resolve a concrete timeout before dispatch (see
+        # UserPromptSupport.send_prompt_request); this is a defensive fallback so an
+        # unresolved timeout never reaches wait_for() or gets sent to the frontend.
+        if prompt.timeout is None:
+            prompt = prompt.copy(update={"timeout": default_timeout_s})
         self.prompt = prompt
         self.message_event = Event()
         self.received_response: Optional[PromptResponse]

app/user_prompt_support/user_prompt_support.py

@@ -14,7 +14,7 @@
 # limitations under the License.
 #
 from .constants import UserResponseStatusEnum
-from .prompt_request import PromptRequest
+from .prompt_request import PromptRequest, default_timeout_s
 from .prompt_response import PromptResponse
 from .user_prompt_manager import user_prompt_manager
 
@@ -31,12 +31,32 @@ class UserPromptSupport(object):
     async def send_prompt_request(
         self, prompt_request: PromptRequest
     ) -> PromptResponse:
+        if prompt_request.timeout is None:
+            prompt_request = prompt_request.copy(
+                update={"timeout": self.__resolve_prompt_timeout()}
+            )
+
         response = await user_prompt_manager.send_prompt_request(prompt_request)
 
         if response is None:
             raise UserPromptError("No prompt response returned")
         return response
 
+    def __resolve_prompt_timeout(self) -> int:
+        """Resolve the default prompt timeout from the project's th_config.
+
+        Falls back to default_timeout_s when no config is available (e.g. this
+        mixin is used from a class with no `.config` property) or the configured
+        value is missing/invalid.
+        """
+        config = getattr(self, "config", None)
+        if config:
+            th_config = config.get("th_config") or {}
+            value = th_config.get("prompt_timeout_seconds")
+            if isinstance(value, int) and not isinstance(value, bool) and value > 0:
+                return value
+        return default_timeout_s
+
     async def invoke_prompt_and_get_str_response(
         self, prompt_request: PromptRequest
     ) -> str:

test_collections/matter/default_project.config

@@ -28,5 +28,9 @@
     "chip_use_paa_certs":false,
     "trace_log":true
   },
-  "test_parameters": null
+  "test_parameters": null,
+  "th_config": {
+    "prompt_timeout_seconds": 60,
+    "enable_realtime_python_test_logs": null
+  }
 }

test_collections/matter/sdk_tests/support/python_testing/models/test_case.py

@@ -145,9 +145,21 @@ async def step_success(
         self, logger: Any, logs: str, duration: int, request: Any
     ) -> None:
         # Display logs captured during this step only if real-time logging is enabled
-        if settings.ENABLE_REALTIME_PYTHON_TEST_LOGS:
+        if self._realtime_logs_enabled():
             await self._display_step_logs()
 
+    def _realtime_logs_enabled(self) -> bool:
+        """Whether Python test logs should be displayed incrementally per step.
+
+        The project's th_config.enable_realtime_python_test_logs, when explicitly
+        set, overrides the instance-wide ENABLE_REALTIME_PYTHON_TEST_LOGS env var.
+        """
+        th_config = (self.config or {}).get("th_config") or {}
+        override = th_config.get("enable_realtime_python_test_logs")
+        if override is not None:
+            return bool(override)
+        return settings.ENABLE_REALTIME_PYTHON_TEST_LOGS
+
     async def _display_step_logs(self) -> None:
         """Display logs that were captured during the current step."""
         # Validate file path is set and file exists
@@ -265,7 +277,7 @@ async def step_failure(
     ) -> None:
         # Display logs captured during this step before marking as failure
         # only if real-time logging is enabled
-        if settings.ENABLE_REALTIME_PYTHON_TEST_LOGS:
+        if self._realtime_logs_enabled():
             await self._display_step_logs()
 
         failure_msg = "Python test step failure"
@@ -462,7 +474,7 @@ async def cleanup(self) -> None:
         logger.info("Test Cleanup")
         # Log any remaining content that wasn't captured by steps
         # only if real-time logging is enabled
-        if settings.ENABLE_REALTIME_PYTHON_TEST_LOGS:
+        if self._realtime_logs_enabled():
             await self._log_remaining_content()
         else:
             # Use batch logging when real-time logging is disabled
@@ -644,7 +656,7 @@ async def execute(self) -> None:
 
             # Check for any remaining logs that weren't captured by steps
             # or show all logs if real-time logging is disabled
-            if settings.ENABLE_REALTIME_PYTHON_TEST_LOGS:
+            if self._realtime_logs_enabled():
                 await self._log_remaining_content()
             else:
                 # Use batch logging when real-time logging is disabled

test_collections/matter/sdk_tests/support/tests/matter/test_test_environment_config.py

@@ -28,6 +28,7 @@
     default_config_thread_no_ba_port,
     default_config_thread_valid,
     default_matter_config,
+    default_matter_config_with_th_config,
 )
 from test_collections.matter.test_environment_config import TestEnvironmentConfigMatter
 
@@ -38,6 +39,30 @@ def test_create_config_matter_with_valid_config_success() -> None:
     assert config_matter is not None
 
 
+def test_create_config_matter_with_no_th_config_defaults_to_none() -> None:
+    config_matter = TestEnvironmentConfigMatter(**default_matter_config)
+
+    assert config_matter.th_config is None
+
+
+def test_create_config_matter_with_th_config_round_trips() -> None:
+    config_matter = TestEnvironmentConfigMatter(**default_matter_config_with_th_config)
+
+    assert config_matter.th_config is not None
+    assert config_matter.th_config.prompt_timeout_seconds == 120
+    assert config_matter.th_config.enable_realtime_python_test_logs is True
+
+
+def test_create_config_matter_with_invalid_th_config_type_fails() -> None:
+    config = {
+        **default_matter_config,
+        "th_config": {"prompt_timeout_seconds": "not-a-number"},
+    }
+
+    with pytest.raises(TestEnvironmentConfigError):
+        TestEnvironmentConfigMatter(**config)
+
+
 def test_create_config_matter_with_no_config_fails() -> None:
     with pytest.raises(TestEnvironmentConfigError) as e:
         TestEnvironmentConfigMatter()