test changes

tests/conftest.py

@@ -1,47 +1,18 @@
-import socket
-import sys
-from collections.abc import Iterator
+from collections.abc import AsyncIterator
 
 import pytest
-from xprocess import ProcessStarter
-from xprocess import XProcess
+from asgi_lifespan import LifespanManager
+from fastmcp.utilities.tests import ASGIServer
 
-SERVER_HOST = "127.0.0.1"
+from personal_mcp.app import create_app
 
 
 @pytest.fixture
-def fastapi_server_url(xprocess: XProcess, unused_tcp_port: int) -> Iterator[str]:
-    class ServerStarter(ProcessStarter):
-        timeout = 10
-        terminate_on_interrupt = True
-
-        @property
-        def args(self) -> list[str]:
-            return [
-                sys.executable,
-                "-m",
-                "uvicorn",
-                "personal_mcp.app:create_app",
-                "--factory",
-                "--host",
-                SERVER_HOST,
-                "--port",
-                str(unused_tcp_port),
-                "--log-level",
-                "warning",
-                "--no-access-log",
-            ]
-
-        def startup_check(self) -> bool:
-            try:
-                with socket.create_connection((SERVER_HOST, unused_tcp_port), timeout=0.1):
-                    return True
-            except OSError:
-                return False
-
-    process_name = f"personal-mcp-http-{unused_tcp_port}"
-    xprocess.ensure(process_name, ServerStarter)
-    try:
-        yield f"http://{SERVER_HOST}:{unused_tcp_port}/mcp"
-    finally:
-        xprocess.getinfo(process_name).terminate()
+async def http_server() -> AsyncIterator[ASGIServer]:
+    app = create_app()
+    async with LifespanManager(app):
+        yield ASGIServer(
+            url="http://127.0.0.1/mcp",
+            app=app,
+            transport_type="http",
+        )

tests/server_contract.py

@@ -0,0 +1,22 @@
+from fastmcp import Client
+from mcp_types import TextResourceContents
+
+
+async def assert_server_contract(client: Client) -> None:
+    resources = {str(resource.uri) for resource in await client.list_resources()}
+    templates = {str(template.uri_template) for template in await client.list_resource_templates()}
+    prompts = {prompt.name for prompt in await client.list_prompts()}
+
+    assert await client.list_tools() == []
+    assert "skill://pytesting/SKILL.md" in resources
+    assert "resource://docs/{path*}" in templates
+    assert "skill://pytesting/{path*}" in templates
+    assert "pytest-fill-scaffold" in prompts
+
+    skill_content = await client.read_resource("skill://pytesting/SKILL.md")
+    docs_content = await client.read_resource("resource://docs/index.md")
+
+    assert isinstance(skill_content[0], TextResourceContents)
+    assert "# Pytesting" in skill_content[0].text
+    assert isinstance(docs_content[0], TextResourceContents)
+    assert '"format": "markdown"' in docs_content[0].text

tests/test_http.py

@@ -1,11 +1,9 @@
 import pytest
-from fastmcp import Client
-from fastmcp.client.transports import StreamableHttpTransport
-from test_stdio import assert_server_contract
+from fastmcp.utilities.tests import ASGIServer
+from server_contract import assert_server_contract
 
 
 @pytest.mark.integration
-async def test_fastapi_server_enumerates_and_reads_content(fastapi_server_url: str) -> None:
-    transport = StreamableHttpTransport(fastapi_server_url)
-    async with Client(transport) as client:
+async def test_fastapi_server(http_server: ASGIServer) -> None:
+    async with http_server.client() as client:
         await assert_server_contract(client)

tests/test_stdio.py

@@ -4,31 +4,11 @@ from pathlib import Path
 import pytest
 from fastmcp import Client
 from fastmcp.client.transports import StdioTransport
-from mcp_types import TextResourceContents
-
-
-async def assert_server_contract(client: Client) -> None:
-    resources = {str(resource.uri) for resource in await client.list_resources()}
-    templates = {str(template.uri_template) for template in await client.list_resource_templates()}
-    prompts = {prompt.name for prompt in await client.list_prompts()}
-
-    assert await client.list_tools() == []
-    assert "skill://pytesting/SKILL.md" in resources
-    assert "resource://docs/{path*}" in templates
-    assert "skill://pytesting/{path*}" in templates
-    assert "pytest-fill-scaffold" in prompts
-
-    skill_content = await client.read_resource("skill://pytesting/SKILL.md")
-    docs_content = await client.read_resource("resource://docs/index.md")
-
-    assert isinstance(skill_content[0], TextResourceContents)
-    assert "# Pytesting" in skill_content[0].text
-    assert isinstance(docs_content[0], TextResourceContents)
-    assert '"format": "markdown"' in docs_content[0].text
+from server_contract import assert_server_contract
 
 
 @pytest.mark.integration
-async def test_stdio_server_enumerates_and_reads_content() -> None:
+async def test_stdio_server() -> None:
     transport = StdioTransport(
         command=sys.executable,
         args=["-m", "personal_mcp.mcp"],