tegwick
8f2584c1a0
Introduce pytest smoke tests, run/verify scripts, and Makefile targets so the bridge can be developed and validated without a full cluster deploy. Document the local workflow and agent quickstart in README.
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" |