From 6236f2fb78439acce9ec8d87c224f70b9d761200 Mon Sep 17 00:00:00 2001 From: Kjell Tore Guttormsen Date: Fri, 3 Jul 2026 18:37:30 +0200 Subject: [PATCH] test(ingest): load-bearing seam quartet + MAF-free guard, detach-proven (I2) --- tests/test_ingest_loadbearing.py | 258 +++++++++++++++++++++++++++++++ 1 file changed, 258 insertions(+) create mode 100644 tests/test_ingest_loadbearing.py diff --git a/tests/test_ingest_loadbearing.py b/tests/test_ingest_loadbearing.py new file mode 100644 index 0000000..80cde0b --- /dev/null +++ b/tests/test_ingest_loadbearing.py @@ -0,0 +1,258 @@ +"""I2 load-bearing seams (ingest spec §11's table; the method-spec §11 detach-RED regime). + +Each test names its seam and the detach that turns it RED — scenarios are built so the seam +under test is the ONLY thing preventing the observable artifact (the Fase-2 green-but-dead +trap). Pattern: tests/test_step8_promotion_loadbearing.py. + +- VERDICT RESERVATION (§3): a manifest mapping to ``type: verdict`` — otherwise fully valid + and materializable, so detaching the validator WOULD produce a verdict file + link — is + rejected at load, before any source call, and writes NOTHING. RED when the + case-insensitive ``okf_type`` validator is removed. (The reserved-FILENAME half of §3 — + ``index.md``/``promoted-verdict-*`` — is structurally unreachable: the §4 id grammar plus + the mandatory ``ingest-`` prefix make no valid manifest able to produce those names. It + is asserted below as a defensive invariant; its load-bearing bite is the okf_type check.) +- PROVENANCE STAMPING (§7): every generated file carries the 7-key layer, readable by the + UNCHANGED ``okf.parse_frontmatter``. RED when the stamp keys are dropped from rendering. +- NAVIGABILITY (§2/§11): the generated bundle is consumable by the UNCHANGED bundle + navigation — every ``ingest-{id}.md`` reachable via index cross-links, classified by its + ``okf_type``, provenance riding through unknown-field preservation, content rendered by + ``bundle_context``. RED when index linking is detached (file unreachable). +- RE-INGEST LAYER SAFETY (§3/§6/§10) — with the removal filter ACTIVE: the second + materialization uses a REDUCED manifest so the §6 managed-line removal actually fires + while a REAL promoted verdict (``verdicts.promote_verdict``) and the fixture's curated + links are present. An identical-manifest second run never removes anything and would + stay green under a loose filter — exactly the trap this test closes. RED when the + removal/ownership filter over-matches (a bare-substring filter deletes the promoted + managed-form line) or replacement stops honouring the ingest stamp. +- MAF-FREE MODULE (context-seam purity): ``ingest.py`` imports no ``agent_framework``/ + ``mcp`` and no MAF-coupled project modules (only ``okf``/``retrieval``, both + stdlib-pure) — mirrors ``tests/test_okf.py::test_okf_is_maf_free``, which covers only + ``okf.py``. RED the moment e.g. ``contracts`` (→ ``backends`` → MAF) is imported. +""" + +from __future__ import annotations + +import ast +import copy +import json +import shutil +from pathlib import Path +from typing import Any + +import pytest +from pydantic import ValidationError + +from portfolio_optimiser import okf +from portfolio_optimiser.ingest import Extraction, materialize +from portfolio_optimiser.verdicts import Verdict, bundle_candidate_features, promote_verdict + +REPO_ROOT = Path(__file__).resolve().parents[1] +CURATED_BUNDLE = REPO_ROOT / "shared" / "examples" / "bygg-energi-mikro" +INGEST_SOURCE = REPO_ROOT / "src" / "portfolio_optimiser" / "ingest.py" + +_INGESTED_AT = "2026-07-03T12:00:00Z" + +_EXTRACTIONS: list[dict[str, Any]] = [ + {"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}, +] +_FILES = { + "costs.csv": b"item,cost\nled,120\n", + "meta.csv": b"key,value\nowner,anlegg\n", +} + + +def _write_project(tmp_path: Path, extractions: list[dict[str, Any]]) -> Path: + catalogue = tmp_path / "catalogue" + catalogue.mkdir(exist_ok=True) + for name, content in _FILES.items(): + (catalogue / name).write_bytes(content) + manifest = { + "manifest_version": 1, + "source": {"type": "file", "id": "prosjekt-arkiv", "root": "catalogue"}, + "bundle_summary": "Cost extracts from the project archive.", + "extractions": extractions, + } + name = f"manifest-{len(extractions)}.json" + manifest_path = tmp_path / name + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + return manifest_path + + +# --- Seam 1: verdict reservation (§3) ------------------------------------------------------------ + + +def test_verdict_typed_manifest_is_rejected_and_writes_nothing(tmp_path: Path) -> None: + """LOAD-BEARING RESERVATION: the manifest is otherwise fully materializable (real CSVs, + writable target), so detaching the okf_type validator WOULD produce a machine-generated + ``type: verdict`` file that the next run's seeding turns into an "approved" store entry + around the gate. RED when the validator is removed.""" + extractions = copy.deepcopy(_EXTRACTIONS) + extractions[0]["okf_type"] = "Verdict" # case-insensitive rejection (§3/§4) + manifest_path = _write_project(tmp_path, extractions) + bundle_dir = tmp_path / "bundle" + + with pytest.raises(ValidationError): + materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT) + + assert not bundle_dir.exists(), "a verdict-typed manifest must write NOTHING (no index, no files)" + + +def test_reserved_filename_namespace_is_structurally_unreachable() -> None: + """Defensive invariant (§3, honestly non-detachable): for EVERY valid extraction id the + generated name ``ingest-{id}.md`` is disjoint from ``index.md`` and ``promoted-verdict-*`` + by construction — no manifest input can exercise a filename collision with the reserved + namespace, so the load-bearing bite lives in the okf_type test above.""" + extraction = Extraction( + id="promoted-verdict-x", title="t", query="q.csv", okf_type="dataset", max_rows=1 + ) + name = f"ingest-{extraction.id}.md" + assert name.startswith("ingest-") and name != "index.md" + assert not name.startswith("promoted-verdict-") + + +# --- Seam 2: provenance stamping (§7) ------------------------------------------------------------- + + +def test_every_generated_file_carries_the_provenance_layer(tmp_path: Path) -> None: + """LOAD-BEARING PROVENANCE: the §7 layer (all seven keys, ``generated: true``, the + ``{stem}@{hash16}`` manifest reference) is present on EVERY generated file, readable by + the unchanged ``okf.parse_frontmatter``. RED when the stamp keys are dropped.""" + manifest_path = _write_project(tmp_path, copy.deepcopy(_EXTRACTIONS)) + written = materialize(manifest_path, tmp_path / "bundle", ingested_at=_INGESTED_AT) + + assert len(written) == 2 + for path in written: + fm = okf.parse_frontmatter(path) + for key in ( + "type", + "title", + "source_system", + "source_query", + "ingested_at", + "ingest_manifest", + "generated", + ): + assert key in fm, f"provenance key {key!r} missing from {path.name}" + assert fm["generated"] == "true" + assert fm["source_system"] == "prosjekt-arkiv" + assert fm["ingested_at"] == _INGESTED_AT + stem, _, digest = fm["ingest_manifest"].partition("@") + assert stem == manifest_path.stem and len(digest) == 16 + + +# --- Seam 3: navigability via UNCHANGED okf.py (§2/§11) ------------------------------------------- + + +def test_generated_bundle_is_navigable_via_unchanged_okf(tmp_path: Path) -> None: + """LOAD-BEARING NAVIGABILITY (durable version of the planning session's ad-hoc proof): + ``okf.navigate_bundle`` — untouched by I2 — reaches every generated file through index + cross-links, classifies it by okf_type, preserves the provenance fields (unknown-field + preservation, OKF §4), and ``bundle_context`` renders the extracted content. RED when + index linking is detached: the file exists but is unreachable (navigation follows ONLY + index cross-links).""" + manifest_path = _write_project(tmp_path, copy.deepcopy(_EXTRACTIONS)) + bundle_dir = tmp_path / "bundle" + materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT) + + bundle = okf.navigate_bundle(str(bundle_dir)) + by_name = {f.name: f for f in bundle.files} + assert "ingest-costs.md" in by_name and "ingest-meta.md" in by_name + assert by_name["ingest-costs.md"].type == "dataset" + assert by_name["ingest-meta.md"].type == "reference" + assert by_name["ingest-costs.md"].frontmatter["ingest_manifest"] # provenance rides through + context = okf.bundle_context(bundle) + assert "| led | 120 |" in context and "Project costs" in context + + +# --- Seam 4: re-ingest layer safety, removal filter ACTIVE (§3/§6/§10) ---------------------------- + + +def test_reingest_with_active_removal_preserves_promoted_and_curated(tmp_path: Path) -> None: + """LOAD-BEARING LAYER SAFETY: materialize into a copy of the curated fixture bundle, + promote a REAL verdict through ``verdicts.promote_verdict``, then re-materialize with a + REDUCED manifest so the §6 removal filter actually fires. The dropped extraction's file + and link vanish; the promoted verdict's file and exact managed-form index line — and + every curated byte — survive. RED when the filter over-matches (bare substring) or the + stamp ownership rule is detached.""" + bundle_dir = tmp_path / "bundle" + shutil.copytree(CURATED_BUNDLE, bundle_dir) # promotion + ingest WRITE: throwaway copy + curated_index_lines = (bundle_dir / "index.md").read_text(encoding="utf-8").splitlines() + + manifest_two = _write_project(tmp_path, copy.deepcopy(_EXTRACTIONS)) + materialize(manifest_two, bundle_dir, ingested_at=_INGESTED_AT) + + promoted_path = promote_verdict( + str(bundle_dir), + Verdict( + id="I2-REINGEST", + proposal_features=bundle_candidate_features(str(bundle_dir)), + decision="approved", + rationale="godkjent — layer-safety-fixture", + ), + approver="persona", + experiment="i2-reingest", + timestamp="2026-07-03", + ) + promoted_line = f"- [Promotert ekspert-vurdering (gated)]({promoted_path.name})" + curated_bytes_before = { + p.name: p.read_bytes() + for p in sorted(bundle_dir.glob("*.md")) + if not p.name.startswith("ingest-") and p.name != "index.md" + } + + manifest_one = _write_project(tmp_path, copy.deepcopy(_EXTRACTIONS)[:1]) # drop `meta` + materialize(manifest_one, bundle_dir, ingested_at=_INGESTED_AT) + + index_after = (bundle_dir / "index.md").read_text(encoding="utf-8") + assert not (bundle_dir / "ingest-meta.md").exists() + assert "](ingest-meta.md)" not in index_after, "dropped extraction's link must be removed" + assert (bundle_dir / "ingest-costs.md").exists() + assert promoted_path.exists(), "re-ingest deleted the promoted verdict file (§3 violated)" + assert promoted_line in index_after.splitlines(), ( + "the promoted verdict's index link did not survive an ACTIVE removal pass — the " + "filter over-matched the managed-line shape promote_verdict writes" + ) + for line in curated_index_lines: + assert line in index_after.splitlines(), f"curated index line lost: {line!r}" + curated_bytes_after = { + p.name: p.read_bytes() + for p in sorted(bundle_dir.glob("*.md")) + if not p.name.startswith("ingest-") and p.name != "index.md" + } + assert curated_bytes_after == curated_bytes_before, "only stamped files may be replaced" + + +# --- Seam 5: MAF-free module guard ---------------------------------------------------------------- + + +def test_ingest_module_is_maf_free_and_context_layer_pure() -> None: + """The ingest module is a context-seam citizen (D7-portable): NO ``agent_framework``, + NO ``mcp``, and no MAF-coupled project modules — only ``okf``/``retrieval`` (both + stdlib-pure; ``contracts`` would smuggle MAF in via ``backends``). Mirrors + ``test_okf.py::test_okf_is_maf_free``, which covers only ``okf.py``.""" + tree = ast.parse(INGEST_SOURCE.read_text(encoding="utf-8")) + forbidden_roots = {"agent_framework", "mcp"} + allowed_internal = {"portfolio_optimiser.okf", "portfolio_optimiser.retrieval"} + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + root = alias.name.split(".")[0] + assert root not in forbidden_roots, f"MAF import in ingest.py: {alias.name}" + assert root != "portfolio_optimiser", ( + "import whole project modules explicitly via `from` so the guard can " + f"check them: {alias.name}" + ) + elif isinstance(node, ast.ImportFrom): + module = node.module or "" + root = module.split(".")[0] + assert root not in forbidden_roots, f"MAF import in ingest.py: {module}" + if root == "portfolio_optimiser": + if module == "portfolio_optimiser": + imported = {f"portfolio_optimiser.{alias.name}" for alias in node.names} + else: + imported = {module} + assert imported <= allowed_internal, ( + f"ingest.py imports MAF-coupled project module(s): " + f"{sorted(imported - allowed_internal)}" + )