From 9ff75e5b4c4a2d30ee5cefb18a2890dd0ceb4596 Mon Sep 17 00:00:00 2001 From: antonio-amjr <116589331+antonio-amjr@users.noreply.github.com> Date: Fri, 14 Aug 2026 16:51:23 -0300 Subject: [PATCH] [Fix] Massive Logs Support to CLI (#347) * Fixing CLI logs interrupting for some edge cases * Using WS timeout, changing log funcs to async and chunked output and cathing connection close errors * Stop redundant DB commits and unbounded log broadcasts under high log volume * Fixing and adding to the unit tests * Fixing Black linting errors * Wrapping websocket closure with a try/catch block. * Fix PR review findings: task ordering, UTF-8 decoding, and encoding safety - TestUIObserver: move __async_updates to instance state (was a shared class-level list, leaking Task references across runs) - TestUIObserver: broadcast each flush's chunks in order via one sequential task instead of one independent task per chunk - test_harness_client: use explicit UTF-8 encoding for the SDK log file - test_case: read test_output.txt incrementally with a persistent UTF-8 decoder so split multi-byte characters aren't corrupted across reads - test_case: stream display_batch_logs()/_log_remaining_content() instead of loading the whole file into memory * Fixing ui observer unit test --- app/main.py | 8 +- app/socket_connection_manager.py | 24 +- app/test_engine/test_db_observer.py | 52 ++-- app/test_engine/test_runner.py | 2 +- app/test_engine/test_ui_observer.py | 65 ++++- .../test_socket_connection_manager.py | 35 ++- app/tests/test_engine/test_db_observer.py | 26 ++ app/tests/test_engine/test_ui_observer.py | 112 +++++++- .../models/rpc_client/test_harness_client.py | 8 +- .../python_testing/models/test_case.py | 115 +++++++-- .../support/python_testing/models/utils.py | 2 +- .../python_tests/test_python_test_case.py | 244 +++++++++++++++++- 12 files changed, 616 insertions(+), 77 deletions(-) diff --git a/app/main.py b/app/main.py index 5a246314..79746e89 100644 --- a/app/main.py +++ b/app/main.py @@ -20,6 +20,7 @@ from app.api.api_v1.api import api_router from app.core.config import settings from app.test_engine.test_script_manager import test_script_manager +from app.uvicorn_worker import WS_PING_TIMEOUT_S app = FastAPI( title=settings.PROJECT_NAME, openapi_url=f"{settings.API_V1_STR}/openapi.json" @@ -51,4 +52,9 @@ async def startup_event() -> None: if __name__ == "__main__": - uvicorn.run(app, host="0.0.0.0", port=80, log_config=None) + # ws_ping_timeout raised from uvicorn's 20s default: a large log flush can + # keep the event loop busy long enough to miss the keepalive pong and get + # the websocket dropped mid-broadcast (see app.uvicorn_worker). + uvicorn.run( + app, host="0.0.0.0", port=80, log_config=None, ws_ping_timeout=WS_PING_TIMEOUT_S + ) diff --git a/app/socket_connection_manager.py b/app/socket_connection_manager.py index 98bcec69..6c5fff42 100644 --- a/app/socket_connection_manager.py +++ b/app/socket_connection_manager.py @@ -24,7 +24,7 @@ from fastapi.websockets import WebSocketDisconnect from loguru import logger from starlette.websockets import WebSocketState -from websockets.exceptions import ConnectionClosedOK +from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK from app.constants.shared_constants import MessageKeysEnum, MessageTypeEnum from app.constants.websockets_constants import ( @@ -129,7 +129,8 @@ async def broadcast(self, message: Union[str, dict, list]) -> None: # Convert dictionaries and lists to string using json if isinstance(message, dict) or isinstance(message, list): message = json.dumps(message, default=pydantic.json.pydantic_encoder) - for connection in self.active_connections: + # Iterate over a copy: disconnect() below mutates active_connections. + for connection in list(self.active_connections): if connection.type == WebSocketTypeEnum.MAIN: websocket = connection.websocket try: @@ -137,9 +138,24 @@ async def broadcast(self, message: Union[str, dict, list]) -> None: # Starlette raises websockets.exceptions.ConnectionClosedOK # when trying to send to a closed websocket. # https://github.com/encode/starlette/issues/759 - except ConnectionClosedOK: + # ConnectionClosedError is raised on an abrupt drop (e.g. a + # missed websocket keepalive pong) rather than a graceful + # close - treat it the same way so a dead connection doesn't + # keep being retried for the rest of the process's life. + except (ConnectionClosedOK, ConnectionClosedError): if websocket.application_state != WebSocketState.DISCONNECTED: - await websocket.close() + try: + await websocket.close() + except Exception as close_error: + # The transport can already be gone by this point + # (that's the whole reason we're here) - closing + # an already-dead socket failing is expected and + # uninteresting, but must not prevent the + # cleanup below from running. + logger.debug( + f"Error closing already-dead websocket: {close_error}" + ) + self.disconnect(connection) logger.warning( f'Failed to send message: "{message}"' f' to websocket: "{websocket}", connection closed."' diff --git a/app/test_engine/test_db_observer.py b/app/test_engine/test_db_observer.py index f657f584..8d02b865 100644 --- a/app/test_engine/test_db_observer.py +++ b/app/test_engine/test_db_observer.py @@ -13,8 +13,8 @@ # See the License for the specific language governing permissions and # limitations under the License. # +import asyncio from datetime import datetime -from queue import Empty, Queue from typing import Callable, Generator, Union from loguru import logger @@ -30,6 +30,10 @@ from app.test_engine.models import TestCase, TestRun, TestStep, TestSuite from app.test_engine.test_observer import Observer +ExecutionObj = Union[ + TestCaseExecution, TestStepExecution, TestSuiteExecution, TestRunExecution +] + class TestDBObserver(Observer): __test__ = False # Needed to indicate to PyTest that this is not a "test" @@ -38,15 +42,19 @@ def __init__( self, db_generator: Callable[[], Generator[Session, None, None]] = get_db ) -> None: self.__db_generator = db_generator - self.data_queue: Queue = Queue() - - def apply_updates(self) -> None: - while not self.data_queue.empty(): - try: - data = self.data_queue.get(timeout=0.1) - self.__save(data) - except Empty: - pass + # Keyed by id(execution_obj) instead of a plain queue: dispatch() can + # re-notify the SAME object many times (e.g. once per 0.5s log-flush + # tick for the whole run) before apply_updates() ever drains this - + # a dict collapses those into a single pending save per distinct + # object instead of committing the same (increasingly large) object + # redundantly once per notification. + self.__pending: dict[int, ExecutionObj] = {} + + async def apply_updates(self) -> None: + pending = self.__pending + self.__pending = {} + for data in pending.values(): + await self.__save(data) def dispatch( self, observable: Union[TestRun, TestSuite, TestCase, TestStep] @@ -62,6 +70,9 @@ def dispatch( elif isinstance(observable, TestStep): self.__onTestStepUpdate(observable) + def __enqueue(self, execution_obj: ExecutionObj) -> None: + self.__pending[id(execution_obj)] = execution_obj + def __onTestRunUpdate(self, observable: "TestRun") -> None: logger.debug("Test Run Observer received", observable) test_run_execution = observable.test_run_execution @@ -74,7 +85,7 @@ def __onTestRunUpdate(self, observable: "TestRun") -> None: if self.isCompleted(observable.state): test_run_execution.completed_at = datetime.now() - self.data_queue.put(test_run_execution) + self.__enqueue(test_run_execution) def __onTestSuiteUpdate(self, observable: "TestSuite") -> None: logger.debug("Test Suite Observer received", observable) @@ -89,7 +100,7 @@ def __onTestSuiteUpdate(self, observable: "TestSuite") -> None: if self.isCompleted(observable.state): observable.test_suite_execution.completed_at = datetime.now() - self.data_queue.put(observable.test_suite_execution) + self.__enqueue(observable.test_suite_execution) def __onTestCaseUpdate(self, observable: "TestCase") -> None: logger.debug("Test Case Observer received", observable) @@ -104,7 +115,7 @@ def __onTestCaseUpdate(self, observable: "TestCase") -> None: if self.isCompleted(observable.state): observable.test_case_execution.completed_at = datetime.now() - self.data_queue.put(observable.test_case_execution) + self.__enqueue(observable.test_case_execution) def __onTestStepUpdate(self, observable: "TestStep") -> None: logger.debug("Test Step Observer received", observable) @@ -121,14 +132,9 @@ def __onTestStepUpdate(self, observable: "TestStep") -> None: if self.isCompleted(observable.state): observable.test_step_execution.completed_at = datetime.now() - self.data_queue.put(observable.test_step_execution) + self.__enqueue(observable.test_step_execution) - def __save( - self, - execution_obj: Union[ - TestCaseExecution, TestStepExecution, TestSuiteExecution, TestRunExecution - ], - ) -> None: + async def __save(self, execution_obj: ExecutionObj) -> None: # We get the session from the model it self to avoid overriding values when # using a different session insp = inspect(execution_obj) @@ -139,7 +145,11 @@ def __save( session = next(self.__db_generator()) session.add(execution_obj) session.expire_on_commit = False - session.commit() + # session.commit() is a blocking, synchronous SQLAlchemy call. It can + # be a large write (e.g. a run's full log), so keep it off the event + # loop rather than stalling every other coroutine (websocket pings, + # other requests) for however long it takes. + await asyncio.to_thread(session.commit) logger.debug( f"Saved {execution_obj.__class__} {execution_obj.id}" f" with state {execution_obj.state}" diff --git a/app/test_engine/test_runner.py b/app/test_engine/test_runner.py index 49b8a7c3..18e53097 100644 --- a/app/test_engine/test_runner.py +++ b/app/test_engine/test_runner.py @@ -181,7 +181,7 @@ async def run(self) -> None: self.test_run.unsubscribe([ui_observer, db_observer]) # Flush all pending DB updates - db_observer.apply_updates() + await db_observer.apply_updates() # Ensure all state updates are sent to the frontend await ui_observer.complete_tasks() diff --git a/app/test_engine/test_ui_observer.py b/app/test_engine/test_ui_observer.py index b0bc30ca..7fc294c9 100644 --- a/app/test_engine/test_ui_observer.py +++ b/app/test_engine/test_ui_observer.py @@ -38,11 +38,30 @@ class TestUpdateTypeEnum(str, Enum): TEST_CASE = "Test Case" +# Maximum number of log entries broadcast in a single websocket message. A +# dense burst of log lines (e.g. a large conformance report) can otherwise +# accumulate thousands of entries into one flush, producing a single +# multi-MB message whose synchronous JSON serialization (see +# SocketConnectionManager.broadcast) has no yield point. Splitting into +# multiple smaller messages, sent in order within a single task (see +# __handle_test_run_log), lets the event loop interleave other work (e.g. +# websocket keepalives) between chunks. +LOG_RECORDS_BROADCAST_CHUNK_SIZE = 200 + + class TestUIObserver(Observer): __test__ = False - __async_updates: list[Task] = [] - __last_seen_run_state: Optional[TestStateEnum] = None - __last_seen_run_log_len = 0 + + def __init__(self) -> None: + # Instance state - must not be class-level mutable defaults. A list + # declared at class scope is shared across every TestUIObserver + # instance until an instance reassigns it; TestRunner creates a + # fresh instance per run, so a shared, ever-growing task list would + # leak Task references (and their retained coroutine state) across + # every run for the life of the process. + self.__async_updates: list[Task] = [] + self.__last_seen_run_state: Optional[TestStateEnum] = None + self.__last_seen_run_log_len = 0 def dispatch( self, observable: Union[TestRun, TestSuite, TestCase, TestStep] @@ -81,7 +100,11 @@ def __handle_test_run_log(self, test_run: TestRun) -> None: log_len = len(test_run.log) if log_len > self.__last_seen_run_log_len: new_entries = test_run.log[self.__last_seen_run_log_len :] - self.__send_log_records_message(new_entries) + chunks = [ + new_entries[i : i + LOG_RECORDS_BROADCAST_CHUNK_SIZE] + for i in range(0, len(new_entries), LOG_RECORDS_BROADCAST_CHUNK_SIZE) + ] + self.__send_log_records_messages_in_order(chunks) self.__last_seen_run_log_len = log_len def __onTestSuiteUpdate(self, observable: TestSuite) -> None: @@ -138,13 +161,33 @@ def __send_test_update_message(self, update_payload: dict) -> None: } ) - def __send_log_records_message(self, log_entries: list[TestRunLogEntry]) -> None: - self.__send_message( - { - MessageKeysEnum.TYPE: MessageTypeEnum.TEST_LOG_RECORDS, - MessageKeysEnum.PAYLOAD: log_entries, - } - ) + def __send_log_records_messages_in_order( + self, chunks: list[list[TestRunLogEntry]] + ) -> None: + """Broadcast each chunk of one flush as its own TEST_LOG_RECORDS + message, in order. + + Scheduled as a single task for the whole flush, not one task per + chunk: chunks must be delivered in order, but independently + scheduled tasks don't guarantee that - SocketConnectionManager. + broadcast()'s websocket.send_text() is genuine async I/O that can + yield under backpressure, which could let a later chunk's task + complete first. Awaiting each chunk's broadcast sequentially within + one task guarantees order while still yielding the event loop + between chunks (each await is itself a yield point). + """ + + async def _send_in_order() -> None: + for chunk in chunks: + await socket_connection_manager.broadcast( + { + MessageKeysEnum.TYPE: MessageTypeEnum.TEST_LOG_RECORDS, + MessageKeysEnum.PAYLOAD: chunk, + } + ) + + task = create_task(_send_in_order()) + self.__async_updates.append(task) def __send_message(self, message: dict[str, Any]) -> None: # enqueue update diff --git a/app/tests/socket_connection_manager/test_socket_connection_manager.py b/app/tests/socket_connection_manager/test_socket_connection_manager.py index 395e2fd8..63123d04 100644 --- a/app/tests/socket_connection_manager/test_socket_connection_manager.py +++ b/app/tests/socket_connection_manager/test_socket_connection_manager.py @@ -20,7 +20,7 @@ import pytest from fastapi import WebSocket from starlette.websockets import WebSocketState -from websockets.exceptions import ConnectionClosedOK +from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK from app.constants.shared_constants import MessageKeysEnum, MessageTypeEnum from app.constants.websockets_constants import ( @@ -203,6 +203,39 @@ async def test_broadcast_failed_for_ConnectionClosed() -> None: socket_connection_manager.active_connections.clear() +@pytest.mark.asyncio +async def test_broadcast_failed_for_ConnectionClosedError() -> None: + """ + Tests that broadcast() also handles ConnectionClosedError (an abrupt + drop, e.g. from a missed keepalive ping under event-loop stall) the same + way as ConnectionClosedOK: the socket is closed and the dead connection + is removed from active_connections so it isn't retried for the rest of + the process's life (regression test for issue #1072). + """ + test_message = "Test" + socket_connection_manager.active_connections.clear() + + # Add a websocket object to the "active_connections" list to imitate + # an existing active connection + socket = mock.MagicMock(spec=WebSocket) + socket.application_state = WebSocketState.CONNECTED + connection = WebSocketConnection(socket, WebSocketTypeEnum.MAIN) + + socket_connection_manager.active_connections.append(connection) + assert len(socket_connection_manager.active_connections) == 1 + + # Force an abrupt connection-closed exception + socket.send_text.side_effect = ConnectionClosedError(rcvd=None, sent=None) + + await socket_connection_manager.broadcast(message=test_message) + socket.send_text.assert_called_once_with(test_message) + socket.close.assert_called_once() + assert connection not in socket_connection_manager.active_connections + + # Cleanup + socket_connection_manager.active_connections.clear() + + @pytest.mark.asyncio async def test_broadcast_failed_for_RuntimeError() -> None: """ diff --git a/app/tests/test_engine/test_db_observer.py b/app/tests/test_engine/test_db_observer.py index 10382ec3..d4d99f2c 100644 --- a/app/tests/test_engine/test_db_observer.py +++ b/app/tests/test_engine/test_db_observer.py @@ -14,6 +14,7 @@ # limitations under the License. # import asyncio +from unittest import mock import pytest from sqlalchemy.orm import Session @@ -281,3 +282,28 @@ def test_test_db_observer_update(db: Session) -> None: db.close() test_db_observer.dispatch(test_step) assert TestStateEnum.EXECUTING == test_step.test_step_execution.state + + +@pytest.mark.asyncio +async def test_test_db_observer_dedups_repeated_run_updates(db: Session) -> None: + """Repeated dispatch() calls for the same TestRunExecution before + apply_updates() runs must collapse into a single commit, instead of one + redundant commit per dispatch (regression test for issue #1072's + DB-write storm, where ~1,400 redundant commits of the same growing log + blocked the event loop for the whole duration of a large test run).""" + test_script_manager = TestScriptManager() + test_db_observer = TestDBObserver() + + test_run_execution = create_test_run_execution_with_some_test_cases(db=db) + test_run = test_script_manager.get_test_run(db, test_run_execution) + test_run.state = TestStateEnum.EXECUTING + + with mock.patch.object(Session, "commit") as mock_commit: + # Simulate several rapid dispatch() calls for the same object, as + # happens once per ~0.5s log-flush tick during a run. + for _ in range(5): + test_db_observer.dispatch(test_run) + + await test_db_observer.apply_updates() + + assert mock_commit.call_count == 1 diff --git a/app/tests/test_engine/test_ui_observer.py b/app/tests/test_engine/test_ui_observer.py index f11c3ca6..af2b76ef 100644 --- a/app/tests/test_engine/test_ui_observer.py +++ b/app/tests/test_engine/test_ui_observer.py @@ -13,6 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. # +import asyncio from typing import Any, Dict from unittest import mock @@ -28,15 +29,32 @@ from app.models.test_run_execution import TestRunExecution from app.schemas.test_run_log_entry import TestRunLogEntry from app.test_engine.models import TestRun -from app.test_engine.test_ui_observer import TestUIObserver, TestUpdateTypeEnum +from app.test_engine.test_ui_observer import ( + LOG_RECORDS_BROADCAST_CHUNK_SIZE, + TestUIObserver, + TestUpdateTypeEnum, +) + + +def _log_record_payloads(broadcast_mock: mock.AsyncMock) -> list[list[TestRunLogEntry]]: + """Extract only the TEST_LOG_RECORDS payloads from a mocked broadcast()'s + calls, ignoring the TEST_UPDATE state-change message that notify() also + fires on the first call (when state differs from the observer's + initial/None last-seen state).""" + return [ + call.args[0][MessageKeysEnum.PAYLOAD] + for call in broadcast_mock.call_args_list + if call.args[0][MessageKeysEnum.TYPE] == MessageTypeEnum.TEST_LOG_RECORDS + ] @pytest.mark.asyncio async def test_test_ui_observer_test_run_log(db: Session) -> None: ui_observer = TestUIObserver() - with mock.patch.object( - ui_observer, "_TestUIObserver__send_log_records_message" - ) as send_log_mock: + with mock.patch( + "app.test_engine.test_ui_observer.socket_connection_manager.broadcast", + new_callable=mock.AsyncMock, + ) as broadcast_mock: run = TestRun(test_run_execution=TestRunExecution()) run.subscribe([ui_observer]) @@ -47,13 +65,15 @@ async def test_test_ui_observer_test_run_log(db: Session) -> None: ] run.log = log_entries run.notify() - send_log_mock.assert_called_once_with(log_entries) - send_log_mock.reset_mock() + await ui_observer.complete_tasks() + assert _log_record_payloads(broadcast_mock) == [log_entries] + broadcast_mock.reset_mock() # Assert send_log is not called when no new logs are added run.notify() - send_log_mock.assert_not_called() - send_log_mock.reset_mock() + await ui_observer.complete_tasks() + assert _log_record_payloads(broadcast_mock) == [] + broadcast_mock.reset_mock() # Assert only new log events are in call additional_log_entries = [ @@ -63,11 +83,83 @@ async def test_test_ui_observer_test_run_log(db: Session) -> None: run.log.extend(additional_log_entries) assert len(run.log) == 4 run.notify() - send_log_mock.assert_called_once_with(additional_log_entries) + await ui_observer.complete_tasks() + assert _log_record_payloads(broadcast_mock) == [additional_log_entries] + + +@pytest.mark.asyncio +async def test_test_ui_observer_test_run_log_chunks_large_batches(db: Session) -> None: + """A single flush containing more entries than the broadcast chunk size + must be split into multiple smaller messages instead of one large one + (regression test for issue #1072's unbounded-broadcast-batch bug, where + a dense burst of log lines could become a single multi-MB websocket + message with no yield point during serialization).""" + ui_observer = TestUIObserver() + with mock.patch( + "app.test_engine.test_ui_observer.socket_connection_manager.broadcast", + new_callable=mock.AsyncMock, + ) as broadcast_mock: + run = TestRun(test_run_execution=TestRunExecution()) + run.subscribe([ui_observer]) + + extra = 50 + log_entries = [ + TestRunLogEntry(level="info", timestamp=float(i), message=f"Message{i}") + for i in range(LOG_RECORDS_BROADCAST_CHUNK_SIZE + extra) + ] + run.log = log_entries + run.notify() + await ui_observer.complete_tasks() + + chunks = _log_record_payloads(broadcast_mock) + assert len(chunks) == 2 + assert chunks[0] == log_entries[:LOG_RECORDS_BROADCAST_CHUNK_SIZE] + assert chunks[1] == log_entries[LOG_RECORDS_BROADCAST_CHUNK_SIZE:] - # cleanup + +@pytest.mark.asyncio +async def test_test_ui_observer_test_run_log_chunks_delivered_in_order( + db: Session, +) -> None: + """Chunks of one flush must be broadcast in order, even though the + actual sends happen inside an awaited task rather than synchronously + (regression test: chunks used to be scheduled as independently-created + tasks, which don't guarantee delivery order relative to each other if + websocket.send_text() ever actually yields, e.g. under backpressure).""" + ui_observer = TestUIObserver() + send_order: list[int] = [] + + async def _recording_broadcast(message: dict) -> None: + # Ignore the TEST_UPDATE state-change message notify() also fires - + # only TEST_LOG_RECORDS chunks are relevant to ordering here. + if message[MessageKeysEnum.TYPE] != MessageTypeEnum.TEST_LOG_RECORDS: + return + # Simulate send_text() genuinely yielding control (e.g. under + # backpressure) - if chunks were sent via independent tasks, this + # would let a later chunk's task finish first. + payload = message[MessageKeysEnum.PAYLOAD] + first_entry_index = int(payload[0].message.removeprefix("Message")) + await asyncio.sleep(0) + send_order.append(first_entry_index) + + with mock.patch( + "app.test_engine.test_ui_observer.socket_connection_manager.broadcast", + side_effect=_recording_broadcast, + ): + run = TestRun(test_run_execution=TestRunExecution()) + run.subscribe([ui_observer]) + + extra = 50 + log_entries = [ + TestRunLogEntry(level="info", timestamp=float(i), message=f"Message{i}") + for i in range(LOG_RECORDS_BROADCAST_CHUNK_SIZE + extra) + ] + run.log = log_entries + run.notify() await ui_observer.complete_tasks() + assert send_order == [0, LOG_RECORDS_BROADCAST_CHUNK_SIZE] + def __expected_test_run_log_dict() -> Dict[str, Any]: return { diff --git a/test_collections/matter/sdk_tests/support/python_testing/models/rpc_client/test_harness_client.py b/test_collections/matter/sdk_tests/support/python_testing/models/rpc_client/test_harness_client.py index be42563a..8ffa5ef3 100644 --- a/test_collections/matter/sdk_tests/support/python_testing/models/rpc_client/test_harness_client.py +++ b/test_collections/matter/sdk_tests/support/python_testing/models/rpc_client/test_harness_client.py @@ -191,7 +191,13 @@ def main() -> None: # TODO: find a better solution. # This is a temporary workaround since Python Tests # are generating a big amount of log - with open(EXECUTION_LOG_OUTPUT, "w") as f: + # buffering=1 (line buffering) ensures each line reaches disk as it's + # printed, instead of sitting in a block buffer until the file closes + # at the very end of the test run. encoding is explicit since every + # downstream reader of this file assumes UTF-8 - the process locale + # default (which varies by environment, e.g. ASCII in a minimal + # container) is not a safe assumption to share with them implicitly. + with open(EXECUTION_LOG_OUTPUT, "w", buffering=1, encoding="utf-8") as f: with redirect_stdout(f): # Check if 'commission' was passed if th_args[TH_COMMISSION_ARGUMENT]: diff --git a/test_collections/matter/sdk_tests/support/python_testing/models/test_case.py b/test_collections/matter/sdk_tests/support/python_testing/models/test_case.py index d9c7e594..92cc0186 100644 --- a/test_collections/matter/sdk_tests/support/python_testing/models/test_case.py +++ b/test_collections/matter/sdk_tests/support/python_testing/models/test_case.py @@ -13,6 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. # +import codecs import re from asyncio import sleep from inspect import iscoroutinefunction @@ -68,6 +69,9 @@ LOG_BATCH_SIZE = 50 # Number of log lines to send per batch LOG_BATCH_DELAY = 0.01 # Delay in seconds between batches (10ms) +# Marker prefix printed by the SDK before each test step's output +STEP_MARKER_PREFIX = "***** Test Step " + # Custom type variable used to annotate the factory method in PythonTestCase. T = TypeVar("T", bound="PythonTestCase") @@ -97,11 +101,20 @@ def __init__(self, test_case_execution: TestCaseExecution) -> None: self.test_stop_called = False self.test_socket = None self.file_output_path: Optional[Path] = None - self.current_python_step_number = 0 + self.current_python_step_name: Optional[str] = None self._cached_file_content: str = "" self._last_file_size: int = 0 self._last_logged_position: int = 0 # Track last logged position self._remaining_content_logged: bool = False + # Incremental UTF-8 decoder for _read_file_incrementally(). A plain + # `open(..., encoding="utf-8")` decodes each call's bytes in + # isolation - if a multi-byte character's bytes are split across two + # incremental reads (very possible while the file is still growing), + # errors="replace" doesn't defer it, it corrupts it into U+FFFD + # immediately with no way to recover. An incremental decoder holds + # back any incomplete trailing bytes and completes the character + # once the rest arrives on the next call. + self._utf8_decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") # Move to the next step if the test case has additional steps apart from the 2 # deafult ones @@ -138,7 +151,7 @@ def step_skipped(self, name: str, expression: str) -> None: self.current_test_step.mark_as_not_applicable(skiped_msg) def step_start(self, name: str) -> None: - self.current_python_step_number += 1 + self.current_python_step_name = name self.step_over() async def step_success( @@ -171,13 +184,16 @@ async def _display_step_logs(self) -> None: logger.debug(f"Test output file does not exist: {self.file_output_path}") return + if not self.current_python_step_name: + return + try: # Read file content with incremental caching for performance content = self._read_file_incrementally() # Extract logs for the current step (returns tuple of logs and end position) step_logs, end_pos = self._extract_logs_for_step( - content, self.current_python_step_number + content, self.current_python_step_name ) if step_logs: @@ -204,6 +220,13 @@ async def _display_step_logs(self) -> None: def _read_file_incrementally(self) -> str: """Read file incrementally, caching content to avoid re-reading entire file. + Reads in binary mode and feeds bytes through a persistent + incremental UTF-8 decoder, rather than independently decoding each + call's bytes in text mode. Binary-mode seeks are always byte-safe + (unlike seeking a text-mode handle to an arbitrary byte offset), and + the incremental decoder correctly withholds any incomplete trailing + multi-byte character across calls instead of corrupting it. + Returns: Full content of the file up to current position """ @@ -220,14 +243,18 @@ def _read_file_incrementally(self) -> str: # File has grown, read only new content if current_size > self._last_file_size and self._cached_file_content: - with open(self.file_output_path, "r", encoding="utf-8") as f: + with open(self.file_output_path, "rb") as f: f.seek(self._last_file_size) - new_content = f.read() - self._cached_file_content += new_content + new_bytes = f.read() + self._cached_file_content += self._utf8_decoder.decode(new_bytes) else: - # First read or file was truncated - with open(self.file_output_path, "r", encoding="utf-8") as f: - self._cached_file_content = f.read() + # First read or file was truncated - reset decoder state too, + # so any bytes withheld from a previous (now-stale) read + # don't get prepended to unrelated new content. + self._utf8_decoder.reset() + with open(self.file_output_path, "rb") as f: + new_bytes = f.read() + self._cached_file_content = self._utf8_decoder.decode(new_bytes) self._last_file_size = current_size return self._cached_file_content @@ -236,29 +263,35 @@ def _read_file_incrementally(self) -> str: return self._cached_file_content def _extract_logs_for_step( - self, content: str, step_number: int + self, content: str, step_name: str ) -> tuple[list[str], int]: """Extract logs for a specific test step from the full log content. Args: content: Full content of the test output file - step_number: The step number to extract logs for + step_name: The step name (as passed to step_start) to extract logs for Returns: Tuple of: - list of log lines for the specified step. - End position of step content """ - current_step_marker = f"***** Test Step {step_number} :" - next_step_marker = f"***** Test Step {step_number + 1} :" + current_step_marker = f"{STEP_MARKER_PREFIX}{step_name}" - # Find the start position of current step - start_idx = content.find(current_step_marker) + # Search starting at the last logged position rather than from the + # beginning of the file, so repeated/similar step labels across loop + # iterations resolve to the correct occurrence and the whole + # (growing) file isn't rescanned on every step. + start_idx = content.find(current_step_marker, self._last_logged_position) if start_idx == -1: - return ([], 0) + # No match yet (e.g. the SDK hasn't written this step's marker to + # disk yet) — don't regress the cursor, just report nothing new. + return ([], self._last_logged_position) - # Find the start position of next step - next_idx = content.find(next_step_marker, start_idx) + # Find the start position of the next step's marker + next_idx = content.find( + STEP_MARKER_PREFIX, start_idx + len(current_step_marker) + ) # Extract the section between current and next step if next_idx != -1: @@ -478,7 +511,7 @@ async def cleanup(self) -> None: await self._log_remaining_content() else: # Use batch logging when real-time logging is disabled - self.display_batch_logs() + await self.display_batch_logs() async def _log_remaining_content(self) -> None: """Log any content from the test output file that wasn't logged yet.""" @@ -498,7 +531,16 @@ async def _log_remaining_content(self) -> None: return try: - # Read the full file content + # Reuse the incremental reader (rather than a second, independent + # full read) to pick up anything appended since the last call and + # fold it into the cache. This used to deliberately bypass the + # cache "so this catch-all reliably recovers everything + # regardless of whether the incremental/per-step read path hit + # an error earlier" - that hedge existed because + # _read_file_incrementally() used to be able to silently corrupt + # content at chunk boundaries. Now that it decodes incrementally + # and byte-safely, there's no need to duplicate the whole file + # in memory a second time here. content = self._read_file_incrementally() # Check if there's content after the last logged position @@ -507,8 +549,16 @@ async def _log_remaining_content(self) -> None: if remaining_content.strip(): logger.info("---- Remaining logs not captured by steps ----") - # Just log all remaining content directly - logger.log(PYTHON_TEST_LEVEL, remaining_content) + # Log in batches (like _display_step_logs) instead of one + # unbroken call, so a large backlog doesn't monopolize the + # event loop for an extended stretch in one go. + remaining_lines = remaining_content.split("\n") + for i in range(0, len(remaining_lines), LOG_BATCH_SIZE): + batch = remaining_lines[i : i + LOG_BATCH_SIZE] + for line in batch: + logger.log(PYTHON_TEST_LEVEL, line) + if i + LOG_BATCH_SIZE < len(remaining_lines): + await sleep(LOG_BATCH_DELAY) logger.info("---- End of remaining logs ----") # Mark as logged to prevent duplicate calls @@ -521,7 +571,7 @@ async def _log_remaining_content(self) -> None: f"Unexpected error while logging remaining content: {e}", exc_info=True ) - def display_batch_logs(self) -> None: + async 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, @@ -542,9 +592,24 @@ def display_batch_logs(self) -> None: try: logger.info("---- Start of Python test logs ----") - with open(self.file_output_path, "r", encoding="utf-8") as f: + with open( + self.file_output_path, "r", encoding="utf-8", errors="replace" + ) as f: + # Iterate the file object directly instead of readlines(): + # file iteration is already lazily buffered by Python, so + # this never materializes the whole (potentially 100MB+) + # file as one in-memory list before logging/pacing even + # starts. + batch_count = 0 for line in f: logger.log(PYTHON_TEST_LEVEL, line.rstrip("\n")) + batch_count += 1 + if batch_count >= LOG_BATCH_SIZE: + batch_count = 0 + # Yield to the event loop between batches, so a + # large file doesn't monopolize it for an extended + # stretch in one go. + await sleep(LOG_BATCH_DELAY) logger.info("---- End of Python test logs ----") except (IOError, OSError) as e: logger.warning(f"Failed to read test output file: {e}") @@ -660,7 +725,7 @@ async def execute(self) -> None: await self._log_remaining_content() else: # Use batch logging when real-time logging is disabled - self.display_batch_logs() + await self.display_batch_logs() self.current_test_step.mark_as_completed() finally: diff --git a/test_collections/matter/sdk_tests/support/python_testing/models/utils.py b/test_collections/matter/sdk_tests/support/python_testing/models/utils.py index 87a24f40..fbf675a6 100644 --- a/test_collections/matter/sdk_tests/support/python_testing/models/utils.py +++ b/test_collections/matter/sdk_tests/support/python_testing/models/utils.py @@ -234,7 +234,7 @@ def log_test_output_file(logger: loguru.Logger) -> None: file_output_path = sdk_tests_path / TEST_OUTPUT_FILE_PATH if file_output_path.exists(): - with open(file_output_path, "r") as f: + with open(file_output_path, "r", encoding="utf-8", errors="replace") as f: content = f.read() if content.strip(): # Only log if there's actual content logger.log(PYTHON_TEST_LEVEL, content) diff --git a/test_collections/matter/sdk_tests/support/tests/python_tests/test_python_test_case.py b/test_collections/matter/sdk_tests/support/tests/python_tests/test_python_test_case.py index bd6e835d..002b6b29 100644 --- a/test_collections/matter/sdk_tests/support/tests/python_tests/test_python_test_case.py +++ b/test_collections/matter/sdk_tests/support/tests/python_tests/test_python_test_case.py @@ -26,7 +26,7 @@ from app.models.test_case_execution import TestCaseExecution from app.models.test_run_execution import TestRunExecution from app.models.test_suite_execution import TestSuiteExecution -from app.test_engine.logger import test_engine_logger +from app.test_engine.logger import PYTHON_TEST_LEVEL, test_engine_logger from ...models.matter_test_models import MatterTestStep, MatterTestType from ...python_testing.models import PythonTestCase @@ -820,3 +820,245 @@ def test_realtime_logs_enabled_defers_to_env_when_th_config_absent() -> None: ) as mock_settings: mock_settings.ENABLE_REALTIME_PYTHON_TEST_LOGS = True assert instance._realtime_logs_enabled() is True + + +def test_extract_logs_for_step_matches_composite_label() -> None: + """Regression test for issue #1072: composite/alphanumeric step labels + (e.g. from privilege-level loops such as TC-ACE-2.4) must be matched + correctly, instead of never matching (as a synthetic integer counter + would).""" + test = python_test_instance() + case_class: Type[PythonTestCase] = PythonTestCase.class_factory( + test=test, python_test_version="version", mandatory=False + ) + instance = case_class(TestCaseExecution()) + + content = ( + "***** Test Step 3a_kView : desc\n" + "lineA\n" + "***** Test Step 3b_kView : desc\n" + "lineB\n" + ) + + step_logs, end_pos = instance._extract_logs_for_step(content, "3a_kView : desc") + + assert "lineA" in step_logs + assert not any("lineB" in line for line in step_logs) + assert end_pos == content.index("***** Test Step 3b_kView : desc") + + +def test_extract_logs_for_step_matches_plain_integer_label() -> None: + """Backward-compatibility guard: plain integer-style step labels (the + common case for tests without loops) must keep matching correctly.""" + test = python_test_instance() + case_class: Type[PythonTestCase] = PythonTestCase.class_factory( + test=test, python_test_version="version", mandatory=False + ) + instance = case_class(TestCaseExecution()) + + content = ( + "***** Test Step 1 : first description\n" + "line1\n" + "***** Test Step 2 : second description\n" + "line2\n" + ) + + step_logs, end_pos = instance._extract_logs_for_step( + content, "1 : first description" + ) + + assert "line1" in step_logs + assert not any("line2" in line for line in step_logs) + assert end_pos == content.index("***** Test Step 2 : second description") + + +def test_extract_logs_for_step_uses_cursor_not_first_occurrence() -> None: + """Regression test: extraction must search from the last logged + position, not always from the start of the file, so a marker that + appears more than once (e.g. a reused label across loop iterations) + resolves to the correct, later occurrence.""" + test = python_test_instance() + case_class: Type[PythonTestCase] = PythonTestCase.class_factory( + test=test, python_test_version="version", mandatory=False + ) + instance = case_class(TestCaseExecution()) + + marker = "***** Test Step 3a_kView : desc" + content = f"{marker}\nfirst pass\n{marker}\nsecond pass\n" + + first_logs, first_end = instance._extract_logs_for_step(content, "3a_kView : desc") + assert "first pass" in first_logs + + instance._last_logged_position = first_end + second_logs, second_end = instance._extract_logs_for_step( + content, "3a_kView : desc" + ) + + assert "second pass" in second_logs + assert not any("first pass" in line for line in second_logs) + assert second_end == len(content) + + +def test_step_start_stores_step_name() -> None: + """step_start must keep the actual SDK step label (used for marker + matching), not a synthetic incrementing counter.""" + test = python_test_instance() + case_class: Type[PythonTestCase] = PythonTestCase.class_factory( + test=test, python_test_version="version", mandatory=False + ) + instance = case_class(TestCaseExecution()) + + instance.step_start("3a_kView : some description") + + assert instance.current_python_step_name == "3a_kView : some description" + + +@pytest.mark.asyncio +async def test_log_remaining_content_recovers_everything_missed(tmp_path: Path) -> None: + """_log_remaining_content must recover all content after the last logged + position (via the incremental reader, which picks up anything appended + since the last call) so it acts as a reliable safety net regardless of + why the per-step path may have missed content.""" + test = python_test_instance() + case_class: Type[PythonTestCase] = PythonTestCase.class_factory( + test=test, python_test_version="version", mandatory=False + ) + instance = case_class(TestCaseExecution()) + + prefix = "already displayed content" + remaining = "\nremaining line 1\nremaining line 2\n" + output_file = tmp_path / "test_output.txt" + output_file.write_text(prefix + remaining) + + instance.file_output_path = output_file + instance._last_logged_position = len(prefix) + + with mock.patch( + "test_collections.matter.sdk_tests.support.python_testing.models.test_case" + ".logger" + ) as mock_logger: + await instance._log_remaining_content() + + mock_logger.info.assert_any_call("---- Remaining logs not captured by steps ----") + # _log_remaining_content logs one line at a time (batched/paced), not the + # whole remaining string in a single call - assert every line was logged. + remaining_lines = remaining.split("\n") + for line in remaining_lines: + mock_logger.log.assert_any_call(PYTHON_TEST_LEVEL, line) + assert mock_logger.log.call_count == len(remaining_lines) + mock_logger.info.assert_any_call("---- End of remaining logs ----") + assert instance._remaining_content_logged is True + + +@pytest.mark.asyncio +async def test_log_remaining_content_is_idempotent(tmp_path: Path) -> None: + """_log_remaining_content must only log the remaining content once, even + if called multiple times (e.g. from both execute() and cleanup()).""" + test = python_test_instance() + case_class: Type[PythonTestCase] = PythonTestCase.class_factory( + test=test, python_test_version="version", mandatory=False + ) + instance = case_class(TestCaseExecution()) + + prefix = "already displayed content" + remaining = "\nremaining line\n" + output_file = tmp_path / "test_output.txt" + output_file.write_text(prefix + remaining) + + instance.file_output_path = output_file + instance._last_logged_position = len(prefix) + + with mock.patch( + "test_collections.matter.sdk_tests.support.python_testing.models.test_case" + ".logger" + ) as mock_logger: + await instance._log_remaining_content() + await instance._log_remaining_content() + + # One log.log call per line on the first (non-idempotent) invocation; + # the second call must not log anything more. + assert mock_logger.log.call_count == len(remaining.split("\n")) + + +@pytest.mark.asyncio +async def test_display_batch_logs_logs_every_line(tmp_path: Path) -> None: + """display_batch_logs must log every line of the test output file, + wrapped in start/end banners.""" + test = python_test_instance() + case_class: Type[PythonTestCase] = PythonTestCase.class_factory( + test=test, python_test_version="version", mandatory=False + ) + instance = case_class(TestCaseExecution()) + + output_file = tmp_path / "test_output.txt" + output_file.write_text("line one\nline two\n") + instance.file_output_path = output_file + + with mock.patch( + "test_collections.matter.sdk_tests.support.python_testing.models.test_case" + ".logger" + ) as mock_logger: + await instance.display_batch_logs() + + mock_logger.info.assert_any_call("---- Start of Python test logs ----") + mock_logger.log.assert_any_call(PYTHON_TEST_LEVEL, "line one") + mock_logger.log.assert_any_call(PYTHON_TEST_LEVEL, "line two") + mock_logger.info.assert_any_call("---- End of Python test logs ----") + assert mock_logger.log.call_count == 2 + + +@pytest.mark.asyncio +async def test_display_batch_logs_is_idempotent(tmp_path: Path) -> None: + """display_batch_logs must only log the file's content once, even if + called multiple times.""" + test = python_test_instance() + case_class: Type[PythonTestCase] = PythonTestCase.class_factory( + test=test, python_test_version="version", mandatory=False + ) + instance = case_class(TestCaseExecution()) + + output_file = tmp_path / "test_output.txt" + output_file.write_text("line one\nline two\n") + instance.file_output_path = output_file + + with mock.patch( + "test_collections.matter.sdk_tests.support.python_testing.models.test_case" + ".logger" + ) as mock_logger: + await instance.display_batch_logs() + await instance.display_batch_logs() + + assert mock_logger.log.call_count == 2 + + +def test_read_file_incrementally_handles_split_utf8_character(tmp_path: Path) -> None: + """A multi-byte UTF-8 character whose bytes arrive split across two + incremental reads (very possible while the file is still growing) must + be correctly assembled once complete, not corrupted into a replacement + character in the meantime (regression test for issue #1072 follow-up: + _read_file_incrementally() used to decode each call's bytes in + isolation, with no memory of a previous call's incomplete trailing + bytes).""" + test = python_test_instance() + case_class: Type[PythonTestCase] = PythonTestCase.class_factory( + test=test, python_test_version="version", mandatory=False + ) + instance = case_class(TestCaseExecution()) + + full_text = "hello café world" # "é" is a 2-byte UTF-8 character + full_bytes = full_text.encode("utf-8") + # Split right after the first byte of "é", so it's incomplete on disk. + split_point = full_bytes.index("é".encode("utf-8")) + 1 + + output_file = tmp_path / "test_output.txt" + output_file.write_bytes(full_bytes[:split_point]) + instance.file_output_path = output_file + + first_read = instance._read_file_incrementally() + assert "�" not in first_read + assert first_read == "hello caf" + + # The rest of the character (and the rest of the file) now arrives. + output_file.write_bytes(full_bytes) + second_read = instance._read_file_incrementally() + assert second_read == full_text