tests/test_join_contract.py
back to source
"""The topology-join contract, and whether the central claim is falsifiable.
Before this existed, "a booted machine joins the topology as a compute node"
had no state of the world that would refute it: nothing left the machine and
nothing recorded an outcome, so a node that never joined was byte-identical to
one that did. That is not an unfinished feature, it is an unfalsifiable claim.
The contract closes it in both directions:
inside a capability document and a join.state outcome exist on disk
outside a configured coordinator either received a capability or did not
and the property the whole thing exists for: a join that FAILS is recorded as
failed, never as joined.
It also has to stay on the right side of the no-telemetry commitment
(docs/PRINCIPLES.md, CCC 1907). Off by default, only ever to the operator's
own coordinator, inspectable before sending, refusal a first-class state.
"""
import http.server
import os
import shutil
import subprocess
import tempfile
import threading
import time
import unittest
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 Coordinator:
"""Stands in for an operator's coordinator; records what it received."""
def __init__(self, status=200):
self.received = []
received = self.received
class Handler(http.server.BaseHTTPRequestHandler):
def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
received.append(self.rfile.read(length).decode("utf-8", "replace"))
self.send_response(status)
self.end_headers()
def do_GET(self):
self.send_response(405)
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/join" % self.port
def stop(self):
self.httpd.shutdown()
self.httpd.server_close()
class JoinCase(unittest.TestCase):
def setUp(self):
self.work = tempfile.mkdtemp()
self.rundir = os.path.join(self.work, "run")
os.makedirs(self.rundir)
def tearDown(self):
shutil.rmtree(self.work, ignore_errors=True)
def numctl(self, *args, env_extra=None, timeout=60):
env = dict(os.environ)
env["NUMOS_RUNDIR"] = self.rundir.replace("\\", "/")
env["NUMOS_LIB"] = LIB
env.pop("NUMOS_COORDINATOR", None)
if env_extra:
env.update(env_extra)
return subprocess.run([_bash(), NUMCTL] + list(args),
capture_output=True, text=True, env=env,
timeout=timeout)
def read(self, name):
path = os.path.join(self.rundir, name)
if not os.path.exists(path):
return None
with open(path) as handle:
return handle.read()
def join_bg(self, env_extra=None, settle=6):
"""`join` is a longrun that never returns; run it and let it settle."""
env = dict(os.environ)
env["NUMOS_RUNDIR"] = self.rundir.replace("\\", "/")
env["NUMOS_LIB"] = LIB
env.pop("NUMOS_COORDINATOR", None)
if env_extra:
env.update(env_extra)
# Kill the whole tree, not just the shell. `join` is a longrun that
# sleeps in a loop, so killing only the bash parent orphans the sleep
# child; across the tests in this file those accumulate and were the
# likely cause of a one-off failure that did not reproduce.
kwargs = {}
if os.name == "nt":
kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
else:
kwargs["start_new_session"] = True
proc = subprocess.Popen([_bash(), NUMCTL, "join"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
env=env, **kwargs)
# Poll for the outcome rather than sleeping a fixed 6s. The wall-clock
# settle was the last false-positive channel in this suite: under load
# from leaked children it raced, so a contention failure was
# indistinguishable from a real one -- which meant revert-the-fix
# evidence for THIS contract could not be trusted.
# Wait for a TERMINAL outcome, not merely for the file to appear.
# `pending` is written before the POST is attempted, so polling on
# existence returned while the request was still in flight and the
# failure path read as `pending`. The condition has to be the thing
# actually being asserted.
terminal = ("disabled", "joined", "failed")
statefile = os.path.join(self.rundir, "join.state")
deadline = time.time() + settle
while time.time() < deadline:
try:
with open(statefile) as handle:
if handle.read().strip().startswith(terminal):
break
except OSError:
pass
if proc.poll() is not None:
break
time.sleep(0.1)
if os.name == "nt":
subprocess.run(["taskkill", "/F", "/T", "/PID", str(proc.pid)],
capture_output=True)
else:
import signal
try:
os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
except (ProcessLookupError, PermissionError):
pass
proc.kill()
proc.wait(timeout=30)
return proc
class TestCapability(JoinCase):
def test_capability_is_written_to_disk(self):
result = self.numctl("capability")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIsNotNone(self.read("capability"))
def test_it_reports_an_architecture(self):
self.numctl("capability")
self.assertRegex(self.read("capability"), r"(?m)^arch \S+$")
def test_unmeasurable_fields_are_dashes_not_guesses(self):
"""A capability is a claim a coordinator schedules against. A node that
cannot count its cores must say so, not invent a number."""
self.numctl("capability")
body = self.read("capability")
for field in ("cores", "ram_mb"):
line = [l for l in body.split("\n") if l.startswith(field + " ")]
self.assertTrue(line, "%s missing entirely" % field)
value = line[0].split(" ", 1)[1]
self.assertTrue(value == "-" or value.isdigit(),
"%s is neither a measurement nor '-': %r"
% (field, value))
def test_a_degraded_node_advertises_that_it_is_degraded(self):
"""The point of persisting degraded: a node must not advertise a
capability it does not have."""
with open(os.path.join(self.rundir, "degraded"), "w") as handle:
handle.write("1\n")
self.numctl("capability")
self.assertIn("degraded 1", self.read("capability"))
def test_a_healthy_node_advertises_degraded_zero(self):
self.numctl("capability")
self.assertIn("degraded 0", self.read("capability"))
def test_supported_ops_come_from_the_booted_state_not_a_registry(self):
state = os.path.join(self.work, "numos.state")
with open(state, "w", newline="\n") as handle:
handle.write("V 1\nU op-alpha-v1 longrun on-failure 1 2 - - - - true\n"
"U mount-proc oneshot never 0 0 - - - - true\n")
self.numctl("capability",
env_extra={"NUMOS_STATE": state.replace("\\", "/")})
body = self.read("capability")
self.assertIn("op op-alpha-v1", body)
self.assertNotIn("op mount-proc", body,
"a non-op unit was advertised as a supported op")
class TestJoinIsOffByDefault(JoinCase):
"""No coordinator means no network contact, ever."""
def test_no_coordinator_records_disabled(self):
self.join_bg()
self.assertEqual((self.read("join.state") or "").strip(), "disabled")
def test_capability_is_still_produced_when_join_is_disabled(self):
self.join_bg()
self.assertIsNotNone(self.read("capability"),
"the node must stay observable without joining")
def test_nothing_is_sent_when_no_coordinator_is_configured(self):
coordinator = Coordinator()
try:
self.join_bg() # NUMOS_COORDINATOR deliberately unset
self.assertEqual(coordinator.received, [],
"a beacon fired with no coordinator configured")
finally:
coordinator.stop()
class TestJoinWhenConfigured(JoinCase):
def test_a_successful_join_sends_the_capability_and_records_joined(self):
coordinator = Coordinator(status=200)
try:
self.join_bg(env_extra={"NUMOS_COORDINATOR": coordinator.url})
self.assertEqual(len(coordinator.received), 1,
"the coordinator received nothing")
self.assertIn("arch ", coordinator.received[0])
self.assertTrue((self.read("join.state") or "").startswith("joined"))
finally:
coordinator.stop()
def test_it_sends_exactly_what_was_written_to_disk(self):
"""Inspectable-before-send is only meaningful if the bytes match."""
coordinator = Coordinator(status=200)
try:
self.join_bg(env_extra={"NUMOS_COORDINATOR": coordinator.url})
self.assertEqual(coordinator.received[0].strip(),
(self.read("capability") or "").strip())
finally:
coordinator.stop()
def test_an_unreachable_coordinator_is_recorded_as_failed(self):
"""The falsifiability property. A machine that did not join must not
look like one that did."""
self.join_bg(env_extra={"NUMOS_COORDINATOR": "http://127.0.0.1:1/join"})
state = (self.read("join.state") or "").strip()
self.assertTrue(state.startswith("failed"),
"an unreachable coordinator produced %r, not failed" % state)
self.assertNotIn("joined", state)
def test_a_rejecting_coordinator_is_recorded_as_failed(self):
coordinator = Coordinator(status=503)
try:
self.join_bg(env_extra={"NUMOS_COORDINATOR": coordinator.url})
self.assertTrue((self.read("join.state") or "").startswith("failed"))
finally:
coordinator.stop()
def test_a_failed_join_does_not_stop_the_node(self):
"""Subsidiarity: an unreachable coordinator must not stop a machine
supervising itself."""
proc = self.join_bg(
env_extra={"NUMOS_COORDINATOR": "http://127.0.0.1:1/join"})
self.assertIsNotNone(self.read("capability"))
# It was killed by the harness because it stayed alive, which is the
# behaviour under test; a clean early exit would mean it gave up.
self.assertNotEqual(proc.returncode, 0)
class TestFalsifiability(JoinCase):
"""The three worlds must be distinguishable from one another.
This is the whole reason the contract exists. Before it, all three were
byte-identical.
"""
def test_disabled_joined_and_failed_are_all_distinguishable(self):
self.join_bg()
disabled = (self.read("join.state") or "").strip()
self.tearDown(); self.setUp()
coordinator = Coordinator(status=200)
try:
self.join_bg(env_extra={"NUMOS_COORDINATOR": coordinator.url})
joined = (self.read("join.state") or "").strip()
finally:
coordinator.stop()
self.tearDown(); self.setUp()
self.join_bg(env_extra={"NUMOS_COORDINATOR": "http://127.0.0.1:1/join"})
failed = (self.read("join.state") or "").strip()
self.assertEqual(len({disabled, joined, failed}), 3,
"join outcomes are not distinguishable: %r %r %r"
% (disabled, joined, failed))
if __name__ == "__main__":
unittest.main()