tests/test_export.py

back to source

import json
import os
import shutil
import tempfile
import unittest

from numos.export_state import build_state, export
from numos.state import parse, verify
from numos.validate import validate, ValidationError

OPS = ["SwarmExecutor_v1", "ObserverChain_v1", "MetaGraphSync_v1"]


class TestBuildState(unittest.TestCase):
    def test_every_op_becomes_a_unit(self):
        names = [u["name"] for u in build_state(OPS)["units"]]
        for op in OPS:
            self.assertIn("op-" + op.lower().replace("_", "-"), names)

    def test_infra_units_are_present_alongside_ops(self):
        names = [u["name"] for u in build_state(OPS)["units"]]
        self.assertIn("mount-proc", names)

    def test_op_units_are_longrun_and_restart_on_failure(self):
        unit = [u for u in build_state(OPS)["units"]
                if u["name"] == "op-swarmexecutor-v1"][0]
        self.assertEqual(unit["kind"], "longrun")
        self.assertEqual(unit["restart"], "on-failure")

    def test_op_units_depend_on_join(self):
        unit = [u for u in build_state(OPS)["units"]
                if u["name"] == "op-swarmexecutor-v1"][0]
        self.assertIn("join", unit["requires"])

    def test_built_state_passes_validation(self):
        validate(build_state(OPS))

    def test_all_nine_arch_targets_included(self):
        self.assertEqual(len(build_state(OPS)["arch_targets"]), 9)

    def test_op_name_with_a_space_is_refused_at_export(self):
        # The op name comes over HTTP from the documented --ops-url CLI, and
        # unit_name_for_op passes the space straight through into the unit
        # name, shifting every later field of the U record.
        with self.assertRaises(ValidationError) as ctx:
            build_state(["Bad Op_v1"])
        self.assertIn("op-bad op-v1", str(ctx.exception))


class TestExport(unittest.TestCase):
    def setUp(self):
        self.out = tempfile.mkdtemp()

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

    def test_writes_both_artifacts(self):
        export(OPS, self.out)
        self.assertTrue(os.path.exists(os.path.join(self.out, "numos.state")))
        self.assertTrue(os.path.exists(os.path.join(self.out, "numos.state.json")))

    def test_written_state_self_verifies(self):
        export(OPS, self.out)
        with open(os.path.join(self.out, "numos.state")) as handle:
            verify(handle.read())

    def test_export_is_byte_identical_across_runs(self):
        export(OPS, self.out)
        with open(os.path.join(self.out, "numos.state"), "rb") as handle:
            first = handle.read()
        export(OPS, self.out)
        with open(os.path.join(self.out, "numos.state"), "rb") as handle:
            second = handle.read()
        self.assertEqual(first, second)

    def test_op_order_does_not_change_output(self):
        export(OPS, self.out)
        with open(os.path.join(self.out, "numos.state"), "rb") as handle:
            first = handle.read()
        export(list(reversed(OPS)), self.out)
        with open(os.path.join(self.out, "numos.state"), "rb") as handle:
            second = handle.read()
        self.assertEqual(first, second)

    def test_json_view_round_trips_to_the_same_state(self):
        export(OPS, self.out)
        with open(os.path.join(self.out, "numos.state")) as handle:
            from_lines = parse(handle.read())
        with open(os.path.join(self.out, "numos.state.json")) as handle:
            from_json = json.load(handle)
        self.assertEqual(from_lines["units"], from_json["units"])

    def test_state_uses_lf_endings_only(self):
        export(OPS, self.out)
        with open(os.path.join(self.out, "numos.state"), "rb") as handle:
            self.assertNotIn(b"\r", handle.read())


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