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

#!/usr/bin/env python3
"""RED-first contract tests for verbatim-hashed normative fragments."""

from __future__ import annotations

import hashlib
import importlib.util
import tempfile
import unittest
from pathlib import Path


MODULE_PATH = Path(__file__).parents[2] / "framework/tools/lease-broker/normative_fragments.py"


def shipped_module():
    # Each test reaches the shipped implementation; no test doubles or local
    # reimplementation of construction are allowed on this admission surface.
    assert MODULE_PATH.is_file(), f"shipped construction module is missing: {MODULE_PATH}"
    spec = importlib.util.spec_from_file_location("normative_fragments", MODULE_PATH)
    if spec is None or spec.loader is None:
        raise RuntimeError("unable to load normative fragment construction")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def fragment(module, source_id: str, content: bytes):
    return module.NormativeFragment(
        source_id=source_id,
        content=content,
        expected_sha256=hashlib.sha256(content).hexdigest(),
    )


def valid_fragments(module):
    return [
        fragment(module, "authority/constitution", b"Constitution\n"),
        fragment(module, "authority/runtime", b"Runtime\n"),
    ]


class NormativeFragmentsTest(unittest.TestCase):
    def test_t24_claude_and_pi_builders_are_byte_identical_and_one_way(self) -> None:
        module = shipped_module()
        fragments = valid_fragments(module)

        claude = module.build_for_claude(
            manifest_version=1,
            generator_version="wi4-test",
            fragments=fragments,
        )
        pi = module.build_for_pi(
            manifest_version=1,
            generator_version="wi4-test",
            fragments=fragments,
        )

        self.assertEqual(claude.injectionDecision, "ACCEPTED")
        self.assertTrue(claude.promotion)
        self.assertEqual(claude.b_payload, pi.b_payload)
        self.assertEqual(claude.h_payload, pi.h_payload)
        self.assertEqual(
            claude.h_payload,
            hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + module.length_frame([claude.b_payload])).hexdigest(),
        )
        self.assertNotIn(b"h_payload", claude.b_payload)

        mutated_fragment = module.build_for_claude(
            manifest_version=1,
            generator_version="wi4-test",
            fragments=[
                fragment(module, "authority/constitution", b"Constitution changed\n"),
                fragment(module, "authority/runtime", b"Runtime\n"),
            ],
        )
        mutated_metadata = module.build_for_claude(
            manifest_version=2,
            generator_version="wi4-test",
            fragments=fragments,
        )
        self.assertNotEqual(claude.h_payload, mutated_fragment.h_payload)
        self.assertNotEqual(claude.h_payload, mutated_metadata.h_payload)

    def test_length_framing_and_domain_separation_prevent_ambiguous_construction(self) -> None:
        module = shipped_module()
        left = module.length_frame([b"ab", b"c"])
        right = module.length_frame([b"a", b"bc"])

        self.assertNotEqual(left, right)
        self.assertNotEqual(
            hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + left).hexdigest(),
            hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + right).hexdigest(),
        )
        self.assertNotEqual(
            hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + left).hexdigest(),
            hashlib.sha256(b"other-context\x00" + left).hexdigest(),
        )

    def test_source_invalidation_missing_refuses_real_construction_and_promotion(self) -> None:
        module = shipped_module()
        missing = module.NormativeFragment(
            source_id="authority/missing",
            content=None,
            expected_sha256=hashlib.sha256(b"missing").hexdigest(),
        )

        result = module.build_for_claude(
            manifest_version=1,
            generator_version="wi4-test",
            fragments=[missing],
        )

        self.assertEqual(result.injectionDecision, "REFUSED")
        self.assertFalse(result.promotion)
        self.assertEqual(result.source_reason, "missing")
        self.assertIsNone(result.b_payload)
        self.assertIsNone(result.h_payload)

    def test_source_invalidation_oversize_refuses_real_construction_and_promotion(self) -> None:
        module = shipped_module()
        content = b"x" * (module.MAX_FRAGMENT_BYTES + 1)
        oversize = fragment(module, "authority/oversize", content)

        result = module.build_for_pi(
            manifest_version=1,
            generator_version="wi4-test",
            fragments=[oversize],
        )

        self.assertEqual(result.injectionDecision, "REFUSED")
        self.assertFalse(result.promotion)
        self.assertEqual(result.source_reason, "oversize")
        self.assertIsNone(result.b_payload)
        self.assertIsNone(result.h_payload)

    def test_source_invalidation_hash_mismatch_refuses_real_construction_and_promotion(self) -> None:
        module = shipped_module()
        mismatch = module.NormativeFragment(
            source_id="authority/hash-mismatch",
            content=b"trusted bytes",
            expected_sha256=hashlib.sha256(b"different bytes").hexdigest(),
        )

        result = module.build_for_claude(
            manifest_version=1,
            generator_version="wi4-test",
            fragments=[mismatch],
        )

        self.assertEqual(result.injectionDecision, "REFUSED")
        self.assertFalse(result.promotion)
        self.assertEqual(result.source_reason, "hash-mismatch")
        self.assertIsNone(result.b_payload)
        self.assertIsNone(result.h_payload)


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