"""S5.2 — Varsling: unit/contract tests for ``notify.py`` (Notifier protocol + notifiers). Detach-seam (RED-when-detached) twins live in ``tests/test_notify_loadbearing.py``. Tests may import MAF-tainted modules freely (``verdicts``) — only ``notify.py`` itself must stay MAF-free. """ from __future__ import annotations import inspect import io import json import pytest from pydantic import ValidationError from portfolio_optimiser.notify import ( ConsoleNotifier, FileNotifier, Notifier, NotifierConfig, NotifyRefused, WebhookNotifier, _urllib_post, _verdict_payload, build_notifier, ) from portfolio_optimiser.verdicts import ProposalFeatures, Verdict, capture_verdict, verdict_to_dict def _make_post(recorder: list[tuple[str, str]]): """A canned webhook transport: records ``(url, body)``, opens no socket. Local to this file (mirrors ``test_ingest_http._make_get`` locality; never conftest).""" def post(url: str, body: str) -> None: recorder.append((url, body)) return post def _verdict() -> Verdict: return capture_verdict( ProposalFeatures( affected_codes=frozenset({"05.2", "03.1"}), measure_type="scope_reduction", claimed_saving_nok=200_000.0, description="LED retrofit", ), "approved", "feasible within range", ) def test_plain_callable_satisfies_notifier() -> None: """The seam promise: any plain callable satisfies the runtime-checkable ``Notifier`` protocol, so ``run_project(notify=lambda v: ...)`` stays valid — a structural superset of ``Callable[[Verdict], None]``, not a new ABC (needs ``@runtime_checkable``).""" assert isinstance(lambda v: None, Notifier) def test_console_emits_id_and_decision() -> None: """ConsoleNotifier writes the verdict id + decision to the injected stream (never a URL).""" stream = io.StringIO() verdict = _verdict() ConsoleNotifier(stream=stream)(verdict) out = stream.getvalue() assert verdict.id in out assert verdict.decision in out def test_payload_matches_verdict_to_dict_oracle() -> None: """DRIFT-ORACLE: the duck-typed ``_verdict_payload`` reproduces ``verdict_to_dict``'s exact NESTED shape (verdicts.py:124-139) without importing it at runtime — the test imports both (tests may pull MAF freely) and pins them equal, so a ``Verdict`` shape change goes RED here instead of silently diverging the notified payload.""" verdict = _verdict() assert _verdict_payload(verdict) == verdict_to_dict(verdict) def test_file_notifier_byte_deterministic(tmp_path) -> None: """FileNotifier appends one JSONL line per verdict — two writes of the same verdict are byte-identical, LF-terminated, nested per the oracle shape, and carry no wall-clock key.""" path = tmp_path / "sub" / "notify.jsonl" verdict = _verdict() notifier = FileNotifier(str(path)) notifier(verdict) first = path.read_bytes() notifier(verdict) second = path.read_bytes() assert second == first * 2 # identical appended line bytes line = first.decode("utf-8") assert line.endswith("\n") payload = json.loads(line) assert payload["id"] == verdict.id assert payload["proposal_features"]["affected_codes"] == ["03.1", "05.2"] assert not any(k in payload for k in ("date", "timestamp", "created", "ts")) def test_webhook_posts_correct_payload() -> None: """The injected transport receives the exact ``(url, nested verdict_to_dict payload)``.""" calls: list[tuple[str, str]] = [] verdict = _verdict() url = "https://hooks.example.test/T000/B000/token" WebhookNotifier(url, allow_egress=True, post=_make_post(calls))(verdict) assert len(calls) == 1 posted_url, body = calls[0] assert (posted_url, json.loads(body)) == (url, verdict_to_dict(verdict)) def test_webhook_default_post_is_urllib_post() -> None: """inspect.signature proves the stdlib default WITHOUT opening a socket — ``_urllib_post`` is the single socket path; the offline suite never lets it reach a socket (the only in-suite call uses a scheme-less URL that fails at parse time).""" sig = inspect.signature(WebhookNotifier.__init__) assert sig.parameters["post"].default is _urllib_post def test_config_fail_fast() -> None: """Config is fail-fast (mirrors ``GoalContract``): a webhook config without a url refuses at validation; an unknown type refuses at the factory; and the factory alone canNOT grant egress — ``allow_egress`` is a code kwarg (default False → ``NotifyRefused``), never a config field.""" with pytest.raises(ValidationError): NotifierConfig(type="webhook") # url missing → model_validator raises with pytest.raises(ValueError, match="unknown notifier type"): build_notifier(NotifierConfig(type="smoke-signal")) webhook_cfg = NotifierConfig(type="webhook", url="https://hooks.example.test/T0/B0/tok") with pytest.raises(NotifyRefused): build_notifier(webhook_cfg) # allow_egress defaults False → fail-closed notifier = build_notifier(webhook_cfg, allow_egress=True) assert isinstance(notifier, WebhookNotifier) def test_build_notifier_dispatches_console_and_file(tmp_path) -> None: """Factory-glue coverage (the units are tested above): the console and file branches return the right notifier types, and a file config without a path is rejected at CONSTRUCTION (guards ``_required_fields_by_type`` — the ``assert config.path is not None`` in ``build_notifier`` is unreachable by construction). RED only if dispatch/validator break — e.g. the file branch returning ConsoleNotifier flips the isinstance.""" assert isinstance(build_notifier(NotifierConfig(type="console")), ConsoleNotifier) assert isinstance( build_notifier(NotifierConfig(type="file", path=str(tmp_path / "out.jsonl"))), FileNotifier, ) with pytest.raises(ValidationError): NotifierConfig(type="file") # path missing → model_validator raises def test_webhook_config_rejects_schemeless_url() -> None: """Fail-fast scheme gate (S5.2 remediation): a scheme-less webhook url is rejected at config construction — BEFORE it can reach ``_urllib_post``, whose bare ``ValueError`` would carry the secret-bearing URL verbatim. The str/repr absence assertions guard ``hide_input_in_errors``: without that flag pydantic v2 echoes ``input_value='…secret-token'`` in the error display (remove the flag → RED here).""" with pytest.raises(ValidationError) as excinfo: NotifierConfig(type="webhook", url="hooks.example.test/T000/B000/secret-token") assert "secret-token" not in str(excinfo.value) assert "secret-token" not in repr(excinfo.value) # http:// and https:// stay accepted (the gate rejects ONLY scheme-less pastes) NotifierConfig(type="webhook", url="https://hooks.example.test/x") NotifierConfig(type="webhook", url="http://hooks.example.test/x") def test_public_contract_exported() -> None: """The declared B11 contract is public authoring API: importable from the package top and listed in ``__all__`` (the brief-permitted export path — run.py's seam stays byte-intact).""" import portfolio_optimiser exported = ( "Notifier", "ConsoleNotifier", "FileNotifier", "WebhookNotifier", "build_notifier", "NotifierConfig", "NotifyError", "NotifyRefused", ) for name in exported: assert name in portfolio_optimiser.__all__, f"{name} missing from __all__" assert hasattr(portfolio_optimiser, name), f"{name} not importable from package top"