62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
|
"""Smoke tests for the MCP bridge HTTP surface (no cluster required)."""
|
||
|
|
|
||
|
|
from fastapi.testclient import TestClient
|
||
|
|
|
||
|
|
from app.main import PROMPTS, RESOURCES, TOOLS, app
|
||
|
|
|
||
|
|
client = TestClient(app)
|
||
|
|
|
||
|
|
EXPECTED_TOOL_NAMES = {
|
||
|
|
"promql.query",
|
||
|
|
"loki.query",
|
||
|
|
"k8s.get",
|
||
|
|
"k8s.events",
|
||
|
|
"inventory.snapshot",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_healthz_returns_ok():
|
||
|
|
response = client.get("/healthz")
|
||
|
|
assert response.status_code == 200
|
||
|
|
body = response.json()
|
||
|
|
assert body["status"] == "ok"
|
||
|
|
assert isinstance(body["ts"], int)
|
||
|
|
|
||
|
|
|
||
|
|
def test_mcp_schema_exposes_resources_tools_and_prompts():
|
||
|
|
response = client.get("/mcp/schema")
|
||
|
|
assert response.status_code == 200
|
||
|
|
body = response.json()
|
||
|
|
|
||
|
|
assert "resources" in body
|
||
|
|
assert "tools" in body
|
||
|
|
assert "prompts" in body
|
||
|
|
|
||
|
|
assert len(body["resources"]) == len(RESOURCES)
|
||
|
|
assert len(body["tools"]) == len(TOOLS)
|
||
|
|
assert len(body["prompts"]) == len(PROMPTS)
|
||
|
|
|
||
|
|
tool_names = {tool["name"] for tool in body["tools"]}
|
||
|
|
assert tool_names == EXPECTED_TOOL_NAMES
|
||
|
|
|
||
|
|
for tool in body["tools"]:
|
||
|
|
assert "inputSchema" in tool
|
||
|
|
assert tool["inputSchema"]["type"] == "object"
|
||
|
|
|
||
|
|
|
||
|
|
def test_mcp_resource_returns_saved_query():
|
||
|
|
uri = "res://dashboards/top-pods-by-cpu.promql"
|
||
|
|
response = client.get("/mcp/resource", params={"uri": uri})
|
||
|
|
assert response.status_code == 200
|
||
|
|
body = response.json()
|
||
|
|
assert body["uri"] == uri
|
||
|
|
assert body["mimeType"] == "text/plain"
|
||
|
|
assert "container_cpu_usage_seconds_total" in body["content"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_mcp_resource_unknown_uri():
|
||
|
|
response = client.get("/mcp/resource", params={"uri": "res://does-not-exist"})
|
||
|
|
assert response.status_code == 200
|
||
|
|
body = response.json()
|
||
|
|
assert body["error"] == "not found"
|
||
|
|
assert body["uri"] == "res://does-not-exist"
|