diff --git a/tests/integration/test_main_lifespan.py b/tests/integration/test_main_lifespan.py new file mode 100644 index 0000000..db9295d --- /dev/null +++ b/tests/integration/test_main_lifespan.py @@ -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