crypto_trading/backend/tests/test_notify.py

from __future__ import annotations

import asyncio

from app.config import Settings
from app.notify import FanoutNotifier, Notifier, WebhookNotifier, build_notifier
from app.runtime import Runtime


class RecordingNotifier(Notifier):
    def __init__(self, name: str = "rec", fail: bool = False) -> None:
        self.name = name
        self.fail = fail
        self.messages: list[tuple[str, str]] = []

    async def send(self, level: str, message: str) -> None:
        if self.fail:
            raise RuntimeError("boom")
        self.messages.append((level, message))


async def test_fanout_sends_to_all() -> None:
    first = RecordingNotifier("a")
    second = RecordingNotifier("b")
    await FanoutNotifier([first, second]).send("trade", "hello")
    assert first.messages == [("trade", "hello")]
    assert second.messages == [("trade", "hello")]


async def test_fanout_survives_a_failing_channel() -> None:
    bad = RecordingNotifier("bad", fail=True)
    good = RecordingNotifier("good")
    await FanoutNotifier([bad, good]).send("error", "still delivered")
    assert good.messages == [("error", "still delivered")]


def test_build_notifier_none_without_config() -> None:
    assert build_notifier(Settings(telegram_bot_token="", notify_webhook_url="")) is None


def test_build_notifier_telegram_requires_both_fields() -> None:
    assert build_notifier(Settings(telegram_bot_token="t", telegram_chat_id="")) is None
    notifier = build_notifier(Settings(telegram_bot_token="t", telegram_chat_id="c"))
    assert notifier is not None


def test_build_notifier_webhook() -> None:
    notifier = build_notifier(Settings(notify_webhook_url="https://example.com/hook"))
    assert isinstance(notifier, FanoutNotifier)
    assert any(isinstance(n, WebhookNotifier) for n in notifier.notifiers)


async def test_runtime_notifies_on_trade_events() -> None:
    recorder = RecordingNotifier()
    settings = Settings(telegram_bot_token="", notify_webhook_url="")
    runtime = Runtime(settings, notifier=recorder)
    runtime._broadcast({"type": "event", "data": {"level": "trade", "message": "opened long"}})
    await asyncio.gather(*list(runtime._notify_tasks))
    assert recorder.messages == [("trade", "opened long")]


async def test_runtime_ignores_info_events() -> None:
    recorder = RecordingNotifier()
    settings = Settings(telegram_bot_token="", notify_webhook_url="")
    runtime = Runtime(settings, notifier=recorder)
    runtime._broadcast({"type": "event", "data": {"level": "info", "message": "started"}})
    await asyncio.gather(*list(runtime._notify_tasks))
    assert recorder.messages == []


async def test_notify_levels_are_configurable() -> None:
    recorder = RecordingNotifier()
    settings = Settings(telegram_bot_token="", notify_webhook_url="", notify_levels="info,trade")
    runtime = Runtime(settings, notifier=recorder)
    runtime._broadcast({"type": "event", "data": {"level": "info", "message": "started"}})
    await asyncio.gather(*list(runtime._notify_tasks))
    assert recorder.messages == [("info", "started")]


async def test_start_and_stop_notify_lifecycle(settings, store, market) -> None:
    from app.broker.paper import PaperBroker
    from app.schemas import StartManyRequest, StrategySpec

    settings.notify_lifecycle = True
    recorder = RecordingNotifier()
    runtime = Runtime(
        settings,
        store,
        feed_factory=lambda _s: market,
        broker_factory=lambda _s: PaperBroker(fee_bps=0, slippage_bps=0),
        notifier=recorder,
    )
    await runtime.start_many(StartManyRequest(strategies=[StrategySpec(strategy_key="sma_cross")]))
    await runtime.stop()
    await asyncio.gather(*list(runtime._notify_tasks))

    levels = [level for level, _ in recorder.messages]
    assert "info" in levels
    assert any("Started" in msg for _, msg in recorder.messages)
    assert any("Stopped" in msg for _, msg in recorder.messages)
    await runtime.shutdown()


async def test_kill_notifies(settings, store, market) -> None:
    from app.broker.paper import PaperBroker
    from app.schemas import StartManyRequest, StrategySpec

    recorder = RecordingNotifier()
    runtime = Runtime(
        settings,
        store,
        feed_factory=lambda _s: market,
        broker_factory=lambda _s: PaperBroker(fee_bps=0, slippage_bps=0),
        notifier=recorder,
    )
    await runtime.start_many(StartManyRequest(strategies=[StrategySpec(strategy_key="sma_cross")]))
    await runtime.kill("panic")
    await asyncio.gather(*list(runtime._notify_tasks))
    assert any(level == "error" and "KILL SWITCH" in msg for level, msg in recorder.messages)
    await runtime.shutdown()


async def test_summary_message_lists_strategies(settings, store, market) -> None:
    from app.broker.paper import PaperBroker
    from app.schemas import StartManyRequest, StrategySpec

    runtime = Runtime(
        settings,
        store,
        feed_factory=lambda _s: market,
        broker_factory=lambda _s: PaperBroker(fee_bps=0, slippage_bps=0),
        notifier=RecordingNotifier(),
    )
    await runtime.start_many(
        StartManyRequest(
            strategies=[
                StrategySpec(strategy_key="sma_cross", weight=1),
                StrategySpec(strategy_key="ema_cross", weight=3),
            ],
            starting_balance=10000,
        )
    )
    message = runtime.summary_message()
    assert "Portfolio" in message
    assert "SMA Crossover" in message
    assert "EMA Crossover" in message
    assert "75%" in message  # the weighted strategy's allocation
    await runtime.shutdown()


async def test_summary_loop_runs(settings, store, market) -> None:
    from app.broker.paper import PaperBroker

    settings.summary_interval_minutes = 1  # -> 60s; we only check it starts
    recorder = RecordingNotifier()
    runtime = Runtime(
        settings,
        store,
        feed_factory=lambda _s: market,
        broker_factory=lambda _s: PaperBroker(fee_bps=0, slippage_bps=0),
        notifier=recorder,
    )
    runtime._start_summary_loop()
    assert runtime._summary_task is not None
    runtime._stop_summary_loop()
    assert runtime._summary_task is None