Added SDKPythonTestResult class in order to handle the python test re…

…sponses

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

@@ -13,10 +13,39 @@
 # 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, 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 SDKPythonTestResult(BaseModel):
+    type: SDKPythonTestResultEnum
+    count: Optional[int]
+    duration: Optional[str]
+    name: Optional[str]
+    filename: Optional[str]
+    exception: Any
+    expression: Optional[str]
+    logger: Any
+    logs: Any
+    request: Any
+    received: Any
 
 
 class SDKPythonTestRunnerHooks(TestRunnerHooks):
@@ -38,58 +67,77 @@ def is_finished(self) -> bool:
         return SDKPythonTestRunnerHooks.finished
 
     def start(self, count: int) -> None:
-        self.results.put({"start": {"count": count}})
+        self.results.put(
+            SDKPythonTestResult(type=SDKPythonTestResultEnum.START, count=count)
+        )
 
     def stop(self, duration: int) -> None:
-        self.results.put({"stop": {"duration": duration}})
+        self.results.put(
+            SDKPythonTestResult(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}}
+            SDKPythonTestResult(
+                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(
+            SDKPythonTestResult(
+                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}})
+        self.results.put(
+            SDKPythonTestResult(
+                type=SDKPythonTestResultEnum.STEP_SKIPPED, expression=expression
+            )
+        )
 
     def step_start(self, name: str) -> None:
-        self.results.put({"step_start": {"name": name}})
+        self.results.put(
+            SDKPythonTestResult(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,
-                }
-            }
+            SDKPythonTestResult(
+                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,
-                }
-            }
+            SDKPythonTestResult(
+                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(SDKPythonTestResult(type=SDKPythonTestResultEnum.STEP_UNKNOWN))
 
     def step_manual(self) -> None:
-        self.results.put({"step_manual": {}})
+        self.results.put(SDKPythonTestResult(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,7 @@
 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 SDKPythonTestResult, SDKPythonTestRunnerHooks
 
 # Custom type variable used to annotate the factory method in PythonTestCase.
 T = TypeVar("T", bound="PythonTestCase")
@@ -54,33 +54,41 @@ def __init__(self, test_case_execution: TestCaseExecution) -> None:
         self.__runned = 0
         self.test_stop_called = False
 
-    def start(self, count: int) -> None:
+    def start(self, count: int, **kwargs: Any) -> None:
         pass
 
-    def stop(self, duration: int) -> None:
+    def stop(self, duration: int, **kwargs: Any) -> None:
         if not self.test_stop_called:
             self.current_test_step.mark_as_completed()
 
-    def test_start(self, filename: str, name: str, count: int) -> None:
+    def test_start(self, filename: str, name: str, count: int, **kwargs: Any) -> None:
         self.next_step()
 
-    def test_stop(self, exception: Exception, duration: int) -> None:
+    def test_stop(self, exception: Exception, duration: int, **kwargs: Any) -> None:
         self.test_stop_called = True
         self.current_test_step.mark_as_completed()
 
-    def step_skipped(self, name: str, expression: str) -> None:
+    def step_skipped(self, name: str, expression: str, **kwargs: Any) -> None:
         self.current_test_step.mark_as_not_applicable("Test step skipped")
         self.next_step()
 
-    def step_start(self, name: str) -> None:
+    def step_start(self, name: str, **kwargs: Any) -> None:
         pass
 
-    def step_success(self, logger: Any, logs: str, duration: int, request: Any) -> None:
+    def step_success(
+        self, logger: Any, logs: str, duration: int, request: Any, **kwargs: Any
+    ) -> None:
         # TODO Handle Logs properly
         self.next_step()
 
     def step_failure(
-        self, logger: Any, logs: str, duration: int, request: Any, received: Any
+        self,
+        logger: Any,
+        logs: str,
+        duration: int,
+        request: Any,
+        received: Any,
+        **kwargs: Any,
     ) -> None:
         # TODO Handle Logs properly
         self.next_step()
@@ -178,9 +186,8 @@ 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: SDKPythonTestResult) -> None:
+        self.__call_function_from_name(self, update.type.value, update.__dict__)
 
     def __call_function_from_name(self, obj, func_name, kwargs) -> None:  # type: ignore
         func = getattr(obj, func_name, None)