tests/test_op_dispatch.py
back to source
"""The op-dispatch contract, and what op units actually are.
A review found "the 196 op-derived units are exported but referenced by no
BootPhase, so nothing starts them" and classed it a wiring gap. The gap was
the modelling. They were `longrun` with `restart: on-failure` -- which says a
node should boot 196 self-restarting daemons. An op is work: dispatched, run,
finished. Wiring them into a phase would have started 196 daemons that
immediately and permanently failed.
So they are a dispatch catalogue: they enumerate what this node can be asked
to do, and `numctl capability` derives ops_supported from exactly these
records. Nothing requires them from a phase, and that is correct.
Dispatch itself refuses in two directions rather than pretending, because a
node that cannot do work must not look like one that did it.
"""
import http.server
import os
import shutil
import subprocess
import tempfile
import threading
import unittest
from numos.export_state import build_state
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
NUMCTL = os.path.join(ROOT, "boot", "numctl").replace("\\", "/")
LIB = os.path.join(ROOT, "boot", "lib").replace("\\", "/")
def _bash():
found = shutil.which("bash")
if found:
return found
for candidate in (r"C:\Program Files\Git\bin\bash.exe",
r"C:\Program Files\Git\usr\bin\bash.exe",
"/usr/bin/bash", "/bin/bash"):
if os.path.isfile(candidate):
return candidate
raise unittest.SkipTest("bash not found")
class Runtime:
"""Stands in for an op runtime; records the paths it was asked to run."""
def __init__(self, status=200):
self.paths = []
paths = self.paths
class Handler(http.server.BaseHTTPRequestHandler):
def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
self.rfile.read(length)
paths.append(self.path)
self.send_response(status)
self.end_headers()
def log_message(self, *a):
pass
self.httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
self.port = self.httpd.server_address[1]
threading.Thread(target=self.httpd.serve_forever, daemon=True).start()
@property
def url(self):
return "http://127.0.0.1:%d/ops" % self.port
def stop(self):
self.httpd.shutdown()
self.httpd.server_close()
class TestOpUnitsAreACatalogue(unittest.TestCase):
def test_op_units_are_oneshot_not_daemons(self):
"""The correction. An op is work, not a service."""
units = [u for u in build_state(["SwarmExecutor_v1"])["units"]
if u["name"].startswith("op-")]
self.assertTrue(units)
for unit in units:
self.assertEqual(unit["kind"], "oneshot",
"an op modelled as a daemon would be booted and "
"restarted forever")
self.assertEqual(unit["restart"], "never")
def test_no_boot_phase_requires_an_op_unit(self):
"""Deliberate: booting the catalogue would start work nobody asked for."""
state = build_state(["SwarmExecutor_v1", "ObserverChain_v1"])
required = set()
for phase in state["boot_phases"]:
required.update(phase["required_units"])
op_units = {u["name"] for u in state["units"]
if u["name"].startswith("op-")}
self.assertEqual(required & op_units, set(),
"a boot phase requires an op unit; the catalogue "
"would be started at boot")
def test_an_op_unit_still_requires_join(self):
"""A real precondition now, not boot ordering: no work before joining."""
units = [u for u in build_state(["SwarmExecutor_v1"])["units"]
if u["name"].startswith("op-")]
for unit in units:
self.assertIn("join", unit["requires"])
class DispatchCase(unittest.TestCase):
def setUp(self):
self.work = tempfile.mkdtemp()
self.rundir = os.path.join(self.work, "run")
os.makedirs(self.rundir)
self.state = os.path.join(self.work, "numos.state")
with open(self.state, "w", newline="\n") as handle:
handle.write("V 1\n"
"U op-swarmexecutor-v1 oneshot never 0 0 - - join - "
"numctl run-op SwarmExecutor_v1\n")
def tearDown(self):
shutil.rmtree(self.work, ignore_errors=True)
def run_op(self, op, env_extra=None):
env = dict(os.environ)
env["NUMOS_RUNDIR"] = self.rundir.replace("\\", "/")
env["NUMOS_LIB"] = LIB
env["NUMOS_STATE"] = self.state.replace("\\", "/")
env.pop("NUMOS_OP_RUNTIME", None)
if env_extra:
env.update(env_extra)
return subprocess.run([_bash(), NUMCTL, "run-op", op],
capture_output=True, text=True, env=env,
timeout=90)
def op_state(self, unit="op-swarmexecutor-v1"):
path = os.path.join(self.rundir, "ops", "%s.state" % unit)
if not os.path.exists(path):
return None
with open(path) as handle:
return handle.read().strip()
class TestDispatchRefusals(DispatchCase):
def test_an_undeclared_op_is_refused(self):
"""A node must not run what it does not advertise: capability derives
ops_supported from these same records."""
result = self.run_op("SomethingElse_v9")
self.assertNotEqual(result.returncode, 0)
self.assertIn("numctl: HALT:", result.stderr)
self.assertIn("not declared", result.stderr)
def test_no_runtime_configured_is_unavailable_not_success(self):
result = self.run_op("SwarmExecutor_v1")
self.assertNotEqual(result.returncode, 0,
"dispatch succeeded with no runtime configured")
self.assertIn("unavailable", result.stderr)
self.assertEqual(self.op_state(), "unavailable")
def test_nothing_is_dispatched_when_no_runtime_is_configured(self):
runtime = Runtime()
try:
self.run_op("SwarmExecutor_v1") # NUMOS_OP_RUNTIME unset
self.assertEqual(runtime.paths, [],
"work was dispatched with no runtime configured")
finally:
runtime.stop()
def test_a_missing_state_is_refused_rather_than_assumed(self):
env = {"NUMOS_STATE": os.path.join(self.work, "gone.state")}
result = self.run_op("SwarmExecutor_v1", env_extra=env)
self.assertNotEqual(result.returncode, 0)
self.assertIn("cannot confirm", result.stderr)
class TestDispatchWhenConfigured(DispatchCase):
def test_a_declared_op_is_dispatched_and_recorded(self):
runtime = Runtime(status=200)
try:
result = self.run_op("SwarmExecutor_v1",
env_extra={"NUMOS_OP_RUNTIME": runtime.url})
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(len(runtime.paths), 1)
self.assertIn("SwarmExecutor_v1", runtime.paths[0])
self.assertTrue((self.op_state() or "").startswith("ok"))
finally:
runtime.stop()
def test_a_rejecting_runtime_is_recorded_as_failed(self):
runtime = Runtime(status=500)
try:
result = self.run_op("SwarmExecutor_v1",
env_extra={"NUMOS_OP_RUNTIME": runtime.url})
self.assertNotEqual(result.returncode, 0)
self.assertTrue((self.op_state() or "").startswith("failed"))
finally:
runtime.stop()
def test_an_unreachable_runtime_is_recorded_as_failed(self):
result = self.run_op(
"SwarmExecutor_v1",
env_extra={"NUMOS_OP_RUNTIME": "http://127.0.0.1:1/ops"})
self.assertNotEqual(result.returncode, 0)
self.assertTrue((self.op_state() or "").startswith("failed"))
def test_never_run_dispatched_and_failed_are_distinguishable(self):
"""Same falsifiability property as the join contract, one layer in."""
never = self.op_state()
runtime = Runtime(status=200)
try:
self.run_op("SwarmExecutor_v1",
env_extra={"NUMOS_OP_RUNTIME": runtime.url})
ok = self.op_state()
finally:
runtime.stop()
self.run_op("SwarmExecutor_v1",
env_extra={"NUMOS_OP_RUNTIME": "http://127.0.0.1:1/ops"})
failed = self.op_state()
self.assertIsNone(never)
self.assertEqual(len({ok, failed}), 2,
"dispatched and failed are indistinguishable: %r %r"
% (ok, failed))
if __name__ == "__main__":
unittest.main()