tests/test_numinit_failure.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("\\", "/")


# A later phase's unit prints this. Nothing else in numinit's output
# contains it, so its presence or absence is the only direct evidence of
# whether the later phase actually ran. Asserting on the unit's *name*
# proves nothing: numos_run_unit prints nothing outside dry run, so the
# name can never appear on stdout whether the phase ran or not.
LATER_MARKER = "NUMOS_LATER_PHASE_RAN"


def unit(name, exec_="true"):
    return {"name": name, "kind": "oneshot", "restart": "never",
            "backoff_ms": 0, "backoff_max_ms": 0, "arch_mask": [],
            "health_probe": None, "requires": [], "after": [], "exec": exec_}


def marker_unit(name="later"):
    return unit(name, exec_="echo %s" % LATER_MARKER)


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 run_main(state_path):
    """Run numinit end to end for real (no dry run) so failing units fail."""
    env = dict(os.environ)
    env["NUMOS_STATE"] = state_path
    env.pop("NUMOS_DRY_RUN", None)
    proc = subprocess.run(["bash", NUMINIT], capture_output=True, text=True, env=env)
    return proc.returncode, proc.stdout.strip(), proc.stderr.strip()


class TestHaltMode(unittest.TestCase):
    def test_halt_phase_failure_stops_boot_with_named_reason(self):
        path = state_file([unit("boom", exec_="false"), marker_unit()],
                          [{"ordinal": 10, "name": "critical", "on_failure": "halt",
                            "required_units": ["boom"]},
                           {"ordinal": 20, "name": "after", "on_failure": "continue",
                            "required_units": ["later"]}])
        try:
            code, out, err = run_main(path)
            self.assertNotEqual(code, 0)
            self.assertIn("numos: HALT:", err)
            self.assertIn("critical", err)
            # The halt has to actually stop the walk, not just print.
            self.assertNotIn(LATER_MARKER, out)
            self.assertNotIn("boot complete", out)
        finally:
            os.unlink(path)

    def test_the_marker_appears_when_the_later_phase_does_run(self):
        """Positive control for the assertion above.

        Without this, assertNotIn(LATER_MARKER) would be indistinguishable
        from asserting on a string the program can never emit -- which is
        exactly the defect this test replaced. Same state, same later phase,
        only the first phase's outcome differs.
        """
        path = state_file([unit("fine", exec_="true"), marker_unit()],
                          [{"ordinal": 10, "name": "critical", "on_failure": "halt",
                            "required_units": ["fine"]},
                           {"ordinal": 20, "name": "after", "on_failure": "continue",
                            "required_units": ["later"]}])
        try:
            code, out, err = run_main(path)
            self.assertEqual(code, 0, err)
            self.assertIn(LATER_MARKER, out)
            self.assertIn("boot complete", out)
        finally:
            os.unlink(path)


class TestDegradeMode(unittest.TestCase):
    def test_degrade_continues_boot_and_marks_node(self):
        path = state_file([unit("boom", exec_="false"), unit("later")],
                          [{"ordinal": 10, "name": "net", "on_failure": "degrade",
                            "required_units": ["boom"]},
                           {"ordinal": 20, "name": "after", "on_failure": "continue",
                            "required_units": ["later"]}])
        try:
            code, out, _ = run_main(path)
            self.assertEqual(code, 0)
            self.assertIn("numos: DEGRADED: net", out)
            self.assertIn("numos: boot complete degraded=1", out)
        finally:
            os.unlink(path)


class TestContinueMode(unittest.TestCase):
    def test_continue_logs_but_does_not_degrade(self):
        path = state_file([unit("boom", exec_="false")],
                          [{"ordinal": 10, "name": "clock", "on_failure": "continue",
                            "required_units": ["boom"]}])
        try:
            code, out, _ = run_main(path)
            self.assertEqual(code, 0)
            self.assertIn("numos: WARN: phase clock failed", out)
            self.assertIn("degraded=0", out)
        finally:
            os.unlink(path)


class TestHappyPath(unittest.TestCase):
    def test_all_phases_succeed_reports_not_degraded(self):
        path = state_file([unit("a"), unit("b")],
                          [{"ordinal": 10, "name": "one", "on_failure": "halt",
                            "required_units": ["a"]},
                           {"ordinal": 20, "name": "two", "on_failure": "halt",
                            "required_units": ["b"]}])
        try:
            code, out, _ = run_main(path)
            self.assertEqual(code, 0)
            self.assertIn("numos: boot complete degraded=0", out)
        finally:
            os.unlink(path)


class TestUnknownPolicyHalts(unittest.TestCase):
    def test_unknown_on_failure_value_halts_with_named_reason(self):
        # This state file bypasses numos/validate.py's ON_FAILURE enum check
        # (render() does not validate) to simulate numinit being handed a
        # state file that never passed through the exporter's validation --
        # numinit is PID 1 and must fail closed on its own, not rely on an
        # upstream guarantee. The required unit fails so numos_run_phase
        # returns nonzero and numos_handle_phase_failure's `*)` arm is
        # actually reached.
        path = state_file([unit("boom", exec_="false")],
                          [{"ordinal": 10, "name": "weird", "on_failure": "explode",
                            "required_units": ["boom"]}])
        try:
            code, out, err = run_main(path)
            self.assertNotEqual(code, 0)
            self.assertIn("numos: HALT:", err)
            self.assertIn("weird", err)
        finally:
            os.unlink(path)


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