feat(fase2a): herd load_verdicts_from_dir — vokabular-SKIP + caps (S2.5)
This commit is contained in:
parent
a706184bdd
commit
0a227a103b
2 changed files with 126 additions and 7 deletions
|
|
@ -7,6 +7,7 @@ genuine two-arg ``extend_instructions(source_id, instructions)`` GA signature
|
|||
Critical Fase 1 risk. Pattern: tests/spikes/test_d_verdictstore.py + real SessionContext.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
|
@ -19,7 +20,9 @@ from portfolio_optimiser.verdicts import (
|
|||
VerdictStore,
|
||||
bundle_candidate_features,
|
||||
capture_verdict,
|
||||
load_verdicts_from_dir,
|
||||
seed_store_from_bundle,
|
||||
write_verdict,
|
||||
)
|
||||
|
||||
_BUNDLE_DIR = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro"
|
||||
|
|
@ -145,3 +148,74 @@ def test_seed_store_retrieval_matches_the_candidate() -> None:
|
|||
hits = store.retrieve(query, k=3)
|
||||
assert len(hits) == 1
|
||||
assert "0.82" in hits[0].rationale
|
||||
|
||||
|
||||
# --- S2.5 (Step 7): inbox herding — vocabulary SKIP + rationale-cap (skip) + file-count (fail-fast) ---
|
||||
|
||||
|
||||
def _feats(code: str, magnitude: float = 1.0) -> ProposalFeatures:
|
||||
return ProposalFeatures(
|
||||
affected_codes=frozenset({code}), measure_type="m", claimed_saving_nok=magnitude
|
||||
)
|
||||
|
||||
|
||||
def test_load_skips_unknown_decision_vocabulary(tmp_path) -> None:
|
||||
"""T-2.5a: an inbox verdict whose decision is outside the binary run-path vocabulary
|
||||
{approved, rejected} is SKIPPED (never enters the store) — tolerant load, never raises. Detach
|
||||
the vocabulary check → the ``banana`` verdict enters the store → RED."""
|
||||
write_verdict(str(tmp_path), capture_verdict(_feats("X"), "banana", "weird decision"))
|
||||
write_verdict(str(tmp_path), capture_verdict(_feats("Y", 2.0), "approved", "fine"))
|
||||
|
||||
decisions = [v.decision for v in load_verdicts_from_dir(str(tmp_path))]
|
||||
assert "banana" not in decisions
|
||||
assert "approved" in decisions
|
||||
|
||||
|
||||
def test_load_skips_oversized_rationale_with_log(tmp_path, caplog) -> None:
|
||||
"""T-2.5b: a rationale over ``max_rationale_len`` is SKIPPED and logged (caplog-observable) — a
|
||||
per-file tolerant skip, NOT a raise (contrast the file-count cap)."""
|
||||
write_verdict(str(tmp_path), capture_verdict(_feats("X"), "approved", "x" * 500))
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
loaded = load_verdicts_from_dir(str(tmp_path), max_rationale_len=100)
|
||||
|
||||
assert loaded == []
|
||||
assert any("rationale" in r.message.lower() for r in caplog.records)
|
||||
|
||||
|
||||
def test_load_fails_fast_over_max_files(tmp_path) -> None:
|
||||
"""T-2.5c: a file count over ``max_files`` per merge is a fail-fast RAISE (aggregate guard,
|
||||
contrast the per-file tolerant skips)."""
|
||||
for i in range(3):
|
||||
write_verdict(
|
||||
str(tmp_path), capture_verdict(_feats(f"C{i}", float(i + 1)), "approved", "r")
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
load_verdicts_from_dir(str(tmp_path), max_files=2)
|
||||
|
||||
|
||||
def test_load_within_caps_is_unchanged(tmp_path) -> None:
|
||||
"""Control: with vocabulary-valid decisions and no cap breach, the loader behaves exactly as
|
||||
before — the tolerant contract (which the Step-7 loop relies on) is intact."""
|
||||
write_verdict(str(tmp_path), capture_verdict(_feats("X"), "approved", "ok"))
|
||||
write_verdict(str(tmp_path), capture_verdict(_feats("Y", 2.0), "rejected", "no"))
|
||||
|
||||
loaded = load_verdicts_from_dir(str(tmp_path), max_rationale_len=100, max_files=10)
|
||||
assert {v.decision for v in loaded} == {"approved", "rejected"}
|
||||
|
||||
|
||||
def test_no_inbox_json_uses_approved_with_adjustment() -> None:
|
||||
"""Assumption 3 (TDD guard): no shipped/test ``.json`` uses the promotion-only decision
|
||||
``approved_with_adjustment`` — the vocabulary SKIP would now silently drop it. It lives only in
|
||||
bundle-seed frontmatter + the promotion gate ``_APPROVED_DECISIONS``, never a binary run-path
|
||||
inbox file."""
|
||||
root = Path(__file__).resolve().parents[1]
|
||||
offenders = [
|
||||
p
|
||||
for p in root.rglob("*.json")
|
||||
if ".venv" not in p.parts
|
||||
and ".git" not in p.parts
|
||||
and "approved_with_adjustment" in p.read_text(encoding="utf-8", errors="ignore")
|
||||
]
|
||||
assert offenders == [], f"inbox JSON with promotion-only decision: {offenders}"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue