tests/test_numctl.py

back to source

"""Floor numctl: seeded unit execs must not name a vapor binary.

identity-init and join are the only implemented commands. Everything else
halts with a named reason. Control-socket query/command remains residual.
"""

import os
import shutil
import subprocess
import tempfile
import time
import unittest

from numos.state import render

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
NUMCTL = os.path.join(ROOT, "boot", "numctl").replace("\\", "/")
NUMINIT = os.path.join(ROOT, "boot", "numinit.sh").replace("\\", "/")
BOOT = os.path.join(ROOT, "boot").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")


class NumctlCase(unittest.TestCase):
    def setUp(self):
        self.work = tempfile.mkdtemp()
        self.rundir = os.path.join(self.work, "run").replace("\\", "/")

    def tearDown(self):
        shutil.rmtree(self.work, ignore_errors=True)

    def run_numctl(self, *args, timeout=30):
        env = dict(os.environ)
        env["NUMOS_RUNDIR"] = self.rundir
        # boot/ on PATH so bare `numctl` resolves the way seed units expect.
        env["PATH"] = BOOT + os.pathsep + env.get("PATH", "")
        return subprocess.run(
            [_bash(), NUMCTL] + list(args),
            capture_output=True, text=True, env=env, timeout=timeout)

    def test_identity_init_writes_an_identity_file(self):
        proc = self.run_numctl("identity-init")
        self.assertEqual(proc.returncode, 0, proc.stderr)
        path = os.path.join(self.work, "run", "identity")
        self.assertTrue(os.path.isfile(path), "identity file was not written")
        with open(path) as handle:
            first = handle.read().strip()
        self.assertTrue(first, "identity file is empty")
        # Idempotent: second call must not rotate the identity.
        proc2 = self.run_numctl("identity-init")
        self.assertEqual(proc2.returncode, 0, proc2.stderr)
        with open(path) as handle:
            second = handle.read().strip()
        self.assertEqual(first, second)

    def test_unknown_command_halts_with_named_reason(self):
        proc = self.run_numctl("explode")
        self.assertNotEqual(proc.returncode, 0)
        self.assertIn("numctl: HALT:", proc.stderr)
        self.assertIn("unknown command", proc.stderr)

    def test_status_halts_because_control_socket_is_absent(self):
        proc = self.run_numctl("status")
        self.assertNotEqual(proc.returncode, 0)
        self.assertIn("numctl: HALT:", proc.stderr)
        self.assertIn("control socket not implemented", proc.stderr)

    def test_run_op_without_state_halts_named(self):
        # Bare PATH invoke has no NUMOS_STATE: refuse before inventing work.
        proc = self.run_numctl("run-op", "SwarmExecutor_v1")
        self.assertNotEqual(proc.returncode, 0)
        self.assertIn("numctl: HALT:", proc.stderr)
        self.assertIn("SwarmExecutor_v1", proc.stderr)
        self.assertIn("NUMOS_STATE", proc.stderr)

    def test_join_stays_alive_as_a_longrun_floor(self):
        env = dict(os.environ)
        env["NUMOS_RUNDIR"] = self.rundir
        env["PATH"] = BOOT + os.pathsep + env.get("PATH", "")
        proc = subprocess.Popen(
            [_bash(), NUMCTL, "join"],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            env=env)
        try:
            time.sleep(0.3)
            self.assertIsNone(proc.poll(), "join floor exited immediately")
        finally:
            proc.terminate()
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()
                proc.wait(timeout=5)

    def test_seeded_identity_unit_runs_against_the_floor(self):
        """The seed's real exec string must work when numctl is on PATH."""
        text = render({
            "version": 1,
            "arch_targets": [],
            "units": [{
                "name": "identity", "kind": "oneshot", "restart": "never",
                "backoff_ms": 0, "backoff_max_ms": 0, "arch_mask": [],
                "health_probe": None, "requires": [], "after": [],
                "exec": "numctl identity-init",
            }],
            "boot_phases": [{
                "ordinal": 10, "name": "identity", "on_failure": "halt",
                "required_units": ["identity"],
            }],
            "health": [],
        })
        state = os.path.join(self.work, "numos.state").replace("\\", "/")
        with open(state, "w", newline="\n") as handle:
            handle.write(text)
        env = dict(os.environ)
        env["NUMOS_STATE"] = state
        env["NUMOS_RUNDIR"] = self.rundir
        env["NUMOS_MAX_TICKS"] = "1"
        env["NUMOS_NO_SLEEP"] = "1"
        env["PATH"] = BOOT + os.pathsep + env.get("PATH", "")
        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=60)
        self.assertEqual(proc.returncode, 0, proc.stderr)
        self.assertIn("boot complete", proc.stdout)
        self.assertTrue(
            os.path.isfile(os.path.join(self.work, "run", "identity")),
            "seeded identity unit did not create the identity file")


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