tests/test_numinit_steady.py
back to source
"""Steady state: what makes numinit an init system rather than a boot script.
Until this exists, numinit walks the phases and returns - and a PID 1 that
exits panics the kernel. It also means a longrun unit is started and then
never watched, and HealthPredicate records round-trip through the state
format without anything ever executing them.
The loop is tick-driven and bounded by NUMOS_MAX_TICKS so it can be tested;
in a real boot NUMOS_MAX_TICKS is unset and it never returns.
"""
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():
"""Resolve bash for Windows hosts where Git bash is installed but not PATH."""
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 (install Git for Windows or put bash on PATH)")
def unit(name, exec_="true", kind="oneshot", restart="never",
backoff_ms=0, backoff_max_ms=0, health_probe=None,
requires=None, after=None):
return {"name": name, "kind": kind, "restart": restart,
"backoff_ms": backoff_ms, "backoff_max_ms": backoff_max_ms,
"arch_mask": [], "health_probe": health_probe,
"requires": requires or [], "after": after or [], "exec": exec_}
def health(name, probe="true", interval_s=1, threshold=1,
local_action="none", on_fire=None):
return {"name": name, "interval_s": interval_s, "threshold": threshold,
"local_action": local_action, "on_fire": on_fire, "probe": probe}
class SteadyCase(unittest.TestCase):
def setUp(self):
self.work = tempfile.mkdtemp()
def tearDown(self):
shutil.rmtree(self.work, ignore_errors=True)
def state_file(self, units, phases=None, healths=None):
text = render({"version": 1, "arch_targets": [], "units": units,
"boot_phases": phases or [], "health": healths or []})
path = os.path.join(self.work, "numos.state")
with open(path, "w", newline="\n") as handle:
handle.write(text)
return path.replace("\\", "/")
def run_init(self, state, ticks=3, extra=None):
env = dict(os.environ)
env["NUMOS_STATE"] = state
env["NUMOS_RUNDIR"] = os.path.join(self.work, "run").replace("\\", "/")
env["NUMOS_MAX_TICKS"] = str(ticks)
env["NUMOS_NO_SLEEP"] = "1"
env.pop("NUMOS_DRY_RUN", None)
env.pop("NUMOS_SOURCE_ONLY", None)
if extra:
env.update(extra)
proc = subprocess.run([_bash(), NUMINIT], capture_output=True,
text=True, env=env, timeout=120)
return proc.returncode, proc.stdout, proc.stderr
def marker(self, name):
return os.path.join(self.work, name).replace("\\", "/")
class TestLoopRuns(SteadyCase):
def test_steady_state_runs_after_the_phase_walk(self):
state = self.state_file(
[unit("a")],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["a"]}])
code, out, err = self.run_init(state, ticks=2)
self.assertEqual(code, 0, err)
self.assertIn("boot complete", out)
self.assertIn("steady state", out)
def test_a_halted_boot_never_reaches_steady_state(self):
state = self.state_file(
[unit("boom", exec_="false")],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["boom"]}])
code, out, err = self.run_init(state, ticks=2)
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
self.assertNotIn("steady state", out)
class TestLongrunSupervision(SteadyCase):
"""A longrun unit is backgrounded by the phase walk. Steady state is what
notices it died and applies its restart policy."""
def _state(self, restart, script):
counter = self.marker("count")
exec_ = "sh -c '%s'" % script.replace("COUNTER", counter)
return self.state_file(
[unit("d", exec_=exec_, kind="longrun", restart=restart,
backoff_ms=1, backoff_max_ms=4)],
[{"ordinal": 10, "name": "p", "on_failure": "continue",
"required_units": ["d"]}]), counter
def test_restart_always_relaunches_a_unit_that_exits(self):
state, counter = self._state("always", 'echo x >> COUNTER')
code, out, err = self.run_init(state, ticks=4)
self.assertEqual(code, 0, err)
with open(counter) as handle:
runs = len(handle.readlines())
self.assertGreater(runs, 1, "restart=always did not relaunch: %s" % out)
self.assertIn("restart d", out)
def test_restart_never_does_not_relaunch(self):
state, counter = self._state("never", 'echo x >> COUNTER')
code, out, err = self.run_init(state, ticks=4)
self.assertEqual(code, 0, err)
with open(counter) as handle:
runs = len(handle.readlines())
self.assertEqual(runs, 1, "restart=never relaunched: %s" % out)
self.assertIn("restart=never", out)
def test_restart_on_failure_relaunches_only_a_failing_unit(self):
state, counter = self._state("on-failure", 'echo x >> COUNTER; exit 1')
_, out, _ = self.run_init(state, ticks=4)
with open(counter) as handle:
failing_runs = len(handle.readlines())
self.tearDown()
self.setUp()
state, counter = self._state("on-failure", 'echo x >> COUNTER; exit 0')
_, out2, _ = self.run_init(state, ticks=4)
with open(counter) as handle:
ok_runs = len(handle.readlines())
self.assertGreater(failing_runs, 1, "a failing on-failure unit was not restarted")
self.assertEqual(ok_runs, 1, "a successful on-failure unit was restarted: %s" % out2)
class TestHealthPredicates(SteadyCase):
def test_a_passing_probe_never_fires(self):
state = self.state_file(
[unit("a")],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["a"]}],
[health("ok", probe="true", interval_s=1, threshold=1)])
_, out, err = self.run_init(state, ticks=3)
self.assertNotIn("HEALTH-FIRE", out, err)
def test_a_failing_probe_fires_only_after_its_threshold(self):
state = self.state_file(
[unit("a")],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["a"]}],
[health("bad", probe="false", interval_s=1, threshold=3)])
_, out, _ = self.run_init(state, ticks=2)
self.assertNotIn("HEALTH-FIRE", out,
"fired before reaching the threshold")
_, out, _ = self.run_init(state, ticks=5)
self.assertIn("HEALTH-FIRE bad", out)
def test_interval_gates_how_often_a_probe_runs(self):
counter = self.marker("probes")
state = self.state_file(
[unit("a")],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["a"]}],
[health("slow", probe="sh -c 'echo p >> %s'" % counter,
interval_s=4, threshold=99)])
self.run_init(state, ticks=4)
with open(counter) as handle:
self.assertEqual(len(handle.readlines()), 1,
"interval_s=4 should gate the probe to one run in 4 ticks")
def test_a_recovered_probe_resets_its_failure_count(self):
flag = self.marker("flag")
# Fails while the flag is absent; the first probe creates it, so the
# second probe passes. With threshold 2 and no reset, that would fire.
probe = "sh -c '[ -f %s ] || { touch %s; exit 1; }'" % (flag, flag)
state = self.state_file(
[unit("a")],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["a"]}],
[health("flappy", probe=probe, interval_s=1, threshold=2)])
_, out, _ = self.run_init(state, ticks=5)
self.assertIn("recovered", out)
self.assertNotIn("HEALTH-FIRE", out,
"failure count was not reset by a passing probe")
def test_degrade_node_action_marks_the_node_degraded(self):
state = self.state_file(
[unit("a")],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["a"]}],
[health("disk", probe="false", interval_s=1, threshold=1,
local_action="degrade-node")])
_, out, _ = self.run_init(state, ticks=3)
self.assertIn("HEALTH-FIRE disk", out)
self.assertIn("degraded=1", out)
def test_an_unknown_local_action_halts_rather_than_being_ignored(self):
state = self.state_file(
[unit("a")],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["a"]}],
[health("weird", probe="false", interval_s=1, threshold=1,
local_action="teleport")])
code, _, err = self.run_init(state, ticks=3)
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
self.assertIn("teleport", err)
def test_an_on_fire_swarm_is_queued_rather_than_blocking(self):
rundir = os.path.join(self.work, "run")
state = self.state_file(
[unit("a")],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["a"]}],
[health("net", probe="false", interval_s=1, threshold=1,
local_action="none", on_fire="swarm-heal-net")])
_, out, _ = self.run_init(state, ticks=3)
self.assertIn("queued escalation", out)
queue = os.path.join(rundir, "escalations.queue")
self.assertTrue(os.path.exists(queue), "escalation was not persisted")
with open(queue) as handle:
self.assertIn("swarm-heal-net", handle.read())
if __name__ == "__main__":
unittest.main()