tests/test_distro.py

back to source

# SPDX-License-Identifier: MIT
import json
import os
import tempfile
import unittest

from numos.distro import (
    DISTRO_PROFILES,
    floor_manifest,
    get_profile,
    kernel_manifest,
    mermaid_for_profile,
    profile_graph,
    profile_summary,
    resolve_floor,
    resolve_kernel,
)
from numos.export_state import build_state, export
from numos.state import render, verify


class TestDistroProfiles(unittest.TestCase):
    def test_self_init_and_fleet_exist(self):
        self.assertIn("self-init", DISTRO_PROFILES)
        self.assertIn("fleet", DISTRO_PROFILES)

    def test_self_init_has_no_net_or_join(self):
        p = get_profile("self-init")
        names = {u["name"] for u in p["units"]}
        self.assertIn("hold", names)
        self.assertIn("identity", names)
        self.assertNotIn("net-up", names)
        self.assertNotIn("join", names)
        self.assertNotIn("clock-sync", names)

    def test_self_init_excludes_op_units(self):
        self.assertFalse(get_profile("self-init")["include_op_units"])
        self.assertTrue(get_profile("fleet")["include_op_units"])

    def test_self_init_state_validates_and_hashes(self):
        state = build_state([], profile="self-init")
        text = render(state)
        verify(text)
        self.assertIn("U hold ", text)
        self.assertIn("P 30 complete ", text)
        self.assertNotIn("U net-up ", text)

    def test_fleet_still_has_join(self):
        state = build_state([], profile="fleet")
        text = render(state)
        self.assertIn("U join ", text)
        self.assertIn("U net-up ", text)

    def test_floor_resolve_x86_64(self):
        floor = resolve_floor("x86_64", "self-init")
        self.assertEqual(floor["install_name"], "numos-floor")
        self.assertTrue(floor["url"])
        self.assertEqual(len(floor["sha256"]), 64)
        self.assertIn("sh", floor["applets"])
        self.assertIn("mount", floor["applets"])

    def test_floor_manifest_json(self):
        m = floor_manifest("x86_64", "self-init")
        self.assertEqual(m["profile"], "self-init")
        self.assertEqual(m["install_name"], "numos-floor")

    def test_export_writes_floor_and_profile(self):
        with tempfile.TemporaryDirectory() as tmp:
            export([], tmp, profile="self-init", arch="x86_64")
            self.assertTrue(os.path.isfile(os.path.join(tmp, "numos.state")))
            with open(os.path.join(tmp, "distro.floor.json"), encoding="utf-8") as f:
                floor = json.load(f)
            self.assertEqual(floor["install_name"], "numos-floor")
            with open(os.path.join(tmp, "distro.profile.json"), encoding="utf-8") as f:
                prof = json.load(f)
            self.assertEqual(prof["profile"], "self-init")
            self.assertIn("hold", prof["unit_names"])
            with open(os.path.join(tmp, "distro.graph.json"), encoding="utf-8") as f:
                graph = json.load(f)
            self.assertGreaterEqual(graph["counts"]["units"], 5)
            self.assertTrue(os.path.isfile(os.path.join(tmp, "distro.graph.mmd")))

    def test_profile_graph_has_layered_nodes(self):
        g = profile_graph("self-init")
        kinds = {n["kind"] for n in g["nodes"]}
        self.assertIn("BootPhase", kinds)
        self.assertIn("OSUnit", kinds)
        self.assertIn("HealthPredicate", kinds)
        self.assertIn("FloorProvider", kinds)
        layers = {n["layer"] for n in g["nodes"]}
        self.assertEqual(layers, {"META", "GENESIS", "MGS", "SHADOW"})

    def test_mermaid_mentions_hold_and_phases(self):
        mmd = mermaid_for_profile("self-init")
        self.assertIn("flowchart TD", mmd)
        self.assertIn("hold", mmd)
        self.assertIn("mount", mmd)

    def test_profile_summary_keys(self):
        s = profile_summary("self-init")
        self.assertEqual(s["profile"], "self-init")
        self.assertIn("x86_64", s["floor_arches"])
        self.assertIn("x86_64", s["kernel_arches"])

    def test_kernel_resolve_x86_64(self):
        k = resolve_kernel("x86_64", "self-init")
        self.assertEqual(k["product"], "third-party-kernel")
        self.assertEqual(k["kind"], "deb")
        self.assertIn("linux-image", k["url"])
        m = kernel_manifest("x86_64", "self-init")
        self.assertIn("asec", m)
        self.assertEqual(m["install_name"], "vmlinuz")

    def test_kernel_missing_arch(self):
        with self.assertRaises(KeyError):
            resolve_kernel("riscv64", "self-init")

    def test_profile_graph_has_kernel_provider(self):
        g = profile_graph("self-init")
        kinds = {n["kind"] for n in g["nodes"]}
        self.assertIn("KernelProvider", kinds)
        rels = {e["rel"] for e in g["edges"]}
        self.assertIn("kernel_for", rels)

    def test_export_writes_kernel_manifest(self):
        with tempfile.TemporaryDirectory() as tmp:
            export([], tmp, profile="self-init", arch="x86_64")
            with open(os.path.join(tmp, "distro.kernel.json"), encoding="utf-8") as f:
                k = json.load(f)
            self.assertEqual(k["product"], "third-party-kernel")
            self.assertTrue(k["url"])


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