[Feature] Improve CLI Realtime Logs (#100)

* Adding Execution Tree to the Realtime log with step jump feature

* Refining the refresh feature of the CLI's Realtime log

* Using now a per-client broadcast pattern for multiple simultaneous connections

Also, dropped events during reconnection are now buffered and replayed instead of silently lost.

* Fixing related unit tests for CLI

Author: antonio-amjr

tests/test_run/test_log_stream_handler.py

@@ -46,10 +46,11 @@ def test_is_running_initially_false(self):
             h = LogStreamHandler()
         assert h.is_running is False
 
-    def test_log_queue_is_queue(self):
+    def test_clients_is_empty_set(self):
         with patch("th_cli.test_run.log_stream_handler.LogsHTTPServer"):
             h = LogStreamHandler()
-        assert isinstance(h.log_queue, queue.Queue)
+        assert isinstance(h._clients, set)
+        assert len(h._clients) == 0
 
     def test_log_file_path_initially_none(self):
         with patch("th_cli.test_run.log_stream_handler.LogsHTTPServer"):
@@ -163,24 +164,23 @@ def test_sets_is_running_false(self):
 
         assert h.is_running is False
 
-    def test_puts_none_sentinel_into_queue(self):
+    def test_broadcasts_none_sentinel_on_stop(self):
         h, mock_srv = _make_handler()
         h.is_running = True
+        client_q = queue.Queue()
+        h._clients.add(client_q)
 
         with patch("th_cli.test_run.log_stream_handler.logger"):
             h.stop()
 
-        assert h.log_queue.get_nowait() is None
+        assert client_q.get_nowait() is None
 
-    def test_skips_sentinel_when_queue_full(self):
+    def test_stop_does_not_raise_when_client_queue_full(self):
         h, mock_srv = _make_handler()
         h.is_running = True
-        # Fill the queue to capacity
-        for _ in range(h.log_queue.maxsize):
-            try:
-                h.log_queue.put_nowait("entry")
-            except queue.Full:
-                break
+        client_q = queue.Queue(maxsize=1)
+        client_q.put_nowait("existing")  # fill to capacity
+        h._clients.add(client_q)
 
         with patch("th_cli.test_run.log_stream_handler.logger"):
             h.stop()  # must not raise
@@ -196,45 +196,54 @@ class TestAddLogEntry:
     def test_noop_when_not_running(self):
         h, _ = _make_handler()
         h.is_running = False
-        h.add_log_entry(message="ignored")
-        assert h.log_queue.empty()
+        with patch.object(h, "_broadcast") as mock_bcast:
+            h.add_log_entry(message="ignored")
+        mock_bcast.assert_not_called()
 
     def test_adds_entry_to_queue_when_running(self):
         h, _ = _make_handler()
         h.is_running = True
+        client_q = queue.Queue()
+        h._clients.add(client_q)
         h.add_log_entry(message="hello", level="INFO")
-        entry = h.log_queue.get_nowait()
+        entry = client_q.get_nowait()
         assert entry["message"] == "hello"
         assert entry["level"] == "INFO"
 
     def test_auto_generates_timestamp_when_not_provided(self):
         h, _ = _make_handler()
         h.is_running = True
+        client_q = queue.Queue()
+        h._clients.add(client_q)
         h.add_log_entry(message="msg")
-        entry = h.log_queue.get_nowait()
+        entry = client_q.get_nowait()
         assert "timestamp" in entry
         assert entry["timestamp"] is not None
 
     def test_uses_provided_timestamp(self):
         h, _ = _make_handler()
         h.is_running = True
+        client_q = queue.Queue()
+        h._clients.add(client_q)
         h.add_log_entry(message="msg", timestamp="2025-01-01T00:00:00")
-        entry = h.log_queue.get_nowait()
+        entry = client_q.get_nowait()
         assert entry["timestamp"] == "2025-01-01T00:00:00"
 
     def test_level_uppercased(self):
         h, _ = _make_handler()
         h.is_running = True
+        client_q = queue.Queue()
+        h._clients.add(client_q)
         h.add_log_entry(message="msg", level="warning")
-        entry = h.log_queue.get_nowait()
+        entry = client_q.get_nowait()
         assert entry["level"] == "WARNING"
 
     def test_silently_drops_when_queue_full(self):
         h, _ = _make_handler()
         h.is_running = True
-        # Fill queue to max
-        for _ in range(h.log_queue.maxsize):
-            h.log_queue.put_nowait({"message": "x"})
+        client_q = queue.Queue(maxsize=1)
+        client_q.put_nowait({"message": "x"})  # fill to capacity
+        h._clients.add(client_q)
         h.add_log_entry(message="overflow")  # must not raise
 
 

tests/test_run/test_logs_http_server.py

@@ -45,6 +45,12 @@ def _make_handler(path="/", server_attrs=None):
     handler.path = path
 
     mock_server = MagicMock()
+    # Set sensible defaults for the broadcast-pattern attributes so stream_logs()
+    # doesn't hit MagicMock internals (e.g. deepcopy of threading.Lock).
+    mock_server.active_clients = set()
+    mock_server.clients_lock = threading.Lock()
+    mock_server.tree_state = {}   # empty dict avoids deepcopy branch
+    mock_server.tree_lock = None
     for attr, value in (server_attrs or {}).items():
         setattr(mock_server, attr, value)
     handler.server = mock_server
@@ -64,6 +70,14 @@ def _make_handler(path="/", server_attrs=None):
     return handler
 
 
+def _pre_filled_queue(*items):
+    """Return a queue pre-loaded with items for use when patching queue.Queue."""
+    q = queue.Queue()
+    for item in items:
+        q.put(item)
+    return q
+
+
 # ---------------------------------------------------------------------------
 # do_GET routing
 # ---------------------------------------------------------------------------
@@ -202,49 +216,41 @@ def test_data_is_valid_json(self):
 @pytest.mark.unit
 class TestStreamLogs:
     def test_sends_sse_headers(self):
-        q = queue.Queue()
-        q.put(None)  # immediate end-of-stream
-        h = _make_handler(server_attrs={"log_queue": q})
-
-        with patch("th_cli.test_run.logs_http_server.logger"):
-            h.stream_logs()
+        h = _make_handler()
+        with patch("th_cli.test_run.logs_http_server.queue.Queue", return_value=_pre_filled_queue(None)):
+            with patch("th_cli.test_run.logs_http_server.logger"):
+                h.stream_logs()
 
         assert h._response_code == 200
         assert h._headers_sent.get("Content-Type") == "text/event-stream"
 
     def test_sends_end_event_on_none_sentinel(self):
-        q = queue.Queue()
-        q.put(None)
-        h = _make_handler(server_attrs={"log_queue": q})
+        h = _make_handler()
         sent_events = []
-        original_send = h._send_sse_event
         h._send_sse_event = lambda ev, data: sent_events.append(ev) or True
 
-        with patch("th_cli.test_run.logs_http_server.logger"):
-            h.stream_logs()
+        with patch("th_cli.test_run.logs_http_server.queue.Queue", return_value=_pre_filled_queue(None)):
+            with patch("th_cli.test_run.logs_http_server.logger"):
+                h.stream_logs()
 
         assert "end" in sent_events
 
     def test_streams_log_entries_from_queue(self):
-        q = queue.Queue()
-        q.put({"message": "hello", "level": "INFO", "timestamp": "2025-01-01"})
-        q.put(None)
-        h = _make_handler(server_attrs={"log_queue": q})
+        h = _make_handler()
         sent_events = []
         h._send_sse_event = lambda ev, data: sent_events.append((ev, data)) or True
 
-        with patch("th_cli.test_run.logs_http_server.logger"):
-            h.stream_logs()
+        entry = {"message": "hello", "level": "INFO", "timestamp": "2025-01-01"}
+        with patch("th_cli.test_run.logs_http_server.queue.Queue", return_value=_pre_filled_queue(entry, None)):
+            with patch("th_cli.test_run.logs_http_server.logger"):
+                h.stream_logs()
 
         log_events = [d for ev, d in sent_events if ev == "log"]
         assert len(log_events) == 1
         assert log_events[0]["message"] == "hello"
 
     def test_stops_when_send_sse_returns_false(self):
-        q = queue.Queue()
-        q.put({"message": "msg", "level": "INFO", "timestamp": "t"})
-        q.put({"message": "msg2", "level": "INFO", "timestamp": "t"})
-        h = _make_handler(server_attrs={"log_queue": q})
+        h = _make_handler()
         call_count = [0]
 
         def _send(ev, data):
@@ -254,33 +260,33 @@ def _send(ev, data):
             return False  # disconnect immediately
 
         h._send_sse_event = _send
+        entry1 = {"message": "msg", "level": "INFO", "timestamp": "t"}
+        entry2 = {"message": "msg2", "level": "INFO", "timestamp": "t"}
+        with patch("th_cli.test_run.logs_http_server.queue.Queue", return_value=_pre_filled_queue(entry1, entry2)):
+            with patch("th_cli.test_run.logs_http_server.logger"):
+                h.stream_logs()
 
-        with patch("th_cli.test_run.logs_http_server.logger"):
-            h.stream_logs()
-
-        # Should not have kept reading after disconnect
         assert call_count[0] <= 3
 
-    def test_no_log_queue_on_server_returns_early(self):
-        h = _make_handler(server_attrs={"log_queue": None})
+    def test_no_active_clients_on_server_returns_early(self):
+        h = _make_handler(server_attrs={"active_clients": None, "clients_lock": None})
         with patch("th_cli.test_run.logs_http_server.logger"):
             h.stream_logs()
-        # Should have sent 200 headers but not crashed
         assert h._response_code == 200
 
     def test_debug_log_at_100_entries(self):
-        """Covers line 134: `if sent_count % 100 == 0` debug message."""
-        q = queue.Queue()
-        # Put 100 log entries then sentinel
-        for i in range(100):
-            q.put({"message": f"msg{i}", "level": "INFO", "timestamp": "t"})
-        q.put(None)
-
-        h = _make_handler(server_attrs={"log_queue": q})
-        with patch("th_cli.test_run.logs_http_server.logger") as mock_logger:
-            h.stream_logs()
+        """Covers line `if sent_count % 100 == 0` debug message."""
+        entries = [{"message": f"msg{i}", "level": "INFO", "timestamp": "t"} for i in range(100)]
+        entries.append(None)  # sentinel
+        pre_q = queue.Queue()
+        for e in entries:
+            pre_q.put(e)
+
+        h = _make_handler()
+        with patch("th_cli.test_run.logs_http_server.queue.Queue", return_value=pre_q):
+            with patch("th_cli.test_run.logs_http_server.logger") as mock_logger:
+                h.stream_logs()
 
-        # 100 entries should have triggered the debug log
         mock_logger.debug.assert_called()
 
 
@@ -381,7 +387,8 @@ def test_server_thread_initially_none(self):
 class TestLogsHTTPServerStart:
     def test_creates_threading_http_server(self):
         srv = LogsHTTPServer(port=0)
-        q = queue.Queue()
+        clients = set()
+        lock = threading.Lock()
 
         with patch("th_cli.test_run.logs_http_server.ThreadingHTTPServer") as mock_cls:
             mock_ths = MagicMock()
@@ -390,23 +397,32 @@ def test_creates_threading_http_server(self):
                 mock_thread = MagicMock()
                 mock_thread_cls.return_value = mock_thread
                 with patch("th_cli.test_run.logs_http_server.logger"):
-                    srv.start(log_queue=q, test_run_title="Run")
+                    srv.start(active_clients=clients, clients_lock=lock, tree_state={}, test_run_title="Run")
 
         mock_cls.assert_called_once()
         mock_thread.start.assert_called_once()
 
     def test_sets_server_attributes(self):
         srv = LogsHTTPServer(port=0)
-        q = queue.Queue()
+        clients = set()
+        lock = threading.Lock()
 
         with patch("th_cli.test_run.logs_http_server.ThreadingHTTPServer") as mock_cls:
             mock_ths = MagicMock()
             mock_cls.return_value = mock_ths
             with patch("th_cli.test_run.logs_http_server.threading.Thread", return_value=MagicMock()):
                 with patch("th_cli.test_run.logs_http_server.logger"):
-                    srv.start(log_queue=q, test_run_title="MyTitle", local_ip="1.2.3.4", log_file_path="/tmp/f.log")
-
-        assert mock_ths.log_queue is q
+                    srv.start(
+                        active_clients=clients,
+                        clients_lock=lock,
+                        tree_state={},
+                        test_run_title="MyTitle",
+                        local_ip="1.2.3.4",
+                        log_file_path="/tmp/f.log",
+                    )
+
+        assert mock_ths.active_clients is clients
+        assert mock_ths.clients_lock is lock
         assert mock_ths.test_run_title == "MyTitle"
         assert mock_ths.local_ip == "1.2.3.4"
         assert mock_ths.log_file_path == "/tmp/f.log"
@@ -416,7 +432,7 @@ def test_propagates_oserror(self):
         with patch("th_cli.test_run.logs_http_server.ThreadingHTTPServer", side_effect=OSError("port in use")):
             with patch("th_cli.test_run.logs_http_server.logger"):
                 with pytest.raises(OSError):
-                    srv.start(log_queue=queue.Queue(), test_run_title="run")
+                    srv.start(active_clients=set(), clients_lock=threading.Lock(), tree_state={}, test_run_title="run")
 
 
 # ---------------------------------------------------------------------------