super/box
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(work): import hashlib and wire heal subparser into main CLI

Browse Source
This commit is contained in:
operator
2026-10-09 23:13:43 +00:00
parent c3b4b1cd28
commit f75977ca6e
202 changed files with 4185 additions and 4142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_box_jobs_https.py
+51 -19
View File
@@ -37,6 +37,7 @@ def _load(name, relpath):
exec_constrained = _load("exec_constrained_jobs", "bin/exec-constrained.py")
box_ctl = _load("box_ctl_jobstest", "bin/box-ctl.py")
def _box_ctl(*args, stdin=None):
@@ -45,6 +46,37 @@ def _box_ctl(*args, stdin=None):
input=stdin, capture_output=True, text=True, timeout=120)
class _InProcResult:
"""Minimal CompletedProcess stand-in (returncode/stdout/stderr)."""
def __init__(self, returncode, stdout):
self.returncode = returncode
self.stdout = stdout
self.stderr = ""
def _box_ctl_inproc(*args, stdin=None):
"""In-process _box_ctl (proven pattern from test_box_loop_https).
Real main(argv) with stdout captured and sys.stdin patched when a
body is given (job-put reads the raw definition from stdin).
"""
import io
from contextlib import redirect_stdout, nullcontext
from unittest import mock
buf = io.StringIO()
returncode = 0
stdin_ctx = (mock.patch.object(sys, "stdin", io.StringIO(stdin))
if stdin is not None else nullcontext())
with stdin_ctx:
with redirect_stdout(buf):
try:
box_ctl.main(["box-ctl.py", *args])
except SystemExit as e:
returncode = e.code if isinstance(e.code, int) else 1
return _InProcResult(returncode, buf.getvalue())
def _job_def(name, **over):
d = {"name": name, "description": "unit test job",
"schedule": "manual", "agent": "opm",
@@ -166,7 +198,7 @@ class ExecJobOpsTests(unittest.TestCase):
class BoxCtlJobsTests(unittest.TestCase):
def test_job_next_dry_run_live(self):
r = _box_ctl("job-next", MISSING_ID)
r = _box_ctl_inproc("job-next", MISSING_ID)
self.assertEqual(r.returncode, 0, r.stderr)
data = json.loads(r.stdout)
self.assertTrue(data["ok"])
@@ -174,55 +206,55 @@ class BoxCtlJobsTests(unittest.TestCase):
self.assertFalse(data["would_dispatch"])
def test_job_put_rejects_before_write(self):
r = _box_ctl("job-put", "Bad_Name!", stdin="{}")
r = _box_ctl_inproc("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")
r = _box_ctl_inproc("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")))
r = _box_ctl_inproc("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))
r = _box_ctl_inproc("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!")
r = _box_ctl_inproc("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)
r = _box_ctl_inproc("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)
r = _box_ctl_inproc("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)
r = _box_ctl_inproc("job-chain", EXISTING_JOB, EXISTING_JOB)
self.assertEqual(json.loads(r.stdout)["code"], "INVALID_JOB")
r = _box_ctl("job-chain", MISSING_JOB, EXISTING_JOB)
r = _box_ctl_inproc("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!")
r = _box_ctl_inproc(verb, "Bad_Name!")
self.assertNotEqual(r.returncode, 0)
self.assertEqual(json.loads(r.stdout)["code"], "BAD_NAME")
r = _box_ctl(verb, MISSING_JOB)
r = _box_ctl_inproc(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")
r = _box_ctl_inproc("quality-validate", "job-put", "my-job")
self.assertTrue(json.loads(r.stdout)["valid"], r.stdout)
r = _box_ctl("quality-validate", "job-trigger", EXISTING_JOB)
r = _box_ctl_inproc("quality-validate", "job-trigger", EXISTING_JOB)
self.assertTrue(json.loads(r.stdout)["valid"], r.stdout)
r = _box_ctl("quality-validate", "job-chain", "a", "b")
r = _box_ctl_inproc("quality-validate", "job-chain", "a", "b")
self.assertTrue(json.loads(r.stdout)["valid"], r.stdout)
r = _box_ctl("quality-validate", "job-next", MISSING_ID)
r = _box_ctl_inproc("quality-validate", "job-next", MISSING_ID)
self.assertTrue(json.loads(r.stdout)["valid"], r.stdout)
r = _box_ctl("quality-validate", "timer-stop", EXISTING_JOB)
r = _box_ctl_inproc("quality-validate", "timer-stop", EXISTING_JOB)
self.assertTrue(json.loads(r.stdout)["valid"], r.stdout)
r = _box_ctl("quality-validate", "job-put", "Bad_Name!")
r = _box_ctl_inproc("quality-validate", "job-put", "Bad_Name!")
self.assertFalse(json.loads(r.stdout)["valid"], r.stdout)