"""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 import traceback 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: socket-opening CALLS confined to the _urllib_post seam ------------------------- def test_no_socket_call_outside_the_post_seam() -> None: """LOAD-BEARING (socket confinement): (a) ``notify.py`` never imports ``socket`` at all (no raw sockets needed); (b) every CALL to ``urlopen`` / ``Request`` / ``socket.socket`` has ``_urllib_post`` as its nearest-enclosing function — no other function opens a connection. The module-level ``urllib`` IMPORT is ALLOWED (the single sanctioned import, exactly as ``ingest.py`` imports urllib at module scope) — the guard scopes the forbidden thing to socket-opening *calls*, not imports (AST, not substring — the ``test_okf.py:216`` lesson). Detach point: add a ``urlopen(...)`` call to any notifier body → RED.""" tree = ast.parse((_SRC_DIR / "notify.py").read_text("utf-8")) imported: set[str] = set() for node in ast.walk(tree): if isinstance(node, ast.Import): imported |= {a.name.split(".")[0] for a in node.names} elif isinstance(node, ast.ImportFrom): imported.add((node.module or "").split(".")[0]) assert "socket" not in imported, "notify.py must never import socket" offending: list[str] = [] def visit(node: ast.AST, enclosing: str | None) -> None: if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): enclosing = node.name if isinstance(node, ast.Call): fn = node.func name = ( fn.id if isinstance(fn, ast.Name) else fn.attr if isinstance(fn, ast.Attribute) else None ) if name in {"urlopen", "Request", "socket"} and enclosing != "_urllib_post": offending.append(f"{name} called in {enclosing or ''}") for child in ast.iter_child_nodes(node): visit(child, enclosing) visit(tree, None) assert offending == [], ( f"socket-opening calls must live ONLY in _urllib_post, found: {offending}" ) # --- 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). The injected ``OSError`` EMBEDS the url in its message, so the traceback assertion guards the severed cause-chain: restore ``from exc`` → the URL-bearing cause renders via ``traceback.format_exception`` → RED. Positive control: the url IS delivered to a successful transport — so the negative assertion has teeth. Detach (put the url in the error message, or restore ``from exc``) → RED.""" url = "https://hooks.example.test/T000/B000/secret-token" def failing_post(post_url: str, body: str) -> None: raise OSError(f"connection refused for {url}") 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) assert url not in "".join(traceback.format_exception(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_webhook_malformed_url_never_leaks_through_real_transport() -> None: """LOAD-BEARING (secret discipline, real transport): a scheme-less webhook URL makes the DEFAULT ``_urllib_post`` raise a bare ``ValueError`` carrying the URL verbatim (``unknown url type: 'hooks…secret'``) — the raise happens at URL parsing, BEFORE any socket, so the suite stays socket-free. Direct construction bypasses ``NotifierConfig``'s scheme gate, so the catch in ``WebhookNotifier.__call__`` is the actual safety net. Detach points: narrow the catch back to ``(URLError, OSError)`` → the ValueError escapes unscrubbed → RED; restore ``from exc`` → the URL renders via ``__cause__`` in ``traceback.format_exception`` → RED.""" url = "hooks.example.test/T000/B000/secret-token" notifier = WebhookNotifier(url, allow_egress=True) # default _urllib_post, no injection with pytest.raises(NotifyError) as excinfo: notifier(_verdict()) assert "secret-token" not in str(excinfo.value) assert "secret-token" not in repr(excinfo.value) assert "secret-token" not in "".join(traceback.format_exception(excinfo.value)) 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.`` 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)}" )