feat(stdio): harden local MCP transport and add end-to-end tests

Reserve stdout for the JSON-RPC stream: _configure_stderr_logging() pins all
logging to stderr (and rewrites any stray stdout handler) so a log line can
never corrupt the stdio protocol. Extract a pure, testable build_server() from
_serve(). Add end-to-end tests over the mcp in-memory transport (initialize +
tools/list + tools/call), covering a successful round trip and a policy denial
surfaced as an MCP error.
This commit is contained in:
2026-06-27 15:19:42 +02:00
parent f660fa32c1
commit 5d4a98d06e
2 changed files with 117 additions and 13 deletions
+70
View File
@@ -73,6 +73,19 @@ def test_default_audit_log_path_is_user_scoped() -> None:
assert "aegis-gitea-mcp" in str(path)
def test_configure_stderr_logging_keeps_stdout_clean(stdio_env: None) -> None:
"""No log handler may target stdout (it is the JSON-RPC channel)."""
import logging
import sys
# Simulate a library that attached a stdout handler.
root = logging.getLogger()
root.addHandler(logging.StreamHandler(sys.stdout))
stdio_app._configure_stderr_logging()
for handler in logging.getLogger().handlers:
assert getattr(handler, "stream", sys.stderr) is not sys.stdout
def test_main_exits_when_env_missing(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("GITEA_URL", raising=False)
monkeypatch.delenv("GITEA_TOKEN", raising=False)
@@ -138,3 +151,60 @@ async def test_dispatch_pins_owner_login_and_returns(
assert result["name"] == "app"
# The dispatch pinned the trusted PAT owner onto the request context.
assert get_gitea_user_login() == "alice"
# --- End-to-end over the MCP protocol (in-memory transport) -----------------
async def test_build_server_exposes_registry_tools(stdio_env: None) -> None:
"""build_server() wires every registry tool, including gitea_request."""
from mcp.shared.memory import create_connected_server_and_client_session
server = stdio_app.build_server()
async with create_connected_server_and_client_session(server) as client:
await client.initialize()
listed = await client.list_tools()
names = {t.name for t in listed.tools}
assert "get_repository_info" in names
assert "gitea_request" in names
assert len(names) >= 40
async def test_stdio_tool_call_round_trip(stdio_env: None, monkeypatch: pytest.MonkeyPatch) -> None:
"""A tools/call over the MCP protocol dispatches through the shared core."""
from mcp.shared.memory import create_connected_server_and_client_session
monkeypatch.setattr(stdio_app, "_owner_login", "alice")
patcher = _patch_gitea_client(
get_repository={"owner": {"login": "acme"}, "name": "app", "full_name": "acme/app"}
)
try:
server = stdio_app.build_server()
async with create_connected_server_and_client_session(server) as client:
await client.initialize()
result = await client.call_tool("get_repository_info", {"owner": "acme", "repo": "app"})
finally:
patcher.stop()
assert result.isError is False
text = "".join(getattr(block, "text", "") for block in result.content)
assert "app" in text
async def test_stdio_tool_call_policy_denial_is_reported(
stdio_env: None, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A write tool denied by WRITE_MODE surfaces as an MCP error, not a crash."""
from mcp.shared.memory import create_connected_server_and_client_session
monkeypatch.setattr(stdio_app, "_owner_login", "alice")
server = stdio_app.build_server()
async with create_connected_server_and_client_session(server) as client:
await client.initialize()
result = await client.call_tool(
"create_issue", {"owner": "acme", "repo": "app", "title": "x"}
)
assert result.isError is True
text = "".join(getattr(block, "text", "") for block in result.content)
assert "write mode is disabled" in text