"""Tests for loop + strategy writes over HTTPS (no SSH). Covers the loop/strategy expansion: box-relay.sh (agent client) -> exec-constrained.py named ops -> box-ctl.py backend verbs -> followups/variables/strategy state. Safe mutations only. Live execution is limited to side-effect-free paths: loop-remediate --dry-run (all writes guarded), strat-reset on a probe key that can never exist (returns False, no write), and validation-failure paths (which fail before any side effect). loop-resolve always appends to job-log, and vars-reset/rollback/strat-set mutate live fleet state, so those success paths are covered by quality-validate (dry-run) plus unit tests — never executed here. Live-socket round-trips are intentionally NOT covered here; instead we assert the exact argv each op builds. """ import importlib.util import json import subprocess import sys import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent BOX_CTL = REPO_ROOT / "bin" / "box-ctl.py" RELAY = REPO_ROOT / "bin" / "box-relay.sh" MISSING_VAR = "definitely-no-such-var-xyz" # A strategy key no agent will ever set: reset returns False, no write. PROBE_TYPE = "heartbeat" PROBE_SUBTYPE = "ZZZ_PROBE_NO_SUCH_SUBTYPE" def _load(name, relpath): spec = importlib.util.spec_from_file_location(name, REPO_ROOT / relpath) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod exec_constrained = _load("exec_constrained_loop", "bin/exec-constrained.py") def _box_ctl(*args): return subprocess.run( [sys.executable, str(BOX_CTL), *args], capture_output=True, text=True, timeout=180) class ExecLoopOpsTests(unittest.TestCase): def test_ops_registered_and_side_effecting(self): spec = exec_constrained.OPS for op in ("loop.remediate", "loop.resolve", "strat.set", "strat.reset", "vars.reset", "vars.rollback"): self.assertIn(op, spec) self.assertTrue(spec[op]["side_effecting"]) def test_known_identities_only(self): p = exec_constrained.permitted for op in ("loop.remediate", "loop.resolve", "strat.set", "strat.reset", "vars.reset", "vars.rollback"): self.assertFalse(p("some-unknown-identity", op)) self.assertTrue(p("operator-646", op)) self.assertFalse(p("exec-canary", "loop.remediate")) def test_remediate_validate(self): v = exec_constrained.OPS["loop.remediate"]["validate"] self.assertEqual(v({}), {"dry_run": False}) self.assertTrue(v({"dry_run": True})["dry_run"]) with self.assertRaises(exec_constrained.OpError): v({"bogus": 1}) def test_resolve_validate(self): v = exec_constrained.OPS["loop.resolve"]["validate"] good = v({"dm_id": "bdf7beb6", "note": "looks good"}) self.assertEqual(good["dm_id"], "bdf7beb6") self.assertEqual(good["note"], "looks good") self.assertIsNone(v({"dm_id": "bdf7beb6"})["note"]) with self.assertRaises(exec_constrained.OpError): v({"dm_id": "xyz!"}) with self.assertRaises(exec_constrained.OpError): v({"dm_id": "bdf7beb6", "note": " "}) def test_strat_set_validate(self): v = exec_constrained.OPS["strat.set"]["validate"] good = v({"type": "job", "priority": "important", "nudges": 3}) self.assertEqual(good["type"], "job") self.assertEqual(good["priority"], "important") # Typos must fail: the backend silently maps unknown types to MANUAL. with self.assertRaises(exec_constrained.OpError, msg="typo type"): v({"type": "wkae"}) with self.assertRaises(exec_constrained.OpError): v({"type": "job", "priority": "urgent"}) with self.assertRaises(exec_constrained.OpError): v({"type": "job", "timeout_s": "soon"}) with self.assertRaises(exec_constrained.OpError): v({"type": "job", "agent": "nope"}) with self.assertRaises(exec_constrained.OpError): v({"priority": "normal"}) def test_strat_reset_validate(self): v = exec_constrained.OPS["strat.reset"]["validate"] good = v({"type": "heartbeat", "subtype": "DM", "agent": "opm"}) self.assertEqual(good, {"type": "heartbeat", "subtype": "DM", "agent": "opm"}) with self.assertRaises(exec_constrained.OpError): v({"type": "wkae"}) with self.assertRaises(exec_constrained.OpError): v({"type": "job", "subtype": "has space"}) def test_vars_validate(self): vr = exec_constrained.OPS["vars.reset"]["validate"] self.assertEqual(vr({"name": "max_nudge_count"})["name"], "max_nudge_count") with self.assertRaises(exec_constrained.OpError): vr({"name": "has space"}) vb = exec_constrained.OPS["vars.rollback"]["validate"] self.assertIsNone(vb({"name": "max_nudge_count"})["revision"]) self.assertEqual(vb({"name": "x", "revision": 2})["revision"], 2) with self.assertRaises(exec_constrained.OpError): vb({"name": "x", "revision": 0}) with self.assertRaises(exec_constrained.OpError): vb({"name": "x", "revision": "a\nb"}) def test_build_argv_shapes(self): rem = exec_constrained.OPS["loop.remediate"] self.assertEqual(rem["build"]({"dry_run": False})[-1], "loop-remediate") argv = rem["build"]({"dry_run": True}) self.assertEqual(argv[-2:], ["loop-remediate", "--dry-run"]) res = exec_constrained.OPS["loop.resolve"] argv = res["build"]({"dm_id": "abc123", "note": None}) self.assertEqual(argv[-2:], ["loop-resolve", "abc123"]) argv = res["build"]({"dm_id": "abc123", "note": "n"}) self.assertEqual(argv[-3:], ["loop-resolve", "abc123", "n"]) st = exec_constrained.OPS["strat.set"] argv = st["build"]({"type": "job", "subtype": None, "agent": None, "track": None, "priority": "normal", "timeout_s": None, "nudges": 2, "escalate": None}) self.assertEqual(argv[-3], "strat-set") payload = json.loads(argv[-1]) self.assertEqual(payload["priority"], "normal") self.assertEqual(payload["nudges"], 2) sr = exec_constrained.OPS["strat.reset"] argv = sr["build"]({"type": "job", "subtype": "DM", "agent": "opm"}) self.assertEqual(argv[-4:], ["strat-reset", "job", "DM", "--agent", "opm"][-4:]) vrt = exec_constrained.OPS["vars.reset"] self.assertEqual(vrt["build"]({"name": "x"})[-2:], ["vars-reset", "x"]) vrb = exec_constrained.OPS["vars.rollback"] argv = vrb["build"]({"name": "x", "revision": 2}) self.assertEqual(argv[-3:], ["vars-rollback", "x", "2"]) self.assertIsInstance(argv, list) class BoxCtlLoopTests(unittest.TestCase): def test_remediate_dry_run_live(self): r = _box_ctl("loop-remediate", "--dry-run") self.assertEqual(r.returncode, 0, r.stderr) data = json.loads(r.stdout) self.assertTrue(data["ok"]) self.assertTrue(data["dry_run"]) self.assertIn("remediated", data) self.assertIn("escalated", data) def test_strat_reset_probe_key_live(self): r = _box_ctl("strat-reset", PROBE_TYPE, PROBE_SUBTYPE) self.assertEqual(r.returncode, 0, r.stderr) data = json.loads(r.stdout) self.assertTrue(data["ok"]) self.assertFalse(data["reset"]) def test_strat_set_rejects_before_write(self): r = _box_ctl("strat-set") self.assertNotEqual(r.returncode, 0) self.assertEqual(json.loads(r.stdout)["code"], "BAD_NAME") r = _box_ctl("strat-set", "job", "not json") self.assertEqual(json.loads(r.stdout)["code"], "STRAT_ERROR") bad = json.dumps({"priority": "urgent"}) r = _box_ctl("strat-set", "job", bad) self.assertEqual(json.loads(r.stdout)["code"], "STRAT_ERROR") def test_vars_rejects_unknown_before_write(self): r = _box_ctl("vars-reset", MISSING_VAR) self.assertNotEqual(r.returncode, 0) self.assertEqual(json.loads(r.stdout)["code"], "VARS_ERROR") r = _box_ctl("vars-rollback", MISSING_VAR) self.assertNotEqual(r.returncode, 0) self.assertEqual(json.loads(r.stdout)["code"], "VARS_ERROR") def test_quality_validate_loop_verbs(self): cases = [ (["loop-remediate"], True), (["loop-remediate", "--dry-run"], True), (["loop-resolve", "bdf7beb6"], True), (["loop-resolve", "bdf7beb6", "note"], True), (["loop-resolve", "xyz!"], False), (["strat-set", "job"], True), (["strat-set"], False), (["strat-reset", "job", "DM", "--agent", "opm"], True), (["strat-reset", "--agent", "nope"], False), (["vars-reset", "max_nudge_count"], True), (["vars-rollback", "max_nudge_count", "2"], True), (["vars-get", "loop_health_threshold"], True), (["vars-set", "max_nudge_count", "5"], True), (["vars-reset"], False), (["vars-get", "has space"], False), ] for args, valid in cases: r = _box_ctl("quality-validate", *args) self.assertEqual(json.loads(r.stdout)["valid"], valid, args) class BoxRelayLoopTests(unittest.TestCase): def test_relay_help_lists_loop_commands(self): r = subprocess.run(["bash", str(RELAY), "help"], capture_output=True, text=True, timeout=30) self.assertEqual(r.returncode, 0, r.stderr) for line in ("box loop remediate", "box loop resolve", "box strat set", "box strat reset", "box vars reset", "box vars rollback"): self.assertIn(line, r.stdout) def test_relay_maps_loop_commands_to_ops(self): text = RELAY.read_text() for op in ('"loop.remediate"', '"loop.resolve"', '"strat.set"', '"strat.reset"', '"vars.reset"', '"vars.rollback"'): self.assertIn(op, text) if __name__ == "__main__": unittest.main()