mcdbus/tests/test_elicitation.py

"""Tests for elicitation (human-in-the-loop confirmation) on mutation tools."""

import os
from unittest.mock import AsyncMock, patch

import pytest
from fastmcp.exceptions import ToolError
from fastmcp.server.elicitation import (
    AcceptedElicitation,
    CancelledElicitation,
    DeclinedElicitation,
)

from mcdbus._interaction import _confirm_or_abort


class TestConfirmOrAbort:
    """Unit tests for the _confirm_or_abort helper."""

    async def test_accepted_proceeds(self):
        ctx = AsyncMock()
        ctx.elicit = AsyncMock(return_value=AcceptedElicitation(data="confirm"))
        # Should return without raising
        await _confirm_or_abort(ctx, "Test message", "test_op")

    async def test_declined_raises_tool_error(self):
        ctx = AsyncMock()
        ctx.elicit = AsyncMock(return_value=DeclinedElicitation())
        with pytest.raises(ToolError, match="cancelled by user"):
            await _confirm_or_abort(ctx, "Test message", "test_op")

    async def test_accepted_wrong_key_raises(self):
        ctx = AsyncMock()
        ctx.elicit = AsyncMock(return_value=AcceptedElicitation(data="deny"))
        with pytest.raises(ToolError, match="cancelled by user"):
            await _confirm_or_abort(ctx, "Test message", "test_op")

    async def test_cancelled_proceeds_by_default(self):
        ctx = AsyncMock()
        ctx.elicit = AsyncMock(return_value=CancelledElicitation())
        # Client doesn't support elicitation — should proceed silently
        await _confirm_or_abort(ctx, "Test message", "test_op")

    async def test_cancelled_hard_fails_when_required(self):
        ctx = AsyncMock()
        ctx.elicit = AsyncMock(return_value=CancelledElicitation())
        with patch.dict(os.environ, {"MCDBUS_REQUIRE_ELICITATION": "1"}):
            with pytest.raises(ToolError, match="Elicitation required"):
                await _confirm_or_abort(ctx, "Test message", "test_op")

    async def test_elicit_exception_treated_as_cancelled(self):
        """When ctx.elicit() throws (client lacks protocol support), proceed."""
        ctx = AsyncMock()
        ctx.elicit = AsyncMock(side_effect=Exception("Method not found"))
        # Should not raise — exception is treated as CancelledElicitation
        await _confirm_or_abort(ctx, "Test message", "test_op")

    async def test_elicit_exception_hard_fails_when_required(self):
        """When ctx.elicit() throws and MCDBUS_REQUIRE_ELICITATION is set, fail."""
        ctx = AsyncMock()
        ctx.elicit = AsyncMock(side_effect=Exception("Method not found"))
        with patch.dict(os.environ, {"MCDBUS_REQUIRE_ELICITATION": "1"}):
            with pytest.raises(ToolError, match="Elicitation required"):
                await _confirm_or_abort(ctx, "Test message", "test_op")

    async def test_cancelled_notify_fallback_approved(self):
        """MCDBUS_NOTIFY_CONFIRM=1 + user approves → proceeds normally."""
        ctx = AsyncMock()
        ctx.elicit = AsyncMock(return_value=CancelledElicitation())
        mock_mgr = AsyncMock()
        mock_mgr.get_bus = AsyncMock(return_value=AsyncMock())
        with (
            patch.dict(os.environ, {"MCDBUS_NOTIFY_CONFIRM": "1"}),
            patch("mcdbus._interaction.get_mgr", return_value=mock_mgr),
            patch("mcdbus._interaction.notify_confirm", return_value=True),
        ):
            await _confirm_or_abort(ctx, "Test message", "test_op")

    async def test_cancelled_notify_fallback_denied(self):
        """MCDBUS_NOTIFY_CONFIRM=1 + user denies → ToolError."""
        ctx = AsyncMock()
        ctx.elicit = AsyncMock(return_value=CancelledElicitation())
        mock_mgr = AsyncMock()
        mock_mgr.get_bus = AsyncMock(return_value=AsyncMock())
        with (
            patch.dict(os.environ, {"MCDBUS_NOTIFY_CONFIRM": "1"}),
            patch("mcdbus._interaction.get_mgr", return_value=mock_mgr),
            patch("mcdbus._interaction.notify_confirm", return_value=False),
        ):
            with pytest.raises(ToolError, match="denied via desktop notification"):
                await _confirm_or_abort(ctx, "Test message", "test_op")

    async def test_cancelled_notify_unavailable_proceeds(self):
        """Notification service failure → falls through with warning."""
        ctx = AsyncMock()
        ctx.elicit = AsyncMock(return_value=CancelledElicitation())
        mock_mgr = AsyncMock()
        mock_mgr.get_bus = AsyncMock(return_value=AsyncMock())
        with (
            patch.dict(os.environ, {"MCDBUS_NOTIFY_CONFIRM": "1"}),
            patch("mcdbus._interaction.get_mgr", return_value=mock_mgr),
            patch(
                "mcdbus._interaction.notify_confirm",
                side_effect=RuntimeError("No notification service"),
            ),
        ):
            # Should proceed without raising
            await _confirm_or_abort(ctx, "Test message", "test_op")

    async def test_require_elicitation_overrides_notify(self):
        """MCDBUS_REQUIRE_ELICITATION takes precedence over MCDBUS_NOTIFY_CONFIRM."""
        ctx = AsyncMock()
        ctx.elicit = AsyncMock(return_value=CancelledElicitation())
        with (
            patch.dict(os.environ, {
                "MCDBUS_REQUIRE_ELICITATION": "1",
                "MCDBUS_NOTIFY_CONFIRM": "1",
            }),
            patch("mcdbus._interaction.notify_confirm") as mock_notify,
        ):
            with pytest.raises(ToolError, match="Elicitation required"):
                await _confirm_or_abort(ctx, "Test message", "test_op")
            mock_notify.assert_not_called()

    async def test_notify_not_attempted_without_env(self):
        """Without MCDBUS_NOTIFY_CONFIRM, notify_confirm is never called."""
        ctx = AsyncMock()
        ctx.elicit = AsyncMock(return_value=CancelledElicitation())
        with patch("mcdbus._interaction.notify_confirm") as mock_notify:
            await _confirm_or_abort(ctx, "Test message", "test_op")
            mock_notify.assert_not_called()


class TestCallMethodElicitation:
    """Verify call_method triggers elicitation on system bus but not session bus."""

    async def test_session_bus_no_elicitation(self, client):
        """Session bus calls should not trigger elicitation."""
        # Ping on session bus — should succeed without any elicitation
        result = await client.call_tool("call_method", {
            "bus": "session",
            "service": "org.freedesktop.DBus",
            "object_path": "/org/freedesktop/DBus",
            "interface": "org.freedesktop.DBus.Peer",
            "method": "Ping",
        })
        text = result.content[0].text
        assert "void" in text.lower()


class TestSetPropertyElicitation:
    """Verify set_property always triggers elicitation."""

    async def test_invalid_property_still_validates_first(self, client):
        """Validation errors should fire before elicitation."""
        with pytest.raises(ToolError, match="Invalid D-Bus"):
            await client.call_tool("set_property", {
                "bus": "session",
                "service": "not-valid",
                "object_path": "/foo",
                "interface": "org.test.Iface",
                "property_name": "Foo",
                "value": '"bar"',
                "signature": "s",
            })