tests/test_numinit_kind.py
back to source
"""Unit `kind` semantics in numinit's phase walk.
Every unit used to run synchronously, so a `longrun` unit -- a daemon that
by definition never returns -- blocked the phase walk forever. The seeded
default state has exactly that shape (`join` is longrun and is phase 50's
required unit), so the shipped configuration hung PID 1. These tests hold
the fix: longrun starts and returns, oneshot still runs to completion and
still propagates its failure.
The longrun discriminator is deliberately not a wall-clock threshold.
Spawning a process under git-bash on Windows costs a large and variable
fraction of a second, and the seeded phase walk runs sixteen units, so any
fixed time bound would be either flaky or too loose to fail. Instead the
daemon runs until the test releases a lock file and only then writes a
marker: if numinit exits with the marker still absent, the phase walk
demonstrably did not wait for the daemon. If it does wait, numinit never
exits and the subprocess timeout fires.
"""
import os
import subprocess
import tempfile
import time
import unittest
from numos import seed
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("\\", "/")
# Long enough for the whole seeded phase walk on a slow box; a blocked
# phase walk hits it and the test goes red on TimeoutExpired.
BOOT_TIMEOUT_SECONDS = 240
def unit(name, kind="oneshot", exec_="true", requires=None, after=None):
return {"name": name, "kind": kind, "restart": "never",
"backoff_ms": 0, "backoff_max_ms": 0, "arch_mask": [],
"health_probe": None, "requires": requires or [],
"after": after or [], "exec": exec_}
def state_text(units, phases):
return render({"version": 1, "arch_targets": [], "units": units,
"boot_phases": phases, "health": []})
def write_state(text):
handle = tempfile.NamedTemporaryFile("w", suffix=".numos", delete=False, newline="\n")
handle.write(text)
handle.close()
return handle.name.replace("\\", "/")
def temp_path(suffix):
"""Reserve a temp file name without leaving the file behind."""
handle = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
handle.close()
os.unlink(handle.name)
return handle.name.replace("\\", "/")
def try_unlink(path):
"""Delete a file a still-running background unit may still hold open.
On Windows a backgrounded longrun unit inherits numinit's stdout handle
and keeps the file locked until it exits, so deletion is best effort.
"""
for _ in range(20):
try:
os.unlink(path)
return
except FileNotFoundError:
return
except OSError:
time.sleep(0.1)
class Daemon(object):
"""A longrun exec that runs until released, then records that it ended.
Its marker file is the evidence: present means the daemon finished,
absent means it was still running.
"""
def __init__(self):
self.lock = temp_path(".lock")
self.marker = temp_path(".marker")
with open(self.lock, "w") as handle:
handle.write("held\n")
@property
def exec_(self):
return ("while [ -e '%s' ]; do sleep 1; done; echo ENDED >> '%s'"
% (self.lock, self.marker))
def finished(self):
return os.path.exists(self.marker)
def release(self):
try_unlink(self.lock)
def cleanup(self):
self.release()
try_unlink(self.marker)
def run_main(state_path, dry_run=False, release=None,
timeout=BOOT_TIMEOUT_SECONDS):
"""Run numinit end to end and return (rc, stdout, stderr, elapsed).
stdout/stderr go to real files rather than pipes: a backgrounded longrun
unit inherits the write end of a pipe and would hold it open for its
whole life, so a pipe-reading parent would block even though numinit
itself already exited -- which is the very thing under test. `release`
runs the instant numinit exits, before anything is read or deleted, so
an assertion can distinguish "still running" from "already finished".
"""
env = dict(os.environ)
env["NUMOS_STATE"] = state_path
if dry_run:
env["NUMOS_DRY_RUN"] = "1"
else:
env.pop("NUMOS_DRY_RUN", None)
out_handle = tempfile.NamedTemporaryFile("w+", delete=False)
err_handle = tempfile.NamedTemporaryFile("w+", delete=False)
out_name, err_name = out_handle.name, err_handle.name
try:
started = time.time()
proc = subprocess.run(["bash", NUMINIT], stdout=out_handle, stderr=err_handle,
env=env, timeout=timeout)
elapsed = time.time() - started
if release is not None:
release()
out_handle.close()
err_handle.close()
with open(out_name) as handle:
out = handle.read()
with open(err_name) as handle:
err = handle.read()
return proc.returncode, out, err, elapsed
finally:
if release is not None:
release()
for handle in (out_handle, err_handle):
if not handle.closed:
handle.close()
try_unlink(out_name)
try_unlink(err_name)
class TestLongrunDoesNotBlock(unittest.TestCase):
def test_longrun_unit_does_not_block_the_phase_walk(self):
daemon = Daemon()
path = write_state(state_text(
[unit("daemon", kind="longrun", exec_=daemon.exec_),
unit("later", exec_="echo NUMOS_LATER_RAN")],
[{"ordinal": 10, "name": "svc", "on_failure": "halt",
"required_units": ["daemon"]},
{"ordinal": 20, "name": "after", "on_failure": "halt",
"required_units": ["later"]}]))
try:
code, out, err, _ = run_main(path, release=daemon.release)
self.assertEqual(code, 0, err)
self.assertIn("numos: boot complete degraded=0", out)
self.assertIn("NUMOS_LATER_RAN", out)
finally:
daemon.cleanup()
os.unlink(path)
def test_boot_completes_while_the_longrun_unit_is_still_running(self):
"""The assertion that actually discriminates C1.
The daemon writes its marker only when it ends. numinit reaching
`boot complete` with no marker on disk means the phase walk moved on
while the daemon was still up -- the exact behavior a synchronous
run cannot produce.
"""
daemon = Daemon()
path = write_state(state_text(
[unit("daemon", kind="longrun", exec_=daemon.exec_)],
[{"ordinal": 10, "name": "svc", "on_failure": "halt",
"required_units": ["daemon"]}]))
try:
code, out, err, _ = run_main(path)
still_running = not daemon.finished()
self.assertEqual(code, 0, err)
self.assertIn("numos: boot complete", out)
self.assertTrue(still_running,
"the longrun unit had already ended when numinit "
"exited: the phase walk waited on it")
finally:
daemon.cleanup()
os.unlink(path)
def test_seeded_default_state_shape_boots_instead_of_hanging(self):
"""The seeded phase/unit graph, with execs this host can actually run.
seed.INFRA_UNITS' real execs (mount, ip, ntpd, numctl) do not exist
on the development machine, so the literal seed state cannot be run
to completion here -- that is the QEMU gate's job, and no claim is
made that it boots. What is reproducible is the shape that hung: the
seeded phases, the seeded DAG, and `join` still longrun.
"""
daemon = Daemon()
units = []
for src in seed.INFRA_UNITS:
copy = dict(src)
copy["exec"] = daemon.exec_ if src["kind"] == "longrun" else "true"
units.append(copy)
self.assertEqual([u["name"] for u in units if u["kind"] == "longrun"],
["join"], "the seed's longrun unit set changed")
path = write_state(state_text(units, seed.BOOT_PHASES))
try:
code, out, err, _ = run_main(path)
still_running = not daemon.finished()
self.assertEqual(code, 0, err)
self.assertIn("numos: boot complete", out)
self.assertTrue(still_running,
"phase 50's longrun unit had ended when numinit "
"exited: the seeded state still blocks the boot")
finally:
daemon.cleanup()
os.unlink(path)
def test_dry_run_reports_every_kind(self):
path = write_state(state_text(
[unit("one", kind="oneshot"),
unit("daemon", kind="longrun", exec_="sleep 60"),
unit("grouped", kind="target", requires=["daemon"])],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["one", "grouped"]}]))
try:
code, out, err, _ = run_main(path, dry_run=True)
self.assertEqual(code, 0, err)
self.assertEqual(sorted(l for l in out.split("\n") if l.startswith("RUN ")),
["RUN daemon", "RUN grouped", "RUN one"])
finally:
os.unlink(path)
class TestOneshotStaysSynchronous(unittest.TestCase):
def test_oneshot_units_complete_before_the_next_phase_starts(self):
marker = temp_path(".marker")
path = write_state(state_text(
[unit("slow", exec_="sleep 2; echo first >> '%s'" % marker),
unit("fast", exec_="echo second >> '%s'" % marker)],
[{"ordinal": 10, "name": "one", "on_failure": "halt",
"required_units": ["slow"]},
{"ordinal": 20, "name": "two", "on_failure": "halt",
"required_units": ["fast"]}]))
try:
code, out, err, elapsed = run_main(path)
self.assertEqual(code, 0, err)
self.assertIn("numos: boot complete", out)
with open(marker) as handle:
self.assertEqual(handle.read().split(), ["first", "second"])
self.assertGreaterEqual(
elapsed, 2.0,
"a oneshot unit was backgrounded: boot returned before it finished")
finally:
os.unlink(path)
try_unlink(marker)
def test_oneshot_failure_still_halts_the_boot(self):
path = write_state(state_text(
[unit("boom", exec_="false")],
[{"ordinal": 10, "name": "critical", "on_failure": "halt",
"required_units": ["boom"]}]))
try:
code, _, err, _ = run_main(path)
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
self.assertIn("critical", err)
finally:
os.unlink(path)
def test_target_kind_runs_synchronously_and_propagates_failure(self):
path = write_state(state_text(
[unit("grouped", kind="target", exec_="false")],
[{"ordinal": 10, "name": "critical", "on_failure": "halt",
"required_units": ["grouped"]}]))
try:
code, _, err, _ = run_main(path)
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
finally:
os.unlink(path)
class TestUnknownKindFailsClosed(unittest.TestCase):
def test_unknown_kind_halts_with_a_named_reason(self):
# numinit is PID 1: it validates kind itself rather than trusting
# that the state came through numos/validate.py's KINDS enum.
path = write_state(state_text(
[unit("weird", kind="perpetual")],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["weird"]}]))
try:
code, _, err, _ = run_main(path)
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
self.assertIn("perpetual", err)
finally:
os.unlink(path)
if __name__ == "__main__":
unittest.main()