379 lines
15 KiB
Python
379 lines
15 KiB
Python
"""S5.1 HITL pending registry + dimension→expert routing (roadmap E, målbilde §3).
|
|
|
|
Behaviour + fail-fast tests for the operator inspection tool ``python -m
|
|
portfolio_optimiser.hitl pending|route``. The load-bearing detach seams (id-join,
|
|
route classification, transitive-MAF probe, inbox-predicate parity) live in
|
|
``test_hitl_loadbearing.py``.
|
|
|
|
Outbox artefacts are built via the REAL ``write_outbox`` and inbox verdicts via the REAL
|
|
``write_verdict`` (mirrors ``test_outbox_loadbearing.py`` / ``test_step7_async_loop_loadbearing.py``),
|
|
so a shape drift in either writer breaks these tests rather than passing green-but-dead.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from portfolio_optimiser.ir import AffectedItem, SavingsProposal
|
|
from portfolio_optimiser.outbox import write_outbox
|
|
from portfolio_optimiser.provenance import Citation, ProvenanceStamp
|
|
from portfolio_optimiser.retrieval import TextSpan
|
|
from portfolio_optimiser.validator import ValidatedProposal
|
|
from portfolio_optimiser.verdicts import ProposalFeatures, Verdict, write_verdict
|
|
|
|
from portfolio_optimiser import hitl
|
|
|
|
_PROVENANCE = ProvenanceStamp(
|
|
citations=[Citation(file="f.md", locator=TextSpan(start_index=0, end_index=5), snippet="hi")],
|
|
model="synthetic",
|
|
role="proposer",
|
|
validator_decision="validated",
|
|
token_usage=8,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_model_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Defensive hermetic env (copy of ``test_live_dry_run.py:20-29``): clear the S4.1 out-of-tree
|
|
overrides. hitl's CLI never reads these, but importing the package eagerly loads ``run`` — keep
|
|
the assertions insensitive to the operator's Foundry-configured environment."""
|
|
monkeypatch.delenv("PORTFOLIO_MODEL_MAP", raising=False)
|
|
monkeypatch.delenv("PORTFOLIO_FOUNDRY_PROJECT_ENDPOINT", raising=False)
|
|
|
|
|
|
def _make_validated(measure: str, codes: list[str], claimed: float = 200.0) -> ValidatedProposal:
|
|
items = [AffectedItem(code=c, quantity=1000.0, unit_cost=1.0) for c in codes]
|
|
proposal = SavingsProposal(
|
|
project_id="P1", measure=measure, affected_items=items, claimed_saving_nok=claimed
|
|
)
|
|
return ValidatedProposal(proposal=proposal, p10=100.0, p50=150.0, p90=200.0, nominal_feasible=180.0)
|
|
|
|
|
|
def _write_proposal(
|
|
outbox: Path, run_id: str, *, verdict_id: str, measure: str = "LED-retrofit", codes: list[str] | None = None
|
|
) -> None:
|
|
"""Persist one run's outbox pair via the real writer, keyed on ``verdict_id``."""
|
|
write_outbox(
|
|
str(outbox),
|
|
run_id,
|
|
outcome=_make_validated(measure, codes or ["05.2"]),
|
|
provenance=_PROVENANCE,
|
|
checker_verdict="approve",
|
|
verdict_id=verdict_id,
|
|
)
|
|
|
|
|
|
def _drop_verdict(inbox: Path, verdict_id: str, *, decision: str = "approved") -> None:
|
|
write_verdict(
|
|
str(inbox),
|
|
Verdict(
|
|
id=verdict_id,
|
|
proposal_features=ProposalFeatures(
|
|
affected_codes=frozenset({"05.2"}),
|
|
measure_type="scope_reduction",
|
|
claimed_saving_nok=200.0,
|
|
),
|
|
decision=decision,
|
|
rationale="expert reviewed (test)",
|
|
),
|
|
)
|
|
|
|
|
|
# --- pending() behaviour --------------------------------------------------------------------------
|
|
|
|
|
|
def test_pending_lists_unjudged_proposal(tmp_path: Path) -> None:
|
|
"""A proposal in the outbox with an EMPTY inbox is listed pending (run_id + verdict_id +
|
|
outcome_type surfaced, codes/measure captured)."""
|
|
outbox = tmp_path / "outbox"
|
|
inbox = tmp_path / "inbox"
|
|
_write_proposal(outbox, "run-1", verdict_id="vid-1", measure="LED-retrofit", codes=["05.2"])
|
|
|
|
result = hitl.pending(str(outbox), str(inbox))
|
|
|
|
assert len(result) == 1
|
|
p = result[0]
|
|
assert p.run_id == "run-1"
|
|
assert p.verdict_id == "vid-1"
|
|
assert p.outcome_type == "validated"
|
|
assert p.measure == "LED-retrofit"
|
|
assert p.codes == frozenset({"05.2"})
|
|
|
|
|
|
def test_pending_tolerant_to_missing_dir_and_foreign_files(tmp_path: Path) -> None:
|
|
"""Missing outbox → []; foreign / half-written files are skipped, never raised."""
|
|
inbox = tmp_path / "inbox"
|
|
assert hitl.pending(str(tmp_path / "does-not-exist"), str(inbox)) == []
|
|
|
|
outbox = tmp_path / "outbox"
|
|
_write_proposal(outbox, "run-1", verdict_id="vid-1")
|
|
(outbox / "notes.txt").write_text("not json", encoding="utf-8")
|
|
(outbox / "broken-proposal.json").write_text("{ half written", encoding="utf-8")
|
|
# a foreign file in the inbox must not blow up the id-set read either
|
|
inbox.mkdir()
|
|
(inbox / "golden.json").write_text("{ not a verdict", encoding="utf-8")
|
|
|
|
result = hitl.pending(str(outbox), str(inbox))
|
|
assert [p.run_id for p in result] == ["run-1"]
|
|
|
|
|
|
def test_pending_skips_orphan_proposal_without_outcome(tmp_path: Path) -> None:
|
|
"""A proposal file with no matching outcome (orphan from a half-written live run) is skipped."""
|
|
outbox = tmp_path / "outbox"
|
|
inbox = tmp_path / "inbox"
|
|
_write_proposal(outbox, "run-1", verdict_id="vid-1")
|
|
(outbox / "run-1-outcome.json").unlink() # orphan the proposal
|
|
|
|
assert hitl.pending(str(outbox), str(inbox)) == []
|
|
|
|
|
|
def test_pending_is_deterministic(tmp_path: Path) -> None:
|
|
"""Output order is stable across calls, sorted by (run_id, verdict_id)."""
|
|
outbox = tmp_path / "outbox"
|
|
inbox = tmp_path / "inbox"
|
|
_write_proposal(outbox, "run-c", verdict_id="vid-3")
|
|
_write_proposal(outbox, "run-a", verdict_id="vid-1")
|
|
_write_proposal(outbox, "run-b", verdict_id="vid-2")
|
|
|
|
first = [(p.run_id, p.verdict_id) for p in hitl.pending(str(outbox), str(inbox))]
|
|
second = [(p.run_id, p.verdict_id) for p in hitl.pending(str(outbox), str(inbox))]
|
|
assert first == second == [("run-a", "vid-1"), ("run-b", "vid-2"), ("run-c", "vid-3")]
|
|
|
|
|
|
def test_hitl_registered_maf_free() -> None:
|
|
"""Meta: hitl.py is registered in the MAF-free guard list, so ``test_okf_is_maf_free`` actually
|
|
scans it — otherwise the MAF-free source claim would be green-but-dead."""
|
|
from tests.test_okf import _MAF_FREE_MODULES
|
|
|
|
assert "hitl.py" in _MAF_FREE_MODULES
|
|
|
|
|
|
# --- load_routing_config() fail-fast ---------------------------------------------------------------
|
|
|
|
|
|
def _write_config(tmp_path: Path, data: dict) -> Path:
|
|
p = tmp_path / "routing-config.json"
|
|
p.write_text(json.dumps(data), encoding="utf-8")
|
|
return p
|
|
|
|
|
|
_WELLFORMED_CONFIG = {
|
|
"_note": "ignored underscore key",
|
|
"entries": [
|
|
{
|
|
"id": "energi",
|
|
"allowed_measure_types": ["scope_reduction"],
|
|
"allowed_code_prefixes": ["05"],
|
|
"expert": "Ola Energi",
|
|
},
|
|
{"id": "vei", "allowed_code_prefixes": ["03"], "expert": "Kari Vei"},
|
|
],
|
|
}
|
|
|
|
|
|
def test_routing_config_loads_wellformed(tmp_path: Path) -> None:
|
|
"""A well-formed config loads: entries + experts present, ``_``-prefixed top-level keys stripped."""
|
|
config = hitl.load_routing_config(_write_config(tmp_path, _WELLFORMED_CONFIG))
|
|
assert [e.id for e in config.entries] == ["energi", "vei"]
|
|
assert {e.expert for e in config.entries} == {"Ola Energi", "Kari Vei"}
|
|
assert config.entries[1].allowed_measure_types == frozenset() # empty default = "any measure"
|
|
|
|
|
|
def test_routing_config_missing_expert_raises(tmp_path: Path) -> None:
|
|
"""An entry missing the required ``expert`` field fails validation."""
|
|
bad = {"entries": [{"id": "energi", "allowed_code_prefixes": ["05"]}]}
|
|
with pytest.raises(ValidationError):
|
|
hitl.load_routing_config(_write_config(tmp_path, bad))
|
|
|
|
|
|
def test_routing_config_duplicate_id_raises(tmp_path: Path) -> None:
|
|
"""Two entries with the same ``id`` raise (ValueError via the after-validator)."""
|
|
dup = {
|
|
"entries": [
|
|
{"id": "energi", "expert": "A"},
|
|
{"id": "energi", "expert": "B"},
|
|
]
|
|
}
|
|
with pytest.raises(ValueError):
|
|
hitl.load_routing_config(_write_config(tmp_path, dup))
|
|
|
|
|
|
def test_routing_config_wrong_types_raise(tmp_path: Path) -> None:
|
|
"""A non-list ``entries`` (wrong shape) raises."""
|
|
with pytest.raises(ValidationError):
|
|
hitl.load_routing_config(_write_config(tmp_path, {"entries": {"id": "x", "expert": "y"}}))
|
|
|
|
|
|
def test_routing_config_missing_file_raises(tmp_path: Path) -> None:
|
|
"""A missing config file fails fast with ``FileNotFoundError`` (mirrors ``load_pricing``)."""
|
|
with pytest.raises(FileNotFoundError):
|
|
hitl.load_routing_config(tmp_path / "no-such-routing-config.json")
|
|
|
|
|
|
def test_routing_config_valid_variant_loads(tmp_path: Path) -> None:
|
|
"""Control: a minimal valid config (single entry, no code/measure constraints) loads — proving
|
|
the raises above fire only on the bad case, not always."""
|
|
config = hitl.load_routing_config(
|
|
_write_config(tmp_path, {"entries": [{"id": "generalist", "expert": "Per"}]})
|
|
)
|
|
assert config.entries[0].expert == "Per"
|
|
assert config.entries[0].allowed_code_prefixes == frozenset()
|
|
|
|
|
|
# --- route() classification -----------------------------------------------------------------------
|
|
|
|
|
|
def test_route_by_code_prefix_with_measure_optional(tmp_path: Path) -> None:
|
|
"""A proposal with a REAL prose measure routes by code prefix when the entry sets no measure gate
|
|
(``allowed_measure_types`` empty = "any") — proving measure-optional works on real prose, not a
|
|
fixture-tuned token."""
|
|
outbox = tmp_path / "outbox"
|
|
inbox = tmp_path / "inbox"
|
|
_write_proposal(
|
|
outbox, "run-1", verdict_id="v1", measure="LED-retrofit av kontorbelysning", codes=["05.2"]
|
|
)
|
|
config = hitl.RoutingConfig(
|
|
entries=[
|
|
hitl.RoutingEntry(id="energi", allowed_code_prefixes=frozenset({"05"}), expert="Ola")
|
|
]
|
|
)
|
|
routed = hitl.route(str(outbox), str(inbox), config)
|
|
assert len(routed) == 1
|
|
r = routed[0]
|
|
assert r.expert == "Ola"
|
|
assert r.dimension_id == "energi"
|
|
assert r.ambiguous is False
|
|
assert r.pending.measure == "LED-retrofit av kontorbelysning"
|
|
|
|
|
|
def test_route_unroutable_when_no_entry_matches(tmp_path: Path) -> None:
|
|
"""A proposal whose codes match no entry is emitted UNROUTABLE (expert/dimension None), never
|
|
silently dropped."""
|
|
outbox = tmp_path / "outbox"
|
|
inbox = tmp_path / "inbox"
|
|
_write_proposal(outbox, "run-1", verdict_id="v1", codes=["99.9"])
|
|
config = hitl.RoutingConfig(
|
|
entries=[
|
|
hitl.RoutingEntry(id="energi", allowed_code_prefixes=frozenset({"05"}), expert="Ola")
|
|
]
|
|
)
|
|
routed = hitl.route(str(outbox), str(inbox), config)
|
|
assert len(routed) == 1
|
|
assert routed[0].expert is None
|
|
assert routed[0].dimension_id is None
|
|
assert routed[0].ambiguous is False
|
|
|
|
|
|
def test_route_ambiguous_uses_sorted_first_tie_break(tmp_path: Path) -> None:
|
|
"""A proposal admitted by two entries routes to the sorted-first ``entry.id`` (deterministic) and
|
|
is flagged ``ambiguous``."""
|
|
outbox = tmp_path / "outbox"
|
|
inbox = tmp_path / "inbox"
|
|
_write_proposal(outbox, "run-1", verdict_id="v1", codes=["05.2"])
|
|
config = hitl.RoutingConfig(
|
|
entries=[
|
|
hitl.RoutingEntry(id="b-energi", allowed_code_prefixes=frozenset({"05"}), expert="Beta"),
|
|
hitl.RoutingEntry(id="a-energi", allowed_code_prefixes=frozenset({"05"}), expert="Alpha"),
|
|
]
|
|
)
|
|
r = hitl.route(str(outbox), str(inbox), config)[0]
|
|
assert r.ambiguous is True
|
|
assert r.dimension_id == "a-energi"
|
|
assert r.expert == "Alpha"
|
|
|
|
|
|
def test_route_measure_constrained_entry_filters(tmp_path: Path) -> None:
|
|
"""A measure-constrained entry filters out a proposal whose prose measure is not in
|
|
``allowed_measure_types`` (unroutable); the same entry routes a matching measure — proving the
|
|
gate fires only on the mismatch."""
|
|
inbox = tmp_path / "inbox"
|
|
config = hitl.RoutingConfig(
|
|
entries=[
|
|
hitl.RoutingEntry(
|
|
id="energi",
|
|
allowed_measure_types=frozenset({"scope_reduction"}),
|
|
allowed_code_prefixes=frozenset({"05"}),
|
|
expert="Ola",
|
|
)
|
|
]
|
|
)
|
|
miss = tmp_path / "miss"
|
|
_write_proposal(miss, "run-1", verdict_id="v1", measure="rate_renegotiation", codes=["05.2"])
|
|
assert hitl.route(str(miss), str(inbox), config)[0].expert is None
|
|
|
|
hit = tmp_path / "hit"
|
|
_write_proposal(hit, "run-1", verdict_id="v1", measure="scope_reduction", codes=["05.2"])
|
|
assert hitl.route(str(hit), str(inbox), config)[0].expert == "Ola"
|
|
|
|
|
|
# --- CLI: python -m portfolio_optimiser.hitl pending|route ----------------------------------------
|
|
|
|
|
|
def test_cli_pending_lists_unjudged(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""``pending`` subcommand → rc 0 and one greppable ``run_id verdict_id outcome_type`` line per
|
|
un-judged proposal."""
|
|
outbox = tmp_path / "outbox"
|
|
inbox = tmp_path / "inbox"
|
|
_write_proposal(outbox, "run-1", verdict_id="v1")
|
|
rc = hitl.main(["pending", "--outbox-dir", str(outbox), "--verdict-dir", str(inbox)])
|
|
assert rc == 0
|
|
out = capsys.readouterr().out
|
|
assert "run-1 v1 validated" in out
|
|
|
|
|
|
def test_cli_route_lists_expert_and_unroutable(
|
|
tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
"""``route`` subcommand → rc 0; a routed proposal carries its expert + ``dim:``, an unmatched one
|
|
carries ``UNROUTABLE`` (and no ``dim:``)."""
|
|
outbox = tmp_path / "outbox"
|
|
inbox = tmp_path / "inbox"
|
|
_write_proposal(outbox, "run-1", verdict_id="v1", codes=["05.2"])
|
|
_write_proposal(outbox, "run-2", verdict_id="v2", codes=["99.9"])
|
|
cfg = _write_config(
|
|
tmp_path, {"entries": [{"id": "energi", "allowed_code_prefixes": ["05"], "expert": "Ola"}]}
|
|
)
|
|
rc = hitl.main(
|
|
["route", "--outbox-dir", str(outbox), "--verdict-dir", str(inbox), "--routing-config", str(cfg)]
|
|
)
|
|
assert rc == 0
|
|
out = capsys.readouterr().out
|
|
assert "run-1 v1" in out and "Ola" in out and "dim:energi" in out
|
|
assert "run-2 v2" in out and "UNROUTABLE" in out
|
|
|
|
|
|
def test_cli_route_malformed_config_returns_rc1(
|
|
tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
"""A malformed routing config → rc 1, a structured ``hitl:`` message to stderr, NO traceback."""
|
|
outbox = tmp_path / "outbox"
|
|
inbox = tmp_path / "inbox"
|
|
_write_proposal(outbox, "run-1", verdict_id="v1")
|
|
bad_cfg = _write_config(tmp_path, {"entries": [{"id": "energi"}]}) # missing required expert
|
|
rc = hitl.main(
|
|
["route", "--outbox-dir", str(outbox), "--verdict-dir", str(inbox), "--routing-config", str(bad_cfg)]
|
|
)
|
|
assert rc == 1
|
|
captured = capsys.readouterr()
|
|
assert "hitl:" in captured.err
|
|
assert "Traceback" not in captured.err
|
|
|
|
|
|
def test_cli_pending_output_is_deterministic(
|
|
tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
"""Two identical ``pending`` invocations produce byte-identical stdout."""
|
|
outbox = tmp_path / "outbox"
|
|
inbox = tmp_path / "inbox"
|
|
_write_proposal(outbox, "run-b", verdict_id="v2")
|
|
_write_proposal(outbox, "run-a", verdict_id="v1")
|
|
hitl.main(["pending", "--outbox-dir", str(outbox), "--verdict-dir", str(inbox)])
|
|
first = capsys.readouterr().out
|
|
hitl.main(["pending", "--outbox-dir", str(outbox), "--verdict-dir", str(inbox)])
|
|
second = capsys.readouterr().out
|
|
assert first == second
|
|
assert first.index("run-a") < first.index("run-b") # sorted
|