173 lines
7.2 KiB
Python
173 lines
7.2 KiB
Python
"""S5.2 — Varsling: load-bearing detach seams for ``notify.py``. Each test goes RED when the
|
|
guarded mechanism is removed; a bare happy-path pass would not catch the regression.
|
|
|
|
Seams pinned here:
|
|
- **MAF-free registration** — the direct-import guard (``test_okf_is_maf_free``) actually scans
|
|
``notify.py`` (otherwise the MAF-free claim is green-but-dead);
|
|
- **transitive import-graph MAF-freedom** — a TYPE_CHECKING-aware static BFS over notify's
|
|
first-party RUNTIME import edges (model: the S51 hitl probe, forbidden roots
|
|
``{agent_framework, mcp}`` ONLY — notify legitimately imports ``urllib`` for the webhook seam).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from portfolio_optimiser.notify import NotifyError, NotifyRefused, WebhookNotifier
|
|
from portfolio_optimiser.verdicts import ProposalFeatures, Verdict, capture_verdict
|
|
|
|
_SRC_DIR = Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser"
|
|
_MAF_ROOTS = {"agent_framework", "mcp"}
|
|
|
|
|
|
def _verdict() -> Verdict:
|
|
return capture_verdict(
|
|
ProposalFeatures(
|
|
affected_codes=frozenset({"05.2"}),
|
|
measure_type="scope_reduction",
|
|
claimed_saving_nok=200_000.0,
|
|
),
|
|
"approved",
|
|
"expert reviewed (test)",
|
|
)
|
|
|
|
|
|
# --- LOAD-BEARING: egress gate (both branches, same url + same recording post) --------------------
|
|
|
|
|
|
def test_webhook_gate_refuses_without_optin_and_posts_with_optin() -> None:
|
|
"""LOAD-BEARING (no-silent-egress): without explicit per-run opt-in the webhook notifier
|
|
refuses at CONSTRUCTION (``NotifyRefused``) and the transport is NEVER touched; with
|
|
``allow_egress=True`` the same url + same transport delivers exactly one post. Detach the
|
|
``if not allow_egress: raise`` gate → the False branch stops raising (silent egress) → RED."""
|
|
url = "https://hooks.example.test/T000/B000/secret-token"
|
|
posts: list[tuple[str, str]] = []
|
|
|
|
def recording_post(post_url: str, body: str) -> None:
|
|
posts.append((post_url, body))
|
|
|
|
with pytest.raises(NotifyRefused):
|
|
WebhookNotifier(url, allow_egress=False, post=recording_post)
|
|
assert posts == [] # gate refused BEFORE any transport access
|
|
|
|
WebhookNotifier(url, allow_egress=True, post=recording_post)(_verdict())
|
|
assert len(posts) == 1
|
|
|
|
|
|
# --- LOAD-BEARING: transport failures never leak the secret-bearing URL ---------------------------
|
|
|
|
|
|
def test_webhook_error_never_leaks_url() -> None:
|
|
"""LOAD-BEARING (secret discipline): a Slack/Teams webhook URL embeds the receiver secret, so
|
|
the wrapped ``NotifyError`` must NOT contain it (STRICTER than ``ingest.py:314``, which may
|
|
include its url). Positive control: the url IS delivered to a successful transport — so the
|
|
negative assertion has teeth. Detach (put the url in the error message) → RED."""
|
|
url = "https://hooks.example.test/T000/B000/secret-token"
|
|
|
|
def failing_post(post_url: str, body: str) -> None:
|
|
raise OSError("connection refused")
|
|
|
|
notifier = WebhookNotifier(url, allow_egress=True, post=failing_post)
|
|
with pytest.raises(NotifyError) as excinfo:
|
|
notifier(_verdict())
|
|
assert url not in str(excinfo.value)
|
|
assert url not in repr(excinfo.value)
|
|
|
|
# Positive control: the same url reaches a working transport verbatim.
|
|
posts: list[tuple[str, str]] = []
|
|
|
|
def recording_post(post_url: str, body: str) -> None:
|
|
posts.append((post_url, body))
|
|
|
|
WebhookNotifier(url, allow_egress=True, post=recording_post)(_verdict())
|
|
assert posts[0][0] == url
|
|
|
|
|
|
def test_notify_registered_maf_free() -> None:
|
|
"""Meta: notify.py is registered in the MAF-free guard list, so ``test_okf_is_maf_free``
|
|
actually scans it — otherwise the MAF-free claim would be green-but-dead (never checked)."""
|
|
from tests.test_okf import _MAF_FREE_MODULES
|
|
|
|
assert "notify.py" in _MAF_FREE_MODULES
|
|
|
|
|
|
def _runtime_import_nodes(tree: ast.Module) -> list[ast.Import | ast.ImportFrom]:
|
|
"""Every ``Import``/``ImportFrom`` node OUTSIDE an ``if TYPE_CHECKING:`` body. notify's
|
|
``Verdict`` import is type-only (TYPE_CHECKING) and must NOT count as a runtime edge —
|
|
the probe walks the runtime graph, exactly what D7 portability is about."""
|
|
skip: set[int] = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.If):
|
|
test = node.test
|
|
if (isinstance(test, ast.Name) and test.id == "TYPE_CHECKING") or (
|
|
isinstance(test, ast.Attribute) and test.attr == "TYPE_CHECKING"
|
|
):
|
|
for child in node.body:
|
|
for sub in ast.walk(child):
|
|
skip.add(id(sub))
|
|
return [
|
|
node
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, (ast.Import, ast.ImportFrom)) and id(node) not in skip
|
|
]
|
|
|
|
|
|
def _first_party_submodule_edges(tree: ast.Module) -> set[str]:
|
|
"""``portfolio_optimiser.<name>`` RUNTIME submodule import edges — modeled on
|
|
``test_hitl_loadbearing._first_party_submodule_edges`` (same package-top/``__init__``
|
|
discount), restricted to non-TYPE_CHECKING nodes via ``_runtime_import_nodes``."""
|
|
edges: set[str] = set()
|
|
for node in _runtime_import_nodes(tree):
|
|
if isinstance(node, ast.ImportFrom):
|
|
parts = (node.module or "").split(".")
|
|
if len(parts) >= 2 and parts[0] == "portfolio_optimiser" and parts[1] != "__init__":
|
|
edges.add(parts[1])
|
|
else:
|
|
for alias in node.names:
|
|
parts = alias.name.split(".")
|
|
if len(parts) >= 2 and parts[0] == "portfolio_optimiser" and parts[1] != "__init__":
|
|
edges.add(parts[1])
|
|
return edges
|
|
|
|
|
|
def _imports_maf(tree: ast.Module) -> bool:
|
|
for node in _runtime_import_nodes(tree):
|
|
if isinstance(node, ast.Import):
|
|
if any(a.name.split(".")[0] in _MAF_ROOTS for a in node.names):
|
|
return True
|
|
elif (node.module or "").split(".")[0] in _MAF_ROOTS:
|
|
return True
|
|
return False
|
|
|
|
|
|
def test_notify_transitive_import_graph_is_maf_free() -> None:
|
|
"""LOAD-BEARING: no module reachable from ``notify.py``'s OWN first-party RUNTIME import edges
|
|
imports ``agent_framework``/``mcp``. Static AST BFS (never traversing ``__init__.py``;
|
|
TYPE_CHECKING edges discounted — the ``Verdict`` annotation is type-only by design). Detach
|
|
point (documented): promote the ``verdicts`` import to runtime (e.g. ``from
|
|
portfolio_optimiser.verdicts import verdict_to_dict``) → the walk reaches ``verdicts.py``
|
|
(``agent_framework`` at :29) → RED. A clean notify → GREEN."""
|
|
seen: set[str] = set()
|
|
queue = list(
|
|
_first_party_submodule_edges(ast.parse((_SRC_DIR / "notify.py").read_text("utf-8")))
|
|
)
|
|
maf_bearing: list[str] = []
|
|
while queue:
|
|
mod = queue.pop()
|
|
if mod in seen:
|
|
continue
|
|
seen.add(mod)
|
|
mod_file = _SRC_DIR / f"{mod}.py"
|
|
if not mod_file.is_file():
|
|
continue
|
|
tree = ast.parse(mod_file.read_text("utf-8"))
|
|
if _imports_maf(tree):
|
|
maf_bearing.append(mod)
|
|
queue.extend(_first_party_submodule_edges(tree) - seen)
|
|
assert maf_bearing == [], (
|
|
f"notify's transitive first-party import graph reaches MAF-bearing module(s): "
|
|
f"{sorted(maf_bearing)}"
|
|
)
|