Kjell Tore Guttormsen
37547fe292
The context sets, the packaged knowledge bases and the example bundles are replaced by one fictitious example set about IT operations in an invented organisation: three context sets (serverrom-2027, driftsavtale-2027 and the two-base drift-og-avtale-2027), two synthetic knowledge bases under src/portfolio_optimiser/data/kunnskapsbaser and two example bundles under src/portfolio_optimiser/data/bundles. Numbers, codes and structural values in tests and fixtures are kept; names, ids and wording change. Dated measurement documents that only recorded runs on the replaced material are deleted. Gate figures measured on the new set are not comparable with earlier ones. The exclusion gate from the previous commit is green: 0 tracked files hit outside the shared/ subtree. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1022 lines
38 KiB
Python
1022 lines
38 KiB
Python
"""S5.3 CLI-parity tests for ``run.main()`` single-project flags (Steps 2/4/5).
|
|
|
|
In-process ``run.main([argv])`` + rc + ``capsys`` substring asserts (never subprocess),
|
|
mirroring ``tests/test_live_dry_run.py``. Every arm is offline — it stops before the first
|
|
model call (``debate.run``), so no socket/network is exercised (brief NFR). The bundle
|
|
fixture ``shared/examples/bygg-energi-mikro`` (project ``BYGG-KONTOR-NORD``) supplies citable
|
|
content so the dry-run reaches its offline return.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from portfolio_optimiser import run
|
|
from portfolio_optimiser.dimension import Dimension
|
|
from portfolio_optimiser.ledger import LedgerEntry, SavingsLedger
|
|
from portfolio_optimiser.run import run_project
|
|
from portfolio_optimiser.verdicts import ProposalFeatures, capture_verdict, write_verdict
|
|
|
|
BUNDLE_DIR = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro"
|
|
_PID = "BYGG-KONTOR-NORD"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_model_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Hermetic env (verbatim from ``test_live_dry_run.py``): clear the S4.1 out-of-tree overrides
|
|
so these CLI assertions read the BUNDLED map/config, not the operator's Foundry environment."""
|
|
monkeypatch.delenv("PORTFOLIO_MODEL_MAP", raising=False)
|
|
monkeypatch.delenv("PORTFOLIO_FOUNDRY_PROJECT_ENDPOINT", raising=False)
|
|
|
|
|
|
def _write_dimension(tmp_path: Path) -> Path:
|
|
"""A valid dimension scope config on disk (loaded fail-fast by ``load_dimension``)."""
|
|
dim = Dimension(
|
|
id="energi",
|
|
label="Energi",
|
|
allowed_measure_types=frozenset({"energy_efficiency"}),
|
|
)
|
|
p = tmp_path / "dim.json"
|
|
p.write_text(dim.model_dump_json(), encoding="utf-8")
|
|
return p
|
|
|
|
|
|
# --- Step 2: --dimension-config / --outbox-dir / --run-id wiring + structured refusal -------------
|
|
|
|
|
|
def test_dimension_config_flag_parses_offline(tmp_path, capsys) -> None:
|
|
"""(a) ``--dimension-config <valid>`` + ``--live-dry-run`` → rc 0 (flag parsed, loader invoked,
|
|
offline — stops before any model call)."""
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--bundle-dir",
|
|
str(BUNDLE_DIR),
|
|
"--dimension-config",
|
|
str(_write_dimension(tmp_path)),
|
|
"--live-dry-run",
|
|
]
|
|
)
|
|
assert rc == 0
|
|
assert "LIVE-DRY-RUN OK" in capsys.readouterr().out
|
|
|
|
|
|
def test_outbox_dir_with_run_id_writes_runconfig_offline(tmp_path) -> None:
|
|
"""(b) ``--outbox-dir`` + ``--run-id`` + ``--live-dry-run`` → rc 0 AND ``<tmp>/r1-runconfig.json``
|
|
written offline (via ``write_run_config``, before the dry-run return)."""
|
|
outbox = tmp_path / "out"
|
|
outbox.mkdir()
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--bundle-dir",
|
|
str(BUNDLE_DIR),
|
|
"--outbox-dir",
|
|
str(outbox),
|
|
"--run-id",
|
|
"r1",
|
|
"--live-dry-run",
|
|
]
|
|
)
|
|
assert rc == 0
|
|
assert (outbox / "r1-runconfig.json").is_file()
|
|
|
|
|
|
def test_outbox_dir_without_run_id_refuses(tmp_path, capsys) -> None:
|
|
"""(c) RED guard: ``--outbox-dir`` WITHOUT ``--run-id`` → rc 1 structured refusal. ``run_project``'s
|
|
step-0 fail-fast (no wall-clock default) surfaces through the CLI refusal wrapper, no traceback."""
|
|
outbox = tmp_path / "out"
|
|
outbox.mkdir()
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--outbox-dir",
|
|
str(outbox),
|
|
"--live-dry-run",
|
|
]
|
|
)
|
|
assert rc == 1
|
|
assert "refused" in capsys.readouterr().err.lower()
|
|
|
|
|
|
def test_dimension_config_missing_file_refuses(capsys) -> None:
|
|
"""(d) ``--dimension-config <nonexistent>`` → rc 1 structured refusal. ``load_dimension`` raises
|
|
``FileNotFoundError``, caught by the WIDENED dry-run handler (not just ``ValueError`` — Pass-2 #1)."""
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--dimension-config",
|
|
"/nonexistent-dim-config.json",
|
|
"--live-dry-run",
|
|
]
|
|
)
|
|
assert rc == 1
|
|
assert "refused" in capsys.readouterr().err.lower()
|
|
|
|
|
|
# --- Step 4: mode-exclusivity refusals + backward-compat pin -------------------------------------
|
|
|
|
|
|
def _met_portfolio_goal(tmp_path: Path) -> tuple[Path, Path]:
|
|
"""A portfolio-hard goal (1 øre) already met by a 1-øre ledger — so any RED (pre-refusal)
|
|
fall-through into the portfolio dispatch stops OFFLINE at the goal check (no client, no socket)."""
|
|
goals = tmp_path / "goals.json"
|
|
goals.write_text('{"portfolio": {"absolute_ore": 1, "mode": "hard"}}', encoding="utf-8")
|
|
led = SavingsLedger()
|
|
led.add_realized(
|
|
LedgerEntry(
|
|
project_id="KONTOR-IT-E1",
|
|
dimension="energi",
|
|
candidate_identity="c1",
|
|
amount_ore=1,
|
|
verdict_id="v1",
|
|
provenance="x",
|
|
)
|
|
)
|
|
ledger = tmp_path / "ledger.json"
|
|
led.save(str(ledger))
|
|
return goals, ledger
|
|
|
|
|
|
def test_goals_without_portfolio_refuses(tmp_path, capsys) -> None:
|
|
"""(a) ``--goals`` without ``--portfolio`` → rc 1 refusal (the flag belongs to portfolio mode).
|
|
``--live-dry-run`` keeps the RED (pre-refusal) fall-through offline."""
|
|
goals = tmp_path / "goals.json"
|
|
goals.write_text('{"portfolio": {"absolute_ore": 1, "mode": "hard"}}', encoding="utf-8")
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--bundle-dir",
|
|
str(BUNDLE_DIR),
|
|
"--goals",
|
|
str(goals),
|
|
"--live-dry-run",
|
|
]
|
|
)
|
|
assert rc == 1
|
|
assert "refused" in capsys.readouterr().err.lower()
|
|
|
|
|
|
def test_ledger_without_portfolio_refuses(tmp_path, capsys) -> None:
|
|
"""(a') ``--ledger`` without ``--portfolio`` → rc 1 refusal."""
|
|
ledger = tmp_path / "ledger.json"
|
|
ledger.write_text("[]", encoding="utf-8")
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--bundle-dir",
|
|
str(BUNDLE_DIR),
|
|
"--ledger",
|
|
str(ledger),
|
|
"--live-dry-run",
|
|
]
|
|
)
|
|
assert rc == 1
|
|
assert "refused" in capsys.readouterr().err.lower()
|
|
|
|
|
|
def test_portfolio_with_single_project_flag_refuses(tmp_path, capsys) -> None:
|
|
"""(c) ``--portfolio`` combined with a single-project-only flag (``--outbox-dir``) → rc 1 refusal
|
|
naming the offending flag. The met portfolio goal keeps the RED fall-through offline."""
|
|
goals, ledger = _met_portfolio_goal(tmp_path)
|
|
rc = run.main(
|
|
[
|
|
"--portfolio",
|
|
"--goals",
|
|
str(goals),
|
|
"--ledger",
|
|
str(ledger),
|
|
"--outbox-dir",
|
|
str(tmp_path / "ob"),
|
|
]
|
|
)
|
|
assert rc == 1
|
|
err = capsys.readouterr().err.lower()
|
|
assert "refused" in err
|
|
assert "--outbox-dir" in err
|
|
|
|
|
|
def test_single_project_mode_without_docs_dir_refuses(capsys) -> None:
|
|
"""(b) single-project mode with a pid but no ``--docs-dir`` → rc 1 refusal (the Step-3
|
|
compensating guard for the relaxed argparse ``required=``)."""
|
|
rc = run.main([_PID])
|
|
assert rc == 1
|
|
assert "refused" in capsys.readouterr().err.lower()
|
|
|
|
|
|
def test_single_project_mode_without_pid_refuses(capsys) -> None:
|
|
"""(b') single-project mode with ``--docs-dir`` but no PROJECT_ID → rc 1 refusal."""
|
|
rc = run.main(["--docs-dir", str(BUNDLE_DIR)])
|
|
assert rc == 1
|
|
assert "refused" in capsys.readouterr().err.lower()
|
|
|
|
|
|
def test_legacy_single_project_invocation_still_succeeds(capsys) -> None:
|
|
"""Backward-compat pin: the legacy invocation (positional pid + ``--docs-dir`` + ``--bundle-dir``
|
|
+ ``--live-dry-run``) still returns rc 0 — the existing CLI contract survives Step 3's
|
|
``nargs='?'``/``required`` relaxation."""
|
|
rc = run.main(
|
|
[_PID, "--docs-dir", str(BUNDLE_DIR), "--bundle-dir", str(BUNDLE_DIR), "--live-dry-run"]
|
|
)
|
|
assert rc == 0
|
|
assert "LIVE-DRY-RUN OK" in capsys.readouterr().out
|
|
|
|
|
|
# --- Step 5: confirm coverage for the already-wired flags (never re-wired; run.py untouched) ------
|
|
|
|
|
|
def test_verdict_dir_ingested_at_main_level_offline(tmp_path, capsys) -> None:
|
|
"""Step 5 (SC2 second half): ``--verdict-dir`` is exercised at ``main()`` level — the previously
|
|
untested already-wired flag. The async inbox is ingested (``load_verdicts_from_dir``,
|
|
``run.py:287``) BEFORE the dry-run cut (``run.py:335``), so a dropped verdict is threaded through
|
|
``main()`` offline without raising. ``--bundle-dir``'s ``main()``-level coverage already exists
|
|
in ``tests/test_live_dry_run.py:32-49`` and is NOT re-tested here (never re-wired)."""
|
|
inbox = tmp_path / "inbox"
|
|
feats = ProposalFeatures(
|
|
affected_codes=frozenset({"ENERGI-TOTAL-EL"}),
|
|
measure_type="energy_efficiency",
|
|
claimed_saving_nok=30000.0,
|
|
description="LED-retrofit",
|
|
)
|
|
write_verdict(str(inbox), capture_verdict(feats, "approved", "expert reviewed (sim)"))
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--bundle-dir",
|
|
str(BUNDLE_DIR),
|
|
"--verdict-dir",
|
|
str(inbox),
|
|
"--live-dry-run",
|
|
]
|
|
)
|
|
assert rc == 0
|
|
assert "LIVE-DRY-RUN OK" in capsys.readouterr().out
|
|
|
|
|
|
# --- S5.4: --report / --json read-only value-report mode ------------------------------------------
|
|
|
|
|
|
def _report_ledger(tmp_path: Path) -> Path:
|
|
"""A saved ledger with >=2 projects and one cross-dimension overlap (``c-a`` under both
|
|
``energi`` and ``lisens`` in KONTOR -> counted once, flagged), for the value-report arms.
|
|
portfolio_total = 1234567 (overlap once) + 500000 = 1734567 øre."""
|
|
led = SavingsLedger()
|
|
led.add_realized(
|
|
LedgerEntry(
|
|
project_id="KONTOR-IT-E1",
|
|
dimension="energi",
|
|
candidate_identity="c-a",
|
|
amount_ore=1234567,
|
|
verdict_id="v1",
|
|
provenance="p1",
|
|
)
|
|
)
|
|
led.add_realized(
|
|
LedgerEntry(
|
|
project_id="KONTOR-IT-E1",
|
|
dimension="lisens",
|
|
candidate_identity="c-a",
|
|
amount_ore=1234567,
|
|
verdict_id="v2",
|
|
provenance="p2", # cross-dimension overlap on (KONTOR-IT-E1, c-a)
|
|
)
|
|
)
|
|
led.add_realized(
|
|
LedgerEntry(
|
|
project_id="NETT-SIKR-TP",
|
|
dimension="energi",
|
|
candidate_identity="c-b",
|
|
amount_ore=500000,
|
|
verdict_id="v3",
|
|
provenance="p3",
|
|
)
|
|
)
|
|
p = tmp_path / "ledger.json"
|
|
led.save(str(p))
|
|
return p
|
|
|
|
|
|
def test_report_prints_table_rc0(tmp_path, capsys) -> None:
|
|
"""SC3: ``--report --ledger <f>`` -> rc 0; stdout carries a per-project row + the portfolio-total
|
|
NOK string. Dispatched FIRST, so no PROJECT_ID/--docs-dir is needed (no single-project refusal)."""
|
|
rc = run.main(["--report", "--ledger", str(_report_ledger(tmp_path))])
|
|
out = capsys.readouterr().out
|
|
assert rc == 0
|
|
assert "KONTOR-IT-E1" in out # a per-project row
|
|
assert "17\xa0345,67\xa0kr" in out # portfolio total 1734567 øre
|
|
|
|
|
|
def test_report_json_rc0_parses_rollup(tmp_path, capsys) -> None:
|
|
"""SC4: ``--report ... --json`` -> rc 0 and ``json.loads(stdout)`` yields the roll-up
|
|
(int portfolio total, per_project dict, overlaps as JSON lists, provenance list)."""
|
|
rc = run.main(["--report", "--ledger", str(_report_ledger(tmp_path)), "--json"])
|
|
out = capsys.readouterr().out
|
|
assert rc == 0
|
|
payload = json.loads(out)
|
|
assert payload["portfolio_total_ore"] == 1734567
|
|
assert payload["per_project"]["KONTOR-IT-E1"] == 1234567
|
|
assert isinstance(payload["overlaps"], list)
|
|
assert ["KONTOR-IT-E1", "c-a"] in payload["overlaps"] # tuple serialized as a JSON array
|
|
assert isinstance(payload["provenance"], list)
|
|
assert len(payload["provenance"]) == 3 # one ProvenanceLine per ledger entry
|
|
|
|
|
|
def test_report_missing_ledger_file_rc1(capsys) -> None:
|
|
"""SC5: ``--report --ledger /nonexistent`` -> rc 1, stderr non-empty, NO table on stdout (a load
|
|
failure must never masquerade as a real zero-savings result)."""
|
|
rc = run.main(["--report", "--ledger", "/nonexistent-ledger.json"])
|
|
cap = capsys.readouterr()
|
|
assert rc == 1
|
|
assert cap.err.strip()
|
|
assert cap.out == ""
|
|
|
|
|
|
def test_report_malformed_ledger_rc1(tmp_path, capsys) -> None:
|
|
"""SC5: a malformed-row ledger file -> rc 1 (``ValidationError`` surfaced as a refusal)."""
|
|
bad = tmp_path / "bad.json"
|
|
bad.write_text('[{"project_id": "P1"}]', encoding="utf-8") # missing required fields
|
|
rc = run.main(["--report", "--ledger", str(bad)])
|
|
assert rc == 1
|
|
assert "refused" in capsys.readouterr().err.lower()
|
|
|
|
|
|
def test_report_empty_dict_ledger_rc1(tmp_path, capsys) -> None:
|
|
"""SC5 masquerade guard: a valid-JSON ``{}`` ledger must NOT load as an empty ledger and print a
|
|
misleading ``0,00 kr`` at rc 0 — a malformed file masquerading as a real zero-savings result is
|
|
the exact failure SC5's fail-fast refusal exists to prevent."""
|
|
bad = tmp_path / "empty-obj.json"
|
|
bad.write_text("{}", encoding="utf-8")
|
|
rc = run.main(["--report", "--ledger", str(bad)])
|
|
cap = capsys.readouterr()
|
|
assert rc == 1
|
|
assert "refused" in cap.err.lower()
|
|
assert cap.out == "" # no table, no "0,00 kr"
|
|
|
|
|
|
def test_report_nonlist_ledger_rc1_no_traceback(tmp_path, capsys) -> None:
|
|
"""SC5: a valid-JSON but wrong-shape ledger (bare scalar / object-with-keys) -> rc 1 refusal, NOT
|
|
an uncaught ``TypeError`` traceback ('rc 1, no traceback')."""
|
|
bad = tmp_path / "scalar.json"
|
|
bad.write_text("42", encoding="utf-8")
|
|
rc = run.main(["--report", "--ledger", str(bad)])
|
|
cap = capsys.readouterr()
|
|
assert rc == 1
|
|
assert "refused" in cap.err.lower()
|
|
assert "Traceback" not in cap.err
|
|
assert cap.out == ""
|
|
|
|
|
|
def test_report_without_ledger_rc1_no_traceback(capsys) -> None:
|
|
"""Major-#1 guard: ``--report`` with no ``--ledger`` -> rc 1 with a 'requires --ledger' message
|
|
and no traceback (guards ``SavingsLedger.load(None)`` -> ``Path(None)`` TypeError)."""
|
|
rc = run.main(["--report"])
|
|
err = capsys.readouterr().err
|
|
assert rc == 1
|
|
assert "requires --ledger" in err
|
|
assert "Traceback" not in err
|
|
|
|
|
|
def test_report_with_portfolio_refuses(tmp_path, capsys) -> None:
|
|
"""Major-#3 partition: ``--report`` + ``--portfolio`` -> rc 1 (mode-exclusive)."""
|
|
rc = run.main(["--report", "--portfolio", "--ledger", str(_report_ledger(tmp_path))])
|
|
assert rc == 1
|
|
assert "refused" in capsys.readouterr().err.lower()
|
|
|
|
|
|
def test_report_with_goals_refuses(tmp_path, capsys) -> None:
|
|
"""Major-#3 / P2-2 allowlist: ``--report`` + ``--goals`` (a config flag) -> rc 1. The allowlist
|
|
rejects config flags too, not just the two mode flags — else ``--goals`` would be silently
|
|
dropped, whereas bare ``--goals`` is refused (adding ``--report`` must not suppress a refusal)."""
|
|
goals = tmp_path / "goals.json"
|
|
goals.write_text('{"portfolio": {"absolute_ore": 1, "mode": "hard"}}', encoding="utf-8")
|
|
rc = run.main(["--report", "--goals", str(goals), "--ledger", str(_report_ledger(tmp_path))])
|
|
assert rc == 1
|
|
assert "refused" in capsys.readouterr().err.lower()
|
|
|
|
|
|
def test_json_without_report_refuses(tmp_path, capsys) -> None:
|
|
"""Major-#3 partition: ``--json`` without ``--report`` -> rc 1 (a stray --json is never silently
|
|
ignored)."""
|
|
rc = run.main(["--json", "--ledger", str(_report_ledger(tmp_path))])
|
|
assert rc == 1
|
|
assert "refused" in capsys.readouterr().err.lower()
|
|
|
|
|
|
# --- S3.1 SC6: --semantic-retrieval opt-in (default OFF) --------------------------------------
|
|
|
|
_ENERGY_REPLY = (
|
|
'{"measure":"LED-retrofit av kontorbelysning","affected_items":'
|
|
'[{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}],"claimed_saving_nok":30000}'
|
|
)
|
|
_VERDICT_INPUT = {"decision": "approved", "rationale": "expert reviewed (sim)"}
|
|
|
|
# The marker lives in the RATIONALE, which is what ``format_fewshot`` emits into the prompt.
|
|
# "0.37" appears nowhere in the bundle (grep-verified), so its presence in a prompt can only have
|
|
# come through the ExpeL fold — it cannot be leaked by bundle context.
|
|
_TIE_MARKER = "realiseringsgrad=0.37"
|
|
|
|
# The pair ties on the STRUCTURAL score while carrying different code sets: each keeps the query's
|
|
# own code and adds one foreign code, so Jaccard is equal (0.700 each) but the canonical embedding
|
|
# string differs and cosine has something to separate. The previous fixture tied by being
|
|
# structurally IDENTICAL and differing only in prose — a shape the framework cannot mint, since
|
|
# ``_mint_id`` ignores description and both candidates would collapse onto one id.
|
|
_MARKER_EXTRA_CODE = "01.1"
|
|
_DISTRACTOR_EXTRA_CODE = "01.4"
|
|
|
|
|
|
def _tied_pair_verdicts():
|
|
"""The tied pair, minted through ``capture_verdict`` — the real minting path.
|
|
|
|
The bundle's own seed verdict is deliberately NOT included: its features are identical to the
|
|
query, making it a perfect cosine match that would win every hybrid ranking and so could never
|
|
demonstrate a tie-break."""
|
|
from portfolio_optimiser.verdicts import (
|
|
VerdictStore,
|
|
bundle_candidate_features,
|
|
capture_verdict,
|
|
)
|
|
|
|
query = bundle_candidate_features(str(BUNDLE_DIR))
|
|
|
|
def tied(extra_code: str, decision: str, rationale: str):
|
|
return capture_verdict(
|
|
ProposalFeatures(
|
|
affected_codes=query.affected_codes | {extra_code},
|
|
measure_type=query.measure_type,
|
|
# ``description == measure`` is what both live minting paths emit
|
|
# (``run._features_of`` / ``verdicts.features_from_ir``).
|
|
claimed_saving_nok=query.claimed_saving_nok,
|
|
description=query.measure_type,
|
|
),
|
|
decision,
|
|
rationale,
|
|
)
|
|
|
|
distractor = tied(
|
|
_DISTRACTOR_EXTRA_CODE,
|
|
"rejected",
|
|
"ingen realiseringsdata for dette tiltaket",
|
|
)
|
|
marker = tied(_MARKER_EXTRA_CODE, "approved", f"tidligere LED-dom [{_TIE_MARKER}]")
|
|
return VerdictStore(verdicts=[distractor, marker]), marker.id, distractor.id
|
|
|
|
|
|
# Minted once at module scope so the tests can name the ids. ``_MARKER_ID`` sorts AFTER
|
|
# ``_DISTRACTOR_ID``, so the structural ``(-similarity, id)`` key puts the DISTRACTOR first and
|
|
# only the cosine term can overturn it.
|
|
_MARKER_ID = _tied_pair_verdicts()[1]
|
|
_DISTRACTOR_ID = _tied_pair_verdicts()[2]
|
|
|
|
|
|
def _tied_pair_store():
|
|
return _tied_pair_verdicts()[0]
|
|
|
|
|
|
def _bundle_reference_project(tmp_path: Path):
|
|
"""A single bundle-backed project for the portfolio arm, pointed at the SAME bundle the tied
|
|
pair was derived from — a different bundle would yield a different query and untie the pair.
|
|
Topology mirrors ``tests/test_portfolio_learning_loadbearing.py::_bundle_kplus1``."""
|
|
from portfolio_optimiser.reference_domain import Project
|
|
|
|
docs = tmp_path / "portfolio-docs"
|
|
docs.mkdir()
|
|
(docs / "cost.txt").write_text(
|
|
"LED-retrofit av lysrorarmaturer i kontorlokaler reduserte energikostnaden.",
|
|
encoding="utf-8",
|
|
)
|
|
return Project(
|
|
id=_PID,
|
|
name="Bundle-backed portfolio project",
|
|
description="bundle-backed project for the semantic-retrieval forwarding proof",
|
|
currency="NOK",
|
|
cost_items=(),
|
|
docs_dir=str(docs),
|
|
verdict_input=_VERDICT_INPUT,
|
|
bundle_dir=str(BUNDLE_DIR),
|
|
verdict_dir=None,
|
|
)
|
|
|
|
|
|
def _generation_prompts(sink: list[str]) -> list[str]:
|
|
return [p for p in sink if "SavingsProposal" in p]
|
|
|
|
|
|
def test_semantic_retrieval_flag_parses_offline(tmp_path, capsys) -> None:
|
|
"""(a) the flag is accepted in single-project mode and the run still stops offline.
|
|
|
|
Carries ``--verdict-dir`` because the flag is now refused without it (Step 3) — the inbox path
|
|
need not exist, since the Steg-7 load is deliberately tolerant of a missing folder."""
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--bundle-dir",
|
|
str(BUNDLE_DIR),
|
|
"--verdict-dir",
|
|
str(tmp_path / "inbox"),
|
|
"--semantic-retrieval",
|
|
"--live-dry-run",
|
|
]
|
|
)
|
|
assert rc == 0
|
|
assert "LIVE-DRY-RUN OK" in capsys.readouterr().out
|
|
|
|
|
|
def test_semantic_retrieval_without_bundle_dir_is_refused(tmp_path, capsys) -> None:
|
|
"""The flag is REFUSED, never ignored: without ``--bundle-dir`` the Step-1 fold never runs, so
|
|
the opt-in could not affect anything the run emits. The refusal names the missing flag verbatim
|
|
(the repo's offender-naming idiom) and carries no traceback."""
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--verdict-dir",
|
|
str(tmp_path / "inbox"),
|
|
"--semantic-retrieval",
|
|
]
|
|
)
|
|
err = capsys.readouterr().err
|
|
assert rc == 1
|
|
assert "run refused:" in err
|
|
assert "--bundle-dir" in err
|
|
assert "Traceback" not in err
|
|
|
|
|
|
def test_semantic_retrieval_without_verdict_dir_is_refused(capsys) -> None:
|
|
"""Same contract on the other half: ``--verdict-dir`` is the ONLY route by which ``main()`` can
|
|
hand ``run_project`` a non-empty store, so without it the fold's ``store.verdicts`` guard
|
|
short-circuits and the flag is inert."""
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--bundle-dir",
|
|
str(BUNDLE_DIR),
|
|
"--semantic-retrieval",
|
|
]
|
|
)
|
|
err = capsys.readouterr().err
|
|
assert rc == 1
|
|
assert "run refused:" in err
|
|
assert "--verdict-dir" in err
|
|
assert "Traceback" not in err
|
|
|
|
|
|
def test_semantic_retrieval_with_both_dirs_is_not_refused(tmp_path, capsys) -> None:
|
|
"""The refusal must be exactly as wide as the condition that makes the flag inert — with both
|
|
flags present the run proceeds (here to the offline dry-run return), not to a refusal."""
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--bundle-dir",
|
|
str(BUNDLE_DIR),
|
|
"--verdict-dir",
|
|
str(tmp_path / "inbox"),
|
|
"--semantic-retrieval",
|
|
"--live-dry-run",
|
|
]
|
|
)
|
|
out = capsys.readouterr()
|
|
assert rc == 0
|
|
assert "LIVE-DRY-RUN OK" in out.out
|
|
assert "--semantic-retrieval" not in out.err
|
|
|
|
|
|
def test_embedder_config_valid_file_parses_offline(tmp_path, capsys) -> None:
|
|
"""``--embedder-config <valid>`` is accepted and the run still stops offline — the registry is
|
|
reachable from the CLI, not merely importable."""
|
|
cfg = tmp_path / "embedder.json"
|
|
cfg.write_text('{"type": "fake"}', encoding="utf-8")
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--bundle-dir",
|
|
str(BUNDLE_DIR),
|
|
"--verdict-dir",
|
|
str(tmp_path / "inbox"),
|
|
"--semantic-retrieval",
|
|
"--embedder-config",
|
|
str(cfg),
|
|
"--live-dry-run",
|
|
]
|
|
)
|
|
assert rc == 0
|
|
assert "LIVE-DRY-RUN OK" in capsys.readouterr().out
|
|
|
|
|
|
def test_embedder_config_missing_file_refuses(tmp_path, capsys) -> None:
|
|
"""Fail-fast startup config: a missing file refuses the run (rc 1, no traceback), surfacing
|
|
through the existing structured-refusal handler."""
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--bundle-dir",
|
|
str(BUNDLE_DIR),
|
|
"--verdict-dir",
|
|
str(tmp_path / "inbox"),
|
|
"--semantic-retrieval",
|
|
"--embedder-config",
|
|
"/nonexistent-embedder-config.json",
|
|
"--live-dry-run",
|
|
]
|
|
)
|
|
err = capsys.readouterr().err
|
|
assert rc == 1
|
|
assert "refused" in err.lower()
|
|
assert "Traceback" not in err
|
|
|
|
|
|
def test_embedder_config_unknown_type_refuses(tmp_path, capsys) -> None:
|
|
"""A config naming an embedder outside the closed registry is REFUSED, never resolved — this
|
|
is the CLI-level face of the no-import-path rule."""
|
|
cfg = tmp_path / "embedder.json"
|
|
cfg.write_text('{"type": "my_pkg.mod:NetworkEmbedder"}', encoding="utf-8")
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--bundle-dir",
|
|
str(BUNDLE_DIR),
|
|
"--verdict-dir",
|
|
str(tmp_path / "inbox"),
|
|
"--semantic-retrieval",
|
|
"--embedder-config",
|
|
str(cfg),
|
|
"--live-dry-run",
|
|
]
|
|
)
|
|
err = capsys.readouterr().err
|
|
assert rc == 1
|
|
assert "refused" in err.lower()
|
|
assert "my_pkg" not in sys.modules
|
|
|
|
|
|
def test_embedder_config_without_semantic_retrieval_is_refused(tmp_path, capsys) -> None:
|
|
"""The embedder is consulted ONLY by the hybrid ranker ``--semantic-retrieval`` builds (proved
|
|
by ``test_injected_embedder_is_never_consulted_with_the_flag_off`` below), so without that flag
|
|
the config is loaded and then dropped. That is the silent-ignore this CLI's flag contract
|
|
refuses — the same ground on which ``--semantic-retrieval`` itself is refused when it cannot
|
|
take effect."""
|
|
cfg = tmp_path / "embedder.json"
|
|
cfg.write_text('{"type": "fake"}', encoding="utf-8")
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--bundle-dir",
|
|
str(BUNDLE_DIR),
|
|
"--verdict-dir",
|
|
str(tmp_path / "inbox"),
|
|
"--embedder-config",
|
|
str(cfg),
|
|
"--live-dry-run",
|
|
]
|
|
)
|
|
err = capsys.readouterr().err
|
|
assert rc == 1
|
|
assert "--embedder-config" in err
|
|
assert "--semantic-retrieval" in err
|
|
|
|
|
|
def test_embedder_config_without_semantic_retrieval_is_refused_in_portfolio_mode(
|
|
tmp_path, capsys
|
|
) -> None:
|
|
"""Mode-independent: BOTH run modes gate the embedder on the same flag (``run_portfolio``
|
|
forwards it to ``run_project``, which builds the ranker or nothing), so a refusal scoped to
|
|
single-project mode would leave portfolio mode with the silent drop."""
|
|
cfg = tmp_path / "embedder.json"
|
|
cfg.write_text('{"type": "fake"}', encoding="utf-8")
|
|
rc = run.main(["--portfolio", "--embedder-config", str(cfg)])
|
|
err = capsys.readouterr().err
|
|
assert rc == 1
|
|
assert "--embedder-config" in err
|
|
assert "--semantic-retrieval" in err
|
|
|
|
|
|
def test_embedder_config_is_refused_before_the_scripted_banner(tmp_path, capsys) -> None:
|
|
"""The refusal sits ABOVE the scripted door, mirroring the required-args hoist: a refused run
|
|
must never print the honesty banner, which claims a scripted loop actually closed."""
|
|
cfg = tmp_path / "embedder.json"
|
|
cfg.write_text('{"type": "fake"}', encoding="utf-8")
|
|
replies = tmp_path / "replies.json"
|
|
replies.write_text(
|
|
json.dumps({"proposer": _ENERGY_REPLY, "checker": "VERDICT: APPROVE"}), encoding="utf-8"
|
|
)
|
|
rc = run.main(
|
|
[
|
|
_PID,
|
|
"--docs-dir",
|
|
str(BUNDLE_DIR),
|
|
"--bundle-dir",
|
|
str(BUNDLE_DIR),
|
|
"--verdict-dir",
|
|
str(tmp_path / "inbox"),
|
|
"--embedder-config",
|
|
str(cfg),
|
|
"--scripted-replies",
|
|
str(replies),
|
|
]
|
|
)
|
|
out = capsys.readouterr()
|
|
assert rc == 1
|
|
assert "SCRIPTED OFFLINE RUN" not in out.out
|
|
|
|
|
|
def test_report_with_embedder_config_is_refused(tmp_path, capsys) -> None:
|
|
"""--report stays an ALLOWLIST: a new config flag must be refused there like every other one,
|
|
else it would be silently dropped."""
|
|
cfg = tmp_path / "embedder.json"
|
|
cfg.write_text('{"type": "fake"}', encoding="utf-8")
|
|
ledger_file = tmp_path / "ledger.json"
|
|
SavingsLedger(entries=[]).save(str(ledger_file))
|
|
rc = run.main(["--report", "--ledger", str(ledger_file), "--embedder-config", str(cfg)])
|
|
assert rc == 1
|
|
assert "refused" in capsys.readouterr().err.lower()
|
|
|
|
|
|
def test_semantic_retrieval_is_not_refused_in_portfolio_mode(capsys) -> None:
|
|
"""The flag is valid in BOTH modes (like --dimension-config), so the portfolio partition must
|
|
not name it. Probed via a run that IS refused for a different flag: the refusal lists
|
|
``--docs-dir`` and must NOT mention ``--semantic-retrieval``."""
|
|
rc = run.main(["--portfolio", "--semantic-retrieval", "--docs-dir", str(BUNDLE_DIR)])
|
|
err = capsys.readouterr().err
|
|
assert rc == 1
|
|
assert "--docs-dir" in err
|
|
assert "--semantic-retrieval" not in err
|
|
|
|
|
|
def test_report_with_semantic_retrieval_is_refused(tmp_path, capsys) -> None:
|
|
"""--report is an ALLOWLIST: only --ledger/--json ride along. A silently-dropped
|
|
--semantic-retrieval would break that partition."""
|
|
ledger_file = tmp_path / "ledger.json"
|
|
SavingsLedger(entries=[]).save(str(ledger_file))
|
|
rc = run.main(["--report", "--ledger", str(ledger_file), "--semantic-retrieval"])
|
|
assert rc == 1
|
|
assert "refused" in capsys.readouterr().err.lower()
|
|
|
|
|
|
async def test_semantic_retrieval_on_swaps_the_fewshot_reaching_the_prompt(
|
|
make_recording_client_factory,
|
|
) -> None:
|
|
"""(c) the substantive run-level proof: with ``semantic_retrieval=True`` and ``top_k=1``, the
|
|
verdict that only cosine can surface is the one whose rationale reaches the hypothesis prompt.
|
|
Drives the REAL Step-1 fold via the recording client — not ``--live-dry-run``, which returns
|
|
before the fold."""
|
|
factory, recorded = make_recording_client_factory(_ENERGY_REPLY)
|
|
|
|
await run_project(
|
|
_PID,
|
|
"local",
|
|
docs_dir=str(BUNDLE_DIR),
|
|
bundle_dir=str(BUNDLE_DIR),
|
|
verdict_input=_VERDICT_INPUT,
|
|
store=_tied_pair_store(),
|
|
client_factory=factory,
|
|
top_k=1,
|
|
semantic_retrieval=True,
|
|
)
|
|
|
|
gen_prompts = _generation_prompts(recorded)
|
|
assert gen_prompts, "the generation call must have happened"
|
|
assert any(_TIE_MARKER in p for p in gen_prompts), (
|
|
"the cosine-surfaced verdict did not reach the hypothesis prompt — "
|
|
"--semantic-retrieval is not installing the HybridRanker before the Step-1 fold"
|
|
)
|
|
assert any(_MARKER_ID in p for p in gen_prompts)
|
|
|
|
|
|
async def test_semantic_retrieval_off_leaves_the_structural_pick_in_the_prompt(
|
|
make_recording_client_factory,
|
|
) -> None:
|
|
"""CAUSALITY CONTROL — the identical run with the flag OFF must carry the STRUCTURAL winner
|
|
instead, and no marker. This is what makes the positive above load-bearing: the swap is caused
|
|
by the flag, not by the fixture merely containing the marker."""
|
|
factory, recorded = make_recording_client_factory(_ENERGY_REPLY)
|
|
|
|
await run_project(
|
|
_PID,
|
|
"local",
|
|
docs_dir=str(BUNDLE_DIR),
|
|
bundle_dir=str(BUNDLE_DIR),
|
|
verdict_input=_VERDICT_INPUT,
|
|
store=_tied_pair_store(),
|
|
client_factory=factory,
|
|
top_k=1,
|
|
)
|
|
|
|
assert all(_TIE_MARKER not in p for p in recorded), (
|
|
"the marker reached a prompt with --semantic-retrieval OFF — the default path is not the "
|
|
"structural ranking, or the assertion is not load-bearing"
|
|
)
|
|
gen_prompts = _generation_prompts(recorded)
|
|
assert any(_DISTRACTOR_ID in p for p in gen_prompts), (
|
|
"the structural winner did not reach the prompt — the default fold is broken"
|
|
)
|
|
|
|
|
|
async def test_semantic_retrieval_does_not_leak_into_a_reused_store(
|
|
make_recording_client_factory,
|
|
) -> None:
|
|
"""LEAK CONTROL — the opt-in must not outlive the run that asked for it.
|
|
|
|
The store is caller-owned. Installing the ranker on it (``store.retriever = ...``) meant a
|
|
flag-ON run silently governed every LATER retrieval on that same object, so a subsequent run
|
|
with the flag OFF still ranked semantically. Here the SAME store instance is driven twice:
|
|
once with the flag on, once without. The second run must give the STRUCTURAL pick, and the
|
|
store must be left exactly as the caller handed it over.
|
|
|
|
Detach point: assign ``store.retriever`` in ``run_project`` instead of passing the ranker per
|
|
call → RED."""
|
|
store = _tied_pair_store()
|
|
assert store.retriever is None # precondition: the caller handed over a clean store
|
|
|
|
factory_on, _ = make_recording_client_factory(_ENERGY_REPLY)
|
|
await run_project(
|
|
_PID,
|
|
"local",
|
|
docs_dir=str(BUNDLE_DIR),
|
|
bundle_dir=str(BUNDLE_DIR),
|
|
verdict_input=_VERDICT_INPUT,
|
|
store=store,
|
|
client_factory=factory_on,
|
|
top_k=1,
|
|
semantic_retrieval=True,
|
|
)
|
|
|
|
factory_off, recorded_off = make_recording_client_factory(_ENERGY_REPLY)
|
|
await run_project(
|
|
_PID,
|
|
"local",
|
|
docs_dir=str(BUNDLE_DIR),
|
|
bundle_dir=str(BUNDLE_DIR),
|
|
verdict_input=_VERDICT_INPUT,
|
|
store=store,
|
|
client_factory=factory_off,
|
|
top_k=1,
|
|
)
|
|
|
|
assert store.retriever is None, (
|
|
"run_project mutated the caller's store — the opt-in leaked out of the run that asked "
|
|
"for it"
|
|
)
|
|
assert all(_TIE_MARKER not in p for p in _generation_prompts(recorded_off)), (
|
|
"the flag-OFF run still ranked semantically — the previous run's opt-in leaked through "
|
|
"the shared store"
|
|
)
|
|
|
|
|
|
async def test_run_portfolio_forwards_semantic_retrieval_to_each_project(
|
|
make_recording_client_factory, monkeypatch, tmp_path
|
|
) -> None:
|
|
"""WIRING — ``run_portfolio(semantic_retrieval=True)`` must reach each project's Step-1 fold.
|
|
|
|
No test in the repo drove the portfolio path with this flag before: the pre-loop
|
|
``store.retriever`` install was covered only by the single-project tests, so deleting the
|
|
forwarding would have gone unnoticed.
|
|
|
|
Detach point: drop ``semantic_retrieval=semantic_retrieval`` from the ``run_project`` call
|
|
inside ``run_portfolio`` → RED."""
|
|
factory, recorded = make_recording_client_factory(_ENERGY_REPLY)
|
|
store = _tied_pair_store()
|
|
|
|
project = _bundle_reference_project(tmp_path)
|
|
monkeypatch.setattr("portfolio_optimiser.run.load_reference_projects", lambda: [project])
|
|
|
|
await run.run_portfolio(
|
|
[project.id],
|
|
profile="local",
|
|
store=store,
|
|
client_factory=factory,
|
|
top_k=1,
|
|
semantic_retrieval=True,
|
|
)
|
|
|
|
gen_prompts = _generation_prompts(recorded)
|
|
assert gen_prompts, "the generation call must have happened"
|
|
assert any(_TIE_MARKER in p for p in gen_prompts), (
|
|
"the cosine-surfaced verdict did not reach the hypothesis prompt — run_portfolio is not "
|
|
"forwarding semantic_retrieval to run_project"
|
|
)
|
|
|
|
|
|
async def test_run_portfolio_without_the_flag_keeps_the_structural_pick(
|
|
make_recording_client_factory, monkeypatch, tmp_path
|
|
) -> None:
|
|
"""CAUSALITY CONTROL for the portfolio arm — the identical pass with the flag absent must
|
|
carry the structural winner and no marker."""
|
|
factory, recorded = make_recording_client_factory(_ENERGY_REPLY)
|
|
store = _tied_pair_store()
|
|
|
|
project = _bundle_reference_project(tmp_path)
|
|
monkeypatch.setattr("portfolio_optimiser.run.load_reference_projects", lambda: [project])
|
|
|
|
await run.run_portfolio(
|
|
[project.id],
|
|
profile="local",
|
|
store=store,
|
|
client_factory=factory,
|
|
top_k=1,
|
|
)
|
|
|
|
assert all(_TIE_MARKER not in p for p in recorded), (
|
|
"the marker reached a prompt without the flag — the portfolio default is not structural"
|
|
)
|
|
|
|
|
|
def _counting_embedder(calls: list) -> object:
|
|
"""A real embedder (the shipped ``FakeEmbedder``) that records every consultation, so the
|
|
flag-ON arm completes normally and the difference between the arms is the CALL COUNT alone."""
|
|
from portfolio_optimiser.semretrieval import FakeEmbedder
|
|
|
|
fake = FakeEmbedder()
|
|
|
|
def embed(features):
|
|
calls.append(features)
|
|
return fake(features)
|
|
|
|
return embed
|
|
|
|
|
|
async def test_injected_embedder_is_never_consulted_with_the_flag_off(
|
|
make_recording_client_factory,
|
|
) -> None:
|
|
"""THE MEASUREMENT BEHIND THE REFUSAL — an injected embedder is consulted ZERO times when
|
|
``semantic_retrieval`` is off, because the ``HybridRanker`` that holds it is only built when
|
|
the flag is on; the structural retriever takes no embedder at all.
|
|
|
|
This is why ``--embedder-config`` without ``--semantic-retrieval`` is REFUSED rather than
|
|
WIRED (contrast ``--scripted-replies`` in portfolio mode, where a seam to wire existed). If
|
|
someone later gives the embedder a job on the default path, this test goes RED and the CLI
|
|
refusal above becomes wrong — which is exactly the signal wanted."""
|
|
calls: list = []
|
|
factory, _ = make_recording_client_factory(_ENERGY_REPLY)
|
|
|
|
await run_project(
|
|
_PID,
|
|
"local",
|
|
docs_dir=str(BUNDLE_DIR),
|
|
bundle_dir=str(BUNDLE_DIR),
|
|
verdict_input=_VERDICT_INPUT,
|
|
store=_tied_pair_store(),
|
|
client_factory=factory,
|
|
top_k=1,
|
|
embedder=_counting_embedder(calls),
|
|
)
|
|
|
|
assert calls == [], (
|
|
"the injected embedder was consulted with semantic_retrieval OFF — the CLI refusal of "
|
|
"--embedder-config without --semantic-retrieval is no longer justified"
|
|
)
|
|
|
|
|
|
async def test_injected_embedder_is_consulted_with_the_flag_on(
|
|
make_recording_client_factory,
|
|
) -> None:
|
|
"""CAUSALITY CONTROL — the identical run with the flag ON consults the same embedder. Without
|
|
this the zero above could equally mean the embedder was never reachable at all."""
|
|
calls: list = []
|
|
factory, _ = make_recording_client_factory(_ENERGY_REPLY)
|
|
|
|
await run_project(
|
|
_PID,
|
|
"local",
|
|
docs_dir=str(BUNDLE_DIR),
|
|
bundle_dir=str(BUNDLE_DIR),
|
|
verdict_input=_VERDICT_INPUT,
|
|
store=_tied_pair_store(),
|
|
client_factory=factory,
|
|
top_k=1,
|
|
embedder=_counting_embedder(calls),
|
|
semantic_retrieval=True,
|
|
)
|
|
|
|
assert calls, "the injected embedder was never consulted with the flag ON"
|