"""Notification delivery — LOAD-BEARING (S5.2-analog; §8; §11; K10). The seam this file keeps alive: a run/HITL pass can DELIVER a notification (console, file, or webhook) WITHOUT breaking the no-silent-egress invariant. A webhook — the one transport that leaves the machine — fires ONLY behind an explicit per-run opt-in flag (mirroring ingest-spec §8: "the flag is a run argument, never a config field — the config cannot grant itself network access"). The transport is INJECTED: the suite passes a canned transport, so no socket is ever opened; the real transport lives behind one clearly-marked seam function that the suite never calls. Three detached seams proven RED here: * Opt-in gate: build a webhook notifier WITHOUT the opt-in flag → it MUST refuse fail-fast (``EgressNotPermitted``) and its transport MUST stay unfired. Detach point: drop the ``if not egress_opt_in: raise`` guard → the webhook constructs and its transport fires on ``notify`` → RED. * Payload structure: the canned transport receives the STRUCTURED event (``event`` / ``summary`` / ``fields`` as deterministic JSON), not a prose blob. Detach point: change the encoded shape → RED. * No socket outside the seam: an AST grep-guard proves ``notify.py`` carries no network import ANYWHERE except inside ``default_webhook_transport`` (the one injectable seam). Detach point: hoist the ``urllib`` import to module scope (or add any ``socket``/``httpx`` path elsewhere) → RED. Key assumption (stack-local, no shared spec): the payload shape is this stack's own — divergence from the MAF sibling is accepted and documented in the module docstring. Pinned here by asserting the exact structured payload. """ from __future__ import annotations import ast import json from pathlib import Path import pytest from portfolio_optimiser_claude.notify import ( ConsoleNotifier, EgressNotPermitted, FileNotifier, Notification, NotifyConfig, WebhookNotifier, build_notifiers, emit, notify_config_from_args, ) SRC_PKG = Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser_claude" _NETWORK_MODULES = {"socket", "urllib", "http", "https", "requests", "httpx", "anthropic", "ssl"} _SEAM_FUNC = "default_webhook_transport" class _CannedTransport: """A canned webhook transport — records calls, opens NO socket (§11 suite).""" def __init__(self) -> None: self.calls: list[tuple[str, bytes]] = [] def send(self, url: str, payload: bytes) -> None: self.calls.append((url, payload)) def _event() -> Notification: return Notification( event="run.completed", summary="run r-001: validated", fields={"run_id": "r-001", "decision": "validated", "attempts": 1}, ) # --- the opt-in gate (LOAD-BEARING: webhook egress is opt-in per run) ------------------------ class TestWebhookOptInGate: """§8 mirror: the webhook refuses fail-fast WITHOUT the per-run opt-in flag.""" def test_webhook_without_optin_refuses_and_never_fires(self) -> None: # LOAD-BEARING (the opt-in seam). Detach point: drop the guard in # WebhookNotifier.__init__ → construction succeeds → transport fires on # notify → this RED. Restore from the implemented copy, never git checkout. canned = _CannedTransport() with pytest.raises(EgressNotPermitted): WebhookNotifier("https://hooks.example/x", canned.send, egress_opt_in=False) assert canned.calls == [] # nothing left the machine def test_build_notifiers_refuses_webhook_without_optin(self) -> None: # The run-argument seam: the CLI flag threads the opt-in into build; a # webhook_url without the flag is refused BEFORE any transport exists. canned = _CannedTransport() with pytest.raises(EgressNotPermitted): build_notifiers( NotifyConfig(webhook_url="https://hooks.example/x", webhook_egress_opt_in=False), webhook_transport=canned.send, ) assert canned.calls == [] def test_optin_webhook_constructs_and_fires(self) -> None: canned = _CannedTransport() notifier = WebhookNotifier("https://hooks.example/x", canned.send, egress_opt_in=True) notifier.notify(_event()) assert len(canned.calls) == 1 # --- the payload structure (LOAD-BEARING: structured, deterministic) ------------------------- class TestPayloadStructure: """The canned transport receives the structured event, not a prose blob.""" def test_canned_transport_receives_structured_payload(self) -> None: # LOAD-BEARING (payload shape, stack-local). Detach point: change the # encoded shape (e.g. drop ``fields`` or emit prose) → RED. canned = _CannedTransport() notifiers = build_notifiers( NotifyConfig(webhook_url="https://hooks.example/x", webhook_egress_opt_in=True), webhook_transport=canned.send, ) emit(notifiers, _event()) assert len(canned.calls) == 1 url, payload = canned.calls[0] assert url == "https://hooks.example/x" assert json.loads(payload) == { "event": "run.completed", "summary": "run r-001: validated", "fields": {"run_id": "r-001", "decision": "validated", "attempts": 1}, } def test_payload_is_byte_deterministic(self) -> None: # Same event → identical bytes (sorted keys) — a notification is not a # source of run-to-run drift. a, b = _CannedTransport(), _CannedTransport() WebhookNotifier("https://x", a.send, egress_opt_in=True).notify(_event()) WebhookNotifier("https://x", b.send, egress_opt_in=True).notify(_event()) assert a.calls[0][1] == b.calls[0][1] # --- no socket outside the injectable seam (LOAD-BEARING: AST grep-guard) -------------------- def _import_nodes(tree: ast.AST) -> list[ast.stmt]: return [n for n in ast.walk(tree) if isinstance(n, (ast.Import, ast.ImportFrom))] def _modules_of(node: ast.stmt) -> set[str]: if isinstance(node, ast.Import): return {alias.name.split(".")[0] for alias in node.names} if isinstance(node, ast.ImportFrom) and node.module: return {node.module.split(".")[0]} return set() class TestNoSocketOutsideSeam: """LOAD-BEARING (§11): the ONLY network path in notify.py is the seam function.""" def test_no_network_import_outside_the_seam(self) -> None: # Detach point: hoist the urllib import out of default_webhook_transport # to module scope, or add any socket/httpx path elsewhere → an import # node OUTSIDE the seam references a network module → RED. tree = ast.parse((SRC_PKG / "notify.py").read_text("utf-8")) seam = next( n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == _SEAM_FUNC ) seam_node_ids = {id(n) for n in _import_nodes(seam)} # Positive control: the seam DOES import a network module, and this very # machinery detects it. Without this, `outside == set()` would hold just # as well if `_import_nodes` returned nothing, `_modules_of` resolved no # names, or `_NETWORK_MODULES` were empty — i.e. if the detector were # incapable of ever flagging anything. It proves the matcher, not just # the absence. inside: set[str] = set() for node in _import_nodes(seam): inside |= _modules_of(node) & _NETWORK_MODULES assert inside, "the seam should carry the ONLY network import; detector matched none" outside: set[str] = set() for node in _import_nodes(tree): if id(node) not in seam_node_ids: outside |= _modules_of(node) & _NETWORK_MODULES assert outside == set() def test_the_seam_is_real_not_hollow(self) -> None: # Green-but-dead guard: the seam MUST actually carry the network import # it isolates — deleting it (so the guard passes vacuously) is RED here. tree = ast.parse((SRC_PKG / "notify.py").read_text("utf-8")) seam = next( n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == _SEAM_FUNC ) seam_modules: set[str] = set() for node in _import_nodes(seam): seam_modules |= _modules_of(node) assert "urllib" in seam_modules # --- the local notifiers (console, file — no network) ---------------------------------------- class TestLocalNotifiers: """console and file notifiers deliver locally — never a socket.""" def test_console_notifier_prints_event_and_summary( self, capsys: pytest.CaptureFixture[str] ) -> None: ConsoleNotifier().notify(_event()) out = capsys.readouterr().out assert "run.completed" in out assert "run r-001: validated" in out def test_file_notifier_writes_deterministic_json(self, tmp_path: Path) -> None: FileNotifier(tmp_path).notify(_event()) written = tmp_path / "run.completed.json" assert written.is_file() body = json.loads(written.read_text("utf-8")) assert body["event"] == "run.completed" assert body["fields"]["run_id"] == "r-001" # deterministic house JSON: sorted keys, trailing newline assert written.read_text("utf-8").endswith("}\n") def test_build_notifiers_composes_the_configured_set(self, tmp_path: Path) -> None: notifiers = build_notifiers(NotifyConfig(console=True, file_dir=tmp_path)) assert [type(n).__name__ for n in notifiers] == ["ConsoleNotifier", "FileNotifier"] def test_no_config_builds_nothing(self) -> None: assert build_notifiers(NotifyConfig()) == [] # --- the CLI-arg mapping (shared by run.py and hitl.py) -------------------------------------- class TestNotifyConfigFromArgs: """notify_config_from_args maps the shared CLI flags into a NotifyConfig.""" def test_flags_map_into_config(self, tmp_path: Path) -> None: import argparse from portfolio_optimiser_claude.notify import add_notify_args parser = argparse.ArgumentParser() add_notify_args(parser) args = parser.parse_args( [ "--notify-console", "--notify-file", str(tmp_path), "--notify-webhook", "https://x", "--allow-webhook-egress", ] ) cfg = notify_config_from_args(args) assert cfg == NotifyConfig( console=True, file_dir=tmp_path, webhook_url="https://x", webhook_egress_opt_in=True, ) def test_defaults_are_all_off(self) -> None: import argparse from portfolio_optimiser_claude.notify import add_notify_args parser = argparse.ArgumentParser() add_notify_args(parser) cfg = notify_config_from_args(parser.parse_args([])) assert cfg == NotifyConfig()