operator
372 lines
15 KiB
Python
372 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""test_box_runtime.py — Runtime state sensing + `box runtime` management.
|
||
|
||
Covers: runtime_state classification (approval-pending/working/open-prompt),
|
||
muse argv approval-posture parsing, runtime_rows assembly (mocked tmux),
|
||
and the `box runtime` CLI surface.
|
||
"""
|
||
|
||
import json
|
||
import sys
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
|
||
REPO_ROOT = Path("/home/super/Projects/NetVM")
|
||
BIN_DIR = REPO_ROOT / "bin"
|
||
sys.path.insert(0, str(BIN_DIR))
|
||
|
||
import muse_choice_watcher as w
|
||
|
||
PROMPT = "❯" # Muse TUI input glyph (U+276F)
|
||
|
||
STATE_OPEN = (
|
||
"Some completed agent output here.\n"
|
||
"\n"
|
||
"───────────────────────────────────\n"
|
||
+ PROMPT + "\n"
|
||
"───────────────────────────────────\n"
|
||
" muse-spark-1.3-con… · Auto-review\n"
|
||
)
|
||
|
||
STATE_WORKING = (
|
||
"Partial agent output...\n"
|
||
"\n"
|
||
"… — running (10s · esc to interrupt)\n"
|
||
"\n"
|
||
"───────────────────────────────────\n"
|
||
+ PROMPT + "\n"
|
||
"───────────────────────────────────\n"
|
||
" muse-spark-1.3-con… · Auto-review\n"
|
||
)
|
||
|
||
STATE_WORKING_CUT = (
|
||
"Partial agent output...\n"
|
||
"◆ Calling tools (5m 53s · esc to in\n"
|
||
"\n"
|
||
"───────────────────────────────────\n"
|
||
+ PROMPT + "\n"
|
||
)
|
||
|
||
STATE_APPROVAL = (
|
||
"───────────────────────────────────\n"
|
||
"Would you like to run the following\n"
|
||
"\n"
|
||
" $ tmux capture-pane -p\n"
|
||
"\n"
|
||
"› 1. Yes, proceed (y)\n"
|
||
" 2. No, and tell Muse Code what to do instead\n"
|
||
)
|
||
|
||
STATE_SHELL = "[super@bl NetVM]$ printf 'hi'\nhi\n[super@bl NetVM]$ "
|
||
|
||
|
||
class TestRuntimeState(unittest.TestCase):
|
||
def test_open_prompt(self):
|
||
st = w.runtime_state(STATE_OPEN)
|
||
self.assertEqual(st["state"], "open-prompt")
|
||
self.assertIsNone(st["match"])
|
||
|
||
def test_working(self):
|
||
st = w.runtime_state(STATE_WORKING)
|
||
self.assertEqual(st["state"], "working")
|
||
|
||
def test_working_edge_cut_indicator(self):
|
||
st = w.runtime_state(STATE_WORKING_CUT)
|
||
self.assertEqual(st["state"], "working")
|
||
|
||
def test_approval_pending(self):
|
||
st = w.runtime_state(STATE_APPROVAL)
|
||
self.assertEqual(st["state"], "approval-pending")
|
||
self.assertEqual(st["match"]["kind"], "muse-approval")
|
||
self.assertEqual(st["match"]["key"], "1")
|
||
|
||
def test_approval_beats_working(self):
|
||
st = w.runtime_state(STATE_WORKING + STATE_APPROVAL)
|
||
self.assertEqual(st["state"], "approval-pending")
|
||
|
||
def test_working_beats_open_prompt(self):
|
||
# A working pane still renders its prompt footer.
|
||
st = w.runtime_state(STATE_WORKING)
|
||
self.assertEqual(st["state"], "working")
|
||
|
||
def test_shell_is_unknown(self):
|
||
self.assertEqual(w.runtime_state(STATE_SHELL)["state"], "unknown")
|
||
|
||
def test_empty_is_unknown(self):
|
||
self.assertEqual(w.runtime_state("")["state"], "unknown")
|
||
self.assertEqual(w.runtime_state(None)["state"], "unknown")
|
||
|
||
|
||
class TestApprovalFlags(unittest.TestCase):
|
||
def test_bare_is_not_auto(self):
|
||
p = w.muse_approval_flags(["/home/super/.local/bin/muse-bin-1.4.3"])
|
||
self.assertFalse(p["auto_approve"])
|
||
self.assertEqual(p["flags"], [])
|
||
|
||
def test_yolo(self):
|
||
p = w.muse_approval_flags(["muse", "--yolo"])
|
||
self.assertTrue(p["auto_approve"])
|
||
self.assertIn("yolo", p["flags"])
|
||
|
||
def test_disable_approval(self):
|
||
p = w.muse_approval_flags(["muse", "--disable-approval"])
|
||
self.assertTrue(p["auto_approve"])
|
||
|
||
def test_approval_mode_never(self):
|
||
p = w.muse_approval_flags(["muse", "--approval-mode", "never"])
|
||
self.assertTrue(p["auto_approve"])
|
||
self.assertIn("approval-mode=never", p["flags"])
|
||
|
||
def test_approval_mode_equals(self):
|
||
p = w.muse_approval_flags(["muse", "--approval-mode=never"])
|
||
self.assertTrue(p["auto_approve"])
|
||
|
||
def test_approval_mode_on_request_is_not_auto(self):
|
||
p = w.muse_approval_flags(["muse", "--approval-mode", "on-request"])
|
||
self.assertFalse(p["auto_approve"])
|
||
self.assertIn("approval-mode=on-request", p["flags"])
|
||
|
||
def test_empty_argv(self):
|
||
p = w.muse_approval_flags([])
|
||
self.assertFalse(p["auto_approve"])
|
||
|
||
|
||
def _tmux_result(returncode=0, stdout="", stderr=""):
|
||
r = mock.Mock()
|
||
r.returncode = returncode
|
||
r.stdout = stdout
|
||
r.stderr = stderr
|
||
return r
|
||
|
||
|
||
class TestRuntimeRows(unittest.TestCase):
|
||
LISTING = ("muse\t1\t%37\tmuse-bin-1.4.3-R5018.1\t2880158\t71\t27\n"
|
||
"muse\t1\t%38\tbash\t2880200\t100\t30\n")
|
||
|
||
def _patched(self, tmux_stdout=LISTING, tmux_rc=0, captures=None,
|
||
children=None, cmdlines=None, watcher=None):
|
||
captures = captures or {}
|
||
cmdlines = cmdlines or {}
|
||
children = children or {}
|
||
return (mock.patch.object(w, "_tmux", return_value=_tmux_result(
|
||
tmux_rc, tmux_stdout)),
|
||
mock.patch.object(w, "capture_pane",
|
||
side_effect=lambda s, p: captures.get(p)),
|
||
mock.patch.object(w, "_child_pids",
|
||
side_effect=lambda p: children.get(p, [])),
|
||
mock.patch.object(w, "_cmdline",
|
||
side_effect=lambda p: cmdlines.get(p, [])),
|
||
mock.patch.object(w, "is_running", return_value=watcher))
|
||
|
||
def test_rows_shape(self):
|
||
patches = self._patched(
|
||
captures={"%37": STATE_OPEN, "%38": STATE_SHELL},
|
||
children={2880158: [2881158]},
|
||
cmdlines={2881158: ["/home/super/.local/bin/muse-bin-1.4.3",
|
||
"--disable-approval"]},
|
||
watcher=1234)
|
||
with patches[0], patches[1], patches[2], patches[3], patches[4]:
|
||
rows = w.runtime_rows("/tmp/sock")
|
||
self.assertEqual(len(rows), 2)
|
||
muse = rows[0]
|
||
self.assertEqual(muse["pane"], "%37")
|
||
self.assertTrue(muse["is_muse"])
|
||
self.assertTrue(muse["auto_approve"])
|
||
self.assertEqual(muse["approval_flags"], ["disable-approval"])
|
||
self.assertEqual(muse["state"], "open-prompt")
|
||
self.assertTrue(muse["watcher_alive"])
|
||
self.assertEqual(muse["watcher_pid"], 1234)
|
||
self.assertEqual(muse["width"], 71)
|
||
self.assertEqual(muse["height"], 27)
|
||
self.assertFalse(muse["squeezed"])
|
||
shell = rows[1]
|
||
self.assertFalse(shell["is_muse"])
|
||
self.assertIsNone(shell["auto_approve"])
|
||
self.assertEqual(shell["state"], "unknown")
|
||
|
||
def test_bare_muse_reports_not_auto(self):
|
||
patches = self._patched(
|
||
captures={"%37": STATE_WORKING, "%38": STATE_SHELL},
|
||
children={2880158: [2881158]},
|
||
cmdlines={2881158: ["/home/super/.local/bin/muse-bin-1.4.3"]})
|
||
with patches[0], patches[1], patches[2], patches[3], patches[4]:
|
||
rows = w.runtime_rows("/tmp/sock")
|
||
self.assertFalse(rows[0]["auto_approve"])
|
||
self.assertEqual(rows[0]["state"], "working")
|
||
self.assertFalse(rows[0]["watcher_alive"])
|
||
|
||
def test_approval_pending_row_carries_kind(self):
|
||
patches = self._patched(captures={"%37": STATE_APPROVAL,
|
||
"%38": STATE_SHELL})
|
||
with patches[0], patches[1], patches[2], patches[3], patches[4]:
|
||
rows = w.runtime_rows("/tmp/sock")
|
||
self.assertEqual(rows[0]["state"], "approval-pending")
|
||
self.assertEqual(rows[0]["prompt_kind"], "muse-approval")
|
||
self.assertEqual(rows[0]["prompt_key"], "1")
|
||
|
||
def test_tmux_failure_returns_empty(self):
|
||
patches = self._patched(tmux_rc=1, tmux_stdout="")
|
||
with patches[0], patches[1], patches[2], patches[3], patches[4]:
|
||
self.assertEqual(w.runtime_rows("/tmp/sock"), [])
|
||
|
||
def test_vanished_pane_skipped(self):
|
||
patches = self._patched(captures={"%37": None, "%38": STATE_SHELL})
|
||
with patches[0], patches[1], patches[2], patches[3], patches[4]:
|
||
rows = w.runtime_rows("/tmp/sock")
|
||
self.assertEqual([r["pane"] for r in rows], ["%38"])
|
||
|
||
def test_pane_state_found_and_missing(self):
|
||
patches = self._patched(captures={"%37": STATE_OPEN,
|
||
"%38": STATE_SHELL})
|
||
with patches[0], patches[1], patches[2], patches[3], patches[4]:
|
||
hit = w.pane_state("/tmp/sock", "%37")
|
||
miss = w.pane_state("/tmp/sock", "%99")
|
||
self.assertEqual(hit["pane"], "%37")
|
||
self.assertEqual(miss["error"], "no_such_pane")
|
||
|
||
def test_rows_flag_squeezed(self):
|
||
listing = ("muse\t1\t%37\tmuse-bin-1.4\t2880158\t35\t7\n"
|
||
"muse\t1\t%38\tbash\t2880200\t35\t7\n")
|
||
patches = self._patched(
|
||
tmux_stdout=listing,
|
||
captures={"%37": STATE_OPEN, "%38": STATE_SHELL})
|
||
with patches[0], patches[1], patches[2], patches[3], patches[4]:
|
||
rows = w.runtime_rows("/tmp/sock")
|
||
self.assertTrue(rows[0]["squeezed"])
|
||
self.assertEqual(rows[0]["width"], 35)
|
||
self.assertEqual(rows[0]["height"], 7)
|
||
self.assertTrue(rows[1]["squeezed"])
|
||
|
||
|
||
class TestNodeFromSession(unittest.TestCase):
|
||
def test_conforming_sessions(self):
|
||
self.assertEqual(w.node_from_session("pip--worker--01"), "pip")
|
||
self.assertEqual(w.node_from_session("muse--repair--09"), "muse")
|
||
self.assertEqual(w.node_from_session("646--watch--a1"), "646")
|
||
self.assertEqual(w.node_from_session("PIP--X--1"), "pip")
|
||
|
||
def test_adhoc_sessions_have_no_node(self):
|
||
for name in ("muse", "repair-09", "swarm-worker", "main",
|
||
"", None):
|
||
self.assertIsNone(w.node_from_session(name), name)
|
||
|
||
def test_rejects_unknown_node_and_single_dash(self):
|
||
self.assertIsNone(w.node_from_session("foo--worker--01"))
|
||
self.assertIsNone(w.node_from_session("pip-worker-01"))
|
||
self.assertIsNone(w.node_from_session("pip--worker"))
|
||
|
||
def test_runtime_rows_carry_node(self):
|
||
listing = "pip--audit--a1\t1\t%37\tbash\t2880158\t80\t24\n"
|
||
r = mock.Mock(returncode=0, stdout=listing, stderr="")
|
||
with mock.patch.object(w, "_tmux", return_value=r), \
|
||
mock.patch.object(w, "capture_pane",
|
||
return_value=STATE_SHELL), \
|
||
mock.patch.object(w, "is_running", return_value=None):
|
||
rows = w.runtime_rows("/tmp/sock")
|
||
self.assertEqual(len(rows), 1)
|
||
self.assertEqual(rows[0]["node"], "pip")
|
||
|
||
|
||
class TestSpreadTargets(unittest.TestCase):
|
||
def _row(self, pane, is_muse, squeezed):
|
||
return {"socket": "/tmp/s", "session": "muse", "window": "1",
|
||
"pane": pane, "is_muse": is_muse, "squeezed": squeezed}
|
||
|
||
def test_selects_squeezed_muse_only(self):
|
||
rows = [self._row("%22", True, False),
|
||
self._row("%23", True, True),
|
||
self._row("%38", False, True)]
|
||
targets = w.spread_targets(rows)
|
||
self.assertEqual([t["pane"] for t in targets], ["%23"])
|
||
|
||
def test_empty_when_nothing_squeezed(self):
|
||
rows = [self._row("%22", True, False)]
|
||
self.assertEqual(w.spread_targets(rows), [])
|
||
|
||
def test_tolerates_missing_keys(self):
|
||
self.assertEqual(w.spread_targets([{"pane": "%1"}]), [])
|
||
|
||
|
||
class TestBoxRuntimeCLI(unittest.TestCase):
|
||
def _box(self, *argv, timeout=60):
|
||
import subprocess
|
||
cmd = [sys.executable, str(BIN_DIR / "super-cli.py"),
|
||
"runtime"] + list(argv)
|
||
return subprocess.run(cmd, capture_output=True, text=True,
|
||
timeout=timeout)
|
||
|
||
def test_list_empty_socket_json(self):
|
||
r = self._box("list", "--socket", "/nonexistent.sock", "--json")
|
||
self.assertEqual(r.returncode, 0, r.stderr[:500])
|
||
data = json.loads(r.stdout)
|
||
self.assertTrue(data["ok"])
|
||
self.assertEqual(data["runtimes"], [])
|
||
|
||
def test_list_muse_only_flag_accepted(self):
|
||
r = self._box("list", "--socket", "/nonexistent.sock", "--json",
|
||
"--muse-only")
|
||
self.assertEqual(r.returncode, 0, r.stderr[:500])
|
||
self.assertTrue(json.loads(r.stdout)["ok"])
|
||
|
||
def test_subcommand_help(self):
|
||
for sub in ("list", "send", "launch", "layout", "spread"):
|
||
r = self._box(sub, "--help")
|
||
self.assertEqual(r.returncode, 0, sub)
|
||
|
||
def test_launch_dry_run_injects_approve(self):
|
||
r = self._box("launch", "--session", "probe-x",
|
||
"--dry-run", "--json")
|
||
self.assertEqual(r.returncode, 0, r.stderr[:500])
|
||
data = json.loads(r.stdout)
|
||
self.assertTrue(data["ok"])
|
||
self.assertTrue(data["dry_run"])
|
||
self.assertEqual(data["injected"], ["--disable-approval"])
|
||
self.assertIn("--disable-approval", data["cmdline"])
|
||
self.assertIn("muse-code", data["cmdline"])
|
||
|
||
def test_launch_dry_run_respects_caller_flags(self):
|
||
r = self._box("launch", "--session", "probe-x",
|
||
"--dry-run", "--json", "--", "--yolo")
|
||
self.assertEqual(r.returncode, 0, r.stderr[:500])
|
||
data = json.loads(r.stdout)
|
||
self.assertEqual(data["injected"], [])
|
||
self.assertIn("--yolo", data["cmdline"])
|
||
self.assertNotIn("--disable-approval", data["cmdline"])
|
||
|
||
def test_send_missing_pane_json(self):
|
||
r = self._box("send", "--socket", "/nonexistent.sock",
|
||
"%99", "hi", "--json")
|
||
self.assertEqual(r.returncode, 0, r.stderr[:500])
|
||
data = json.loads(r.stdout)
|
||
self.assertFalse(data["ok"])
|
||
self.assertEqual(data["error"], "no_such_pane")
|
||
|
||
def test_layout_empty_socket_json(self):
|
||
r = self._box("layout", "--socket", "/nonexistent.sock", "--json")
|
||
self.assertEqual(r.returncode, 0, r.stderr[:500])
|
||
data = json.loads(r.stdout)
|
||
self.assertTrue(data["ok"])
|
||
self.assertEqual(data["runtimes"], [])
|
||
self.assertIn("width", data["minimum"])
|
||
self.assertIn("height", data["minimum"])
|
||
|
||
def test_spread_empty_socket_json(self):
|
||
r = self._box("spread", "--socket", "/nonexistent.sock", "--json")
|
||
self.assertEqual(r.returncode, 0, r.stderr[:500])
|
||
data = json.loads(r.stdout)
|
||
self.assertTrue(data["ok"])
|
||
self.assertEqual(data["spread"], [])
|
||
|
||
def test_spread_dry_run_empty_socket_json(self):
|
||
r = self._box("spread", "--socket", "/nonexistent.sock",
|
||
"--dry-run", "--json")
|
||
self.assertEqual(r.returncode, 0, r.stderr[:500])
|
||
data = json.loads(r.stdout)
|
||
self.assertTrue(data["dry_run"])
|
||
self.assertEqual(data["targets"], [])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|