operator
284 lines
12 KiB
Python
284 lines
12 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
|
|
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_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()
|