Ryan Malloy
5fa1eb36ef
Three-pillar fix from Hamilton review: Code quality — validate_signature() for D-Bus spec compliance, MCDBUS_TIMEOUT env var, replace 13 error-as-success returns with ToolError, monotonic clock deadline on tree walks, sanitize D-Bus error messages, fix resource connection leak via module-level BusManager, hasattr guards in conftest. Elicitation — ctx.elicit() confirmation for system bus call_method and all set_property calls, graceful degradation when client lacks elicitation support, MCDBUS_REQUIRE_ELICITATION for hard-fail mode. Permission docs — four-layer guide (systemd sandboxing, dbus-broker policy, polkit rules, xdg-dbus-proxy) with ready-to-deploy example configs validated against xmllint, bash -n, and systemd-analyze.
84 lines
3.2 KiB
Python
84 lines
3.2 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")
|
|
|
|
|
|
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",
|
|
})
|