Show python logs at the end of the test execution

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

@@ -16,7 +16,7 @@
 import re
 from asyncio import sleep
 from multiprocessing.managers import BaseManager
-from typing import Any, Type, TypeVar
+from typing import Any, Generator, Type, TypeVar, cast
 
 from app.models import TestCaseExecution
 from app.test_engine.logger import test_engine_logger as logger
@@ -31,7 +31,12 @@
     SDKPythonTestResultBase,
     SDKPythonTestRunnerHooks,
 )
-from .utils import EXECUTABLE, RUNNER_CLASS_PATH, generate_command_arguments
+from .utils import (
+    EXECUTABLE,
+    RUNNER_CLASS_PATH,
+    generate_command_arguments,
+    handle_logs,
+)
 
 # Custom type variable used to annotate the factory method in PythonTestCase.
 T = TypeVar("T", bound="PythonTestCase")
@@ -56,14 +61,16 @@ def __init__(self, test_case_execution: TestCaseExecution) -> None:
         self.__runned = 0
         self.test_stop_called = False
 
-    def next_step(self) -> None:
-        # Python tests that don't follow the template only have the default step "Start
-        # Python test", but inside the file there can be more than one test case, so the
-        # hooks steps methods will continue to be called
-        if len(self.test_steps) == 1:
+    # Move to the next step if the test case has additional steps apart from the 2
+    # deafult ones
+    def step_over(self) -> None:
+        # Python tests that don't follow the template only have the default steps "Start
+        # Python test" and "Show test logs", but inside the file there can be more than
+        # one test case, so the hooks' step methods will continue to be called
+        if len(self.test_steps) == 2:
             return
 
-        super().next_step()
+        self.next_step()
 
     def start(self, count: int) -> None:
         pass
@@ -73,29 +80,28 @@ def stop(self, duration: int) -> None:
             self.current_test_step.mark_as_completed()
 
     def test_start(self, filename: str, name: str, count: int) -> None:
-        self.next_step()
+        self.step_over()
 
     def test_stop(self, exception: Exception, duration: int) -> None:
         self.test_stop_called = True
-        self.current_test_step.mark_as_completed()
 
     def step_skipped(self, name: str, expression: str) -> None:
         self.current_test_step.mark_as_not_applicable("Test step skipped")
-        self.next_step()
+        self.step_over()
 
     def step_start(self, name: str) -> None:
         pass
 
     def step_success(self, logger: Any, logs: str, duration: int, request: Any) -> None:
         # TODO Handle Logs properly
-        self.next_step()
+        self.step_over()
 
     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()
+        self.step_over()
 
     def step_unknown(self) -> None:
         self.__runned += 1
@@ -172,13 +178,12 @@ async def execute(self) -> None:
             if self.chip_tool.pics_file_created:
                 command.append(f" --PICS {PICS_FILE_PATH}")
 
-            # TODO Ignoring stream from docker execution
-            self.chip_tool.send_command(
+            exec_result = self.chip_tool.send_command(
                 command,
                 prefix=EXECUTABLE,
                 is_stream=True,
                 is_socket=False,
-            ).output
+            )
 
             while ((update := test_runner_hooks.update_test()) is not None) or (
                 not test_runner_hooks.is_finished()
@@ -189,6 +194,21 @@ async def execute(self) -> None:
 
                 self.__handle_update(update)
 
+            # Step: Show test logs
+
+            # Python tests that don't follow the template only have the 2 default steps
+            # and, at this point, will still be in the first step because of the
+            # step_over method. So we have to explicitly move on to the next step here.
+            # The tests that do follow the template will have additional steps and will
+            # have already been moved to the correct step by the hooks' step methods.
+            if len(self.test_steps) == 2:
+                self.next_step()
+
+            logger.info("---- Start of Python test logs ----")
+            handle_logs(cast(Generator, exec_result.output), logger)
+            logger.info("---- End of Python test logs ----")
+
+            self.current_test_step.mark_as_completed()
         finally:
             pass
 
@@ -208,3 +228,4 @@ def create_test_steps(self) -> None:
         for step in self.python_test.steps:
             python_test_step = TestStep(step.label)
             self.test_steps.append(python_test_step)
+        self.test_steps.append(TestStep("Show test logs"))