tests/test_state.py

back to source

import unittest
from numos.state import render, parse, verify, StateError

SAMPLE = {
    "version": 1,
    "arch_targets": [
        {"canonical": "x86_64", "aliases": ["x86_64", "amd64"],
         "zig_triple": "x86_64-linux-musl", "busybox_variant": "busybox-x86_64",
         "artifact_sha256": "ab" * 32},
        {"canonical": "riscv64", "aliases": ["riscv64"],
         "zig_triple": None, "busybox_variant": "busybox-riscv64",
         "artifact_sha256": None},
    ],
    "boot_phases": [
        {"ordinal": 20, "name": "net", "on_failure": "degrade", "required_units": ["ip-up"]},
        {"ordinal": 10, "name": "mount", "on_failure": "halt", "required_units": ["mount-proc"]},
    ],
    "units": [
        {"name": "mount-proc", "kind": "oneshot", "restart": "never",
         "backoff_ms": 0, "backoff_max_ms": 0, "arch_mask": [], "health_probe": None,
         "requires": [], "after": [], "exec": "mount -t proc proc /proc"},
        {"name": "ip-up", "kind": "oneshot", "restart": "on-failure",
         "backoff_ms": 100, "backoff_max_ms": 800, "arch_mask": [], "health_probe": None,
         "requires": ["mount-proc"], "after": [], "exec": "ip link set eth0 up"},
    ],
    "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 _sorted_state(state):
    """Return a copy of state with all lists sorted in render order."""
    import copy
    s = copy.deepcopy(state)
    s["arch_targets"].sort(key=lambda a: "A %s" % a["canonical"])
    s["boot_phases"].sort(key=lambda p: "P %d" % p["ordinal"])
    s["units"].sort(key=lambda u: "U %s" % u["name"])
    s["health"].sort(key=lambda h: "X %s" % h["name"])
    return s


class TestRender(unittest.TestCase):
    def test_blocks_appear_in_order(self):
        tags = [ln[0] for ln in render(SAMPLE).strip().split("\n")]
        self.assertEqual(tags, sorted(tags, key="VCAPUX".index))

    def test_null_fields_render_as_dash(self):
        line = [ln for ln in render(SAMPLE).split("\n") if ln.startswith("A riscv64")][0]
        self.assertEqual(line, "A riscv64 riscv64 - busybox-riscv64 -")

    def test_empty_lists_render_as_dash(self):
        line = [ln for ln in render(SAMPLE).split("\n") if ln.startswith("U mount-proc")][0]
        self.assertTrue(line.endswith("- - - - mount -t proc proc /proc"))

    def test_records_are_sorted_within_block(self):
        phases = [ln for ln in render(SAMPLE).split("\n") if ln.startswith("P ")]
        self.assertEqual(phases[0], "P 10 mount halt mount-proc")

    def test_render_is_deterministic(self):
        self.assertEqual(render(SAMPLE), render(SAMPLE))

    def test_c_line_is_stamped(self):
        self.assertEqual(len([ln for ln in render(SAMPLE).split("\n") if ln.startswith("C ")]), 1)


class TestRoundTrip(unittest.TestCase):
    def test_parse_inverts_render(self):
        self.assertEqual(parse(render(SAMPLE)), _sorted_state(SAMPLE))

    def test_exec_with_spaces_survives(self):
        unit = [u for u in parse(render(SAMPLE))["units"] if u["name"] == "ip-up"][0]
        self.assertEqual(unit["exec"], "ip link set eth0 up")

    def test_probe_with_spaces_and_quotes_survives(self):
        probe = parse(render(SAMPLE))["health"][0]["probe"]
        self.assertEqual(probe, "df -P / | awk 'NR==2 && $5+0 < 95'")


class TestVerify(unittest.TestCase):
    def test_good_state_verifies(self):
        verify(render(SAMPLE))

    def test_tampered_body_raises(self):
        bad = render(SAMPLE).replace("busybox-riscv64", "busybox-EVIL")
        with self.assertRaises(StateError):
            verify(bad)

    def test_missing_c_line_raises(self):
        bad = "\n".join(ln for ln in render(SAMPLE).split("\n") if not ln.startswith("C "))
        with self.assertRaises(StateError):
            verify(bad)


if __name__ == "__main__":
    unittest.main()