feat(ingest): index generation and stamped-replacement semantics (I2)

This commit is contained in:
Kjell Tore Guttormsen 2026-07-03 18:33:56 +02:00
commit 676c11a498
2 changed files with 188 additions and 1 deletions

View file

@ -250,3 +250,105 @@ def test_sql_source_has_no_connector_in_i2(tmp_path: Path) -> None:
)
with pytest.raises(IngestError):
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
# --- Step 4: index generation + stamped-replacement semantics ---
_TWO_EXTRACTIONS = [
{"id": "costs", "title": "Project costs", "query": "costs.csv", "okf_type": "dataset", "max_rows": 100},
{"id": "meta", "title": "Catalogue metadata", "query": "meta.csv", "okf_type": "reference", "max_rows": 10},
]
_TWO_FILES = {
"costs.csv": b"item,cost\nled,120\n",
"meta.csv": b"key,value\nowner,anlegg\n",
}
def _bundle_bytes(bundle_dir: Path) -> dict[str, bytes]:
return {p.name: p.read_bytes() for p in sorted(bundle_dir.iterdir()) if p.is_file()}
def test_fresh_dir_gets_index_with_bundle_summary_and_links_in_manifest_order(
tmp_path: Path,
) -> None:
manifest_path, bundle_dir = _project(
tmp_path, extractions=copy.deepcopy(_TWO_EXTRACTIONS), files=dict(_TWO_FILES)
)
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
index = (bundle_dir / "index.md").read_text(encoding="utf-8")
assert index.startswith("Cost extracts from the project archive.\n")
link_lines = [line for line in index.splitlines() if line.startswith("- [")]
assert link_lines == [
"- [Project costs](ingest-costs.md)",
"- [Catalogue metadata](ingest-meta.md)",
]
def test_curated_index_line_preserved_byte_for_byte(tmp_path: Path) -> None:
manifest_path, bundle_dir = _project(tmp_path)
bundle_dir.mkdir()
curated = "Hand-written summary.\n\n- [Curated note](note.md)\n"
(bundle_dir / "index.md").write_bytes(curated.encode("utf-8"))
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
index = (bundle_dir / "index.md").read_text(encoding="utf-8")
assert index.startswith(curated) # §6: every unmanaged line preserved verbatim
assert "- [Project costs](ingest-costs.md)" in index
def test_removed_extraction_drops_file_and_index_link(tmp_path: Path) -> None:
manifest_path, bundle_dir = _project(
tmp_path, extractions=copy.deepcopy(_TWO_EXTRACTIONS), files=dict(_TWO_FILES)
)
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
# Re-materialize with `meta` dropped from the manifest.
manifest_path2, _ = _project(tmp_path, files=dict(_TWO_FILES))
materialize(manifest_path2, bundle_dir, ingested_at=_INGESTED_AT)
assert not (bundle_dir / "ingest-meta.md").exists()
index = (bundle_dir / "index.md").read_text(encoding="utf-8")
assert "](ingest-meta.md)" not in index # §6: link to removed ingest file is gone
assert "- [Project costs](ingest-costs.md)" in index # sibling link intact
def test_changed_title_refreshes_label_in_place(tmp_path: Path) -> None:
manifest_path, bundle_dir = _project(
tmp_path, extractions=copy.deepcopy(_TWO_EXTRACTIONS), files=dict(_TWO_FILES)
)
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
renamed = copy.deepcopy(_TWO_EXTRACTIONS)
renamed[0]["title"] = "Renamed costs"
manifest_path2, _ = _project(tmp_path, extractions=renamed, files=dict(_TWO_FILES))
materialize(manifest_path2, bundle_dir, ingested_at=_INGESTED_AT)
link_lines = [
line
for line in (bundle_dir / "index.md").read_text(encoding="utf-8").splitlines()
if line.startswith("- [")
]
# §6: the label IS the title — refreshed in place, position preserved (no stale label,
# no move-to-end).
assert link_lines == [
"- [Renamed costs](ingest-costs.md)",
"- [Catalogue metadata](ingest-meta.md)",
]
def test_unstamped_collision_fails_and_leaves_bundle_unmodified(tmp_path: Path) -> None:
manifest_path, bundle_dir = _project(tmp_path)
bundle_dir.mkdir()
(bundle_dir / "index.md").write_bytes(b"Curated.\n")
# A curated file occupying a generated filename WITHOUT the ingest stamp (§3): never
# overwrite curated content.
(bundle_dir / "ingest-costs.md").write_bytes(b"---\ntype: note\n---\n\nCurated body.\n")
before = _bundle_bytes(bundle_dir)
with pytest.raises(IngestError):
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
assert _bundle_bytes(bundle_dir) == before
def test_rematerialization_twice_is_byte_identical_including_index(tmp_path: Path) -> None:
manifest_path, bundle_dir = _project(
tmp_path, extractions=copy.deepcopy(_TWO_EXTRACTIONS), files=dict(_TWO_FILES)
)
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
first = _bundle_bytes(bundle_dir)
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
assert _bundle_bytes(bundle_dir) == first # §10, index.md included