fix(s52): resolve ConsoleNotifier default stream at call time

This commit is contained in:
Kjell Tore Guttormsen 2026-07-17 03:18:00 +02:00
commit 7ec60618b0
2 changed files with 24 additions and 4 deletions

View file

@ -47,13 +47,18 @@ class NotifyRefused(Exception):
class ConsoleNotifier:
"""One-line verdict summary to a text stream (default stdout). Never receives a URL."""
"""One-line verdict summary to a text stream (default stdout). Never receives a URL.
def __init__(self, stream: TextIO = sys.stdout) -> None:
The default stream is resolved at CALL time (``None`` current ``sys.stdout``), so a later
``redirect_stdout``/capture is honored an import-time ``sys.stdout`` default would bind the
original stream object and bypass the redirect."""
def __init__(self, stream: TextIO | None = None) -> None:
self._stream = stream
def __call__(self, verdict: Verdict) -> None:
print(f"[notify] verdict {verdict.id} decision={verdict.decision}", file=self._stream)
stream = self._stream if self._stream is not None else sys.stdout
print(f"[notify] verdict {verdict.id} decision={verdict.decision}", file=stream)
def _verdict_payload(verdict: Verdict) -> dict[str, Any]:

View file

@ -12,12 +12,14 @@ Seams pinned here:
from __future__ import annotations
import ast
import contextlib
import io
import traceback
from pathlib import Path
import pytest
from portfolio_optimiser.notify import NotifyError, NotifyRefused, WebhookNotifier
from portfolio_optimiser.notify import ConsoleNotifier, NotifyError, NotifyRefused, WebhookNotifier
from portfolio_optimiser.verdicts import ProposalFeatures, Verdict, capture_verdict
_SRC_DIR = Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser"
@ -154,6 +156,19 @@ def test_webhook_malformed_url_never_leaks_through_real_transport() -> None:
assert "secret-token" not in "".join(traceback.format_exception(excinfo.value))
def test_console_default_stream_resolves_at_call_time() -> None:
"""LOAD-BEARING (call-time stream resolution): ``ConsoleNotifier()``'s default stream is the
CURRENT ``sys.stdout`` at call time, not the object bound at import time so later
``redirect_stdout``/capture is honored. The notifier is constructed BEFORE the redirect on
purpose. Detach point: rebind the default to import-time ``sys.stdout`` (e.g.
``stream: TextIO = sys.stdout`` in ``__init__``) writes bypass the redirect RED."""
verdict = _verdict()
notifier = ConsoleNotifier() # constructed before the redirect — no explicit stream
with contextlib.redirect_stdout(io.StringIO()) as buf:
notifier(verdict)
assert verdict.id in buf.getvalue()
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)."""