"""Materialization (ingest-spec §5): staging, collision gate, replacement. Three explicit inputs — manifest, bundle dir, ingested_at (validated, no wall-clock default). All extractions execute and render IN MEMORY before the first disk mutation; the §3 collision gate runs before any mutation; only ingest-stamped files are ever replaced. LF-only bytes, one trailing newline. """ from __future__ import annotations import hashlib import json import logging import sqlite3 from pathlib import Path from typing import Any import pytest from llm_ingestion_okf.errors import IngestError, MaterializationError, NetworkGateError from llm_ingestion_okf.materialize import IngestResult, materialize_bundle INGESTED_AT = "2026-07-16T12:00:00Z" def write_manifest(directory: Path, data: dict[str, Any]) -> Path: directory.mkdir(parents=True, exist_ok=True) path = directory / "manifest.json" path.write_text(json.dumps(data), encoding="utf-8") return path def stamp_of(manifest_path: Path) -> str: h16 = hashlib.sha256(manifest_path.read_bytes()).hexdigest()[:16] return f"{manifest_path.stem}@{h16}" def file_manifest_data(extractions: list[dict[str, Any]] | None = None) -> dict[str, Any]: return { "manifest_version": 1, "source": {"type": "file", "id": "catalogue-1", "root": "data"}, "bundle_summary": "A test bundle.", "extractions": extractions or [ { "id": "orders", "title": "Orders", "query": "orders.csv", "okf_type": "dataset", "max_rows": 100, } ], } @pytest.fixture def file_setup(tmp_path: Path) -> tuple[Path, Path]: """A manifest + CSV catalogue in tmp/src, and an empty bundle target path.""" src = tmp_path / "src" manifest_path = write_manifest(src, file_manifest_data()) (src / "data").mkdir() (src / "data" / "orders.csv").write_text("a,b\n1,x\n2,y\n", encoding="utf-8", newline="") return manifest_path, tmp_path / "bundle" # --- error hierarchy --- def test_materialization_error_is_ingest_error() -> None: assert issubclass(MaterializationError, IngestError) assert issubclass(NetworkGateError, IngestError) # --- ingested_at: explicit, validated, no wall-clock default --- @pytest.mark.parametrize( "bad", [ "2026-07-16", "2026-07-16T12:00:00", "2026-07-16T12:00:00+00:00", "2026-07-16 12:00:00Z", "16-07-2026T12:00:00Z", "", ], ) def test_bad_ingested_at_rejected_before_any_mutation( file_setup: tuple[Path, Path], bad: str ) -> None: manifest_path, bundle = file_setup with pytest.raises(MaterializationError): materialize_bundle(manifest_path, bundle, bad) assert not bundle.exists() def test_missing_manifest_raises_typed_error(tmp_path: Path) -> None: with pytest.raises(IngestError): materialize_bundle(tmp_path / "nope.json", tmp_path / "bundle", INGESTED_AT) # --- file source: exact §5 bytes --- def test_file_source_concept_file_exact_bytes(file_setup: tuple[Path, Path]) -> None: manifest_path, bundle = file_setup materialize_bundle(manifest_path, bundle, INGESTED_AT) expected = ( "---\n" "type: dataset\n" "title: Orders\n" "source_system: catalogue-1\n" "source_query: orders.csv\n" f"ingested_at: {INGESTED_AT}\n" f"ingest_manifest: {stamp_of(manifest_path)}\n" "generated: true\n" "---\n" "\n" "| a | b |\n" "| --- | --- |\n" "| 1 | x |\n" "| 2 | y |\n" ) assert (bundle / "ingest-orders.md").read_bytes() == expected.encode("utf-8") def test_title_renders_identically_in_frontmatter_and_index(tmp_path: Path) -> None: # §4: the title "becomes the `title` frontmatter AND the index link # label" — one value, two sites, so the two MUST agree. The frontmatter # renderer single-lines its values; the index label was written raw, so # a title carrying a whitespace run diverged between the two. src = tmp_path / "src" manifest_path = write_manifest( src, file_manifest_data( [ { "id": "orders", "title": "Energiforbruk 2024", "query": "orders.csv", "okf_type": "dataset", "max_rows": 100, } ] ), ) (src / "data").mkdir() (src / "data" / "orders.csv").write_text("a,b\n1,x\n", encoding="utf-8", newline="") bundle = tmp_path / "bundle" materialize_bundle(manifest_path, bundle, ingested_at=INGESTED_AT) concept = (bundle / "ingest-orders.md").read_text(encoding="utf-8") index = (bundle / "index.md").read_text(encoding="utf-8") frontmatter_title = next( line.partition(":")[2].strip() for line in concept.splitlines() if line.startswith("title:") ) index_label = index.partition("- [")[2].partition("]")[0] assert frontmatter_title == index_label def test_title_emitted_verbatim_not_collapsed(tmp_path: Path) -> None: # §5 mandates whitespace-run collapse ONLY for `source_query` # (ingest-spec.md:140-141). `title` is validated single-line at manifest # load and emitted verbatim — validation, not repair — so an internal # whitespace run survives at BOTH the frontmatter and the index-label site. src = tmp_path / "src" manifest_path = write_manifest( src, file_manifest_data( [ { "id": "orders", "title": "Energiforbruk 2024", "query": "orders.csv", "okf_type": "dataset", "max_rows": 100, } ] ), ) (src / "data").mkdir() (src / "data" / "orders.csv").write_text("a,b\n1,x\n", encoding="utf-8", newline="") bundle = tmp_path / "bundle" materialize_bundle(manifest_path, bundle, ingested_at=INGESTED_AT) concept = (bundle / "ingest-orders.md").read_text(encoding="utf-8") index = (bundle / "index.md").read_text(encoding="utf-8") assert "title: Energiforbruk 2024\n" in concept assert "- [Energiforbruk 2024](ingest-orders.md)" in index def test_result_lists_written_concept_files(file_setup: tuple[Path, Path]) -> None: manifest_path, bundle = file_setup result = materialize_bundle(manifest_path, bundle, INGESTED_AT) assert isinstance(result, IngestResult) assert result.written == (bundle / "ingest-orders.md",) def test_result_exposes_manifest_stamp(file_setup: tuple[Path, Path]) -> None: """The §5 provenance stamp ({stem}@{sha256(raw)[:16]}) is returned on the result so consumers never have to recompute it from the manifest bytes.""" manifest_path, bundle = file_setup result = materialize_bundle(manifest_path, bundle, INGESTED_AT) assert result.stamp == stamp_of(manifest_path) content = (bundle / "ingest-orders.md").read_text(encoding="utf-8") assert f"ingest_manifest: {result.stamp}\n" in content def test_relative_root_resolves_against_manifest_dir( tmp_path: Path, file_setup: tuple[Path, Path], monkeypatch: pytest.MonkeyPatch ) -> None: """Pinned decision: a relative root resolves against the manifest file's directory — never the process cwd (reproducibility).""" manifest_path, bundle = file_setup elsewhere = tmp_path / "elsewhere" elsewhere.mkdir() monkeypatch.chdir(elsewhere) result = materialize_bundle(manifest_path, bundle, INGESTED_AT) assert len(result.written) == 1 # --- sql source end-to-end --- def test_sql_source_end_to_end(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: db = tmp_path / "fixture.db" with sqlite3.connect(db) as conn: conn.execute("CREATE TABLE t (id INTEGER, name TEXT)") conn.executemany("INSERT INTO t VALUES (?, ?)", [(1, "alpha"), (2, None)]) monkeypatch.setenv("OKF_TEST_DB", str(db)) data = file_manifest_data( [ { "id": "things", "title": "Things", "query": "SELECT id,\n name FROM t ORDER BY id", "okf_type": "dataset", "max_rows": 10, } ] ) data["source"] = {"type": "sql", "id": "db-1", "connection_ref": "OKF_TEST_DB"} manifest_path = write_manifest(tmp_path / "src", data) bundle = tmp_path / "bundle" materialize_bundle(manifest_path, bundle, INGESTED_AT) content = (bundle / "ingest-things.md").read_text(encoding="utf-8") # §5: whitespace runs (incl. newlines) in source_query collapse to single spaces. assert "source_query: SELECT id, name FROM t ORDER BY id\n" in content assert content.endswith("| 1 | alpha |\n| 2 | |\n") # --- fresh index generation (§6, creation path) --- def test_fresh_index_exact_bytes(tmp_path: Path) -> None: src = tmp_path / "src" data = file_manifest_data( [ { "id": "orders", "title": "Orders", "query": "orders.csv", "okf_type": "dataset", "max_rows": 10, }, { "id": "refunds", "title": "Refunds", "query": "refunds.csv", "okf_type": "dataset", "max_rows": 10, }, ] ) manifest_path = write_manifest(src, data) (src / "data").mkdir() (src / "data" / "orders.csv").write_text("a\n1\n", encoding="utf-8", newline="") (src / "data" / "refunds.csv").write_text("b\n2\n", encoding="utf-8", newline="") bundle = tmp_path / "bundle" materialize_bundle(manifest_path, bundle, INGESTED_AT) expected = "A test bundle.\n- [Orders](ingest-orders.md)\n- [Refunds](ingest-refunds.md)\n" assert (bundle / "index.md").read_bytes() == expected.encode("utf-8") # --- §3 collision gate: BEFORE any mutation --- def test_collision_with_unstamped_file_fails_without_mutation( file_setup: tuple[Path, Path], ) -> None: manifest_path, bundle = file_setup bundle.mkdir() curated = bundle / "ingest-orders.md" curated.write_text("hand-curated content, no stamp\n", encoding="utf-8") stale = bundle / "ingest-stale.md" stale.write_text( "---\ntype: dataset\ngenerated: true\ningest_manifest: old@0000\n---\n\nold\n", encoding="utf-8", ) with pytest.raises(MaterializationError): materialize_bundle(manifest_path, bundle, INGESTED_AT) # Gate fired BEFORE any mutation: curated file untouched, stale stamped file still there. assert curated.read_text(encoding="utf-8") == "hand-curated content, no stamp\n" assert stale.is_file() # --- §5 replacement: only ingest-stamped files are replaced --- def test_replacement_removes_stale_stamped_files(file_setup: tuple[Path, Path]) -> None: # A prior run of THIS manifest (stem `manifest`, older content -> older # sha) left ingest-stale.md, which the current run no longer produces. # §5 replacement reclaims it because the stem still names this manifest. manifest_path, bundle = file_setup bundle.mkdir() stale = bundle / "ingest-stale.md" stale.write_text( "---\ntype: dataset\ngenerated: true\ningest_manifest: manifest@0000\n---\n\nold\n", encoding="utf-8", ) materialize_bundle(manifest_path, bundle, INGESTED_AT) assert not stale.exists() assert (bundle / "ingest-orders.md").is_file() def test_replacement_leaves_another_manifests_stamped_file(file_setup: tuple[Path, Path]) -> None: # The mirror of the case above: a stale file stamped by a DIFFERENT # manifest (stem `other`) is NOT this manifest's to remove, so it survives. manifest_path, bundle = file_setup bundle.mkdir() foreign = bundle / "ingest-foreign.md" foreign.write_text( "---\ntype: dataset\ngenerated: true\ningest_manifest: other@0000\n---\n\nold\n", encoding="utf-8", ) materialize_bundle(manifest_path, bundle, INGESTED_AT) assert foreign.is_file() assert (bundle / "ingest-orders.md").is_file() def test_second_manifest_does_not_delete_first_manifests_stamped_file(tmp_path: Path) -> None: # §10.2 per-manifest ownership: two manifests writing into ONE bundle each # own only the files whose stamp names them by stem. Running manifest B # must leave the ingest file that manifest A stamped in place — a narrower # replacement blast radius than "remove every stamped file". (This does not # close the operator-copy restriction, which stays documented in # ingest-spec.md:69-72, not enforced.) src = tmp_path / "src" src.mkdir() (src / "data").mkdir() (src / "data" / "orders.csv").write_text("a,b\n1,x\n", encoding="utf-8", newline="") alpha = src / "alpha.json" alpha.write_text( json.dumps( file_manifest_data( [ { "id": "alpha-thing", "title": "Alpha", "query": "orders.csv", "okf_type": "dataset", "max_rows": 10, } ] ) ), encoding="utf-8", ) beta = src / "beta.json" beta.write_text( json.dumps( file_manifest_data( [ { "id": "beta-thing", "title": "Beta", "query": "orders.csv", "okf_type": "dataset", "max_rows": 10, } ] ) ), encoding="utf-8", ) bundle = tmp_path / "bundle" materialize_bundle(alpha, bundle, INGESTED_AT) alpha_file = bundle / "ingest-alpha-thing.md" assert alpha_file.is_file() materialize_bundle(beta, bundle, INGESTED_AT) # A's stamped file survives B's run, and B's own file is written alongside it. assert alpha_file.is_file() assert (bundle / "ingest-beta-thing.md").is_file() # A's index link survives too — B only removes links to files it owns. index = (bundle / "index.md").read_text(encoding="utf-8") assert "- [Alpha](ingest-alpha-thing.md)" in index assert "- [Beta](ingest-beta-thing.md)" in index def test_curated_files_survive_byte_identical(file_setup: tuple[Path, Path]) -> None: manifest_path, bundle = file_setup bundle.mkdir() curated = bundle / "notes.md" curated_bytes = b"---\ntype: note\n---\n\ncurated, no stamp\n" curated.write_bytes(curated_bytes) materialize_bundle(manifest_path, bundle, INGESTED_AT) assert curated.read_bytes() == curated_bytes # --- in-memory staging: extraction failure leaves the disk untouched --- def test_failed_extraction_means_zero_disk_mutation(tmp_path: Path) -> None: src = tmp_path / "src" data = file_manifest_data( [ { "id": "good", "title": "Good", "query": "good.csv", "okf_type": "dataset", "max_rows": 10, }, {"id": "bad", "title": "Bad", "query": "bad.csv", "okf_type": "dataset", "max_rows": 1}, ] ) manifest_path = write_manifest(src, data) (src / "data").mkdir() (src / "data" / "good.csv").write_text("a\n1\n", encoding="utf-8", newline="") (src / "data" / "bad.csv").write_text("a\n1\n2\n", encoding="utf-8", newline="") bundle = tmp_path / "bundle" with pytest.raises(IngestError): materialize_bundle(manifest_path, bundle, INGESTED_AT) assert not bundle.exists() # --- determinism and idempotence (§10) --- def test_two_runs_into_fresh_dirs_are_byte_identical(file_setup: tuple[Path, Path]) -> None: manifest_path, bundle = file_setup other = bundle.parent / "bundle2" materialize_bundle(manifest_path, bundle, INGESTED_AT) materialize_bundle(manifest_path, other, INGESTED_AT) files = sorted(path.name for path in bundle.iterdir()) assert files == sorted(path.name for path in other.iterdir()) for name in files: assert (bundle / name).read_bytes() == (other / name).read_bytes() def test_rerun_over_own_output_is_idempotent(file_setup: tuple[Path, Path]) -> None: manifest_path, bundle = file_setup materialize_bundle(manifest_path, bundle, INGESTED_AT) before = {path.name: path.read_bytes() for path in bundle.iterdir()} materialize_bundle(manifest_path, bundle, INGESTED_AT) after = {path.name: path.read_bytes() for path in bundle.iterdir()} assert after == before # incl. no duplicated index links # --- §8: source calls are logged (which source, when, row count) --- def test_source_call_logged( file_setup: tuple[Path, Path], caplog: pytest.LogCaptureFixture ) -> None: manifest_path, bundle = file_setup with caplog.at_level(logging.INFO, logger="llm_ingestion_okf.materialize"): materialize_bundle(manifest_path, bundle, INGESTED_AT) joined = " ".join(record.getMessage() for record in caplog.records) assert "catalogue-1" in joined assert INGESTED_AT in joined assert "rows=2" in joined # --- §8 network gate: http refused fail-fast without the per-run opt-in --- def test_http_without_opt_in_refused_fail_fast(tmp_path: Path) -> None: """Load-bearing (spec §11): the manifest cannot grant itself network access.""" data = file_manifest_data() data["source"] = {"type": "http", "id": "api-1", "base_url": "https://example.test"} data["extractions"][0]["query"] = "/orders" manifest_path = write_manifest(tmp_path / "src", data) bundle = tmp_path / "bundle" with pytest.raises(NetworkGateError): materialize_bundle(manifest_path, bundle, INGESTED_AT) assert not bundle.exists()