tests/test_validate.py
back to source
import unittest
from numos.state import render, parse
from numos.validate import validate, find_cycle, ValidationError
def unit(name, requires=None, after=None, exec_="true", **overrides):
record = {"name": name, "kind": "oneshot", "restart": "never",
"backoff_ms": 0, "backoff_max_ms": 0, "arch_mask": [],
"health_probe": None, "requires": requires or [],
"after": after or [], "exec": exec_}
record.update(overrides)
return record
def state(units=None, phases=None, health=None):
return {"version": 1, "arch_targets": [], "units": units or [],
"boot_phases": phases or [], "health": health or []}
class TestFindCycle(unittest.TestCase):
def test_acyclic_returns_none(self):
self.assertIsNone(find_cycle([unit("a"), unit("b", requires=["a"])]))
def test_direct_cycle_detected(self):
cycle = find_cycle([unit("a", requires=["b"]), unit("b", requires=["a"])])
self.assertIsNotNone(cycle)
self.assertIn("a", cycle)
self.assertIn("b", cycle)
def test_cycle_through_after_edges_detected(self):
self.assertIsNotNone(find_cycle([unit("a", after=["b"]), unit("b", after=["a"])]))
def test_self_cycle_detected(self):
self.assertIsNotNone(find_cycle([unit("a", requires=["a"])]))
class TestValidate(unittest.TestCase):
def test_valid_state_passes(self):
validate(state(units=[unit("a")], phases=[
{"ordinal": 10, "name": "mount", "on_failure": "halt", "required_units": ["a"]}]))
def test_cycle_rejected_at_export_not_boot(self):
with self.assertRaises(ValidationError) as ctx:
validate(state(units=[unit("a", requires=["b"]), unit("b", requires=["a"])]))
self.assertIn("cycle", str(ctx.exception))
def test_duplicate_ordinal_rejected(self):
with self.assertRaises(ValidationError) as ctx:
validate(state(phases=[
{"ordinal": 10, "name": "mount", "on_failure": "halt", "required_units": []},
{"ordinal": 10, "name": "net", "on_failure": "halt", "required_units": []}]))
self.assertIn("ordinal", str(ctx.exception))
def test_phase_referencing_unknown_unit_rejected(self):
with self.assertRaises(ValidationError) as ctx:
validate(state(phases=[
{"ordinal": 10, "name": "mount", "on_failure": "halt",
"required_units": ["ghost"]}]))
self.assertIn("ghost", str(ctx.exception))
def test_unit_requiring_unknown_unit_rejected(self):
with self.assertRaises(ValidationError) as ctx:
validate(state(units=[unit("a", requires=["ghost"])]))
self.assertIn("ghost", str(ctx.exception))
def test_duplicate_unit_name_rejected(self):
with self.assertRaises(ValidationError) as ctx:
validate(state(units=[unit("a"), unit("a")]))
self.assertIn("duplicate", str(ctx.exception))
def test_bad_on_failure_rejected(self):
with self.assertRaises(ValidationError):
validate(state(phases=[
{"ordinal": 10, "name": "m", "on_failure": "explode", "required_units": []}]))
def test_bad_kind_rejected(self):
with self.assertRaises(ValidationError) as ctx:
bad_unit = {"name": "badkind", "kind": "invalid", "restart": "never",
"backoff_ms": 0, "backoff_max_ms": 0, "arch_mask": [],
"health_probe": None, "requires": [], "after": [], "exec": "true"}
validate(state(units=[bad_unit]))
exc_str = str(ctx.exception)
self.assertIn("badkind", exc_str)
self.assertIn("invalid", exc_str)
def test_bad_restart_rejected(self):
with self.assertRaises(ValidationError) as ctx:
bad_unit = {"name": "badrestart", "kind": "oneshot", "restart": "invalid",
"backoff_ms": 0, "backoff_max_ms": 0, "arch_mask": [],
"health_probe": None, "requires": [], "after": [], "exec": "true"}
validate(state(units=[bad_unit]))
exc_str = str(ctx.exception)
self.assertIn("badrestart", exc_str)
self.assertIn("invalid", exc_str)
class TestFieldSeparators(unittest.TestCase):
"""A space in any field but the last one shifts the whole record.
The shifted record is still hash-valid and passes both verifiers, so
export-time validation is the only place it can be caught.
"""
def test_unit_name_with_a_space_is_rejected(self):
with self.assertRaises(ValidationError) as ctx:
validate(state(units=[unit("op-bad op-v1")]))
message = str(ctx.exception)
self.assertIn("op-bad op-v1", message)
self.assertIn("name", message)
self.assertIn("space", message)
def test_the_shifted_record_would_otherwise_reach_numinit(self):
"""Show what validate() is actually preventing, not just that it raises.
Rendered without the guard, the U record's fields all move left by
one: numinit would read kind=op-v1 and restart=longrun, and parse()
would die on int('on-failure') with a raw ValueError.
"""
bad = unit("op-bad op-v1", kind="longrun", restart="on-failure")
line = [ln for ln in render({"version": 1, "arch_targets": [],
"boot_phases": [], "units": [bad],
"health": []}).split("\n")
if ln.startswith("U ")][0]
self.assertEqual(line.split(" ")[2], "op-v1") # would be read as kind
self.assertEqual(line.split(" ")[3], "longrun") # would be read as restart
with self.assertRaises(ValueError):
parse(render({"version": 1, "arch_targets": [], "boot_phases": [],
"units": [bad], "health": []}))
with self.assertRaises(ValidationError):
validate(state(units=[bad]))
def test_unit_kind_with_a_space_is_rejected(self):
with self.assertRaises(ValidationError) as ctx:
validate(state(units=[unit("a", kind="one shot")]))
self.assertIn("kind", str(ctx.exception))
def test_requires_element_with_a_space_is_rejected(self):
with self.assertRaises(ValidationError) as ctx:
validate(state(units=[unit("a"), unit("b", requires=["a b"])]))
self.assertIn("requires", str(ctx.exception))
def test_requires_element_with_a_comma_is_rejected(self):
with self.assertRaises(ValidationError) as ctx:
validate(state(units=[unit("a"), unit("b", requires=["a,b"])]))
self.assertIn("comma", str(ctx.exception))
def test_requires_element_equal_to_the_empty_marker_is_rejected(self):
with self.assertRaises(ValidationError) as ctx:
validate(state(units=[unit("a", requires=["-"])]))
self.assertIn("marker", str(ctx.exception))
def test_phase_name_with_a_space_is_rejected(self):
with self.assertRaises(ValidationError) as ctx:
validate(state(phases=[{"ordinal": 10, "name": "early mount",
"on_failure": "halt", "required_units": []}]))
self.assertIn("early mount", str(ctx.exception))
def test_exec_may_contain_spaces_because_it_is_last_on_the_line(self):
validate(state(units=[unit("a", exec_="mount -t proc proc /proc")]))
def test_exec_may_not_contain_a_newline(self):
with self.assertRaises(ValidationError) as ctx:
validate(state(units=[unit("a", exec_="true\nU forged oneshot")]))
self.assertIn("newline", str(ctx.exception))
def test_empty_exec_is_rejected(self):
with self.assertRaises(ValidationError) as ctx:
validate(state(units=[unit("a", exec_="")]))
self.assertIn("exec", str(ctx.exception))
def test_health_probe_may_contain_spaces_and_quotes(self):
validate(state(health=[{"name": "disk", "interval_s": 30, "threshold": 3,
"local_action": "degrade-node", "on_fire": None,
"probe": "df -P / | awk 'NR==2 && $5+0 < 95'"}]))
def test_arch_alias_with_a_space_is_rejected(self):
bad = {"canonical": "x86_64", "aliases": ["x86_64", "amd 64"],
"zig_triple": None, "busybox_variant": "busybox-x86_64",
"artifact_sha256": None}
with self.assertRaises(ValidationError) as ctx:
validate({"version": 1, "arch_targets": [bad], "units": [],
"boot_phases": [], "health": []})
self.assertIn("aliases", str(ctx.exception))
if __name__ == "__main__":
unittest.main()