Feature/785 make log synchronization configurable

.version_information

@@ -1 +1 @@
-v2.14-beta3+fall2025
+v2.15-beta1+winter2025

app/core/config.py

@@ -107,6 +107,9 @@ def get_emails_enabled(cls, v: bool, values: Dict[str, Any]) -> bool:
     NOTIFIER_TO: str = "xiole.chip.test@gmail.com"
     NOTIFIER_SUBJECT: str = "CHIP Tool Crash Log"
 
+    # Python Test Logging
+    ENABLE_REALTIME_PYTHON_TEST_LOGS: bool = False
+
     class Config:
         case_sensitive = True
 

test_collections/matter/config.py

@@ -23,9 +23,9 @@ class MatterSettings(BaseSettings):
 
     # SDK Docker Image
     SDK_DOCKER_IMAGE: str = "connectedhomeip/chip-cert-bins"
-    SDK_DOCKER_TAG: str = "f902839abf1de0d17956de34889b6ad997e2c5e4"
+    SDK_DOCKER_TAG: str = "ca9d1118e097fe947b2aec1ba84f265d6cf2447e"
     # SDK SHA: used to fetch tests (YAML and Python) from SDK.
-    SDK_SHA: str = "a95f163c4d527b7a793fe4f89e55af331f40b87a"
+    SDK_SHA: str = "ca9d1118e097fe947b2aec1ba84f265d6cf2447e"
 
     class Config:
         case_sensitive = True

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

@@ -21,6 +21,7 @@
 from socket import SocketIO
 from typing import Any, Optional, Type, TypeVar
 
+from app.core.config import settings
 from app.models import TestCaseExecution
 from app.test_engine.logger import PYTHON_TEST_LEVEL
 from app.test_engine.logger import test_engine_logger as logger
@@ -142,8 +143,9 @@ def step_start(self, name: str) -> None:
     async def step_success(
         self, logger: Any, logs: str, duration: int, request: Any
     ) -> None:
-        # Display logs captured during this step
-        await self._display_step_logs()
+        # Display logs captured during this step only if real-time logging is enabled
+        if settings.ENABLE_REALTIME_PYTHON_TEST_LOGS:
+            await self._display_step_logs()
 
     async def _display_step_logs(self) -> None:
         """Display logs that were captured during the current step."""
@@ -261,7 +263,9 @@ async def step_failure(
         self, logger: Any, logs: str, duration: int, request: Any, received: Any
     ) -> None:
         # Display logs captured during this step before marking as failure
-        await self._display_step_logs()
+        # only if real-time logging is enabled
+        if settings.ENABLE_REALTIME_PYTHON_TEST_LOGS:
+            await self._display_step_logs()
 
         failure_msg = "Python test step failure"
         if logs:
@@ -450,7 +454,12 @@ async def setup(self) -> None:
     async def cleanup(self) -> None:
         logger.info("Test Cleanup")
         # Log any remaining content that wasn't captured by steps
-        await self._log_remaining_content()
+        # only if real-time logging is enabled
+        if settings.ENABLE_REALTIME_PYTHON_TEST_LOGS:
+            await self._log_remaining_content()
+        else:
+            # Use batch logging when real-time logging is disabled
+            self.display_batch_logs()
 
     async def _log_remaining_content(self) -> None:
         """Log any content from the test output file that wasn't logged yet."""
@@ -493,6 +502,38 @@ async def _log_remaining_content(self) -> None:
                 f"Unexpected error while logging remaining content: {e}", exc_info=True
             )
 
+    def display_batch_logs(self) -> None:
+        """Batch logging method for when real-time logging is disabled.
+
+        This method logs all test output at once after test execution completes,
+        rather than displaying logs incrementally as each step executes.
+        """
+        # Check idempotency flag to prevent duplicate logging
+        if getattr(self, "_batch_logs_displayed", False):
+            return
+
+        # Validate file path is set
+        if not self.file_output_path:
+            logger.debug("Test output file not found, skipping log display")
+            return
+
+        if not self.file_output_path.exists():
+            logger.debug(f"Test output file does not exist: {self.file_output_path}")
+            return
+
+        try:
+            logger.info("---- Start of Python test logs ----")
+            with open(self.file_output_path, "r", encoding="utf-8") as f:
+                for line in f:
+                    logger.log(PYTHON_TEST_LEVEL, line.rstrip("\n"))
+            logger.info("---- End of Python test logs ----")
+        except (IOError, OSError) as e:
+            logger.warning(f"Failed to read test output file: {e}")
+        except Exception as e:
+            logger.error(f"Unexpected error while reading logs: {e}", exc_info=True)
+
+        setattr(self, "_batch_logs_displayed", True)
+
     async def execute(self) -> None:
         try:
             logger.info("Running Python Test: " + self.python_test.name)
@@ -567,7 +608,12 @@ async def execute(self) -> None:
                 self.skip_to_last_step()
 
             # Check for any remaining logs that weren't captured by steps
-            await self._log_remaining_content()
+            # or show all logs if real-time logging is disabled
+            if settings.ENABLE_REALTIME_PYTHON_TEST_LOGS:
+                await self._log_remaining_content()
+            else:
+                # Use batch logging when real-time logging is disabled
+                self.display_batch_logs()
 
             self.current_test_step.mark_as_completed()
         finally: