generated from personal-projects/leo-claude-mktplace
Issue #25 - Docker multi-service infrastructure: - Create docker/Dockerfile with multi-stage build, git support, port 8080 - Create docker/docker-compose.yml with app + Caddy services - Create docker/Caddyfile for HTTPS termination and reverse proxy - Create docker/.env.example with configuration template Issue #26 - Startup scripts and tests: - Create scripts/start.sh for production startup with env validation - Create scripts/healthcheck.sh for Docker health checks - Add health endpoint tests to test_mcp_endpoints.py - Fix middleware order (HealthCheckBypass must wrap BearerAuth) - Fix pyproject.toml testpaths to use 'tests' directory - Update test_config.py for new defaults (0.0.0.0:8080) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
138 lines
3.8 KiB
Python
138 lines
3.8 KiB
Python
"""Tests for MCP protocol endpoints and health checks."""
|
|
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)
|
|
|
|
|
|
# =============================================================================
|
|
# Health Endpoint Tests
|
|
# =============================================================================
|
|
|
|
|
|
def test_health_endpoint(client):
|
|
"""Test GET /health returns status ok."""
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "ok"
|
|
|
|
|
|
def test_health_endpoint_no_auth_required(mock_env):
|
|
"""Test health endpoint works even with AUTH_TOKEN set."""
|
|
with patch.dict("os.environ", {**mock_env, "AUTH_TOKEN": "secret123"}):
|
|
app = create_app()
|
|
client = TestClient(app)
|
|
# Health should bypass auth
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "ok"
|
|
|
|
|
|
# =============================================================================
|
|
# MCP Protocol Tests
|
|
# =============================================================================
|
|
|
|
|
|
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
|