1
0
Fork 0

feat(ingest): I3 — D7-speil av ingest (filkatalog/CSV), bygget fra commons-spec alene

Speiler MAF I2 fra shared/ingest-spec.md alene: manifest → CSV-konnektor →
materialisert OKF-bundle, byte-identisk med den delte golden-fasiten.

- ingest.py: ManifestContract (pydantic, fail-fast, file-kilde, verdict-reservasjon
  §3, id-grammatikk, max_rows), CSV-konnektor (boundary-checked fail-closed),
  materialisering (§5-frontmatter eksakt rekkefølge, markdown-tabell m/ escaping,
  LF-only, SHA-256 manifest-stamp), index-generering (§6), replacement §3/§5.
- okf.py: _parse_index_entry — tolererer frontmatterløs index (method-spec §3:
  index rendres via body = summary, ikke som typet concept-fil). Golden var
  spec-konform; D7-okf var strengere enn standarden. Scoped: non-index concept-
  filer krever fortsatt type (honesty-test).
- examples/ingest-golden-file/: repo-lokal golden (byte-frossen kopi av I2s fasit).
- Speiltester (I2s load-bearing-sett, alle detach-bevist røde): golden byte-fasit
  + mutasjonskontroller · provenance/navigability/verdict-reservasjon/re-ingest-safety
  · kontrakt fail-fast/max_rows/boundary/kollisjon · spec-integritet §11.
- docs/2026-07-04-I3-brief.md: brief + de to operatør-avgjorte beslutningene.

Suite 239 passed uten nøkkel/nettverk (189 + 50 nye) · ruff + mypy --strict rene.

[skip-docs] README + docs/extending.md er bevisst utsatt til I7 per sesjonsplan
(programmet batcher ingest-doc der, avgrenset til det D7 faktisk har — CSV nå,
SQL/HTTP senere). Endringen er dokumentert i docs/2026-07-04-I3-brief.md.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017MM6BWb1hWmJZuXFZ7rjxT
This commit is contained in:
Kjell Tore Guttormsen 2026-07-04 06:12:43 +02:00
commit e03bd79876
15 changed files with 958 additions and 2 deletions

138
tests/test_ingest.py Normal file
View file

@ -0,0 +1,138 @@
"""Ingest unit + fail-fast contract tests (ingest-spec §4, §5, §8).
The manifest is schema-validated fail-fast BEFORE any source call (§4, the startup-contract
discipline). These tests pin the malformed-manifest rejections, the §5 cell-escaping rules,
and the §8 size cap / fail-closed path resolution / no-overwrite collision.
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from pydantic import ValidationError
from portfolio_optimiser_claude.ingest import (
ManifestContract,
_escape_cell,
load_manifest,
materialize,
)
GOLDEN = Path(__file__).resolve().parents[1] / "examples" / "ingest-golden-file"
INGESTED_AT = "2026-07-03T12:00:00Z"
def _valid() -> dict:
return {
"manifest_version": 1,
"source": {"type": "file", "id": "arkiv", "root": "fixture"},
"bundle_summary": "s",
"extractions": [
{"id": "e", "title": "T", "query": "e.csv", "okf_type": "dataset", "max_rows": 5}
],
}
def _write_case(tmp_path: Path, manifest: dict, csvs: dict[str, str]) -> Path:
case = tmp_path / "case"
(case / "fixture").mkdir(parents=True)
(case / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
for name, text in csvs.items():
(case / "fixture" / name).write_text(text, encoding="utf-8")
return case
class TestManifestValidation:
"""§4: a malformed manifest never starts a run (fail-fast)."""
def test_valid_manifest_loads_and_stamps(self) -> None:
contract = ManifestContract(**_valid())
assert contract.manifest_version == 1
assert contract.source.id == "arkiv"
def test_stamp_is_stem_at_sha256_16(self, tmp_path: Path) -> None:
case = _write_case(tmp_path, _valid(), {"e.csv": "a\n1\n"})
loaded = load_manifest(case / "manifest.json")
stem, _, digest = loaded.stamp.partition("@")
assert stem == "manifest"
assert len(digest) == 16 and all(c in "0123456789abcdef" for c in digest)
@pytest.mark.parametrize(
"mutate",
[
lambda m: m.pop("manifest_version"),
lambda m: m.__setitem__("manifest_version", 2),
lambda m: m.pop("source"),
lambda m: m.pop("bundle_summary"),
lambda m: m.__setitem__("extractions", []),
lambda m: m["source"].__setitem__("id", "Bad_Id"),
lambda m: m["source"].__setitem__("type", "sql"),
lambda m: m["extractions"][0].__setitem__("id", "Bad Id"),
lambda m: m["extractions"][0].__setitem__("title", "two\nlines"),
lambda m: m["extractions"][0].__setitem__("max_rows", 0),
lambda m: m["extractions"][0].__setitem__("max_rows", -1),
],
)
def test_malformed_manifest_is_rejected(self, mutate) -> None:
manifest = _valid()
mutate(manifest)
with pytest.raises(ValidationError):
ManifestContract(**manifest)
def test_duplicate_extraction_ids_are_rejected(self) -> None:
manifest = _valid()
manifest["extractions"].append(dict(manifest["extractions"][0]))
with pytest.raises(ValidationError):
ManifestContract(**manifest)
class TestCellEscaping:
"""§5: text verbatim with backslash → \\\\, pipe → \\|, newline → single space."""
def test_backslash_then_pipe_order(self) -> None:
assert _escape_cell("\\|") == "\\\\\\|"
def test_pipe_escaped(self) -> None:
assert _escape_cell("a|b") == "a\\|b"
def test_newline_becomes_single_space(self) -> None:
assert _escape_cell("x\ny") == "x y"
assert _escape_cell("x\r\ny") == "x y"
def test_plain_text_verbatim(self) -> None:
assert _escape_cell("007") == "007"
assert _escape_cell("1.50") == "1.50"
class TestSecurityFrame:
"""§8: size cap fail-fast, path resolution fail-closed, curated never overwritten."""
def test_extraction_exceeding_max_rows_fails(self, tmp_path: Path) -> None:
manifest = _valid()
manifest["extractions"][0]["max_rows"] = 1
case = _write_case(tmp_path, manifest, {"e.csv": "col\n1\n2\n"}) # 2 data rows > 1
with pytest.raises(ValueError, match="max_rows"):
materialize(case / "manifest.json", tmp_path / "bundle", INGESTED_AT)
def test_query_escaping_root_is_refused(self, tmp_path: Path) -> None:
manifest = _valid()
manifest["extractions"][0]["query"] = "../secret.csv"
case = _write_case(tmp_path, manifest, {"e.csv": "col\n1\n"})
(case / "secret.csv").write_text("col\nx\n", encoding="utf-8")
with pytest.raises(ValueError, match="escapes"):
materialize(case / "manifest.json", tmp_path / "bundle", INGESTED_AT)
def test_collision_with_non_ingest_file_fails(self, tmp_path: Path) -> None:
case = _write_case(tmp_path, _valid(), {"e.csv": "col\n1\n"})
bundle = tmp_path / "bundle"
bundle.mkdir()
# A curated (non-stamped) file already occupies the generated name.
(bundle / "ingest-e.md").write_text(
"---\ntype: reference\ntitle: hand\n---\n\nCurated.\n", encoding="utf-8"
)
with pytest.raises(ValueError, match="collides"):
materialize(case / "manifest.json", bundle, INGESTED_AT)
# The curated file is untouched — never overwritten.
assert "Curated." in (bundle / "ingest-e.md").read_text(encoding="utf-8")

View file

@ -0,0 +1,86 @@
"""Ingest golden regression (ingest-spec §11) — the shared byte-for-byte fasit.
D7 mirror of MAF I2. Consumes the repo-local ``examples/ingest-golden-file/``
(a byte-frozen copy of MAF I2's golden — the SAME shared extraction, so this test
proves D7's independent implementation reproduces MAF's exact bytes from the shared
spec alone). ``expected-bundle/`` is compared file by file, byte for byte.
The mutation controls prove the net is taut: one changed determinism input (a CSV
cell, the timestamp, the manifest bytes) must diverge from the golden bytes.
"""
from __future__ import annotations
import shutil
from pathlib import Path
from portfolio_optimiser_claude.ingest import materialize
GOLDEN = Path(__file__).resolve().parents[1] / "examples" / "ingest-golden-file"
def _ingested_at() -> str:
return (GOLDEN / "ingested-at.txt").read_text(encoding="utf-8").strip()
def test_materializes_golden_byte_for_byte(tmp_path: Path) -> None:
bundle = tmp_path / "bundle"
bundle.mkdir()
materialize(GOLDEN / "manifest.json", bundle, _ingested_at())
expected = GOLDEN / "expected-bundle"
expected_names = sorted(p.name for p in expected.iterdir())
produced_names = sorted(p.name for p in bundle.iterdir())
assert produced_names == expected_names # no missing, no extra files
for name in expected_names:
assert (bundle / name).read_bytes() == (expected / name).read_bytes(), name
def test_reingest_is_idempotent(tmp_path: Path) -> None:
# ingest-spec §10: same source content, manifest, ingested_at → byte-identical,
# repeated runs idempotent.
bundle = tmp_path / "bundle"
bundle.mkdir()
materialize(GOLDEN / "manifest.json", bundle, _ingested_at())
first = {p.name: p.read_bytes() for p in bundle.iterdir()}
materialize(GOLDEN / "manifest.json", bundle, _ingested_at())
second = {p.name: p.read_bytes() for p in bundle.iterdir()}
assert first == second
class TestMutationControl:
"""One changed determinism input → divergence from the golden bytes."""
def _case(self, tmp_path: Path) -> Path:
case = tmp_path / "case"
shutil.copytree(GOLDEN, case)
return case
def test_changed_cell_diverges(self, tmp_path: Path) -> None:
case = self._case(tmp_path)
csv = case / "fixture" / "costs.csv"
csv.write_bytes(csv.read_bytes().replace(b"120000", b"999999"))
bundle = tmp_path / "bundle"
bundle.mkdir()
materialize(case / "manifest.json", bundle, _ingested_at())
golden_bytes = (GOLDEN / "expected-bundle" / "ingest-costs.md").read_bytes()
assert (bundle / "ingest-costs.md").read_bytes() != golden_bytes
def test_changed_timestamp_diverges(self, tmp_path: Path) -> None:
bundle = tmp_path / "bundle"
bundle.mkdir()
materialize(GOLDEN / "manifest.json", bundle, "2099-01-01T00:00:00Z")
golden_bytes = (GOLDEN / "expected-bundle" / "ingest-costs.md").read_bytes()
assert (bundle / "ingest-costs.md").read_bytes() != golden_bytes
def test_changed_manifest_bytes_change_the_stamp(self, tmp_path: Path) -> None:
# The ingest_manifest stamp is SHA-256 of the manifest's raw bytes (§5) —
# a whitespace-only edit still re-stamps every generated file.
case = self._case(tmp_path)
manifest = case / "manifest.json"
manifest.write_bytes(manifest.read_bytes() + b"\n")
bundle = tmp_path / "bundle"
bundle.mkdir()
materialize(manifest, bundle, _ingested_at())
golden_bytes = (GOLDEN / "expected-bundle" / "ingest-costs.md").read_bytes()
assert (bundle / "ingest-costs.md").read_bytes() != golden_bytes

View file

@ -0,0 +1,178 @@
"""Load-bearing ingest seams (ingest-spec §11) — D7 mirror of MAF I2's set.
Each test must go RED when its seam is detached (the method-spec §11 regime): a
grønn-men-død test is the failure mode the rule exists for. The seams mirrored here:
- Provenance stamping a generated file carries the §7 provenance layer, in order.
- Navigability the generated bundle is consumable by the UNCHANGED ``okf`` navigation
(index links included), incl. the frontmatter-less generated index.
- Verdict reservation a manifest mapping to ``type: verdict`` is rejected fail-fast,
before any source call.
- Re-ingest layer safety re-materialization over a bundle carrying a promoted verdict
preserves the verdict file AND its index link (curated content always survives).
"""
from __future__ import annotations
import json
import shutil
from pathlib import Path
import pytest
from pydantic import ValidationError
from portfolio_optimiser_claude import okf
from portfolio_optimiser_claude.ingest import ManifestContract, load_manifest, materialize
GOLDEN = Path(__file__).resolve().parents[1] / "examples" / "ingest-golden-file"
INGESTED_AT = (GOLDEN / "ingested-at.txt").read_text(encoding="utf-8").strip()
_PROVENANCE_KEYS = ("source_system", "source_query", "ingested_at", "ingest_manifest", "generated")
def _materialized(tmp_path: Path) -> Path:
bundle = tmp_path / "bundle"
bundle.mkdir()
materialize(GOLDEN / "manifest.json", bundle, INGESTED_AT)
return bundle
class TestProvenanceStamping:
"""Seam: a generated file carries the §7 provenance layer (RED if it stops)."""
def test_generated_file_carries_the_provenance_layer(self, tmp_path: Path) -> None:
bundle = _materialized(tmp_path)
concept = okf.parse_concept_file(bundle / "ingest-costs.md")
for key in _PROVENANCE_KEYS:
assert key in concept.frontmatter, f"provenance key {key} detached"
assert concept.frontmatter["source_system"] == "prosjekt-arkiv"
assert concept.frontmatter["ingested_at"] == INGESTED_AT
assert concept.frontmatter["generated"] == "true"
stamp = load_manifest(GOLDEN / "manifest.json").stamp
assert concept.frontmatter["ingest_manifest"] == stamp
def test_provenance_keys_are_in_the_spec_order(self, tmp_path: Path) -> None:
# §5: exactly these keys, in exactly this order — the chain is a contract.
bundle = _materialized(tmp_path)
lines = (bundle / "ingest-costs.md").read_text(encoding="utf-8").splitlines()
keys = [ln.split(":", 1)[0] for ln in lines[1 : lines.index("---", 1)]]
assert keys == [
"type",
"title",
"source_system",
"source_query",
"ingested_at",
"ingest_manifest",
"generated",
]
class TestNavigability:
"""Seam: the generated bundle is consumable by the UNCHANGED okf navigation."""
def test_generated_bundle_navigates_via_unchanged_okf(self, tmp_path: Path) -> None:
bundle = _materialized(tmp_path)
names = [c.path.name for c in okf.navigate_bundle(bundle)]
# Every generated ingest file must be REACHABLE via index cross-links — an
# unlinked generated file is unreachable (RED if index linking detaches).
assert names == ["index.md", "ingest-costs.md", "ingest-edge.md"]
def test_generated_context_renders_the_extracted_rows(self, tmp_path: Path) -> None:
bundle = _materialized(tmp_path)
context = okf.bundle_context(bundle)
assert "led-retrofit" in context and "requires vendor quote" in context
def test_generated_index_is_frontmatterless_and_still_navigates(self, tmp_path: Path) -> None:
# The generated index carries no frontmatter (§6 shape) — navigation must
# still work, exercising the okf entry-point relaxation end to end.
bundle = _materialized(tmp_path)
assert not (bundle / "index.md").read_text(encoding="utf-8").startswith("---")
assert okf.navigate_bundle(bundle)[0].path.name == "index.md"
class TestVerdictReservation:
"""Seam: an ingest mapping to the verdict layer is rejected fail-fast (§3)."""
def _manifest(self, okf_type: str) -> dict:
return {
"manifest_version": 1,
"source": {"type": "file", "id": "x", "root": "fixture"},
"bundle_summary": "s",
"extractions": [
{"id": "e", "title": "T", "query": "e.csv", "okf_type": okf_type, "max_rows": 1}
],
}
def test_verdict_okf_type_is_rejected(self) -> None:
with pytest.raises(ValidationError):
ManifestContract(**self._manifest("verdict"))
def test_verdict_reservation_is_case_insensitive(self) -> None:
with pytest.raises(ValidationError):
ManifestContract(**self._manifest("Verdict"))
def test_rejection_is_fail_fast_before_any_source_call(self, tmp_path: Path) -> None:
# A verdict manifest never touches the source: no bundle is written.
manifest = tmp_path / "manifest.json"
manifest.write_text(json.dumps(self._manifest("verdict")), encoding="utf-8")
bundle = tmp_path / "bundle"
with pytest.raises(ValidationError):
materialize(manifest, bundle, INGESTED_AT)
assert not bundle.exists() or not any(bundle.iterdir())
class TestReingestLayerSafety:
"""Seam: re-ingest preserves a promoted verdict AND its index link (§3, §6)."""
def _promote(self, bundle: Path) -> None:
# Simulate the promotion gate writing a verdict file + its (neutral) index link,
# plus a curated file + link. Neither carries the ingest stamp.
(bundle / "promoted-verdict-led.md").write_text(
"---\ntype: verdict\ndecision: approved\ndescription: LED holds up.\n---\n\nBody.\n",
encoding="utf-8",
)
(bundle / "curated-note.md").write_text(
"---\ntype: reference\ntitle: Curated\n---\n\nHand-written.\n", encoding="utf-8"
)
index = bundle / "index.md"
index.write_text(
index.read_text(encoding="utf-8")
+ "- [promoted](promoted-verdict-led.md)\n"
+ "- [Curated](curated-note.md)\n",
encoding="utf-8",
)
def test_promoted_verdict_and_curated_survive_reingest(self, tmp_path: Path) -> None:
bundle = _materialized(tmp_path)
self._promote(bundle)
materialize(GOLDEN / "manifest.json", bundle, INGESTED_AT) # re-ingest
assert (bundle / "promoted-verdict-led.md").is_file() # verdict file survives
assert (bundle / "curated-note.md").is_file() # curated file survives
index_text = (bundle / "index.md").read_text(encoding="utf-8")
assert "- [promoted](promoted-verdict-led.md)" in index_text # verdict link survives
assert "- [Curated](curated-note.md)" in index_text # curated link survives
# ...and the ingest files were still refreshed.
assert (bundle / "ingest-costs.md").is_file()
assert "](ingest-costs.md)" in index_text
def test_reingest_drops_a_stale_ingest_file_and_its_link(self, tmp_path: Path) -> None:
# A manifest that no longer generates ingest-edge → the stale file AND its index
# link are removed; the promoted/curated content is untouched (RED if replacement
# over-reaches or under-reaches).
bundle = _materialized(tmp_path)
self._promote(bundle)
# The shrunk manifest must sit beside its own fixture/ (root resolves relative to
# the manifest dir), so copy the golden case and rewrite the manifest there.
case = tmp_path / "case"
shutil.copytree(GOLDEN, case)
shrunk = json.loads((case / "manifest.json").read_text(encoding="utf-8"))
shrunk["extractions"] = [e for e in shrunk["extractions"] if e["id"] == "costs"]
(case / "manifest.json").write_text(json.dumps(shrunk), encoding="utf-8")
materialize(case / "manifest.json", bundle, INGESTED_AT)
assert not (bundle / "ingest-edge.md").exists() # stale ingest file removed
index_text = (bundle / "index.md").read_text(encoding="utf-8")
assert "](ingest-edge.md)" not in index_text # stale ingest link removed
assert "- [promoted](promoted-verdict-led.md)" in index_text # promoted survives
assert (bundle / "ingest-costs.md").is_file() # kept extraction survives

View file

@ -0,0 +1,70 @@
"""Spec-integrity seam for the ingest spec (ingest-spec §11).
The D7 analog of MAF's I1 framework-guard: this repo consumes ``shared/ingest-spec.md``
UNCHANGED from commons, and this test keeps the contract honest it goes RED when the
spec goes missing, names a concrete agent toolkit (the framework-neutrality rule), or
stops documenting a contract field. It is the load-bearing guard the ingest layer relies
on to keep being implementable "from this spec alone".
"""
from __future__ import annotations
from pathlib import Path
import pytest
SPEC = Path(__file__).resolve().parents[1] / "shared" / "ingest-spec.md"
# Concrete agent toolkits / vendor stacks the framework-neutral spec MUST NOT name.
_FORBIDDEN_TOOLKITS = (
"claude",
"anthropic",
"openai",
"gpt",
"gemini",
"llama",
"langchain",
"autogen",
"crewai",
"semantic kernel",
"microsoft agent framework",
"agent sdk",
"bedrock",
"vertex",
"foundry",
"maf",
)
# Every field of the machine-readable contracts the D7 implementation depends on — the
# spec's §12 cross-check table must keep documenting each (spec-integrity).
_CONTRACT_FIELDS = (
"manifest_version",
"source",
"bundle_summary",
"extractions",
"source_system",
"source_query",
"ingested_at",
"ingest_manifest",
"generated",
"okf_type",
"max_rows",
"root",
)
def test_spec_is_present() -> None:
# RED if the spec goes missing (the layer stops being implementable from spec alone).
assert SPEC.is_file(), "ingest-spec.md missing — subtree pull the commons contract"
def test_spec_names_no_agent_toolkit() -> None:
text = SPEC.read_text(encoding="utf-8").lower()
present = [tok for tok in _FORBIDDEN_TOOLKITS if tok in text]
assert not present, f"framework-neutral spec names a concrete toolkit: {present}"
@pytest.mark.parametrize("field", _CONTRACT_FIELDS)
def test_spec_documents_contract_field(field: str) -> None:
text = SPEC.read_text(encoding="utf-8")
assert field in text, f"contract field {field!r} is no longer documented in the spec"

View file

@ -54,6 +54,32 @@ class TestNavigation:
names = [c.path.name for c in navigate_bundle(bundle)]
assert names == ["index.md", "b.md", "a.md"]
def test_frontmatterless_index_navigates(self, tmp_path: Path) -> None:
# method-spec §3 Step 1: the index contributes "the index body (the summary)"
# and is NOT a "non-index concept file" rendered as a `## {type}: {title}`
# section — so a generated index carrying only the summary + links (ingest
# spec §6 shape, frozen in the ingest golden) is valid. The index MAY omit
# frontmatter; okf must navigate it via its body, never raise.
_write(tmp_path / "index.md", "Summary line.\n- [A](a.md)\n")
_write(tmp_path / "a.md", "---\ntype: project\ntitle: A\n---\nBody A.")
concepts = navigate_bundle(tmp_path)
assert [c.path.name for c in concepts] == ["index.md", "a.md"]
index = concepts[0]
assert index.type == "index" # default type, no KeyError for downstream .type reads
assert index.body == "Summary line.\n- [A](a.md)" # body = the whole file
# The summary flows into the read-context as the leading section.
assert bundle_context(tmp_path).startswith("Summary line.")
def test_frontmatterless_tolerance_is_scoped_to_the_index(self, tmp_path: Path) -> None:
# LOAD-BEARING (honesty): the relaxation is for the index ENTRY POINT only.
# A NON-index concept file without frontmatter is still malformed and MUST
# raise — else the relaxation has silently weakened the `type`-required rule
# for concept files (method-spec §3 Step 1). RED if the tolerance leaks.
_write(tmp_path / "index.md", "Summary.\n- [A](a.md)\n")
_write(tmp_path / "a.md", "no frontmatter here\n")
with pytest.raises(ValueError, match="frontmatter"):
navigate_bundle(tmp_path)
def test_out_of_bundle_and_escaping_targets_are_skipped(self, tmp_path: Path) -> None:
# A target containing a path separator is out-of-bundle — skipped, never
# raised; ../-escapes never resolve outside the bundle (fail-closed).