From 890cf2e1cfba4c2bbbd3e39daf351f858b8d4b08 Mon Sep 17 00:00:00 2001 From: Romulo Quidute Filho Date: Wed, 6 Dec 2023 16:54:40 +0000 Subject: [PATCH] Added SDKPythonTestResult class in order to handle the python test responses --- .../models/python_testing_hooks_proxy.py | 100 +++++++++++++----- .../python_testing/models/test_case.py | 31 +++--- 2 files changed, 93 insertions(+), 38 deletions(-) diff --git a/test_collections/sdk_tests/support/python_testing/models/python_testing_hooks_proxy.py b/test_collections/sdk_tests/support/python_testing/models/python_testing_hooks_proxy.py index a418d5cd..f8c0a6ec 100644 --- a/test_collections/sdk_tests/support/python_testing/models/python_testing_hooks_proxy.py +++ b/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 diff --git a/test_collections/sdk_tests/support/python_testing/models/test_case.py b/test_collections/sdk_tests/support/python_testing/models/test_case.py index 0883e4b2..9f7490bd 100644 --- a/test_collections/sdk_tests/support/python_testing/models/test_case.py +++ b/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)