Kjell Tore Guttormsen
1747238a83
The §3 ownership scan classified every ingest-stamped file as replaceable,
so materializing one manifest into a bundle deleted the ingest files another
manifest had stamped there — and removed their index links too.
Narrow ownership to files whose stamp names the running manifest by stem.
The stamp is `{stem}@{sha256[:16]}`; matching on the stem, not the whole
stamp, lets an edited manifest still reclaim the files a prior run of the
same manifest wrote (the stem is stable across content edits), while a
different manifest sharing the bundle keeps its own files. Implements the
trinn-e §10.2 decision. This does not close the operator-copy restriction
(a generated file copied into curated content), which stays documented in
ingest-spec.md:69-72, not enforced.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HBbjgS5A55RVavoyjJC4FX
491 lines
18 KiB
Python
491 lines
18 KiB
Python
"""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()
|