6krrt/tests/test_notifier.py

"""Unit tests for the desktop notification channel (notifier.py).

All tests stub ``subprocess.run`` — no real ``notify-send`` calls.
"""

from __future__ import annotations

import logging
import time
from types import SimpleNamespace
from unittest import mock

import pytest

from config import NotificationsConfig
from notifier import AlertEvent, Notifier

# ── fixtures ──────────────────────────────────────────────────────────


@pytest.fixture
def notifier() -> Notifier:
    """Default notifier with one desktop channel (min_severity=warning)."""
    cfg = NotificationsConfig()
    return Notifier(cfg)


@pytest.fixture
def event() -> AlertEvent:
    """A sample critical trigger event."""
    return AlertEvent(
        dedup_key="test-session-1",
        severity="critical",
        state="trigger",
        title="Agent stuck",
        summary="No progress for 30 minutes",
    )


# ── tests ─────────────────────────────────────────────────────────────


class TestDesktopNotify:
    """Tests for desktop notification delivery via notify-send."""

    def test_deliver_critical(self, notifier: Notifier, event: AlertEvent) -> None:
        """A critical event fires notify-send with -u critical."""
        with mock.patch("notifier.subprocess.run") as mock_run:
            notifier.deliver(event)

        mock_run.assert_called_once_with(
            ["notify-send", "-u", "critical", "Agent stuck",
             "No progress for 30 minutes"],
            check=False,
            timeout=5,
        )

    def test_deliver_skips_below_min_severity(
        self, notifier: Notifier
    ) -> None:
        """An info event is skipped when channel min_severity is warning."""
        info_event = AlertEvent(
            dedup_key="info-test",
            severity="info",
            state="trigger",
            title="Trivial",
            summary="Nothing to see",
        )
        with mock.patch("notifier.subprocess.run") as mock_run:
            notifier.deliver(info_event)

        mock_run.assert_not_called()

    def test_deliver_rate_limited(self, notifier: Notifier) -> None:
        """A second delivery within the rate window is suppressed.

        We deliver once, record the timestamp as if the caller did, then
        deliver again — the second should be rate-limited.
        """
        store: dict[str, float] = {}
        # Notifier that reads from our store but does not write to it
        rate_notifier = Notifier(
            NotificationsConfig(),
            last_fired=lambda key: store.get(key),
        )

        ev = AlertEvent(
            dedup_key="dup",
            severity="warning",
            state="trigger",
            title="Dup",
            summary="Duplicate check",
        )

        with mock.patch("notifier.subprocess.run") as mock_run:
            # First delivery — no last_fired entry, should fire
            rate_notifier.deliver(ev)
            # Simulate the caller recording the fired timestamp
            store["default"] = time.monotonic()
            # Second delivery — within rate window, should be suppressed
            rate_notifier.deliver(ev)

        assert mock_run.call_count == 1

    def test_deliver_resolve_not_rate_limited(
        self, notifier: Notifier
    ) -> None:
        """A resolve event bypasses the rate limit.

        Deliver trigger, record timestamp, deliver resolve — both fire.
        """
        store: dict[str, float] = {}
        rate_notifier = Notifier(
            NotificationsConfig(),
            last_fired=lambda key: store.get(key),
        )

        trigger_event = AlertEvent(
            dedup_key="test-alert",
            severity="critical",
            state="trigger",
            title="Something wrong",
            summary="Critical issue detected",
        )
        resolve_event = AlertEvent(
            dedup_key="test-alert",
            severity="critical",
            state="resolve",
            title="Something wrong",
            summary="Issue resolved",
        )

        with mock.patch("notifier.subprocess.run") as mock_run:
            rate_notifier.deliver(trigger_event)
            store["default"] = time.monotonic()  # simulate caller recording
            rate_notifier.deliver(resolve_event)

        assert mock_run.call_count == 2

    def test_deliver_failure_logged(
        self, caplog: pytest.LogCaptureFixture, notifier: Notifier,
        event: AlertEvent,
    ) -> None:
        """FileNotFoundError from notify-send is caught and logged."""
        caplog.set_level(logging.WARNING, logger="notifier")

        with mock.patch(
            "notifier.subprocess.run", side_effect=FileNotFoundError
        ):
            notifier.deliver(event)

        assert len(caplog.records) >= 1
        assert "notify-send not installed" in caplog.text
        assert "test-session-1" in caplog.text


class TestUnknownChannel:
    """Tests for handling of unknown channel types."""

    def test_unknown_channel_type_skips(
        self, caplog: pytest.LogCaptureFixture, notifier: Notifier,
        event: AlertEvent,
    ) -> None:
        """An unknown channel type (not 'desktop') logs a warning and skips."""
        caplog.set_level(logging.WARNING, logger="notifier")

        fake_channel = SimpleNamespace(type="unknown", name="test-channel")
        notifier._dispatch(fake_channel, event)

        assert any(
            "unknown channel type" in r.getMessage() for r in caplog.records
        )
        assert "test-channel" in caplog.text