operator
b7e45010c3
- Add clean line-by-line transcript style toggle ('b' key / /clean / /boxed)
- Implement right-click context menus for chat list, fleet agent panel, and transcript
- Add cooldown mode lock bypass (2-second double-confirm force sync)
- Implement dual copy support: clean text (strip reply metadata) vs full context
- Add clickable [📋 Copy] and [📑+ Context] buttons to message headers
- Add unit test suites for transcript cleaning, context menus, rate limits, and copy actions
193 lines
8.7 KiB
Python
193 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
test_scrollback.py — Comprehensive unit and integration test suite for:
|
|
1. Transcript scrollback rendering, line wrapping, and frame cache.
|
|
2. Jump-to-top ('g', Home) and bounds-capped scrolling ('k', Up arrow, PgUp).
|
|
3. Immediate downward movement ('j', Down arrow, PgDn) without 9999 offset blockage.
|
|
4. Jump-to-bottom ('G', End) re-engaging auto_scroll = True.
|
|
5. Thread and node switching resetting transcript scroll offset to 0.
|
|
6. Message send input line execution resetting scroll offset to bottom.
|
|
7. Mouse wheel scrolling within bounds.
|
|
"""
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import curses
|
|
|
|
REPO_ROOT = Path("/home/super/Projects/NetVM")
|
|
TUI_PATH = REPO_ROOT / "bin" / "muse-tui.py"
|
|
|
|
spec = importlib.util.spec_from_file_location("muse_tui", str(TUI_PATH))
|
|
muse_mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(muse_mod)
|
|
|
|
|
|
class TestScrollbackSubsystem(unittest.TestCase):
|
|
"""Test suite covering transcript scrollback, bounds clamping, and cache performance."""
|
|
|
|
def setUp(self):
|
|
self.mock_stdscr = MagicMock()
|
|
self.mock_stdscr.getmaxyx.return_value = (30, 100)
|
|
self.tui = muse_mod.MuseTUI(self.mock_stdscr, initial_mode="muse", initial_node="muse")
|
|
self.tui.safe_addstr = MagicMock()
|
|
|
|
# Seed realistic history cache (50 messages, multi-line)
|
|
test_msgs = []
|
|
for i in range(50):
|
|
role = "user" if i % 2 == 0 else "assistant"
|
|
test_msgs.append({
|
|
"role": role,
|
|
"text": f"Message {i}: This is a test message to simulate rich chat transcript history.\nLine 2 of message {i}",
|
|
"seq": i,
|
|
"message_id": f"msg-{i:04d}",
|
|
})
|
|
self.tui.data.history_cache[("muse", "test_thread")] = test_msgs
|
|
self.tui.data.active_node = "muse"
|
|
self.tui.data.active_thread_id = "test_thread"
|
|
self.tui.data.active_thread_title = "Test Thread"
|
|
|
|
def test_initial_render_and_line_caching(self):
|
|
"""Verify _render_transcript populates cache and calculates total lines."""
|
|
self.tui._render_transcript(y=2, x=24, h=26, w=76)
|
|
|
|
total_lines = self.tui.transcript_total_lines
|
|
self.assertGreater(total_lines, 50, "50 multi-line messages should yield > 50 rendered lines")
|
|
self.assertIsNotNone(self.tui._transcript_cache_key)
|
|
self.assertGreater(len(self.tui._transcript_cache_lines), 0)
|
|
self.assertTrue(self.tui.auto_scroll)
|
|
self.assertEqual(self.tui.transcript_scroll_offset, 0)
|
|
|
|
# Re-render with same parameters: should hit cache without recomputing lines
|
|
cached_lines_ref = self.tui._transcript_cache_lines
|
|
self.tui._render_transcript(y=2, x=24, h=26, w=76)
|
|
self.assertIs(self.tui._transcript_cache_lines, cached_lines_ref)
|
|
|
|
def test_jump_to_top_and_down_scrolling(self):
|
|
"""Verify 'g' jumps to top without 9999 offset bug, and 'j' immediately scrolls down."""
|
|
# 1. Initial render to establish dimensions
|
|
self.tui._render_transcript(y=2, x=24, h=26, w=76)
|
|
max_scroll = max(0, self.tui.transcript_total_lines - self.tui.transcript_height)
|
|
self.assertGreater(max_scroll, 0)
|
|
|
|
# 2. Press 'g' (or Home) to jump to top
|
|
self.tui.focus_pane = "transcript"
|
|
handled = self.tui._handle_normal_key(ord('g'))
|
|
self.assertTrue(handled)
|
|
self.assertFalse(self.tui.auto_scroll)
|
|
self.assertEqual(self.tui.transcript_scroll_offset, max_scroll, "Jump to top must equal max_scroll, not 9999")
|
|
|
|
# 3. Press 'j' (Down arrow) to scroll down by 3 lines
|
|
handled = self.tui._handle_normal_key(ord('j'))
|
|
self.assertTrue(handled)
|
|
self.assertEqual(self.tui.transcript_scroll_offset, max_scroll - 3, "Down arrow must immediately decrement offset")
|
|
|
|
# 4. Render frame: bounds check should preserve offset
|
|
self.tui._render_transcript(y=2, x=24, h=26, w=76)
|
|
self.assertEqual(self.tui.transcript_scroll_offset, max_scroll - 3)
|
|
|
|
def test_jump_to_bottom_re_engages_auto_scroll(self):
|
|
"""Verify 'G' (or End) resets offset to 0 and re-engages auto_scroll."""
|
|
self.tui._render_transcript(y=2, x=24, h=26, w=76)
|
|
self.tui.focus_pane = "transcript"
|
|
|
|
# Scroll up
|
|
self.tui._handle_normal_key(ord('k'))
|
|
self.assertFalse(self.tui.auto_scroll)
|
|
self.assertGreater(self.tui.transcript_scroll_offset, 0)
|
|
|
|
# Press 'G' to jump to bottom
|
|
handled = self.tui._handle_normal_key(ord('G'))
|
|
self.assertTrue(handled)
|
|
self.assertTrue(self.tui.auto_scroll)
|
|
self.assertEqual(self.tui.transcript_scroll_offset, 0)
|
|
|
|
def test_uncapped_scroll_up_prevented(self):
|
|
"""Verify repeated 'k' / PageUp keys never exceed max_scroll."""
|
|
self.tui._render_transcript(y=2, x=24, h=26, w=76)
|
|
self.tui.focus_pane = "transcript"
|
|
max_scroll = max(0, self.tui.transcript_total_lines - self.tui.transcript_height)
|
|
|
|
# Hammer scroll up 100 times
|
|
for _ in range(100):
|
|
self.tui._handle_normal_key(ord('k'))
|
|
|
|
self.assertEqual(self.tui.transcript_scroll_offset, max_scroll, "Scroll offset must be capped at max_scroll")
|
|
|
|
# Render frame
|
|
self.tui._render_transcript(y=2, x=24, h=26, w=76)
|
|
self.assertEqual(self.tui.transcript_scroll_offset, max_scroll)
|
|
|
|
def test_mouse_wheel_scrolling(self):
|
|
"""Verify mouse wheel up scrolls up and mouse wheel down scrolls back to auto_scroll."""
|
|
self.tui._render_transcript(y=2, x=24, h=26, w=76)
|
|
content_y, content_h = 2, 26
|
|
sidebar_w = 24
|
|
transcript_mx = sidebar_w + 10 # Mouse inside transcript pane
|
|
transcript_my = content_y + 5
|
|
|
|
# Mouse wheel up mask
|
|
wheel_up_mask = 0x10000
|
|
self.tui._handle_mouse(transcript_mx, transcript_my, wheel_up_mask)
|
|
self.assertFalse(self.tui.auto_scroll)
|
|
self.assertEqual(self.tui.transcript_scroll_offset, 3)
|
|
|
|
# Mouse wheel down mask (curses.BUTTON5_PRESSED or 0x200000)
|
|
wheel_down_mask = getattr(curses, "BUTTON5_PRESSED", 0x200000)
|
|
self.tui._handle_mouse(transcript_mx, transcript_my, wheel_down_mask)
|
|
self.assertEqual(self.tui.transcript_scroll_offset, 0)
|
|
self.assertTrue(self.tui.auto_scroll)
|
|
|
|
def test_target_switch_resets_scroll(self):
|
|
"""Verify switching active node or thread automatically resets scroll state."""
|
|
self.tui._render_transcript(y=2, x=24, h=26, w=76)
|
|
self.tui.focus_pane = "transcript"
|
|
self.tui._handle_normal_key(ord('g')) # Scrolled all the way up
|
|
self.assertFalse(self.tui.auto_scroll)
|
|
self.assertGreater(self.tui.transcript_scroll_offset, 0)
|
|
|
|
# Switch to another thread
|
|
self.tui.data.history_cache[("muse", "other_thread")] = [
|
|
{"role": "user", "text": "Hello new thread", "seq": 1, "message_id": "m1"}
|
|
]
|
|
self.tui.data.active_thread_id = "other_thread"
|
|
|
|
# Next render detects target switch
|
|
self.tui._render_transcript(y=2, x=24, h=26, w=76)
|
|
self.assertTrue(self.tui.auto_scroll, "Target switch must restore auto_scroll")
|
|
self.assertEqual(self.tui.transcript_scroll_offset, 0, "Target switch must reset transcript_scroll_offset")
|
|
|
|
def test_message_send_resets_scroll(self):
|
|
"""Verify sending a chat message resets scroll offset to follow live response."""
|
|
self.tui._render_transcript(y=2, x=24, h=26, w=76)
|
|
self.tui.focus_pane = "transcript"
|
|
self.tui._handle_normal_key(ord('g'))
|
|
self.assertFalse(self.tui.auto_scroll)
|
|
self.assertGreater(self.tui.transcript_scroll_offset, 0)
|
|
|
|
# Mock async send to avoid actual network / subprocess calls
|
|
with patch.object(self.tui, "_async_send_message"):
|
|
self.tui._execute_input_line("Test message from operator")
|
|
|
|
self.assertTrue(self.tui.auto_scroll, "Message send must restore auto_scroll")
|
|
self.assertEqual(self.tui.transcript_scroll_offset, 0, "Message send must reset scroll offset to 0")
|
|
|
|
def test_insert_mode_typing_and_batch_processing(self):
|
|
"""Verify insert mode appends characters and updates cursor cleanly."""
|
|
self.tui.editor_mode = "INSERT"
|
|
for ch in "hello netvm":
|
|
handled = self.tui._handle_insert_key(ord(ch))
|
|
self.assertTrue(handled)
|
|
self.assertEqual(self.tui.input_buf, "hello netvm")
|
|
self.assertEqual(self.tui.input_cursor, 11)
|
|
|
|
# Backspace
|
|
self.tui._handle_insert_key(curses.KEY_BACKSPACE)
|
|
self.assertEqual(self.tui.input_buf, "hello netv")
|
|
self.assertEqual(self.tui.input_cursor, 10)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|