operator
439 lines
18 KiB
Python
439 lines
18 KiB
Python
"""test_box_tui_render.py — Headless verification of Box TUI and Work Pipeline screens."""
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
import os
|
|
import sys
|
|
import curses
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
BIN_DIR = REPO_ROOT / "bin"
|
|
if str(BIN_DIR) not in sys.path:
|
|
sys.path.insert(0, str(BIN_DIR))
|
|
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location("muse_tui", str(BIN_DIR / "muse-tui.py"))
|
|
muse_tui = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(muse_tui)
|
|
|
|
spec_fleet = importlib.util.spec_from_file_location("box_fleet_tui", str(BIN_DIR / "box-fleet-tui.py"))
|
|
box_fleet_tui = importlib.util.module_from_spec(spec_fleet)
|
|
spec_fleet.loader.exec_module(box_fleet_tui)
|
|
|
|
|
|
class TestBoxTUIHeadless(unittest.TestCase):
|
|
def setUp(self):
|
|
self.mock_stdscr = MagicMock()
|
|
self.mock_stdscr.getmaxyx.return_value = (40, 120)
|
|
self.mock_stdscr.getch.return_value = -1
|
|
|
|
def test_box_tui_initialization_work_tab(self):
|
|
app = muse_tui.MuseTUI(self.mock_stdscr, initial_mode="box", initial_tab="work")
|
|
self.assertEqual(app.mode, "box")
|
|
self.assertEqual(app.box_tab, 7)
|
|
self.assertIn("8: Work Pipeline", muse_tui.BOX_TABS)
|
|
|
|
def test_render_work_view(self):
|
|
app = muse_tui.MuseTUI(self.mock_stdscr, initial_mode="box", initial_tab="work")
|
|
# Populate mock cognitive cache and work issues
|
|
app.data.cognitive_cache = {
|
|
"muse": {
|
|
"node": "muse",
|
|
"status": "IDLE",
|
|
"cognitive_lock": False,
|
|
"screen": {"url": "https://muse.ai/", "title": "Chat - muse"}
|
|
},
|
|
"pip": {
|
|
"node": "pip",
|
|
"status": "SIDECHAT_IDLE",
|
|
"cognitive_lock": False,
|
|
"screen": {"url": "https://muse.ai/thread/123", "title": "Chat - pip"}
|
|
}
|
|
}
|
|
app.data.work_issues_cache = [
|
|
{"number": 218, "state": "open", "title": "Autonomous TUI upgrade", "assignee": {"username": "dev"}}
|
|
]
|
|
app.data.work_prs_cache = [
|
|
{"number": 219, "state": "open", "merged": False, "title": "feat: box tui", "head": {"ref": "feature/tui"}}
|
|
]
|
|
|
|
# Call _render_work_view directly
|
|
app._render_work_view(2, 0, 35, 120)
|
|
self.assertTrue(self.mock_stdscr.addstr.called)
|
|
|
|
def test_box_tui_key_actions_on_work_tab(self):
|
|
app = muse_tui.MuseTUI(self.mock_stdscr, initial_mode="box", initial_tab="work")
|
|
app.data.nodes = ["muse", "pip", "646", "opm", "def", "dev"]
|
|
|
|
# 'w' opens work_dispatch modal
|
|
handled = app._handle_key(ord('w'))
|
|
self.assertTrue(handled)
|
|
self.assertEqual(app.modal, "work_dispatch")
|
|
|
|
# Esc closes modal
|
|
handled = app._handle_key(27)
|
|
self.assertTrue(handled)
|
|
self.assertIsNone(app.modal)
|
|
|
|
# Tab navigation: '1' switches to Chat (tab 0), '8' switches to Work (tab 7)
|
|
app._handle_key(ord('1'))
|
|
self.assertEqual(app.box_tab, 0)
|
|
app._handle_key(ord('8'))
|
|
self.assertEqual(app.box_tab, 7)
|
|
|
|
def test_work_dispatch_modal_text_input_and_cycle(self):
|
|
app = muse_tui.MuseTUI(self.mock_stdscr, initial_mode="box", initial_tab="work")
|
|
app.data.nodes = ["muse", "pip", "646", "opm", "def", "dev"]
|
|
app.modal = "work_dispatch"
|
|
app.modal_input_buf = ""
|
|
app.modal_input_cursor = 0
|
|
app.dispatch_target_node = "dev"
|
|
|
|
# Type chars 'Build #12'
|
|
for c in "Build #12":
|
|
app._handle_key(ord(c))
|
|
self.assertEqual(app.modal_input_buf, "Build #12")
|
|
|
|
# Tab cycles worker
|
|
app._handle_key(ord('\t'))
|
|
self.assertNotEqual(app.dispatch_target_node, "dev")
|
|
|
|
def test_agent_thoughts_modal(self):
|
|
app = muse_tui.MuseTUI(self.mock_stdscr, initial_mode="box", initial_tab="work")
|
|
app.selected_thought_node = "def"
|
|
app.agent_thought_data = {
|
|
"screen": {
|
|
"url": "https://muse.ai/",
|
|
"title": "Chat - def",
|
|
"is_generating": False,
|
|
"recent_paragraphs": ["Inspecting workspace...", "Code updated."]
|
|
}
|
|
}
|
|
app.modal = "agent_thoughts"
|
|
# Render modal
|
|
app._render_modal(40, 120)
|
|
self.assertTrue(self.mock_stdscr.addstr.called)
|
|
|
|
# Esc closes modal
|
|
app._handle_key(27)
|
|
self.assertIsNone(app.modal)
|
|
|
|
def test_agent_profile_menu_modal(self):
|
|
app = muse_tui.MuseTUI(self.mock_stdscr, initial_mode="box", initial_tab="work")
|
|
app.selected_menu_node = "pip"
|
|
app.menu_tab = "activity"
|
|
app.agent_menu_data = {
|
|
"tab": "activity",
|
|
"data": {
|
|
"items": [{"title": "Session started", "detail": "Active", "time": "12m ago"}]
|
|
}
|
|
}
|
|
app.modal = "agent_menu"
|
|
# Render modal
|
|
app._render_modal(40, 120)
|
|
self.assertTrue(self.mock_stdscr.addstr.called)
|
|
|
|
# Tab key cycles menu tabs
|
|
with patch.object(app, "_async_load_menu") as mock_load:
|
|
app._handle_key(ord('\t'))
|
|
mock_load.assert_called_with("pip", "upcoming")
|
|
|
|
def test_transcript_sub_header_render_and_mouse_click(self):
|
|
app = muse_tui.MuseTUI(self.mock_stdscr, initial_mode="muse")
|
|
app.data.active_node = "dev"
|
|
app.focus_pane = "transcript"
|
|
app._render_transcript(0, 0, 30, 100)
|
|
self.assertIsNotNone(app._btn_case_actions_bounds)
|
|
btn_y, b_start, b_end = app._btn_case_actions_bounds
|
|
|
|
# Click [w:Case] (offset +2)
|
|
with patch.object(app, "_open_agent_menu") as mock_menu:
|
|
handled = app._handle_mouse(b_start + 2, btn_y, curses.BUTTON1_CLICKED)
|
|
self.assertTrue(handled)
|
|
self.assertEqual(app.modal, "work_dispatch")
|
|
app.modal = None
|
|
|
|
# Click [f:Refocus] (offset +12)
|
|
with patch.object(app, "_async_refocus_main") as mock_refocus:
|
|
handled = app._handle_mouse(b_start + 12, btn_y, curses.BUTTON1_CLICKED)
|
|
self.assertTrue(handled)
|
|
mock_refocus.assert_called_with("dev")
|
|
|
|
# Click [t:Thoughts] (offset +25)
|
|
with patch.object(app, "_open_agent_thoughts") as mock_thoughts:
|
|
handled = app._handle_mouse(b_start + 25, btn_y, curses.BUTTON1_CLICKED)
|
|
self.assertTrue(handled)
|
|
mock_thoughts.assert_called_with("dev")
|
|
|
|
# Click [p:Menu] (offset +38)
|
|
with patch.object(app, "_open_agent_menu") as mock_menu:
|
|
handled = app._handle_mouse(b_start + 38, btn_y, curses.BUTTON1_CLICKED)
|
|
self.assertTrue(handled)
|
|
mock_menu.assert_called_with("dev")
|
|
|
|
# Click [h:Heal] (offset +46)
|
|
with patch.object(app, "_async_heal_agent") as mock_heal:
|
|
handled = app._handle_mouse(b_start + 46, btn_y, curses.BUTTON1_CLICKED)
|
|
self.assertTrue(handled)
|
|
mock_heal.assert_called_with("dev")
|
|
|
|
def test_agent_context_twelve_actions_and_hotkeys(self):
|
|
app = muse_tui.MuseTUI(self.mock_stdscr, initial_mode="muse")
|
|
app.data.active_node = "646"
|
|
app.context_agent = {"node": "646", "status": {}, "index": 0}
|
|
app.modal = "agent_context"
|
|
app._render_modal(40, 120)
|
|
|
|
self.assertEqual(len(app._agent_context_actions), 12)
|
|
|
|
# Hotkey 2 / w: Dispatch
|
|
app.modal = "agent_context"
|
|
app._handle_key(ord('2'))
|
|
self.assertEqual(app.modal, "work_dispatch")
|
|
self.assertEqual(app.dispatch_target_node, "646")
|
|
|
|
# Hotkey 3 / t: Thoughts
|
|
app.modal = "agent_context"
|
|
with patch.object(app, "_open_agent_thoughts") as mock_th:
|
|
app._handle_key(ord('3'))
|
|
mock_th.assert_called_with("646")
|
|
|
|
# Hotkey 4 / p: Menu
|
|
app.modal = "agent_context"
|
|
with patch.object(app, "_open_agent_menu") as mock_menu:
|
|
app._handle_key(ord('4'))
|
|
mock_menu.assert_called_with("646")
|
|
|
|
# Hotkey 5 / f: Refocus
|
|
app.modal = "agent_context"
|
|
with patch.object(app, "_async_refocus_main") as mock_ref:
|
|
app._handle_key(ord('5'))
|
|
mock_ref.assert_called_with("646")
|
|
|
|
# Hotkey 7 / h: Heal
|
|
app.modal = "agent_context"
|
|
with patch.object(app, "_async_heal_agent") as mock_heal:
|
|
app._handle_key(ord('7'))
|
|
mock_heal.assert_called_with("646")
|
|
|
|
# Hotkey 8 / l: Loopback
|
|
app.modal = "agent_context"
|
|
with patch.object(app, "_async_run_loopback") as mock_loop:
|
|
app._handle_key(ord('8'))
|
|
mock_loop.assert_called_once()
|
|
|
|
def test_work_dispatch_presets_f1_to_f4(self):
|
|
app = muse_tui.MuseTUI(self.mock_stdscr, initial_mode="muse")
|
|
app.modal = "work_dispatch"
|
|
app.modal_input_buf = ""
|
|
app.modal_input_cursor = 0
|
|
app.dispatch_target_node = "dev"
|
|
|
|
app._handle_key(curses.KEY_F1)
|
|
self.assertEqual(app.modal_input_buf, "Build: Implement feature specification and tests")
|
|
|
|
app._handle_key(curses.KEY_F2)
|
|
self.assertEqual(app.modal_input_buf, "Bugfix: Investigate and resolve error trace")
|
|
|
|
app._handle_key(curses.KEY_F3)
|
|
self.assertEqual(app.modal_input_buf, "Review: Code review and verify pull request changes")
|
|
|
|
app._handle_key(curses.KEY_F4)
|
|
self.assertEqual(app.modal_input_buf, "Audit: Pre-flight health and security audit")
|
|
|
|
# Enter triggers dispatch
|
|
with patch("threading.Thread") as mock_thread:
|
|
app._handle_key(10)
|
|
mock_thread.assert_called()
|
|
self.assertIsNone(app.modal)
|
|
|
|
def test_main_panel_normal_mode_orchestration_hotkeys(self):
|
|
app = muse_tui.MuseTUI(self.mock_stdscr, initial_mode="muse")
|
|
app.data.active_node = "opm"
|
|
app.focus_pane = "transcript"
|
|
app.editor_mode = "NORMAL"
|
|
|
|
# 'w' -> work_dispatch
|
|
app._handle_key(ord('w'))
|
|
self.assertEqual(app.modal, "work_dispatch")
|
|
self.assertEqual(app.dispatch_target_node, "opm")
|
|
app.modal = None
|
|
|
|
# 'f' -> refocus
|
|
with patch.object(app, "_async_refocus_main") as mock_ref:
|
|
app._handle_key(ord('f'))
|
|
mock_ref.assert_called_with("opm")
|
|
|
|
# 't' -> thoughts
|
|
with patch.object(app, "_open_agent_thoughts") as mock_th:
|
|
app._handle_key(ord('t'))
|
|
mock_th.assert_called_with("opm")
|
|
|
|
# 'p' -> menu
|
|
with patch.object(app, "_open_agent_menu") as mock_menu:
|
|
app._handle_key(ord('p'))
|
|
mock_menu.assert_called_with("opm")
|
|
|
|
# 'h' / 'H' -> heal
|
|
with patch.object(app, "_async_heal_agent") as mock_heal:
|
|
app._handle_key(ord('h'))
|
|
mock_heal.assert_called_with("opm")
|
|
|
|
def test_work_pipeline_assign_and_merge_hotkeys(self):
|
|
app = muse_tui.MuseTUI(self.mock_stdscr, initial_mode="box", initial_tab="work")
|
|
app.box_tab = 7
|
|
app.work_sel_idx = 1 # 646 in all_workers
|
|
|
|
with patch.object(app, "_async_merge_pr") as mock_merge:
|
|
app._handle_key(ord('m'))
|
|
mock_merge.assert_called_once()
|
|
|
|
with patch.object(app, "_async_assign_ticket") as mock_assign:
|
|
app._handle_key(ord('a'))
|
|
mock_assign.assert_called_with("646")
|
|
|
|
def test_muse_tui_all_tabs_fuzz_dimensions_and_corrupt_data(self):
|
|
dimensions = [(8, 20), (12, 40), (24, 80), (50, 160)]
|
|
|
|
corrupt_dataset = {
|
|
"cognitive": {
|
|
"muse": None,
|
|
"pip": {"status": None, "cognitive_lock": None, "screen": None},
|
|
"dev": {"status": "THINKING", "cognitive_lock": True, "screen": {"url": None, "title": None}}
|
|
},
|
|
"issues": [
|
|
{"number": 1, "state": None, "title": None, "assignee": None, "labels": None},
|
|
{"number": 2, "state": "open", "title": "Test", "assignee": {"username": None}, "labels": [None, {"name": None}]}
|
|
],
|
|
"prs": [
|
|
{"number": 1, "state": None, "merged": False, "title": None, "head": None},
|
|
{"number": 2, "state": "open", "merged": True, "title": "PR test", "head": {"ref": None}}
|
|
],
|
|
"approvals": [
|
|
{"node": None, "status": None, "target": None},
|
|
{"node": "pip", "status": "INPUT_WAIT", "input_waits": [None, {"task": None}]}
|
|
],
|
|
"jobs": [
|
|
{"name": None, "agent": None, "schedule": None, "timeout": None, "description": None}
|
|
],
|
|
"tmux": ["", "invalid:session:colon"],
|
|
"dm_logs": [
|
|
{"type": None, "agent": None, "to": None, "target": None, "ts": None, "id": None}
|
|
]
|
|
}
|
|
|
|
for h, w in dimensions:
|
|
stdscr = MagicMock()
|
|
stdscr.getmaxyx.return_value = (h, w)
|
|
stdscr.getch.return_value = -1
|
|
|
|
app = muse_tui.MuseTUI(stdscr, initial_mode="box", initial_tab="chat")
|
|
app._term_dims = (h, w)
|
|
|
|
# Test with empty caches
|
|
app.data.cognitive_cache = {}
|
|
app.data.work_issues_cache = []
|
|
app.data.work_prs_cache = []
|
|
app.data.approvals_cache = []
|
|
app.data.jobs_cache = []
|
|
app.data.dm_logs_cache = []
|
|
|
|
content_h = max(1, h - 4)
|
|
for tab_idx in range(8):
|
|
app.box_tab = tab_idx
|
|
if tab_idx == 0:
|
|
app._render_muse_view(2, 0, content_h, w)
|
|
elif tab_idx == 1:
|
|
app._render_fleet_table(2, 0, content_h, w)
|
|
elif tab_idx == 2:
|
|
app._render_approvals_view(2, 0, content_h, w)
|
|
elif tab_idx == 3:
|
|
app._render_jobs_view(2, 0, content_h, w)
|
|
elif tab_idx == 4:
|
|
with patch("subprocess.run") as mock_sub:
|
|
mock_sub.return_value = MagicMock(returncode=0, stdout="s1: 1 windows\n")
|
|
app._render_tmux_view(2, 0, content_h, w)
|
|
elif tab_idx == 5:
|
|
app._render_dm_logs_view(2, 0, content_h, w)
|
|
elif tab_idx == 6:
|
|
app._render_ssh_view(2, 0, content_h, w)
|
|
elif tab_idx == 7:
|
|
app._render_work_view(2, 0, content_h, w)
|
|
|
|
# Test with corrupt / null caches
|
|
app.data.cognitive_cache = corrupt_dataset["cognitive"]
|
|
app.data.work_issues_cache = corrupt_dataset["issues"]
|
|
app.data.work_prs_cache = corrupt_dataset["prs"]
|
|
app.data.approvals_cache = corrupt_dataset["approvals"]
|
|
app.data.jobs_cache = corrupt_dataset["jobs"]
|
|
app.data.dm_logs_cache = corrupt_dataset["dm_logs"]
|
|
|
|
for tab_idx in range(8):
|
|
app.box_tab = tab_idx
|
|
if tab_idx == 0:
|
|
app._render_muse_view(2, 0, content_h, w)
|
|
elif tab_idx == 1:
|
|
app._render_fleet_table(2, 0, content_h, w)
|
|
elif tab_idx == 2:
|
|
app._render_approvals_view(2, 0, content_h, w)
|
|
elif tab_idx == 3:
|
|
app._render_jobs_view(2, 0, content_h, w)
|
|
elif tab_idx == 4:
|
|
with patch("subprocess.run") as mock_sub:
|
|
mock_sub.return_value = MagicMock(returncode=0, stdout="s1: 1 windows\n")
|
|
app._render_tmux_view(2, 0, content_h, w)
|
|
elif tab_idx == 5:
|
|
app._render_dm_logs_view(2, 0, content_h, w)
|
|
elif tab_idx == 6:
|
|
app._render_ssh_view(2, 0, content_h, w)
|
|
elif tab_idx == 7:
|
|
app._render_work_view(2, 0, content_h, w)
|
|
|
|
def test_box_fleet_tui_all_tabs_fuzz_dimensions_and_corrupt_data(self):
|
|
dimensions = [(8, 20), (12, 40), (24, 80), (50, 160)]
|
|
for h, w in dimensions:
|
|
stdscr = MagicMock()
|
|
stdscr.getmaxyx.return_value = (h, w)
|
|
stdscr.getch.return_value = -1
|
|
|
|
fleet_app = box_fleet_tui.BoxFleetTUI(stdscr)
|
|
|
|
# 1. Empty snapshot
|
|
fleet_app.snapshot = {}
|
|
renderers = [fleet_app._render_timers, fleet_app._render_harvest,
|
|
fleet_app._render_followups, fleet_app._render_approvals,
|
|
fleet_app._render_activity, fleet_app._render_runtimes,
|
|
fleet_app._render_work]
|
|
|
|
for tab_idx, renderer in enumerate(renderers):
|
|
fleet_app.current_tab = tab_idx
|
|
renderer(h, w)
|
|
|
|
# 2. Corrupted / null snapshot
|
|
fleet_app.snapshot = {
|
|
"timers": {"rows": [{"timer": None, "next": None, "next_rel": None, "last": None, "last_rel": None, "activates": None}]},
|
|
"harvest": {"rows": [{"freshness": None, "agent": None, "thread": None, "watermark": None, "last": None, "ago": None}]},
|
|
"followups": {"counts": {"pending": 2}, "by_recipient": {"pip": 1}, "oldest_pending": None},
|
|
"approvals": {"total_pending": 1, "checked": 1, "unreachable": [], "pending": [{"node": None, "status": None, "target": None, "title": None}]},
|
|
"activity": {"rows": [{"freshness": None, "agent": None, "last": None, "last_rel": None, "source": None, "detail": None}]},
|
|
"runtimes": {"agents": [{"live": True, "briefed": None, "enabled": True, "session": None, "hat": None, "pane": None, "state": None, "mode": None, "watcher": None}], "queue": {}, "plan": {}},
|
|
"work": {
|
|
"cognitive": {"dev": None},
|
|
"issues": [{"number": None, "state": None, "assignee": None, "created_at": None, "title": None}],
|
|
"loopback_active": True
|
|
}
|
|
}
|
|
|
|
for tab_idx, renderer in enumerate(renderers):
|
|
fleet_app.current_tab = tab_idx
|
|
renderer(h, w)
|
|
|
|
fleet_app._render_header(w)
|
|
fleet_app._render_footer(h, w)
|
|
fleet_app._render_help(h, w)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|