portfolio-optimiser-claude/tests/test_notify_loadbearing.py

"""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()