feat(systemd): add and enable continuous tmux-auto-approver user daemon
This commit is contained in:
operator
@@ -239,6 +239,35 @@ class TestRuntimeRows(unittest.TestCase):
|
||||
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",
|
||||
|
||||
@@ -261,6 +261,18 @@ class ControlsTests(unittest.TestCase):
|
||||
self.assertTrue(controls.set_switch(mock.Mock(),
|
||||
"Transparent proxy", True))
|
||||
|
||||
def test_set_switch_never_reclicks(self):
|
||||
# A fallback re-click would undo a slow commit (switches
|
||||
# toggle), so a poll miss returns False without real Click.
|
||||
off = [{"label": "Transparent proxy row", "aria": "",
|
||||
"checked": False}]
|
||||
with mock.patch.object(controls, "cdp_evaluate",
|
||||
side_effect=[off, "CLICKED"] + [off] * 10), \
|
||||
mock.patch.object(controls, "real_click") as click:
|
||||
self.assertFalse(controls.set_switch(mock.Mock(),
|
||||
"Transparent proxy", True))
|
||||
click.assert_not_called()
|
||||
|
||||
def test_listers_reject_wrong_types(self):
|
||||
with mock.patch.object(controls, "cdp_evaluate",
|
||||
return_value="error"):
|
||||
@@ -332,6 +344,13 @@ class TogglesTests(unittest.TestCase):
|
||||
res = toggles.get_toggle("pip", "permissions.web_access")
|
||||
self.assertEqual((res["ok"], res["value"]), (True, "always_ask"))
|
||||
|
||||
def test_set_toggle_readonly_raises_before_cdp(self):
|
||||
with mock.patch.object(toggles, "get_cdp_ws") as g:
|
||||
with self.assertRaises(toggles.MenuError):
|
||||
toggles.set_toggle("pip", "data_controls.ai_improvement",
|
||||
"off")
|
||||
g.assert_not_called()
|
||||
|
||||
def test_set_toggle_bad_value_raises_before_cdp(self):
|
||||
with mock.patch.object(toggles, "get_cdp_ws") as g:
|
||||
with self.assertRaises(toggles.MenuError):
|
||||
@@ -552,6 +571,34 @@ class TabContractTests(unittest.TestCase):
|
||||
return_value=sws):
|
||||
self.assertTrue(data_controls.ai_improvement(mock.Mock()))
|
||||
|
||||
def test_set_ai_improvement_noop(self):
|
||||
row = {"label": "Help improve our AI models", "aria": "",
|
||||
"checked": True}
|
||||
with mock.patch.object(dialog, "goto_tab", return_value=True), \
|
||||
mock.patch.object(controls, "list_switches",
|
||||
return_value=[row]), \
|
||||
mock.patch.object(data_controls, "real_click") as click, \
|
||||
mock.patch("time.sleep"):
|
||||
self.assertTrue(data_controls.set_ai_improvement(
|
||||
mock.Mock(), True))
|
||||
click.assert_not_called()
|
||||
|
||||
def test_set_ai_improvement_real_click(self):
|
||||
off = {"label": "Help improve our AI models", "aria": "",
|
||||
"checked": False}
|
||||
on = {"label": "Help improve our AI models", "aria": "",
|
||||
"checked": True}
|
||||
with mock.patch.object(dialog, "goto_tab", return_value=True), \
|
||||
mock.patch.object(controls, "list_switches",
|
||||
side_effect=[[off], [on]]), \
|
||||
mock.patch.object(data_controls, "cdp_evaluate",
|
||||
return_value={"x": 1, "y": 2}), \
|
||||
mock.patch.object(data_controls, "real_click") as click, \
|
||||
mock.patch("time.sleep"):
|
||||
self.assertTrue(data_controls.set_ai_improvement(
|
||||
mock.Mock(), True))
|
||||
click.assert_called_once()
|
||||
|
||||
def test_data_controls_single_switch(self):
|
||||
one = [{"label": "Help improve the model", "aria": "",
|
||||
"checked": True}]
|
||||
|
||||
@@ -1086,5 +1086,370 @@ class TestExplicitPhrase(unittest.TestCase):
|
||||
"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"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user