354 lines
13 KiB
Python
354 lines
13 KiB
Python
"""Unit and integration tests for the TryMe TPK FastMCP HTTP server."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import socket
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
import uvicorn
|
|
import yaml
|
|
from fastmcp import Client
|
|
from fastmcp.client.transports import StreamableHttpTransport
|
|
from fastmcp.exceptions import ToolError
|
|
|
|
from app.config import Settings, get_version
|
|
from app.main import app, mcp
|
|
from tests.conftest import UpstreamMockState
|
|
|
|
EXPECTED_TOOLS = {
|
|
# System
|
|
"healthcheck",
|
|
# Authentication
|
|
"get_auth_config",
|
|
"auth0_login",
|
|
"auth0_callback",
|
|
"dev_login",
|
|
"logout",
|
|
# Users & API Keys
|
|
"get_me",
|
|
"update_me",
|
|
"get_my_scores",
|
|
"list_api_keys",
|
|
"create_api_key",
|
|
"expire_api_key",
|
|
# Quizzes
|
|
"list_quizzes",
|
|
"create_quiz",
|
|
"get_quiz",
|
|
"update_quiz",
|
|
"delete_quiz",
|
|
"submit_quiz",
|
|
"vote_quiz",
|
|
# Favourites
|
|
"list_favourites",
|
|
"add_favourite",
|
|
"reorder_favourites",
|
|
"remove_favourite",
|
|
# Moderation
|
|
"claim_initial_moderator",
|
|
"list_users_for_moderation",
|
|
"ban_user",
|
|
"unban_user",
|
|
"list_quizzes_for_moderation",
|
|
"moderator_remove_quiz",
|
|
}
|
|
|
|
|
|
def _unwrap_result(result: Any) -> Any:
|
|
"""Extract structured/JSON payload from a FastMCP CallToolResult."""
|
|
if getattr(result, "structured_content", None) is not None:
|
|
sc = result.structured_content
|
|
if isinstance(sc, dict) and set(sc.keys()) == {"result"}:
|
|
return sc["result"]
|
|
return sc
|
|
if getattr(result, "data", None) is not None:
|
|
return result.data
|
|
for item in getattr(result, "content", []):
|
|
if hasattr(item, "text"):
|
|
return json.loads(item.text)
|
|
return None
|
|
|
|
|
|
def test_default_settings_bind_all_interfaces(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Default configuration binds HTTP server to 0.0.0.0:8000 on /mcp."""
|
|
for var in ("MCP_HOST", "MCP_PORT", "MCP_PATH", "TRYME_BASE_URL"):
|
|
monkeypatch.delenv(var, raising=False)
|
|
|
|
cfg = Settings.from_env()
|
|
assert cfg.mcp_host == "0.0.0.0"
|
|
assert cfg.mcp_port == 8000
|
|
assert cfg.mcp_path == "/mcp"
|
|
assert cfg.tryme_base_url == "https://tryme.tpk.pw"
|
|
assert get_version() == "0.1"
|
|
|
|
|
|
async def test_custom_http_routes_health_and_openapi() -> None:
|
|
"""Verify /health, /openapi.yaml, and /openapi.json on the HTTP ASGI app."""
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as http:
|
|
health_resp = await http.get("/health")
|
|
assert health_resp.status_code == 200
|
|
health_data = health_resp.json()
|
|
assert health_data["status"] == "ok"
|
|
assert health_data["service"] == "trymemcp"
|
|
assert health_data["transport"] == "http"
|
|
assert health_data["host"] == "0.0.0.0"
|
|
|
|
yaml_resp = await http.get("/openapi.yaml")
|
|
assert yaml_resp.status_code == 200
|
|
parsed_yaml = yaml.safe_load(yaml_resp.text)
|
|
assert parsed_yaml["openapi"] == "3.1.0"
|
|
assert len(parsed_yaml["x-mcp-tools"]) == 29
|
|
|
|
json_resp = await http.get("/openapi.json")
|
|
assert json_resp.status_code == 200
|
|
assert json_resp.json()["info"]["title"] == "TryMe TPK FastMCP HTTP Server"
|
|
|
|
|
|
async def test_all_29_openapi_tools_registered() -> None:
|
|
"""Ensure all 29 operations from https://tryme.tpk.pw/openapi.json are exposed."""
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
names = {t.name for t in tools}
|
|
assert names == EXPECTED_TOOLS
|
|
assert len(names) == 29
|
|
|
|
|
|
async def test_system_and_auth_tools(mock_upstream: UpstreamMockState) -> None:
|
|
"""Test system healthcheck and authentication tools."""
|
|
async with Client(mcp) as client:
|
|
res = _unwrap_result(await client.call_tool("healthcheck", {}))
|
|
assert res["status"] == "ok"
|
|
assert mock_upstream.calls[-1].path == "/health"
|
|
|
|
cfg = _unwrap_result(await client.call_tool("get_auth_config", {}))
|
|
assert cfg["auth0_enabled"] is True
|
|
|
|
login_res = _unwrap_result(await client.call_tool("auth0_login", {"return_to": "/dashboard"}))
|
|
assert login_res["status_code"] == 302
|
|
assert "authorize" in login_res["redirect_url"]
|
|
assert mock_upstream.calls[-1].query == {"return_to": "/dashboard"}
|
|
|
|
cb_res = _unwrap_result(
|
|
await client.call_tool("auth0_callback", {"code": "abc", "state": "xyz"})
|
|
)
|
|
assert cb_res["status_code"] == 302
|
|
assert mock_upstream.calls[-1].query == {"code": "abc", "state": "xyz"}
|
|
|
|
dev_res = _unwrap_result(
|
|
await client.call_tool(
|
|
"dev_login",
|
|
{"nickname": "quizmaster", "email": "qm@example.com", "as_moderator": True},
|
|
)
|
|
)
|
|
assert dev_res["access_token"] == "dev-jwt-token"
|
|
assert mock_upstream.calls[-1].body == {
|
|
"nickname": "quizmaster",
|
|
"email": "qm@example.com",
|
|
"as_moderator": True,
|
|
}
|
|
|
|
logout_res = _unwrap_result(
|
|
await client.call_tool("logout", {"authorization": "raw-jwt-token"})
|
|
)
|
|
assert logout_res == {"status": "logged_out"}
|
|
assert mock_upstream.calls[-1].headers["authorization"] == "Bearer raw-jwt-token"
|
|
|
|
|
|
async def test_users_and_api_keys_tools(mock_upstream: UpstreamMockState) -> None:
|
|
"""Test user profile, scores, and API key lifecycle tools."""
|
|
async with Client(mcp) as client:
|
|
# Uses env fallback X-API-Key when no explicit key is passed
|
|
await client.call_tool("get_me", {})
|
|
assert mock_upstream.calls[-1].path == "/api/v1/users/me"
|
|
assert mock_upstream.calls[-1].headers["x-api-key"] == "env-default-key"
|
|
|
|
# Explicit api_key overrides env default
|
|
await client.call_tool(
|
|
"update_me",
|
|
{
|
|
"nickname": "new_nick",
|
|
"age": 16,
|
|
"public_website_consent": True,
|
|
"api_key": "override-key",
|
|
},
|
|
)
|
|
last = mock_upstream.calls[-1]
|
|
assert last.method == "PATCH"
|
|
assert last.headers["x-api-key"] == "override-key"
|
|
assert last.body == {
|
|
"nickname": "new_nick",
|
|
"age": 16,
|
|
"public_website_consent": True,
|
|
}
|
|
|
|
scores = _unwrap_result(await client.call_tool("get_my_scores", {}))
|
|
assert isinstance(scores, list)
|
|
assert mock_upstream.calls[-1].path == "/api/v1/users/me/scores"
|
|
|
|
keys = _unwrap_result(await client.call_tool("list_api_keys", {}))
|
|
assert isinstance(keys, list)
|
|
assert mock_upstream.calls[-1].path == "/api/v1/users/me/api-keys"
|
|
|
|
await client.call_tool("create_api_key", {"name": "ci-agent", "expires_in_days": 30})
|
|
assert mock_upstream.calls[-1].method == "POST"
|
|
assert mock_upstream.calls[-1].body == {"name": "ci-agent", "expires_in_days": 30}
|
|
|
|
await client.call_tool("expire_api_key", {"key_id": 42})
|
|
assert mock_upstream.calls[-1].path == "/api/v1/users/me/api-keys/42/expire"
|
|
|
|
|
|
async def test_quiz_tools(mock_upstream: UpstreamMockState) -> None:
|
|
"""Test quiz listing, CRUD, submission, and voting tools."""
|
|
async with Client(mcp) as client:
|
|
await client.call_tool(
|
|
"list_quizzes",
|
|
{"q": "science", "mode": "multiple_choice", "mine_only": True},
|
|
)
|
|
assert mock_upstream.calls[-1].query == {
|
|
"q": "science",
|
|
"mode": "multiple_choice",
|
|
"mine_only": "true",
|
|
}
|
|
|
|
await client.call_tool(
|
|
"create_quiz",
|
|
{
|
|
"title": "Space Trivia",
|
|
"description": "Planets and stars",
|
|
"is_public": True,
|
|
"questions": [
|
|
{
|
|
"question_type": "multiple_choice",
|
|
"prompt": "Closest planet to the Sun?",
|
|
"options": [
|
|
{"id": "a", "text": "Mercury"},
|
|
{"id": "b", "text": "Venus"},
|
|
],
|
|
"correct_answer": "a",
|
|
"accepted_answers": [],
|
|
}
|
|
],
|
|
},
|
|
)
|
|
created_call = mock_upstream.calls[-1]
|
|
assert created_call.method == "POST"
|
|
assert created_call.path == "/api/v1/quizzes"
|
|
assert created_call.body["title"] == "Space Trivia"
|
|
assert created_call.body["questions"][0]["correct_answer"] == "a"
|
|
|
|
await client.call_tool("get_quiz", {"slug_or_id": "space-trivia"})
|
|
assert mock_upstream.calls[-1].path == "/api/v1/quizzes/space-trivia"
|
|
|
|
await client.call_tool(
|
|
"update_quiz",
|
|
{"quiz_id": 7, "title": "Updated Space Trivia", "is_public": False},
|
|
)
|
|
assert mock_upstream.calls[-1].method == "PUT"
|
|
assert mock_upstream.calls[-1].path == "/api/v1/quizzes/7"
|
|
|
|
del_res = _unwrap_result(await client.call_tool("delete_quiz", {"quiz_id": 7}))
|
|
assert del_res == {"status": "deleted", "quiz_id": 7}
|
|
assert mock_upstream.calls[-1].method == "DELETE"
|
|
|
|
await client.call_tool(
|
|
"submit_quiz",
|
|
{"quiz_id": 7, "answers": {"101": "a", "102": "42"}},
|
|
)
|
|
assert mock_upstream.calls[-1].path == "/api/v1/quizzes/7/submit"
|
|
assert mock_upstream.calls[-1].body == {"answers": {"101": "a", "102": "42"}}
|
|
|
|
await client.call_tool("vote_quiz", {"quiz_id": 7, "value": 1})
|
|
assert mock_upstream.calls[-1].path == "/api/v1/quizzes/7/vote"
|
|
assert mock_upstream.calls[-1].body == {"value": 1}
|
|
|
|
|
|
async def test_favourites_and_moderation_tools(mock_upstream: UpstreamMockState) -> None:
|
|
"""Test favourites management and admin moderation tools."""
|
|
async with Client(mcp) as client:
|
|
await client.call_tool("list_favourites", {})
|
|
assert mock_upstream.calls[-1].path == "/api/v1/favourites"
|
|
|
|
await client.call_tool("add_favourite", {"quiz_id": 12, "position": 0})
|
|
assert mock_upstream.calls[-1].body == {"quiz_id": 12, "position": 0}
|
|
|
|
await client.call_tool("reorder_favourites", {"quiz_ids": [12, 5, 9]})
|
|
assert mock_upstream.calls[-1].method == "PUT"
|
|
assert mock_upstream.calls[-1].body == {"quiz_ids": [12, 5, 9]}
|
|
|
|
await client.call_tool("remove_favourite", {"quiz_id": 12})
|
|
assert mock_upstream.calls[-1].method == "DELETE"
|
|
assert mock_upstream.calls[-1].path == "/api/v1/favourites/12"
|
|
|
|
await client.call_tool("claim_initial_moderator", {})
|
|
assert mock_upstream.calls[-1].path == "/api/v1/admin/claim"
|
|
|
|
await client.call_tool("list_users_for_moderation", {})
|
|
assert mock_upstream.calls[-1].path == "/api/v1/admin/users"
|
|
|
|
await client.call_tool("ban_user", {"user_id": 99, "reason": "Spamming quizzes"})
|
|
assert mock_upstream.calls[-1].path == "/api/v1/admin/users/99/ban"
|
|
assert mock_upstream.calls[-1].body == {"reason": "Spamming quizzes"}
|
|
|
|
await client.call_tool("unban_user", {"user_id": 99})
|
|
assert mock_upstream.calls[-1].path == "/api/v1/admin/users/99/unban"
|
|
|
|
await client.call_tool("list_quizzes_for_moderation", {})
|
|
assert mock_upstream.calls[-1].path == "/api/v1/admin/quizzes"
|
|
|
|
mod_del = _unwrap_result(await client.call_tool("moderator_remove_quiz", {"quiz_id": 55}))
|
|
assert mod_del == {"status": "removed", "quiz_id": 55}
|
|
assert mock_upstream.calls[-1].path == "/api/v1/admin/quizzes/55"
|
|
|
|
|
|
async def test_upstream_error_raises_tool_error(mock_upstream: UpstreamMockState) -> None:
|
|
"""Verify upstream 4xx/5xx responses surface cleanly as MCP ToolErrors."""
|
|
mock_upstream.next_status = 401
|
|
mock_upstream.next_json = {"detail": "Invalid or expired API key"}
|
|
|
|
async with Client(mcp) as client:
|
|
with pytest.raises(ToolError, match="401"):
|
|
await client.call_tool("get_me", {})
|
|
|
|
|
|
async def test_live_http_server_on_all_interfaces_and_header_passthrough(
|
|
mock_upstream: UpstreamMockState,
|
|
) -> None:
|
|
"""Start HTTP server bound to 0.0.0.0 and verify MCP HTTP client & header forwarding."""
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.bind(("0.0.0.0", 0))
|
|
port = sock.getsockname()[1]
|
|
|
|
http_app = mcp.http_app(path="/mcp", transport="http")
|
|
config = uvicorn.Config(http_app, host="0.0.0.0", port=port, log_level="error")
|
|
server = uvicorn.Server(config)
|
|
task = asyncio.create_task(server.serve(sockets=[sock]))
|
|
|
|
try:
|
|
while not server.started:
|
|
await asyncio.sleep(0.02)
|
|
|
|
transport = StreamableHttpTransport(
|
|
url=f"http://127.0.0.1:{port}/mcp",
|
|
headers={
|
|
"X-API-Key": "client-header-api-key",
|
|
"Cookie": "tryme_session=session-cookie-abc",
|
|
},
|
|
)
|
|
async with Client(transport) as client:
|
|
tools = await client.list_tools()
|
|
assert len(tools) == 29
|
|
|
|
await client.call_tool("get_me", {})
|
|
last_call = mock_upstream.calls[-1]
|
|
assert last_call.headers["x-api-key"] == "client-header-api-key"
|
|
assert "tryme_session=session-cookie-abc" in last_call.headers.get("cookie", "")
|
|
finally:
|
|
server.should_exit = True
|
|
await task
|
|
sock.close()
|