Ryan Malloy
89419a36c2
ctx.elicit() throws an exception (not CancelledElicitation) when the MCP client doesn't implement the elicitation/create JSON-RPC method. Wrap the call in try/except to treat protocol-level rejection the same as CancelledElicitation. Found during live testing with Claude Code CLI which doesn't support MCP elicitation yet.
99 lines
4.0 KiB
Python
99 lines
4.0 KiB
Python
"""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")
|
|
|
|
|
|
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",
|
|
})
|