box/tests/test_box_run.py

"""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()