Add python tests model classes

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

@@ -13,10 +13,91 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+from enum import Enum
 from queue import Empty, Queue
-from typing import Any, Union
+from typing import Any, Literal, Optional, Union
 
 from matter_yamltests.hooks import TestRunnerHooks
+from pydantic import BaseModel
+
+
+class SDKPythonTestResultEnum(str, Enum):
+    START = "start"
+    STOP = "stop"
+    TEST_START = "test_start"
+    TEST_STOP = "test_stop"
+    STEP_SKIPPED = "step_skipped"
+    STEP_START = "step_start"
+    STEP_SUCCESS = "step_success"
+    STEP_FAILURE = "step_failure"
+    STEP_UNKNOWN = "step_unknown"
+    STEP_MANUAL = "step_manual"
+
+
+class SDKPythonTestResultBase(BaseModel):
+    type: SDKPythonTestResultEnum
+
+    def params_dict(self) -> dict:
+        return self.dict(exclude={"type"})
+
+
+class SDKPythonTestResultStart(SDKPythonTestResultBase):
+    type: Literal[SDKPythonTestResultEnum.START]
+    count: int
+
+
+class SDKPythonTestResultStop(SDKPythonTestResultBase):
+    type: Literal[SDKPythonTestResultEnum.STOP]
+    duration: int
+
+
+class SDKPythonTestResultTestStart(SDKPythonTestResultBase):
+    type: Literal[SDKPythonTestResultEnum.TEST_START]
+    filename: Optional[str]
+    name: Optional[str]
+    count: Optional[int]
+
+
+class SDKPythonTestResultTestStop(SDKPythonTestResultBase):
+    type: Literal[SDKPythonTestResultEnum.TEST_STOP]
+    duration: Optional[str]
+    exception: Any
+
+
+class SDKPythonTestResultStepSkipped(SDKPythonTestResultBase):
+    type: Literal[SDKPythonTestResultEnum.STEP_SKIPPED]
+    name: Optional[str]
+    expression: Optional[str]
+
+
+class SDKPythonTestResultStepStart(SDKPythonTestResultBase):
+    type: Literal[SDKPythonTestResultEnum.STEP_START]
+    name: Optional[str]
+
+
+class SDKPythonTestResultStepSuccess(SDKPythonTestResultBase):
+    type: Literal[SDKPythonTestResultEnum.STEP_SUCCESS]
+    logger: Any
+    logs: Any
+    duration: int
+    request: Any
+
+
+class SDKPythonTestResultStepFailure(SDKPythonTestResultBase):
+    type: Literal[SDKPythonTestResultEnum.STEP_FAILURE]
+    logger: Any
+    logs: Any
+    duration: int
+    request: Any
+    received: Any
+
+
+class SDKPythonTestResultStepUnknown(SDKPythonTestResultBase):
+    type: Literal[SDKPythonTestResultEnum.STEP_UNKNOWN]
+
+
+class SDKPythonTestResultStepManual(SDKPythonTestResultBase):
+    type: Literal[SDKPythonTestResultEnum.STEP_MANUAL]
 
 
 class SDKPythonTestRunnerHooks(TestRunnerHooks):
@@ -38,58 +119,85 @@ def is_finished(self) -> bool:
         return SDKPythonTestRunnerHooks.finished
 
     def start(self, count: int) -> None:
-        self.results.put({"start": {"count": count}})
+        self.results.put(
+            SDKPythonTestResultStart(type=SDKPythonTestResultEnum.START, count=count)
+        )
 
     def stop(self, duration: int) -> None:
-        self.results.put({"stop": {"duration": duration}})
+        self.results.put(
+            SDKPythonTestResultStop(
+                type=SDKPythonTestResultEnum.STOP, duration=duration
+            )
+        )
         SDKPythonTestRunnerHooks.finished = True
 
     def test_start(self, filename: str, name: str, count: int) -> None:
         self.results.put(
-            {"test_start": {"filename": filename, "name": name, "count": count}}
+            SDKPythonTestResultTestStart(
+                type=SDKPythonTestResultEnum.TEST_START,
+                filename=filename,
+                name=name,
+                count=count,
+            )
         )
 
     def test_stop(self, exception: Exception, duration: int) -> None:
-        self.results.put({"test_stop": {"exception": exception, "duration": duration}})
+        self.results.put(
+            SDKPythonTestResultTestStop(
+                type=SDKPythonTestResultEnum.TEST_STOP,
+                exception=exception,
+                duration=duration,
+            )
+        )
 
-    def step_skipped(self, name: str, expression: str) -> None:
-        self.results.put({"step_skipped": {"name": name, "expression": expression}})
+    def SDKPythonTestResultStepSkipped(self, name: str, expression: str) -> None:
+        self.results.put(
+            SDKPythonTestResultStepSkipped(
+                type=SDKPythonTestResultEnum.STEP_SKIPPED, expression=expression
+            )
+        )
 
     def step_start(self, name: str) -> None:
-        self.results.put({"step_start": {"name": name}})
+        self.results.put(
+            SDKPythonTestResultStepStart(
+                type=SDKPythonTestResultEnum.STEP_START, name=name
+            )
+        )
 
     def step_success(self, logger: Any, logs: Any, duration: int, request: Any) -> None:
         self.results.put(
-            {
-                "step_success": {
-                    "logger": logger,
-                    "logs": logs,
-                    "duration": duration,
-                    "request": request,
-                }
-            }
+            SDKPythonTestResultStepSuccess(
+                type=SDKPythonTestResultEnum.STEP_SUCCESS,
+                logger=logger,
+                logs=logs,
+                duration=duration,
+                request=request,
+            )
         )
 
     def step_failure(
         self, logger: Any, logs: Any, duration: int, request: Any, received: Any
     ) -> None:
         self.results.put(
-            {
-                "step_failure": {
-                    "logger": logger,
-                    "logs": logs,
-                    "duration": duration,
-                    "request": request,
-                    "received": received,
-                }
-            }
+            SDKPythonTestResultStepFailure(
+                type=SDKPythonTestResultEnum.STEP_FAILURE,
+                logger=logger,
+                logs=logs,
+                duration=duration,
+                request=request,
+                received=received,
+            )
         )
 
     def step_unknown(self) -> None:
-        self.results.put({"step_unknown": {}})
+        self.results.put(
+            SDKPythonTestResultStepUnknown(type=SDKPythonTestResultEnum.STEP_UNKNOWN)
+        )
 
     def step_manual(self) -> None:
-        self.results.put({"step_manual": {}})
+        self.results.put(
+            SDKPythonTestResultStepManual(type=SDKPythonTestResultEnum.STEP_MANUAL)
+        )
 
     def step_start_list(self) -> None:
         pass

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

@@ -24,7 +24,10 @@
 from app.test_engine.models import TestCase, TestStep
 
 from .python_test_models import PythonTest
-from .python_testing_hooks_proxy import SDKPythonTestRunnerHooks
+from .python_testing_hooks_proxy import (
+    SDKPythonTestResultBase,
+    SDKPythonTestRunnerHooks,
+)
 
 # Custom type variable used to annotate the factory method in PythonTestCase.
 T = TypeVar("T", bound="PythonTestCase")
@@ -83,6 +86,7 @@ def step_failure(
         self, logger: Any, logs: str, duration: int, request: Any, received: Any
     ) -> None:
         # TODO Handle Logs properly
+        self.mark_step_failure("Python test step failure")
         self.next_step()
 
     def step_unknown(self) -> None:
@@ -178,14 +182,13 @@ async def execute(self) -> None:
         finally:
             pass
 
-    def __handle_update(self, update: dict) -> None:
-        for func_name, kwargs in update.items():
-            self.__call_function_from_name(self, func_name, kwargs)
+    def __handle_update(self, update: SDKPythonTestResultBase) -> None:
+        self.__call_function_from_name(update.type.value, update.params_dict())
 
-    def __call_function_from_name(self, obj, func_name, kwargs) -> None:  # type: ignore
-        func = getattr(obj, func_name, None)
+    def __call_function_from_name(self, func_name: str, kwargs: Any) -> None:
+        func = getattr(self, func_name, None)
         if not func:
-            raise AttributeError(f"{func_name} is not a method of {obj}")
+            raise AttributeError(f"{func_name} is not a method of {self}")
         if not callable(func):
             raise TypeError(f"{func_name} is not callable")
         func(**kwargs)