Spaces:
Paused
Paused
File size: 3,654 Bytes
c88c4a7 ede187b c88c4a7 3ccdaee | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """End-to-end MCP wire tests over the mounted lipsync endpoint."""
import pytest
from starlette.testclient import TestClient
from papers_mcp import server
from tests.conftest import CACHE_DIR, LIPSYNC_REPO
MCP_HEADERS = {
"Accept": "application/json, text/event-stream",
"Content-Type": "application/json",
}
def rpc(method: str, params: dict) -> dict:
return {"jsonrpc": "2.0", "id": 1, "method": method, "params": params}
@pytest.fixture(scope="module")
def client(monkeypatch_module, lipsync_corpus) -> TestClient:
monkeypatch_module.setattr(server, "CORPORA", {"lipsync": LIPSYNC_REPO})
monkeypatch_module.setattr(server, "DATA_DIR", CACHE_DIR)
app = server.create_app()
with TestClient(app) as test_client: # runs lifespan: sync + load + index
yield test_client
def test_tools_are_listed(client: TestClient) -> None:
resp = client.post("/lipsync/mcp", json=rpc("tools/list", {}), headers=MCP_HEADERS)
assert resp.status_code == 200
tools = {t["name"] for t in resp.json()["result"]["tools"]}
assert tools == {"search_papers", "get_paper", "get_citations", "list_recent"}
def test_search_papers_tool(client: TestClient) -> None:
resp = client.post(
"/lipsync/mcp",
json=rpc(
"tools/call",
{"name": "search_papers", "arguments": {"query": "latent diffusion lip sync SyncNet"}},
),
headers=MCP_HEADERS,
)
text = resp.json()["result"]["content"][0]["text"]
assert "2412.09262" in text
def test_get_paper_tool(client: TestClient) -> None:
resp = client.post(
"/lipsync/mcp",
json=rpc("tools/call", {"name": "get_paper", "arguments": {"paper_id": "2412.09262"}}),
headers=MCP_HEADERS,
)
text = resp.json()["result"]["content"][0]["text"]
assert "LatentSync" in text and len(text) > 5000
def test_unknown_paper_id_is_an_error(client: TestClient) -> None:
resp = client.post(
"/lipsync/mcp",
json=rpc("tools/call", {"name": "get_paper", "arguments": {"paper_id": "0000.00000"}}),
headers=MCP_HEADERS,
)
assert resp.json()["result"]["isError"] is True
def test_blank_query_is_an_error(client: TestClient) -> None:
resp = client.post(
"/lipsync/mcp",
json=rpc("tools/call", {"name": "search_papers", "arguments": {"query": " "}}),
headers=MCP_HEADERS,
)
assert resp.json()["result"]["isError"] is True
def test_get_citations_tool(client: TestClient) -> None:
resp = client.post(
"/lipsync/mcp",
json=rpc("tools/call", {"name": "get_citations", "arguments": {"paper_id": "2412.09262"}}),
headers=MCP_HEADERS,
)
text = resp.json()["result"]["content"][0]["text"]
assert "Cites" in text and "Cited by" in text
def test_list_recent_tool(client: TestClient) -> None:
import re
from datetime import date, timedelta
resp = client.post(
"/lipsync/mcp",
json=rpc("tools/call", {"name": "list_recent", "arguments": {"days": 365}}),
headers=MCP_HEADERS,
)
text = resp.json()["result"]["content"][0]["text"]
dates = re.findall(r", (\d{4}-\d{2}-\d{2})\)", text)
assert len(dates) > 5
assert dates == sorted(dates, reverse=True) # newest first
assert min(dates) >= (date.today() - timedelta(days=365)).isoformat()
def test_list_recent_days_over_cap_is_an_error(client: TestClient) -> None:
resp = client.post(
"/lipsync/mcp",
json=rpc("tools/call", {"name": "list_recent", "arguments": {"days": 366}}),
headers=MCP_HEADERS,
)
assert resp.json()["result"]["isError"] is True
|