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>
449 lines
19 KiB
Python
449 lines
19 KiB
Python
"""S3.1 — semantic retrieval unit tests (SC4 here; SC1/SC3/SC7 added by later steps).
|
|
|
|
The embedder must be a PURE function of the structured features: identical features yield a
|
|
byte-identical float64 vector in this process AND in a fresh process started with a different
|
|
``PYTHONHASHSEED`` — the cross-process leg is what rules out Python's salted builtin ``hash()``
|
|
(the exact failure mode that would make retrieval order drift between runs).
|
|
|
|
Components are non-negative, so ``cosine`` stays in ``[0, 1]`` and blends with the structural
|
|
score (also ``[0, 1]``) without a scale mismatch.
|
|
|
|
Pattern: tests/test_verdicts.py (determinism) + tests/test_retrieval.py (fixtures).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from portfolio_optimiser.semretrieval import (
|
|
EMBED_DIM,
|
|
Embedder,
|
|
EmbedderConfig,
|
|
FakeEmbedder,
|
|
HybridRanker,
|
|
StructuralRetriever,
|
|
build_embedder,
|
|
cosine,
|
|
load_embedder_config,
|
|
load_vector_store,
|
|
save_vector_store,
|
|
)
|
|
from portfolio_optimiser.verdicts import (
|
|
ProposalFeatures,
|
|
Verdict,
|
|
VerdictStore,
|
|
similarity,
|
|
)
|
|
|
|
_SRC = Path(__file__).resolve().parents[1] / "src"
|
|
|
|
|
|
def _features(
|
|
codes: frozenset[str] = frozenset({"05.2", "03.1"}),
|
|
measure_type: str = "scope_reduction",
|
|
saving: float = 220_000.0,
|
|
description: str = "licence scope reduction near head office",
|
|
) -> types.SimpleNamespace:
|
|
"""Duck-typed stand-in for ``ProposalFeatures`` — semretrieval never imports ``verdicts``
|
|
at runtime, so the unit tests do not need the real (MAF-bound) type either."""
|
|
return types.SimpleNamespace(
|
|
affected_codes=codes,
|
|
measure_type=measure_type,
|
|
claimed_saving_nok=saving,
|
|
description=description,
|
|
)
|
|
|
|
|
|
def test_fake_embedder_is_pure() -> None:
|
|
embedder = FakeEmbedder()
|
|
assert np.array_equal(embedder(_features()), embedder(_features()))
|
|
|
|
|
|
def test_fake_embedder_has_fixed_float64_dimension() -> None:
|
|
vec = FakeEmbedder()(_features())
|
|
assert vec.shape == (EMBED_DIM,)
|
|
assert vec.dtype == np.dtype("<f8")
|
|
|
|
|
|
def test_fake_embedder_is_insensitive_to_code_set_ordering() -> None:
|
|
"""``affected_codes`` is a set — the canonical string sorts it, so insertion order
|
|
cannot leak into the vector."""
|
|
a = FakeEmbedder()(_features(codes=frozenset({"05.2", "03.1"})))
|
|
b = FakeEmbedder()(_features(codes=frozenset({"03.1", "05.2"})))
|
|
assert np.array_equal(a, b)
|
|
|
|
|
|
def test_distinct_features_give_distinct_vectors() -> None:
|
|
"""SC4, NARROWED: "distinct" means distinct in the EMBEDDED fields — the structural triple.
|
|
The description-only variant moved to the test below, which asserts the opposite."""
|
|
embedder = FakeEmbedder()
|
|
baseline = embedder(_features())
|
|
for variant in (
|
|
_features(codes=frozenset({"09.1"})),
|
|
_features(measure_type="rate_renegotiation"),
|
|
_features(saving=900_000.0), # different magnitude bucket
|
|
):
|
|
assert not np.array_equal(baseline, embedder(variant))
|
|
|
|
|
|
def test_features_differing_only_in_description_embed_identically() -> None:
|
|
"""SC4's narrowing, asserted POSITIVELY rather than left implicit — this is the same
|
|
assertion the loop above used to carry, with the opposite sign.
|
|
|
|
``description`` is excluded from the embedding on purpose: ``verdicts.similarity`` already
|
|
ignores text by design and ``verdicts._mint_id`` already hashes only the structural triple, so
|
|
two proposals differing only in prose are ONE verdict as far as the rest of the system is
|
|
concerned. Leaving the embedding as the sole place surface text still counted made the
|
|
framework's own restatement of a query outrank real expert prose (defect P1).
|
|
|
|
Detach point: put ``features.description`` back into ``_canonical_feature_string`` → RED."""
|
|
embedder = FakeEmbedder()
|
|
assert np.array_equal(
|
|
embedder(_features()),
|
|
embedder(_features(description="totally different wording")),
|
|
)
|
|
|
|
|
|
def test_fake_embedder_components_are_non_negative() -> None:
|
|
"""Non-negative components keep ``cosine`` in [0, 1], matching the structural score's
|
|
range — no scale mismatch when the two are blended by HybridRanker."""
|
|
vec = FakeEmbedder()(_features())
|
|
assert bool((vec >= 0.0).all())
|
|
|
|
|
|
def test_fake_embedder_is_bit_stable_across_processes(tmp_path: Path) -> None:
|
|
"""RED for any ``hash()``-based implementation: the child runs with a different
|
|
``PYTHONHASHSEED``, so a salted hash would produce different bytes."""
|
|
embedder = FakeEmbedder()
|
|
here = tmp_path / "here.npy"
|
|
np.save(here, embedder(_features()))
|
|
|
|
there = tmp_path / "there.npy"
|
|
script = (
|
|
"import types, numpy as np\n"
|
|
"from portfolio_optimiser.semretrieval import FakeEmbedder\n"
|
|
"f = types.SimpleNamespace(\n"
|
|
" affected_codes=frozenset({'05.2', '03.1'}),\n"
|
|
" measure_type='scope_reduction',\n"
|
|
" claimed_saving_nok=220000.0,\n"
|
|
" description='licence scope reduction near head office',\n"
|
|
")\n"
|
|
f"np.save({str(there)!r}, FakeEmbedder()(f))\n"
|
|
)
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", script],
|
|
env={"PATH": "/usr/bin:/bin", "PYTHONPATH": str(_SRC), "PYTHONHASHSEED": "12345"},
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 0, result.stderr
|
|
assert here.read_bytes() == there.read_bytes()
|
|
|
|
|
|
def test_cosine_of_identical_unit_vectors_is_one() -> None:
|
|
"""Approximate by necessity: a float64 dot of a normalized vector with itself lands within
|
|
an ulp or two of 1.0, never exactly on it. Ranking never depends on the exact value —
|
|
``HybridRanker`` rounds to 9 decimals and breaks remaining ties on ``id``."""
|
|
vec = FakeEmbedder()(_features())
|
|
assert cosine(vec, vec) == pytest.approx(1.0)
|
|
|
|
|
|
def test_cosine_is_bounded_to_unit_interval() -> None:
|
|
embedder = FakeEmbedder()
|
|
score = cosine(embedder(_features()), embedder(_features(codes=frozenset({"09.1"}))))
|
|
assert 0.0 <= score <= 1.0
|
|
|
|
|
|
def test_cosine_of_zero_norm_is_zero() -> None:
|
|
"""A zero vector must never produce NaN — NaN in a sort key silently corrupts ranking."""
|
|
zeros = np.zeros(EMBED_DIM, dtype="<f8")
|
|
vec = FakeEmbedder()(_features())
|
|
assert cosine(zeros, vec) == 0.0
|
|
assert cosine(vec, zeros) == 0.0
|
|
assert cosine(zeros, zeros) == 0.0
|
|
|
|
|
|
@pytest.mark.parametrize("bad_value", [np.nan, np.inf, -np.inf])
|
|
def test_cosine_refuses_a_non_finite_vector(bad_value: float) -> None:
|
|
"""The zero-norm guard tests ``norm == 0.0``, which a NaN or inf norm passes straight through:
|
|
measured, ``cosine(unit, nan_vector)`` and ``cosine(unit, inf_vector)`` BOTH returned ``nan``.
|
|
|
|
Refuse rather than repair, and deliberately NOT symmetric with the zero-norm branch above: a
|
|
zero vector is a legitimate, handled state (``FakeEmbedder`` returns ``np.zeros`` by design),
|
|
whereas a non-finite component only ever means the injected embedder is broken. Coercing it to
|
|
``0.0`` would launder that into "no semantic similarity" and let ranking proceed on a forged
|
|
signal — the same reasoning that makes ``read_spend`` raise on corrupt content instead of
|
|
reading it as zero."""
|
|
vec = FakeEmbedder()(_features())
|
|
bad = np.full(EMBED_DIM, bad_value, dtype="<f8")
|
|
with pytest.raises(ValueError, match="non-finite"):
|
|
cosine(vec, bad)
|
|
with pytest.raises(ValueError, match="non-finite"):
|
|
cosine(bad, vec)
|
|
|
|
|
|
# --- SC1: the Retriever seam is additive — the DEFAULT ranking is byte-identical to today ---
|
|
|
|
_QUERY = ProposalFeatures(
|
|
affected_codes=frozenset({"05.2", "03.1"}),
|
|
measure_type="scope_reduction",
|
|
claimed_saving_nok=220_000, # bucket [100k, 500k)
|
|
description="licence scope reduction near head office",
|
|
)
|
|
|
|
|
|
def _store_with_true_match_and_decoys() -> tuple[VerdictStore, str]:
|
|
"""Mirrors ``tests/test_verdicts.py::_store_with_true_match_and_decoys`` — the true match
|
|
shares the STRUCTURED fields with the query but uses different wording; the decoys share
|
|
surface text but differ structurally."""
|
|
true_match = Verdict(
|
|
id="TRUE",
|
|
proposal_features=ProposalFeatures(
|
|
affected_codes=frozenset({"05.2", "03.1"}),
|
|
measure_type="scope_reduction",
|
|
claimed_saving_nok=200_000,
|
|
description="zzz totally unrelated wording alpha beta",
|
|
),
|
|
decision="approved",
|
|
rationale="prior scope reduction on the same codes was approved",
|
|
)
|
|
decoy_low = Verdict(
|
|
id="DECOY-LOW",
|
|
proposal_features=ProposalFeatures(
|
|
affected_codes=frozenset({"09.1"}),
|
|
measure_type="rate_renegotiation",
|
|
claimed_saving_nok=50_000,
|
|
description="licence scope reduction near head office",
|
|
),
|
|
decision="rejected",
|
|
rationale="surface-text decoy",
|
|
)
|
|
decoy_high = Verdict(
|
|
id="DECOY-HIGH",
|
|
proposal_features=ProposalFeatures(
|
|
affected_codes=frozenset({"21.2"}),
|
|
measure_type="material_substitution",
|
|
claimed_saving_nok=700_000,
|
|
description="licence scope reduction extra words",
|
|
),
|
|
decision="rejected",
|
|
rationale="surface-text decoy",
|
|
)
|
|
return VerdictStore(verdicts=[decoy_low, true_match, decoy_high]), "TRUE"
|
|
|
|
|
|
def test_default_retriever_is_none() -> None:
|
|
"""Opt-in seam: a store built the ordinary way carries no retriever, so nothing about
|
|
default retrieval changes unless a caller deliberately installs one."""
|
|
store, _ = _store_with_true_match_and_decoys()
|
|
assert store.retriever is None
|
|
|
|
|
|
def test_default_retrieve_matches_the_pre_seam_structural_sort() -> None:
|
|
"""SC1 — with no retriever installed, ``retrieve`` must reproduce the exact ordering the
|
|
pre-seam inline sort produced: key ``(-similarity, id)``, sliced to k."""
|
|
store, _ = _store_with_true_match_and_decoys()
|
|
expected = [
|
|
v.id
|
|
for v in sorted(
|
|
store.verdicts,
|
|
key=lambda v: (-similarity(_QUERY, v.proposal_features), v.id),
|
|
)[:3]
|
|
]
|
|
assert [h.id for h in store.retrieve(_QUERY, k=3)] == expected
|
|
|
|
|
|
def test_structural_retriever_is_the_wired_default_not_dead_code() -> None:
|
|
"""SC1 — the default path goes through ``StructuralRetriever``, so it is exercised code,
|
|
not an unused parallel implementation that can silently rot."""
|
|
store, _ = _store_with_true_match_and_decoys()
|
|
direct = StructuralRetriever(similarity).rank(_QUERY, store.verdicts, 3)
|
|
assert [h.id for h in store.retrieve(_QUERY, k=3)] == [v.id for v in direct]
|
|
|
|
|
|
def test_retrieve_still_rejects_non_positive_k() -> None:
|
|
"""The pre-seam ``k <= 0`` guard must survive the delegation rewrite
|
|
(asserted independently by tests/test_verdicts.py::test_retrieve_rejects_non_positive_k)."""
|
|
store, _ = _store_with_true_match_and_decoys()
|
|
with pytest.raises(ValueError):
|
|
store.retrieve(_QUERY, k=0)
|
|
|
|
|
|
def test_installed_retriever_actually_routes_retrieval() -> None:
|
|
"""Delegation proof: a sentinel retriever must receive the call and its result must be
|
|
what ``retrieve`` returns — otherwise the seam is decorative."""
|
|
store, _ = _store_with_true_match_and_decoys()
|
|
calls: list[tuple[ProposalFeatures, int, int]] = []
|
|
|
|
class _Sentinel:
|
|
def rank(self, query: ProposalFeatures, candidates: list[Verdict], k: int) -> list[Verdict]:
|
|
calls.append((query, len(candidates), k))
|
|
return list(reversed(candidates))[:k]
|
|
|
|
store.retriever = _Sentinel()
|
|
hits = store.retrieve(_QUERY, k=2)
|
|
assert calls == [(_QUERY, 3, 2)]
|
|
assert [h.id for h in hits] == ["DECOY-HIGH", "TRUE"]
|
|
|
|
|
|
def test_hybrid_ranker_is_injectable_and_routes_retrieval() -> None:
|
|
"""A HybridRanker installed on the store must be the ranker ``retrieve`` delegates to."""
|
|
store, _ = _store_with_true_match_and_decoys()
|
|
ranker = HybridRanker(FakeEmbedder(), similarity)
|
|
expected = [v.id for v in ranker.rank(_QUERY, store.verdicts, 3)]
|
|
store.retriever = ranker
|
|
assert [h.id for h in store.retrieve(_QUERY, k=3)] == expected
|
|
|
|
|
|
def test_hybrid_ranker_at_weight_zero_equals_the_structural_ranking() -> None:
|
|
"""Weight 0 removes the semantic term entirely — the blend must then collapse onto the
|
|
structural ordering. This is the control the Step-5 detach test relies on."""
|
|
store, _ = _store_with_true_match_and_decoys()
|
|
hybrid = HybridRanker(FakeEmbedder(), similarity, weight=0.0)
|
|
structural = StructuralRetriever(similarity)
|
|
assert [v.id for v in hybrid.rank(_QUERY, store.verdicts, 3)] == [
|
|
v.id for v in structural.rank(_QUERY, store.verdicts, 3)
|
|
]
|
|
|
|
|
|
# --- The embedder config seam: a CLOSED registry, never an import path -------------------------
|
|
|
|
|
|
def test_build_embedder_returns_the_shipped_fake_for_the_known_type() -> None:
|
|
"""The registry's one shipped entry. Mirrors ``tests/test_notify.py``'s build_notifier tests."""
|
|
built = build_embedder(EmbedderConfig(type="fake"))
|
|
assert isinstance(built, FakeEmbedder)
|
|
assert isinstance(built, Embedder) # satisfies the runtime-checkable protocol
|
|
|
|
|
|
def test_build_embedder_rejects_an_unknown_type() -> None:
|
|
"""Closed vocabulary: an unknown type RAISES rather than falling back to a default, and the
|
|
message names the seam so the constraint is discoverable at the failure."""
|
|
with pytest.raises(ValueError, match="unknown embedder type"):
|
|
build_embedder(EmbedderConfig(type="nope"))
|
|
|
|
|
|
def test_build_embedder_refuses_an_import_path_instead_of_importing_it() -> None:
|
|
"""SECURITY — the registry must never resolve a dotted ``module:Class`` string from config.
|
|
|
|
Doing so would be arbitrary code execution at config-load time and would hand a config file the
|
|
ability to import a module that opens a socket — routing around this module's no-network guard,
|
|
which is scoped to ``semretrieval.py`` and blind to a third module by construction. The config
|
|
is accepted as DATA and then refused; nothing is imported."""
|
|
hostile = EmbedderConfig(type="evil_pkg.embedders:NetworkEmbedder")
|
|
with pytest.raises(ValueError, match="unknown embedder type"):
|
|
build_embedder(hostile)
|
|
assert "evil_pkg" not in sys.modules
|
|
|
|
|
|
def test_embedder_config_rejects_an_empty_type() -> None:
|
|
with pytest.raises(ValidationError):
|
|
EmbedderConfig(type="")
|
|
|
|
|
|
def test_load_embedder_config_is_fail_fast_on_a_missing_file(tmp_path: Path) -> None:
|
|
"""Authoritative startup config — absence RAISES, in deliberate contrast to the tolerant
|
|
verdict-inbox RAW layer which skips bad files."""
|
|
with pytest.raises(FileNotFoundError):
|
|
load_embedder_config(tmp_path / "nope.json")
|
|
|
|
|
|
def test_load_embedder_config_is_fail_fast_on_a_malformed_file(tmp_path: Path) -> None:
|
|
bad = tmp_path / "embedder.json"
|
|
bad.write_text('{"type": 17}', encoding="utf-8")
|
|
with pytest.raises(ValidationError):
|
|
load_embedder_config(bad)
|
|
|
|
|
|
def test_load_embedder_config_round_trips_a_valid_file(tmp_path: Path) -> None:
|
|
good = tmp_path / "embedder.json"
|
|
good.write_text('{"type": "fake"}', encoding="utf-8")
|
|
assert isinstance(build_embedder(load_embedder_config(good)), FakeEmbedder)
|
|
|
|
|
|
# --- SC3 / SC7: the vector store is byte-deterministic and fails fast on a row/line mismatch ---
|
|
|
|
|
|
def test_vector_store_is_byte_identical_regardless_of_insertion_order(tmp_path: Path) -> None:
|
|
"""SC3 — the store sorts by id before embedding, so the same verdicts written in a different
|
|
order must produce byte-identical artifacts. Pattern:
|
|
tests/test_ledger.py::test_save_is_byte_identical_regardless_of_order."""
|
|
store, _ = _store_with_true_match_and_decoys()
|
|
forward, reverse = tmp_path / "forward", tmp_path / "reverse"
|
|
|
|
save_vector_store(forward, store.verdicts, FakeEmbedder())
|
|
save_vector_store(reverse, list(reversed(store.verdicts)), FakeEmbedder())
|
|
|
|
assert (forward / "vectors.npy").read_bytes() == (reverse / "vectors.npy").read_bytes()
|
|
assert (forward / "vectors.jsonl").read_bytes() == (reverse / "vectors.jsonl").read_bytes()
|
|
|
|
|
|
def test_vector_store_round_trips_ids_sorted(tmp_path: Path) -> None:
|
|
store, _ = _store_with_true_match_and_decoys()
|
|
save_vector_store(tmp_path, store.verdicts, FakeEmbedder())
|
|
|
|
loaded = load_vector_store(tmp_path)
|
|
assert loaded is not None
|
|
ids, matrix = loaded
|
|
assert ids == sorted(v.id for v in store.verdicts)
|
|
assert matrix.shape == (len(ids), EMBED_DIM)
|
|
assert matrix.dtype == np.dtype("<f8")
|
|
|
|
embedder = FakeEmbedder()
|
|
by_id = {v.id: v for v in store.verdicts}
|
|
for row, verdict_id in enumerate(ids):
|
|
assert np.array_equal(matrix[row], embedder(by_id[verdict_id].proposal_features))
|
|
|
|
|
|
def test_vector_store_jsonl_is_one_sorted_object_per_line(tmp_path: Path) -> None:
|
|
"""The id sidecar follows the repo's deterministic on-disk idiom: sorted keys, LF endings,
|
|
trailing newline — so a diff of two runs is empty rather than noisy."""
|
|
store, _ = _store_with_true_match_and_decoys()
|
|
save_vector_store(tmp_path, store.verdicts, FakeEmbedder())
|
|
|
|
raw = (tmp_path / "vectors.jsonl").read_bytes().decode("utf-8")
|
|
assert raw.endswith("\n")
|
|
assert "\r" not in raw
|
|
assert [json.loads(line) for line in raw.splitlines()] == [
|
|
{"id": v.id} for v in sorted(store.verdicts, key=lambda v: v.id)
|
|
]
|
|
|
|
|
|
def test_ranking_is_stable_over_twenty_repetitions() -> None:
|
|
"""The NFR asks for >=20 repetitions, not a single re-run: BLAS reduction order is the thing
|
|
that would make this drift, and it is pinned at module import."""
|
|
base, _ = _store_with_true_match_and_decoys()
|
|
ranker = HybridRanker(FakeEmbedder(), similarity)
|
|
first = [v.id for v in ranker.rank(_QUERY, base.verdicts, 3)]
|
|
for _ in range(20):
|
|
assert [v.id for v in ranker.rank(_QUERY, base.verdicts, 3)] == first
|
|
|
|
|
|
def test_missing_vector_store_loads_as_none(tmp_path: Path) -> None:
|
|
"""Tolerant on absence — the store is a rebuildable cache, so a caller with no store simply
|
|
degrades to structural ranking rather than crashing."""
|
|
assert load_vector_store(tmp_path / "never-written") is None
|
|
assert load_vector_store(tmp_path) is None # directory exists but holds no artifacts
|
|
|
|
|
|
def test_vector_store_row_line_mismatch_fails_fast(tmp_path: Path) -> None:
|
|
"""SC7 — a truncated sidecar would silently mis-map every vector to the wrong verdict id.
|
|
That must raise, not rank garbage (mirrors SavingsLedger.load's fail-fast posture)."""
|
|
store, _ = _store_with_true_match_and_decoys()
|
|
save_vector_store(tmp_path, store.verdicts, FakeEmbedder())
|
|
|
|
sidecar = tmp_path / "vectors.jsonl"
|
|
kept = sidecar.read_text(encoding="utf-8").splitlines()[:-1]
|
|
sidecar.write_text("".join(line + "\n" for line in kept), encoding="utf-8")
|
|
|
|
with pytest.raises(ValueError, match="row"):
|
|
load_vector_store(tmp_path)
|