"""Ingest golden regression (ingest-spec §11) — the shared byte-for-byte fasit. D7 mirror of MAF I2. Consumes the repo-local ``examples/ingest-golden-file/`` (a byte-frozen copy of MAF I2's golden — the SAME shared extraction, so this test proves D7's independent implementation reproduces MAF's exact bytes from the shared spec alone). ``expected-bundle/`` is compared file by file, byte for byte. The mutation controls prove the net is taut: one changed determinism input (a CSV cell, the timestamp, the manifest bytes) must diverge from the golden bytes. """ from __future__ import annotations import shutil from pathlib import Path from portfolio_optimiser_claude.ingest import materialize GOLDEN = Path(__file__).resolve().parents[1] / "examples" / "ingest-golden-file" def _ingested_at() -> str: return (GOLDEN / "ingested-at.txt").read_text(encoding="utf-8").strip() def test_materializes_golden_byte_for_byte(tmp_path: Path) -> None: bundle = tmp_path / "bundle" bundle.mkdir() materialize(GOLDEN / "manifest.json", bundle, _ingested_at()) expected = GOLDEN / "expected-bundle" expected_names = sorted(p.name for p in expected.iterdir()) produced_names = sorted(p.name for p in bundle.iterdir()) assert produced_names == expected_names # no missing, no extra files for name in expected_names: assert (bundle / name).read_bytes() == (expected / name).read_bytes(), name def test_reingest_is_idempotent(tmp_path: Path) -> None: # ingest-spec §10: same source content, manifest, ingested_at → byte-identical, # repeated runs idempotent. bundle = tmp_path / "bundle" bundle.mkdir() materialize(GOLDEN / "manifest.json", bundle, _ingested_at()) first = {p.name: p.read_bytes() for p in bundle.iterdir()} materialize(GOLDEN / "manifest.json", bundle, _ingested_at()) second = {p.name: p.read_bytes() for p in bundle.iterdir()} assert first == second class TestMutationControl: """One changed determinism input → divergence from the golden bytes.""" def _case(self, tmp_path: Path) -> Path: case = tmp_path / "case" shutil.copytree(GOLDEN, case) return case def test_changed_cell_diverges(self, tmp_path: Path) -> None: case = self._case(tmp_path) csv = case / "fixture" / "costs.csv" csv.write_bytes(csv.read_bytes().replace(b"120000", b"999999")) bundle = tmp_path / "bundle" bundle.mkdir() materialize(case / "manifest.json", bundle, _ingested_at()) golden_bytes = (GOLDEN / "expected-bundle" / "ingest-costs.md").read_bytes() assert (bundle / "ingest-costs.md").read_bytes() != golden_bytes def test_changed_timestamp_diverges(self, tmp_path: Path) -> None: bundle = tmp_path / "bundle" bundle.mkdir() materialize(GOLDEN / "manifest.json", bundle, "2099-01-01T00:00:00Z") golden_bytes = (GOLDEN / "expected-bundle" / "ingest-costs.md").read_bytes() assert (bundle / "ingest-costs.md").read_bytes() != golden_bytes def test_changed_manifest_bytes_change_the_stamp(self, tmp_path: Path) -> None: # The ingest_manifest stamp is SHA-256 of the manifest's raw bytes (§5) — # a whitespace-only edit still re-stamps every generated file. case = self._case(tmp_path) manifest = case / "manifest.json" manifest.write_bytes(manifest.read_bytes() + b"\n") bundle = tmp_path / "bundle" bundle.mkdir() materialize(manifest, bundle, _ingested_at()) golden_bytes = (GOLDEN / "expected-bundle" / "ingest-costs.md").read_bytes() assert (bundle / "ingest-costs.md").read_bytes() != golden_bytes