From ffba865ddab4ce46b61e3a32336a1431eef8b991 Mon Sep 17 00:00:00 2001 From: aamj Date: Thu, 6 Aug 2026 11:50:05 -0300 Subject: [PATCH 1/5] Max retained logs added for the CLI log viewer. Also, the websocket closure was postponed for when inactive and now yields to the event loop every 200 records intead of whole batch --- th_cli/test_run/log_viewer.html | 27 ++++++++++++-- th_cli/test_run/websocket.py | 66 ++++++++++++++++++++++++++------- 2 files changed, 75 insertions(+), 18 deletions(-) diff --git a/th_cli/test_run/log_viewer.html b/th_cli/test_run/log_viewer.html index 4e38ca6..64888fc 100644 --- a/th_cli/test_run/log_viewer.html +++ b/th_cli/test_run/log_viewer.html @@ -532,6 +532,12 @@ let lastBatchTime = 0; const BATCH_INTERVAL_MS = 50; const MAX_BATCH_SIZE = 50; + // Cap how many entries are kept in memory/rendered in the DOM. A run + // can produce hundreds of thousands of log lines; retaining all of + // them here (unbounded array + DOM nodes) is what freezes the tab. + // The "Download Logs" button reads the full file from disk directly, + // so it isn't affected by this cap. + const MAX_RETAINED_LOGS = 5000; function connectToLogStream() {{ const statusIndicator = document.getElementById('statusIndicator'); @@ -616,17 +622,30 @@ const fragment = document.createDocumentFragment(); const batchSize = Math.min(logBatchQueue.length, MAX_BATCH_SIZE); const batch = logBatchQueue.splice(0, batchSize); - + batch.forEach(entry => {{ allLogs.push(entry); logCount++; - + const logDiv = createLogElement(entry); fragment.appendChild(logDiv); }}); - + container.appendChild(fragment); - + + // Trim oldest entries once past the retention cap, so memory and + // DOM size stay bounded regardless of total run volume. + if (allLogs.length > MAX_RETAINED_LOGS) {{ + const excess = allLogs.length - MAX_RETAINED_LOGS; + allLogs.splice(0, excess); + for (let i = 0; i < excess; i++) {{ + const oldest = container.firstElementChild; + if (oldest) {{ + container.removeChild(oldest); + }} + }} + }} + document.getElementById('logCount').textContent = logCount; if (autoScroll) {{ diff --git a/th_cli/test_run/websocket.py b/th_cli/test_run/websocket.py index e60da91..ac993cd 100644 --- a/th_cli/test_run/websocket.py +++ b/th_cli/test_run/websocket.py @@ -13,6 +13,8 @@ # See the License for the specific language governing permissions and # limitations under the License. # +import asyncio + import click import websockets from loguru import logger @@ -56,6 +58,19 @@ WEBSOCKET_MAX_MESSAGE_SIZE = 32 * 1024 * 1024 # 32MB +# After the test run reaches a terminal state, the backend may still have a +# trailing batch of log records queued/in-flight (it flushes and broadcasts +# any pending log entries *after* sending the terminal state update - see +# TestLogHandler.finish()/TestUIObserver.complete_tasks() on the backend). +# Keep draining for a short grace period instead of closing immediately, so +# that trailing batch isn't dropped by a socket we already hung up on. +DRAIN_TIMEOUT_S = 5.0 + +# Yield to the event loop every N log records while processing one batch, so +# a very large batch doesn't block the websocket read loop for its entire +# duration. +LOG_RECORD_YIELD_INTERVAL = 200 + class TestRunSocket: def __init__( @@ -68,6 +83,7 @@ def __init__( self.project_config_dict = project_config_dict or {} self.two_way_talk_handler = two_way_talk_handler self._chip_server_info_displayed = False + self._run_finished = False # Track test step errors for logging # Key: (suite_index, case_index), Value: list of error strings from all steps self.test_case_step_errors: dict[tuple[int, int], list[str]] = {} @@ -85,9 +101,19 @@ async def connect_websocket(self) -> None: try: while True: try: - message = await socket.recv() + if self._run_finished: + # Drain any trailing messages for a short grace + # period instead of closing the instant the + # terminal state update arrives. + message = await asyncio.wait_for(socket.recv(), timeout=DRAIN_TIMEOUT_S) + else: + message = await socket.recv() except websockets.exceptions.ConnectionClosedOK: break + except asyncio.TimeoutError: + # No more trailing messages arrived during the + # drain grace period - safe to close now. + break # skip messages that are bytes, as we're expecting a string.\ if not isinstance(message, str): @@ -103,7 +129,13 @@ async def connect_websocket(self) -> None: click.echo(colorize_error(f"Received invalid socket message: {message}"), err=True) click.echo(colorize_error(e.json()), err=True) finally: - pass # Cleanup if needed + if self._run_finished: + try: + await socket.close() + except websockets.exceptions.ConnectionClosedError: + # Backend closed connection without completing handshake + # This is acceptable as test run completed successfully + pass except websockets.exceptions.ConnectionClosed: # Handle case where backend doesn't complete close handshake properly # This can happen with long-running test executions @@ -112,7 +144,7 @@ async def connect_websocket(self) -> None: async def __handle_incoming_socket_message(self, socket: WebSocketClientProtocol, message: SocketMessage) -> None: if isinstance(message.payload, TestUpdate): - await self.__handle_test_update(socket=socket, update=message.payload) + await self.__handle_test_update(update=message.payload) elif isinstance(message.payload, PromptRequest): # Debug: log the message type logger.debug(f"Received prompt with type: {message.type}") @@ -129,7 +161,7 @@ async def __handle_incoming_socket_message(self, socket: WebSocketClientProtocol two_way_talk_handler=self.two_way_talk_handler, ) elif message.type == MessageTypeEnum.TEST_LOG_RECORDS and isinstance(message.payload, list): - self.__handle_log_record(message.payload) + await self.__handle_log_record(message.payload) elif isinstance(message.payload, TimeOutNotification): # ignore time_out_notification as we handle timeout our selves pass @@ -139,7 +171,7 @@ async def __handle_incoming_socket_message(self, socket: WebSocketClientProtocol err=True, ) - async def __handle_test_update(self, socket: WebSocketClientProtocol, update: TestUpdate) -> None: + async def __handle_test_update(self, update: TestUpdate) -> None: if isinstance(update.body, TestStepUpdate): self.__log_test_step_update(update.body) elif isinstance(update.body, TestCaseUpdate): @@ -149,13 +181,12 @@ async def __handle_test_update(self, socket: WebSocketClientProtocol, update: Te elif isinstance(update.body, TestRunUpdate): await self.__log_test_run_update(update.body) if update.body.state != "executing": - # Test run ended disconnect. - try: - await socket.close() - except websockets.exceptions.ConnectionClosedError: - # Backend closed connection without completing handshake - # This is acceptable as test run completed successfully - pass + # Test run ended. Don't close immediately - the backend may + # still be flushing/broadcasting a trailing batch of log + # entries after this message; let the read loop keep + # draining for a short grace period (see DRAIN_TIMEOUT_S) + # before actually closing. + self._run_finished = True async def __log_test_run_update(self, update: TestRunUpdate) -> None: # Display CHIP server info when test run starts executing (SDK container already running) @@ -284,9 +315,16 @@ def __log_test_step_update(self, update: TestStepUpdate) -> None: self.test_case_step_errors.setdefault(case_key, []).extend(update.errors) logger.debug(f"Tracked {len(update.errors)} error(s) for test case {case_key}: {update.errors}") - def __handle_log_record(self, records: list[TestLogRecord]) -> None: - for record in records: + async def __handle_log_record(self, records: list[TestLogRecord]) -> None: + # Batches can contain tens of thousands of entries after a large test + # case run. Yield periodically instead of logging the whole batch in + # one uninterrupted stretch, so the websocket read loop (and any + # other pending work, e.g. prompt handling) doesn't stall for the + # entire duration of processing one message. + for i, record in enumerate(records): logger.log(record.level, record.message) + if (i + 1) % LOG_RECORD_YIELD_INTERVAL == 0: + await asyncio.sleep(0) def __suite(self, index: int) -> TestSuiteExecution: return self.run.test_suite_executions[index] From c04a811b49ed01e057a70d1fed45525680731fef Mon Sep 17 00:00:00 2001 From: aamj Date: Mon, 10 Aug 2026 16:29:53 -0300 Subject: [PATCH 2/5] Cap the rendering to 2000 lines and changed log viewer download logs feature --- tests/test_run/test_websocket_socket.py | 43 ++++++--- th_cli/commands/run_tests.py | 1 + th_cli/test_run/log_stream_handler.py | 36 +++---- th_cli/test_run/log_viewer.html | 119 ++++++++++-------------- th_cli/test_run/logging.py | 13 ++- th_cli/test_run/logs_http_server.py | 78 ++++++---------- 6 files changed, 142 insertions(+), 148 deletions(-) diff --git a/tests/test_run/test_websocket_socket.py b/tests/test_run/test_websocket_socket.py index daf531c..bc5f965 100644 --- a/tests/test_run/test_websocket_socket.py +++ b/tests/test_run/test_websocket_socket.py @@ -28,7 +28,14 @@ TestSuiteExecution, TestSuiteMetadata, ) -from th_cli.test_run.socket_schemas import TestCaseUpdate, TestRunUpdate, TestStepUpdate, TestSuiteUpdate, TestUpdate +from th_cli.test_run.socket_schemas import ( + TestCaseUpdate, + TestLogRecord, + TestRunUpdate, + TestStepUpdate, + TestSuiteUpdate, + TestUpdate, +) from th_cli.test_run.websocket import TestRunSocket # --------------------------------------------------------------------------- @@ -304,7 +311,7 @@ async def test_step_update_routed_correctly(self): ), ) with patch.object(s, "_TestRunSocket__log_test_step_update") as mock_fn: - await s._TestRunSocket__handle_test_update(socket=AsyncMock(), update=update) + await s._TestRunSocket__handle_test_update(update=update) mock_fn.assert_called_once() @@ -319,7 +326,7 @@ async def test_case_update_routed_correctly(self): body=TestCaseUpdate(state="passed", test_case_execution_index=0, test_suite_execution_index=0), ) with patch.object(s, "_TestRunSocket__log_test_case_update") as mock_fn: - await s._TestRunSocket__handle_test_update(socket=AsyncMock(), update=update) + await s._TestRunSocket__handle_test_update(update=update) mock_fn.assert_called_once() @@ -333,28 +340,40 @@ async def test_suite_update_routed_correctly(self): body=TestSuiteUpdate(state="passed", test_suite_execution_index=0), ) with patch.object(s, "_TestRunSocket__log_test_suite_update") as mock_fn: - await s._TestRunSocket__handle_test_update(socket=AsyncMock(), update=update) + await s._TestRunSocket__handle_test_update(update=update) mock_fn.assert_called_once() @pytest.mark.asyncio - async def test_run_update_executing_does_not_close_socket(self): + async def test_run_update_executing_leaves_run_not_finished(self): s = _make_socket() - mock_socket = AsyncMock() update = TestUpdate(test_type="test_run", body=TestRunUpdate(state="executing", test_run_execution_id=1)) with patch.object(s, "_TestRunSocket__log_test_run_update", new_callable=AsyncMock): - await s._TestRunSocket__handle_test_update(socket=mock_socket, update=update) + await s._TestRunSocket__handle_test_update(update=update) - mock_socket.close.assert_not_called() + assert s._run_finished is False @pytest.mark.asyncio - async def test_run_update_non_executing_closes_socket(self): + async def test_run_update_non_executing_marks_run_finished(self): s = _make_socket() - mock_socket = AsyncMock() update = TestUpdate(test_type="test_run", body=TestRunUpdate(state="passed", test_run_execution_id=1)) with patch.object(s, "_TestRunSocket__log_test_run_update", new_callable=AsyncMock): - await s._TestRunSocket__handle_test_update(socket=mock_socket, update=update) + await s._TestRunSocket__handle_test_update(update=update) + + assert s._run_finished is True + + @pytest.mark.asyncio + async def test_handle_log_record_logs_every_record(self): + s = _make_socket() + records = [ + TestLogRecord(level="INFO", timestamp=0.0, message=f"msg{i}") for i in range(3) + ] + + with patch("th_cli.test_run.websocket.logger") as mock_logger: + await s._TestRunSocket__handle_log_record(records) - mock_socket.close.assert_called_once() + assert mock_logger.log.call_count == 3 + for record in records: + mock_logger.log.assert_any_call(record.level, record.message) diff --git a/th_cli/commands/run_tests.py b/th_cli/commands/run_tests.py index 6b7532a..beeac27 100644 --- a/th_cli/commands/run_tests.py +++ b/th_cli/commands/run_tests.py @@ -224,6 +224,7 @@ async def run_tests( execution_pics=execution_pics, project_id=project_id, ) + test_logging.set_download_run_id(new_test_run.id) if _contains_webrtc_two_way_talk(selected_tests_dict): _webrtc_handler = TwoWayTalkHandler(port=8999) _webrtc_handler.start_waiting() diff --git a/th_cli/test_run/log_stream_handler.py b/th_cli/test_run/log_stream_handler.py index 0c56aca..03b1869 100644 --- a/th_cli/test_run/log_stream_handler.py +++ b/th_cli/test_run/log_stream_handler.py @@ -28,7 +28,7 @@ class LogStreamHandler: def __init__(self, port: int = 8998): """Initialize the log stream handler. - + Args: port: Port number for the HTTP server (default: 8998) """ @@ -36,47 +36,49 @@ def __init__(self, port: int = 8998): self.http_server = LogsHTTPServer(port=port) self.log_queue: queue.Queue = queue.Queue(maxsize=1000) self.is_running = False - self.log_file_path: Optional[str] = None - - def start(self, test_run_title: str = "Test Execution", log_file_path: Optional[str] = None) -> str: + + def start(self, test_run_title: str = "Test Execution") -> str: """Start the log streaming HTTP server. - + Args: test_run_title: Title of the test run for display - log_file_path: Path to the log file for download functionality - + Returns: URL where logs can be viewed """ if self.is_running: logger.warning("Log stream handler already running") return self._get_log_viewer_url() - + try: - # Store log file path for download functionality - self.log_file_path = log_file_path - # Get local IP address local_ip = self._get_local_ip() - + # Start HTTP server self.http_server.start( log_queue=self.log_queue, test_run_title=test_run_title, local_ip=local_ip, - log_file_path=log_file_path, ) - + self.is_running = True - + viewer_url = f"http://{local_ip}:{self.port}" logger.info(f"Log stream viewer started: {viewer_url}") - + return viewer_url - + except Exception as e: logger.error(f"Failed to start log stream handler: {e}") raise + + def set_run_id(self, run_id: int) -> None: + """Tell the HTTP server which run's log to link "Download Logs" to, + once the run has been created (its id isn't known when the server + starts). + """ + if self.is_running: + self.http_server.set_run_id(run_id) def stop(self): """Stop the log streaming HTTP server.""" diff --git a/th_cli/test_run/log_viewer.html b/th_cli/test_run/log_viewer.html index 64888fc..25abd74 100644 --- a/th_cli/test_run/log_viewer.html +++ b/th_cli/test_run/log_viewer.html @@ -184,6 +184,8 @@ cursor: pointer; transition: all 0.2s; font-weight: 500; + display: inline-block; + text-decoration: none; }} .btn:hover {{ @@ -496,7 +498,7 @@
- + Download Logs Logs: 0
@@ -522,6 +524,14 @@