bouncer/tests/test_notify.py

"""Tests for push notification module."""

from __future__ import annotations

import time
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from bouncer.config import BouncerConfig, ProxyConfig
from bouncer.notify import Notifier

# -- helpers -----------------------------------------------------------------

def _cfg(**overrides: object) -> BouncerConfig:
    defaults: dict[str, object] = {
        "notify_url": "https://ntfy.sh/bouncer",
        "notify_on_highlight": True,
        "notify_on_privmsg": True,
        "notify_cooldown": 60,
        "notify_proxy": False,
    }
    defaults.update(overrides)
    return BouncerConfig(**defaults)  # type: ignore[arg-type]


def _proxy() -> ProxyConfig:
    return ProxyConfig(host="127.0.0.1", port=1080)


def _notifier(**overrides: object) -> Notifier:
    return Notifier(_cfg(**overrides), _proxy())


# -- enabled -----------------------------------------------------------------

class TestEnabled:
    def test_enabled_with_url(self) -> None:
        n = _notifier(notify_url="https://ntfy.sh/test")
        assert n.enabled is True

    def test_disabled_without_url(self) -> None:
        n = _notifier(notify_url="")
        assert n.enabled is False


# -- should_notify -----------------------------------------------------------

class TestShouldNotify:
    def test_pm_triggers(self) -> None:
        n = _notifier()
        assert n.should_notify("sender", "mynick", "hello", "mynick") is True

    def test_highlight_triggers(self) -> None:
        n = _notifier()
        assert n.should_notify("sender", "#channel", "hey mynick!", "mynick") is True

    def test_highlight_case_insensitive(self) -> None:
        n = _notifier()
        assert n.should_notify("sender", "#channel", "hey MYNICK!", "mynick") is True

    def test_normal_channel_msg_does_not_trigger(self) -> None:
        n = _notifier()
        assert n.should_notify("sender", "#channel", "hello world", "mynick") is False

    def test_disabled_does_not_trigger(self) -> None:
        n = _notifier(notify_url="")
        assert n.should_notify("sender", "mynick", "hello", "mynick") is False

    def test_pm_disabled(self) -> None:
        n = _notifier(notify_on_privmsg=False)
        assert n.should_notify("sender", "mynick", "hello", "mynick") is False

    def test_highlight_disabled(self) -> None:
        n = _notifier(notify_on_highlight=False)
        assert n.should_notify("sender", "#channel", "hey mynick!", "mynick") is False

    def test_cooldown_respected(self) -> None:
        n = _notifier(notify_cooldown=60)
        n._last_sent = time.monotonic()  # just sent
        assert n.should_notify("sender", "mynick", "hello", "mynick") is False

    def test_cooldown_expired(self) -> None:
        n = _notifier(notify_cooldown=60)
        n._last_sent = time.monotonic() - 120  # expired
        assert n.should_notify("sender", "mynick", "hello", "mynick") is True

    def test_channel_prefixes(self) -> None:
        """Targets starting with #, &, +, ! are channels, not PMs."""
        n = _notifier()
        for prefix in ("#", "&", "+", "!"):
            target = f"{prefix}channel"
            assert n.should_notify("sender", target, "hello", "mynick") is False


# -- _is_ntfy ---------------------------------------------------------------

class TestIsNtfy:
    def test_ntfy_sh(self) -> None:
        n = _notifier(notify_url="https://ntfy.sh/mytopic")
        assert n._is_ntfy() is True

    def test_self_hosted_ntfy(self) -> None:
        n = _notifier(notify_url="https://ntfy.example.com/mytopic")
        assert n._is_ntfy() is True

    def test_generic_webhook(self) -> None:
        n = _notifier(notify_url="https://hooks.example.com/webhook")
        assert n._is_ntfy() is False


# -- send --------------------------------------------------------------------

class TestSend:
    @pytest.mark.asyncio
    async def test_ntfy_sends_post(self) -> None:
        n = _notifier(notify_url="https://ntfy.sh/bouncer")

        mock_resp = AsyncMock()
        mock_resp.status = 200
        mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
        mock_resp.__aexit__ = AsyncMock(return_value=False)

        mock_session = AsyncMock()
        mock_session.post = MagicMock(return_value=mock_resp)
        mock_session.__aenter__ = AsyncMock(return_value=mock_session)
        mock_session.__aexit__ = AsyncMock(return_value=False)

        with patch("bouncer.notify.aiohttp.ClientSession", return_value=mock_session):
            await n.send("libera", "user", "#test", "hello world")

        mock_session.post.assert_called_once()
        call_kwargs = mock_session.post.call_args
        assert call_kwargs[1]["data"] == b"hello world"
        assert "Title" in call_kwargs[1]["headers"]
        assert n._last_sent > 0

    @pytest.mark.asyncio
    async def test_webhook_sends_json(self) -> None:
        n = _notifier(notify_url="https://hooks.example.com/webhook")

        mock_resp = AsyncMock()
        mock_resp.status = 200
        mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
        mock_resp.__aexit__ = AsyncMock(return_value=False)

        mock_session = AsyncMock()
        mock_session.post = MagicMock(return_value=mock_resp)
        mock_session.__aenter__ = AsyncMock(return_value=mock_session)
        mock_session.__aexit__ = AsyncMock(return_value=False)

        with patch("bouncer.notify.aiohttp.ClientSession", return_value=mock_session):
            await n.send("libera", "user", "#test", "hello world")

        call_kwargs = mock_session.post.call_args
        assert call_kwargs[1]["json"]["network"] == "libera"
        assert call_kwargs[1]["json"]["sender"] == "user"

    @pytest.mark.asyncio
    async def test_proxy_connector_used(self) -> None:
        n = _notifier(notify_url="https://ntfy.sh/test", notify_proxy=True)

        mock_resp = AsyncMock()
        mock_resp.status = 200
        mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
        mock_resp.__aexit__ = AsyncMock(return_value=False)

        mock_session = AsyncMock()
        mock_session.post = MagicMock(return_value=mock_resp)
        mock_session.__aenter__ = AsyncMock(return_value=mock_session)
        mock_session.__aexit__ = AsyncMock(return_value=False)

        with patch("bouncer.notify.aiohttp.ClientSession", return_value=mock_session):
            with patch("bouncer.notify.Notifier._send_ntfy", new_callable=AsyncMock):
                with patch("aiohttp_socks.ProxyConnector.from_url") as mock_proxy:
                    mock_proxy.return_value = MagicMock()
                    await n.send("libera", "user", "#test", "hello")
                    mock_proxy.assert_called_once_with("socks5://127.0.0.1:1080")

    @pytest.mark.asyncio
    async def test_send_error_does_not_raise(self) -> None:
        """Send errors are logged, not propagated."""
        n = _notifier(notify_url="https://ntfy.sh/test")
        with patch("bouncer.notify.aiohttp.ClientSession", side_effect=Exception("boom")):
            await n.send("libera", "user", "#test", "hello")  # should not raise