[Feature] Updating Step Start Hook to Receive Endpoint

app/test_engine/models/manual_test_case.py

@@ -193,7 +193,7 @@ async def execute(self) -> None:
             if isinstance(step, ManualVerificationTestStep):
                 await step.prompt_verification_step()
             else:
-                logger.error(f"Unsupported test test {step.__class__}")
+                logger.error(f"Unsupported test step {step.__class__}")
             self.next_step()
 
         if isinstance(self.current_test_step, ManualLogUploadStep):

app/test_engine/models/test_step.py

@@ -29,11 +29,17 @@ class TestStep(TestObservable):
 
     __test__ = False  # Needed to indicate to PyTest that this is not a "test"
 
-    def __init__(self, name: str, state: TestStateEnum = TestStateEnum.PENDING) -> None:
+    def __init__(
+        self,
+        name: str,
+        endpoint: int | None = None,
+        state: TestStateEnum = TestStateEnum.PENDING,
+    ) -> None:
         super().__init__()
         self.errors: List[str] = []
         self.failures: List[str] = []
         self.name = name
+        self.endpoint = endpoint
         self.__state = state
         self.test_step_execution: Optional[TestStepExecution] = None
 
@@ -90,7 +96,12 @@ def mark_as_completed(self) -> None:
         else:
             self.state = TestStateEnum.PASSED
 
-        logger.info(f"Test Step Completed [{self.state.name}]: {self.name}")
+        endpoint_info = (
+            f" for endpoint {self.endpoint}" if self.endpoint is not None else ""
+        )
+        logger.info(
+            f"Test Step Completed [{self.state.name}]: {self.name}{endpoint_info}"
+        )
         self.__print_log_separator()
 
     def __print_log_separator(self) -> None:

app/test_engine/test_ui_observer.py

@@ -119,6 +119,7 @@ def __onTestStepUpdate(self, observable: TestStep) -> None:
                 "test_suite_execution_index": test_suite_execution.execution_index,
                 "test_case_execution_index": test_case_execution.execution_index,
                 "test_step_execution_index": test_step_execution.execution_index,
+                "endpoint": observable.endpoint,
                 "state": observable.state,
                 "errors": observable.errors,
                 "failures": observable.failures,

app/tests/test_engine/test_ui_observer.py

@@ -147,7 +147,7 @@ def __expected_test_case_dict(index: int, suite_index: int) -> Dict[str, Any]:
 
 
 def __expected_test_step_dict(
-    index: int, case_index: int, suite_index: int
+    index: int, case_index: int, suite_index: int, endpoint: int | None = None
 ) -> Dict[str, Any]:
     return {
         MessageKeysEnum.TYPE: MessageTypeEnum.TEST_UPDATE,
@@ -157,6 +157,7 @@ def __expected_test_step_dict(
                 "test_suite_execution_index": suite_index,
                 "test_case_execution_index": case_index,
                 "test_step_execution_index": index,
+                "endpoint": endpoint,
                 "state": TestStateEnum.EXECUTING,
                 "errors": [],
                 "failures": [],

test_collections/matter/sdk_tests/support/models/matter_test_models.py

@@ -33,6 +33,10 @@ class MatterTestType(Enum):
 
 class MatterTestStep(BaseModel):
     label: str
+    # Pydantic will fail parsing YAML files since they are using endpoints as variables
+    # instead of numbers only.
+    # So endpoint has to be a String as well to pass the file parsing.
+    endpoint: Optional[int | str] = None
     PICS: Optional[str] = None
     verification: Optional[str] = None
     command: Optional[str]

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

@@ -288,18 +288,20 @@ def __retrieve_steps(method: FunctionDefType) -> List[MatterTestStep]:
 
             step_name = step.args[ARG_STEP_DESCRIPTION_INDEX].value
             parsed_step_name = step_name
+            python_steps.append(
+                MatterTestStep(
+                    label=parsed_step_name,
+                    command=None,
+                    arguments=None,
+                    is_commissioning=arg_is_commissioning,
+                )
+            )
         except Exception as e:
             logger.warning(
                 f"Failed parsing step name from {method.name}, Error:{str(e)}"
             )
             parsed_step_name = "UNABLE TO PARSE TEST STEP NAME"
 
-        python_steps.append(
-            MatterTestStep(
-                label=parsed_step_name, is_commissioning=arg_is_commissioning
-            )
-        )
-
     return python_steps
 
 

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

@@ -44,56 +44,57 @@ def params_dict(self) -> dict:
 
 
 class SDKPythonTestResultStart(SDKPythonTestResultBase):
-    type = SDKPythonTestResultEnum.START
+    type: SDKPythonTestResultEnum = SDKPythonTestResultEnum.START
     count: int
 
 
 class SDKPythonTestResultStop(SDKPythonTestResultBase):
-    type = SDKPythonTestResultEnum.STOP
+    type: SDKPythonTestResultEnum = SDKPythonTestResultEnum.STOP
     duration: int
 
 
 class SDKPythonTestResultTestStart(SDKPythonTestResultBase):
-    type = SDKPythonTestResultEnum.TEST_START
+    type: SDKPythonTestResultEnum = SDKPythonTestResultEnum.TEST_START
     filename: Optional[str]
     name: Optional[str]
     count: Optional[int]
     steps: Optional[list[str]]
 
 
 class SDKPythonTestResultTestStop(SDKPythonTestResultBase):
-    type = SDKPythonTestResultEnum.TEST_STOP
+    type: SDKPythonTestResultEnum = SDKPythonTestResultEnum.TEST_STOP
     duration: Optional[int]
     exception: Any
 
 
 class SDKPythonTestResultTestSkipped(SDKPythonTestResultBase):
-    type = SDKPythonTestResultEnum.TEST_SKIPPED
+    type: SDKPythonTestResultEnum = SDKPythonTestResultEnum.TEST_SKIPPED
     filename: Optional[str]
     name: Optional[str]
 
 
 class SDKPythonTestResultStepSkipped(SDKPythonTestResultBase):
-    type = SDKPythonTestResultEnum.STEP_SKIPPED
+    type: SDKPythonTestResultEnum = SDKPythonTestResultEnum.STEP_SKIPPED
     name: Optional[str]
     expression: Optional[str]
 
 
 class SDKPythonTestResultStepStart(SDKPythonTestResultBase):
-    type = SDKPythonTestResultEnum.STEP_START
+    type: SDKPythonTestResultEnum = SDKPythonTestResultEnum.STEP_START
     name: Optional[str]
+    endpoint: Optional[int]
 
 
 class SDKPythonTestResultStepSuccess(SDKPythonTestResultBase):
-    type = SDKPythonTestResultEnum.STEP_SUCCESS
+    type: SDKPythonTestResultEnum = SDKPythonTestResultEnum.STEP_SUCCESS
     logger: Any
     logs: Any
     duration: int
     request: Any
 
 
 class SDKPythonTestResultStepFailure(SDKPythonTestResultBase):
-    type = SDKPythonTestResultEnum.STEP_FAILURE
+    type: SDKPythonTestResultEnum = SDKPythonTestResultEnum.STEP_FAILURE
     logger: Any
     logs: Any
     duration: int
@@ -102,15 +103,15 @@ class SDKPythonTestResultStepFailure(SDKPythonTestResultBase):
 
 
 class SDKPythonTestResultStepUnknown(SDKPythonTestResultBase):
-    type = SDKPythonTestResultEnum.STEP_UNKNOWN
+    type: SDKPythonTestResultEnum = SDKPythonTestResultEnum.STEP_UNKNOWN
 
 
 class SDKPythonTestResultStepManual(SDKPythonTestResultBase):
-    type = SDKPythonTestResultEnum.STEP_MANUAL
+    type: SDKPythonTestResultEnum = SDKPythonTestResultEnum.STEP_MANUAL
 
 
 class SDKPythonTestResultShowPrompt(SDKPythonTestResultBase):
-    type = SDKPythonTestResultEnum.SHOW_PROMPT
+    type: SDKPythonTestResultEnum = SDKPythonTestResultEnum.SHOW_PROMPT
     msg: str
     placeholder: Optional[str]
     default_value: Optional[str]
@@ -159,10 +160,12 @@ def test_skipped(self, filename: str, name: str) -> None:
         self.results.put(SDKPythonTestResultTestSkipped(filename=filename, name=name))
 
     def step_skipped(self, name: str, expression: str) -> None:
-        self.results.put(SDKPythonTestResultStepSkipped(expression=expression))
+        self.results.put(
+            SDKPythonTestResultStepSkipped(name=None, expression=expression)
+        )
 
-    def step_start(self, name: str) -> None:
-        self.results.put(SDKPythonTestResultStepStart(name=name))
+    def step_start(self, name: str, endpoint: Optional[int] = None) -> None:
+        self.results.put(SDKPythonTestResultStepStart(name=name, endpoint=endpoint))
 
     def step_success(self, logger: Any, logs: Any, duration: int, request: Any) -> None:
         self.results.put(
@@ -198,6 +201,7 @@ def show_prompt(
         msg: str,
         placeholder: Optional[str] = None,
         default_value: Optional[str] = None,
+        endpoint_id: Optional[int] = None,
     ) -> None:
         self.results.put(
             SDKPythonTestResultShowPrompt(

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

@@ -118,8 +118,9 @@ def step_skipped(self, name: str, expression: str) -> None:
         else:
             self.current_test_step.mark_as_not_applicable(skiped_msg)
 
-    def step_start(self, name: str) -> None:
+    def step_start(self, name: str, endpoint: int | None = None) -> None:
         self.step_over()
+        self.current_test_step.endpoint = endpoint
 
     def step_success(self, logger: Any, logs: str, duration: int, request: Any) -> None:
         pass

test_collections/matter/sdk_tests/support/yaml_tests/models/chip_test.py

@@ -112,7 +112,7 @@ def test_start(
         # since there is step execute outside runner context
         self.next_step()
 
-    def test_stop(self, duration: int) -> None:
+    def test_stop(self, exception: Exception, duration: int) -> None:
         self.current_test_step.mark_as_completed()
 
     def step_skipped(self, name: str, expression: str) -> None:
@@ -127,7 +127,7 @@ def step_skipped(self, name: str, expression: str) -> None:
         self.__skipped += 1
         self.next_step()
 
-    def step_start(self, request: TestStep) -> None:
+    def step_start(self, request: TestStep, endpoint: int | None = None) -> None:
         if (
             self.server_type == ChipServerType.CHIP_APP
             and
@@ -140,6 +140,9 @@ def step_start(self, request: TestStep) -> None:
                 self.__prompt_user_for_controller_action(prompt), loop=loop
             )
         self.__index += 1
+        # The TestStep request object already has the endpoint information for YAML
+        # tests and it will be stored in the current test step observable below.
+        self.current_test_step.endpoint = request.endpoint
 
     def step_success(
         self, logger: Any, logs: Any, duration: int, request: TestStep
@@ -180,14 +183,15 @@ async def step_manual(self) -> None:
                 self.__prompt_user_manual_step(step), OUTCOME_TIMEOUT_S
             )
         except asyncio.TimeoutError:
-            self.current_test_step.append_failure("Prompt timed out.")
+            step.append_failure("Prompt timed out.")
         self.next_step()
 
     def show_prompt(
         self,
         msg: str,
         placeholder: Optional[str] = None,
         default_value: Optional[str] = None,
+        endpoint_id: Optional[int] = None,
     ) -> None:
         pass
 

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

@@ -170,7 +170,8 @@ def _append_automated_test_step(self, yaml_step: MatterTestStep) -> None:
             )
             return
 
-        step = TestStep(yaml_step.label)
+        endpoint = yaml_step.endpoint if isinstance(yaml_step.endpoint, int) else None
+        step = TestStep(yaml_step.label, endpoint=endpoint)
         if yaml_step.command == "UserPrompt":
             step = ManualVerificationTestStep(
                 name=yaml_step.label,