generated from personal-projects/leo-claude-mktplace
lmiranda
a5390a3086
- Add POST /mcp endpoint using StreamableHTTPServerTransport - Add HEAD /mcp endpoint returning protocol version header - Remove old custom REST endpoints (/tools/list, /tools/call) - Integrate marketplace tools via MCP Server decorators - Add comprehensive MCP endpoint tests - Update README with correct MCP protocol usage The implementation properly handles: - JSON-RPC 2.0 message format - SSE (Server-Sent Events) responses - Protocol version negotiation - Tool filtering integration - Authentication middleware Tests verify: - HEAD /mcp returns correct headers - POST /mcp handles initialize requests - Proper error handling for missing headers Closes #24 Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
109 lines
2.8 KiB
Python
109 lines
2.8 KiB
Python
"""Tests for MCP protocol endpoints."""
|
|
import pytest
|
|
import json
|
|
import re
|
|
from starlette.testclient import TestClient
|
|
from gitea_mcp_remote.server_http import create_app
|
|
from unittest.mock import patch
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_env():
|
|
"""Mock environment variables for testing."""
|
|
env = {
|
|
"GITEA_URL": "https://gitea.example.com",
|
|
"GITEA_TOKEN": "test_token",
|
|
"GITEA_OWNER": "test_owner",
|
|
}
|
|
with patch.dict("os.environ", env):
|
|
yield env
|
|
|
|
|
|
@pytest.fixture
|
|
def client(mock_env):
|
|
"""Create test client."""
|
|
app = create_app()
|
|
return TestClient(app)
|
|
|
|
|
|
def parse_sse_message(sse_text: str) -> dict:
|
|
"""Parse SSE message data."""
|
|
data_match = re.search(r'data: (.+)', sse_text)
|
|
if data_match:
|
|
return json.loads(data_match.group(1))
|
|
return None
|
|
|
|
|
|
def test_mcp_head_endpoint(client):
|
|
"""Test HEAD /mcp returns protocol version header."""
|
|
response = client.head("/mcp")
|
|
assert response.status_code == 200
|
|
assert "x-mcp-protocol-version" in response.headers
|
|
assert response.headers["x-mcp-protocol-version"] == "2024-11-05"
|
|
|
|
|
|
def test_mcp_initialize(client):
|
|
"""Test MCP initialize request."""
|
|
initialize_request = {
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "initialize",
|
|
"params": {
|
|
"protocolVersion": "2024-11-05",
|
|
"capabilities": {},
|
|
"clientInfo": {
|
|
"name": "test-client",
|
|
"version": "1.0.0"
|
|
}
|
|
}
|
|
}
|
|
|
|
response = client.post(
|
|
"/mcp",
|
|
json=initialize_request,
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json, text/event-stream"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
# Parse SSE response
|
|
data = parse_sse_message(response.text)
|
|
assert data is not None
|
|
assert data.get("jsonrpc") == "2.0"
|
|
assert "result" in data
|
|
assert data["result"].get("protocolVersion") == "2024-11-05"
|
|
assert "serverInfo" in data["result"]
|
|
assert data["result"]["serverInfo"]["name"] == "gitea-mcp-remote"
|
|
|
|
|
|
def test_mcp_missing_accept_header(client):
|
|
"""Test MCP request without required Accept header."""
|
|
initialize_request = {
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "initialize",
|
|
"params": {
|
|
"protocolVersion": "2024-11-05",
|
|
"capabilities": {},
|
|
"clientInfo": {
|
|
"name": "test-client",
|
|
"version": "1.0.0"
|
|
}
|
|
}
|
|
}
|
|
|
|
response = client.post(
|
|
"/mcp",
|
|
json=initialize_request,
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json" # Missing text/event-stream
|
|
}
|
|
)
|
|
|
|
# Should return error about missing accept header
|
|
assert response.status_code == 406
|