tests/test_numinit_order.py

back to source

import os
import subprocess
import tempfile
import unittest

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, requires=None, after=None, exec_="true"):
    return {"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_}


def state_file(units, phases):
    text = render({"version": 1, "arch_targets": [], "units": units,
                   "boot_phases": 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):
    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_DRY_RUN"] = "1"
    proc = subprocess.run(["bash", "-c", full], capture_output=True, text=True, env=env)
    return proc.returncode, proc.stdout.strip(), proc.stderr.strip()


class TestPhaseOrdering(unittest.TestCase):
    def test_phases_run_in_ordinal_order_not_file_order(self):
        path = state_file(
            [unit("a"), unit("b")],
            [{"ordinal": 30, "name": "late", "on_failure": "halt", "required_units": ["b"]},
             {"ordinal": 10, "name": "early", "on_failure": "halt", "required_units": ["a"]}])
        try:
            _, out, _ = sh("numos_phase_ordinals", path)
            self.assertEqual(out.split(), ["10", "30"])
        finally:
            os.unlink(path)


class TestDagResolution(unittest.TestCase):
    def test_requires_edge_orders_dependency_first(self):
        path = state_file([unit("a"), unit("b", requires=["a"])],
                          [{"ordinal": 10, "name": "p", "on_failure": "halt",
                            "required_units": ["b"]}])
        try:
            _, out, _ = sh("numos_resolve_order b", path)
            self.assertEqual(out.split(), ["a", "b"])
        finally:
            os.unlink(path)

    def test_after_edge_orders_without_pulling_in_failure(self):
        path = state_file([unit("a"), unit("b", after=["a"])],
                          [{"ordinal": 10, "name": "p", "on_failure": "halt",
                            "required_units": ["b"]}])
        try:
            _, out, _ = sh("numos_resolve_order b", path)
            self.assertEqual(out.split(), ["a", "b"])
        finally:
            os.unlink(path)

    def test_transitive_chain_fully_ordered(self):
        path = state_file(
            [unit("a"), unit("b", requires=["a"]), unit("c", requires=["b"])],
            [{"ordinal": 10, "name": "p", "on_failure": "halt", "required_units": ["c"]}])
        try:
            _, out, _ = sh("numos_resolve_order c", path)
            self.assertEqual(out.split(), ["a", "b", "c"])
        finally:
            os.unlink(path)

    def test_diamond_emits_each_unit_once(self):
        path = state_file(
            [unit("a"), unit("b", requires=["a"]), unit("c", requires=["a"]),
             unit("d", requires=["b", "c"])],
            [{"ordinal": 10, "name": "p", "on_failure": "halt", "required_units": ["d"]}])
        try:
            _, out, _ = sh("numos_resolve_order d", path)
            names = out.split()
            self.assertEqual(len(names), 4)
            self.assertEqual(names[0], "a")
            self.assertEqual(names[-1], "d")
        finally:
            os.unlink(path)


class TestUnitFieldExtraction(unittest.TestCase):
    def test_exec_with_spaces_extracted_whole(self):
        path = state_file([unit("a", exec_="mount -t proc proc /proc")],
                          [{"ordinal": 10, "name": "p", "on_failure": "halt",
                            "required_units": ["a"]}])
        try:
            _, out, _ = sh("numos_unit_field a exec", path)
            self.assertEqual(out, "mount -t proc proc /proc")
        finally:
            os.unlink(path)


class TestDryRun(unittest.TestCase):
    def test_phase_reports_each_unit_in_order(self):
        path = state_file([unit("a"), unit("b", requires=["a"])],
                          [{"ordinal": 10, "name": "p", "on_failure": "halt",
                            "required_units": ["b"]}])
        try:
            _, out, _ = sh("numos_run_phase 10", path)
            self.assertEqual([l for l in out.split("\n") if l.startswith("RUN ")],
                             ["RUN a", "RUN b"])
        finally:
            os.unlink(path)


class TestFailClosedHalting(unittest.TestCase):
    # These halts are all reached through a command substitution
    # ($(numos_resolve_order ...)) inside numos_run_phase. `exit` inside a
    # substitution only kills the subshell, so a caller that never checks the
    # substitution's status would carry on and return 0 -- a silent no-op
    # instead of a halt. Assert on exit status, not just stderr content: a
    # broken (status-blind) implementation prints the same HALT text.
    def test_dependency_cycle_halts_nonzero_through_run_phase(self):
        path = state_file(
            [unit("a", requires=["b"]), unit("b", requires=["a"])],
            [{"ordinal": 10, "name": "p", "on_failure": "halt", "required_units": ["a"]}])
        try:
            rc, _, err = sh("numos_run_phase 10", path)
            self.assertNotEqual(rc, 0)
            self.assertIn("numos: HALT:", err)
        finally:
            os.unlink(path)

    def test_dangling_requires_halts_nonzero_through_run_phase(self):
        path = state_file(
            [unit("a", requires=["ghost"])],
            [{"ordinal": 10, "name": "p", "on_failure": "halt", "required_units": ["a"]}])
        try:
            rc, _, err = sh("numos_run_phase 10", path)
            self.assertNotEqual(rc, 0)
            self.assertIn("numos: HALT:", err)
        finally:
            os.unlink(path)


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