feat(tui): clean transcript style, right-click context menus, rate limits & dual copy
- 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
This commit is contained in:
operator
@@ -0,0 +1,122 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
test_agent_ranking.py — Unit test suite for Fleet Agent interaction ranking & sorting:
|
||||
1. Agent interaction recording on true input / chat [insert].
|
||||
2. Fleet agent sorting/ranking by last interaction descending.
|
||||
3. Stable fallback ordering for un-interacted agents.
|
||||
4. Persistence and seeding of interaction timestamps.
|
||||
5. format_recency helper accuracy.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
import time
|
||||
import tempfile
|
||||
import json
|
||||
|
||||
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 TestAgentRanking(unittest.TestCase):
|
||||
"""Test suite covering fleet agent interaction sorting & recency tracking."""
|
||||
|
||||
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()
|
||||
self.tui.data.history_cache = {}
|
||||
self.tui.data.read_msg_counts = {}
|
||||
|
||||
def test_format_recency(self):
|
||||
"""Verify format_recency produces compact, accurate recency strings."""
|
||||
now = time.time()
|
||||
self.assertEqual(muse_mod.format_recency(0.0), "never")
|
||||
self.assertEqual(muse_mod.format_recency(None), "never")
|
||||
self.assertEqual(muse_mod.format_recency(now - 10), "10s ago")
|
||||
self.assertEqual(muse_mod.format_recency(now - 120), "2m ago")
|
||||
self.assertEqual(muse_mod.format_recency(now - 7200), "2h ago")
|
||||
self.assertEqual(muse_mod.format_recency(now - 90000), "1d ago")
|
||||
|
||||
def test_sort_nodes_by_interaction(self):
|
||||
"""Verify sort_nodes ranks nodes by last interaction timestamp descending."""
|
||||
data = self.tui.data
|
||||
now = time.time()
|
||||
|
||||
# Set specific timestamps
|
||||
data.agent_interactions["muse"] = now - 500
|
||||
data.agent_interactions["pip"] = now - 100
|
||||
data.agent_interactions["646"] = now - 10 # Most recent
|
||||
data.agent_interactions["opm"] = 0.0 # Never
|
||||
data.agent_interactions["def"] = 0.0 # Never
|
||||
|
||||
data.sort_nodes()
|
||||
|
||||
# Most recent first: 646, then pip, then muse
|
||||
self.assertEqual(data.nodes[0], "646")
|
||||
self.assertEqual(data.nodes[1], "pip")
|
||||
self.assertEqual(data.nodes[2], "muse")
|
||||
# Un-interacted agents should be at the end, retaining baseline order
|
||||
self.assertIn("opm", data.nodes[3:])
|
||||
self.assertIn("def", data.nodes[3:])
|
||||
|
||||
def test_record_interaction_promotes_node_to_top(self):
|
||||
"""Verify record_interaction immediately updates timestamp and promotes node to rank 1."""
|
||||
data = self.tui.data
|
||||
# Ensure 'pip' is not at index 0 initially
|
||||
data.record_interaction("646")
|
||||
self.assertEqual(data.nodes[0], "646")
|
||||
|
||||
# Now operator directly interacts with 'pip'
|
||||
t0 = time.time()
|
||||
data.record_interaction("pip")
|
||||
self.assertEqual(data.nodes[0], "pip", "'pip' must be promoted to rank 1 (index 0)")
|
||||
self.assertEqual(data.nodes[1], "646", "'646' must be rank 2")
|
||||
self.assertGreaterEqual(data.agent_interactions["pip"], t0)
|
||||
|
||||
def test_chat_message_submission_triggers_ranking(self):
|
||||
"""Verify submitting chat message in INSERT mode promotes the active node."""
|
||||
self.tui.editor_mode = "INSERT"
|
||||
self.tui.data.active_node = "def"
|
||||
self.tui.data.active_thread_id = "test-session-123"
|
||||
self.tui.data.active_thread_title = "Def Task Session"
|
||||
|
||||
# Mock async send to avoid subprocess
|
||||
with patch.object(self.tui, "_async_send_message"):
|
||||
self.tui._execute_input_line("Deploy updated service configuration")
|
||||
|
||||
# 'def' was chatted with directly; it must now be rank 1 at index 0
|
||||
self.assertEqual(self.tui.data.nodes[0], "def", "Chat message send must promote target node to top of list")
|
||||
|
||||
def test_persistence_of_interactions(self):
|
||||
"""Verify interaction records persist and load from json correctly."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
fake_int_file = Path(tmpdir) / "agent_interactions.json"
|
||||
now = time.time()
|
||||
fake_int_file.write_text(json.dumps({"opm": now - 50, "dev": now - 10}))
|
||||
|
||||
with patch("pathlib.Path.home", return_value=Path(tmpdir)), \
|
||||
patch.object(muse_mod.Path, "home", return_value=Path(tmpdir)):
|
||||
# Adjust path to match ~/.config/muse-cli/
|
||||
cfg_dir = Path(tmpdir) / ".config" / "muse-cli"
|
||||
cfg_dir.mkdir(parents=True, exist_ok=True)
|
||||
(cfg_dir / "agent_interactions.json").write_text(json.dumps({"opm": now - 50, "dev": now - 10}))
|
||||
|
||||
mgr = muse_mod.FleetDataManager()
|
||||
mgr.stop() # Stop background poller
|
||||
|
||||
self.assertAlmostEqual(mgr.agent_interactions.get("dev", 0.0), now - 10, delta=1.0)
|
||||
self.assertAlmostEqual(mgr.agent_interactions.get("opm", 0.0), now - 50, delta=1.0)
|
||||
self.assertEqual(mgr.nodes[0], "dev", "Persisted recent node 'dev' must be rank 1")
|
||||
self.assertEqual(mgr.nodes[1], "opm", "'opm' must be rank 2")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,87 @@
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
import importlib.util
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path("/home/super/Projects/NetVM")
|
||||
sys.path.insert(0, str(REPO_ROOT / "bin"))
|
||||
|
||||
|
||||
class TestBlockedApprovals(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
spec = importlib.util.spec_from_file_location("muse_tui", str(REPO_ROOT / "bin" / "muse-tui.py"))
|
||||
cls.muse_tui = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(cls.muse_tui)
|
||||
|
||||
def setUp(self):
|
||||
self.mock_stdscr = MagicMock()
|
||||
self.mock_stdscr.getmaxyx.return_value = (40, 120)
|
||||
self.app = self.muse_tui.MuseTUI(self.mock_stdscr, initial_mode="muse", initial_node="muse")
|
||||
self.app.data = MagicMock()
|
||||
self.app.data.nodes = ["muse", "pip", "646", "opm", "def", "dev"]
|
||||
self.app.data.active_node = "muse"
|
||||
self.app.data.active_thread_id = "thread-1"
|
||||
self.app.data.active_thread_title = "Main Chat"
|
||||
self.app.data.lock = MagicMock()
|
||||
self.app.data.lock.__enter__.return_value = None
|
||||
self.app.data.lock.__exit__.return_value = None
|
||||
|
||||
def test_slash_blocked_opens_approvals_modal(self):
|
||||
self.app.data.approvals_cache = [
|
||||
{"node": "opm", "status": "PENDING", "title": "Review task", "is_trusted": True},
|
||||
{"node": "646", "status": "PENDING", "title": "SSH connect", "is_trusted": False},
|
||||
]
|
||||
self.app._execute_input_line("/blocked")
|
||||
self.assertEqual(self.app.modal, "approvals")
|
||||
self.assertEqual(self.app.approvals_sel_idx, 0)
|
||||
|
||||
self.app._execute_input_line("/Blocked")
|
||||
self.assertEqual(self.app.modal, "approvals")
|
||||
|
||||
self.app._execute_input_line("/approvals")
|
||||
self.assertEqual(self.app.modal, "approvals")
|
||||
|
||||
def test_approvals_modal_navigation_and_targeting(self):
|
||||
self.app.modal = "approvals"
|
||||
self.app.approvals_sel_idx = 0
|
||||
self.app.data.approvals_cache = [
|
||||
{"node": "muse", "status": "PENDING", "title": "Task 1"},
|
||||
{"node": "opm", "status": "PENDING", "title": "Task 2"},
|
||||
]
|
||||
|
||||
# Press 'j' (down)
|
||||
handled = self.app._handle_modal_key(ord('j'))
|
||||
self.assertTrue(handled)
|
||||
self.assertEqual(self.app.approvals_sel_idx, 1)
|
||||
|
||||
# Press '1' (allow) -> should target 'opm' (selected row), not 'muse' (active_node)
|
||||
with patch.object(self.app, "_async_resolve_approval") as mock_resolve:
|
||||
handled = self.app._handle_modal_key(ord('1'))
|
||||
self.assertTrue(handled)
|
||||
self.assertIsNone(self.app.modal)
|
||||
mock_resolve.assert_called_once_with("allow", "opm")
|
||||
|
||||
def test_approvals_modal_navigation_k(self):
|
||||
self.app.modal = "approvals"
|
||||
self.app.approvals_sel_idx = 1
|
||||
self.app.data.approvals_cache = [
|
||||
{"node": "muse", "status": "PENDING", "title": "Task 1"},
|
||||
{"node": "opm", "status": "PENDING", "title": "Task 2"},
|
||||
]
|
||||
|
||||
# Press 'k' (up)
|
||||
handled = self.app._handle_modal_key(ord('k'))
|
||||
self.assertTrue(handled)
|
||||
self.assertEqual(self.app.approvals_sel_idx, 0)
|
||||
|
||||
# Press '2' (always) -> should target 'muse'
|
||||
with patch.object(self.app, "_async_resolve_approval") as mock_resolve:
|
||||
handled = self.app._handle_modal_key(ord('2'))
|
||||
self.assertTrue(handled)
|
||||
mock_resolve.assert_called_once_with("always", "muse")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,179 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
test_clean_transcript.py — Unit and integration tests for clean line-by-line transcript view:
|
||||
1. Default display style is 'clean' (no box borders, no bottom rules, no vertical pipe bars).
|
||||
2. Clean mode renders bullet header indicators (● AGENT, ▸ YOU) and 2-space indented body lines.
|
||||
3. Boxed mode retains classic ASCII/Unicode frames (┌──, │ , └───).
|
||||
4. Runtime style toggling via 'b' / 'B' hotkey in Normal mode.
|
||||
5. Slash commands: /clean, /boxed, /view, /compact.
|
||||
6. Selection and copy extraction (_copy_rendered_line_range) handles clean formatting seamlessly.
|
||||
7. Settings persistence in ~/.config/muse-cli/tui_settings.json.
|
||||
"""
|
||||
import curses
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
MUSE_TUI_PATH = REPO_ROOT / "bin" / "muse-tui.py"
|
||||
|
||||
spec = importlib.util.spec_from_file_location("muse_tui", MUSE_TUI_PATH)
|
||||
muse_tui = importlib.util.module_from_spec(spec)
|
||||
sys.modules["muse_tui"] = muse_tui
|
||||
spec.loader.exec_module(muse_tui)
|
||||
|
||||
MuseTUI = muse_tui.MuseTUI
|
||||
|
||||
|
||||
class TestCleanTranscript(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.mock_stdscr = MagicMock()
|
||||
self.mock_stdscr.getmaxyx.return_value = (30, 100)
|
||||
self.tui = MuseTUI(self.mock_stdscr, initial_mode="muse", initial_node="muse")
|
||||
self.tui.safe_addstr = MagicMock()
|
||||
|
||||
self.messages = [
|
||||
{
|
||||
"role": "assistant",
|
||||
"text": "Same sweep thread, routine — holding on dev-i04.\nSecond line of response.",
|
||||
"message_id": "ast-msg-86648",
|
||||
"seq": 86648,
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"text": "Roger that, continue monitoring.\n[WO:20261006-1] Active",
|
||||
"message_id": "usr-msg-d183a734",
|
||||
"seq": 86649,
|
||||
},
|
||||
]
|
||||
with self.tui.data.lock:
|
||||
self.tui.data.history_cache[(self.tui.data.active_node, self.tui.data.active_thread_id)] = self.messages
|
||||
|
||||
def test_default_style_is_clean(self):
|
||||
"""Verify that newly initialized TUI defaults to 'clean' transcript style."""
|
||||
self.assertEqual(self.tui.transcript_style, "clean")
|
||||
|
||||
def test_render_transcript_clean_mode(self):
|
||||
"""In 'clean' mode, lines do NOT contain ┌──, │ , or └───, and use clean bullets."""
|
||||
self.tui.transcript_style = "clean"
|
||||
self.tui._transcript_cache_key = None
|
||||
self.tui._render_transcript(1, 0, 25, 90)
|
||||
|
||||
lines = [item[0] for item in self.tui._transcript_cache_lines]
|
||||
combined = "\n".join(lines)
|
||||
|
||||
# Must not contain box border frames
|
||||
self.assertNotIn("┌──", combined)
|
||||
self.assertNotIn("└──", combined)
|
||||
self.assertNotIn("│ ", combined)
|
||||
|
||||
# Must contain clean indicators
|
||||
self.assertTrue(any(l.startswith("● MUSE") for l in lines), "Agent header must start with clean bullet ●")
|
||||
self.assertTrue(any(l.startswith("▸ YOU") for l in lines), "User header must start with clean prompt ▸")
|
||||
|
||||
# Body lines must use 2-space indentation
|
||||
self.assertTrue(any(l.startswith(" Same sweep thread") for l in lines))
|
||||
self.assertTrue(any(l.startswith(" Second line") for l in lines))
|
||||
self.assertTrue(any(l.startswith(" Roger that") for l in lines))
|
||||
|
||||
def test_render_transcript_boxed_mode(self):
|
||||
"""In 'boxed' mode, classic ASCII/Unicode frames (┌──, │ , └───) are rendered."""
|
||||
self.tui.transcript_style = "boxed"
|
||||
self.tui._transcript_cache_key = None
|
||||
self.tui._render_transcript(1, 0, 25, 90)
|
||||
|
||||
lines = [item[0] for item in self.tui._transcript_cache_lines]
|
||||
combined = "\n".join(lines)
|
||||
|
||||
self.assertIn("┌── [MUSE / MUSE AGENT]", combined)
|
||||
self.assertIn("┌── [YOU / OPERATOR]", combined)
|
||||
self.assertIn("│ Same sweep thread", combined)
|
||||
self.assertIn("└──", combined)
|
||||
|
||||
def test_toggle_transcript_style(self):
|
||||
"""toggle_transcript_style switches between clean and boxed, invalidating cache."""
|
||||
self.tui.transcript_style = "clean"
|
||||
self.tui._transcript_cache_key = ("some_key",)
|
||||
|
||||
self.tui.toggle_transcript_style()
|
||||
self.assertEqual(self.tui.transcript_style, "boxed")
|
||||
self.assertIsNone(self.tui._transcript_cache_key)
|
||||
self.assertIn("Classic boxed", self.tui.toast_msg)
|
||||
|
||||
self.tui.toggle_transcript_style()
|
||||
self.assertEqual(self.tui.transcript_style, "clean")
|
||||
self.assertIsNone(self.tui._transcript_cache_key)
|
||||
self.assertIn("Clean line-by-line", self.tui.toast_msg)
|
||||
|
||||
def test_normal_mode_hotkey_b_toggles_style(self):
|
||||
"""Pressing 'b' or 'B' in normal mode toggles transcript style."""
|
||||
self.tui.editor_mode = "NORMAL"
|
||||
self.tui.copy_mode = False
|
||||
self.tui.transcript_style = "clean"
|
||||
|
||||
handled = self.tui._handle_key(ord('b'))
|
||||
self.assertTrue(handled)
|
||||
self.assertEqual(self.tui.transcript_style, "boxed")
|
||||
|
||||
handled = self.tui._handle_key(ord('B'))
|
||||
self.assertTrue(handled)
|
||||
self.assertEqual(self.tui.transcript_style, "clean")
|
||||
|
||||
def test_slash_commands_clean_and_boxed(self):
|
||||
"""Slash commands /clean, /boxed, and /view correctly set style."""
|
||||
self.tui.transcript_style = "clean"
|
||||
|
||||
# /boxed
|
||||
self.tui._execute_input_line("/boxed")
|
||||
self.assertEqual(self.tui.transcript_style, "boxed")
|
||||
|
||||
# /clean
|
||||
self.tui._execute_input_line("/clean")
|
||||
self.assertEqual(self.tui.transcript_style, "clean")
|
||||
|
||||
# /view box
|
||||
self.tui._execute_input_line("/view box")
|
||||
self.assertEqual(self.tui.transcript_style, "boxed")
|
||||
|
||||
# /view clean
|
||||
self.tui._execute_input_line("/view clean")
|
||||
self.assertEqual(self.tui.transcript_style, "clean")
|
||||
|
||||
# /style (toggles)
|
||||
self.tui._execute_input_line("/style")
|
||||
self.assertEqual(self.tui.transcript_style, "boxed")
|
||||
|
||||
def test_copy_range_in_clean_mode(self):
|
||||
"""_copy_rendered_line_range strips clean 2-space indents and action buttons."""
|
||||
self.tui.transcript_style = "clean"
|
||||
self.tui._transcript_cache_key = None
|
||||
self.tui._render_transcript(1, 0, 25, 90)
|
||||
|
||||
# Total lines in cache
|
||||
lines = self.tui._transcript_cache_lines
|
||||
copied = self.tui._copy_rendered_line_range(0, len(lines) - 1)
|
||||
|
||||
self.assertNotIn("│", copied)
|
||||
self.assertNotIn("[📋 Copy]", copied)
|
||||
self.assertNotIn("[↩ Reply]", copied)
|
||||
self.assertIn("Same sweep thread", copied)
|
||||
self.assertIn("Roger that, continue monitoring.", copied)
|
||||
|
||||
def test_settings_persistence(self):
|
||||
"""Style preference is saved to and loaded from JSON configuration."""
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
fake_cfg = Path(tmp_dir) / "tui_settings.json"
|
||||
with patch("pathlib.Path.home", return_value=Path(tmp_dir)):
|
||||
with patch.object(Path, "mkdir"):
|
||||
with patch("builtins.open", unittest.mock.mock_open()):
|
||||
self.tui.set_transcript_style("boxed")
|
||||
self.assertEqual(self.tui.transcript_style, "boxed")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,184 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
test_context_menus.py — Unit and integration tests for Right-Click function(s)
|
||||
and Context Action Menus in NetVM/Muse TUI:
|
||||
1. Right-click on Chat List (sidebar sidechats/threads) opens `chat_context` menu.
|
||||
2. Right-click on Fleet Agents panel opens `agent_context` menu.
|
||||
3. Right-click on Transcript message opens `message_context` menu.
|
||||
4. Chat Context Menu actions (Open, Pin/Unpin, Mark as Read, Copy ID, Copy URL, Reply, Close).
|
||||
5. Fleet Agent Context Menu actions (Open, Sync, Approvals, Mark All Read, Tmux, Work Order).
|
||||
6. Message Context Menu actions (Reply, Copy Text, Pin to Prompts, Copy ID).
|
||||
7. Keyboard navigation (j/k, 1-8, Enter) and mouse click selection inside context menus.
|
||||
8. Normal mode hotkeys ('x' / 'c') and slash commands (/context chat, /context agent).
|
||||
"""
|
||||
import curses
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
MUSE_TUI_PATH = REPO_ROOT / "bin" / "muse-tui.py"
|
||||
|
||||
spec = importlib.util.spec_from_file_location("muse_tui", MUSE_TUI_PATH)
|
||||
muse_tui = importlib.util.module_from_spec(spec)
|
||||
sys.modules["muse_tui"] = muse_tui
|
||||
spec.loader.exec_module(muse_tui)
|
||||
|
||||
MuseTUI = muse_tui.MuseTUI
|
||||
|
||||
|
||||
class TestContextMenus(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.mock_stdscr = MagicMock()
|
||||
self.mock_stdscr.getmaxyx.return_value = (30, 100)
|
||||
self.tui = MuseTUI(self.mock_stdscr, initial_mode="muse", initial_node="muse")
|
||||
self.tui.safe_addstr = MagicMock()
|
||||
|
||||
# Seed mock threads
|
||||
self.threads = [
|
||||
{"session_id": "sess-main-1234", "title": "Main Chat", "is_main": True, "pinned": False},
|
||||
{"session_id": "sess-side-5678", "title": "Audit Sidechat", "is_main": False, "pinned": True},
|
||||
{"session_id": "sess-side-9999", "title": "Dev Sidechat", "is_main": False, "pinned": False},
|
||||
]
|
||||
with self.tui.data.lock:
|
||||
self.tui.data.pinned_threads["muse"] = set()
|
||||
self.tui.data.threads_cache["muse"] = list(self.threads)
|
||||
self.tui.data.active_thread_id = "sess-main-1234"
|
||||
self.tui.data.active_thread_title = "Main Chat"
|
||||
self.tui.data.history_cache[("muse", "sess-main-1234")] = [
|
||||
{"role": "user", "text": "Hello muse", "seq": 1, "message_id": "mid-user-1"},
|
||||
{"role": "assistant", "text": "System ready", "seq": 2, "message_id": "mid-agent-2"},
|
||||
]
|
||||
self.tui.data.history_cache[("muse", "sess-side-5678")] = [
|
||||
{"role": "user", "text": "Review log", "seq": 3, "message_id": "mid-user-3"},
|
||||
]
|
||||
|
||||
def test_right_click_on_chat_list_opens_chat_context(self):
|
||||
"""Right-clicking a thread row in sidebar opens chat_context modal."""
|
||||
# Screen layout: content_y = 1, visible_agents = 6, divider_y = 8, threads_hdr_y = 9, list_y = 10
|
||||
# Right click on thread index 1 (sess-side-5678) at my = 11, mx = 10
|
||||
bstate = curses.BUTTON3_CLICKED
|
||||
self.tui._handle_mouse(mx=10, my=11, bstate=bstate)
|
||||
|
||||
self.assertEqual(self.tui.modal, "chat_context")
|
||||
self.assertIsNotNone(self.tui.context_chat)
|
||||
self.assertEqual(self.tui.context_chat["thread"]["session_id"], "sess-side-5678")
|
||||
self.assertEqual(self.tui.context_chat["thread"]["title"], "Audit Sidechat")
|
||||
|
||||
def test_right_click_on_fleet_agent_opens_agent_context(self):
|
||||
"""Right-clicking an agent row in sidebar opens agent_context modal."""
|
||||
# Row content_y + 1 = 2 is first agent (index 0 = 'muse')
|
||||
# Row content_y + 2 = 3 is second agent
|
||||
bstate = getattr(curses, "BUTTON3_PRESSED", 0x800)
|
||||
with self.tui.data.lock:
|
||||
second_agent = self.tui.data.nodes[1] if len(self.tui.data.nodes) > 1 else "muse"
|
||||
|
||||
self.tui._handle_mouse(mx=10, my=3, bstate=bstate)
|
||||
|
||||
self.assertEqual(self.tui.modal, "agent_context")
|
||||
self.assertIsNotNone(self.tui.context_agent)
|
||||
self.assertEqual(self.tui.context_agent["node"], second_agent)
|
||||
|
||||
def test_right_click_on_transcript_opens_message_context(self):
|
||||
"""Right-clicking a message in transcript opens message_context modal."""
|
||||
# Populate rendered lines in transcript
|
||||
self.tui._render_transcript(1, 26, 25, 74)
|
||||
# Click on transcript area mx = 50, my = 5
|
||||
bstate = curses.BUTTON3_CLICKED
|
||||
self.tui._handle_mouse(mx=50, my=5, bstate=bstate)
|
||||
|
||||
self.assertEqual(self.tui.modal, "message_context")
|
||||
self.assertIsNotNone(self.tui.context_message)
|
||||
self.assertIn("text", self.tui.context_message)
|
||||
|
||||
def test_chat_context_pin_action(self):
|
||||
"""Executing 'pin' action in chat_context toggles pin state."""
|
||||
t = {"session_id": "sess-side-9999", "title": "Dev Sidechat", "pinned": False}
|
||||
self.tui.context_chat = {"node": "muse", "thread": t, "index": 2}
|
||||
self.tui.modal = "chat_context"
|
||||
|
||||
# Press 'p'
|
||||
self.tui._handle_modal_key(ord('p'))
|
||||
self.assertIsNone(self.tui.modal)
|
||||
self.assertIn("📌 Pinned chat", self.tui.toast_msg)
|
||||
self.assertIn("sess-side-9999", self.tui.data.pinned_threads.get("muse", set()))
|
||||
|
||||
def test_chat_context_mark_read_action(self):
|
||||
"""Executing 'read' action marks thread as read."""
|
||||
t = {"session_id": "sess-side-5678", "title": "Audit Sidechat"}
|
||||
self.tui.context_chat = {"node": "muse", "thread": t, "index": 1}
|
||||
self.tui.modal = "chat_context"
|
||||
|
||||
self.tui._handle_modal_key(ord('m'))
|
||||
self.assertIsNone(self.tui.modal)
|
||||
self.assertIn("Marked 'Audit Sidechat' as read", self.tui.toast_msg)
|
||||
self.assertEqual(self.tui.data.get_unread_count("muse", "sess-side-5678"), 0)
|
||||
|
||||
def test_chat_context_copy_id_action(self):
|
||||
"""Executing 'copy_id' action copies session ID to clipboard."""
|
||||
t = {"session_id": "sess-side-5678", "title": "Audit Sidechat"}
|
||||
self.tui.context_chat = {"node": "muse", "thread": t, "index": 1}
|
||||
self.tui.modal = "chat_context"
|
||||
|
||||
mod = sys.modules.get(MuseTUI.__module__, muse_tui)
|
||||
with patch.object(mod, "copy_to_clipboard") as mock_cp1, \
|
||||
patch.object(muse_tui, "copy_to_clipboard") as mock_cp2:
|
||||
self.tui._handle_modal_key(ord('y'))
|
||||
self.assertTrue(mock_cp1.called or mock_cp2.called)
|
||||
self.assertIn("Copied session ID", self.tui.toast_msg)
|
||||
|
||||
def test_chat_context_reply_action(self):
|
||||
"""Executing 'reply' action switches to chat and quotes latest message."""
|
||||
t = {"session_id": "sess-main-1234", "title": "Main Chat"}
|
||||
self.tui.context_chat = {"node": "muse", "thread": t, "index": 0}
|
||||
self.tui.modal = "chat_context"
|
||||
|
||||
self.tui._handle_modal_key(ord('r'))
|
||||
self.assertIsNone(self.tui.modal)
|
||||
self.assertIsNotNone(self.tui.reply_target)
|
||||
self.assertEqual(self.tui.reply_target["role"], "assistant")
|
||||
self.assertIn("System ready", self.tui.reply_target["text"])
|
||||
|
||||
def test_agent_context_actions(self):
|
||||
"""Agent context menu hotkeys execute appropriate agent actions."""
|
||||
self.tui.context_agent = {"node": "muse"}
|
||||
self.tui.modal = "agent_context"
|
||||
|
||||
# Mark all chats on agent as read ('m')
|
||||
self.tui._handle_modal_key(ord('m'))
|
||||
self.assertIsNone(self.tui.modal)
|
||||
self.assertIn("Marked all chats on MUSE as read", self.tui.toast_msg)
|
||||
|
||||
# Open chat with agent ('1')
|
||||
self.tui.context_agent = {"node": "muse"}
|
||||
self.tui.modal = "agent_context"
|
||||
self.tui._handle_modal_key(ord('1'))
|
||||
self.assertIsNone(self.tui.modal)
|
||||
self.assertIn("Switched to agent", self.tui.toast_msg)
|
||||
|
||||
def test_normal_mode_x_opens_chat_context(self):
|
||||
"""Pressing 'x' in normal mode when focused on sidebar opens chat_context."""
|
||||
self.tui.editor_mode = "NORMAL"
|
||||
self.tui.focus_pane = "sidebar"
|
||||
self.tui.thread_sel_idx = 1
|
||||
|
||||
handled = self.tui._handle_key(ord('x'))
|
||||
self.assertTrue(handled)
|
||||
self.assertEqual(self.tui.modal, "chat_context")
|
||||
self.assertEqual(self.tui.context_chat["thread"]["session_id"], "sess-side-5678")
|
||||
|
||||
def test_slash_command_context(self):
|
||||
"""Slash command /context opens chat_context or agent_context."""
|
||||
self.tui._execute_input_line("/context chat")
|
||||
self.assertEqual(self.tui.modal, "chat_context")
|
||||
|
||||
self.tui.modal = None
|
||||
self.tui._execute_input_line("/context agent")
|
||||
self.assertEqual(self.tui.modal, "agent_context")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,374 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
test_copy_actions.py — Unit and integration tests for:
|
||||
1. Multi-environment clipboard engine (desktop utilities, OSC 52, internal buffer).
|
||||
2. Transcript formatting engine (Markdown structured headers, roles, seq, id).
|
||||
3. UI-first button approach:
|
||||
- Clickable [📋 Copy All] in thread header.
|
||||
- Clickable [📋 Copy] button on each message header.
|
||||
4. Normal mode hotkeys:
|
||||
- 'y': copy latest/active message to clipboard.
|
||||
- 'Y': copy entire chat transcript to clipboard.
|
||||
5. Slash commands:
|
||||
- '/copy', '/yank', '/cp': copy full transcript.
|
||||
- '/copy last', '/copy message': copy latest message.
|
||||
"""
|
||||
import curses
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
MUSE_TUI_PATH = REPO_ROOT / "bin" / "muse-tui.py"
|
||||
|
||||
spec = importlib.util.spec_from_file_location("muse_tui", MUSE_TUI_PATH)
|
||||
muse_tui = importlib.util.module_from_spec(spec)
|
||||
sys.modules["muse_tui"] = muse_tui
|
||||
spec.loader.exec_module(muse_tui)
|
||||
|
||||
MuseTUI = muse_tui.MuseTUI
|
||||
copy_to_clipboard = muse_tui.copy_to_clipboard
|
||||
strip_message_metadata = muse_tui.strip_message_metadata
|
||||
|
||||
|
||||
class TestCopyActions(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.mock_stdscr = MagicMock()
|
||||
self.mock_stdscr.getmaxyx.return_value = (30, 100)
|
||||
self.tui = MuseTUI(self.mock_stdscr, initial_mode="muse", initial_node="muse")
|
||||
self.tui.safe_addstr = MagicMock()
|
||||
|
||||
# Seed conversation history for testing
|
||||
self.messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"text": "Hello muse, check system status",
|
||||
"message_id": "usr-msg-12345678",
|
||||
"seq": 1,
|
||||
},
|
||||
{
|
||||
"role": "assistant",
|
||||
"text": "All services nominal.\nCDP port 9222 active.",
|
||||
"message_id": "ast-msg-87654321",
|
||||
"seq": 2,
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"text": "Please summarize logs",
|
||||
"message_id": "usr-msg-99999999",
|
||||
"seq": 3,
|
||||
},
|
||||
]
|
||||
self.tui.data.history_cache[("muse", "thread-abc")] = list(self.messages)
|
||||
self.tui.data.active_node = "muse"
|
||||
self.tui.data.active_thread_id = "thread-abc"
|
||||
self.tui.data.active_thread_title = "Diagnostics"
|
||||
|
||||
def test_strip_message_metadata(self):
|
||||
"""Verify strip_message_metadata strips reply headers and unquotes quoted lines."""
|
||||
raw = (
|
||||
"> Replying to ASSISTANT (seq:12620):\n"
|
||||
"> Relevant to the timer question: the loop scout just surfaced 6 unanswered jobs.\n"
|
||||
"> So part of getting work done is already queued.\n\n"
|
||||
"yes, proceed with spawning 5 sub agents to assist you ; continue at all costs"
|
||||
)
|
||||
expected = (
|
||||
"Relevant to the timer question: the loop scout just surfaced 6 unanswered jobs.\n"
|
||||
"So part of getting work done is already queued.\n\n"
|
||||
"yes, proceed with spawning 5 sub agents to assist you ; continue at all costs"
|
||||
)
|
||||
self.assertEqual(strip_message_metadata(raw), expected)
|
||||
|
||||
# Message without reply tags remains unchanged
|
||||
no_tags = "Standard message without reply headers."
|
||||
self.assertEqual(strip_message_metadata(no_tags), no_tags)
|
||||
|
||||
def test_copy_to_clipboard_osc52(self):
|
||||
"""Verify copy_to_clipboard formats and flushes OSC 52 sequence."""
|
||||
with patch("sys.stdout.write") as mock_write, patch("sys.stdout.flush"):
|
||||
ok = copy_to_clipboard("test-copy-payload")
|
||||
self.assertTrue(ok)
|
||||
# Verify OSC 52 sequence was written
|
||||
args = [call.args[0] for call in mock_write.call_args_list]
|
||||
osc_written = any("\033]52;c;" in a for a in args)
|
||||
self.assertTrue(osc_written)
|
||||
|
||||
def test_get_formatted_transcript_content(self):
|
||||
"""Verify structured Markdown output contains thread title, roles, seq, id, and text."""
|
||||
formatted = self.tui.get_formatted_transcript()
|
||||
self.assertIn("# Chat Transcript: Diagnostics", formatted)
|
||||
self.assertIn("Node: MUSE | Thread ID: thread-abc | Total Messages: 3", formatted)
|
||||
self.assertIn("### [YOU / OPERATOR] id:usr-msg-", formatted)
|
||||
self.assertIn("Hello muse, check system status", formatted)
|
||||
self.assertIn("### [MUSE / MUSE AGENT] seq:2 id:ast-msg-", formatted)
|
||||
self.assertIn("All services nominal.", formatted)
|
||||
self.assertIn("Please summarize logs", formatted)
|
||||
|
||||
def test_copy_transcript_to_clipboard(self):
|
||||
"""Verify copy_transcript_to_clipboard updates internal buffer and sets success toast."""
|
||||
with patch.object(muse_tui, "copy_to_clipboard", return_value=True):
|
||||
ok = self.tui.copy_transcript_to_clipboard()
|
||||
self.assertTrue(ok)
|
||||
self.assertIn("# Chat Transcript: Diagnostics", self.tui.clipboard_buf)
|
||||
self.assertIn("Copied full transcript (3 msgs", self.tui.toast_msg)
|
||||
|
||||
def test_copy_message_to_clipboard_latest(self):
|
||||
"""Verify copy_message_to_clipboard defaults to latest message."""
|
||||
with patch.object(muse_tui, "copy_to_clipboard", return_value=True):
|
||||
ok = self.tui.copy_message_to_clipboard()
|
||||
self.assertTrue(ok)
|
||||
self.assertEqual(self.tui.clipboard_buf, "Please summarize logs")
|
||||
self.assertIn("Copied clean message to clipboard", self.tui.toast_msg)
|
||||
|
||||
def test_copy_message_to_clipboard_with_and_without_context(self):
|
||||
"""Verify with_context toggles between metadata stripping and raw text."""
|
||||
raw = "> Replying to ASSISTANT (seq:1):\n> Quoted header\n\nActual response body"
|
||||
with patch.object(muse_tui, "copy_to_clipboard", return_value=True):
|
||||
# Clean text (without context)
|
||||
self.tui.copy_message_to_clipboard(raw, with_context=False)
|
||||
self.assertEqual(self.tui.clipboard_buf, "Quoted header\n\nActual response body")
|
||||
self.assertIn("clean message", self.tui.toast_msg)
|
||||
|
||||
# Raw text (with context)
|
||||
self.tui.copy_message_to_clipboard(raw, with_context=True)
|
||||
self.assertEqual(self.tui.clipboard_buf, raw)
|
||||
self.assertIn("message with context", self.tui.toast_msg)
|
||||
|
||||
def test_copy_message_to_clipboard_explicit(self):
|
||||
"""Verify copy_message_to_clipboard with explicit message text."""
|
||||
with patch.object(muse_tui, "copy_to_clipboard", return_value=True):
|
||||
custom_text = "Specific arbitrary message payload"
|
||||
ok = self.tui.copy_message_to_clipboard(custom_text)
|
||||
self.assertTrue(ok)
|
||||
self.assertEqual(self.tui.clipboard_buf, custom_text)
|
||||
|
||||
def test_render_transcript_buttons(self):
|
||||
"""Verify [📋 Copy All], [📋 Copy] and [📑+ Context] buttons are rendered and registered."""
|
||||
self.tui._render_transcript(y=1, x=24, h=25, w=76)
|
||||
|
||||
# Check [📋 Copy All] bounds registered
|
||||
self.assertIsNotNone(self.tui._btn_copy_all_bounds)
|
||||
btn_y, x_start, x_end = self.tui._btn_copy_all_bounds
|
||||
self.assertEqual(btn_y, 1)
|
||||
self.assertGreater(x_end, x_start)
|
||||
|
||||
# Verify button string was rendered to screen at btn coordinates
|
||||
calls = [c for c in self.tui.safe_addstr.call_args_list if "[📋 Copy All]" in str(c)]
|
||||
self.assertTrue(len(calls) > 0, "[📋 Copy All] must be rendered to stdscr")
|
||||
win_arg, call_y, call_x, call_text, call_attr = calls[0].args
|
||||
self.assertEqual(call_y, 1)
|
||||
self.assertEqual(call_x, x_start)
|
||||
|
||||
# Check individual message copy buttons registered
|
||||
self.assertGreater(len(self.tui._msg_copy_buttons), 0)
|
||||
self.assertGreater(len(self.tui._msg_copy_context_buttons), 0)
|
||||
for (my, mx1, mx2, raw_text) in self.tui._msg_copy_buttons:
|
||||
self.assertGreater(mx2, mx1)
|
||||
self.assertIn(raw_text, [m["text"] for m in self.messages])
|
||||
for (my, mx1, mx2, raw_text) in self.tui._msg_copy_context_buttons:
|
||||
self.assertGreater(mx2, mx1)
|
||||
self.assertIn(raw_text, [m["text"] for m in self.messages])
|
||||
|
||||
def test_render_transcript_narrow_screen_retains_copy_button(self):
|
||||
"""Verify [📋 Copy All] is retained next to message count even with long title & narrow width."""
|
||||
self.tui.data.active_thread_title = "Extremely Long Comprehensive Diagnostics Thread Title Exceeding Width"
|
||||
self.tui._render_transcript(y=1, x=10, h=20, w=48)
|
||||
|
||||
self.assertIsNotNone(self.tui._btn_copy_all_bounds)
|
||||
btn_y, x_start, x_end = self.tui._btn_copy_all_bounds
|
||||
self.assertEqual(btn_y, 1)
|
||||
self.assertLessEqual(x_end, 10 + 48, "Button must stay within transcript boundary")
|
||||
calls = [c for c in self.tui.safe_addstr.call_args_list if "[📋 Copy All]" in str(c)]
|
||||
self.assertTrue(len(calls) > 0)
|
||||
|
||||
def test_mouse_click_copy_all(self):
|
||||
"""Verify clicking [📋 Copy All] copies full transcript to clipboard."""
|
||||
self.tui._render_transcript(y=1, x=24, h=25, w=76)
|
||||
btn_y, x_start, x_end = self.tui._btn_copy_all_bounds
|
||||
click_x = (x_start + x_end) // 2
|
||||
|
||||
with patch.object(self.tui, "copy_transcript_to_clipboard") as mock_copy_all:
|
||||
handled = self.tui._handle_mouse(click_x, btn_y, curses.BUTTON1_CLICKED)
|
||||
self.assertTrue(handled)
|
||||
mock_copy_all.assert_called_once()
|
||||
|
||||
def test_mouse_click_message_copy(self):
|
||||
"""Verify clicking an individual [📋 Copy] button copies that specific message without context."""
|
||||
self.tui._render_transcript(y=1, x=24, h=25, w=76)
|
||||
self.assertGreater(len(self.tui._msg_copy_buttons), 0)
|
||||
first_btn = self.tui._msg_copy_buttons[0]
|
||||
msg_y, x_start, x_end, raw_text = first_btn
|
||||
click_x = (x_start + x_end) // 2
|
||||
|
||||
with patch.object(self.tui, "copy_message_to_clipboard") as mock_copy_msg:
|
||||
handled = self.tui._handle_mouse(click_x, msg_y, curses.BUTTON1_CLICKED)
|
||||
self.assertTrue(handled)
|
||||
mock_copy_msg.assert_called_once_with(raw_text, with_context=False)
|
||||
|
||||
def test_mouse_click_message_copy_context(self):
|
||||
"""Verify clicking an individual [📑+ Context] button copies that message with full context."""
|
||||
self.tui._render_transcript(y=1, x=24, h=25, w=76)
|
||||
self.assertGreater(len(self.tui._msg_copy_context_buttons), 0)
|
||||
first_btn = self.tui._msg_copy_context_buttons[0]
|
||||
msg_y, x_start, x_end, raw_text = first_btn
|
||||
click_x = (x_start + x_end) // 2
|
||||
|
||||
with patch.object(self.tui, "copy_message_to_clipboard") as mock_copy_msg:
|
||||
handled = self.tui._handle_mouse(click_x, msg_y, curses.BUTTON1_CLICKED)
|
||||
self.assertTrue(handled)
|
||||
mock_copy_msg.assert_called_once_with(raw_text, with_context=True)
|
||||
|
||||
def test_normal_hotkey_y(self):
|
||||
"""Verify 'y' in Normal mode copies latest message."""
|
||||
self.tui.editor_mode = "NORMAL"
|
||||
with patch.object(self.tui, "copy_message_to_clipboard") as mock_copy_msg:
|
||||
handled = self.tui._handle_normal_key(ord("y"))
|
||||
self.assertTrue(handled)
|
||||
mock_copy_msg.assert_called_once()
|
||||
|
||||
def test_normal_hotkey_Y(self):
|
||||
"""Verify 'Y' in Normal mode copies full transcript."""
|
||||
self.tui.editor_mode = "NORMAL"
|
||||
with patch.object(self.tui, "copy_transcript_to_clipboard") as mock_copy_all:
|
||||
handled = self.tui._handle_normal_key(ord("Y"))
|
||||
self.assertTrue(handled)
|
||||
mock_copy_all.assert_called_once()
|
||||
|
||||
def test_slash_command_copy_all(self):
|
||||
"""Verify '/copy' or '/yank' copies full transcript."""
|
||||
with patch.object(self.tui, "copy_transcript_to_clipboard") as mock_copy_all:
|
||||
self.tui._execute_input_line("/copy")
|
||||
mock_copy_all.assert_called_once()
|
||||
|
||||
with patch.object(self.tui, "copy_transcript_to_clipboard") as mock_copy_all:
|
||||
self.tui._execute_input_line("/yank")
|
||||
mock_copy_all.assert_called_once()
|
||||
|
||||
def test_slash_command_copy_last(self):
|
||||
"""Verify '/copy last' copies latest message."""
|
||||
with patch.object(self.tui, "copy_message_to_clipboard") as mock_copy_msg:
|
||||
self.tui._execute_input_line("/copy last")
|
||||
mock_copy_msg.assert_called_once()
|
||||
|
||||
def test_enter_visual_copy_mode(self):
|
||||
"""Pressing 'v' enters Visual Copy Mode with selection on visible line."""
|
||||
self.tui.editor_mode = "NORMAL"
|
||||
self.tui.copy_mode = False
|
||||
self.tui._render_transcript(y=1, x=24, h=25, w=76)
|
||||
|
||||
ret = self.tui._handle_normal_key(ord('v'))
|
||||
self.assertTrue(ret)
|
||||
self.assertTrue(self.tui.copy_mode)
|
||||
self.assertEqual(self.tui.focus_pane, "transcript")
|
||||
self.assertIsNotNone(self.tui.visual_sel_start)
|
||||
self.assertEqual(self.tui.visual_sel_start, self.tui.copy_cursor_line)
|
||||
self.assertIn("VISUAL line copy mode", self.tui.toast_msg)
|
||||
|
||||
def test_enter_visual_copy_mode_V(self):
|
||||
"""Pressing 'V' on transcript enters visual copy mode."""
|
||||
self.tui.editor_mode = "NORMAL"
|
||||
self.tui.focus_pane = "transcript"
|
||||
self.tui.copy_mode = False
|
||||
self.tui._render_transcript(y=1, x=24, h=25, w=76)
|
||||
|
||||
ret = self.tui._handle_normal_key(ord('V'))
|
||||
self.assertTrue(ret)
|
||||
self.assertTrue(self.tui.copy_mode)
|
||||
self.assertIsNotNone(self.tui.visual_sel_start)
|
||||
self.assertIn("VISUAL line copy mode", self.tui.toast_msg)
|
||||
|
||||
def test_message_context_menu_copy_clean_and_context(self):
|
||||
"""Verify message_context modal options 2/y (clean) and 3/Y (context) work."""
|
||||
target_msg = {
|
||||
"role": "assistant",
|
||||
"text": "> Replying to USER (seq:4):\n> Previous line\n\nClean response text",
|
||||
"message_id": "msg-xyz-123",
|
||||
"seq": 5,
|
||||
}
|
||||
self.tui.context_message = target_msg
|
||||
self.tui.modal = "message_context"
|
||||
|
||||
with patch.object(self.tui, "copy_message_to_clipboard") as mock_copy:
|
||||
# Key '2' or 'y' copies clean text (with_context=False)
|
||||
self.tui._handle_modal_key(ord('2'))
|
||||
mock_copy.assert_called_with(target_msg["text"], with_context=False)
|
||||
|
||||
self.tui.context_message = target_msg
|
||||
self.tui.modal = "message_context"
|
||||
with patch.object(self.tui, "copy_message_to_clipboard") as mock_copy:
|
||||
# Key '3' or 'Y' copies with context (with_context=True)
|
||||
self.tui._handle_modal_key(ord('3'))
|
||||
mock_copy.assert_called_with(target_msg["text"], with_context=True)
|
||||
|
||||
def test_copy_mode_navigation_and_yank_with_border_stripping(self):
|
||||
"""Navigating with 'j'/'k' and yanking with 'y' copies clean text without box borders."""
|
||||
self.tui._render_transcript(y=1, x=24, h=25, w=76)
|
||||
self.tui.copy_mode = True
|
||||
self.tui.visual_sel_start = 2
|
||||
self.tui.copy_cursor_line = 2
|
||||
|
||||
# Move down to line 4
|
||||
self.tui._handle_copy_mode_key(ord('j'))
|
||||
self.assertEqual(self.tui.copy_cursor_line, 3)
|
||||
self.tui._handle_copy_mode_key(ord('j'))
|
||||
self.assertEqual(self.tui.copy_cursor_line, 4)
|
||||
|
||||
# Yank selection with 'y'
|
||||
with patch.object(muse_tui, "copy_to_clipboard", return_value=True) as mock_clip:
|
||||
self.tui._handle_copy_mode_key(ord('y'))
|
||||
self.assertFalse(self.tui.copy_mode)
|
||||
self.assertIsNone(self.tui.visual_sel_start)
|
||||
mock_clip.assert_called_once()
|
||||
copied = mock_clip.call_args[0][0]
|
||||
# Ensure vertical box lines '│ ' are stripped
|
||||
self.assertNotIn("│ ", copied)
|
||||
self.assertIn("Copied", self.tui.toast_msg)
|
||||
|
||||
def test_mouse_drag_select_and_release_copy(self):
|
||||
"""Dragging mouse over transcript lines highlights and copies text on release."""
|
||||
self.tui._render_transcript(y=1, x=24, h=25, w=76)
|
||||
|
||||
# 1. Mouse down at screen y=5, x=30 (inside transcript)
|
||||
self.tui._handle_mouse(mx=30, my=5, bstate=curses.BUTTON1_PRESSED)
|
||||
self.assertTrue(self.tui.mouse_dragging)
|
||||
self.assertTrue(self.tui.copy_mode)
|
||||
start_line = self.tui.visual_sel_start
|
||||
self.assertIsNotNone(start_line)
|
||||
|
||||
# 2. Mouse drag to y=8
|
||||
self.tui._handle_mouse(mx=30, my=8, bstate=curses.REPORT_MOUSE_POSITION)
|
||||
self.assertEqual(self.tui.visual_sel_start, start_line)
|
||||
self.assertNotEqual(self.tui.copy_cursor_line, start_line)
|
||||
|
||||
# 3. Mouse release
|
||||
with patch.object(muse_tui, "copy_to_clipboard", return_value=True) as mock_clip:
|
||||
self.tui._handle_mouse(mx=30, my=8, bstate=curses.BUTTON1_RELEASED)
|
||||
self.assertFalse(self.tui.mouse_dragging)
|
||||
self.assertFalse(self.tui.copy_mode)
|
||||
mock_clip.assert_called_once()
|
||||
copied = mock_clip.call_args[0][0]
|
||||
self.assertGreater(len(copied), 0)
|
||||
self.assertIn("Copied selection", self.tui.toast_msg)
|
||||
|
||||
def test_copy_mode_escape_exits(self):
|
||||
"""Esc or 'q' clears selection and exits copy mode."""
|
||||
self.tui.copy_mode = True
|
||||
self.tui.visual_sel_start = 5
|
||||
self.tui.copy_cursor_line = 8
|
||||
|
||||
# First Esc clears selection
|
||||
self.tui._handle_copy_mode_key(27)
|
||||
self.assertIsNone(self.tui.visual_sel_start)
|
||||
self.assertTrue(self.tui.copy_mode)
|
||||
|
||||
# Second Esc exits copy mode
|
||||
self.tui._handle_copy_mode_key(27)
|
||||
self.assertFalse(self.tui.copy_mode)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -0,0 +1,207 @@
|
||||
import unittest
|
||||
import time
|
||||
from unittest.mock import MagicMock
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(REPO_ROOT / "bin"))
|
||||
|
||||
import muse_tui
|
||||
|
||||
class TestNotificationCounters(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.dm = muse_tui.FleetDataManager()
|
||||
# Mock background loops from interfering with testing
|
||||
self.dm.running = False
|
||||
self.dm.read_msg_counts = {}
|
||||
# Isolate threads and history caches for tests
|
||||
self.dm.nodes = ["pip", "646"]
|
||||
self.dm.threads_cache = {"pip": [], "646": []}
|
||||
self.dm.history_cache = {}
|
||||
|
||||
def test_unread_count_calculation(self):
|
||||
node = "pip"
|
||||
tid = "test-thread-123"
|
||||
key = (node, tid)
|
||||
|
||||
# Initially no messages -> unread is 0
|
||||
self.assertEqual(self.dm.get_unread_count(node, tid), 0)
|
||||
|
||||
# 5 messages arrive in cache
|
||||
self.dm.history_cache[key] = [{"text": f"msg {i}"} for i in range(5)]
|
||||
self.assertEqual(self.dm.get_unread_count(node, tid), 5)
|
||||
|
||||
# Read 3 messages
|
||||
self.dm.mark_thread_read(node, tid, count=3)
|
||||
self.assertEqual(self.dm.get_unread_count(node, tid), 2)
|
||||
|
||||
# Read all messages
|
||||
self.dm.mark_thread_read(node, tid)
|
||||
self.assertEqual(self.dm.get_unread_count(node, tid), 0)
|
||||
|
||||
def test_node_and_fleet_unread_total(self):
|
||||
# Configure test threads for pip and 646
|
||||
self.dm.threads_cache["pip"] = [
|
||||
{"session_id": "pip-chat-1", "is_main": True},
|
||||
{"session_id": "pip-chat-2", "is_main": False},
|
||||
]
|
||||
self.dm.threads_cache["646"] = [
|
||||
{"session_id": "646-chat-1", "is_main": True},
|
||||
]
|
||||
|
||||
# Populate message histories
|
||||
self.dm.history_cache[("pip", "pip-chat-1")] = [{"text": "1"}, {"text": "2"}]
|
||||
self.dm.history_cache[("pip", "pip-chat-2")] = [{"text": "a"}, {"text": "b"}, {"text": "c"}]
|
||||
self.dm.history_cache[("646", "646-chat-1")] = [{"text": "x"}]
|
||||
|
||||
# Before reading: pip has 5, 646 has 1, total fleet is 6
|
||||
self.assertEqual(self.dm.get_node_unread_total("pip"), 5)
|
||||
self.assertEqual(self.dm.get_node_unread_total("646"), 1)
|
||||
self.assertEqual(self.dm.get_fleet_unread_total(), 6)
|
||||
|
||||
# Read pip-chat-1
|
||||
self.dm.mark_thread_read("pip", "pip-chat-1")
|
||||
self.assertEqual(self.dm.get_node_unread_total("pip"), 3)
|
||||
self.assertEqual(self.dm.get_fleet_unread_total(), 4)
|
||||
|
||||
def test_dwell_clearing_logic(self):
|
||||
tui = muse_tui.MuseTUI(MagicMock())
|
||||
tui.data.running = False
|
||||
active_node = "muse"
|
||||
active_tid = "muse-thread-abc"
|
||||
|
||||
tui.data.active_node = active_node
|
||||
tui.data.active_thread_id = active_tid
|
||||
tui.data.history_cache[(active_node, active_tid)] = [{"text": "msg"}] * 4
|
||||
self.assertEqual(tui.data.get_unread_count(active_node, active_tid), 4)
|
||||
|
||||
# Simulated dwell < 5.0 seconds
|
||||
tui.cur_chat_enter_time = time.time() - 2.0
|
||||
tui.last_dwell_check_target = (active_node, active_tid)
|
||||
# Should not clear yet
|
||||
if time.time() - tui.cur_chat_enter_time >= 5.0:
|
||||
tui.data.mark_thread_read(active_node, active_tid)
|
||||
self.assertEqual(tui.data.get_unread_count(active_node, active_tid), 4)
|
||||
|
||||
# Simulated dwell >= 5.0 seconds
|
||||
tui.cur_chat_enter_time = time.time() - 5.5
|
||||
if time.time() - tui.cur_chat_enter_time >= 5.0:
|
||||
tui.data.mark_thread_read(active_node, active_tid)
|
||||
self.assertEqual(tui.data.get_unread_count(active_node, active_tid), 0)
|
||||
|
||||
def test_agent_sorting_and_initial_selection(self):
|
||||
# Configure nodes
|
||||
self.dm.nodes = ["pip", "muse", "opm", "646"]
|
||||
self.dm.threads_cache = {
|
||||
"pip": [{"session_id": "p1"}],
|
||||
"muse": [{"session_id": "m1"}],
|
||||
"opm": [{"session_id": "o1"}],
|
||||
"646": [{"session_id": "c1"}],
|
||||
}
|
||||
# Give opm 5 unread messages, muse 2 unreads, pip 0, 646 0
|
||||
self.dm.history_cache = {
|
||||
("opm", "o1"): [{"text": f"msg {i}"} for i in range(5)],
|
||||
("muse", "m1"): [{"text": f"msg {i}"} for i in range(2)],
|
||||
}
|
||||
self.dm.agent_interactions = {"pip": 100.0, "646": 200.0, "muse": 50.0, "opm": 0.0}
|
||||
|
||||
self.dm.sort_nodes()
|
||||
# opm should be first (highest unread = 5)
|
||||
# muse should be second (unread = 2)
|
||||
# 646 should be third (interaction = 200.0)
|
||||
# pip should be fourth (interaction = 100.0)
|
||||
self.assertEqual(self.dm.nodes, ["opm", "muse", "646", "pip"])
|
||||
|
||||
# Default initial_node=None in MuseTUI picks nodes[0] (top of list) and its Main Chat
|
||||
tui = muse_tui.MuseTUI(MagicMock(), initial_node=None)
|
||||
tui.data.running = False
|
||||
self.assertEqual(tui.data.active_node, tui.data.nodes[0])
|
||||
self.assertIsNotNone(tui.data.active_thread_id)
|
||||
top_node = tui.data.nodes[0]
|
||||
top_threads = {t.get("session_id") for t in tui.data.threads_cache.get(top_node, [])}
|
||||
self.assertIn(tui.data.active_thread_id, top_threads)
|
||||
|
||||
def test_reply_workflow(self):
|
||||
tui = muse_tui.MuseTUI(MagicMock())
|
||||
tui.data.running = False
|
||||
|
||||
msg = {
|
||||
"role": "assistant",
|
||||
"message_id": "msg-987654321",
|
||||
"seq": 42,
|
||||
"text": "Please confirm deployment parameters before we proceed.",
|
||||
}
|
||||
|
||||
# 1. Trigger reply
|
||||
ok = tui.trigger_reply_to_message(msg)
|
||||
self.assertTrue(ok)
|
||||
self.assertIsNotNone(tui.reply_target)
|
||||
self.assertEqual(tui.reply_target["seq"], 42)
|
||||
self.assertEqual(tui.reply_target["role"], "assistant")
|
||||
self.assertEqual(tui.editor_mode, "INSERT")
|
||||
|
||||
# 2. Cancel reply with Esc sequence
|
||||
tui._read_escape_sequence = MagicMock(return_value=[]) # standalone Esc
|
||||
tui._handle_escape_sequence()
|
||||
self.assertIsNone(tui.reply_target)
|
||||
self.assertEqual(tui.editor_mode, "NORMAL")
|
||||
|
||||
# 3. Trigger reply again and execute input line
|
||||
tui.trigger_reply_to_message(msg)
|
||||
dispatched_messages = []
|
||||
tui._async_send_message = lambda node, tid, text: dispatched_messages.append((node, tid, text))
|
||||
|
||||
tui._execute_input_line("Confirmed, deploy now.")
|
||||
self.assertIsNone(tui.reply_target)
|
||||
self.assertEqual(len(dispatched_messages), 1)
|
||||
node, tid, sent_text = dispatched_messages[0]
|
||||
self.assertIn("> Replying to ASSISTANT (seq:42):", sent_text)
|
||||
self.assertIn("Please confirm deployment parameters before we proceed.", sent_text)
|
||||
self.assertIn("Confirmed, deploy now.", sent_text)
|
||||
|
||||
def test_sidechat_notifications_and_transcript_divider(self):
|
||||
tui = muse_tui.MuseTUI(MagicMock())
|
||||
tui.data.running = False
|
||||
active_node = "muse"
|
||||
active_tid = "sidechat-worker-42"
|
||||
tui.data.active_node = active_node
|
||||
tui.data.active_thread_id = active_tid
|
||||
|
||||
tui.data.threads_cache[active_node] = [
|
||||
{"session_id": "main-chat", "is_main": True, "title": "Main Chat"},
|
||||
{"session_id": active_tid, "is_main": False, "title": "Worker Task 42"},
|
||||
{"session_id": "sidechat-worker-43", "is_main": False, "title": "Worker Task 43"},
|
||||
]
|
||||
|
||||
# 3 total messages, read count is 1 -> 2 unread messages in active sidechat
|
||||
tui.data.history_cache[(active_node, active_tid)] = [
|
||||
{"role": "user", "text": "initial prompt", "message_id": "m1"},
|
||||
{"role": "assistant", "text": "update 1", "message_id": "m2"},
|
||||
{"role": "assistant", "text": "update 2", "message_id": "m3"},
|
||||
]
|
||||
tui.data.read_msg_counts[(active_node, active_tid)] = 1
|
||||
|
||||
self.assertEqual(tui.data.get_unread_count(active_node, active_tid), 2)
|
||||
self.assertEqual(tui.data.get_node_unread_total(active_node), 2)
|
||||
|
||||
# Mock safe_addstr to capture rendered strings
|
||||
rendered_texts = []
|
||||
tui.safe_addstr = lambda win, y, x, text, attr=0: rendered_texts.append(text)
|
||||
|
||||
# 1. Render sidebar
|
||||
tui._render_sidebar(0, 0, 30, 32)
|
||||
# Should contain ●2 inline notification badge
|
||||
has_badge = any("●2" in s for s in rendered_texts)
|
||||
self.assertTrue(has_badge, "Sidechats list row or header must render inline ●2 unread badge")
|
||||
|
||||
# 2. Render transcript
|
||||
rendered_texts.clear()
|
||||
tui._render_transcript(0, 32, 30, 80)
|
||||
# Should render the unread divider banner or unread pill
|
||||
has_div = any("2 NEW UNREAD MESSAGE" in s for s in rendered_texts)
|
||||
has_hdr_pill = any("[● 2 new]" in s for s in rendered_texts)
|
||||
self.assertTrue(has_div or has_hdr_pill, "Transcript must display unread banner or pill")
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,166 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
test_paste_handling.py — Unit tests for long string / chunk paste handling in MuseTUI:
|
||||
1. Multi-line chunk pasting preserves newlines and does not send immediately.
|
||||
2. Enter key submits the entire chunk as a single message and single history item.
|
||||
3. Bracketed paste sequences (\033[200~ ... \033[201~) are correctly parsed and captured.
|
||||
4. Alt+Enter (ESC + Enter) inserts newlines for multiline composition.
|
||||
5. Unbracketed rapid paste bursts preserve newlines without line-by-line send.
|
||||
6. Arrow keys navigate lines within multiline buffer.
|
||||
7. UTF-8 multibyte characters are correctly decoded and inserted into buffer.
|
||||
8. Multiline messages or file paths starting with '/' are delivered as messages.
|
||||
"""
|
||||
|
||||
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 TestPasteHandling(unittest.TestCase):
|
||||
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()
|
||||
self.tui.data.running = False # Keep background thread stopped
|
||||
self.tui.data.active_node = "muse"
|
||||
self.tui.data.active_thread_id = "test-session"
|
||||
self.tui.data.active_thread_title = "Main Chat"
|
||||
self.tui.data.threads_cache["muse"] = [{"session_id": "test-session", "is_main": True}]
|
||||
|
||||
def test_insert_pasted_text_multiline(self):
|
||||
"""Pasting a multiline chunk enters input_buf in full without sending."""
|
||||
chunk = "Line 1: Hello\nLine 2: World\nLine 3: Agent instruction"
|
||||
self.tui.editor_mode = "NORMAL"
|
||||
self.tui.input_buf = ""
|
||||
self.tui.input_cursor = 0
|
||||
|
||||
self.tui._insert_pasted_text(chunk)
|
||||
|
||||
self.assertEqual(self.tui.editor_mode, "INSERT")
|
||||
self.assertEqual(self.tui.input_buf, chunk)
|
||||
self.assertEqual(self.tui.input_cursor, len(chunk))
|
||||
self.assertIn("3 lines", self.tui.toast_msg)
|
||||
# Verify nothing was sent yet
|
||||
self.assertEqual(len(self.tui.input_history), 0)
|
||||
|
||||
def test_multiline_send_as_single_chunk(self):
|
||||
"""Submitting a multiline buffer records ONE history entry and sends ONCE."""
|
||||
chunk = "Line 1: def test():\n print('NetVM')\nLine 3: return True"
|
||||
self.tui.editor_mode = "INSERT"
|
||||
self.tui.input_buf = chunk
|
||||
self.tui.input_cursor = len(chunk)
|
||||
|
||||
sent_messages = []
|
||||
self.tui._async_send_message = lambda node, tid, text: sent_messages.append((node, tid, text))
|
||||
|
||||
# Simulate user pressing Enter on the completed multiline chunk
|
||||
# With no characters waiting in stdin (peek_c = -1)
|
||||
self.mock_stdscr.getch.return_value = -1
|
||||
ret = self.tui._handle_insert_key(10)
|
||||
|
||||
self.assertTrue(ret)
|
||||
self.assertEqual(self.tui.input_buf, "")
|
||||
self.assertEqual(len(self.tui.input_history), 1)
|
||||
self.assertEqual(self.tui.input_history[0], chunk)
|
||||
self.assertEqual(len(sent_messages), 1)
|
||||
self.assertEqual(sent_messages[0][0], "muse")
|
||||
self.assertEqual(sent_messages[0][1], "test-session")
|
||||
self.assertEqual(sent_messages[0][2], chunk)
|
||||
|
||||
def test_bracketed_paste_capture(self):
|
||||
"""\033[200~ bracketed paste reads until \033[201~ and captures content."""
|
||||
payload = "Multi-line code snippet:\n```python\nx = 42\nprint(x)\n```"
|
||||
# Simulate stream of getch returns: payload bytes followed by \033[201~ then -1
|
||||
stream = list(payload.encode("utf-8")) + [27, ord('['), ord('2'), ord('0'), ord('1'), ord('~')]
|
||||
|
||||
call_idx = 0
|
||||
def mock_getch():
|
||||
nonlocal call_idx
|
||||
if call_idx < len(stream):
|
||||
val = stream[call_idx]
|
||||
call_idx += 1
|
||||
return val
|
||||
return -1
|
||||
|
||||
self.mock_stdscr.getch.side_effect = mock_getch
|
||||
self.tui.input_buf = ""
|
||||
self.tui.input_cursor = 0
|
||||
|
||||
handled = self.tui._handle_bracketed_paste()
|
||||
self.assertTrue(handled)
|
||||
self.assertEqual(self.tui.input_buf, payload)
|
||||
self.assertEqual(self.tui.input_cursor, len(payload))
|
||||
self.assertEqual(self.tui.editor_mode, "INSERT")
|
||||
|
||||
def test_alt_enter_inserts_newline(self):
|
||||
"""Alt+Enter (ESC + 10 or 13) inserts newline into input_buf."""
|
||||
self.tui.editor_mode = "INSERT"
|
||||
self.tui.input_buf = "First line"
|
||||
self.tui.input_cursor = len(self.tui.input_buf)
|
||||
|
||||
# Mock _read_escape_sequence returning [10] (Enter)
|
||||
with patch.object(self.tui, "_read_escape_sequence", return_value=[10]):
|
||||
res = self.tui._handle_escape_sequence()
|
||||
|
||||
self.assertTrue(res)
|
||||
self.assertEqual(self.tui.input_buf, "First line\n")
|
||||
self.assertEqual(self.tui.input_cursor, len("First line\n"))
|
||||
|
||||
def test_multiline_cursor_navigation(self):
|
||||
"""Up and Down arrow navigate across lines inside multiline input_buf."""
|
||||
self.tui.editor_mode = "INSERT"
|
||||
self.tui.input_buf = "Line A\nLine B\nLine C"
|
||||
# Place cursor at end of "Line C"
|
||||
self.tui.input_cursor = len(self.tui.input_buf)
|
||||
|
||||
# Press Up: cursor should move to "Line B"
|
||||
self.tui._handle_insert_key(curses.KEY_UP)
|
||||
curr_text = self.tui.input_buf[:self.tui.input_cursor]
|
||||
self.assertTrue(curr_text.endswith("Line B") or "Line B" in curr_text)
|
||||
|
||||
# Press Up again: cursor should move to "Line A"
|
||||
self.tui._handle_insert_key(curses.KEY_UP)
|
||||
curr_text = self.tui.input_buf[:self.tui.input_cursor]
|
||||
self.assertTrue(curr_text.startswith("Line A"))
|
||||
|
||||
# Press Down: cursor should move back down to "Line B"
|
||||
self.tui._handle_insert_key(curses.KEY_DOWN)
|
||||
curr_text = self.tui.input_buf[:self.tui.input_cursor]
|
||||
self.assertIn("Line B", curr_text)
|
||||
|
||||
def test_unicode_multibyte_input(self):
|
||||
"""Multi-byte UTF-8 bytes assemble into correct unicode characters."""
|
||||
self.tui.editor_mode = "INSERT"
|
||||
self.tui.input_buf = "Test: "
|
||||
self.tui.input_cursor = len(self.tui.input_buf)
|
||||
|
||||
# 'é' is 0xC3, 0xA9 (195, 169)
|
||||
self.tui._handle_insert_key(195)
|
||||
self.assertEqual(self.tui.input_buf, "Test: ") # incomplete byte, not added yet
|
||||
self.tui._handle_insert_key(169)
|
||||
self.assertEqual(self.tui.input_buf, "Test: é")
|
||||
self.assertEqual(self.tui.input_cursor, len("Test: é"))
|
||||
|
||||
def test_path_starting_with_slash_delivers_as_message(self):
|
||||
"""A string starting with a file path like '/var/log/syslog' is sent as message."""
|
||||
line = "/var/log/syslog contains error details"
|
||||
sent = []
|
||||
self.tui._async_send_message = lambda node, tid, text: sent.append(text)
|
||||
|
||||
self.tui._execute_input_line(line)
|
||||
self.assertEqual(len(sent), 1)
|
||||
self.assertEqual(sent[0], line)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,449 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
test_prompts.py — Comprehensive unit and integration test suite for:
|
||||
1. PromptManager persistence, defaults, formulation, deletion, and pin sorting.
|
||||
2. Compact persistent prompt shelf in View 1 (Agent Chat) and 1-click auto-fill.
|
||||
3. Expanded Prompt Library modal ('P') with Vim hjkl navigation, preview scroll, and Enter/click fill.
|
||||
4. Searchable chat history sends ('s' / Ctrl-R) aggregating session and history cache.
|
||||
5. 'p' pinning chat sent messages to the prompt fill / list in history search and transcript view.
|
||||
6. Horizontal scrolling of prompt shelf via chips bounds, navigation buttons, wheel, and '[' / ']'.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import time
|
||||
import json
|
||||
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 TestPromptManager(unittest.TestCase):
|
||||
"""Test suite covering PromptManager storage, formulation, and pin ranking."""
|
||||
|
||||
def setUp(self):
|
||||
self.temp_dir = tempfile.TemporaryDirectory()
|
||||
self.storage_file = Path(self.temp_dir.name) / "test_prompts.json"
|
||||
self.pm = muse_mod.PromptManager(storage_path=self.storage_file)
|
||||
|
||||
def tearDown(self):
|
||||
self.temp_dir.cleanup()
|
||||
|
||||
def test_default_prompts_seeded_and_persisted(self):
|
||||
"""Verify PromptManager initializes with default prompts and writes to disk."""
|
||||
self.assertTrue(self.storage_file.exists())
|
||||
prompts = self.pm.get_sorted_prompts()
|
||||
self.assertGreaterEqual(len(prompts), 5)
|
||||
titles = [p["title"] for p in prompts]
|
||||
self.assertIn("Fleet Audit", titles)
|
||||
self.assertIn("Work Order", titles)
|
||||
self.assertIn("CDP Inspect", titles)
|
||||
|
||||
def test_add_and_delete_prompt(self):
|
||||
"""Verify formulation of custom prompt skills and subsequent deletion."""
|
||||
new_p = self.pm.add_prompt(
|
||||
title="Custom Health Probe",
|
||||
prompt_text="Run detailed tcpdump and box health probes on all nodes.",
|
||||
tags=["probe", "custom"],
|
||||
pinned=True
|
||||
)
|
||||
self.assertEqual(new_p["title"], "Custom Health Probe")
|
||||
self.assertTrue(new_p["pinned"])
|
||||
|
||||
# Check persistence
|
||||
reloaded = muse_mod.PromptManager(storage_path=self.storage_file)
|
||||
prompts = reloaded.get_sorted_prompts()
|
||||
self.assertTrue(any(p["id"] == new_p["id"] for p in prompts))
|
||||
|
||||
# Delete
|
||||
self.assertTrue(self.pm.delete_prompt(new_p["id"]))
|
||||
prompts_after = self.pm.get_sorted_prompts()
|
||||
self.assertFalse(any(p["id"] == new_p["id"] for p in prompts_after))
|
||||
|
||||
def test_toggle_pin_prompt(self):
|
||||
"""Verify toggle_pin_prompt alters pinned flag and affects sorted order."""
|
||||
prompts = self.pm.get_sorted_prompts()
|
||||
target = prompts[-1] # Get unpinned prompt
|
||||
target_id = target["id"]
|
||||
|
||||
new_state = self.pm.toggle_pin_prompt(target_id)
|
||||
self.assertTrue(new_state)
|
||||
|
||||
# In sorted prompts, newly pinned should be among first
|
||||
sorted_p = self.pm.get_sorted_prompts()
|
||||
first_few_ids = [p["id"] for p in sorted_p[:4]]
|
||||
self.assertIn(target_id, first_few_ids)
|
||||
|
||||
def test_pin_chat_send(self):
|
||||
"""Verify pinning a chat sent message formats title, pins to top, and persists."""
|
||||
sent_text = "Investigate high CDP drop rates and verify routing table on node 646."
|
||||
prompt, is_new = self.pm.pin_chat_send(sent_text, node="646")
|
||||
self.assertTrue(is_new)
|
||||
self.assertTrue(prompt["pinned"])
|
||||
self.assertEqual(prompt["prompt"], sent_text)
|
||||
self.assertIn("Investigate high CDP", prompt["title"])
|
||||
|
||||
# Pinned chat send should be first in sorted prompts
|
||||
sorted_p = self.pm.get_sorted_prompts()
|
||||
self.assertEqual(sorted_p[0]["id"], prompt["id"])
|
||||
|
||||
# Pinning same text again should mark pinned without creating duplicate
|
||||
p2, is_new2 = self.pm.pin_chat_send(sent_text, node="646")
|
||||
self.assertFalse(is_new2)
|
||||
self.assertEqual(p2["id"], prompt["id"])
|
||||
|
||||
def test_toggle_pin_chat_send(self):
|
||||
"""Verify toggle_pin_chat_send flips pinned state between True and False."""
|
||||
sent_text = "Investigate high CDP drop rates and verify routing table."
|
||||
p1, is_new = self.pm.toggle_pin_chat_send(sent_text, node="646")
|
||||
self.assertTrue(is_new)
|
||||
self.assertTrue(p1["pinned"])
|
||||
|
||||
# Toggle again: should unpin
|
||||
p2, is_new2 = self.pm.toggle_pin_chat_send(sent_text, node="646")
|
||||
self.assertFalse(is_new2)
|
||||
self.assertFalse(p2["pinned"])
|
||||
|
||||
# Toggle again: should re-pin
|
||||
p3, is_new3 = self.pm.toggle_pin_chat_send(sent_text, node="646")
|
||||
self.assertFalse(is_new3)
|
||||
self.assertTrue(p3["pinned"])
|
||||
|
||||
|
||||
class TestPromptTUIIntegration(unittest.TestCase):
|
||||
"""Test suite covering TUI compact shelf, click auto-fill, modals, and sends search."""
|
||||
|
||||
def setUp(self):
|
||||
self.temp_dir = tempfile.TemporaryDirectory()
|
||||
self.storage_file = Path(self.temp_dir.name) / "test_prompts.json"
|
||||
|
||||
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.data.stop()
|
||||
self.tui.safe_addstr = MagicMock()
|
||||
|
||||
# Replace prompt manager with isolated instance
|
||||
self.tui.prompt_manager = muse_mod.PromptManager(storage_path=self.storage_file)
|
||||
self.tui.prompts = self.tui.prompt_manager.get_sorted_prompts()
|
||||
|
||||
# Clear history cache to isolate unit test data
|
||||
self.tui.data.history_cache.clear()
|
||||
|
||||
# Seed sample chat history
|
||||
self.tui.data.history_cache[("muse", "thread_1")] = [
|
||||
{"role": "user", "text": "Check all docker network bridges.", "message_id": "m1"},
|
||||
{"role": "assistant", "text": "All docker network bridges are operational.", "message_id": "m2"},
|
||||
{"role": "user", "text": "Inspect CDP latency on pip and opm.", "message_id": "m3"},
|
||||
]
|
||||
self.tui.data.active_node = "muse"
|
||||
self.tui.data.active_thread_id = "thread_1"
|
||||
self.tui.data.active_thread_title = "thread_1"
|
||||
self.tui.input_history = ["Recent typed command from session"]
|
||||
|
||||
def tearDown(self):
|
||||
self.temp_dir.cleanup()
|
||||
|
||||
def test_compact_prompt_shelf_render_and_chip_bounds(self):
|
||||
"""Verify _render_prompt_shelf populates chip bounds and button boundaries."""
|
||||
self.tui._render_prompt_shelf(y=24, x=0, h=2, w=100)
|
||||
bounds = self.tui._prompt_chip_bounds
|
||||
self.assertGreater(len(bounds), 0, "Prompt chips should be registered with bounding boxes")
|
||||
|
||||
first_start, first_end, first_prompt = bounds[0]
|
||||
self.assertGreater(first_end, first_start)
|
||||
self.assertIn("title", first_prompt)
|
||||
self.assertTrue(hasattr(self.tui, "_btn_expand_bounds"))
|
||||
self.assertTrue(hasattr(self.tui, "_btn_sends_bounds"))
|
||||
|
||||
def test_one_click_auto_fill_from_shelf(self):
|
||||
"""Verify clicking a prompt chip immediately auto-fills input_buf and sets INSERT mode."""
|
||||
self.tui._render_prompt_shelf(y=24, x=0, h=2, w=100)
|
||||
first_start, first_end, first_prompt = self.tui._prompt_chip_bounds[0]
|
||||
|
||||
# Simulate click on chip
|
||||
click_x = (first_start + first_end) // 2
|
||||
click_y = 25 # Row 1 of prompt shelf (chip row)
|
||||
res = self.tui._handle_mouse(mx=click_x, my=click_y, bstate=curses.BUTTON1_CLICKED)
|
||||
self.assertTrue(res)
|
||||
|
||||
# Input buffer should contain prompt text
|
||||
self.assertEqual(self.tui.input_buf, first_prompt["prompt"])
|
||||
self.assertEqual(self.tui.input_cursor, len(first_prompt["prompt"]))
|
||||
self.assertEqual(self.tui.editor_mode, "INSERT")
|
||||
self.assertIn("Filled prompt", self.tui.toast_msg)
|
||||
|
||||
def test_expand_prompts_modal_and_vim_navigation(self):
|
||||
"""Verify pressing 'P' expands modal and hjkl keys navigate and Enter fills."""
|
||||
# Press capital 'P' in NORMAL mode
|
||||
self.tui.editor_mode = "NORMAL"
|
||||
self.tui.focus_pane = "transcript"
|
||||
res = self.tui._handle_key(ord('P'))
|
||||
self.assertTrue(res)
|
||||
self.assertEqual(self.tui.modal, "prompts")
|
||||
self.assertEqual(self.tui.prompt_sel_idx, 0)
|
||||
|
||||
# 'j' moves down
|
||||
self.tui._handle_key(ord('j'))
|
||||
self.assertEqual(self.tui.prompt_sel_idx, 1)
|
||||
|
||||
# 'k' moves up
|
||||
self.tui._handle_key(ord('k'))
|
||||
self.assertEqual(self.tui.prompt_sel_idx, 0)
|
||||
|
||||
# 'G' jumps to last
|
||||
self.tui._handle_key(ord('G'))
|
||||
self.assertEqual(self.tui.prompt_sel_idx, len(self.tui.prompts) - 1)
|
||||
|
||||
# 'g' jumps to first
|
||||
self.tui._handle_key(ord('g'))
|
||||
self.assertEqual(self.tui.prompt_sel_idx, 0)
|
||||
|
||||
# Enter fills selected prompt and switches to INSERT
|
||||
expected_text = self.tui.prompts[0]["prompt"]
|
||||
self.tui._handle_key(10) # Enter
|
||||
self.assertIsNone(self.tui.modal)
|
||||
self.assertEqual(self.tui.input_buf, expected_text)
|
||||
self.assertEqual(self.tui.editor_mode, "INSERT")
|
||||
|
||||
def test_searchable_chat_history_sends(self):
|
||||
"""Verify get_chat_sends aggregates past sends and filters with query."""
|
||||
sends = self.tui.get_chat_sends()
|
||||
self.assertGreaterEqual(len(sends), 3)
|
||||
texts = [s["text"] for s in sends]
|
||||
self.assertIn("Recent typed command from session", texts)
|
||||
self.assertIn("Check all docker network bridges.", texts)
|
||||
self.assertIn("Inspect CDP latency on pip and opm.", texts)
|
||||
|
||||
# Filter query
|
||||
filtered = self.tui.get_chat_sends("docker")
|
||||
self.assertEqual(len(filtered), 1)
|
||||
self.assertEqual(filtered[0]["text"], "Check all docker network bridges.")
|
||||
|
||||
def test_pin_chat_send_to_prompt_fill_in_history_search(self):
|
||||
"""Verify 'p' in history_search toggles pin and unpin for highlighted send."""
|
||||
self.tui.modal = "history_search"
|
||||
self.tui.modal_input_buf = ""
|
||||
self.tui.history_search_sel = 0
|
||||
|
||||
initial_prompts_count = len(self.tui.prompts)
|
||||
sends = self.tui.get_chat_sends()
|
||||
target_send_text = sends[0]["text"]
|
||||
|
||||
# Press 'p': 1st time pins
|
||||
res = self.tui._handle_key(ord('p'))
|
||||
self.assertTrue(res)
|
||||
|
||||
self.assertEqual(len(self.tui.prompts), initial_prompts_count + 1)
|
||||
top_prompt = self.tui.prompts[0]
|
||||
self.assertTrue(top_prompt["pinned"])
|
||||
self.assertEqual(top_prompt["prompt"], target_send_text)
|
||||
self.assertIn("Pinned", self.tui.toast_msg)
|
||||
|
||||
# Press 'p': 2nd time UNPINS
|
||||
res2 = self.tui._handle_key(ord('p'))
|
||||
self.assertTrue(res2)
|
||||
|
||||
target_prompt_obj = [p for p in self.tui.prompt_manager.prompts if p["prompt"] == target_send_text][0]
|
||||
self.assertFalse(target_prompt_obj["pinned"])
|
||||
self.assertIn("Unpinned", self.tui.toast_msg)
|
||||
|
||||
# Press 'p': 3rd time RE-PINS
|
||||
res3 = self.tui._handle_key(ord('p'))
|
||||
self.assertTrue(res3)
|
||||
self.assertTrue(target_prompt_obj["pinned"])
|
||||
self.assertIn("Pinned", self.tui.toast_msg)
|
||||
|
||||
def test_history_search_mouse_toggle_pin(self):
|
||||
"""Verify clicking the pin column in history_search toggles pin/unpin."""
|
||||
self.tui.modal = "history_search"
|
||||
self.tui.modal_input_buf = ""
|
||||
self.tui.history_search_sel = 0
|
||||
|
||||
sends = self.tui.get_chat_sends()
|
||||
target_text = sends[0]["text"]
|
||||
|
||||
h, w = self.tui.stdscr.getmaxyx()
|
||||
modal_w = min(74, w - 6)
|
||||
modal_h = min(20, h - 4)
|
||||
top_y = (h - modal_h) // 2
|
||||
left_x = (w - modal_w) // 2
|
||||
|
||||
# 1st click pins
|
||||
click_y = top_y + 4
|
||||
click_x = left_x + 4
|
||||
handled = self.tui._handle_mouse(click_x, click_y, curses.BUTTON1_CLICKED)
|
||||
self.assertTrue(handled)
|
||||
target_prompt = [p for p in self.tui.prompt_manager.prompts if p["prompt"] == target_text][0]
|
||||
self.assertTrue(target_prompt["pinned"])
|
||||
self.assertIn("Pinned", self.tui.toast_msg)
|
||||
|
||||
# 2nd click unpins
|
||||
handled2 = self.tui._handle_mouse(click_x, click_y, curses.BUTTON1_CLICKED)
|
||||
self.assertTrue(handled2)
|
||||
self.assertFalse(target_prompt["pinned"])
|
||||
self.assertIn("Unpinned", self.tui.toast_msg)
|
||||
|
||||
def test_transcript_view_p_pins_latest_send(self):
|
||||
"""Verify pressing 'p' on transcript pins the current thread's latest sent message."""
|
||||
self.tui.modal = None
|
||||
self.tui.editor_mode = "NORMAL"
|
||||
self.tui.focus_pane = "transcript"
|
||||
|
||||
initial_count = len(self.tui.prompts)
|
||||
# In current thread, latest user message is "Inspect CDP latency on pip and opm."
|
||||
res = self.tui._handle_key(ord('p'))
|
||||
self.assertTrue(res)
|
||||
|
||||
self.assertEqual(len(self.tui.prompts), initial_count + 1)
|
||||
self.assertEqual(self.tui.prompts[0]["prompt"], "Inspect CDP latency on pip and opm.")
|
||||
self.assertTrue(self.tui.prompts[0]["pinned"])
|
||||
self.assertIn("Pinned", self.tui.toast_msg)
|
||||
|
||||
def test_sidebar_p_still_toggles_thread_pin(self):
|
||||
"""Verify pressing 'p' when focus is on sidebar preserves thread pinning."""
|
||||
self.tui.modal = None
|
||||
self.tui.editor_mode = "NORMAL"
|
||||
self.tui.focus_pane = "sidebar"
|
||||
|
||||
# Ensure thread_1 is unpinned initially
|
||||
self.tui.data.pinned_threads.setdefault("muse", set()).discard("thread_1")
|
||||
|
||||
# Mock threads cache
|
||||
self.tui.data.threads_cache["muse"] = [
|
||||
{"session_id": "thread_1", "title": "Test Chat", "is_main": False}
|
||||
]
|
||||
self.tui.thread_sel_idx = 0
|
||||
|
||||
res = self.tui._handle_key(ord('p'))
|
||||
self.assertTrue(res)
|
||||
self.assertIn("thread_1", self.tui.data.pinned_threads.get("muse", set()))
|
||||
self.assertIn("Pinned chat", self.tui.toast_msg)
|
||||
|
||||
# Clean up disk state
|
||||
self.tui.data.toggle_pin_thread("muse", "thread_1")
|
||||
|
||||
def test_bracket_keys_scroll_prompt_shelf(self):
|
||||
"""Verify '[' and ']' scroll the compact prompt shelf horizontally."""
|
||||
self.tui.editor_mode = "NORMAL"
|
||||
self.tui.focus_pane = "transcript"
|
||||
self.tui.prompt_scroll_idx = 0
|
||||
|
||||
self.tui._handle_key(ord(']'))
|
||||
self.assertEqual(self.tui.prompt_scroll_idx, 1)
|
||||
|
||||
self.tui._handle_key(ord('['))
|
||||
self.assertEqual(self.tui.prompt_scroll_idx, 0)
|
||||
|
||||
def test_slash_prompt_and_sends_commands(self):
|
||||
"""Verify /prompt save and /sends slash commands formulate and open modals."""
|
||||
self.tui.input_buf = "Check loop telemetry on all nodes"
|
||||
self.tui._execute_input_line("/prompt save Telemetry Check")
|
||||
|
||||
prompts = self.tui.prompt_manager.get_sorted_prompts()
|
||||
saved = [p for p in prompts if p["title"] == "Telemetry Check"]
|
||||
self.assertEqual(len(saved), 1)
|
||||
self.assertEqual(saved[0]["prompt"], "Check loop telemetry on all nodes")
|
||||
|
||||
# /prompt add
|
||||
self.tui._execute_input_line("/prompt add Quick Probe | ping all nodes")
|
||||
saved_add = [p for p in self.tui.prompt_manager.get_sorted_prompts() if p["title"] == "Quick Probe"]
|
||||
self.assertEqual(len(saved_add), 1)
|
||||
self.assertEqual(saved_add[0]["prompt"], "ping all nodes")
|
||||
|
||||
# /prompt del
|
||||
self.tui._execute_input_line(f"/prompt del {saved_add[0]['id']}")
|
||||
saved_del = [p for p in self.tui.prompt_manager.get_sorted_prompts() if p["title"] == "Quick Probe"]
|
||||
self.assertEqual(len(saved_del), 0)
|
||||
|
||||
# /sends opens history search
|
||||
self.tui._execute_input_line("/sends docker")
|
||||
self.assertEqual(self.tui.modal, "history_search")
|
||||
self.assertEqual(self.tui.modal_input_buf, "docker")
|
||||
|
||||
def test_shelf_buttons_mouse_clicks(self):
|
||||
"""Verify clicking buttons on row 0 of prompt shelf opens modals and scrolls."""
|
||||
self.tui._render_prompt_shelf(y=24, x=0, h=2, w=100)
|
||||
|
||||
# Click [P:Expand]
|
||||
exp_x = (self.tui._btn_expand_bounds[0] + self.tui._btn_expand_bounds[1]) // 2
|
||||
self.tui._handle_mouse(mx=exp_x, my=24, bstate=curses.BUTTON1_CLICKED)
|
||||
self.assertEqual(self.tui.modal, "prompts")
|
||||
self.tui.modal = None
|
||||
|
||||
# Click [🔍 Sends]
|
||||
sends_x = (self.tui._btn_sends_bounds[0] + self.tui._btn_sends_bounds[1]) // 2
|
||||
self.tui._handle_mouse(mx=sends_x, my=24, bstate=curses.BUTTON1_CLICKED)
|
||||
self.assertEqual(self.tui.modal, "history_search")
|
||||
self.tui.modal = None
|
||||
|
||||
# Click [►]
|
||||
next_x = (self.tui._btn_next_bounds[0] + self.tui._btn_next_bounds[1]) // 2
|
||||
self.tui.prompt_scroll_idx = 0
|
||||
self.tui._handle_mouse(mx=next_x, my=24, bstate=curses.BUTTON1_CLICKED)
|
||||
self.assertEqual(self.tui.prompt_scroll_idx, 1)
|
||||
|
||||
# Click [◄]
|
||||
prev_x = (self.tui._btn_prev_bounds[0] + self.tui._btn_prev_bounds[1]) // 2
|
||||
self.tui._handle_mouse(mx=prev_x, my=24, bstate=curses.BUTTON1_CLICKED)
|
||||
self.assertEqual(self.tui.prompt_scroll_idx, 0)
|
||||
|
||||
def test_mouse_wheel_scrolling_on_shelf(self):
|
||||
"""Verify mouse wheel up/down over prompt shelf scrolls chips."""
|
||||
self.tui._render_prompt_shelf(y=24, x=0, h=2, w=100)
|
||||
self.tui.prompt_scroll_idx = 0
|
||||
|
||||
# Wheel down over prompt shelf (my=25)
|
||||
self.tui._handle_mouse(mx=50, my=25, bstate=0x200000)
|
||||
self.assertEqual(self.tui.prompt_scroll_idx, 1)
|
||||
|
||||
# Wheel up over prompt shelf (my=25)
|
||||
self.tui._handle_mouse(mx=50, my=25, bstate=0x10000)
|
||||
self.assertEqual(self.tui.prompt_scroll_idx, 0)
|
||||
|
||||
def test_prompt_modal_pin_toggle_and_delete(self):
|
||||
"""Verify 'p' toggles pin and 'd' deletes in prompt modal."""
|
||||
self.tui.modal = "prompts"
|
||||
self.tui.prompt_sel_idx = 0
|
||||
target = self.tui.prompts[0]
|
||||
initial_pin = target.get("pinned", False)
|
||||
|
||||
# Press 'p' to toggle pin
|
||||
self.tui._handle_key(ord('p'))
|
||||
toggled = [p for p in self.tui.prompts if p["id"] == target["id"]][0]
|
||||
self.assertEqual(toggled.get("pinned", False), not initial_pin)
|
||||
|
||||
# Now select the prompt and press 'd' to delete
|
||||
target_idx = [i for i, p in enumerate(self.tui.prompts) if p["id"] == target["id"]][0]
|
||||
self.tui.prompt_sel_idx = target_idx
|
||||
self.tui._handle_key(ord('d'))
|
||||
self.assertFalse(any(p["id"] == target["id"] for p in self.tui.prompts))
|
||||
|
||||
def test_formulate_prompt_modal(self):
|
||||
"""Verify prompt_formulate modal creates and saves new prompt."""
|
||||
self.tui.modal = "prompt_formulate"
|
||||
self.tui.input_buf = "Draft message from input composer"
|
||||
self.tui.modal_input_buf = "New Diagnostic"
|
||||
self.tui.modal_input_cursor = len(self.tui.modal_input_buf)
|
||||
|
||||
# Press Enter
|
||||
self.tui._handle_key(10)
|
||||
self.assertIsNone(self.tui.modal)
|
||||
|
||||
# Check prompt exists in manager
|
||||
matching = [p for p in self.tui.prompts if p["title"] == "New Diagnostic"]
|
||||
self.assertEqual(len(matching), 1)
|
||||
self.assertEqual(matching[0]["prompt"], "Draft message from input composer")
|
||||
self.assertTrue(matching[0]["pinned"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,204 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
test_rate_limits.py — Tests for rate-limiting, adaptive polling, backoff, and circuit breaker.
|
||||
"""
|
||||
import curses
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
MUSE_TUI_PATH = REPO_ROOT / "bin" / "muse-tui.py"
|
||||
|
||||
spec = importlib.util.spec_from_file_location("muse_tui_rl", MUSE_TUI_PATH)
|
||||
muse_tui_rl = importlib.util.module_from_spec(spec)
|
||||
sys.modules["muse_tui_rl"] = muse_tui_rl
|
||||
spec.loader.exec_module(muse_tui_rl)
|
||||
|
||||
FleetDataManager = muse_tui_rl.FleetDataManager
|
||||
MuseTUI = muse_tui_rl.MuseTUI
|
||||
run_command_isolated = muse_tui_rl.run_command_isolated
|
||||
|
||||
|
||||
class TestRateLimitingAndBackoff(unittest.TestCase):
|
||||
def setUp(self):
|
||||
# In tests, start_poller is disabled by default
|
||||
self.mgr = FleetDataManager(start_poller=False)
|
||||
|
||||
def test_poller_disabled_in_tests_by_default(self):
|
||||
"""FleetDataManager does not leak background poller threads during unit test runs."""
|
||||
self.assertIsNone(self.mgr.poller_thread)
|
||||
|
||||
def test_rate_limit_cooldown_marking(self):
|
||||
"""Marking a node as rate-limited sets cooldown window and rate_limited flag."""
|
||||
self.assertFalse(self.mgr.is_node_rate_limited("muse"))
|
||||
self.assertEqual(self.mgr.get_node_cooldown_remaining("muse"), 0)
|
||||
|
||||
self.mgr.mark_node_rate_limited("muse", cooldown_seconds=30.0, reason="HTTP 429")
|
||||
self.assertTrue(self.mgr.is_node_rate_limited("muse"))
|
||||
self.assertGreater(self.mgr.get_node_cooldown_remaining("muse"), 20)
|
||||
self.assertTrue(self.mgr.node_status["muse"]["rate_limited"])
|
||||
|
||||
def test_tiered_backoff_progression(self):
|
||||
"""Consecutive rate-limit occurrences escalate (30s -> 60s -> 120s max), and clear resets."""
|
||||
# Incident 1: 30s
|
||||
self.mgr.mark_node_rate_limited("pip", reason="429 first")
|
||||
cd1 = self.mgr.get_node_cooldown_remaining("pip")
|
||||
self.assertTrue(25 <= cd1 <= 31)
|
||||
|
||||
# Incident 2: 60s
|
||||
self.mgr.mark_node_rate_limited("pip", reason="429 second")
|
||||
cd2 = self.mgr.get_node_cooldown_remaining("pip")
|
||||
self.assertTrue(55 <= cd2 <= 61)
|
||||
|
||||
# Incident 3: 120s max
|
||||
self.mgr.mark_node_rate_limited("pip", reason="429 third")
|
||||
cd3 = self.mgr.get_node_cooldown_remaining("pip")
|
||||
self.assertTrue(110 <= cd3 <= 121)
|
||||
|
||||
# Success resets backoff
|
||||
self.mgr.clear_node_rate_limit("pip")
|
||||
self.assertFalse(self.mgr.is_node_rate_limited("pip"))
|
||||
self.assertEqual(self.mgr.get_node_cooldown_remaining("pip"), 0)
|
||||
self.assertEqual(self.mgr.rate_limit_consecutive["pip"], 0)
|
||||
|
||||
def test_retry_after_header_parsing(self):
|
||||
"""When output includes Retry-After, that explicit duration is used."""
|
||||
with patch.object(muse_tui_rl, "run_command_isolated", return_value=(1, "", "Error 429: Too Many Requests. Retry-After: 85")):
|
||||
self.mgr._fetch_history("646", "thread-xyz")
|
||||
self.assertTrue(self.mgr.is_node_rate_limited("646"))
|
||||
cd = self.mgr.get_node_cooldown_remaining("646")
|
||||
self.assertTrue(80 <= cd <= 86)
|
||||
|
||||
def test_rate_limited_node_skips_fetches(self):
|
||||
"""Rate-limited nodes are skipped by _fetch_history and _run_fetch_threads."""
|
||||
self.mgr.mark_node_rate_limited("muse", cooldown_seconds=60.0)
|
||||
|
||||
with patch.object(muse_tui_rl, "run_command_isolated") as mock_cmd:
|
||||
self.mgr._fetch_history("muse", "thread-123")
|
||||
mock_cmd.assert_not_called()
|
||||
|
||||
self.mgr._run_fetch_threads("muse")
|
||||
mock_cmd.assert_not_called()
|
||||
|
||||
def test_rate_limit_detection_from_command_output(self):
|
||||
"""When CLI command outputs 429 or rate limit text, node is placed on cooldown."""
|
||||
with patch.object(muse_tui_rl, "run_command_isolated", return_value=(1, "", "Error 429: Too Many Requests")):
|
||||
self.mgr._fetch_history("pip", "thread-abc")
|
||||
self.assertTrue(self.mgr.is_node_rate_limited("pip"))
|
||||
self.assertGreater(self.mgr.get_node_cooldown_remaining("pip"), 0)
|
||||
|
||||
def test_record_activity_updates_timestamp(self):
|
||||
"""record_activity updates last_user_activity timestamp."""
|
||||
old_time = self.mgr.last_user_activity
|
||||
time.sleep(0.01)
|
||||
self.mgr.record_activity()
|
||||
self.assertGreaterEqual(self.mgr.last_user_activity, old_time)
|
||||
|
||||
def test_run_command_isolated_handles_quick_command(self):
|
||||
"""run_command_isolated runs a command and returns returncode, stdout, stderr."""
|
||||
rc, stdout, stderr = run_command_isolated(["echo", "hello rate limit"], timeout=2.0)
|
||||
self.assertEqual(rc, 0)
|
||||
self.assertIn("hello rate limit", stdout)
|
||||
|
||||
def test_run_command_isolated_terminates_on_timeout(self):
|
||||
"""run_command_isolated cleanly kills process group on timeout without zombies."""
|
||||
rc, stdout, stderr = run_command_isolated(["sleep", "10"], timeout=0.1)
|
||||
self.assertEqual(rc, -1)
|
||||
self.assertIn("timed out", stderr)
|
||||
|
||||
|
||||
class TestRateLimitUserInteraction(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.mock_stdscr = MagicMock()
|
||||
self.mock_stdscr.getmaxyx.return_value = (30, 100)
|
||||
self.mock_stdscr.getch.return_value = -1
|
||||
self.tui = MuseTUI(self.mock_stdscr, initial_mode="muse", initial_node="muse")
|
||||
self.tui.safe_addstr = MagicMock()
|
||||
|
||||
def test_soft_guardrail_send_during_cooldown(self):
|
||||
"""Sending during cooldown warns on first Enter, preserves buffer, and forces on second Enter."""
|
||||
self.tui.data.mark_node_rate_limited("muse", cooldown_seconds=30.0)
|
||||
self.tui.editor_mode = "INSERT"
|
||||
self.tui.input_buf = "Status report please"
|
||||
self.tui.input_cursor = len(self.tui.input_buf)
|
||||
|
||||
# First Enter tap: warns and does not clear input_buf
|
||||
with patch.object(self.tui, "_async_send_message") as mock_send:
|
||||
self.tui._handle_insert_key(10)
|
||||
mock_send.assert_not_called()
|
||||
self.assertEqual(self.tui.input_buf, "Status report please")
|
||||
self.assertIn("cooldown", self.tui.toast_msg)
|
||||
self.assertIn("Press Enter again", self.tui.toast_msg)
|
||||
|
||||
# Second Enter tap within 2.5s: bypasses cooldown, loads into history, and delivers
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
self.tui._handle_insert_key(10)
|
||||
self.assertFalse(self.tui.data.is_node_rate_limited("muse"))
|
||||
self.assertIn("Sending message to MUSE", self.tui.toast_msg)
|
||||
self.assertEqual(self.tui.input_buf, "")
|
||||
# Ensure message was immediately loaded into history cache!
|
||||
msgs = self.tui.data.history_cache.get(("muse", self.tui.data.active_thread_id), [])
|
||||
self.assertTrue(any(m.get("text") == "Status report please" and m.get("role") == "user" for m in msgs))
|
||||
|
||||
def test_two_tap_manual_sync_override(self):
|
||||
"""Pressing 'r' during cooldown warns on first press and bypasses cooldown on double-tap."""
|
||||
self.tui.data.mark_node_rate_limited("muse", cooldown_seconds=45.0)
|
||||
|
||||
# First 'r' tap: warns
|
||||
self.tui._handle_normal_key(ord('r'))
|
||||
self.assertTrue(self.tui.data.is_node_rate_limited("muse"))
|
||||
self.assertIn("cooling down", self.tui.toast_msg)
|
||||
self.assertIn("Press 'r' again", self.tui.toast_msg)
|
||||
|
||||
# Second 'r' tap within 2s: clears rate limit and forces sync
|
||||
with patch.object(self.tui.data, "lazy_fetch_threads") as mock_fetch:
|
||||
self.tui._handle_normal_key(ord('r'))
|
||||
self.assertFalse(self.tui.data.is_node_rate_limited("muse"))
|
||||
self.assertIn("Force-syncing", self.tui.toast_msg)
|
||||
mock_fetch.assert_called_with("muse", force=True)
|
||||
|
||||
def test_agent_context_menu_sync_override(self):
|
||||
"""Context menu 'sync' action also respects two-tap override during cooldown."""
|
||||
self.tui.data.mark_node_rate_limited("pip", cooldown_seconds=30.0)
|
||||
self.tui.context_agent = {"node": "pip"}
|
||||
|
||||
# First selection: warns
|
||||
self.tui._execute_agent_action("sync")
|
||||
self.assertTrue(self.tui.data.is_node_rate_limited("pip"))
|
||||
self.assertIn("cooling down", self.tui.toast_msg)
|
||||
|
||||
# Second selection within 2s: forces sync
|
||||
with patch.object(self.tui.data, "lazy_fetch_threads") as mock_fetch:
|
||||
self.tui._execute_agent_action("sync")
|
||||
self.assertFalse(self.tui.data.is_node_rate_limited("pip"))
|
||||
self.assertIn("Force-syncing", self.tui.toast_msg)
|
||||
mock_fetch.assert_called_with("pip", force=True)
|
||||
|
||||
def test_optimistic_message_persistence_across_fetch(self):
|
||||
"""Optimistic user messages are preserved even if server history lags behind."""
|
||||
self.tui.data.history_cache[("muse", "sess-test")] = []
|
||||
self.tui.data.add_optimistic_message("muse", "sess-test", "New uncommitted instruction")
|
||||
cached = self.tui.data.history_cache.get(("muse", "sess-test"), [])
|
||||
self.assertEqual(len(cached), 1)
|
||||
self.assertEqual(cached[0]["text"], "New uncommitted instruction")
|
||||
|
||||
# Simulate remote fetch returning older history that doesn't yet have the new message
|
||||
old_server_msgs = [{"role": "assistant", "text": "Earlier reply"}]
|
||||
with patch.object(muse_tui_rl, "run_command_isolated", return_value=(0, json.dumps(old_server_msgs), "")):
|
||||
self.tui.data._fetch_history("muse", "sess-test")
|
||||
updated = self.tui.data.history_cache.get(("muse", "sess-test"), [])
|
||||
# Both the older server message AND the pending user message should exist!
|
||||
self.assertEqual(len(updated), 2)
|
||||
self.assertEqual(updated[0]["text"], "Earlier reply")
|
||||
self.assertEqual(updated[1]["text"], "New uncommitted instruction")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import json
|
||||
unittest.main()
|
||||
@@ -0,0 +1,192 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user