93 lines
4.8 KiB
Python
93 lines
4.8 KiB
Python
"""S3.1 semretrieval — load-bearing guards (SC5 MAF-free, SC8 no-network).
|
|
|
|
``semretrieval`` is the only numpy importer in the package and it sits on the MAF-free side of
|
|
the D7 fault line. Three independent guards hold that line, each with a named detach point:
|
|
|
|
1. **Meta** — the module is registered in ``_MAF_FREE_MODULES``, so the direct-import AST guard
|
|
(``test_okf_is_maf_free``) actually scans it. Without this the MAF-free claim is green-but-dead.
|
|
2. **Transitive + runtime** — the module body is exec'd standalone AND a ``rank()`` call is made,
|
|
then ``sys.modules`` is checked. The AST guard only sees DIRECT imports and only at import
|
|
time; a lazy ``from portfolio_optimiser.verdicts import similarity`` INSIDE ``rank`` would sail
|
|
past both the AST guard and an import-only probe, and trip only this one.
|
|
3. **No network** — an AST sweep for network modules, since a real embeddings client is an
|
|
extension point and must never be smuggled into the offline default.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
_SEMRETRIEVAL = (
|
|
Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser" / "semretrieval.py"
|
|
)
|
|
|
|
# The offline invariant: no verdict data and no proposal text may leave the machine.
|
|
_NETWORK_ROOTS = {"socket", "urllib", "http", "requests", "httpx", "ftplib", "smtplib"}
|
|
|
|
|
|
def test_semretrieval_registered_maf_free() -> None:
|
|
"""Meta: registered in the MAF-free guard list, so ``test_okf_is_maf_free`` scans it.
|
|
Detach point: drop ``semretrieval.py`` from ``_MAF_FREE_MODULES`` → RED."""
|
|
from tests.test_okf import _MAF_FREE_MODULES
|
|
|
|
assert "semretrieval.py" in _MAF_FREE_MODULES
|
|
|
|
|
|
def test_semretrieval_import_and_rank_pull_in_no_maf_and_no_verdicts() -> None:
|
|
"""The genuine seam: exec the module body standalone, then actually RANK, then inspect
|
|
``sys.modules``.
|
|
|
|
Ranking is the moment a lazy import would fire — ``StructuralRetriever``/``HybridRanker``
|
|
need the structural score, and the design INJECTS it as a callable precisely so that
|
|
``verdicts`` (which imports ``agent_framework``) stays out of this module's runtime graph.
|
|
The candidates are duck-typed ``SimpleNamespace`` stand-ins, never real ``ProposalFeatures``
|
|
/``Verdict`` — building those would import ``verdicts`` in the probe itself and defeat the
|
|
very assertion being made.
|
|
|
|
Detach point: replace the injected ``similarity`` with
|
|
``from portfolio_optimiser.verdicts import similarity`` inside ``rank`` → RED."""
|
|
check = (
|
|
"import importlib.util, sys, types\n"
|
|
f"spec = importlib.util.spec_from_file_location('semretrieval_standalone', {str(_SEMRETRIEVAL)!r})\n"
|
|
"mod = importlib.util.module_from_spec(spec)\n"
|
|
"spec.loader.exec_module(mod)\n"
|
|
"\n"
|
|
"def feat(codes, measure, saving, desc):\n"
|
|
" return types.SimpleNamespace(affected_codes=frozenset(codes), measure_type=measure,\n"
|
|
" claimed_saving_nok=saving, description=desc)\n"
|
|
"\n"
|
|
"query = feat(['05.2'], 'scope_reduction', 220000.0, 'query text')\n"
|
|
"candidates = [\n"
|
|
" types.SimpleNamespace(id='A', proposal_features=feat(['05.2'], 'scope_reduction', 200000.0, 'a')),\n"
|
|
" types.SimpleNamespace(id='B', proposal_features=feat(['09.1'], 'rate_renegotiation', 50000.0, 'b')),\n"
|
|
"]\n"
|
|
"sim = lambda q, c: 1.0 if q.measure_type == c.measure_type else 0.0\n"
|
|
"\n"
|
|
"structural = mod.StructuralRetriever(sim).rank(query, candidates, 2)\n"
|
|
"assert [v.id for v in structural] == ['A', 'B'], structural\n"
|
|
"hybrid = mod.HybridRanker(mod.FakeEmbedder(), sim).rank(query, candidates, 2)\n"
|
|
"assert len(hybrid) == 2, hybrid\n"
|
|
"\n"
|
|
"for name in ('agent_framework', 'mcp', 'portfolio_optimiser.verdicts', 'portfolio_optimiser'):\n"
|
|
" assert name not in sys.modules, name + ' leaked into the semretrieval runtime graph'\n"
|
|
)
|
|
result = subprocess.run([sys.executable, "-c", check], capture_output=True, text=True)
|
|
assert result.returncode == 0, result.stderr
|
|
|
|
|
|
def test_semretrieval_imports_no_network_modules() -> None:
|
|
"""SC8 — the offline default must not be able to reach the network. Detach point: add
|
|
``import urllib.request`` (or any ``_NETWORK_ROOTS`` member) to semretrieval.py → RED."""
|
|
tree = ast.parse(_SEMRETRIEVAL.read_text(encoding="utf-8"))
|
|
imported_roots: set[str] = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
imported_roots.update(alias.name.split(".")[0] for alias in node.names)
|
|
elif isinstance(node, ast.ImportFrom) and node.module:
|
|
imported_roots.add(node.module.split(".")[0])
|
|
|
|
leaked = imported_roots & _NETWORK_ROOTS
|
|
assert not leaked, f"semretrieval imports network module(s): {sorted(leaked)}"
|