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
88 lines
3.4 KiB
Python
88 lines
3.4 KiB
Python
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()
|