test: Add tests for MCP tools and resources
This commit is contained in:
+45
-5
@@ -4,12 +4,12 @@ import logging
|
||||
import uuid
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
from httpx import HTTPStatusError
|
||||
import asyncio
|
||||
from mcp import ClientSession
|
||||
from mcp.client.sse import sse_client
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# pytestmark = pytest.mark.asyncio(loop_scope="package")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
async def nc_client() -> NextcloudClient:
|
||||
@@ -35,13 +35,54 @@ async def nc_client() -> NextcloudClient:
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def nc_mcp_client():
|
||||
"""
|
||||
Fixture to create an MCP client session for integration tests.
|
||||
"""
|
||||
logger.info("Creating SSE client")
|
||||
sse_context = sse_client(url="http://127.0.0.1:8000/sse")
|
||||
session_context = None
|
||||
|
||||
try:
|
||||
read, write = await sse_context.__aenter__()
|
||||
session_context = ClientSession(read, write)
|
||||
session = await session_context.__aenter__()
|
||||
await session.initialize()
|
||||
logger.info("MCP client session initialized successfully")
|
||||
|
||||
yield session
|
||||
|
||||
finally:
|
||||
# Clean up in reverse order, ignoring task scope issues
|
||||
if session_context is not None:
|
||||
try:
|
||||
await session_context.__aexit__(None, None, None)
|
||||
except RuntimeError as e:
|
||||
if "cancel scope" in str(e):
|
||||
logger.debug(f"Ignoring cancel scope teardown issue: {e}")
|
||||
else:
|
||||
logger.warning(f"Error closing session: {e}")
|
||||
except Exception as e:
|
||||
logger.warning(f"Error closing session: {e}")
|
||||
|
||||
try:
|
||||
await sse_context.__aexit__(None, None, None)
|
||||
except RuntimeError as e:
|
||||
if "cancel scope" in str(e):
|
||||
logger.debug(f"Ignoring cancel scope teardown issue: {e}")
|
||||
else:
|
||||
logger.warning(f"Error closing SSE client: {e}")
|
||||
except Exception as e:
|
||||
logger.warning(f"Error closing SSE client: {e}")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def temporary_note(nc_client: NextcloudClient):
|
||||
"""
|
||||
Fixture to create a temporary note for a test and ensure its deletion afterward.
|
||||
Yields the created note dictionary.
|
||||
"""
|
||||
asyncio.new_event_loop()
|
||||
|
||||
note_id = None
|
||||
unique_suffix = uuid.uuid4().hex[:8]
|
||||
@@ -87,7 +128,6 @@ async def temporary_note_with_attachment(
|
||||
Yields a tuple: (note_data, attachment_filename, attachment_content).
|
||||
Depends on the temporary_note fixture.
|
||||
"""
|
||||
asyncio.new_event_loop()
|
||||
|
||||
note_data = temporary_note
|
||||
note_id = note_data["id"]
|
||||
|
||||
Reference in New Issue
Block a user