272 lines
13 KiB
Python
272 lines
13 KiB
Python
"""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)}"
|
|
)
|