stack/packages/mosaic/src/lease-broker/promotion_binding_unittest.py

#!/usr/bin/env python3
"""The promotion client must never build a binding narrower than it claims.

RED-first against a real defect: ``build_construction`` skipped any normative
source it could not read (``except OSError: continue``) and promoted whatever
remained. That is not a degraded binding, it is a forged smaller one — the
broker recomputes ``h_source`` / ``h_payload`` from the fragments it is *sent*
(``daemon.py:602-616``), so an omitted fragment is internally consistent and
``PAYLOAD_BINDING_MISMATCH`` cannot fire. Measured before the fix: with only
``USER.md`` readable (964 bytes on the live host), the client produced a
one-fragment construction with ``promotion=True``.

The classification under test mirrors the framework's own file ownership, and
must keep mirroring it:

  * framework-owned, reconciled every upgrade (``install.sh`` FRAMEWORK_OWNED /
    ``config/file-adapter.ts`` FRAMEWORK_OWNED_FILES) plus the per-runtime
    contract — absence is a broken deployment, so it is REFUSED;
  * ``SOUL.md`` / ``USER.md`` — install.sh deliberately does not seed them
    ("generated by `mosaic init`"), so absence is legitimate and ALLOWED.

Unreadable is treated separately from absent for *every* source, optional ones
included: a file that will not open is not a file that was never configured, and
collapsing the two is what let a permission change quietly shrink the law.
"""

from __future__ import annotations

import contextlib
import io
import os
import sys
import tempfile
import unittest
from pathlib import Path

TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
sys.path.insert(0, str(TOOLS))

import lease_promote  # noqa: E402

RUNTIME = "pi"
RUNTIME_CONTRACT = f"runtime/{RUNTIME}/RUNTIME.md"
ALL_SOURCES = (*lease_promote.FRAGMENT_SOURCES, RUNTIME_CONTRACT)
REQUIRED = frozenset(lease_promote.REQUIRED_SOURCES) | {RUNTIME_CONTRACT}
# Derived, never listed: a hand-kept second copy is exactly the drift this file
# exists to catch.
OPTIONAL = tuple(s for s in ALL_SOURCES if s not in REQUIRED)

# chmod 0o000 does not deny root (CAP_DAC_OVERRIDE), so the unreadable
# simulations would fail spuriously in a root container.
runs_unprivileged = unittest.skipIf(
    os.geteuid() == 0, "chmod 0o000 cannot make a file unreadable to root"
)


class PromotionBindingTest(unittest.TestCase):
    def setUp(self) -> None:
        self._previous_home = os.environ.get("MOSAIC_HOME")
        self._temporary = tempfile.TemporaryDirectory()
        self.root = Path(self._temporary.name)
        for source_id in ALL_SOURCES:
            path = self.root / source_id
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_bytes(f"# {source_id}\nnormative bytes\n".encode())
        os.environ["MOSAIC_HOME"] = str(self.root)

    def tearDown(self) -> None:
        for path in self.root.rglob("*"):
            if path.is_file():
                path.chmod(0o644)
        self._temporary.cleanup()
        if self._previous_home is None:
            os.environ.pop("MOSAIC_HOME", None)
        else:
            os.environ["MOSAIC_HOME"] = self._previous_home

    def reset_home(self) -> None:
        """Discard the current home and seed a fresh complete one.

        Each subTest mutates the tree destructively, so it needs a clean start —
        and the old one must be released, not orphaned.
        """
        self.tearDown()
        self.setUp()

    def build(self):
        return lease_promote.build_construction(RUNTIME)

    def source_ids(self) -> list[str]:
        construction, _ = self.build()
        return [f["source_id"] for f in construction["fragments"]]

    # --- the binding is complete when the deployment is complete -------------

    def test_complete_deployment_binds_every_source(self) -> None:
        construction, result = self.build()
        self.assertEqual([f["source_id"] for f in construction["fragments"]], list(ALL_SOURCES))
        self.assertTrue(result.promotion)

    # --- absence: refused for framework-owned, allowed for operator-owned ----

    def test_absent_required_source_is_refused(self) -> None:
        for source_id in sorted(REQUIRED):
            with self.subTest(source=source_id):
                self.reset_home()
                (self.root / source_id).unlink()
                with self.assertRaises(lease_promote.IncompleteBinding) as caught:
                    self.build()
                self.assertIn(source_id, str(caught.exception))

    def test_absent_operator_source_still_binds_the_rest(self) -> None:
        for source_id in OPTIONAL:
            with self.subTest(source=source_id):
                self.reset_home()
                (self.root / source_id).unlink()
                notice = io.StringIO()
                with contextlib.redirect_stderr(notice):
                    bound = self.source_ids()
                self.assertNotIn(source_id, bound)
                for required in lease_promote.REQUIRED_SOURCES:
                    self.assertIn(required, bound)
                # A silent omission is the original defect in miniature: the
                # narrower binding must announce itself.
                self.assertIn(source_id, notice.getvalue())

    # --- unreadable is never the same as absent -----------------------------

    @runs_unprivileged
    def test_unreadable_source_is_refused_even_when_optional(self) -> None:
        for source_id in ALL_SOURCES:
            with self.subTest(source=source_id):
                self.reset_home()
                (self.root / source_id).chmod(0o000)
                with self.assertRaises(lease_promote.IncompleteBinding) as caught:
                    self.build()
                self.assertIn(source_id, str(caught.exception))

    # --- the exact measured regression --------------------------------------

    @runs_unprivileged
    def test_single_readable_source_cannot_promote(self) -> None:
        """The observed failure: only USER.md readable produced a valid binding."""
        for source_id in ALL_SOURCES:
            if source_id != "USER.md":
                (self.root / source_id).chmod(0o000)
        with self.assertRaises(lease_promote.IncompleteBinding):
            self.build()

    @runs_unprivileged
    def test_no_source_readable_cannot_promote(self) -> None:
        for source_id in ALL_SOURCES:
            (self.root / source_id).chmod(0o000)
        with self.assertRaises(lease_promote.IncompleteBinding):
            self.build()

    # --- the classification must not drift from the framework's -------------

    def test_required_set_excludes_only_the_unseeded_sources(self) -> None:
        """`install.sh` decides which files exist; this list must follow it.

        If a source moves between framework-owned and operator-generated
        upstream, this fails and forces the classification to be re-read rather
        than silently inherited.
        """
        self.assertEqual(
            set(lease_promote.REQUIRED_SOURCES),
            {"CONSTITUTION.md", "AGENTS.md", "STANDARDS.md"},
            "REQUIRED_SOURCES changed — re-read install.sh FRAMEWORK_OWNED and "
            "config/file-adapter.ts FRAMEWORK_OWNED_FILES before accepting it",
        )
        self.assertTrue(
            set(lease_promote.REQUIRED_SOURCES) <= set(lease_promote.FRAGMENT_SOURCES),
            "a required source is not in the binding order",
        )


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