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.
293 lines
10 KiB
Python
293 lines
10 KiB
Python
"""Tool-level tests — every MCP tool exercised through Client(mcp) in-memory transport."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from fastmcp import Client
|
|
from fastmcp.exceptions import ToolError
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Discovery tools
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestListServices:
|
|
async def test_returns_services(self, client: Client):
|
|
result = await client.call_tool("list_services", {"bus": "session"})
|
|
text = result.content[0].text
|
|
assert "org.freedesktop.DBus" in text
|
|
assert "D-Bus session bus" in text
|
|
|
|
async def test_include_unique(self, client: Client):
|
|
result = await client.call_tool(
|
|
"list_services", {"bus": "session", "include_unique": True}
|
|
)
|
|
text = result.content[0].text
|
|
assert ":1." in text
|
|
|
|
|
|
class TestIntrospect:
|
|
async def test_dbus_daemon(self, client: Client):
|
|
result = await client.call_tool("introspect", {
|
|
"bus": "session",
|
|
"service": "org.freedesktop.DBus",
|
|
"object_path": "/org/freedesktop/DBus",
|
|
})
|
|
text = result.content[0].text
|
|
assert "org.freedesktop.DBus" in text
|
|
assert "ListNames" in text
|
|
|
|
async def test_invalid_path(self, client: Client):
|
|
with pytest.raises(ToolError, match="Invalid D-Bus object path"):
|
|
await client.call_tool("introspect", {
|
|
"bus": "session",
|
|
"service": "org.freedesktop.DBus",
|
|
"object_path": "not-a-path",
|
|
})
|
|
|
|
|
|
class TestListObjects:
|
|
async def test_dbus_tree(self, client: Client):
|
|
result = await client.call_tool("list_objects", {
|
|
"bus": "session",
|
|
"service": "org.freedesktop.DBus",
|
|
})
|
|
text = result.content[0].text
|
|
assert "Object tree" in text
|
|
assert "/org/freedesktop/DBus" in text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Interaction tools
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCallMethod:
|
|
async def test_ping_void(self, client: Client):
|
|
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()
|
|
|
|
async def test_get_id_returns_hex(self, client: Client):
|
|
result = await client.call_tool("call_method", {
|
|
"bus": "session",
|
|
"service": "org.freedesktop.DBus",
|
|
"object_path": "/org/freedesktop/DBus",
|
|
"interface": "org.freedesktop.DBus",
|
|
"method": "GetId",
|
|
})
|
|
text = result.content[0].text
|
|
# GetId returns a hex string, JSON-encoded with quotes
|
|
bus_id = json.loads(text)
|
|
assert isinstance(bus_id, str)
|
|
assert len(bus_id) > 10
|
|
|
|
async def test_bad_method_raises_tool_error(self, client: Client):
|
|
with pytest.raises(ToolError, match="D-Bus error"):
|
|
await client.call_tool("call_method", {
|
|
"bus": "session",
|
|
"service": "org.freedesktop.DBus",
|
|
"object_path": "/org/freedesktop/DBus",
|
|
"interface": "org.freedesktop.DBus",
|
|
"method": "NoSuchMethod",
|
|
})
|
|
|
|
|
|
class TestGetProperty:
|
|
async def test_read_property(self, client: Client):
|
|
# Read the Features property from the DBus daemon
|
|
result = await client.call_tool("get_property", {
|
|
"bus": "session",
|
|
"service": "org.freedesktop.DBus",
|
|
"object_path": "/org/freedesktop/DBus",
|
|
"interface": "org.freedesktop.DBus",
|
|
"property_name": "Features",
|
|
})
|
|
text = result.content[0].text
|
|
# Features returns an array of strings
|
|
features = json.loads(text)
|
|
assert isinstance(features, list)
|
|
|
|
|
|
class TestGetAllProperties:
|
|
async def test_dbus_daemon(self, client: Client):
|
|
result = await client.call_tool("get_all_properties", {
|
|
"bus": "session",
|
|
"service": "org.freedesktop.DBus",
|
|
"object_path": "/org/freedesktop/DBus",
|
|
"interface": "org.freedesktop.DBus",
|
|
})
|
|
text = result.content[0].text
|
|
assert "Properties of" in text
|
|
assert "Features" in text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSignatureValidation:
|
|
async def test_invalid_signature_rejected(self, client: Client):
|
|
with pytest.raises(ToolError, match="Invalid D-Bus signature"):
|
|
await client.call_tool("call_method", {
|
|
"bus": "session",
|
|
"service": "org.freedesktop.DBus",
|
|
"object_path": "/org/freedesktop/DBus",
|
|
"interface": "org.freedesktop.DBus",
|
|
"method": "GetId",
|
|
"signature": "INVALID!@#",
|
|
"args": '["x"]',
|
|
})
|
|
|
|
async def test_signature_too_long(self, client: Client):
|
|
with pytest.raises(ToolError, match="Signature too long"):
|
|
await client.call_tool("call_method", {
|
|
"bus": "session",
|
|
"service": "org.freedesktop.DBus",
|
|
"object_path": "/org/freedesktop/DBus",
|
|
"interface": "org.freedesktop.DBus",
|
|
"method": "GetId",
|
|
"signature": "s" * 256,
|
|
"args": '["x"]',
|
|
})
|
|
|
|
async def test_malformed_signature_rejected(self, client: Client):
|
|
with pytest.raises(ToolError, match="Malformed D-Bus signature"):
|
|
await client.call_tool("call_method", {
|
|
"bus": "session",
|
|
"service": "org.freedesktop.DBus",
|
|
"object_path": "/org/freedesktop/DBus",
|
|
"interface": "org.freedesktop.DBus",
|
|
"method": "GetId",
|
|
"signature": "((",
|
|
"args": '["x"]',
|
|
})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shortcut tools — existing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSendNotification:
|
|
async def test_sends(self, client: Client):
|
|
result = await client.call_tool("send_notification", {
|
|
"summary": "mcdbus test",
|
|
"body": "automated test notification",
|
|
"timeout": 1000,
|
|
})
|
|
text = result.content[0].text
|
|
assert "Notification sent" in text
|
|
assert "id:" in text
|
|
|
|
|
|
class TestListSystemdUnits:
|
|
async def test_returns_table(self, client: Client):
|
|
result = await client.call_tool("list_systemd_units", {})
|
|
text = result.content[0].text
|
|
assert "Systemd units" in text
|
|
assert "| Unit |" in text
|
|
|
|
async def test_pattern_filter(self, client: Client):
|
|
result = await client.call_tool(
|
|
"list_systemd_units", {"pattern": "dbus*"}
|
|
)
|
|
text = result.content[0].text
|
|
assert "dbus" in text.lower()
|
|
|
|
|
|
class TestMediaPlayerControl:
|
|
async def test_no_players_graceful(self, client: Client):
|
|
# With no explicit player, it auto-discovers; if none found, returns message
|
|
result = await client.call_tool(
|
|
"media_player_control", {"action": "play"}
|
|
)
|
|
text = result.content[0].text
|
|
# Either "No MPRIS media players" or a player response — both are valid
|
|
assert "Player:" in text or "No MPRIS" in text
|
|
|
|
async def test_bad_action(self, client: Client):
|
|
result = await client.call_tool(
|
|
"media_player_control", {"action": "invalid_action"}
|
|
)
|
|
text = result.content[0].text
|
|
# auto-discovers player first; if none found, "No MPRIS" message
|
|
# if player found, "Unknown action" message
|
|
assert "Unknown action" in text or "No MPRIS" in text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shortcut tools — new (service may or may not be available)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestNetworkStatus:
|
|
async def test_returns_status(self, client: Client):
|
|
try:
|
|
result = await client.call_tool("network_status", {})
|
|
except ToolError:
|
|
return # NetworkManager not available on this system
|
|
text = result.content[0].text
|
|
assert "Network Status" in text
|
|
|
|
async def test_has_state(self, client: Client):
|
|
try:
|
|
result = await client.call_tool("network_status", {})
|
|
except ToolError:
|
|
return # NetworkManager not available
|
|
text = result.content[0].text
|
|
assert "State:" in text
|
|
|
|
|
|
class TestBatteryStatus:
|
|
async def test_returns_result(self, client: Client):
|
|
try:
|
|
result = await client.call_tool("battery_status", {})
|
|
except ToolError:
|
|
return # UPower not available on this system
|
|
text = result.content[0].text
|
|
assert (
|
|
"Battery Status" in text
|
|
or "No batteries found" in text
|
|
or "No power devices" in text
|
|
)
|
|
|
|
|
|
class TestBluetoothDevices:
|
|
async def test_returns_result(self, client: Client):
|
|
try:
|
|
result = await client.call_tool("bluetooth_devices", {})
|
|
except ToolError:
|
|
return # bluez not available on this system
|
|
text = result.content[0].text
|
|
assert (
|
|
"Bluetooth Devices" in text
|
|
or "No Bluetooth devices" in text
|
|
or "No Bluetooth objects" in text
|
|
)
|
|
|
|
|
|
class TestKwinWindows:
|
|
async def test_returns_windows(self, client: Client):
|
|
try:
|
|
result = await client.call_tool("kwin_windows", {})
|
|
except ToolError:
|
|
return # KWin not available on this system
|
|
text = result.content[0].text
|
|
assert (
|
|
"KWin Windows" in text
|
|
or "No windows found" in text
|
|
or "No open windows" in text
|
|
)
|
|
|
|
async def test_has_table(self, client: Client):
|
|
try:
|
|
result = await client.call_tool("kwin_windows", {})
|
|
except ToolError:
|
|
return # KWin not available
|
|
text = result.content[0].text
|
|
if "KWin Windows" in text:
|
|
assert "| Window |" in text
|