Resolve issue with UserPrompts in test harness. (#26961)

* Resolve issue with UserPrompts in test harness.

* Update UserPrompt to match legacy behavior if not expected value is set in YAML

* Update documentation and typing for request parameter.

* Add documentation of UserPrompt logging in step_start

* Make reference link a permalink

scripts/py_matter_yamltests/matter_yamltests/hooks.py

@@ -14,6 +14,7 @@
 #    limitations under the License.
 
 from .errors import TestStepError
+from .parser import TestStep
 
 
 class TestParserHooks():
@@ -143,18 +144,18 @@ def step_skipped(self, name: str, expression: str):
         """
         pass
 
-    def step_start(self, name: str):
+    def step_start(self, request: TestStep):
         """
         This method is called when the runner starts running a step from the test.
 
         Parameters
         ----------
-        name: str
-            The name of the test step that is starting.
+        request: TestStep
+            The original request as defined by the test step.
         """
         pass
 
-    def step_success(self, logger, logs, duration: int, request):
+    def step_success(self, logger, logs, duration: int, request: TestStep):
         """
         This method is called when running a step succeeds.
 
@@ -169,12 +170,12 @@ def step_success(self, logger, logs, duration: int, request):
         duration: int
             How long it took to run the test step, in milliseconds.
 
-        request:
+        request: TestStep
             The original request as defined by the test step.
         """
         pass
 
-    def step_failure(self, logger, logs, duration: int, request, received):
+    def step_failure(self, logger, logs, duration: int, request: TestStep, received):
         """
         This method is called when running a step fails.
 
@@ -189,7 +190,7 @@ def step_failure(self, logger, logs, duration: int, request, received):
         duration: int
             How long it took to run the test step, in milliseconds.
 
-        request:
+        request: TestStep
             The original request as defined by the test step.
 
         received:

scripts/py_matter_yamltests/matter_yamltests/parser_config.py

@@ -29,7 +29,7 @@ def get_config(test_file: str):
         config_options = {}
 
         yaml_loader = YamlLoader()
-        _, _, config, _ = yaml_loader.load(test_file)
+        _, _, _, config, _ = yaml_loader.load(test_file)
         config_options = {key: value if not isinstance(
             value, dict) else value['defaultValue'] for key, value in config.items()}
         return config_options

scripts/py_matter_yamltests/matter_yamltests/pseudo_clusters/clusters/log_commands.py

@@ -39,7 +39,17 @@ class LogCommands(PseudoCluster):
     definition = _DEFINITION
 
     async def UserPrompt(self, request):
-        pass
+        expected_value = None
+        for value in request.arguments.get("values", []):
+            if value.get('name') and 'expectedValue' in value['name']:
+                expected_value = value['value']
+                request.responses = [{"values": [{"name": "expectedValue", "value": expected_value}]}]
+
+        if expected_value is not None:
+            input_result = input("")
+            return {"value": {"expectedValue": input_result}}
+
+        return {}
 
     async def Log(self, request):
         pass

scripts/py_matter_yamltests/matter_yamltests/runner.py

@@ -180,11 +180,11 @@ async def _run(self, parser: TestParser, config: TestRunnerConfig):
                     hooks.step_skipped(request.label, request.pics)
                     continue
                 elif not config.adapter:
-                    hooks.step_start(request.label)
+                    hooks.step_start(request)
                     hooks.step_unknown()
                     continue
                 else:
-                    hooks.step_start(request.label)
+                    hooks.step_start(request)
 
                 start = time.time()
                 if config.pseudo_clusters.supports(request):

scripts/tests/yaml/tests_logger.py

@@ -23,6 +23,7 @@
 import click
 from matter_yamltests.errors import TestStepError, TestStepKeyError
 from matter_yamltests.hooks import TestParserHooks, TestRunnerHooks, WebSocketRunnerHooks
+from matter_yamltests.parser import TestStep
 
 
 def _strikethrough(str):
@@ -193,11 +194,16 @@ def step_skipped(self, name: str, expression: str):
         self.__index += 1
         self.__skipped += 1
 
-    def step_start(self, name: str):
+    def step_start(self, request: TestStep):
         if self.__use_test_harness_log_format:
-            print(self.__strings.test_harness_step_start.format(index=self.__index, name=name))
-
-        print(self.__strings.step_start.format(index=self.__index, name=click.style(name, bold=True)), end='')
+            print(self.__strings.test_harness_step_start.format(index=self.__index, name=request.label))
+
+        print(self.__strings.step_start.format(index=self.__index, name=click.style(request.label, bold=True)), end='')
+        # This is to keep previous behavior of UserPrompt. Where it logs USER_PROMPT prior to user input. See link below:
+        # https://github.com/project-chip/connectedhomeip/blob/6644a4b0b0d1272ae325c651b27bd0e7068f3a8a/src/app/tests/suites/commands/log/LogCommands.cpp#L31
+        if request.command == 'UserPrompt':
+            message = request.arguments['values'][0]['value']
+            print("\n" + self.__strings.user_prompt.format(message=f'{message}'))
         # flushing stdout such that the previous print statement is visible on the screen for long running tasks.
         sys.stdout.flush()
 
@@ -208,7 +214,7 @@ def step_unknown(self):
 
         self.__runned += 1
 
-    def step_success(self, logger, logs, duration: int, request):
+    def step_success(self, logger, logs, duration: int, request: TestStep):
         print(self.__strings.step_result.format(state=_SUCCESS, duration=duration))
 
         self.__print_results(logger)
@@ -218,9 +224,6 @@ def step_success(self, logger, logs, duration: int, request):
             elif request.command == 'Log':
                 message = request.arguments['values'][0]['value']
                 print(self.__strings.log.format(message=f'{message}'))
-            elif request.command == 'UserPrompt':
-                message = request.arguments['values'][0]['value']
-                print(self.__strings.user_prompt.format(message=f'{message}'))
 
         if self.__show_adapter_logs:
             self.__log_printer.print(logs)
@@ -230,7 +233,7 @@ def step_success(self, logger, logs, duration: int, request):
         self.__errors += logger.errors
         self.__runned += 1
 
-    def step_failure(self, logger, logs, duration: int, request, received):
+    def step_failure(self, logger, logs, duration: int, request: TestStep, received):
         print(self.__strings.step_result.format(state=_FAILURE, duration=duration))
 
         self.__print_results(logger)