9dc418924b
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>
84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
# 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
|