llm-ingestion-okf/tests/test_load_bearing.py

"""The remaining load-bearing seams of the §11 table, plus the public API.

Every seam in the spec §11 table has one named test that MUST fail when its
seam is detached (verified by hand-mutation, noted per docstring):

| Seam | Named test |
|---|---|
| Provenance stamping | test_provenance_layer_on_every_generated_file (here) |
| Navigability | test_every_generated_file_reachable_via_index_links (here) |
| Verdict reservation | test_verdict_okf_type_rejected (test_manifest.py) |
| Re-ingest layer safety | test_promoted_verdict_and_its_link_survive_reingest (test_index.py) |
| Golden regression | test_golden_case_byte_for_byte (test_golden.py) |
| Network gate | test_network_gate_blocks_before_any_transport_call (test_http_connector.py) |

(The spec-integrity seam belongs to commons, which owns the spec.)
"""

from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any

from llm_ingestion_okf.materialize import materialize_bundle

INGESTED_AT = "2026-07-16T12:00:00Z"

# Independent §7 parsing — deliberately NOT the library's own frontmatter
# parser, so a symmetric render/parse defect cannot mask a missing layer.
_FM_LINE = re.compile(r"^(?P<key>[a-z_]+): (?P<value>.*)$")
_INDEX_LINK = re.compile(r"^- \[[^\]]*\]\((?P<target>[^)]+)\)$", re.MULTILINE)


def parse_frontmatter_independently(path: Path) -> dict[str, str]:
    lines = path.read_text(encoding="utf-8").split("\n")
    assert lines[0] == "---", f"{path.name} does not start with a frontmatter block"
    frontmatter: dict[str, str] = {}
    for line in lines[1:]:
        if line == "---":
            return frontmatter
        match = _FM_LINE.match(line)
        assert match, f"malformed frontmatter line in {path.name}: {line!r}"
        frontmatter[match.group("key")] = match.group("value")
    raise AssertionError(f"{path.name} frontmatter block never closes")


def materialized_bundle(tmp_path: Path) -> tuple[Path, tuple[Path, ...]]:
    data: dict[str, Any] = {
        "manifest_version": 1,
        "source": {"type": "file", "id": "catalogue-1", "root": "data"},
        "bundle_summary": "A test bundle.",
        "extractions": [
            {
                "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,
            },
        ],
    }
    src = tmp_path / "src"
    (src / "data").mkdir(parents=True)
    (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="")
    manifest_path = src / "manifest.json"
    manifest_path.write_text(json.dumps(data), encoding="utf-8")
    bundle = tmp_path / "bundle"
    result = materialize_bundle(manifest_path, bundle, INGESTED_AT)
    return bundle, result.written


def test_provenance_layer_on_every_generated_file(tmp_path: Path) -> None:
    """Load-bearing (spec §11, seam: provenance stamping).

    MUST fail when a generated file no longer carries the §7 layer.
    Hand-mutation check 2026-07-16: dropping any of the five §7 keys from
    _render_concept_file turns this test red.

    `generated` is spelled out here rather than read from the profile, for the
    same reason the frontmatter parser is: a seam that asks the library what it
    should have written cannot catch the library writing the wrong thing. §7
    fixes both sub-keys — the actor is the constant `process:okf-ingest`, and
    `at` repeats `ingested_at` verbatim — so the expected value is derivable by
    hand, which is what makes stating it independently possible at all.
    """
    _, written = materialized_bundle(tmp_path)
    assert written
    for path in written:
        frontmatter = parse_frontmatter_independently(path)
        assert frontmatter["source_system"] == "catalogue-1"
        assert frontmatter["source_query"]
        assert frontmatter["ingested_at"] == INGESTED_AT
        assert re.fullmatch(r"manifest@[0-9a-f]{16}", frontmatter["ingest_manifest"])
        assert frontmatter["generated"] == f"{{ by: process:okf-ingest, at: {INGESTED_AT} }}"


def test_every_generated_file_reachable_via_index_links(tmp_path: Path) -> None:
    """Load-bearing (spec §11, seam: navigability).

    Navigation follows ONLY index cross-links — a generated concept file
    without an index link is unreachable, so every written file must have a
    link whose target resolves. Hand-mutation check 2026-07-16: skipping the
    _link_in_index call in materialize_bundle turns this test red.
    """
    bundle, written = materialized_bundle(tmp_path)
    index = (bundle / "index.md").read_text(encoding="utf-8")
    targets = {match.group("target") for match in _INDEX_LINK.finditer(index)}
    for target in targets:
        assert (bundle / target).is_file(), f"index link target {target!r} does not resolve"
    for path in written:
        assert path.name in targets, f"generated file {path.name!r} has no index link"


def test_public_api_importable_from_package_root() -> None:
    """The library's public surface: one entry point plus the typed errors."""
    import llm_ingestion_okf as pkg

    assert callable(pkg.materialize_bundle)
    assert issubclass(pkg.ManifestError, pkg.IngestError)
    assert issubclass(pkg.SourceError, pkg.IngestError)
    assert issubclass(pkg.RenderError, pkg.IngestError)
    assert issubclass(pkg.MaterializationError, pkg.IngestError)
    assert issubclass(pkg.NetworkGateError, pkg.IngestError)
    assert pkg.IngestResult is not None