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.
53 lines
1.4 KiB
Python
53 lines
1.4 KiB
Python
"""Test fixtures for mcdbus."""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
from fastmcp import Client
|
|
|
|
from mcdbus._bus import BusManager
|
|
from mcdbus.server import mcp
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_mcp_singleton():
|
|
"""Reset event-loop-bound state on the mcp singleton.
|
|
|
|
FastMCP creates an asyncio.Event() at construction time. pytest-asyncio
|
|
creates a fresh loop per test, so after test 1 the Event is bound to a
|
|
dead loop and test 2 would crash. Re-creating these objects keeps every
|
|
test isolated.
|
|
"""
|
|
_PRIVATE_ATTRS = ("_started", "_lifespan_result", "_lifespan_result_set")
|
|
for attr in _PRIVATE_ATTRS:
|
|
if not hasattr(mcp, attr):
|
|
raise AttributeError(
|
|
f"FastMCP removed {attr!r} — fixture needs update "
|
|
f"(fastmcp version may have changed)"
|
|
)
|
|
mcp._started = asyncio.Event()
|
|
mcp._lifespan_result = None
|
|
mcp._lifespan_result_set = False
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
"""MCP client connected via in-memory transport (no HTTP)."""
|
|
async with Client(mcp) as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture
|
|
async def bus_manager():
|
|
"""Provide a BusManager instance, cleaned up after test."""
|
|
mgr = BusManager()
|
|
yield mgr
|
|
await mgr.disconnect_all()
|
|
|
|
|
|
@pytest.fixture
|
|
async def session_bus(bus_manager: BusManager):
|
|
"""Provide a connected session bus."""
|
|
return await bus_manager.get_bus("session")
|