"""Tests for the MCP query tools."""
from __future__ import annotations
import asyncio
from datetime import datetime
import unittest
import mcp_server
def _sample_rows():
base_ts = datetime(2024, 1, 1, 12, 0, 0)
return [
{
"run_id": "run_01",
"ts": base_ts,
"event_type": "DamageDone",
"skill_name": "Curse Explosion",
"damage": 5000,
"is_crit": True,
"is_heavy": False,
"source_name": "Player",
"target_name": "TargetA",
"hit_type": "Normal",
},
{
"run_id": "run_01",
"ts": base_ts,
"event_type": "DamageDone",
"skill_name": "Shadow Burst",
"damage": 2500,
"is_crit": False,
"is_heavy": False,
"source_name": "Player",
"target_name": "TargetA",
"hit_type": "Normal",
},
{
"run_id": "run_02",
"ts": base_ts,
"event_type": "DamageDone",
"skill_name": "Valiant Brawl",
"damage": 3500,
"is_crit": False,
"is_heavy": False,
"source_name": "Player",
"target_name": "TargetB",
"hit_type": "Normal",
},
]
class QueryToolTests(unittest.TestCase):
def test_query_dps_client_parsing(self) -> None:
from app.mcp_client import MCPAnalyzerClient
client = MCPAnalyzerClient(inprocess=True)
sql = "SELECT skill_name, SUM(damage) AS total_damage FROM events GROUP BY 1 ORDER BY total_damage DESC"
payload = client.query(sql)
self.assertIsInstance(payload, dict)
self.assertIn("columns", payload)
self.assertIn("rows", payload)
def setUp(self) -> None:
mcp_server.EVENT_STORE.refresh(_sample_rows())
def test_get_events_schema_has_expected_columns(self) -> None:
schema = asyncio.run(mcp_server.get_events_schema())
column_names = [column["name"] for column in schema["columns"]]
self.assertIn("skill_name", column_names)
self.assertIn("damage", column_names)
def test_query_dps_returns_rows(self) -> None:
sql = "SELECT skill_name, SUM(damage) AS total_damage FROM events GROUP BY 1 ORDER BY total_damage DESC"
result = asyncio.run(mcp_server.query_dps(sql))
self.assertEqual(result["columns"], ["skill_name", "total_damage"])
self.assertEqual(len(result["rows"]), 3)
self.assertEqual(result["rows"][0][0], "Curse Explosion")
def test_non_select_query_is_rejected(self) -> None:
with self.assertRaises(ValueError):
asyncio.run(mcp_server.query_dps("DELETE FROM events"))
def test_get_analysis_packet_schema(self) -> None:
packet = asyncio.run(mcp_server.get_analysis_packet())
for key in [
"meta",
"run_summary",
"runs_last_n",
"top_skills",
"skill_efficiency",
"timeline",
"notes",
]:
self.assertIn(key, packet)
self.assertIsInstance(packet.get("notes"), list)
def test_get_analysis_packet_clamps_bucket_seconds(self) -> None:
packet = asyncio.run(mcp_server.get_analysis_packet(bucket_seconds=7))
self.assertEqual(packet["meta"]["limits"]["bucket_seconds"], 5)
if __name__ == "__main__":
unittest.main()
