107 lines
4 KiB
Python
107 lines
4 KiB
Python
"""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
|
|
|
|
from portfolio_optimiser.notify import (
|
|
ConsoleNotifier,
|
|
FileNotifier,
|
|
Notifier,
|
|
WebhookNotifier,
|
|
_urllib_post,
|
|
_verdict_payload,
|
|
)
|
|
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 and is never called by the offline suite."""
|
|
sig = inspect.signature(WebhookNotifier.__init__)
|
|
assert sig.parameters["post"].default is _urllib_post
|