mcdbus/tests/test_notify_confirm.py

"""Tests for D-Bus notification confirmation fallback."""

import asyncio
from unittest.mock import MagicMock, patch

import pytest
from dbus_fast import MessageType

from mcdbus._notify_confirm import notify_confirm

NID = 42
IFACE = "org.freedesktop.Notifications"


def _make_signal(interface, member, body):
    """Create a mock D-Bus signal message."""
    msg = MagicMock()
    msg.message_type = MessageType.SIGNAL
    msg.interface = interface
    msg.member = member
    msg.body = body
    return msg


@pytest.fixture
def mock_bus():
    """Mock MessageBus with handler registration tracking."""
    bus = MagicMock()
    bus._handlers = []

    def _add(h):
        bus._handlers.append(h)

    def _remove(h):
        if h in bus._handlers:
            bus._handlers.remove(h)

    bus.add_message_handler = MagicMock(side_effect=_add)
    bus.remove_message_handler = MagicMock(side_effect=_remove)
    return bus


def _fire_signal(mock_bus, interface, member, body, delay=0.01):
    """Schedule a signal delivery after a brief delay."""
    async def _deliver():
        await asyncio.sleep(delay)
        msg = _make_signal(interface, member, body)
        for handler in list(mock_bus._handlers):
            handler(msg)
    asyncio.create_task(_deliver())


class TestNotifyConfirm:
    async def test_approve_returns_true(self, mock_bus):
        async def mock_call(bus, destination, path, interface, member,
                            signature="", body=None, timeout=30.0):
            if member == "Notify":
                _fire_signal(mock_bus, IFACE, "ActionInvoked", [NID, "approve"])
                return [NID]
            return None

        with patch("mcdbus._notify_confirm.call_bus_method", side_effect=mock_call):
            result = await notify_confirm(mock_bus, "Test", "Confirm?")
        assert result is True

    async def test_deny_returns_false(self, mock_bus):
        async def mock_call(bus, destination, path, interface, member,
                            signature="", body=None, timeout=30.0):
            if member == "Notify":
                _fire_signal(mock_bus, IFACE, "ActionInvoked", [NID, "deny"])
                return [NID]
            return None

        with patch("mcdbus._notify_confirm.call_bus_method", side_effect=mock_call):
            result = await notify_confirm(mock_bus, "Test", "Confirm?")
        assert result is False

    async def test_dismissed_returns_false(self, mock_bus):
        async def mock_call(bus, destination, path, interface, member,
                            signature="", body=None, timeout=30.0):
            if member == "Notify":
                _fire_signal(mock_bus, IFACE, "NotificationClosed", [NID, 2])
                return [NID]
            return None

        with patch("mcdbus._notify_confirm.call_bus_method", side_effect=mock_call):
            result = await notify_confirm(mock_bus, "Test", "Confirm?")
        assert result is False

    async def test_timeout_returns_false(self, mock_bus):
        async def mock_call(bus, destination, path, interface, member,
                            signature="", body=None, timeout=30.0):
            if member == "Notify":
                return [NID]
            return None

        with patch("mcdbus._notify_confirm.call_bus_method", side_effect=mock_call):
            result = await notify_confirm(mock_bus, "Test", "Confirm?", timeout=0.05)
        assert result is False

    async def test_service_unavailable_raises(self, mock_bus):
        async def mock_call(bus, destination, path, interface, member,
                            signature="", body=None, timeout=30.0):
            if member == "Notify":
                raise RuntimeError("org.freedesktop.DBus.Error.ServiceUnknown")
            return None

        with patch("mcdbus._notify_confirm.call_bus_method", side_effect=mock_call):
            with pytest.raises(RuntimeError, match="ServiceUnknown"):
                await notify_confirm(mock_bus, "Test", "Confirm?")

    async def test_handler_ignores_wrong_nid(self, mock_bus):
        async def mock_call(bus, destination, path, interface, member,
                            signature="", body=None, timeout=30.0):
            if member == "Notify":
                # Wrong nid first, correct nid after
                _fire_signal(mock_bus, IFACE, "ActionInvoked", [999, "approve"], delay=0.01)
                _fire_signal(mock_bus, IFACE, "ActionInvoked", [NID, "deny"], delay=0.03)
                return [NID]
            return None

        with patch("mcdbus._notify_confirm.call_bus_method", side_effect=mock_call):
            result = await notify_confirm(mock_bus, "Test", "Confirm?", timeout=1.0)
        # Wrong nid was ignored; correct nid with "deny" triggered
        assert result is False

    async def test_cleanup_called(self, mock_bus):
        members_called = []

        async def mock_call(bus, destination, path, interface, member,
                            signature="", body=None, timeout=30.0):
            members_called.append(member)
            if member == "Notify":
                _fire_signal(mock_bus, IFACE, "ActionInvoked", [NID, "approve"])
                return [NID]
            return None

        with patch("mcdbus._notify_confirm.call_bus_method", side_effect=mock_call):
            await notify_confirm(mock_bus, "Test", "Confirm?")

        mock_bus.remove_message_handler.assert_called_once()
        assert "CloseNotification" in members_called
        assert members_called.count("RemoveMatch") == 2