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