tests/test_numinit_backoff.py
back to source
import os
import subprocess
import tempfile
import unittest
from numos.canonical import content_hash
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 unit(name, restart, exec_="true", backoff_ms=100, backoff_max_ms=800):
return {"name": name, "kind": "longrun", "restart": restart,
"backoff_ms": backoff_ms, "backoff_max_ms": backoff_max_ms,
"arch_mask": [], "health_probe": None, "requires": [],
"after": [], "exec": exec_}
def state_file(units):
text = render({"version": 1, "arch_targets": [], "units": units,
"boot_phases": [], "health": []})
handle = tempfile.NamedTemporaryFile("w", suffix=".numos", delete=False, newline="\n")
handle.write(text)
handle.close()
return handle.name.replace("\\", "/")
def sh(snippet, state_path, extra_env=None):
full = 'NUMOS_SOURCE_ONLY=1 . "%s"; numos_load_state "%s"; %s' % (
NUMINIT, state_path, snippet)
env = dict(os.environ)
env["NUMOS_SOURCE_ONLY"] = "1"
env["NUMOS_NO_SLEEP"] = "1"
if extra_env:
env.update(extra_env)
proc = subprocess.run(["bash", "-c", full], capture_output=True, text=True, env=env)
return proc.returncode, proc.stdout.strip(), proc.stderr.strip()
def flaky_exec(marker_path):
"""Shell snippet that fails on its first invocation, then succeeds on
every later one (a marker file records that the first call happened)."""
return 'sh -c \'test -f "%s" && exit 0 || { touch "%s"; exit 1; }\'' % (
marker_path, marker_path)
def raw_state_with_backoff(name, backoff_ms, backoff_max_ms=800,
restart="on-failure", exec_="true"):
"""Hand-build a .numos U record with a literal backoff_ms token,
bypassing numos.state.render()'s "%d" formatting -- render() requires
backoff_ms to already be an int, so a malformed (non-numeric) value
can only be exercised by writing the line ourselves. numos_unit_field
reads this text with grep/cut, never through numos.state.parse(), so
the shell's own validation is what has to catch it."""
u_line = "U %s longrun %s %s %s - - - - %s" % (
name, restart, backoff_ms, backoff_max_ms, exec_)
lines = ["V 1", u_line]
stamped = ["V 1", "C " + content_hash(lines), u_line]
text = "".join(ln + "\n" for ln in stamped)
handle = tempfile.NamedTemporaryFile("w", suffix=".numos", delete=False, newline="\n")
handle.write(text)
handle.close()
return handle.name.replace("\\", "/")
class TestBackoffSequence(unittest.TestCase):
def test_doubles_until_ceiling_then_holds(self):
path = state_file([unit("a", "always")])
try:
_, out, _ = sh("numos_backoff_sequence 100 800 6", path)
self.assertEqual(out.split(), ["100", "200", "400", "800", "800", "800"])
finally:
os.unlink(path)
def test_initial_above_ceiling_is_clamped(self):
path = state_file([unit("a", "always")])
try:
_, out, _ = sh("numos_backoff_sequence 5000 800 3", path)
self.assertEqual(out.split(), ["800", "800", "800"])
finally:
os.unlink(path)
class TestSupervise(unittest.TestCase):
def test_restart_never_runs_once_even_on_failure(self):
path = state_file([unit("a", "never", exec_="false")])
try:
_, out, _ = sh("numos_supervise a 5", path)
self.assertEqual([l for l in out.split("\n") if l.startswith("ATTEMPT")],
["ATTEMPT 1"])
finally:
os.unlink(path)
def test_restart_on_failure_retries_until_max_attempts(self):
path = state_file([unit("a", "on-failure", exec_="false")])
try:
_, out, _ = sh("numos_supervise a 3", path)
self.assertEqual([l for l in out.split("\n") if l.startswith("ATTEMPT")],
["ATTEMPT 1", "ATTEMPT 2", "ATTEMPT 3"])
finally:
os.unlink(path)
def test_restart_on_failure_stops_after_success(self):
path = state_file([unit("a", "on-failure", exec_="true")])
try:
_, out, _ = sh("numos_supervise a 3", path)
self.assertEqual([l for l in out.split("\n") if l.startswith("ATTEMPT")],
["ATTEMPT 1"])
finally:
os.unlink(path)
def test_restart_always_retries_even_after_success(self):
path = state_file([unit("a", "always", exec_="true")])
try:
_, out, _ = sh("numos_supervise a 3", path)
self.assertEqual([l for l in out.split("\n") if l.startswith("ATTEMPT")],
["ATTEMPT 1", "ATTEMPT 2", "ATTEMPT 3"])
finally:
os.unlink(path)
class TestSuperviseReturnCode(unittest.TestCase):
"""numos_supervise's return code must reflect whether the unit is
actually up when supervision stops, not just how many ATTEMPT lines
were printed -- a caller cannot act on "gave up after N failures"
if the function claims success (RC=0) regardless of outcome."""
def test_on_failure_exhausts_attempts_returns_nonzero(self):
path = state_file([unit("a", "on-failure", exec_="false")])
try:
code, _, _ = sh("numos_supervise a 3", path)
self.assertNotEqual(code, 0)
finally:
os.unlink(path)
def test_on_failure_succeeds_on_later_attempt_returns_zero(self):
marker = tempfile.NamedTemporaryFile(delete=False)
marker.close()
os.unlink(marker.name)
marker_path = marker.name.replace("\\", "/")
path = state_file([unit("a", "on-failure", exec_=flaky_exec(marker_path))])
try:
code, out, _ = sh("numos_supervise a 3", path)
self.assertEqual(code, 0)
self.assertEqual([l for l in out.split("\n") if l.startswith("ATTEMPT")],
["ATTEMPT 1", "ATTEMPT 2"])
finally:
os.unlink(path)
if os.path.exists(marker_path):
os.unlink(marker_path)
def test_never_with_failing_unit_returns_nonzero(self):
path = state_file([unit("a", "never", exec_="false")])
try:
code, _, _ = sh("numos_supervise a 5", path)
self.assertNotEqual(code, 0)
finally:
os.unlink(path)
def test_always_exhausts_cap_after_success_returns_zero(self):
# Every attempt succeeds, but "always" keeps retrying regardless of
# success until the (test-imposed) attempt cap forces a stop. Our
# documented decision: report the state the unit was left in --
# since the last attempt succeeded, the unit is up, so RC=0.
path = state_file([unit("a", "always", exec_="true")])
try:
code, _, _ = sh("numos_supervise a 3", path)
self.assertEqual(code, 0)
finally:
os.unlink(path)
def test_always_exhausts_cap_all_failing_returns_nonzero(self):
# Complement of the case above: every attempt fails, "always" keeps
# retrying anyway, and the cap forces a stop. The unit was never up
# at any point, so RC must be nonzero -- this is the scenario that
# actually discriminates the fix from the unconditional "return 0"
# bug (the "after a success" case above returns 0 either way).
path = state_file([unit("a", "always", exec_="false")])
try:
code, _, _ = sh("numos_supervise a 3", path)
self.assertNotEqual(code, 0)
finally:
os.unlink(path)
class TestBackoffValidation(unittest.TestCase):
def test_non_numeric_backoff_ms_halts_cleanly(self):
path = raw_state_with_backoff("a", "notanumber")
try:
code, _, err = sh("numos_supervise a 3", path)
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
finally:
os.unlink(path)
def test_non_numeric_backoff_max_ms_halts_cleanly(self):
path = raw_state_with_backoff("a", 100, backoff_max_ms="also-not-a-number")
try:
code, _, err = sh("numos_supervise a 3", path)
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
finally:
os.unlink(path)
class TestSleepArgumentNeverZero(unittest.TestCase):
def test_sub_second_delay_computes_nonzero_argument(self):
path = state_file([unit("a", "always")])
try:
_, out, _ = sh("numos_sleep_arg 100", path)
self.assertNotEqual(out.strip(), "0")
self.assertEqual(out.strip(), "0.100")
finally:
os.unlink(path)
def test_exact_second_delay_computes_nonzero_argument(self):
path = state_file([unit("a", "always")])
try:
_, out, _ = sh("numos_sleep_arg 1000", path)
self.assertNotEqual(out.strip(), "0")
self.assertEqual(out.strip(), "1.000")
finally:
os.unlink(path)
def test_zero_delay_computes_zero_argument(self):
# Only a nonzero delay must never collapse to zero; a genuine
# zero-length backoff is not a throttling bug.
path = state_file([unit("a", "always")])
try:
_, out, _ = sh("numos_sleep_arg 0", path)
self.assertEqual(out.strip(), "0")
finally:
os.unlink(path)
def test_fallback_to_whole_second_sleep_when_fractional_rejected(self):
# Simulate a sleep(1) that rejects fractional arguments (POSIX only
# guarantees whole seconds) by shadowing it on PATH with a fake that
# exits nonzero for any argument containing "." and exits 0
# immediately otherwise -- so this test verifies the fallback branch
# actually gets taken and succeeds, without waiting on a real delay.
fake_bin_dir = tempfile.mkdtemp()
fake_sleep = os.path.join(fake_bin_dir, "sleep")
with open(fake_sleep, "w", newline="\n") as handle:
handle.write(
"#!/bin/sh\n"
"case \"$1\" in\n"
" *.*) exit 1 ;;\n"
" *) exit 0 ;;\n"
"esac\n")
os.chmod(fake_sleep, 0o755)
path = state_file([unit("a", "always")])
try:
env = dict(os.environ)
env["PATH"] = fake_bin_dir + os.pathsep + env.get("PATH", "")
env["NUMOS_NO_SLEEP"] = "0"
code, _, err = sh("numos_sleep_ms 100", path, extra_env=env)
self.assertEqual(code, 0, err)
finally:
os.unlink(path)
os.unlink(fake_sleep)
os.rmdir(fake_bin_dir)
if __name__ == "__main__":
unittest.main()