llm-ingestion-okf/tests/test_materialize.py

"""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"
        f"generated: {{ by: process:okf-ingest, at: {INGESTED_AT} }}\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_a_door_c_import_survives_a_same_stem_materialize_run(
    file_setup: tuple[Path, Path],
) -> None:
    # The §3 scan used to glob every `*.md` regardless of which door wrote it,
    # then unconditionally unlink whatever `_is_ingest_owned` agreed to. The
    # line-oriented parser has no indentation model (pinned in
    # test_two_nested_block_mappings_sharing_a_key_collide_in_the_scalar_parser),
    # so a nested block in an externally-imported concept that happens to
    # share a key name with the ownership markers gets promoted to a
    # top-level `generated`/`ingest_manifest` pair and can spoof ownership —
    # deleting content this door never wrote and does not own.
    manifest_path, bundle = file_setup
    bundle.mkdir()
    imported = bundle / "import-external.md"
    imported.write_text(
        "---\n"
        "type: Concept\n"
        "title: Imported Concept\n"
        "provenance:\n"
        "  generated: true\n"
        "  ingest_manifest: manifest@deadbeef\n"
        "---\n\nExternally imported body.\n",
        encoding="utf-8",
    )

    materialize_bundle(manifest_path, bundle, INGESTED_AT)

    assert imported.is_file()
    assert (bundle / "ingest-orders.md").is_file()


def test_a_door_c_import_forging_the_o2_stamp_survives_a_materialize_run(
    file_setup: tuple[Path, Path],
) -> None:
    """The test above with the value V1 made forgeable. ingest-spec §11 lists
    "Stamp integrity (curated writers)" as a red condition, and V1 changed WHICH
    value satisfies it: the forgeable stamp used to be the literal `true`, and is
    now the O2 mapping naming the ingest actor. The old test keeps exercising a
    value `DEFAULT` still owns, so it did not stop testing anything — but on its
    own it would leave the current threat unexercised.

    The defence is structural rather than a judgement about the value: the §3
    scan globs `ingest-*.md`, so a `import-`-prefixed file is never a candidate
    no matter how perfectly it forges the stamp. Door C writes external concepts
    verbatim and cannot screen this, which is exactly why the namespace, not the
    parse, has to be what holds.
    """
    manifest_path, bundle = file_setup
    bundle.mkdir()
    imported = bundle / "import-external.md"
    forged = (
        "---\n"
        "type: Concept\n"
        "title: Imported Concept\n"
        "provenance:\n"
        f"  generated: {{ by: process:okf-ingest, at: {INGESTED_AT} }}\n"
        "  ingest_manifest: manifest@deadbeef\n"
        "---\n\nExternally imported body.\n"
    )
    imported.write_text(forged, encoding="utf-8")

    materialize_bundle(manifest_path, bundle, INGESTED_AT)

    assert imported.is_file()
    assert imported.read_text(encoding="utf-8") == forged
    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()