60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import AsyncIterator
|
|
from contextlib import asynccontextmanager
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import ASGITransport
|
|
from httpx import AsyncClient
|
|
from mcp import ClientSession
|
|
from mcp.client.streamable_http import streamable_http_client
|
|
|
|
from personal_mcp.web.app import create_app
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client() -> AsyncIterator[AsyncClient]:
|
|
"""Provides an AsyncClient bound to a fresh application instance."""
|
|
app = create_app()
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app),
|
|
base_url="http://testserver",
|
|
timeout=10.0,
|
|
) as test_client:
|
|
yield test_client
|
|
|
|
|
|
@pytest.fixture
|
|
def mcp_endpoint_url() -> str:
|
|
"""Provides the MCP endpoint URL for SDK-based client sessions."""
|
|
return os.getenv("PERSONAL_MCP_TEST_HTTP_URL", "")
|
|
|
|
|
|
@pytest.fixture
|
|
def mcp_session_factory(mcp_endpoint_url: str):
|
|
"""Provides a context manager factory for MCP SDK sessions.
|
|
|
|
Keeping stream/client/session enter and exit in the test task avoids
|
|
cross-task cancel scope teardown errors from async generator fixtures.
|
|
"""
|
|
if not mcp_endpoint_url:
|
|
pytest.skip("Set PERSONAL_MCP_TEST_HTTP_URL to run SDK-backed MCP endpoint tests.")
|
|
|
|
@asynccontextmanager
|
|
async def create_session(*, initialize: bool = True) -> AsyncIterator[ClientSession]:
|
|
async with (
|
|
AsyncClient(timeout=10.0) as http_client,
|
|
streamable_http_client(
|
|
mcp_endpoint_url,
|
|
http_client=http_client,
|
|
) as (read_stream, write_stream, _),
|
|
ClientSession(read_stream, write_stream) as session,
|
|
):
|
|
if initialize:
|
|
await session.initialize()
|
|
yield session
|
|
|
|
return create_session
|