test(mcp): integration tests for pluggable lifespan wiring
Adds tests/integration/test_main_lifespan.py with 5 cases that exercise the end-to-end main.app lifespan: - default (no MCP_SERVICES) mounts both services in BUILTIN order - MCP_SERVICES=files mounts only files - MCP_SERVICES='' mounts zero - MCP_SERVICES=bogus raises RuntimeError at startup - /health on the root app is unaffected by MCP filtering Tests use TestClient and mock main.init_mcp_server to record mount calls, since fastapi-mcp's HTTP endpoints 404 for bare GETs regardless of mount state (the MCP transport requires an initialize handshake before any tool call). Mocking avoids that coupling. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,83 @@
|
||||
# coding=utf-8
|
||||
"""
|
||||
@Time :2026/6/30
|
||||
@Author :tao.chen
|
||||
|
||||
End-to-end tests for main.app's lifespan: verify that MCP_SERVICES env
|
||||
var correctly gates which services get mounted. Mocks main.init_mcp_server
|
||||
to record mount calls rather than relying on fastapi-mcp's HTTP
|
||||
response codes (which 404 for bare GETs regardless of mount state
|
||||
because the MCP transport requires an initialize handshake first).
|
||||
"""
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
def _get_main_app():
|
||||
"""Lazily import main.app so the conftest autouse fixture has run
|
||||
before files_mcp is first imported (via discover_and_filter inside
|
||||
the lifespan)."""
|
||||
import main
|
||||
return main.app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_mount(monkeypatch):
|
||||
"""Replace main.init_mcp_server with a recorder.
|
||||
|
||||
Returns a list that gets appended with mount_path each time a
|
||||
service is mounted. The lifespan inside the TestClient context
|
||||
manager triggers all mounts during __enter__.
|
||||
"""
|
||||
mounted: list[str] = []
|
||||
|
||||
def fake_init(sub_app):
|
||||
class _FakeMcp:
|
||||
def __init__(self, app):
|
||||
self.app = app
|
||||
self.tools = []
|
||||
def mount_http(self, parent, mount_path):
|
||||
mounted.append(mount_path)
|
||||
return _FakeMcp(sub_app)
|
||||
|
||||
monkeypatch.setattr("main.init_mcp_server", fake_init)
|
||||
return mounted
|
||||
|
||||
|
||||
class TestLifespanMounting:
|
||||
def test_default_mounts_both_services(self, monkeypatch, mock_mount):
|
||||
monkeypatch.delenv("MCP_SERVICES", raising=False)
|
||||
with TestClient(_get_main_app()):
|
||||
pass
|
||||
# Both mount paths are present, in BUILTIN_SERVICES order.
|
||||
assert mock_mount == ["/spark-executor-mcp", "/files-mcp"]
|
||||
|
||||
def test_MCP_SERVICES_files_only_mounts_one(self, monkeypatch, mock_mount):
|
||||
monkeypatch.setenv("MCP_SERVICES", "files")
|
||||
with TestClient(_get_main_app()):
|
||||
pass
|
||||
assert mock_mount == ["/files-mcp"]
|
||||
|
||||
def test_MCP_SERVICES_empty_mounts_none(self, monkeypatch, mock_mount):
|
||||
monkeypatch.setenv("MCP_SERVICES", "")
|
||||
with TestClient(_get_main_app()):
|
||||
pass
|
||||
assert mock_mount == []
|
||||
|
||||
def test_MCP_SERVICES_unknown_raises_at_startup(self, monkeypatch, mock_mount):
|
||||
monkeypatch.setenv("MCP_SERVICES", "bogus")
|
||||
with pytest.raises(RuntimeError, match="unknown service names"):
|
||||
with TestClient(_get_main_app()):
|
||||
pass
|
||||
# No mounts attempted
|
||||
assert mock_mount == []
|
||||
|
||||
def test_health_endpoint_unaffected_by_mcp_filtering(self, monkeypatch):
|
||||
# /health is on the root app, independent of MCP service mounting.
|
||||
monkeypatch.delenv("MCP_SERVICES", raising=False)
|
||||
with TestClient(_get_main_app()) as c:
|
||||
assert c.get("/health").status_code == 200
|
||||
|
||||
monkeypatch.setenv("MCP_SERVICES", "")
|
||||
with TestClient(_get_main_app()) as c:
|
||||
assert c.get("/health").status_code == 200
|
||||
Reference in New Issue
Block a user