operator
b7e45010c3
- Add clean line-by-line transcript style toggle ('b' key / /clean / /boxed)
- Implement right-click context menus for chat list, fleet agent panel, and transcript
- Add cooldown mode lock bypass (2-second double-confirm force sync)
- Implement dual copy support: clean text (strip reply metadata) vs full context
- Add clickable [📋 Copy] and [📑+ Context] buttons to message headers
- Add unit test suites for transcript cleaning, context menus, rate limits, and copy actions
167 lines
7.1 KiB
Python
167 lines
7.1 KiB
Python
#!/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()
|