operator
0065d11e97
- bin/muse_choice_watcher.py + systemd/muse-choices-reconcile.*: automatic choice answering and timer reconciliation - bin/digest.py: fleet log and health summarization - docs/BOX-*-HTTPS.md: comprehensive HTTPS execution contracts and API documentation - docs/MUSE-CHOICES-POLICY.md & docs/SUPERVISION-SPEC.md: autonomous execution specs - tests/test_*.py: unit test suites for HTTPS API, choice watcher, fleet heal, and swarm pruning
193 lines
7.6 KiB
Python
193 lines
7.6 KiB
Python
"""Tests for `box run` and `box watch` (headless muse-code tmux runs) in super-cli.py."""
|
|
import argparse
|
|
import importlib.util
|
|
import io
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
"super_cli_box_run", REPO_ROOT / "bin" / "super-cli.py")
|
|
super_cli = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(super_cli)
|
|
|
|
|
|
def _ns(**over):
|
|
kw = dict(prompt="hello from unit test", prompt_file=None, session="ut-boxrun",
|
|
log=None, model=None, effort=None, provider="echo",
|
|
permission_profile=None, approval_mode="never",
|
|
trust_workspace=False, auto_approve=False)
|
|
kw.update(over)
|
|
return argparse.Namespace(**kw)
|
|
|
|
|
|
class FakeCompleted:
|
|
def __init__(self, returncode=0, stderr=""):
|
|
self.returncode = returncode
|
|
self.stderr = stderr
|
|
|
|
|
|
class BoxRunTests(unittest.TestCase):
|
|
def setUp(self):
|
|
for f in ("/tmp/ut-boxrun.prompt", "/tmp/ut-boxrun.sh", "/tmp/ut-boxrun.exit",
|
|
"/tmp/ut-boxrun-watch.sh", "/tmp/utw-watch.sh"):
|
|
try:
|
|
Path(f).unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
self.addCleanup(self._cleanup)
|
|
|
|
def _cleanup(self):
|
|
for f in ("/tmp/ut-boxrun.prompt", "/tmp/ut-boxrun.sh", "/tmp/ut-boxrun.exit",
|
|
"/tmp/ut-boxrun-watch.sh", "/tmp/utw-watch.sh"):
|
|
try:
|
|
Path(f).unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
for log in ("ut-boxrun.log", "ut-boxrun-watch.log", "utw-watch.log"):
|
|
try:
|
|
Path(super_cli.MUSE_TMUX_LOG_DIR / log).unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
def _which(self, name):
|
|
if name == "tmux":
|
|
return "/usr/bin/tmux"
|
|
if "muse-code" in name:
|
|
return "/home/super/.local/bin/muse-code"
|
|
return None
|
|
|
|
def test_spawn_writes_prompt_wrapper_and_tmux_argv(self):
|
|
calls = []
|
|
|
|
def fake_run(argv, **kw):
|
|
calls.append(argv)
|
|
return FakeCompleted(0)
|
|
|
|
with mock.patch.object(super_cli.shutil, "which", side_effect=self._which), \
|
|
mock.patch.object(super_cli.subprocess, "run", side_effect=fake_run):
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
super_cli.cmd_run(_ns())
|
|
out = buf.getvalue()
|
|
|
|
self.assertIn("session: ut-boxrun", out)
|
|
self.assertIn("attach: tmux -S /tmp/tmux-muse.sock attach -t ut-boxrun", out)
|
|
self.assertEqual(Path("/tmp/ut-boxrun.prompt").read_text(encoding="utf-8"),
|
|
"hello from unit test\n")
|
|
wrapper = Path("/tmp/ut-boxrun.sh").read_text(encoding="utf-8")
|
|
self.assertIn("muse-code", wrapper)
|
|
self.assertIn("--provider echo", wrapper)
|
|
self.assertIn("--prompt-file", wrapper)
|
|
self.assertIn("cd /home/super/Projects/NetVM", wrapper)
|
|
new_session = [c for c in calls if "new-session" in c]
|
|
self.assertEqual(len(new_session), 1)
|
|
self.assertIn("/tmp/tmux-muse.sock", new_session[0])
|
|
self.assertIn("ut-boxrun", new_session[0])
|
|
|
|
def test_optional_flags_passed_through(self):
|
|
calls = []
|
|
|
|
def fake_run(argv, **kw):
|
|
calls.append(argv)
|
|
return FakeCompleted(0)
|
|
|
|
ns = _ns(model="m1", effort="low", permission_profile="prof",
|
|
trust_workspace=True, approval_mode="on-request")
|
|
with mock.patch.object(super_cli.shutil, "which", side_effect=self._which), \
|
|
mock.patch.object(super_cli.subprocess, "run", side_effect=fake_run):
|
|
with redirect_stdout(io.StringIO()):
|
|
super_cli.cmd_run(ns)
|
|
wrapper = Path("/tmp/ut-boxrun.sh").read_text(encoding="utf-8")
|
|
for flag in ("--model m1", "--reasoning-effort low", "--permission-profile prof",
|
|
"--trust-workspace", "--approval-mode on-request"):
|
|
self.assertIn(flag, wrapper)
|
|
|
|
def test_no_prompt_exits_2(self):
|
|
with mock.patch.object(super_cli.shutil, "which", side_effect=self._which), \
|
|
mock.patch.object(super_cli.sys.stdin, "isatty", return_value=True):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
super_cli.cmd_run(_ns(prompt=None))
|
|
self.assertEqual(cm.exception.code, 2)
|
|
|
|
def test_missing_muse_code_exits_2(self):
|
|
with mock.patch.object(super_cli.shutil, "which", return_value=None):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
super_cli.cmd_run(_ns())
|
|
self.assertEqual(cm.exception.code, 2)
|
|
|
|
def test_sanitize_session_name(self):
|
|
self.assertEqual(super_cli._sanitize_tmux_name("run:2026/10/06 05.00"), "run-2026-10-06-05-00")
|
|
self.assertEqual(super_cli._sanitize_tmux_name("!!!"), "run")
|
|
|
|
def test_run_auto_approve_spawns_watcher(self):
|
|
calls = []
|
|
|
|
def fake_run(argv, **kw):
|
|
calls.append(argv)
|
|
return FakeCompleted(0)
|
|
|
|
with mock.patch.object(super_cli.shutil, "which", side_effect=self._which), \
|
|
mock.patch.object(super_cli.subprocess, "run", side_effect=fake_run):
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
super_cli.cmd_run(_ns(auto_approve=True))
|
|
out = buf.getvalue()
|
|
self.assertIn("watcher: ut-boxrun-watch", out)
|
|
self.assertIn("watcher-log:", out)
|
|
new_session = [c for c in calls if "new-session" in c]
|
|
self.assertEqual(len(new_session), 2)
|
|
watch_argv = [c for c in new_session if "ut-boxrun-watch" in c]
|
|
self.assertEqual(len(watch_argv), 1)
|
|
self.assertTrue(Path("/tmp/ut-boxrun-watch.sh").exists())
|
|
|
|
def test_write_watch_script_content(self):
|
|
watch_session, script_file = super_cli._write_watch_script("utw")
|
|
self.assertEqual(watch_session, "utw-watch")
|
|
text = Path(script_file).read_text(encoding="utf-8")
|
|
self.assertIn('target="utw"', text)
|
|
self.assertIn('exit_file="/tmp/utw.exit"', text)
|
|
self.assertIn("capture-pane", text)
|
|
self.assertIn('send-keys -t "$target" "1" Enter', text)
|
|
self.assertIn("max=200", text)
|
|
self.assertIn("has-session", text)
|
|
|
|
def test_watch_missing_session_exits_2(self):
|
|
def fake_run(argv, **kw):
|
|
if "has-session" in argv:
|
|
return FakeCompleted(1)
|
|
return FakeCompleted(0)
|
|
|
|
ns = argparse.Namespace(session="nope-missing")
|
|
with mock.patch.object(super_cli.shutil, "which", side_effect=self._which), \
|
|
mock.patch.object(super_cli.subprocess, "run", side_effect=fake_run):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
super_cli.cmd_watch(ns)
|
|
self.assertEqual(cm.exception.code, 2)
|
|
|
|
def test_watch_spawns_watcher_session(self):
|
|
calls = []
|
|
|
|
def fake_run(argv, **kw):
|
|
calls.append(argv)
|
|
return FakeCompleted(0)
|
|
|
|
ns = argparse.Namespace(session="utw")
|
|
with mock.patch.object(super_cli.shutil, "which", side_effect=self._which), \
|
|
mock.patch.object(super_cli.subprocess, "run", side_effect=fake_run):
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
super_cli.cmd_watch(ns)
|
|
out = buf.getvalue()
|
|
self.assertIn("watching: utw", out)
|
|
self.assertIn("watcher: utw-watch", out)
|
|
new_session = [c for c in calls if "new-session" in c]
|
|
self.assertEqual(len(new_session), 1)
|
|
self.assertIn("utw-watch", new_session[0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|