operator
1652 lines
68 KiB
Python
1652 lines
68 KiB
Python
#!/usr/bin/env python3
|
||
"""test_muse_choice_watcher.py — Focused tests for the Muse A/B/C watcher.
|
||
|
||
Covers: prompt matching (A-default regex), once-per-prompt answering policy,
|
||
socket-namespaced state files (the %0-on-two-sockets collision regression),
|
||
and send-keys command construction.
|
||
"""
|
||
|
||
import sys
|
||
import time
|
||
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_ABC = """Some agent output here.
|
||
How should I proceed?
|
||
A. Apply the fix now
|
||
B. Show a diff first
|
||
C. Skip this file
|
||
Reply with A, B, or C:
|
||
"""
|
||
|
||
PROMPT_AB_WRAPPED = """Long thinking output...
|
||
Which approach?
|
||
A. Switch coverage tuples to a single
|
||
cached registry call with ALL_NODES
|
||
B. Keep per-node calls and retry
|
||
Your choice (A/B)?
|
||
"""
|
||
|
||
PROMPT_STALE = (
|
||
"A. Old option one\nB. Old option two\nPick one (A/B)?\n"
|
||
+ "\n".join("filler line %d" % i for i in range(40))
|
||
)
|
||
|
||
|
||
class TestMatcher(unittest.TestCase):
|
||
def test_matches_abc_with_cue(self):
|
||
m = w.find_choice_prompt(PROMPT_ABC)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual(len(m["options"]), 3)
|
||
self.assertTrue(m["options"][0].startswith("A."))
|
||
# Cue scan hits the question line above the options first; either cue
|
||
# line proves the block was recognized as awaiting a reply.
|
||
self.assertIn(m["cue"], ("How should I proceed?", "Reply with A, B, or C:"))
|
||
self.assertEqual(len(m["sig"]), 16)
|
||
|
||
def test_matches_ab_wrapped(self):
|
||
m = w.find_choice_prompt(PROMPT_AB_WRAPPED)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual(len(m["options"]), 2)
|
||
|
||
def test_rejects_single_option(self):
|
||
self.assertIsNone(w.find_choice_prompt("A. Only one option\nSome text\n"))
|
||
|
||
def test_rejects_lettered_list_without_cue(self):
|
||
text = "A. Apples are red\nB. Bananas are yellow\nJust a grocery list.\n"
|
||
self.assertIsNone(w.find_choice_prompt(text))
|
||
|
||
def test_rejects_stale_scrollback(self):
|
||
self.assertIsNone(w.find_choice_prompt(PROMPT_STALE))
|
||
|
||
def test_matches_prompt_with_trailing_blanks(self):
|
||
# Tall panes pad output with blank lines; a live prompt above the
|
||
# padding must still match (scratch-pane regression).
|
||
text = PROMPT_ABC + "\n" * 30
|
||
m = w.find_choice_prompt(text)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual(len(m["options"]), 3)
|
||
|
||
def test_rejects_out_of_order(self):
|
||
text = "B. Second thing\nA. First thing\nWhich (A/B)?\n"
|
||
self.assertIsNone(w.find_choice_prompt(text))
|
||
|
||
def test_rejects_empty(self):
|
||
self.assertIsNone(w.find_choice_prompt(""))
|
||
self.assertIsNone(w.find_choice_prompt(None))
|
||
|
||
def test_sig_stable_and_sensitive(self):
|
||
a = w.find_choice_prompt(PROMPT_ABC)["sig"]
|
||
b = w.find_choice_prompt(PROMPT_ABC)["sig"]
|
||
self.assertEqual(a, b)
|
||
changed = PROMPT_ABC.replace("Apply the fix now", "Apply the fix later")
|
||
c = w.find_choice_prompt(changed)["sig"]
|
||
self.assertNotEqual(a, c)
|
||
|
||
|
||
PROMPT_YN = """Migrating 12 threads...
|
||
Proceed with the migration? (y/n)
|
||
"""
|
||
|
||
PROMPT_YN_BRACKET = """Target file exists.
|
||
Overwrite existing file? [y/N]
|
||
"""
|
||
|
||
PROMPT_NUMBERED = """Requesting permission for: rm -rf /tmp/x
|
||
(1) Allow once
|
||
(2) Always allow
|
||
Selection:
|
||
"""
|
||
|
||
|
||
class TestPromptKinds(unittest.TestCase):
|
||
def test_letter_kind_and_key(self):
|
||
m = w.find_choice_prompt(PROMPT_ABC)
|
||
self.assertEqual(m["kind"], "letter")
|
||
self.assertEqual(m["key"], "A")
|
||
|
||
def test_yn_paren(self):
|
||
m = w.find_choice_prompt(PROMPT_YN)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual((m["kind"], m["key"]), ("yn", "y"))
|
||
|
||
def test_yn_bracket(self):
|
||
m = w.find_choice_prompt(PROMPT_YN_BRACKET)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual((m["kind"], m["key"]), ("yn", "y"))
|
||
|
||
def test_yn_rejects_mid_line_mention(self):
|
||
self.assertIsNone(w.find_choice_prompt("use y/n for confirmation\nok\n"))
|
||
|
||
def test_yn_rejects_stale(self):
|
||
text = "Proceed? (y/n)\n" + "\n".join("filler %d" % i for i in range(10))
|
||
self.assertIsNone(w.find_choice_prompt(text))
|
||
|
||
def test_numbered_kind_and_key(self):
|
||
m = w.find_choice_prompt(PROMPT_NUMBERED)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual((m["kind"], m["key"]), ("numbered", "1"))
|
||
self.assertEqual(len(m["options"]), 2)
|
||
|
||
def test_numbered_rejects_single(self):
|
||
self.assertIsNone(
|
||
w.find_choice_prompt("(1) Only option\nSelection:\n"))
|
||
|
||
def test_numbered_rejects_without_cue(self):
|
||
self.assertIsNone(
|
||
w.find_choice_prompt("(1) one\n(2) two\nSome other text.\n"))
|
||
|
||
def test_letter_beats_yn_priority(self):
|
||
m = w.find_choice_prompt(PROMPT_ABC + "Proceed? (y/n)\n")
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual(m["kind"], "letter")
|
||
|
||
def test_sigs_namespaced_by_kind(self):
|
||
a = w.find_choice_prompt(PROMPT_ABC)["sig"]
|
||
b = w.find_choice_prompt(PROMPT_YN)["sig"]
|
||
c = w.find_choice_prompt(PROMPT_NUMBERED)["sig"]
|
||
self.assertEqual(len({a, b, c}), 3)
|
||
|
||
|
||
class TestPollOnce(unittest.TestCase):
|
||
def _run(self, captures, dry_run=False, prefill_cap=False):
|
||
state = w.WatcherState()
|
||
if prefill_cap:
|
||
import time
|
||
for i in range(w.MAX_ANSWERS_PER_HOUR):
|
||
state.record_answer("old-%d" % i, time.time())
|
||
log = mock.Mock()
|
||
with mock.patch.object(w, "pane_exists", return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
side_effect=captures) as cap, \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True) as send, \
|
||
mock.patch.object(w, "audit") as audit:
|
||
outcomes = [w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=dry_run)
|
||
for _ in range(len(captures) - 1)]
|
||
return state, log, cap, send, audit, outcomes
|
||
|
||
def _log_msgs(self, log):
|
||
return [c.args[1] for c in log.log.call_args_list]
|
||
|
||
def test_letter_answered_with_A(self):
|
||
state, log, cap, send, audit, outcomes = self._run(
|
||
[PROMPT_ABC, PROMPT_ABC, PROMPT_ABC])
|
||
self.assertEqual(outcomes, ["seen", "answered"])
|
||
send.assert_called_once_with("/tmp/s", "%1", "A", enter=True)
|
||
audit.assert_called_once()
|
||
self.assertIn("prompt seen", self._log_msgs(log))
|
||
self.assertIn("prompt stable, answering", self._log_msgs(log))
|
||
|
||
def test_yn_answered_with_y(self):
|
||
state, log, cap, send, audit, outcomes = self._run(
|
||
[PROMPT_YN, PROMPT_YN, PROMPT_YN])
|
||
self.assertEqual(outcomes[-1], "answered")
|
||
send.assert_called_once_with("/tmp/s", "%1", "y", enter=True)
|
||
self.assertEqual(audit.call_args[0][0], "muse-choice-answered")
|
||
self.assertEqual(audit.call_args[1]["extra"]["key"], "y")
|
||
|
||
def test_numbered_answered_with_1(self):
|
||
state, log, cap, send, audit, outcomes = self._run(
|
||
[PROMPT_NUMBERED, PROMPT_NUMBERED, PROMPT_NUMBERED])
|
||
self.assertEqual(outcomes[-1], "answered")
|
||
send.assert_called_once_with("/tmp/s", "%1", "1", enter=True)
|
||
|
||
def test_dry_run_records_without_sending(self):
|
||
state, log, cap, send, audit, outcomes = self._run(
|
||
[PROMPT_YN, PROMPT_YN, PROMPT_YN], dry_run=True)
|
||
self.assertEqual(outcomes, ["seen", "dry-answered"])
|
||
send.assert_not_called()
|
||
audit.assert_not_called()
|
||
self.assertEqual(len(state.answered_sigs), 1)
|
||
|
||
def test_vanished_prompt_skips_send(self):
|
||
state, log, cap, send, audit, outcomes = self._run(
|
||
[PROMPT_YN, PROMPT_YN, "something else entirely\n"])
|
||
self.assertEqual(outcomes, ["seen", "vanished"])
|
||
send.assert_not_called()
|
||
audit.assert_not_called()
|
||
|
||
def test_capped_logs_once(self):
|
||
state, log, cap, send, audit, outcomes = self._run(
|
||
[PROMPT_YN] * 5, prefill_cap=True)
|
||
self.assertTrue(all(o == "capped" for o in outcomes[1:]))
|
||
send.assert_not_called()
|
||
warns = [c for c in log.log.call_args_list
|
||
if c.args[0] == "warn"]
|
||
self.assertEqual(len(warns), 1)
|
||
|
||
def test_gone_and_capture_failed(self):
|
||
state, log = w.WatcherState(), mock.Mock()
|
||
with mock.patch.object(w, "pane_exists", return_value=False):
|
||
self.assertEqual(
|
||
w._poll_once("/tmp/s", "%1", state, log), "gone")
|
||
with mock.patch.object(w, "pane_exists", return_value=True), \
|
||
mock.patch.object(w, "capture_pane", return_value=None):
|
||
self.assertEqual(
|
||
w._poll_once("/tmp/s", "%1", state, log), "capture-failed")
|
||
|
||
|
||
class TestAnswerPolicy(unittest.TestCase):
|
||
def test_needs_stability(self):
|
||
st = w.WatcherState()
|
||
m = w.find_choice_prompt(PROMPT_ABC)
|
||
now = time.time()
|
||
self.assertEqual(st.observe(m, now), "wait")
|
||
self.assertEqual(st.observe(m, now), "answer")
|
||
|
||
def test_once_per_prompt(self):
|
||
st = w.WatcherState()
|
||
m = w.find_choice_prompt(PROMPT_ABC)
|
||
now = time.time()
|
||
st.observe(m, now)
|
||
st.observe(m, now)
|
||
st.record_answer(m["sig"], now)
|
||
self.assertEqual(st.observe(m, now), "none")
|
||
# A new prompt answers again.
|
||
m2 = w.find_choice_prompt(PROMPT_AB_WRAPPED)
|
||
self.assertNotEqual(m2["sig"], m["sig"])
|
||
self.assertEqual(st.observe(m2, now), "wait")
|
||
self.assertEqual(st.observe(m2, now), "answer")
|
||
|
||
def test_none_resets_pending(self):
|
||
st = w.WatcherState()
|
||
m = w.find_choice_prompt(PROMPT_ABC)
|
||
now = time.time()
|
||
st.observe(m, now)
|
||
self.assertEqual(st.observe(None, now), "none")
|
||
self.assertEqual(st.observe(m, now), "wait")
|
||
|
||
def test_hourly_cap(self):
|
||
st = w.WatcherState()
|
||
now = time.time()
|
||
for i in range(w.MAX_ANSWERS_PER_HOUR):
|
||
st.record_answer("sig-%d" % i, now)
|
||
m = w.find_choice_prompt(PROMPT_ABC)
|
||
st.observe(m, now)
|
||
self.assertEqual(st.observe(m, now), "capped")
|
||
|
||
def test_answered_ttl_allows_recovery(self):
|
||
st = w.WatcherState()
|
||
now = time.time()
|
||
m = w.find_choice_prompt(PROMPT_ABC)
|
||
st.observe(m, now)
|
||
st.observe(m, now)
|
||
st.record_answer(m["sig"], now)
|
||
self.assertEqual(st.observe(m, now + 1), "none")
|
||
# Same prompt still present past TTL => stuck dialog, re-answer.
|
||
self.assertEqual(st.observe(m, now + w.ANSWERED_TTL_SECONDS + 1),
|
||
"wait")
|
||
self.assertEqual(st.observe(m, now + w.ANSWERED_TTL_SECONDS + 2),
|
||
"answer")
|
||
|
||
|
||
class TestNamespacing(unittest.TestCase):
|
||
"""Same pane id on different sockets must never share state files."""
|
||
|
||
def test_pidfile_differs_across_sockets(self):
|
||
a = w.pidfile_for("/tmp/tmux-1000/default", "%0")
|
||
b = w.pidfile_for("/tmp/tmux-1000/lte", "%0")
|
||
self.assertNotEqual(a, b)
|
||
|
||
def test_logfile_differs_across_sockets(self):
|
||
a = w.logfile_for("/tmp/tmux-1000/default", "%0")
|
||
b = w.logfile_for("/tmp/tmux-1000/lte", "%0")
|
||
self.assertNotEqual(a, b)
|
||
|
||
def test_slug_guards_same_basename(self):
|
||
a = w.slug_socket("/tmp/a/default")
|
||
b = w.slug_socket("/tmp/b/default")
|
||
self.assertNotEqual(a, b)
|
||
|
||
|
||
class TestSendAnswer(unittest.TestCase):
|
||
def test_sends_A_then_enter_to_exact_pane(self):
|
||
calls = []
|
||
|
||
def fake_tmux(sock, *args, timeout=5):
|
||
calls.append((sock, args))
|
||
r = mock.Mock()
|
||
r.returncode = 0
|
||
return r
|
||
|
||
with mock.patch.object(w, "_tmux", side_effect=fake_tmux):
|
||
self.assertTrue(w.send_answer("/tmp/tmux-1000/default", "%37"))
|
||
self.assertEqual(len(calls), 2)
|
||
self.assertEqual(calls[0][0], "/tmp/tmux-1000/default")
|
||
self.assertEqual(calls[0][1][:3], ("send-keys", "-t", "%37"))
|
||
self.assertEqual(calls[0][1][3], "A")
|
||
self.assertEqual(calls[1][1][3], "Enter")
|
||
|
||
def test_send_failure_returns_false(self):
|
||
def failing(sock, *args, timeout=5):
|
||
r = mock.Mock()
|
||
r.returncode = 1
|
||
return r
|
||
|
||
with mock.patch.object(w, "_tmux", side_effect=failing):
|
||
self.assertFalse(w.send_answer("/tmp/tmux-1000/default", "%37"))
|
||
|
||
|
||
class TestDesiredState(unittest.TestCase):
|
||
def test_default_is_on(self):
|
||
# Policy: undefined desired state means auto-approve on.
|
||
with mock.patch.object(w, "DESIRED_STATE_FILE", "/nonexistent/x.json"):
|
||
st = w.get_desired()
|
||
self.assertTrue(st["enabled"])
|
||
self.assertFalse(st["dry_run"])
|
||
|
||
def test_missing_key_defaults_on(self):
|
||
import json
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/muse-choices.json"
|
||
with open(path, "w") as f:
|
||
json.dump({"dry_run": False}, f)
|
||
with mock.patch.object(w, "DESIRED_STATE_FILE", path):
|
||
self.assertTrue(w.get_desired()["enabled"])
|
||
|
||
def test_explicit_off_is_respected(self):
|
||
import json
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/muse-choices.json"
|
||
with open(path, "w") as f:
|
||
json.dump({"enabled": False, "dry_run": False}, f)
|
||
with mock.patch.object(w, "DESIRED_STATE_FILE", path):
|
||
self.assertFalse(w.get_desired()["enabled"])
|
||
|
||
def test_set_enabled_roundtrip(self):
|
||
import json
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/muse-choices.json"
|
||
with mock.patch.object(w, "DESIRED_STATE_FILE", path), \
|
||
mock.patch.object(w, "audit") as audit:
|
||
st = w.set_enabled(True, dry_run=True, by="tester")
|
||
self.assertTrue(st["enabled"])
|
||
self.assertTrue(st["dry_run"])
|
||
self.assertEqual(st["updated_by"], "tester")
|
||
self.assertTrue(w.get_desired()["enabled"])
|
||
audit.assert_called_once()
|
||
args, _ = audit.call_args
|
||
self.assertEqual(args[0], "muse-choice-enabled")
|
||
with open(path) as f:
|
||
on_disk = json.load(f)
|
||
self.assertTrue(on_disk["enabled"])
|
||
|
||
|
||
class TestAudit(unittest.TestCase):
|
||
def test_local_fallback_shape(self):
|
||
import json
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/box-ctl.jsonl"
|
||
with mock.patch.object(w, "CTL_LOG", path):
|
||
w._audit_local("muse-choice-answered", "default:%37", "watcher",
|
||
{"sig": "abc"})
|
||
with open(path) as f:
|
||
rec = json.loads(f.read())
|
||
self.assertEqual(rec["action"], "muse-choice-answered")
|
||
self.assertEqual(rec["type"], "muse-choice")
|
||
self.assertEqual(rec["name"], "default:%37")
|
||
self.assertEqual(rec["sig"], "abc")
|
||
self.assertIn("ts", rec)
|
||
|
||
def test_audit_prefers_approvals_module(self):
|
||
import sys
|
||
fake = mock.Mock()
|
||
with mock.patch.dict(sys.modules, {"approvals": fake}):
|
||
w.audit("muse-choice-enabled", caller="box",
|
||
extra={"dry_run": False})
|
||
fake.log_box_ctl.assert_called_once_with(
|
||
"muse-choice-enabled", name=None, caller="box",
|
||
extra={"dry_run": False})
|
||
|
||
def test_audit_never_raises(self):
|
||
import sys
|
||
with mock.patch.dict(sys.modules, {"approvals": None}), \
|
||
mock.patch.object(w, "_audit_local", side_effect=OSError("disk")):
|
||
w.audit("muse-choice-answered") # must not raise
|
||
|
||
|
||
class TestReconcile(unittest.TestCase):
|
||
def _patch_common(self, enabled, dry_run=False):
|
||
return (mock.patch.object(w, "get_desired",
|
||
return_value={"enabled": enabled,
|
||
"dry_run": dry_run}),
|
||
mock.patch.object(w, "muse_panes", return_value=["%37"]),
|
||
mock.patch.object(w, "_start_detached", return_value=True),
|
||
mock.patch.object(w, "status_all", return_value=[]),
|
||
mock.patch.object(w, "stop_all", return_value=[]),
|
||
mock.patch.object(w, "audit"),
|
||
mock.patch.object(w.os.path, "exists", return_value=True))
|
||
|
||
def test_enabled_starts_missing(self):
|
||
patches = self._patch_common(True)
|
||
with patches[0], patches[1], patches[2] as start, patches[3], \
|
||
patches[4], patches[5] as audit, patches[6], \
|
||
mock.patch.object(w, "is_running", return_value=None):
|
||
res = w.reconcile(sockets=["/tmp/sock"])
|
||
start.assert_called_once_with("/tmp/sock", "%37", dry_run=False)
|
||
self.assertEqual(res["started"], ["/tmp/sock:%37"])
|
||
audit.assert_called_once() # changed something -> audited
|
||
|
||
def test_enabled_skips_running_and_stays_quiet(self):
|
||
patches = self._patch_common(True)
|
||
with patches[0], patches[1], patches[2] as start, patches[3], \
|
||
patches[4], patches[5] as audit, patches[6], \
|
||
mock.patch.object(w, "is_running", return_value=1234):
|
||
res = w.reconcile(sockets=["/tmp/sock"])
|
||
start.assert_not_called()
|
||
self.assertEqual(res["already"], ["/tmp/sock:%37"])
|
||
audit.assert_not_called() # no change -> no audit noise
|
||
|
||
def test_disabled_stops_all(self):
|
||
patches = self._patch_common(False)
|
||
with patches[0], patches[1], patches[2] as start, patches[3], \
|
||
mock.patch.object(w, "stop_all",
|
||
return_value=[{"pidfile": "x.pid", "pid": 1}]), \
|
||
patches[5] as audit, patches[6]:
|
||
res = w.reconcile(sockets=["/tmp/sock"])
|
||
start.assert_not_called()
|
||
self.assertFalse(res["enabled"])
|
||
self.assertEqual(len(res["stopped"]), 1)
|
||
audit.assert_called_once()
|
||
|
||
def test_prunes_dead_pidfiles(self):
|
||
patches = self._patch_common(True)
|
||
dead = [{"alive": False, "pidfile": "muse-choice-watcher-x.pid"}]
|
||
with patches[0], patches[1], patches[2], \
|
||
mock.patch.object(w, "status_all", return_value=dead), \
|
||
patches[4], patches[5], patches[6], \
|
||
mock.patch.object(w, "is_running", return_value=999), \
|
||
mock.patch.object(w.os, "remove") as rm:
|
||
res = w.reconcile(sockets=["/tmp/sock"])
|
||
rm.assert_called_once()
|
||
self.assertEqual(res["pruned"], ["muse-choice-watcher-x.pid"])
|
||
|
||
|
||
class TestPidfileClaim(unittest.TestCase):
|
||
def test_claims_missing_file(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/x.pid"
|
||
self.assertTrue(w._claim_pidfile(path))
|
||
with open(path) as f:
|
||
self.assertEqual(f.read().strip(), str(w.os.getpid()))
|
||
|
||
def test_refuses_live_other_watcher(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/x.pid"
|
||
with open(path, "w") as f:
|
||
f.write("99999998")
|
||
with mock.patch.object(w, "_pid_alive", return_value=True), \
|
||
mock.patch.object(w, "_pid_is_watcher", return_value=True):
|
||
self.assertFalse(w._claim_pidfile(path))
|
||
|
||
def test_takes_over_dead_pid(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/x.pid"
|
||
with open(path, "w") as f:
|
||
f.write("99999997")
|
||
with mock.patch.object(w, "_pid_alive", return_value=False):
|
||
self.assertTrue(w._claim_pidfile(path))
|
||
|
||
|
||
class TestAnsweredPersistence(unittest.TestCase):
|
||
"""Answered sigs survive restarts: a daemon that restarts while an
|
||
answered prompt is still visible must not answer it again (live
|
||
double-answer -> stray "1" in the input box)."""
|
||
|
||
def test_restart_suppresses_answered_sig(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/x.answered.json"
|
||
s1 = w.WatcherState(persist_path=path)
|
||
now = time.time()
|
||
s1.record_answer("sig-abc", now)
|
||
s2 = w.WatcherState(persist_path=path)
|
||
self.assertEqual(s2.observe({"sig": "sig-abc"}, now + 5),
|
||
"none")
|
||
|
||
def test_expired_sig_not_loaded(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/x.answered.json"
|
||
s1 = w.WatcherState(persist_path=path)
|
||
s1.record_answer("sig-old",
|
||
time.time() - w.ANSWERED_TTL_SECONDS - 10)
|
||
s2 = w.WatcherState(persist_path=path)
|
||
self.assertNotIn("sig-old", s2.answered_sigs)
|
||
|
||
def test_corrupt_store_loads_empty(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/x.answered.json"
|
||
with open(path, "w") as f:
|
||
f.write("not json{{{")
|
||
s = w.WatcherState(persist_path=path)
|
||
self.assertEqual(s.answered_sigs, {})
|
||
|
||
def test_memory_only_without_path(self):
|
||
s = w.WatcherState()
|
||
s.record_answer("sig-x", time.time())
|
||
self.assertIn("sig-x", s.answered_sigs)
|
||
|
||
|
||
class TestPidfileFlock(unittest.TestCase):
|
||
"""Single daemon per pane, kernel-enforced: concurrent starters
|
||
must not pile up (double answers, '11' in the input box)."""
|
||
|
||
def _child_claim(self, path):
|
||
import subprocess
|
||
code = ("import sys; sys.path.insert(0, %r); "
|
||
"import muse_choice_watcher as w; "
|
||
"sys.exit(0 if w._claim_pidfile(%r) else 3)") % (
|
||
str(BIN_DIR), path)
|
||
return subprocess.run([sys.executable, "-c", code],
|
||
capture_output=True, timeout=30)
|
||
|
||
def test_second_process_refused_while_held(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/x.pid"
|
||
self.assertTrue(w._claim_pidfile(path))
|
||
try:
|
||
r = self._child_claim(path)
|
||
self.assertEqual(r.returncode, 3)
|
||
finally:
|
||
w._release_pidfile(path)
|
||
|
||
def test_claim_succeeds_after_release(self):
|
||
import os
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/x.pid"
|
||
self.assertTrue(w._claim_pidfile(path))
|
||
w._release_pidfile(path)
|
||
self.assertFalse(os.path.exists(path))
|
||
r = self._child_claim(path)
|
||
self.assertEqual(r.returncode, 0)
|
||
|
||
|
||
class TestReconcileFailed(unittest.TestCase):
|
||
def test_failed_starts_recorded(self):
|
||
with mock.patch.object(w, "get_desired",
|
||
return_value={"enabled": True,
|
||
"dry_run": False}), \
|
||
mock.patch.object(w, "muse_panes", return_value=["%37"]), \
|
||
mock.patch.object(w, "is_running", return_value=None), \
|
||
mock.patch.object(w, "_start_detached", return_value=False), \
|
||
mock.patch.object(w, "status_all", return_value=[]), \
|
||
mock.patch.object(w, "audit") as audit, \
|
||
mock.patch.object(w.os.path, "exists", return_value=True):
|
||
res = w.reconcile(sockets=["/tmp/sock"])
|
||
self.assertEqual(res["failed"], ["/tmp/sock:%37"])
|
||
self.assertEqual(res["started"], [])
|
||
audit.assert_called_once()
|
||
|
||
|
||
class TestWatchCommand(unittest.TestCase):
|
||
def test_watch_refused_when_claimed(self):
|
||
with mock.patch.object(w, "_claim_pidfile", return_value=False):
|
||
rc = w.main(["watch", "--socket", "/tmp/s", "--pane", "%1"])
|
||
self.assertEqual(rc, 3)
|
||
|
||
def test_watch_runs_loop_and_releases_own_pidfile(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/x.pid"
|
||
with open(path, "w") as f:
|
||
f.write(str(w.os.getpid()))
|
||
with mock.patch.object(w, "pidfile_for", return_value=path), \
|
||
mock.patch.object(w, "_claim_pidfile", return_value=True), \
|
||
mock.patch.object(w, "watch_loop", return_value=0) as loop:
|
||
rc = w.main(["watch", "--socket", "/tmp/s", "--pane", "%1"])
|
||
self.assertEqual(rc, 0)
|
||
loop.assert_called_once_with("/tmp/s", "%1", dry_run=False)
|
||
self.assertFalse(w.os.path.exists(path))
|
||
|
||
|
||
class TestWatchProcs(unittest.TestCase):
|
||
def _mkproc(self, root, pid, argv):
|
||
import os
|
||
d = os.path.join(root, str(pid))
|
||
os.makedirs(d)
|
||
with open(os.path.join(d, "cmdline"), "wb") as f:
|
||
f.write(b"\0".join(x.encode() for x in argv) + b"\0")
|
||
|
||
def test_exact_argv_scan(self):
|
||
import os
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
self._mkproc(td, 111, ["python3", "/x/muse_choice_watcher.py",
|
||
"watch", "--socket", "/tmp/s",
|
||
"--pane", "%1"])
|
||
self._mkproc(td, 222, ["python3", "bin/super-cli.py",
|
||
"muse-choices", "reconcile"])
|
||
self._mkproc(td, 333, ["python3", "/x/muse_choice_watcher.py",
|
||
"reconcile"])
|
||
os.makedirs(os.path.join(td, "self"))
|
||
procs = w._watch_procs(proc_root=td)
|
||
self.assertEqual(procs, [{"pid": 111, "socket": "/tmp/s",
|
||
"pane": "%1"}])
|
||
|
||
def test_missing_root(self):
|
||
self.assertEqual(w._watch_procs(proc_root="/nonexistent-proc"), [])
|
||
|
||
|
||
class TestWatchDuplicate(unittest.TestCase):
|
||
def test_watch_refuses_duplicate_argv(self):
|
||
dup = [{"pid": 9991, "socket": "/tmp/s", "pane": "%1"}]
|
||
with mock.patch.object(w, "_watch_procs", return_value=dup), \
|
||
mock.patch.object(w, "watch_loop") as loop:
|
||
rc = w.main(["watch", "--socket", "/tmp/s", "--pane", "%1"])
|
||
self.assertEqual(rc, 3)
|
||
loop.assert_not_called()
|
||
|
||
def test_watch_allows_different_pane(self):
|
||
other = [{"pid": 9991, "socket": "/tmp/s", "pane": "%2"}]
|
||
with mock.patch.object(w, "_watch_procs", return_value=other), \
|
||
mock.patch.object(w, "_claim_pidfile", return_value=True), \
|
||
mock.patch.object(w, "watch_loop", return_value=0) as loop:
|
||
rc = w.main(["watch", "--socket", "/tmp/s", "--pane", "%1"])
|
||
self.assertEqual(rc, 0)
|
||
loop.assert_called_once()
|
||
|
||
|
||
class TestStopAllOrphans(unittest.TestCase):
|
||
def test_stop_all_kills_orphans(self):
|
||
import signal
|
||
orphan = {"pid": 8888, "socket": "/tmp/s", "pane": "%9"}
|
||
with mock.patch("os.listdir", return_value=[]), \
|
||
mock.patch.object(w, "_watch_procs", return_value=[orphan]), \
|
||
mock.patch("os.kill") as kill:
|
||
res = w.stop_all()
|
||
kill.assert_called_once_with(8888, signal.SIGTERM)
|
||
self.assertEqual(res[0]["status"], "stopped-orphan")
|
||
self.assertEqual(res[0]["pane"], "%9")
|
||
|
||
|
||
class TestStatusOrphans(unittest.TestCase):
|
||
def test_status_lists_orphans(self):
|
||
orphan = {"pid": 99999999, "socket": "/tmp/sock-x", "pane": "%9"}
|
||
with mock.patch.object(w, "_watch_procs", return_value=[orphan]):
|
||
rows = w.status_all()
|
||
orphans = [r for r in rows if r.get("orphan")]
|
||
self.assertEqual(len(orphans), 1)
|
||
self.assertEqual(orphans[0]["pid"], 99999999)
|
||
|
||
|
||
class TestRecentAnswers(unittest.TestCase):
|
||
def test_filters_and_limits(self):
|
||
import json
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/box-ctl.jsonl"
|
||
with open(path, "w") as f:
|
||
f.write('{"action": "other"}\n')
|
||
f.write('not json\n')
|
||
for i in range(3):
|
||
f.write(json.dumps({"action": "muse-choice-answered",
|
||
"sig": "s%d" % i}) + "\n")
|
||
with mock.patch.object(w, "CTL_LOG", path):
|
||
recs = w.recent_answers(limit=2)
|
||
self.assertEqual([r["sig"] for r in recs], ["s1", "s2"])
|
||
|
||
def test_missing_log_returns_empty(self):
|
||
with mock.patch.object(w, "CTL_LOG", "/nonexistent/x.jsonl"):
|
||
self.assertEqual(w.recent_answers(), [])
|
||
|
||
|
||
class TestDaemonUnits(unittest.TestCase):
|
||
def test_reconcile_unit_lets_daemons_survive(self):
|
||
# Load-bearing line: without KillMode=process, systemd kills
|
||
# timer-spawned watchers when the oneshot service exits.
|
||
text = (REPO_ROOT / "systemd" / "muse-choices-reconcile.service"
|
||
).read_text()
|
||
self.assertIn("KillMode=process", text)
|
||
self.assertIn("muse_choice_watcher.py reconcile", text)
|
||
|
||
def test_reconcile_timer_exists(self):
|
||
text = (REPO_ROOT / "systemd" / "muse-choices-reconcile.timer"
|
||
).read_text()
|
||
self.assertIn("OnUnitActiveSec=", text)
|
||
|
||
|
||
class TestBoxWiring(unittest.TestCase):
|
||
def test_box_status_json_shape(self):
|
||
import json
|
||
import subprocess
|
||
cmd = [sys.executable, str(BIN_DIR / "super-cli.py"),
|
||
"muse-choices", "status", "--json"]
|
||
r = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
|
||
self.assertEqual(r.returncode, 0, r.stderr[:500])
|
||
data = json.loads(r.stdout)
|
||
self.assertTrue(data["ok"])
|
||
self.assertIn("enabled", data["desired"])
|
||
self.assertIsInstance(data["watchers"], list)
|
||
self.assertIsInstance(data["recent_answers"], list)
|
||
|
||
|
||
CURSOR = "›" # Muse TUI menu cursor (U+203A)
|
||
|
||
|
||
PROMPT_MUSE_APPROVAL = (
|
||
"Would you like to run the following\n"
|
||
"\n"
|
||
" $ tmux -S /tmp/tmux-1000/default capture-pane -p -t %39\n"
|
||
"\n"
|
||
+ CURSOR + " 1. Yes, proceed (y)\n"
|
||
" 2. No, and tell Muse Code what to do instead\n"
|
||
)
|
||
|
||
PROMPT_MUSE_APPROVAL_WRAPPED = (
|
||
"Would you like to run the following\n"
|
||
"\n"
|
||
" $ ps -eo pid,etime,args | grep\n"
|
||
" \"[m]use_choice_watcher.py\n"
|
||
" watch\" | wc -l; box\n"
|
||
" muse-choices status\n"
|
||
"\n"
|
||
+ CURSOR + " 1. Yes, proceed (y)\n"
|
||
" 2. No, and tell Muse Code what to\n"
|
||
" do instead\n"
|
||
)
|
||
|
||
PROMPT_MUSE_APPROVAL_DECIDED = (
|
||
PROMPT_MUSE_APPROVAL + "approval decision accepted\n"
|
||
)
|
||
|
||
|
||
class TestMuseApproval(unittest.TestCase):
|
||
"""Native Muse TUI approval menu: Would-you-like + 1.Yes/2.No."""
|
||
|
||
def test_matches_native_approval(self):
|
||
m = w.find_choice_prompt(PROMPT_MUSE_APPROVAL)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual((m["kind"], m["key"]), ("muse-approval", "1"))
|
||
self.assertEqual(m["cue"], "Would you like to run the following")
|
||
self.assertEqual(len(m["options"]), 2)
|
||
|
||
def test_matches_wrapped_command_echo(self):
|
||
m = w.find_choice_prompt(PROMPT_MUSE_APPROVAL_WRAPPED)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual((m["kind"], m["key"]), ("muse-approval", "1"))
|
||
|
||
def test_matches_ascii_cursor(self):
|
||
text = PROMPT_MUSE_APPROVAL.replace(CURSOR, ">")
|
||
m = w.find_choice_prompt(text)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual(m["kind"], "muse-approval")
|
||
|
||
def test_matches_cursor_on_no(self):
|
||
text = PROMPT_MUSE_APPROVAL.replace(CURSOR + " 1.", " 1.")
|
||
text = text.replace(" 2.", CURSOR + " 2.")
|
||
m = w.find_choice_prompt(text)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual((m["kind"], m["key"]), ("muse-approval", "1"))
|
||
|
||
def test_matches_observed_live_variant(self):
|
||
# Verbatim shape answered on a live pane: cue with "command?",
|
||
# option 2 ending "(esc)".
|
||
text = ("Would you like to run the following command?\n"
|
||
"\n"
|
||
" $ tmux capture-pane -p\n"
|
||
"\n"
|
||
+ CURSOR + " 1. Yes, proceed (y)\n"
|
||
" 2. No, and tell Muse Code what to do differently (esc)\n")
|
||
m = w.find_choice_prompt(text)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual((m["kind"], m["key"]), ("muse-approval", "1"))
|
||
|
||
def test_rejects_decided_block(self):
|
||
# An already-landed decision must never be double-answered.
|
||
self.assertIsNone(w.find_choice_prompt(PROMPT_MUSE_APPROVAL_DECIDED))
|
||
|
||
def test_rejects_cue_without_pair(self):
|
||
text = "Would you like to run the following\n\n $ foo\n"
|
||
self.assertIsNone(w.find_choice_prompt(text))
|
||
|
||
def test_rejects_pair_without_cue(self):
|
||
text = (CURSOR + " 1. Yes, proceed (y)\n"
|
||
" 2. No, thanks\n")
|
||
self.assertIsNone(w.find_choice_prompt(text))
|
||
|
||
def test_priority_over_yn(self):
|
||
m = w.find_choice_prompt(PROMPT_MUSE_APPROVAL + "Proceed? (y/n)\n")
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual(m["kind"], "muse-approval")
|
||
|
||
def test_sig_namespaced(self):
|
||
a = w.find_choice_prompt(PROMPT_MUSE_APPROVAL)["sig"]
|
||
b = w.find_choice_prompt(PROMPT_ABC)["sig"]
|
||
self.assertNotEqual(a, b)
|
||
|
||
def test_sig_distinguishes_consecutive_approvals(self):
|
||
# Live stuck-state regression: options+cue are byte-identical
|
||
# across command approvals, so the sig must include the $ command
|
||
# or every dialog after the first is swallowed by once-only.
|
||
other = PROMPT_MUSE_APPROVAL.replace(
|
||
"capture-pane -p -t %39", "capture-pane -p -t %29")
|
||
a = w.find_choice_prompt(PROMPT_MUSE_APPROVAL)["sig"]
|
||
b = w.find_choice_prompt(other)["sig"]
|
||
self.assertNotEqual(a, b)
|
||
|
||
def test_second_approval_answers_after_first(self):
|
||
# End-to-end stuck-state regression through the poll loop.
|
||
other = PROMPT_MUSE_APPROVAL.replace(
|
||
"capture-pane -p -t %39", "capture-pane -p -t %29")
|
||
state = w.WatcherState()
|
||
log = mock.Mock()
|
||
captures = [PROMPT_MUSE_APPROVAL] * 3 + [other] * 3
|
||
with mock.patch.object(w, "pane_exists", return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
side_effect=captures), \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True) as send, \
|
||
mock.patch.object(w, "audit"):
|
||
outcomes = [w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
for _ in range(4)]
|
||
self.assertEqual(outcomes,
|
||
["seen", "answered", "seen", "answered"])
|
||
self.assertEqual(send.call_count, 2)
|
||
|
||
def test_poll_answers_with_1(self):
|
||
state = w.WatcherState()
|
||
log = mock.Mock()
|
||
captures = [PROMPT_MUSE_APPROVAL] * 3
|
||
with mock.patch.object(w, "pane_exists", return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
side_effect=captures), \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True) as send, \
|
||
mock.patch.object(w, "audit") as audit:
|
||
outcomes = [w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
for _ in range(len(captures) - 1)]
|
||
self.assertEqual(outcomes, ["seen", "answered"])
|
||
send.assert_called_once_with("/tmp/s", "%1", "1", enter=True)
|
||
audit.assert_called_once()
|
||
self.assertEqual(audit.call_args[1]["extra"]["kind"],
|
||
"muse-approval")
|
||
|
||
|
||
PROMPT_INTERVIEW = (
|
||
"The daemon only approves today. Should this step add deny/escalate\n"
|
||
"decisions informed by helpers, cover more prompt shapes, or both?\n"
|
||
"\n"
|
||
+ CURSOR + " 1. Deny/escalate policy (Recommended) Keep approving by default.\n"
|
||
" 2. More prompt shapes Teach the matcher more UIs.\n"
|
||
" 3. Both Policy plus broader shapes.\n"
|
||
" 4. None of the above Optionally add notes (tab).\n"
|
||
)
|
||
|
||
|
||
class TestInterview(unittest.TestCase):
|
||
"""Agent interview UI: cursor + ordered 1./2. menu + question."""
|
||
|
||
def test_matches_interview(self):
|
||
m = w.find_choice_prompt(PROMPT_INTERVIEW)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual((m["kind"], m["key"]), ("interview", "1"))
|
||
self.assertEqual(len(m["options"]), 4)
|
||
|
||
def test_matches_ascii_cursor(self):
|
||
m = w.find_choice_prompt(PROMPT_INTERVIEW.replace(CURSOR, ">"))
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual(m["kind"], "interview")
|
||
|
||
def test_rejects_prose_list_without_cursor(self):
|
||
# Same shape minus the selection cursor is prose, not a live menu.
|
||
text = PROMPT_INTERVIEW.replace(CURSOR + " ", " ")
|
||
self.assertIsNone(w.find_choice_prompt(text))
|
||
|
||
def test_rejects_single_option(self):
|
||
text = ("Pick one?\n\n" + CURSOR + " 1. Only choice\n")
|
||
self.assertIsNone(w.find_choice_prompt(text))
|
||
|
||
def test_rejects_options_without_question(self):
|
||
text = ("Some statement here.\n\n"
|
||
+ CURSOR + " 1. First\n"
|
||
" 2. Second\n")
|
||
self.assertIsNone(w.find_choice_prompt(text))
|
||
|
||
def test_approval_wins_over_interview(self):
|
||
# A native approval dialog also carries dotted options; the more
|
||
# specific kind must win.
|
||
m = w.find_choice_prompt(PROMPT_MUSE_APPROVAL)
|
||
self.assertEqual(m["kind"], "muse-approval")
|
||
|
||
def test_poll_answers_with_1(self):
|
||
state = w.WatcherState()
|
||
log = mock.Mock()
|
||
captures = [PROMPT_INTERVIEW] * 3
|
||
with mock.patch.object(w, "pane_exists", return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
side_effect=captures), \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True) as send, \
|
||
mock.patch.object(w, "audit") as audit:
|
||
outcomes = [w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
for _ in range(len(captures) - 1)]
|
||
self.assertEqual(outcomes, ["seen", "answered"])
|
||
send.assert_called_once_with("/tmp/s", "%1", "1", enter=True)
|
||
self.assertEqual(audit.call_args[1]["extra"]["kind"], "interview")
|
||
|
||
|
||
class TestLaunchOptOut(unittest.TestCase):
|
||
def test_bare_argv_answers(self):
|
||
self.assertFalse(w.launch_opt_out(["/x/muse-bin-1.4"]))
|
||
self.assertFalse(w.launch_opt_out([]))
|
||
self.assertFalse(w.launch_opt_out(None))
|
||
|
||
def test_auto_flags_answer(self):
|
||
self.assertFalse(w.launch_opt_out(["muse", "--yolo"]))
|
||
self.assertFalse(w.launch_opt_out(["muse", "--disable-approval"]))
|
||
self.assertFalse(
|
||
w.launch_opt_out(["muse", "--approval-mode", "never"]))
|
||
self.assertFalse(
|
||
w.launch_opt_out(["muse", "--approval-mode=never"]))
|
||
|
||
def test_explicit_mode_holds(self):
|
||
self.assertTrue(
|
||
w.launch_opt_out(["muse", "--approval-mode", "on-request"]))
|
||
self.assertTrue(
|
||
w.launch_opt_out(["muse", "--approval-mode", "untrusted"]))
|
||
self.assertTrue(
|
||
w.launch_opt_out(["muse", "--approval-mode=on-request"]))
|
||
|
||
def test_poll_holds_opt_out_pane(self):
|
||
state = w.WatcherState()
|
||
log = mock.Mock()
|
||
captures = [PROMPT_YN] * 4
|
||
with mock.patch.object(w, "pane_exists", return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
side_effect=captures), \
|
||
mock.patch.object(w, "pane_muse_argv",
|
||
return_value=["muse", "--approval-mode",
|
||
"on-request"]), \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True) as send, \
|
||
mock.patch.object(w, "audit") as audit:
|
||
outcomes = [w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
for _ in range(len(captures) - 1)]
|
||
self.assertEqual(outcomes, ["seen", "held", "none"])
|
||
send.assert_not_called()
|
||
audit.assert_not_called()
|
||
msgs = [c.args[1] for c in log.log.call_args_list]
|
||
self.assertIn("held: pane opted out via launch flags", msgs)
|
||
|
||
def test_poll_answers_bare_pane(self):
|
||
state = w.WatcherState()
|
||
log = mock.Mock()
|
||
captures = [PROMPT_YN] * 3
|
||
with mock.patch.object(w, "pane_exists", return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
side_effect=captures), \
|
||
mock.patch.object(w, "pane_muse_argv", return_value=[]), \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True) as send, \
|
||
mock.patch.object(w, "audit"):
|
||
outcomes = [w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
for _ in range(len(captures) - 1)]
|
||
self.assertEqual(outcomes, ["seen", "answered"])
|
||
send.assert_called_once_with("/tmp/s", "%1", "y", enter=True)
|
||
|
||
def test_pane_muse_argv(self):
|
||
listing = mock.Mock(returncode=0, stdout="%1 1000\n%2 2000\n",
|
||
stderr="")
|
||
|
||
def fake_cmdline(pid):
|
||
if pid == 1001:
|
||
return ["/x/muse-bin", "--yolo"]
|
||
return ["/bin/bash"]
|
||
|
||
with mock.patch.object(w, "_tmux",
|
||
return_value=listing) as t, \
|
||
mock.patch.object(w, "_child_pids", return_value=[1001]), \
|
||
mock.patch.object(w, "_cmdline", side_effect=fake_cmdline):
|
||
argv = w.pane_muse_argv("/tmp/s", "%1")
|
||
self.assertEqual(argv, ["/x/muse-bin", "--yolo"])
|
||
t.assert_called_once()
|
||
|
||
def test_pane_muse_argv_missing(self):
|
||
listing = mock.Mock(returncode=0, stdout="%2 2000\n", stderr="")
|
||
with mock.patch.object(w, "_tmux", return_value=listing):
|
||
self.assertEqual(w.pane_muse_argv("/tmp/s", "%1"), [])
|
||
|
||
|
||
PROMPT_COLLAPSED_APPROVAL = (
|
||
"Would you like to run the following\n"
|
||
"\n"
|
||
" $ python3 -m unittest\n"
|
||
" tests.test_muse_choice_watcher\n"
|
||
" 2>&1 | tail -n 3 && cp\n"
|
||
" ʼ 4 command rows omitted\n"
|
||
" ctrl+o view full command\n"
|
||
)
|
||
|
||
|
||
class TestCollapsedApproval(unittest.TestCase):
|
||
"""Collapsed approval: long command hides the 1/2 pair; ctrl+o expands."""
|
||
|
||
def test_matches_collapsed(self):
|
||
m = w.find_choice_prompt(PROMPT_COLLAPSED_APPROVAL)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual(m["kind"], "muse-approval-collapsed")
|
||
self.assertEqual(m["key"], "Enter")
|
||
self.assertFalse(m["enter"])
|
||
|
||
def test_expanded_pair_beats_collapsed(self):
|
||
m = w.find_choice_prompt(PROMPT_MUSE_APPROVAL)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual(m["kind"], "muse-approval")
|
||
|
||
def test_rejects_cue_without_markers(self):
|
||
text = "Would you like to run the following\n\n $ foo\n"
|
||
self.assertIsNone(w.find_choice_prompt(text))
|
||
|
||
def test_rejects_markers_without_cue(self):
|
||
text = (" $ foo bar baz\n"
|
||
" 4 command rows omitted\n"
|
||
" ctrl+o view full command\n")
|
||
self.assertIsNone(w.find_choice_prompt(text))
|
||
|
||
def test_rejects_decided_collapsed(self):
|
||
text = PROMPT_COLLAPSED_APPROVAL + "approval decision accepted\n"
|
||
self.assertIsNone(w.find_choice_prompt(text))
|
||
|
||
def test_poll_sends_bare_enter(self):
|
||
state = w.WatcherState()
|
||
log = mock.Mock()
|
||
captures = [PROMPT_COLLAPSED_APPROVAL] * 3
|
||
calls = []
|
||
|
||
def fake_tmux(sock, *args, timeout=5):
|
||
calls.append((sock, args))
|
||
r = mock.Mock()
|
||
r.returncode = 0
|
||
return r
|
||
|
||
with mock.patch.object(w, "pane_exists", return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
side_effect=captures), \
|
||
mock.patch.object(w, "pane_muse_argv", return_value=[]), \
|
||
mock.patch.object(w, "_tmux", side_effect=fake_tmux), \
|
||
mock.patch.object(w, "audit") as audit:
|
||
outcomes = [w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
for _ in range(len(captures) - 1)]
|
||
self.assertEqual(outcomes, ["seen", "answered"])
|
||
self.assertEqual(len(calls), 1)
|
||
self.assertEqual(calls[0][1][:3], ("send-keys", "-t", "%1"))
|
||
self.assertEqual(calls[0][1][3], "Enter")
|
||
audit.assert_called_once()
|
||
self.assertEqual(audit.call_args[1]["extra"]["kind"],
|
||
"muse-approval-collapsed")
|
||
|
||
|
||
PROMPT_EXPLICIT = (
|
||
"Reply ACCEPT to approve this text as written (the 2 minutes\n"
|
||
"included), or amend anything first. Note: accepting the scope\n"
|
||
"finishes the interview and flips the record to Final.\n"
|
||
)
|
||
|
||
|
||
class TestExplicitPhrase(unittest.TestCase):
|
||
"""Model asks the user to reply an explicit magic word."""
|
||
|
||
def test_matches_accept(self):
|
||
m = w.find_choice_prompt(PROMPT_EXPLICIT)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual((m["kind"], m["key"]),
|
||
("explicit-phrase", "ACCEPT"))
|
||
|
||
def test_matches_other_tokens(self):
|
||
for token in ("YES", "GO", "OK", "CONTINUE", "PROCEED", "ABORT-1"):
|
||
with self.subTest(token=token):
|
||
m = w.find_choice_prompt("Reply %s to confirm.\n" % token)
|
||
self.assertIsNotNone(m)
|
||
self.assertEqual(m["key"], token)
|
||
|
||
def test_rejects_lowercase_prose(self):
|
||
self.assertIsNone(
|
||
w.find_choice_prompt("Please reply soon to confirm.\n"))
|
||
self.assertIsNone(
|
||
w.find_choice_prompt("Reply yes please to continue.\n"))
|
||
|
||
def test_rejects_stale(self):
|
||
text = ("Reply ACCEPT to approve.\n"
|
||
+ "\n".join("filler %d" % i for i in range(12)))
|
||
self.assertIsNone(w.find_choice_prompt(text))
|
||
|
||
def test_freshest_wins(self):
|
||
text = ("Reply YES to confirm.\n"
|
||
"Some agent chatter.\n"
|
||
"Reply ACCEPT to approve this text as written.\n")
|
||
m = w.find_choice_prompt(text)
|
||
self.assertEqual(m["key"], "ACCEPT")
|
||
|
||
def test_poll_answers_with_token(self):
|
||
state = w.WatcherState()
|
||
log = mock.Mock()
|
||
captures = [PROMPT_EXPLICIT] * 3
|
||
with mock.patch.object(w, "pane_exists", return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
side_effect=captures), \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True) as send, \
|
||
mock.patch.object(w, "audit") as audit:
|
||
outcomes = [w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
for _ in range(len(captures) - 1)]
|
||
self.assertEqual(outcomes, ["seen", "answered"])
|
||
send.assert_called_once_with("/tmp/s", "%1", "ACCEPT", enter=True)
|
||
self.assertEqual(audit.call_args[1]["extra"]["kind"],
|
||
"explicit-phrase")
|
||
|
||
|
||
class TestRulesEval(unittest.TestCase):
|
||
def _use_rules(self, tmpdir, data):
|
||
import json
|
||
path = tmpdir + "/rules.json"
|
||
if isinstance(data, str):
|
||
with open(path, "w") as f:
|
||
f.write(data)
|
||
else:
|
||
with open(path, "w") as f:
|
||
json.dump(data, f)
|
||
w._RULES_CACHE = {"key": None, "rules": []}
|
||
return mock.patch.object(w, "RULES_FILE", path)
|
||
|
||
def _rm_approval(self):
|
||
return PROMPT_MUSE_APPROVAL.replace(
|
||
"tmux -S /tmp/tmux-1000/default capture-pane -p -t %39",
|
||
"rm -rf /tmp/scratch")
|
||
|
||
def test_no_file_approves(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td, \
|
||
mock.patch.object(w, "RULES_FILE",
|
||
td + "/missing.json"):
|
||
w._RULES_CACHE = {"key": None, "rules": []}
|
||
d, r = w.evaluate_rules(
|
||
w.find_choice_prompt(PROMPT_ABC))
|
||
self.assertEqual((d, r), ("approve", None))
|
||
|
||
def test_bad_json_fails_open(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with self._use_rules(td, "{not json"):
|
||
d, r = w.evaluate_rules(
|
||
w.find_choice_prompt(PROMPT_ABC), mock.Mock())
|
||
self.assertEqual((d, r), ("approve", None))
|
||
|
||
def test_token_match_holds(self):
|
||
import tempfile
|
||
rules = {"rules": [{"id": "t", "kind": "explicit-phrase",
|
||
"token": ["ABORT", "DELETE"],
|
||
"decision": "hold"}]}
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with self._use_rules(td, rules):
|
||
m = w.find_choice_prompt("Reply ABORT to cancel.\n")
|
||
d, r = w.evaluate_rules(m)
|
||
self.assertEqual(d, "hold")
|
||
self.assertEqual(r["id"], "t")
|
||
|
||
def test_token_nonmatch_approves(self):
|
||
import tempfile
|
||
rules = {"rules": [{"id": "t", "kind": "explicit-phrase",
|
||
"token": ["ABORT"], "decision": "hold"}]}
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with self._use_rules(td, rules):
|
||
m = w.find_choice_prompt("Reply ACCEPT to approve.\n")
|
||
d, r = w.evaluate_rules(m)
|
||
self.assertEqual((d, r), ("approve", None))
|
||
|
||
def test_command_match_holds(self):
|
||
import tempfile
|
||
rules = {"rules": [{"id": "c", "kind": "muse-approval",
|
||
"command": r"\brm\s+-rf?\b",
|
||
"decision": "hold"}]}
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with self._use_rules(td, rules):
|
||
m = w.find_choice_prompt(self._rm_approval())
|
||
self.assertIsNotNone(m)
|
||
d, r = w.evaluate_rules(m)
|
||
self.assertEqual(d, "hold")
|
||
|
||
def test_kind_mismatch_skips(self):
|
||
import tempfile
|
||
rules = {"rules": [{"id": "c", "kind": "yn",
|
||
"command": r"\brm\s+-rf?\b",
|
||
"decision": "hold"}]}
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with self._use_rules(td, rules):
|
||
m = w.find_choice_prompt(self._rm_approval())
|
||
d, r = w.evaluate_rules(m)
|
||
self.assertEqual((d, r), ("approve", None))
|
||
|
||
def test_deny_question_skipped(self):
|
||
import tempfile
|
||
rules = {"rules": [{"id": "d", "kind": "explicit-phrase",
|
||
"token": ["ABORT"], "decision": "deny"}]}
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with self._use_rules(td, rules):
|
||
m = w.find_choice_prompt("Reply ABORT to cancel.\n")
|
||
d, r = w.evaluate_rules(m)
|
||
self.assertEqual((d, r), ("approve", None))
|
||
|
||
def test_deny_approval(self):
|
||
import tempfile
|
||
rules = {"rules": [{"id": "d", "kind": "muse-approval",
|
||
"decision": "deny"}]}
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with self._use_rules(td, rules):
|
||
m = w.find_choice_prompt(PROMPT_MUSE_APPROVAL)
|
||
d, r = w.evaluate_rules(m)
|
||
self.assertEqual(d, "deny")
|
||
|
||
def test_deny_collapsed_downgrades_to_hold(self):
|
||
import tempfile
|
||
rules = {"rules": [{"id": "d",
|
||
"kind": "muse-approval-collapsed",
|
||
"decision": "deny"}]}
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with self._use_rules(td, rules):
|
||
m = w.find_choice_prompt(PROMPT_COLLAPSED_APPROVAL)
|
||
d, r = w.evaluate_rules(m)
|
||
self.assertEqual(d, "hold")
|
||
self.assertEqual(r["downgraded_from"], "deny")
|
||
|
||
def test_unknown_decision_approves(self):
|
||
import tempfile
|
||
rules = {"rules": [{"id": "x", "decision": "explode"}]}
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with self._use_rules(td, rules):
|
||
d, r = w.evaluate_rules(
|
||
w.find_choice_prompt(PROMPT_ABC))
|
||
self.assertEqual(d, "approve")
|
||
|
||
def test_bad_regex_never_matches(self):
|
||
import tempfile
|
||
rules = {"rules": [{"id": "x", "command": "[invalid",
|
||
"decision": "hold"}]}
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with self._use_rules(td, rules):
|
||
d, r = w.evaluate_rules(
|
||
w.find_choice_prompt(PROMPT_MUSE_APPROVAL))
|
||
self.assertEqual((d, r), ("approve", None))
|
||
|
||
|
||
class TestHoldFlow(unittest.TestCase):
|
||
def _polls(self, tmpdir, text, n, hold_rule=True):
|
||
import time
|
||
state = w.WatcherState()
|
||
log = mock.Mock()
|
||
rule = {"id": "t", "reason": "test hold"}
|
||
eff = ("hold", rule) if hold_rule else ("approve", None)
|
||
with mock.patch.object(w, "STATE_DIR", tmpdir), \
|
||
mock.patch.object(w, "pane_exists", return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
side_effect=[text] * (2 * n)), \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True) as send, \
|
||
mock.patch.object(w, "audit") as audit, \
|
||
mock.patch.object(w, "evaluate_rules",
|
||
return_value=eff):
|
||
outcomes = [w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
for _ in range(n)]
|
||
return state, log, send, audit, outcomes
|
||
|
||
def test_hold_suppresses_and_writes_file(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
state, log, send, audit, outcomes = self._polls(
|
||
td, PROMPT_YN, 2)
|
||
self.assertEqual(outcomes, ["seen", "held"])
|
||
send.assert_not_called()
|
||
held_calls = [c for c in audit.call_args_list
|
||
if c.args[0] == "muse-choice-held"]
|
||
self.assertEqual(len(held_calls), 1)
|
||
|
||
def test_held_persists_without_reaudit(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
state, log, send, audit, outcomes = self._polls(
|
||
td, PROMPT_YN, 4)
|
||
self.assertEqual(outcomes,
|
||
["seen", "held", "held", "held"])
|
||
send.assert_not_called()
|
||
held_calls = [c for c in audit.call_args_list
|
||
if c.args[0] == "muse-choice-held"]
|
||
self.assertEqual(len(held_calls), 1)
|
||
|
||
def test_resolve_approve_releases(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with mock.patch.object(w, "STATE_DIR", td):
|
||
state, log, send, audit, outcomes = self._polls(
|
||
td, PROMPT_YN, 2)
|
||
hf = w.read_hold("/tmp/s", "%1")
|
||
self.assertIsNotNone(hf)
|
||
hf["directive"] = "approve" # what box resolve does
|
||
w.write_hold("/tmp/s", "%1", hf)
|
||
with mock.patch.object(w, "pane_exists",
|
||
return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
return_value=PROMPT_YN), \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True) as send2, \
|
||
mock.patch.object(w, "audit") as audit2, \
|
||
mock.patch.object(w, "evaluate_rules") as ev:
|
||
out = w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
self.assertEqual(out, "answered")
|
||
send2.assert_called_once_with("/tmp/s", "%1", "y", enter=True)
|
||
ev.assert_not_called() # release bypasses re-evaluation
|
||
|
||
def test_resolve_deny_sends_negative(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with mock.patch.object(w, "STATE_DIR", td):
|
||
state, log, send, audit, outcomes = self._polls(
|
||
td, PROMPT_YN, 2)
|
||
hf = w.read_hold("/tmp/s", "%1")
|
||
hf["directive"] = "deny"
|
||
w.write_hold("/tmp/s", "%1", hf)
|
||
with mock.patch.object(w, "pane_exists",
|
||
return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
return_value=PROMPT_YN), \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True) as send2, \
|
||
mock.patch.object(w, "audit"):
|
||
out = w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
self.assertEqual(out, "denied")
|
||
send2.assert_called_once_with("/tmp/s", "%1", "n", enter=True)
|
||
|
||
def test_resolve_deny_refused_on_questions(self):
|
||
import tempfile
|
||
text = "Reply ABORT to cancel.\n"
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with mock.patch.object(w, "STATE_DIR", td):
|
||
state, log, send, audit, outcomes = self._polls(
|
||
td, text, 2)
|
||
hf = w.read_hold("/tmp/s", "%1")
|
||
hf["directive"] = "deny"
|
||
w.write_hold("/tmp/s", "%1", hf)
|
||
with mock.patch.object(w, "pane_exists",
|
||
return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
return_value=text), \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True) as send2, \
|
||
mock.patch.object(w, "audit"):
|
||
out = w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
self.assertIsNone(w.read_hold("/tmp/s", "%1")
|
||
["directive"])
|
||
self.assertEqual(out, "held")
|
||
send2.assert_not_called()
|
||
|
||
def test_expiry_releases_to_approve(self):
|
||
import tempfile
|
||
import time
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with mock.patch.object(w, "STATE_DIR", td):
|
||
state, log, send, audit, outcomes = self._polls(
|
||
td, PROMPT_YN, 2)
|
||
hf = w.read_hold("/tmp/s", "%1")
|
||
hf["held_until"] = time.time() - 1
|
||
w.write_hold("/tmp/s", "%1", hf)
|
||
with mock.patch.object(w, "pane_exists",
|
||
return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
return_value=PROMPT_YN), \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True) as send2, \
|
||
mock.patch.object(w, "audit") as audit2, \
|
||
mock.patch.object(w, "evaluate_rules") as ev:
|
||
out = w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
self.assertEqual(out, "answered")
|
||
send2.assert_called_once()
|
||
ev.assert_not_called()
|
||
expired = [c for c in audit2.call_args_list
|
||
if c.args[0] == "muse-choice-hold-expired"]
|
||
self.assertEqual(len(expired), 1)
|
||
|
||
def test_stale_hold_cleared(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with mock.patch.object(w, "STATE_DIR", td):
|
||
state, log, send, audit, outcomes = self._polls(
|
||
td, PROMPT_YN, 2)
|
||
self.assertIsNotNone(w.read_hold("/tmp/s", "%1"))
|
||
with mock.patch.object(w, "pane_exists",
|
||
return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
return_value=PROMPT_ABC), \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True), \
|
||
mock.patch.object(w, "audit"), \
|
||
mock.patch.object(w, "evaluate_rules",
|
||
return_value=("approve", None)):
|
||
out = w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
self.assertIsNone(w.read_hold("/tmp/s", "%1"))
|
||
self.assertEqual(out, "seen")
|
||
|
||
|
||
class TestBoxResolve(unittest.TestCase):
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
import importlib.util
|
||
spec = importlib.util.spec_from_file_location(
|
||
"supercli_test", str(BIN_DIR / "super-cli.py"))
|
||
cls.cli = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(cls.cli)
|
||
|
||
def _ns(self, decision):
|
||
import argparse
|
||
return argparse.Namespace(mc_action="resolve", socket="/tmp/s",
|
||
pane="%1", decision=decision, json=True)
|
||
|
||
def _hold(self, tmpdir, kind="yn"):
|
||
import time
|
||
with mock.patch.object(w, "STATE_DIR", tmpdir):
|
||
w.write_hold("/tmp/s", "%1",
|
||
{"sig": "s1", "kind": kind, "key": "y",
|
||
"text": "t", "rule": "r",
|
||
"held_until": time.time() + 60,
|
||
"directive": None})
|
||
|
||
def _run(self, tmpdir, decision):
|
||
import io
|
||
import json
|
||
from contextlib import redirect_stdout
|
||
buf = io.StringIO()
|
||
with mock.patch.object(w, "STATE_DIR", tmpdir), \
|
||
mock.patch.object(w, "audit"), \
|
||
redirect_stdout(buf):
|
||
self.cli.cmd_muse_choices(self._ns(decision))
|
||
return json.loads(buf.getvalue())
|
||
|
||
def test_resolve_approve_sets_directive(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
self._hold(td)
|
||
data = self._run(td, "approve")
|
||
self.assertTrue(data["ok"])
|
||
with mock.patch.object(w, "STATE_DIR", td):
|
||
hf = w.read_hold("/tmp/s", "%1")
|
||
self.assertEqual(hf["directive"], "approve")
|
||
|
||
def test_resolve_deny_sets_directive(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
self._hold(td, kind="yn")
|
||
data = self._run(td, "deny")
|
||
self.assertTrue(data["ok"])
|
||
with mock.patch.object(w, "STATE_DIR", td):
|
||
hf = w.read_hold("/tmp/s", "%1")
|
||
self.assertEqual(hf["directive"], "deny")
|
||
|
||
def test_resolve_deny_refused_on_questions(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
self._hold(td, kind="interview")
|
||
data = self._run(td, "deny")
|
||
self.assertFalse(data["ok"])
|
||
self.assertIn("D2", data["reason"])
|
||
|
||
def test_resolve_nothing_held(self):
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
data = self._run(td, "approve")
|
||
self.assertTrue(data["ok"])
|
||
self.assertIsNone(data["held"])
|
||
|
||
|
||
class TestPeerAnsweredReload(unittest.TestCase):
|
||
"""A peer watcher answering first must suppress this watcher.
|
||
|
||
Concurrent duplicates share only the on-disk answered store: a
|
||
second watcher that loaded before the peer's answer must re-check
|
||
the store before typing, or both type '1' ('11' in the input box).
|
||
"""
|
||
|
||
def test_disk_answer_suppresses_send(self):
|
||
import json
|
||
import tempfile
|
||
import time
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = td + "/x.answered.json"
|
||
state = w.WatcherState(persist_path=path)
|
||
log = mock.Mock()
|
||
sig = w.find_choice_prompt(PROMPT_YN)["sig"]
|
||
with mock.patch.object(w, "pane_exists", return_value=True), \
|
||
mock.patch.object(w, "capture_pane",
|
||
side_effect=[PROMPT_YN] * 4), \
|
||
mock.patch.object(w, "send_answer",
|
||
return_value=True) as send, \
|
||
mock.patch.object(w, "audit"):
|
||
out1 = w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
self.assertEqual(out1, "seen")
|
||
# Peer answers the same prompt and persists first.
|
||
with open(path, "w") as f:
|
||
json.dump({sig: time.time()}, f)
|
||
out2 = w._poll_once("/tmp/s", "%1", state, log,
|
||
dry_run=False)
|
||
send.assert_not_called()
|
||
self.assertIn(out2, ("duplicate-suppressed", "none"))
|
||
self.assertIn(sig, state.answered_sigs)
|
||
|
||
|
||
class TestCaptureJoinWrapped(unittest.TestCase):
|
||
def test_capture_joins_wrapped_lines(self):
|
||
with mock.patch.object(w, "_tmux") as t:
|
||
t.return_value = mock.Mock(returncode=0, stdout="ok")
|
||
w.capture_pane("/tmp/s", "%1")
|
||
args = t.call_args[0]
|
||
self.assertIn("-J", args)
|
||
|
||
|
||
class TestLogPosture(unittest.TestCase):
|
||
"""Watcher start records the pane's permission posture.
|
||
|
||
The watcher answers with per-choice logging in every mode; bypass
|
||
postures (yolo / approval disabled) additionally audit, since the
|
||
session then acts outside the choice trail.
|
||
"""
|
||
|
||
def test_bypass_audits(self):
|
||
log = mock.Mock()
|
||
with mock.patch.object(w, "pane_muse_argv",
|
||
return_value=["muse", "--yolo"]), \
|
||
mock.patch.object(w, "audit") as audit:
|
||
w._log_posture("/tmp/s", "%1", log)
|
||
log.log.assert_called_once()
|
||
self.assertEqual(log.log.call_args[0][1], "pane posture")
|
||
self.assertEqual(log.log.call_args[1]["mode"], "yolo")
|
||
audit.assert_called_once()
|
||
self.assertEqual(audit.call_args[0][0], "muse-choice-posture")
|
||
self.assertTrue(audit.call_args[1]["extra"]["bypass"])
|
||
|
||
def test_default_logs_without_audit(self):
|
||
log = mock.Mock()
|
||
with mock.patch.object(w, "pane_muse_argv",
|
||
return_value=["muse"]), \
|
||
mock.patch.object(w, "audit") as audit:
|
||
w._log_posture("/tmp/s", "%1", log)
|
||
log.log.assert_called_once()
|
||
audit.assert_not_called()
|
||
|
||
def test_nondefault_profile_audits(self):
|
||
log = mock.Mock()
|
||
with mock.patch.object(w, "pane_muse_argv",
|
||
return_value=["muse", "--permission-profile",
|
||
":unrestricted"]), \
|
||
mock.patch.object(w, "audit") as audit:
|
||
w._log_posture("/tmp/s", "%1", log)
|
||
audit.assert_called_once()
|
||
extra = audit.call_args[1]["extra"]
|
||
self.assertEqual(extra["mode"], ":unrestricted")
|
||
self.assertFalse(extra["bypass"])
|
||
|
||
def test_argv_failure_never_raises(self):
|
||
log = mock.Mock()
|
||
with mock.patch.object(w, "pane_muse_argv",
|
||
side_effect=RuntimeError("tmux")), \
|
||
mock.patch.object(w, "audit") as audit:
|
||
w._log_posture("/tmp/s", "%1", log) # must not raise
|
||
log.log.assert_not_called()
|
||
audit.assert_not_called()
|
||
|
||
|
||
class TestReconcileOrphan(unittest.TestCase):
|
||
"""Reconcile must not start a second watcher when an orphan with a
|
||
lost pidfile is already alive for the pane."""
|
||
|
||
def test_orphan_counts_as_already(self):
|
||
orphan = [{"pid": 7777, "socket": "/tmp/sock", "pane": "%37"}]
|
||
with mock.patch.object(w, "get_desired",
|
||
return_value={"enabled": True,
|
||
"dry_run": False}), \
|
||
mock.patch.object(w, "muse_panes", return_value=["%37"]), \
|
||
mock.patch.object(w, "is_running", return_value=None), \
|
||
mock.patch.object(w, "_watch_procs", return_value=orphan), \
|
||
mock.patch.object(w, "_start_detached") as start, \
|
||
mock.patch.object(w, "status_all", return_value=[]), \
|
||
mock.patch.object(w, "audit"), \
|
||
mock.patch.object(w.os.path, "exists", return_value=True):
|
||
res = w.reconcile(sockets=["/tmp/sock"])
|
||
start.assert_not_called()
|
||
self.assertEqual(res["already"], ["/tmp/sock:%37"])
|
||
self.assertEqual(res["failed"], [])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|