tests/test_quarantine.py
back to source
"""Terminal state for a unit that will never come up.
Backoff bounded the DELAY; nothing bounded the ATTEMPTS. A unit that
crash-looped from boot retried forever at the ceiling, and it was invisible:
no health predicate fired, no capability was retracted, and nothing prompted
an operator to look. "Self-healing" was asserted but nothing distinguished
healing from an infinite retry with no exit condition.
Quarantine is deliberately not a halt. One dead unit must not take a machine
down - the node keeps running, marks itself degraded, and leaves a marker
naming what gave up.
"""
import os
import shutil
import subprocess
import tempfile
import unittest
from numos.state import render
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
NUMINIT = os.path.join(ROOT, "boot", "numinit.sh").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 QuarantineCase(unittest.TestCase):
def setUp(self):
self.work = tempfile.mkdtemp()
self.rundir = os.path.join(self.work, "run")
def tearDown(self):
shutil.rmtree(self.work, ignore_errors=True)
def run_init(self, restart="always", ticks=8, max_attempts="3"):
units = [{"name": "doomed", "kind": "longrun", "restart": restart,
"backoff_ms": 1, "backoff_max_ms": 2, "arch_mask": [],
"health_probe": None, "requires": [], "after": [],
"exec": "exit 1"}]
phases = [{"ordinal": 10, "name": "p", "on_failure": "continue",
"required_units": ["doomed"]}]
text = render({"version": 1, "arch_targets": [], "units": units,
"boot_phases": phases, "health": []})
path = os.path.join(self.work, "numos.state")
with open(path, "w", newline="\n") as handle:
handle.write(text)
env = dict(os.environ)
env.update({
"NUMOS_STATE": path.replace("\\", "/"),
"NUMOS_RUNDIR": self.rundir.replace("\\", "/"),
"NUMOS_MAX_TICKS": str(ticks),
"NUMOS_MAX_ATTEMPTS": max_attempts,
"NUMOS_NO_SLEEP": "1",
})
env.pop("NUMOS_DRY_RUN", None)
env.pop("NUMOS_SOURCE_ONLY", None)
proc = subprocess.run([_bash(), NUMINIT], capture_output=True,
text=True, env=env, timeout=180)
return proc.returncode, proc.stdout, proc.stderr
def marker(self, name="doomed"):
return os.path.join(self.rundir, "units", "%s.quarantined" % name)
class TestQuarantine(QuarantineCase):
def test_a_permanently_failing_unit_is_eventually_quarantined(self):
code, out, err = self.run_init(max_attempts="3")
self.assertEqual(code, 0, err)
self.assertIn("QUARANTINED doomed", out)
self.assertTrue(os.path.exists(self.marker()),
"no terminal marker was written; the unit would retry "
"forever and nothing could observe it")
def test_attempts_stop_at_the_ceiling(self):
"""The bug in one assertion: restarts must not grow without bound."""
_, out, _ = self.run_init(max_attempts="3", ticks=10)
restarts = out.count("numos: restart doomed")
self.assertLessEqual(restarts, 3,
"restarts exceeded NUMOS_MAX_ATTEMPTS=3: %d" % restarts)
self.assertGreater(restarts, 0, "the unit was never retried at all")
def test_quarantine_degrades_the_node_rather_than_halting_it(self):
"""One dead unit must not take the machine down."""
code, out, _ = self.run_init(max_attempts="2")
self.assertEqual(code, 0, "quarantine halted the boot instead of degrading")
self.assertIn("degraded=1", out)
self.assertIn("boot complete", out)
def test_the_marker_records_how_many_attempts_were_made(self):
self.run_init(max_attempts="2")
with open(self.marker()) as handle:
self.assertEqual(handle.read().strip(), "2")
def test_a_quarantined_unit_is_not_restarted_again(self):
"""Terminal means terminal: later ticks must leave it alone."""
_, out, _ = self.run_init(max_attempts="2", ticks=12)
after = out.split("QUARANTINED doomed", 1)[1]
self.assertNotIn("numos: restart doomed", after,
"a quarantined unit was restarted again")
class TestHealthyUnitsAreUnaffected(QuarantineCase):
def test_a_unit_that_stays_up_is_never_quarantined(self):
units = [{"name": "fine", "kind": "longrun", "restart": "always",
"backoff_ms": 1, "backoff_max_ms": 2, "arch_mask": [],
"health_probe": None, "requires": [], "after": [],
# Long enough to still be alive across the bounded tick loop, short
# enough that a leaked copy cannot starve a later module. A
# literal `sleep 30` here orphaned a real 30-second process on
# every run and was the clearest contributor to the suite
# intermittently reporting failures no targeted run reproduced.
"exec": "sleep 5"}]
phases = [{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["fine"]}]
text = render({"version": 1, "arch_targets": [], "units": units,
"boot_phases": phases, "health": []})
path = os.path.join(self.work, "numos.state")
with open(path, "w", newline="\n") as handle:
handle.write(text)
env = dict(os.environ)
env.update({
"NUMOS_STATE": path.replace("\\", "/"),
"NUMOS_RUNDIR": self.rundir.replace("\\", "/"),
"NUMOS_MAX_TICKS": "4",
"NUMOS_MAX_ATTEMPTS": "2",
"NUMOS_NO_SLEEP": "1",
})
env.pop("NUMOS_DRY_RUN", None)
proc = subprocess.run([_bash(), NUMINIT], capture_output=True,
text=True, env=env, timeout=180)
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertNotIn("QUARANTINED", proc.stdout)
self.assertFalse(os.path.exists(self.marker("fine")))
class TestHealthPredicateProducer(unittest.TestCase):
"""The falsifiability pivot.
X records rendered, parsed and round-tripped while numinit's health
machinery ran against an always-empty set: the runtime was complete and
tested, and nothing fed it. With no predicate, a node whose join unit died
on boot and one whose join unit is healthy produce byte-identical output.
"""
def test_the_exported_state_carries_at_least_one_predicate(self):
from numos.export_state import build_state
health = build_state([])["health"]
self.assertTrue(health, "export produced no health predicates; the "
"runtime machinery still has no producer")
def test_the_seeded_predicate_observes_the_join_unit(self):
from numos.export_state import build_state
probes = " ".join(h["probe"] for h in build_state([])["health"])
self.assertIn("join.pid", probes,
"nothing observes the one longrun unit in the seeded state")
def test_the_predicate_survives_render_and_verify(self):
from numos.export_state import build_state
from numos.state import render, parse, verify
text = render(build_state([]))
verify(text)
self.assertIn("X join-alive", text)
self.assertEqual(parse(text)["health"], build_state([])["health"])
def test_its_local_action_degrades_rather_than_restarting(self):
"""A liveness failure this floor cannot diagnose should mark the node
degraded, not thrash the unit."""
from numos.export_state import build_state
actions = [h["local_action"] for h in build_state([])["health"]]
self.assertIn("degrade-node", actions)
if __name__ == "__main__":
unittest.main()