Agileday MCP Server

import os import sys import pytest from unittest.mock import AsyncMock, patch, MagicMock from starlette.testclient import TestClient from contextlib import asynccontextmanager # --- Setup Environment --- os.environ["AGILEDAY_TENANT_ID"] = "test-tenant" os.environ["AGILEDAY_API_TOKEN"] = "test-token" # Add src to path sys.path.append(os.path.join(os.path.dirname(__file__), '../src')) from agileday_server import starlette_app # --- Fixtures --- @pytest.fixture def client(): """Returns a Starlette TestClient for making HTTP requests.""" return TestClient(starlette_app) @pytest.fixture def mock_server_run(): """ Patches the internal MCP server.run method to prevent it from blocking. """ with patch("agileday_server.server.run", new_callable=AsyncMock) as mock_run: yield mock_run @pytest.fixture def mock_sse_transport(mock_server_run): """ Mocks the internal SSE Transport methods to simulate ASGI responses. """ with patch("agileday_server.sse_transport") as mock: # 1. Mock handle_post_message # The real SDK writes a 202 Accepted response to the 'send' channel. # We must replicate that side effect here. async def side_effect_post(scope, receive, send): await send({ 'type': 'http.response.start', 'status': 202, 'headers': [[b'content-type', b'text/plain']] }) await send({ 'type': 'http.response.body', 'body': b'Accepted' }) mock.handle_post_message = AsyncMock(side_effect=side_effect_post) # 2. Mock connect_sse # The real SDK writes 200 OK headers for the Event Stream. # We use asynccontextmanager to replicate the 'async with' behavior. @asynccontextmanager async def side_effect_connect(scope, receive, send): # Simulate starting the stream await send({ 'type': 'http.response.start', 'status': 200, 'headers': [[b'content-type', b'text/event-stream']] }) # Yield mock streams (reader, writer) required by the server yield (AsyncMock(), AsyncMock()) mock.connect_sse.side_effect = side_effect_connect yield mock # --- Tests --- def test_health_check(client): """Ensure the health endpoint returns 200 OK.""" response = client.get("/health") assert response.status_code == 200 assert response.text == "OK" def test_mcp_post_message_accepted(client, mock_sse_transport): """ Verifies that a POST to /mcp returns 202 Accepted. """ response = client.post("/mcp", json={"jsonrpc": "2.0", "method": "ping"}) # We expect the mock to have been called with the raw ASGI arguments assert mock_sse_transport.handle_post_message.call_count == 1 # Check the response assert response.status_code == 202 assert response.text == "Accepted" def test_mcp_get_connects_sse(client, mock_sse_transport): """ Verifies that GET /mcp attempts to establish an SSE stream. """ with client.stream("GET", "/mcp") as response: # Check we got the 200 OK headers simulated by our mock assert response.status_code == 200 # Verify the context manager was entered assert mock_sse_transport.connect_sse.called def test_legacy_sse_endpoint_redirects(client, mock_sse_transport): """ Verifies that the deprecated /sse endpoint still works. """ with client.stream("GET", "/sse") as response: assert response.status_code == 200 assert mock_sse_transport.connect_sse.called def test_method_not_allowed(client): """Ensure we enforce allowed methods (PUT should fail).""" response = client.put("/mcp") assert response.status_code == 405