operator
0065d11e97
- bin/muse_choice_watcher.py + systemd/muse-choices-reconcile.*: automatic choice answering and timer reconciliation - bin/digest.py: fleet log and health summarization - docs/BOX-*-HTTPS.md: comprehensive HTTPS execution contracts and API documentation - docs/MUSE-CHOICES-POLICY.md & docs/SUPERVISION-SPEC.md: autonomous execution specs - tests/test_*.py: unit test suites for HTTPS API, choice watcher, fleet heal, and swarm pruning
247 lines
11 KiB
Python
247 lines
11 KiB
Python
"""Tests for job lifecycle over HTTPS (no SSH).
|
|
|
|
Covers the job-lifecycle expansion:
|
|
box-relay.sh (agent client) -> exec-constrained.py named ops
|
|
-> box-ctl.py backend verbs -> jobs/*.json / systemd / job-dispatch.
|
|
|
|
Safe mutations only: put/trigger/chain/stop/disable. Deletes are
|
|
deliberately NOT exposed. Live writes, triggers, and timer control are
|
|
NEVER executed here: only validation-failure paths (which fail before any
|
|
side effect) plus the read-only job-next dry-run. Live-socket round-trips
|
|
are intentionally NOT covered here; instead we assert the exact argv each
|
|
op builds and execute the fast, side-effect-free argv directly.
|
|
"""
|
|
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"
|
|
|
|
# An existing job used for existence-gated validation (read-only).
|
|
EXISTING_JOB = "heartbeat"
|
|
# Well-formed names that must not exist (validation-failure paths only).
|
|
MISSING_JOB = "definitely-no-such-job-xyz"
|
|
MISSING_ID = "definitely-no-such-job-xyz-20200101-000000-deadbeef"
|
|
|
|
|
|
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_jobs", "bin/exec-constrained.py")
|
|
|
|
|
|
def _box_ctl(*args, stdin=None):
|
|
return subprocess.run(
|
|
[sys.executable, str(BOX_CTL), *args],
|
|
input=stdin, capture_output=True, text=True, timeout=120)
|
|
|
|
|
|
def _job_def(name, **over):
|
|
d = {"name": name, "description": "unit test job",
|
|
"schedule": "manual", "agent": "opm",
|
|
"prompt_template": "test prompt {job_id}", "timeout": 300}
|
|
d.update(over)
|
|
return d
|
|
|
|
|
|
class ExecJobOpsTests(unittest.TestCase):
|
|
def test_ops_registered_with_side_effect_flags(self):
|
|
spec = exec_constrained.OPS
|
|
for op in ("job.put", "job.trigger", "job.chain",
|
|
"cron.timer_stop", "cron.timer_disable"):
|
|
self.assertIn(op, spec)
|
|
self.assertTrue(spec[op]["side_effecting"])
|
|
self.assertIn("job.next", spec)
|
|
self.assertFalse(spec["job.next"]["side_effecting"])
|
|
|
|
def test_no_delete_ops_exposed(self):
|
|
names = set(exec_constrained.OPS)
|
|
self.assertNotIn("job.delete", names)
|
|
self.assertNotIn("cron.timer_delete", names)
|
|
|
|
def test_permissions(self):
|
|
p = exec_constrained.permitted
|
|
self.assertTrue(p("some-unknown-identity", "job.next"))
|
|
for op in ("job.put", "job.trigger", "job.chain",
|
|
"cron.timer_stop", "cron.timer_disable"):
|
|
self.assertFalse(p("some-unknown-identity", op))
|
|
self.assertTrue(p("operator-646", op))
|
|
self.assertFalse(p("exec-canary", "job.next"))
|
|
|
|
def test_job_put_validate(self):
|
|
v = exec_constrained.OPS["job.put"]["validate"]
|
|
good = v({"name": "my-job", "definition": _job_def("my-job")})
|
|
self.assertEqual(good["name"], "my-job")
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"name": "Bad_Name!", "definition": _job_def("x")})
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"name": "my-job", "definition": ["not", "a", "dict"]})
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"name": "my-job", "definition": _job_def("other")})
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"name": "my-job"})
|
|
|
|
def test_job_trigger_validate(self):
|
|
v = exec_constrained.OPS["job.trigger"]["validate"]
|
|
self.assertEqual(v({"name": EXISTING_JOB})["name"], EXISTING_JOB)
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"name": MISSING_JOB})
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"name": "Bad_Name!"})
|
|
|
|
def test_job_chain_validate(self):
|
|
v = exec_constrained.OPS["job.chain"]["validate"]
|
|
good = v({"from": EXISTING_JOB, "to": EXISTING_JOB})
|
|
self.assertFalse(good["on_failure"])
|
|
self.assertTrue(v({"from": EXISTING_JOB, "to": EXISTING_JOB,
|
|
"on_failure": True})["on_failure"])
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"from": MISSING_JOB, "to": EXISTING_JOB})
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"from": EXISTING_JOB})
|
|
|
|
def test_job_next_validate(self):
|
|
v = exec_constrained.OPS["job.next"]["validate"]
|
|
good = v({"job_id": MISSING_ID})
|
|
self.assertEqual(good["job_id"], MISSING_ID)
|
|
self.assertIsNone(good["success"])
|
|
self.assertTrue(v({"job_id": MISSING_ID, "success": True})["success"])
|
|
for bad in ("plainname", "a-20200101-000000-xyz!",
|
|
"UPPER-20200101-000000-deadbeef", ""):
|
|
with self.assertRaises(exec_constrained.OpError, msg=bad):
|
|
v({"job_id": bad})
|
|
|
|
def test_timer_validate(self):
|
|
for op in ("cron.timer_stop", "cron.timer_disable"):
|
|
v = exec_constrained.OPS[op]["validate"]
|
|
self.assertEqual(v({"name": EXISTING_JOB})["name"], EXISTING_JOB)
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"name": MISSING_JOB})
|
|
|
|
def test_build_argv_shapes(self):
|
|
put = exec_constrained.OPS["job.put"]
|
|
argv = put["build"]({"name": "my-job",
|
|
"definition": _job_def("my-job")})
|
|
self.assertEqual(argv[-2:], ["job-put", "my-job"])
|
|
trig = exec_constrained.OPS["job.trigger"]
|
|
self.assertEqual(trig["build"]({"name": EXISTING_JOB})[-2:],
|
|
["job-trigger", EXISTING_JOB])
|
|
chain = exec_constrained.OPS["job.chain"]
|
|
argv = chain["build"]({"from": "a", "to": "b", "on_failure": False})
|
|
self.assertEqual(argv[-3:], ["job-chain", "a", "b"])
|
|
argv = chain["build"]({"from": "a", "to": "b", "on_failure": True})
|
|
self.assertEqual(argv[-4:], ["job-chain", "a", "b", "--on-failure"])
|
|
nxt = exec_constrained.OPS["job.next"]
|
|
self.assertEqual(nxt["build"]({"job_id": "i", "success": None})[-2:],
|
|
["job-next", "i"])
|
|
argv = nxt["build"]({"job_id": "i", "success": False})
|
|
self.assertEqual(argv[-3:], ["job-next", "i", "--fail"])
|
|
stop = exec_constrained.OPS["cron.timer_stop"]
|
|
self.assertEqual(stop["build"]({"name": EXISTING_JOB})[-2:],
|
|
["timer-stop", EXISTING_JOB])
|
|
dis = exec_constrained.OPS["cron.timer_disable"]
|
|
self.assertEqual(dis["build"]({"name": EXISTING_JOB})[-2:],
|
|
["timer-disable", EXISTING_JOB])
|
|
self.assertIsInstance(argv, list)
|
|
|
|
def test_stdin_body_routing(self):
|
|
body = exec_constrained._stdin_body(
|
|
"job.put", {"name": "my-job", "definition": _job_def("my-job")})
|
|
# box-ctl job-put reads the raw definition (not the envelope).
|
|
self.assertEqual(json.loads(body)["name"], "my-job")
|
|
self.assertNotIn("definition", json.loads(body))
|
|
self.assertIsNone(exec_constrained._stdin_body("job.trigger", {}))
|
|
env = exec_constrained._stdin_body("files.read", {"path": "x"})
|
|
self.assertEqual(json.loads(env), {"path": "x"})
|
|
|
|
|
|
class BoxCtlJobsTests(unittest.TestCase):
|
|
def test_job_next_dry_run_live(self):
|
|
r = _box_ctl("job-next", MISSING_ID)
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
data = json.loads(r.stdout)
|
|
self.assertTrue(data["ok"])
|
|
self.assertEqual(data["job_id"], MISSING_ID)
|
|
self.assertFalse(data["would_dispatch"])
|
|
|
|
def test_job_put_rejects_before_write(self):
|
|
r = _box_ctl("job-put", "Bad_Name!", stdin="{}")
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertEqual(json.loads(r.stdout)["code"], "BAD_NAME")
|
|
r = _box_ctl("job-put", "my-job", stdin="not json")
|
|
self.assertEqual(json.loads(r.stdout)["code"], "INVALID_JOB")
|
|
r = _box_ctl("job-put", "my-job",
|
|
stdin=json.dumps(_job_def("other")))
|
|
self.assertEqual(json.loads(r.stdout)["code"], "NAME_MISMATCH")
|
|
bad = _job_def("my-job")
|
|
del bad["agent"]
|
|
r = _box_ctl("job-put", "my-job", stdin=json.dumps(bad))
|
|
self.assertEqual(json.loads(r.stdout)["code"], "INVALID_JOB")
|
|
|
|
def test_job_trigger_rejects_missing(self):
|
|
r = _box_ctl("job-trigger", "Bad_Name!")
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertEqual(json.loads(r.stdout)["code"], "BAD_NAME")
|
|
r = _box_ctl("job-trigger", MISSING_JOB)
|
|
self.assertEqual(json.loads(r.stdout)["code"], "NOT_FOUND")
|
|
|
|
def test_job_chain_rejects_before_write(self):
|
|
r = _box_ctl("job-chain", "Bad_Name!", EXISTING_JOB)
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertEqual(json.loads(r.stdout)["code"], "BAD_NAME")
|
|
r = _box_ctl("job-chain", EXISTING_JOB, EXISTING_JOB)
|
|
self.assertEqual(json.loads(r.stdout)["code"], "INVALID_JOB")
|
|
r = _box_ctl("job-chain", MISSING_JOB, EXISTING_JOB)
|
|
self.assertEqual(json.loads(r.stdout)["code"], "NOT_FOUND")
|
|
|
|
def test_timer_control_rejects_before_action(self):
|
|
for verb in ("timer-stop", "timer-disable"):
|
|
r = _box_ctl(verb, "Bad_Name!")
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertEqual(json.loads(r.stdout)["code"], "BAD_NAME")
|
|
r = _box_ctl(verb, MISSING_JOB)
|
|
self.assertEqual(json.loads(r.stdout)["code"], "NOT_FOUND")
|
|
|
|
def test_quality_validate_job_verbs(self):
|
|
r = _box_ctl("quality-validate", "job-put", "my-job")
|
|
self.assertTrue(json.loads(r.stdout)["valid"], r.stdout)
|
|
r = _box_ctl("quality-validate", "job-trigger", EXISTING_JOB)
|
|
self.assertTrue(json.loads(r.stdout)["valid"], r.stdout)
|
|
r = _box_ctl("quality-validate", "job-chain", "a", "b")
|
|
self.assertTrue(json.loads(r.stdout)["valid"], r.stdout)
|
|
r = _box_ctl("quality-validate", "job-next", MISSING_ID)
|
|
self.assertTrue(json.loads(r.stdout)["valid"], r.stdout)
|
|
r = _box_ctl("quality-validate", "timer-stop", EXISTING_JOB)
|
|
self.assertTrue(json.loads(r.stdout)["valid"], r.stdout)
|
|
r = _box_ctl("quality-validate", "job-put", "Bad_Name!")
|
|
self.assertFalse(json.loads(r.stdout)["valid"], r.stdout)
|
|
|
|
|
|
class BoxRelayJobsTests(unittest.TestCase):
|
|
def test_relay_help_lists_job_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 cron put", "box cron trigger", "box cron chain",
|
|
"box cron next", "box timer stop"):
|
|
self.assertIn(line, r.stdout)
|
|
|
|
def test_relay_maps_job_commands_to_ops(self):
|
|
text = RELAY.read_text()
|
|
for op in ('"job.put"', '"job.trigger"', '"job.chain"', '"job.next"',
|
|
'"cron.timer_stop"', '"cron.timer_disable"'):
|
|
self.assertIn(op, text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|