"""HTTP Server and Transport Tests.
Comprehensive tests for HTTP server functionality:
1. Server starts on configured port
2. Health endpoint returns 200
3. SSE connection established
4. Tool calls work over HTTP
5. Resource reads work over HTTP
6. CORS headers present
Reference: mcp_agent_mail-9z6
"""
from __future__ import annotations
import contextlib
from typing import Any
import pytest
from httpx import ASGITransport, AsyncClient
from mcp_agent_mail import config as _config
from mcp_agent_mail.app import build_mcp_server
from mcp_agent_mail.db import ensure_schema
from mcp_agent_mail.http import build_http_app
def _rpc(method: str, params: dict[str, Any]) -> dict[str, Any]:
"""Create a JSON-RPC 2.0 request payload."""
return {"jsonrpc": "2.0", "id": "1", "method": method, "params": params}
# =============================================================================
# Test: Server Starts on Configured Port
# =============================================================================
class TestServerConfiguration:
"""Test that server respects configuration settings."""
@pytest.mark.asyncio
async def test_server_builds_with_default_config(self, isolated_env):
"""Server builds successfully with default configuration."""
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
assert app is not None
# FastAPI app should have routes
assert len(app.routes) > 0
@pytest.mark.asyncio
async def test_server_uses_configured_path(self, isolated_env, monkeypatch):
"""Server mounts MCP handler at configured HTTP_PATH."""
monkeypatch.setenv("HTTP_PATH", "/custom-mcp/")
with contextlib.suppress(Exception):
_config.clear_settings_cache()
settings = _config.get_settings()
assert settings.http.path == "/custom-mcp/"
server = build_mcp_server()
app = build_http_app(settings, server)
assert app is not None
@pytest.mark.asyncio
async def test_server_builds_with_custom_host_port(self, isolated_env, monkeypatch):
"""Server configuration accepts custom host and port."""
monkeypatch.setenv("HTTP_HOST", "0.0.0.0")
monkeypatch.setenv("HTTP_PORT", "9999")
with contextlib.suppress(Exception):
_config.clear_settings_cache()
settings = _config.get_settings()
assert settings.http.host == "0.0.0.0"
assert settings.http.port == 9999
server = build_mcp_server()
app = build_http_app(settings, server)
assert app is not None
# =============================================================================
# Test: Health Endpoints Return 200
# =============================================================================
class TestHealthEndpoints:
"""Test health check endpoints."""
@pytest.mark.asyncio
async def test_liveness_returns_200(self, isolated_env):
"""Liveness endpoint returns 200 with status 'alive'."""
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.get("/health/liveness")
assert response.status_code == 200
data = response.json()
assert data["status"] == "alive"
@pytest.mark.asyncio
async def test_readiness_returns_200_when_healthy(self, isolated_env):
"""Readiness endpoint returns 200 when database is accessible."""
# Ensure schema exists for readiness check
await ensure_schema()
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.get("/health/readiness")
assert response.status_code == 200
data = response.json()
assert data["status"] == "ready"
@pytest.mark.asyncio
async def test_health_endpoints_bypass_auth(self, isolated_env, monkeypatch):
"""Health endpoints work without authentication."""
monkeypatch.setenv("HTTP_BEARER_TOKEN", "secret-token")
monkeypatch.setenv("HTTP_ALLOW_LOCALHOST_UNAUTHENTICATED", "false")
with contextlib.suppress(Exception):
_config.clear_settings_cache()
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
# No auth header - should still work for health endpoints
r1 = await client.get("/health/liveness")
assert r1.status_code == 200
# Readiness might fail if DB not ready, but should not be 401
r2 = await client.get("/health/readiness")
assert r2.status_code != 401
# =============================================================================
# Test: SSE Connection Established
# =============================================================================
class TestSSEConnection:
"""Test Server-Sent Events (SSE) connection capability."""
@pytest.mark.asyncio
async def test_sse_accept_header_supported(self, isolated_env):
"""Server accepts SSE content type in Accept header."""
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
# Request with SSE Accept header
headers = {"Accept": "text/event-stream"}
response = await client.get(
"/health/liveness",
headers=headers,
)
# Health endpoints return JSON regardless, but should not error
assert response.status_code == 200
@pytest.mark.asyncio
async def test_mcp_endpoint_accepts_sse_header(self, isolated_env):
"""MCP endpoint accepts SSE content negotiation."""
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
# POST to MCP path with SSE in Accept
headers = {
"Accept": "application/json, text/event-stream",
"Content-Type": "application/json",
}
response = await client.post(
settings.http.path,
headers=headers,
json=_rpc("tools/call", {"name": "health_check", "arguments": {}}),
)
# Should get a valid response (200 or 401 if auth required)
assert response.status_code in (200, 401)
# =============================================================================
# Test: Tool Calls Work Over HTTP
# =============================================================================
class TestToolCallsOverHTTP:
"""Test that MCP tool calls work over HTTP transport."""
@pytest.mark.asyncio
async def test_health_check_tool_succeeds(self, isolated_env):
"""health_check tool call returns success over HTTP."""
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.post(
settings.http.path,
json=_rpc("tools/call", {"name": "health_check", "arguments": {}}),
)
assert response.status_code == 200
data = response.json()
# JSON-RPC response should have result
assert "result" in data or "error" not in data
@pytest.mark.asyncio
async def test_tool_call_returns_jsonrpc_format(self, isolated_env):
"""Tool calls return proper JSON-RPC 2.0 format."""
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.post(
settings.http.path,
json=_rpc("tools/call", {"name": "health_check", "arguments": {}}),
)
assert response.status_code == 200
data = response.json()
assert data.get("jsonrpc") == "2.0"
assert "id" in data
@pytest.mark.asyncio
async def test_tool_call_with_bearer_auth(self, isolated_env, monkeypatch):
"""Tool calls work with bearer token authentication."""
monkeypatch.setenv("HTTP_BEARER_TOKEN", "my-secret-token")
monkeypatch.setenv("HTTP_ALLOW_LOCALHOST_UNAUTHENTICATED", "false")
with contextlib.suppress(Exception):
_config.clear_settings_cache()
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
# Without auth -> 401
r1 = await client.post(
settings.http.path,
json=_rpc("tools/call", {"name": "health_check", "arguments": {}}),
)
assert r1.status_code == 401
# With correct auth -> 200
r2 = await client.post(
settings.http.path,
headers={"Authorization": "Bearer my-secret-token"},
json=_rpc("tools/call", {"name": "health_check", "arguments": {}}),
)
assert r2.status_code == 200
# =============================================================================
# Test: Resource Reads Work Over HTTP
# =============================================================================
class TestResourceReadsOverHTTP:
"""Test that MCP resource reads work over HTTP transport."""
@pytest.mark.asyncio
async def test_resources_list_returns_data(self, isolated_env):
"""resources/list returns available resources."""
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.post(
settings.http.path,
json=_rpc("resources/list", {}),
)
assert response.status_code == 200
data = response.json()
assert "result" in data or "error" not in data
@pytest.mark.asyncio
async def test_resource_read_returns_jsonrpc(self, isolated_env):
"""Resource reads return proper JSON-RPC response."""
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.post(
settings.http.path,
json=_rpc("resources/read", {"uri": "resource://projects"}),
)
assert response.status_code == 200
data = response.json()
assert data.get("jsonrpc") == "2.0"
# =============================================================================
# Test: CORS Headers Present
# =============================================================================
class TestCORSHeaders:
"""Test CORS (Cross-Origin Resource Sharing) configuration."""
@pytest.mark.asyncio
async def test_cors_preflight_returns_headers(self, isolated_env, monkeypatch):
"""CORS preflight OPTIONS request returns appropriate headers."""
monkeypatch.setenv("HTTP_CORS_ENABLED", "true")
monkeypatch.setenv("HTTP_CORS_ORIGINS", "http://example.com")
with contextlib.suppress(Exception):
_config.clear_settings_cache()
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.options(
settings.http.path,
headers={
"Origin": "http://example.com",
"Access-Control-Request-Method": "POST",
},
)
assert response.status_code in (200, 204)
# CORS headers should be present
assert "access-control-allow-origin" in response.headers
@pytest.mark.asyncio
async def test_cors_headers_on_response(self, isolated_env, monkeypatch):
"""CORS headers are present on regular responses."""
monkeypatch.setenv("HTTP_CORS_ENABLED", "true")
monkeypatch.setenv("HTTP_CORS_ORIGINS", "*")
with contextlib.suppress(Exception):
_config.clear_settings_cache()
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.post(
settings.http.path,
headers={"Origin": "http://test-origin.com"},
json=_rpc("tools/call", {"name": "health_check", "arguments": {}}),
)
assert response.status_code == 200
# CORS header should be present
assert response.headers.get("access-control-allow-origin") in ("*", "http://test-origin.com")
@pytest.mark.asyncio
async def test_cors_disabled_no_headers(self, isolated_env, monkeypatch):
"""When CORS is disabled, no CORS headers are added."""
monkeypatch.setenv("HTTP_CORS_ENABLED", "false")
with contextlib.suppress(Exception):
_config.clear_settings_cache()
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.get("/health/liveness")
assert response.status_code == 200
# CORS headers should not be present when disabled
# Note: This may vary based on implementation; check for absence
# of origin-specific headers on non-preflight requests
# =============================================================================
# Test: Error Handling
# =============================================================================
class TestHTTPErrorHandling:
"""Test HTTP error handling."""
@pytest.mark.asyncio
async def test_invalid_json_returns_error(self, isolated_env):
"""Invalid JSON payload returns appropriate error."""
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.post(
settings.http.path,
content=b"not valid json{{{",
headers={"Content-Type": "application/json"},
)
# Should return 4xx error for invalid JSON
assert response.status_code >= 400
@pytest.mark.asyncio
async def test_missing_method_returns_error(self, isolated_env):
"""JSON-RPC request without method returns error."""
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.post(
settings.http.path,
json={"jsonrpc": "2.0", "id": "1"}, # Missing method
)
# Should return 200 with JSON-RPC error or 400
assert response.status_code in (200, 400)
if response.status_code == 200:
data = response.json()
assert "error" in data
# =============================================================================
# Test: Request Logging
# =============================================================================
class TestRequestLogging:
"""Test request logging middleware."""
@pytest.mark.asyncio
async def test_request_logs_path_and_status(self, isolated_env, caplog):
"""Requests are logged with path and status."""
import logging
caplog.set_level(logging.DEBUG)
settings = _config.get_settings()
server = build_mcp_server()
app = build_http_app(settings, server)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
await client.get("/health/liveness")
# Logging should have occurred (may use structlog or stdlib)
# This is a smoke test that the request completes
