From 80800e9f199a4a234395fb2da774db9b2217fa35 Mon Sep 17 00:00:00 2001 From: "Antonio Melo Jr." Date: Fri, 17 Apr 2026 20:48:34 +0000 Subject: [PATCH] WIP: Adding Realtime Log feature to CLI --- poetry.lock | 19 +- pyproject.toml | 1 + tests/README.md | 5 +- tests/test_realtime_log_display.py | 481 ++++++++++++++++++++++++ th_cli/test_run/realtime_log_display.py | 303 +++++++++++++++ th_cli/test_run/websocket.py | 47 +++ 6 files changed, 846 insertions(+), 10 deletions(-) create mode 100644 tests/test_realtime_log_display.py create mode 100644 th_cli/test_run/realtime_log_display.py diff --git a/poetry.lock b/poetry.lock index 74fa12a..02e62c8 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.3.2 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.3.4 and should not be changed by hand. [[package]] name = "aioconsole" @@ -863,7 +863,7 @@ version = "4.0.0" description = "Python port of markdown-it. Markdown parsing, done right!" optional = false python-versions = ">=3.10" -groups = ["dev"] +groups = ["main", "dev"] files = [ {file = "markdown_it_py-4.0.0-py3-none-any.whl", hash = "sha256:87327c59b172c5011896038353a81343b6754500a08cd7a4973bb48c6d578147"}, {file = "markdown_it_py-4.0.0.tar.gz", hash = "sha256:cb0a2b4aa34f932c007117b194e945bd74e0ec24133ceb5bac59009cda1cb9f3"}, @@ -998,7 +998,7 @@ version = "0.1.2" description = "Markdown URL utilities" optional = false python-versions = ">=3.7" -groups = ["dev"] +groups = ["main", "dev"] files = [ {file = "mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8"}, {file = "mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba"}, @@ -1530,7 +1530,7 @@ version = "2.19.2" description = "Pygments is a syntax highlighting package written in Python." optional = false python-versions = ">=3.8" -groups = ["dev"] +groups = ["main", "dev"] files = [ {file = "pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b"}, {file = "pygments-2.19.2.tar.gz", hash = "sha256:636cb2477cec7f8952536970bc533bc43743542f70392ae026374600add5b887"}, @@ -1772,19 +1772,20 @@ use-chardet-on-py3 = ["chardet (>=3.0.2,<6)"] [[package]] name = "rich" -version = "14.1.0" +version = "13.9.4" description = "Render rich text, tables, progress bars, syntax highlighting, markdown and more to the terminal" optional = false python-versions = ">=3.8.0" -groups = ["dev"] +groups = ["main", "dev"] files = [ - {file = "rich-14.1.0-py3-none-any.whl", hash = "sha256:536f5f1785986d6dbdea3c75205c473f970777b4a0d6c6dd1b696aa05a3fa04f"}, - {file = "rich-14.1.0.tar.gz", hash = "sha256:e497a48b844b0320d45007cdebfeaeed8db2a4f4bcf49f15e455cfc4af11eaa8"}, + {file = "rich-13.9.4-py3-none-any.whl", hash = "sha256:6049d5e6ec054bf2779ab3358186963bac2ea89175919d699e378b99738c2a90"}, + {file = "rich-13.9.4.tar.gz", hash = "sha256:439594978a49a09530cff7ebc4b5c7103ef57baf48d5ea3184f21d9a2befa098"}, ] [package.dependencies] markdown-it-py = ">=2.2.0" pygments = ">=2.13.0,<3.0.0" +typing-extensions = {version = ">=4.0.0,<5.0", markers = "python_version < \"3.11\""} [package.extras] jupyter = ["ipywidgets (>=7.5.1,<9)"] @@ -2214,4 +2215,4 @@ dev = ["black (>=19.3b0) ; python_version >= \"3.6\"", "pytest (>=4.6.2)"] [metadata] lock-version = "2.1" python-versions = ">=3.10,<4.0" -content-hash = "a378266a49991f7a805d9433196caafaf4eeabbe5e1b13566dcdb71e355f49c6" +content-hash = "239dfe9731ea13632469150c6d2a459dc1775a2abd42cf1a33787b8d4634f632" diff --git a/pyproject.toml b/pyproject.toml index ade63c7..d5d5b31 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,6 +47,7 @@ dependencies = [ "tomli>=1.2.0,<3.0.0", "aiofiles>=23.0.0,<24.0.0", "ffmpeg-python>=0.2.0,<1.0.0", + "rich>=13.0.0,<14.0.0", ] [project.scripts] diff --git a/tests/README.md b/tests/README.md index 6d2ad70..500775b 100644 --- a/tests/README.md +++ b/tests/README.md @@ -12,11 +12,14 @@ tests/ ├── conftest.py # Shared fixtures and configuration ├── test_abort_testing.py # Tests for abort-testing command ├── test_available_tests.py # Tests for available-tests command +├── test_ffmpeg_converter.py # Tests for FFmpeg stream converter ├── test_project_commands.py # Tests for all project commands +├── test_realtime_log_display.py # Tests for realtime log display ├── test_run_tests.py # Tests for run-tests command -├── test_test_run_execution.py # Tests for test-run-execution command +├── test_test_run_execution.py # Tests for test-run-execution command ├── test_test_runner_status.py # Tests for test-runner-status command ├── test_utils.py # Tests for utility functions +├── test_validation.py # Tests for validation functions └── README.md # This file ``` diff --git a/tests/test_realtime_log_display.py b/tests/test_realtime_log_display.py new file mode 100644 index 0000000..7959b22 --- /dev/null +++ b/tests/test_realtime_log_display.py @@ -0,0 +1,481 @@ +# +# Copyright (c) 2026 Project CHIP Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +"""Tests for the realtime log display functionality.""" + +import time +from unittest.mock import Mock, patch + +import pytest + +from th_cli.test_run.realtime_log_display import ( + MAX_DISPLAY_LINES, + REALTIME_LOG_STEP_PATTERNS, + REALTIME_LOG_TEST_CASES, + RealtimeLogDisplay, + get_realtime_display, + reset_realtime_display, +) + + +@pytest.mark.unit +class TestRealtimeLogDisplayConfiguration: + """Test cases for realtime log display configuration constants.""" + + def test_realtime_log_step_patterns_defined(self): + """Test that step patterns are properly defined.""" + assert isinstance(REALTIME_LOG_STEP_PATTERNS, list) + assert len(REALTIME_LOG_STEP_PATTERNS) > 0 + assert "Start Python test" in REALTIME_LOG_STEP_PATTERNS + assert "Show test logs" in REALTIME_LOG_STEP_PATTERNS + + def test_realtime_log_test_cases_defined(self): + """Test that test case IDs are properly defined.""" + assert isinstance(REALTIME_LOG_TEST_CASES, list) + assert len(REALTIME_LOG_TEST_CASES) > 0 + assert "TC-SC-3.5" in REALTIME_LOG_TEST_CASES + + def test_max_display_lines_is_positive(self): + """Test that max display lines is a positive integer.""" + assert isinstance(MAX_DISPLAY_LINES, int) + assert MAX_DISPLAY_LINES > 0 + + +@pytest.mark.unit +class TestRealtimeLogDisplayInit: + """Test cases for RealtimeLogDisplay initialization.""" + + def test_init_with_default_max_lines(self): + """Test initialization with default max lines.""" + display = RealtimeLogDisplay() + assert display.max_lines == MAX_DISPLAY_LINES + assert display.is_active is False + assert len(display.log_buffer) == 0 + + def test_init_with_custom_max_lines(self): + """Test initialization with custom max lines.""" + custom_lines = 20 + display = RealtimeLogDisplay(max_lines=custom_lines) + assert display.max_lines == custom_lines + assert display.log_buffer.maxlen == custom_lines + + def test_init_sets_initial_state(self): + """Test that initialization sets correct initial state.""" + display = RealtimeLogDisplay() + assert display.live_display is None + assert display.current_step_title is None + assert display.current_test_case_id is None + + +@pytest.mark.unit +class TestShouldDisplayForStep: + """Test cases for should_display_for_step method.""" + + def test_returns_true_for_configured_test_case(self): + """Test that configured test cases always show logs.""" + display = RealtimeLogDisplay() + result = display.should_display_for_step("Any step", test_case_id="TC-SC-3.5") + assert result is True + + def test_returns_true_for_matching_step_pattern(self): + """Test that matching step patterns return True.""" + display = RealtimeLogDisplay() + assert display.should_display_for_step("Start Python test") is True + assert display.should_display_for_step("Show test logs") is True + assert display.should_display_for_step("Commissioning device") is True + + def test_returns_false_for_non_matching_step(self): + """Test that non-matching steps return False.""" + display = RealtimeLogDisplay() + result = display.should_display_for_step("Random step title") + assert result is False + + def test_case_insensitive_matching(self): + """Test that pattern matching is case insensitive.""" + display = RealtimeLogDisplay() + assert display.should_display_for_step("start python test") is True + assert display.should_display_for_step("START PYTHON TEST") is True + assert display.should_display_for_step("Start Python Test") is True + + def test_partial_pattern_matching(self): + """Test that partial pattern matching works.""" + display = RealtimeLogDisplay() + # "websocket" pattern should match "WebSocket connection" + assert display.should_display_for_step("WebSocket connection") is True + assert display.should_display_for_step("Establish websocket") is True + + +@pytest.mark.unit +class TestStartStopDisplay: + """Test cases for start_display and stop_display methods.""" + + def test_start_display_sets_active_state(self): + """Test that start_display sets the active state.""" + display = RealtimeLogDisplay() + + with patch.object(display, 'rich_enabled', True): + with patch('th_cli.test_run.realtime_log_display.Live') as mock_live: + mock_live_instance = Mock() + mock_live.return_value = mock_live_instance + + display.start_display("Test Step", test_case_id="TC-SC-3.5") + + assert display.is_active is True + assert display.current_step_title == "Test Step" + assert display.current_test_case_id == "TC-SC-3.5" + + def test_start_display_clears_buffer(self): + """Test that start_display clears the log buffer.""" + display = RealtimeLogDisplay() + # Add a tuple to buffer (time_str, level_text, level_style, message) + display.log_buffer.append(("12:00:00", "INFO", "blue", "old log")) + + with patch.object(display, 'rich_enabled', True): + with patch('th_cli.test_run.realtime_log_display.Live'): + display.start_display("Test Step") + + assert len(display.log_buffer) == 0 + + def test_start_display_stops_previous_display(self): + """Test that starting a new display stops the previous one.""" + display = RealtimeLogDisplay() + + with patch.object(display, 'rich_enabled', True): + with patch('th_cli.test_run.realtime_log_display.Live'): + display.start_display("First Step") + first_active = display.is_active + + display.start_display("Second Step") + + assert first_active is True + assert display.is_active is True + assert display.current_step_title == "Second Step" + + def test_start_display_skips_when_rich_disabled(self): + """Test that start_display skips when Rich is disabled.""" + display = RealtimeLogDisplay() + display.rich_enabled = False + + display.start_display("Test Step") + + assert display.is_active is False + assert display.live_display is None + + def test_stop_display_clears_state(self): + """Test that stop_display clears the state.""" + display = RealtimeLogDisplay() + + with patch.object(display, 'rich_enabled', True): + with patch('th_cli.test_run.realtime_log_display.Live'): + display.start_display("Test Step") + display.stop_display() + + assert display.is_active is False + assert display.live_display is None + assert display.current_step_title is None + + def test_stop_display_when_not_active(self): + """Test that stop_display handles inactive state gracefully.""" + display = RealtimeLogDisplay() + + # Should not raise an exception + display.stop_display() + + assert display.is_active is False + + +@pytest.mark.unit +class TestAddLog: + """Test cases for add_log method.""" + + def test_add_log_when_not_active(self): + """Test that add_log does nothing when display is not active.""" + display = RealtimeLogDisplay() + + display.add_log("Test message") + + assert len(display.log_buffer) == 0 + + def test_add_log_adds_to_buffer(self): + """Test that add_log adds formatted message to buffer.""" + display = RealtimeLogDisplay() + display.is_active = True + + with patch.object(display, 'live_display', Mock()): + display.add_log("Test message", level="INFO") + + assert len(display.log_buffer) == 1 + # Buffer contains tuples: (time_str, level_text, level_style, message) + time_str, level_text, level_style, message = display.log_buffer[0] + assert "INFO" in level_text + assert message == "Test message" + + def test_add_log_with_timestamp(self): + """Test that add_log includes timestamp.""" + display = RealtimeLogDisplay() + display.is_active = True + + with patch.object(display, 'live_display', Mock()): + timestamp = time.time() + display.add_log("Test message", level="INFO", timestamp=timestamp) + + assert len(display.log_buffer) == 1 + # Should contain time format HH:MM:SS in first element of tuple + time_str, _, _, _ = display.log_buffer[0] + assert ":" in time_str + + def test_add_log_truncates_long_messages(self): + """Test that add_log truncates very long messages.""" + display = RealtimeLogDisplay() + display.is_active = True + + with patch.object(display, 'live_display', Mock()): + long_message = "x" * 300 + display.add_log(long_message, level="INFO") + + # Message should be truncated with "..." + _, _, _, message = display.log_buffer[0] + assert "..." in message + assert len(message) < len(long_message) + + def test_add_log_respects_buffer_limit(self): + """Test that log buffer respects max_lines limit.""" + display = RealtimeLogDisplay(max_lines=5) + display.is_active = True + + with patch.object(display, 'live_display', Mock()): + # Add more logs than buffer size + for i in range(10): + display.add_log(f"Message {i}", level="INFO") + + # Buffer should not exceed max_lines + assert len(display.log_buffer) == 5 + # Should contain most recent messages + _, _, _, message = display.log_buffer[-1] + assert "Message 9" == message + + def test_add_log_updates_live_display(self): + """Test that add_log updates the live display.""" + display = RealtimeLogDisplay() + display.is_active = True + + mock_live = Mock() + display.live_display = mock_live + + display.add_log("Test message", level="INFO") + + # Should call update on live display + mock_live.update.assert_called_once() + + +@pytest.mark.unit +class TestIsDisplaying: + """Test cases for is_displaying method.""" + + def test_is_displaying_returns_active_state(self): + """Test that is_displaying returns the current active state.""" + display = RealtimeLogDisplay() + + assert display.is_displaying() is False + + display.is_active = True + assert display.is_displaying() is True + + display.is_active = False + assert display.is_displaying() is False + + +@pytest.mark.unit +class TestGlobalSingleton: + """Test cases for global singleton functions.""" + + def test_get_realtime_display_returns_singleton(self): + """Test that get_realtime_display returns the same instance.""" + # Reset first to ensure clean state + reset_realtime_display() + + instance1 = get_realtime_display() + instance2 = get_realtime_display() + + assert instance1 is instance2 + + def test_reset_realtime_display_creates_new_instance(self): + """Test that reset creates a new instance.""" + instance1 = get_realtime_display() + reset_realtime_display() + instance2 = get_realtime_display() + + assert instance1 is not instance2 + + def test_reset_stops_active_display(self): + """Test that reset stops any active display.""" + display = get_realtime_display() + + with patch.object(display, 'rich_enabled', True): + with patch('th_cli.test_run.realtime_log_display.Live'): + display.start_display("Test") + assert display.is_active is True + + reset_realtime_display() + + # Original instance should have been stopped + assert display.is_active is False + + +@pytest.mark.unit +class TestGetLevelStyle: + """Test cases for _get_level_style method.""" + + def test_get_level_style_known_levels(self): + """Test that known log levels return appropriate styles.""" + display = RealtimeLogDisplay() + + style, text = display._get_level_style("DEBUG") + assert text == "DEBUG " + assert style is not None + + style, text = display._get_level_style("INFO") + assert text == "INFO " + assert style is not None + + style, text = display._get_level_style("WARNING") + assert text == "WARNING " + assert style is not None + + style, text = display._get_level_style("ERROR") + assert text == "ERROR " + assert style is not None + + def test_get_level_style_unknown_level(self): + """Test that unknown log level uses default style.""" + display = RealtimeLogDisplay() + + style, text = display._get_level_style("CUSTOM") + assert "CUSTOM" in text + assert style is not None + + +@pytest.mark.unit +class TestRenderDisplay: + """Test cases for _render_display method.""" + + def test_render_display_with_empty_buffer(self): + """Test rendering with empty log buffer.""" + display = RealtimeLogDisplay() + + panel = display._render_display() + + # Should return a Panel object + assert panel is not None + + def test_render_display_with_logs(self): + """Test rendering with log messages.""" + display = RealtimeLogDisplay() + display.is_active = True + + with patch.object(display, 'live_display', Mock()): + display.add_log("Test message 1", level="INFO") + display.add_log("Test message 2", level="DEBUG") + + panel = display._render_display() + + # Should return a Panel object + assert panel is not None + + def test_render_display_with_test_case_id(self): + """Test rendering includes test case ID in title.""" + display = RealtimeLogDisplay() + display.current_test_case_id = "TC-SC-3.5" + + panel = display._render_display() + + # Panel should include test case ID in title + assert panel is not None + + +@pytest.mark.unit +class TestRealtimeLogDisplayIntegration: + """Integration tests for realtime log display.""" + + def test_complete_display_lifecycle(self): + """Test complete lifecycle: start, add logs, stop.""" + display = RealtimeLogDisplay() + + # Should not display initially + assert display.is_displaying() is False + + with patch.object(display, 'rich_enabled', True): + with patch('th_cli.test_run.realtime_log_display.Live'): + # Start display + display.start_display("Test Step", test_case_id="TC-SC-3.5") + assert display.is_displaying() is True + + # Add some logs + display.add_log("Log 1", level="INFO") + display.add_log("Log 2", level="DEBUG") + assert len(display.log_buffer) == 2 + + # Stop display + display.stop_display() + assert display.is_displaying() is False + + def test_multiple_start_stop_cycles(self): + """Test multiple start/stop cycles work correctly.""" + display = RealtimeLogDisplay() + + with patch.object(display, 'rich_enabled', True): + with patch('th_cli.test_run.realtime_log_display.Live'): + # First cycle + display.start_display("Step 1") + display.add_log("Log 1") + display.stop_display() + + # Second cycle + display.start_display("Step 2") + # Buffer should be cleared + assert len(display.log_buffer) == 0 + display.add_log("Log 2") + assert len(display.log_buffer) == 1 + display.stop_display() + + +@pytest.mark.unit +class TestThreadSafety: + """Test cases for thread safety.""" + + def test_lock_used_for_start_display(self): + """Test that lock is used when starting display.""" + display = RealtimeLogDisplay() + + with patch.object(display.lock, 'acquire') as mock_acquire: + with patch.object(display.lock, 'release') as mock_release: + # Context manager should acquire and release lock + with display.lock: + pass + + mock_acquire.assert_called() + mock_release.assert_called() + + def test_lock_used_for_add_log(self): + """Test that lock is used when adding logs.""" + display = RealtimeLogDisplay() + display.is_active = True + + with patch.object(display, 'live_display', Mock()): + # Lock should be used internally + display.add_log("Test") + + # If we got here without deadlock, thread safety is working + assert True diff --git a/th_cli/test_run/realtime_log_display.py b/th_cli/test_run/realtime_log_display.py new file mode 100644 index 0000000..37f5bdb --- /dev/null +++ b/th_cli/test_run/realtime_log_display.py @@ -0,0 +1,303 @@ +# +# Copyright (c) 2026 Project CHIP Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +""" +Real-time log display manager for CLI test execution. + +This module provides functionality to display Python test container logs +in real-time during specific test steps, especially for steps that require +user interaction or feedback (like TC-SC-3.5 QR code/websocket prompts). +""" +import os +import threading +from collections import deque +from datetime import datetime +from typing import Optional + +from loguru import logger +from rich.console import Console, RenderableType +from rich.live import Live +from rich.panel import Panel +from rich.text import Text + +# Configuration: Test step patterns that should show real-time logs +# These patterns match step titles that require real-time feedback +REALTIME_LOG_STEP_PATTERNS = [ + "Start Python test", + "Show test logs", + "Commissioning", + "Pairing", + "QR Code", + "Manual Pairing Code", + "websocket", + "WebSocket", +] + +# Configuration: Test case IDs that should show real-time logs for all steps +REALTIME_LOG_TEST_CASES = [ + "TC-SC-3.5", # Requires showing QR code/websocket URL during execution + "TC-SC-3.6", + "TC-SC-5.1", + "TC-SC-5.2", + "TC-ACE-1.3" +] + +# Maximum number of log lines to display in the live view +MAX_DISPLAY_LINES = 15 + +# Check if we're in a proper terminal environment +ENABLE_RICH_DISPLAY = os.isatty(1) if hasattr(os, 'isatty') else True + + +class RealtimeLogDisplay: + """ + Manages real-time display of Python test container logs during test execution. + + This class provides a thread-safe way to display logs in real-time using + Rich's Live display feature. It maintains a buffer of recent log messages + and updates the display asynchronously. + """ + + def __init__(self, max_lines: int = MAX_DISPLAY_LINES): + """ + Initialize the realtime log display manager. + + Args: + max_lines: Maximum number of log lines to display (default: 15) + """ + self.max_lines = max_lines + self.log_buffer: deque[tuple[str, str, str, str]] = deque(maxlen=max_lines) + self.is_active = False + self.live_display: Optional[Live] = None + self.console = Console() + self.lock = threading.RLock() + self.current_step_title: Optional[str] = None + self.current_test_case_id: Optional[str] = None + + # Track if Rich display is available + self.rich_enabled = ENABLE_RICH_DISPLAY + + def should_display_for_step(self, step_title: str, test_case_id: Optional[str] = None) -> bool: + """ + Determine if real-time logs should be displayed for a given step. + + Args: + step_title: The title of the test step + test_case_id: Optional test case ID (e.g., "TC-SC-3.5") + + Returns: + True if logs should be displayed in real-time for this step + """ + # Check if test case is in the list of cases that always show logs + if test_case_id and test_case_id in REALTIME_LOG_TEST_CASES: + return True + + # Check if step title matches any of the patterns + step_title_lower = step_title.lower() + return any(pattern.lower() in step_title_lower for pattern in REALTIME_LOG_STEP_PATTERNS) + + def start_display(self, step_title: str, test_case_id: Optional[str] = None) -> None: + """ + Start displaying real-time logs for a test step. + + Args: + step_title: The title of the test step being executed + test_case_id: Optional test case ID + """ + if not self.rich_enabled: + logger.debug("Rich display not available, skipping real-time log display") + return + + with self.lock: + if self.is_active: + logger.debug("Real-time log display already active, stopping previous display") + self.stop_display() + + self.current_step_title = step_title + self.current_test_case_id = test_case_id + self.log_buffer.clear() + self.is_active = True + + # Create Live display with auto-refresh + try: + self.live_display = Live( + self._render_display(), + console=self.console, + refresh_per_second=4, # Update 4 times per second (smooth but not excessive) + transient=True, # Clear display when done + ) + self.live_display.start() + logger.debug(f"Started real-time log display for step: {step_title}") + except Exception as e: + logger.warning(f"Failed to start Rich live display: {e}") + self.is_active = False + self.live_display = None + + def stop_display(self) -> None: + """Stop displaying real-time logs.""" + with self.lock: + if not self.is_active: + return + + try: + if self.live_display: + self.live_display.stop() + self.live_display = None + logger.debug("Stopped real-time log display") + except Exception as e: + logger.warning(f"Error stopping live display: {e}") + finally: + self.is_active = False + self.current_step_title = None + self.current_test_case_id = None + + def add_log(self, log_message: str, level: str = "INFO", timestamp: Optional[float] = None) -> None: + """ + Add a log message to the display buffer. + + Args: + log_message: The log message to display + level: Log level (INFO, DEBUG, WARNING, ERROR, etc.) + timestamp: Optional timestamp for the log message + """ + if not self.is_active: + return + + with self.lock: + # Format timestamp + if timestamp: + try: + dt = datetime.fromtimestamp(timestamp) + time_str = dt.strftime("%H:%M:%S") + except (ValueError, OSError): + time_str = "??:??:??" + else: + time_str = datetime.now().strftime("%H:%M:%S") + + # Get log level style + level_style, level_text = self._get_level_style(level.upper()) + + # Clean and truncate message if too long + clean_message = log_message.strip() + # if len(clean_message) > 180: + # clean_message = clean_message[:177] + "..." + + # Store as tuple for rendering + self.log_buffer.append((time_str, level_text, level_style, clean_message)) + + # Update the live display + if self.live_display and self.is_active: + try: + self.live_display.update(self._render_display()) + except Exception as e: + logger.debug(f"Error updating live display: {e}") + + def _get_level_style(self, level: str) -> tuple[str, str]: + """ + Get style and formatted text for log level. + + Args: + level: Log level string + + Returns: + Tuple of (style_color, formatted_level_text) + """ + level_colors = { + "DEBUG": "dim cyan", + "INFO": "blue", + "WARNING": "yellow", + "ERROR": "red", + "CRITICAL": "bold red", + } + color = level_colors.get(level, "white") + return (color, f"{level:8}") + + def _render_display(self) -> RenderableType: + """ + Render the current log display. + + Returns: + A Rich renderable object (Panel) containing the log display + """ + with self.lock: + # Build the log display text + if not self.log_buffer: + content = Text("Waiting for logs...", style="dim italic") + else: + # Build Text object with proper styling (timestamps, levels, messages) + content = Text() + for time_str, level_text, level_style, message in self.log_buffer: + content.append(f"[{time_str}] ", style="dim") + content.append(level_text, style=level_style) + content.append(f" {message}\n") + + # Remove trailing newline + if content.plain: + content = content[:-1] + + # Create title with step information + title = "Real-time Test Logs" + if self.current_test_case_id: + title = f"Real-time Test Logs - {self.current_test_case_id}" + + # Create panel with border + panel = Panel( + content, + title=title, + title_align="left", + border_style="cyan", + padding=(0, 1), + ) + + return panel + + def is_displaying(self) -> bool: + """ + Check if the display is currently active. + + Returns: + True if display is active, False otherwise + """ + return self.is_active + + +# Global singleton instance for use across the CLI +_global_display_instance: Optional[RealtimeLogDisplay] = None +_instance_lock = threading.Lock() + + +def get_realtime_display() -> RealtimeLogDisplay: + """ + Get the global RealtimeLogDisplay instance (singleton pattern). + + Returns: + The global RealtimeLogDisplay instance + """ + global _global_display_instance + + with _instance_lock: + if _global_display_instance is None: + _global_display_instance = RealtimeLogDisplay() + return _global_display_instance + + +def reset_realtime_display() -> None: + """Reset the global realtime display instance (useful for testing).""" + global _global_display_instance + + with _instance_lock: + if _global_display_instance and _global_display_instance.is_active: + _global_display_instance.stop_display() + _global_display_instance = None diff --git a/th_cli/test_run/websocket.py b/th_cli/test_run/websocket.py index e60da91..c19becc 100644 --- a/th_cli/test_run/websocket.py +++ b/th_cli/test_run/websocket.py @@ -40,6 +40,7 @@ from th_cli.shared_constants import MessageTypeEnum from .prompt_manager import handle_file_upload_request, handle_prompt +from .realtime_log_display import get_realtime_display from .socket_schemas import ( PromptRequest, SocketMessage, @@ -71,6 +72,10 @@ def __init__( # 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]] = {} + # Track current test case for real-time log display + self.current_test_case_id: str | None = None + # Get real-time log display instance + self.realtime_display = get_realtime_display() async def connect_websocket(self) -> None: try: @@ -167,6 +172,12 @@ async def __log_test_run_update(self, update: TestRunUpdate) -> None: colored_state = colorize_state(update.state.value) click.echo(f"{test_run_text} {colored_state}") + # Ensure real-time display is stopped when test run completes + if update.state.value in ("passed", "failed", "error", "cancelled"): + if self.realtime_display.is_displaying(): + logger.debug("Stopping real-time log display as test run completed") + self.realtime_display.stop_display() + async def __display_manual_pairing_code(self) -> None: """Fetch and display manual pairing code after SDK container has started.""" try: @@ -227,6 +238,12 @@ def __log_test_case_update(self, update: TestCaseUpdate) -> None: colored_state = colorize_state(update.state.value) click.echo(f" - {colored_title} {colored_state}") + # Update current test case ID for real-time log display + if update.state.value == "executing": + self.current_test_case_id = public_id + elif update.state.value in ("passed", "failed", "error", "cancelled"): + self.current_test_case_id = None + # Log any errors when a test case fails if update.state.value in ("failed", "error"): all_errors = [] @@ -278,6 +295,17 @@ def __log_test_step_update(self, update: TestStepUpdate) -> None: colored_state = colorize_state(update.state.value) click.echo(f" - {colored_title} {colored_state}") + # Start real-time log display if this step should show logs + if update.state.value == "executing": + if self.realtime_display.should_display_for_step(title, self.current_test_case_id): + logger.debug(f"Starting real-time log display for step: {title}") + self.realtime_display.start_display(title, self.current_test_case_id) + elif update.state.value in ("passed", "failed", "error", "cancelled"): + # Stop real-time log display when step completes + if self.realtime_display.is_displaying(): + logger.debug(f"Stopping real-time log display for step: {title}") + self.realtime_display.stop_display() + # Track test step errors for later use in test case update if update.errors: case_key = (update.test_suite_execution_index, update.test_case_execution_index) @@ -286,7 +314,26 @@ def __log_test_step_update(self, update: TestStepUpdate) -> None: def __handle_log_record(self, records: list[TestLogRecord]) -> None: for record in records: + # Log to logger as before logger.log(record.level, record.message) + + # Also feed to real-time display if active + if self.realtime_display.is_displaying(): + # Extract timestamp - handle both float and string formats + timestamp = None + if isinstance(record.timestamp, float): + timestamp = record.timestamp + elif isinstance(record.timestamp, str): + try: + timestamp = float(record.timestamp) + except (ValueError, TypeError): + pass + + self.realtime_display.add_log( + log_message=record.message, + level=record.level, + timestamp=timestamp + ) def __suite(self, index: int) -> TestSuiteExecution: return self.run.test_suite_executions[index]