tests/test_resource_envelope.py
back to source
"""Resource envelopes: enforce what the floor can, refuse what it cannot.
`OSUnit.resource_envelope` sat in the schema, unenforced by anything, and
docs/PRINCIPLES.md named it as the clearest outstanding gap against
stewardship. Closing it exposes an honest split:
ram_mb enforceable here, via ulimit -v
timeout_s enforceable here, via timeout(1)
cpu_pct NOT enforceable by a POSIX shell. A percentage needs cgroups;
ulimit -t is cumulative CPU seconds, a different quantity.
Declaring a limit nothing applies is a lie told to whoever wrote the state,
so an envelope the floor cannot honour halts rather than running unbounded.
"""
import os
import shutil
import subprocess
import tempfile
import unittest
from numos.state import render, parse, verify
from numos.validate import validate, ValidationError
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")
def unit(name, exec_="true", envelope=None):
u = {"name": name, "kind": "oneshot", "restart": "never",
"backoff_ms": 0, "backoff_max_ms": 0, "arch_mask": [],
"health_probe": None, "requires": [], "after": [], "exec": exec_}
if envelope is not None:
u["resource_envelope"] = envelope
return u
def state(units, phases=None):
return {"version": 1, "arch_targets": [], "units": units,
"boot_phases": phases or [], "health": []}
class TestFormat(unittest.TestCase):
def test_a_unit_without_an_envelope_emits_no_r_record(self):
text = render(state([unit("a")]))
self.assertNotIn("\nR ", text)
def test_an_envelope_renders_an_r_record(self):
text = render(state([unit("a", envelope={"ram_mb": "64"})]))
self.assertIn("R a - 64 -", text)
def test_all_three_fields_render_in_order(self):
text = render(state([unit("a", envelope={
"cpu_pct": "50", "ram_mb": "64", "timeout_s": "30"})]))
self.assertIn("R a 50 64 30", text)
def test_round_trip_preserves_the_envelope(self):
original = state([unit("a", envelope={
"cpu_pct": None, "ram_mb": "64", "timeout_s": None})])
self.assertEqual(parse(render(original)), original)
def test_a_unit_without_an_envelope_round_trips_without_growing_the_key(self):
original = state([unit("a")])
self.assertEqual(parse(render(original)), original)
self.assertNotIn("resource_envelope", parse(render(original))["units"][0])
def test_the_envelope_is_covered_by_the_content_hash(self):
text = render(state([unit("a", envelope={"ram_mb": "64"})]))
verify(text)
tampered = text.replace("R a - 64 -", "R a - 99999 -")
with self.assertRaises(Exception):
verify(tampered)
def test_an_r_record_for_an_unknown_unit_is_rejected(self):
text = render(state([unit("a", envelope={"ram_mb": "64"})]))
bad = text.replace("R a ", "R ghost ")
with self.assertRaises(Exception):
parse(bad)
class TestValidation(unittest.TestCase):
def test_a_valid_envelope_passes(self):
validate(state([unit("a", envelope={"ram_mb": "64", "timeout_s": "30"})]))
def test_a_non_numeric_value_is_rejected_by_name(self):
with self.assertRaises(ValidationError) as ctx:
validate(state([unit("a", envelope={"ram_mb": "lots"})]))
self.assertIn("ram_mb", str(ctx.exception))
self.assertIn("a", str(ctx.exception))
def test_a_cpu_percentage_above_100_is_rejected(self):
with self.assertRaises(ValidationError) as ctx:
validate(state([unit("a", envelope={"cpu_pct": "150"})]))
self.assertIn("cpu_pct", str(ctx.exception))
def test_a_null_field_means_unlimited_and_is_allowed(self):
validate(state([unit("a", envelope={
"cpu_pct": None, "ram_mb": None, "timeout_s": None})]))
class ShellCase(unittest.TestCase):
def setUp(self):
self.work = tempfile.mkdtemp()
def tearDown(self):
shutil.rmtree(self.work, ignore_errors=True)
def run_init(self, units, phases, ticks=1):
text = render(state(units, phases))
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": os.path.join(self.work, "run").replace("\\", "/"),
"NUMOS_MAX_TICKS": str(ticks),
"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
class TestShellEnforcement(ShellCase):
def test_a_unit_with_no_envelope_runs_normally(self):
code, out, err = self.run_init(
[unit("a", exec_="echo RAN_A")],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["a"]}])
self.assertEqual(code, 0, err)
self.assertIn("RAN_A", out)
def test_cpu_pct_is_refused_rather_than_silently_ignored(self):
"""The whole point. An operator who sets cpu_pct expecting protection
must not get a unit that runs unbounded and a boot that says nothing."""
code, out, err = self.run_init(
[unit("a", exec_="echo SHOULD_NOT_RUN",
envelope={"cpu_pct": "50"})],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["a"]}])
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
self.assertIn("cpu_pct", err)
self.assertIn("cgroups", err)
self.assertNotIn("SHOULD_NOT_RUN", out,
"the unit ran despite an unenforceable limit")
def test_a_timeout_kills_a_unit_that_overruns(self):
code, out, err = self.run_init(
[unit("slow", exec_="sleep 30; echo FINISHED",
envelope={"timeout_s": "1"})],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["slow"]}])
self.assertNotEqual(code, 0, "an overrunning unit was not stopped")
self.assertNotIn("FINISHED", out)
def test_a_unit_within_its_timeout_succeeds(self):
code, out, err = self.run_init(
[unit("quick", exec_="echo QUICK_OK",
envelope={"timeout_s": "30"})],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["quick"]}])
self.assertEqual(code, 0, err)
self.assertIn("QUICK_OK", out)
def test_cpu_pct_on_a_longrun_unit_also_halts(self):
"""The envelope must be resolved in the PARENT shell.
A longrun unit is backgrounded, and numos_die inside a background job
kills only that job - the boot would print the halt and carry on,
running the unit with no limit. Same swallowed-exit class this
codebase has hit repeatedly, so it gets its own test rather than
relying on the oneshot case.
"""
code, out, err = self.run_init(
[{"name": "d", "kind": "longrun", "restart": "never",
"backoff_ms": 0, "backoff_max_ms": 0, "arch_mask": [],
"health_probe": None, "requires": [], "after": [],
"exec": "sleep 30", "resource_envelope": {"cpu_pct": "25"}}],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["d"]}])
self.assertNotEqual(code, 0,
"a backgrounded unit's unenforceable limit did not "
"stop the boot")
self.assertIn("cpu_pct", err)
self.assertNotIn("boot complete", out)
def test_a_ram_limit_is_actually_applied_to_the_unit(self):
"""Assert the limit reaches the unit's own process, not just that the
boot survived: the unit reports the ulimit it is running under."""
code, out, err = self.run_init(
[unit("mem", exec_="ulimit -v", envelope={"ram_mb": "64"})],
[{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["mem"]}])
self.assertEqual(code, 0, err)
self.assertIn("65536", out,
"ram_mb=64 should surface as a 65536 KB address-space "
"limit inside the unit; got: %s" % out)
if __name__ == "__main__":
unittest.main()
class TestRestartPathEnvelope(ShellCase):
"""What the restart path can actually be held to.
A review flagged as CRITICAL that numos_restart_unit backgrounds
numos_exec_unit without the parent-shell numos_resolve_envelope call its
sibling numos_run_unit performs -- so an unenforceable cpu_pct check would
run inside the background job and halt nothing.
Tracing it, that is UNREACHABLE. A longrun unit declaring cpu_pct dies at
INITIAL start, because numos_run_unit already resolves in the parent, so it
never survives to be restarted. The first test written for it passed with
the fix reverted -- it was exercising the initial-start path and proving
nothing. The severity was overstated and the test was vacuous; both are
recorded here rather than quietly dropped.
The parent-shell resolve on the restart path is kept as a defensive
symmetry with its sibling, not as a fix for a live defect. What IS
reachable, and is asserted below, is that a unit restarted by the
supervisor still has its enforceable envelope applied.
"""
def test_a_restarted_unit_still_gets_its_ram_limit(self):
marker = os.path.join(self.work, "limits").replace("\\", "/")
code, out, err = self.run_init(
[{"name": "recycler", "kind": "longrun", "restart": "always",
"backoff_ms": 1, "backoff_max_ms": 2, "arch_mask": [],
"health_probe": None, "requires": [], "after": [],
"exec": "ulimit -v >> %s; exit 0" % marker,
"resource_envelope": {"ram_mb": "64"}}],
[{"ordinal": 10, "name": "p", "on_failure": "continue",
"required_units": ["recycler"]}],
ticks=4)
self.assertEqual(code, 0, err)
with open(marker) as handle:
seen = [line.strip() for line in handle if line.strip()]
self.assertGreater(len(seen), 1,
"the unit was never restarted, so this asserts "
"nothing about the restart path: %s" % out)
for value in seen:
self.assertEqual(value, "65536",
"a restarted unit ran without its ram_mb limit")