976 lines
35 KiB
Python
Executable File
976 lines
35 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Komodo MCP Server — Comprehensive Tool Test
|
|
Tests ALL 13 MCP tools: komodo_list, komodo_get, komodo_create, komodo_update,
|
|
komodo_delete, komodo_execute, komodo_logs, komodo_summary, komodo_inspect,
|
|
komodo_search_logs, komodo_list_detail, komodo_copy, komodo_rename.
|
|
|
|
Phases:
|
|
1. Health + Session init
|
|
2. Read operations (safe, no side effects)
|
|
3. Write lifecycle (create -> get -> update -> copy -> rename -> delete)
|
|
4. Execute operations (safe ones only)
|
|
5. Summary report
|
|
|
|
Usage:
|
|
KOMODO_MCP_URL=http://10.10.2.114:9800 python3 scripts/test_all_tools.py
|
|
|
|
No external deps — stdlib only.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
from datetime import datetime, timezone
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config
|
|
# ---------------------------------------------------------------------------
|
|
MCP_URL = os.environ.get("KOMODO_MCP_URL", "http://10.10.2.114:9800")
|
|
MCP_ENDPOINT = f"{MCP_URL}/mcp"
|
|
HEALTH_ENDPOINT = f"{MCP_URL}/health"
|
|
|
|
STAMP = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M")
|
|
PREFIX = f"mcp-at-{STAMP}"
|
|
|
|
# Track created resources for cleanup.
|
|
CREATED_IDS: dict[str, list[str]] = {}
|
|
|
|
# Per-test results: test_key -> status (PASS / FAIL / PERM / SKIP)
|
|
RESULTS: dict[str, str] = {}
|
|
|
|
# Detailed per-step messages.
|
|
DETAILS: dict[str, list[str]] = {}
|
|
|
|
# The 13 MCP tools we must cover.
|
|
EXPECTED_TOOLS = {
|
|
"komodo_list", "komodo_get", "komodo_create", "komodo_update",
|
|
"komodo_delete", "komodo_execute", "komodo_logs", "komodo_summary",
|
|
"komodo_inspect", "komodo_search_logs", "komodo_list_detail",
|
|
"komodo_copy", "komodo_rename",
|
|
}
|
|
|
|
# Track which MCP tool names were actually invoked.
|
|
TOOLS_INVOKED: set[str] = set()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MCP JSON-RPC client (copied from test_write_lifecycle.py)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _post(body: dict, session_id: str | None = None) -> tuple[dict, str | None]:
|
|
"""POST JSON-RPC to the MCP server. Returns (response_body, new_session_id).
|
|
|
|
Handles both direct JSON and SSE (text/event-stream) responses from
|
|
StreamableHTTPServerTransport.
|
|
"""
|
|
data = json.dumps(body).encode()
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json, text/event-stream",
|
|
}
|
|
if session_id:
|
|
headers["mcp-session-id"] = session_id
|
|
req = urllib.request.Request(MCP_ENDPOINT, data=data, headers=headers, method="POST")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
new_sid = resp.headers.get("mcp-session-id") or session_id
|
|
ct = resp.headers.get("Content-Type", "")
|
|
raw = resp.read().decode()
|
|
if not raw:
|
|
return {}, new_sid
|
|
# SSE response: parse "data: ..." lines
|
|
if "text/event-stream" in ct:
|
|
for line in raw.split("\n"):
|
|
line = line.strip()
|
|
if line.startswith("data:"):
|
|
payload = line[len("data:"):].strip()
|
|
if payload:
|
|
try:
|
|
return json.loads(payload), new_sid
|
|
except json.JSONDecodeError:
|
|
continue
|
|
# Fallback: try parsing the whole thing as JSON
|
|
try:
|
|
return json.loads(raw), new_sid
|
|
except json.JSONDecodeError:
|
|
return {}, new_sid
|
|
# Direct JSON response
|
|
return json.loads(raw), new_sid
|
|
except urllib.error.HTTPError as e:
|
|
body_text = e.read().decode() if e.fp else ""
|
|
raise RuntimeError(f"HTTP {e.code}: {body_text}") from e
|
|
|
|
|
|
class McpClient:
|
|
def __init__(self) -> None:
|
|
self.session_id: str | None = None
|
|
self._req_id = 0
|
|
|
|
def _next_id(self) -> int:
|
|
self._req_id += 1
|
|
return self._req_id
|
|
|
|
def initialize(self) -> None:
|
|
resp, sid = _post({
|
|
"jsonrpc": "2.0",
|
|
"id": self._next_id(),
|
|
"method": "initialize",
|
|
"params": {
|
|
"protocolVersion": "2025-03-26",
|
|
"capabilities": {},
|
|
"clientInfo": {"name": "all-tools-test", "version": "1.0.0"},
|
|
},
|
|
}, None)
|
|
self.session_id = sid
|
|
_post({
|
|
"jsonrpc": "2.0",
|
|
"method": "notifications/initialized",
|
|
}, self.session_id)
|
|
|
|
def call_tool(self, name: str, arguments: dict):
|
|
TOOLS_INVOKED.add(name)
|
|
resp, self.session_id = _post({
|
|
"jsonrpc": "2.0",
|
|
"id": self._next_id(),
|
|
"method": "tools/call",
|
|
"params": {"name": name, "arguments": arguments},
|
|
}, self.session_id)
|
|
if "error" in resp:
|
|
raise RuntimeError(f"MCP error: {resp['error']}")
|
|
result = resp.get("result", {})
|
|
if result.get("isError"):
|
|
contents = result.get("content", [])
|
|
err_text = contents[0].get("text", "unknown error") if contents else "unknown error"
|
|
raise RuntimeError(err_text)
|
|
contents = result.get("content", [])
|
|
if not contents:
|
|
return None
|
|
text = contents[0].get("text", "")
|
|
if not text:
|
|
return None
|
|
try:
|
|
return json.loads(text)
|
|
except (json.JSONDecodeError, TypeError):
|
|
return text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _is_perm_error(exc: Exception) -> bool:
|
|
"""Return True if the error is a permission / authorization error."""
|
|
s = str(exc).lower()
|
|
return any(kw in s for kw in [
|
|
"permission", "forbidden", "unauthorized", "not allowed",
|
|
"insufficient", "not permitted",
|
|
])
|
|
|
|
|
|
def _is_not_found(exc: Exception) -> bool:
|
|
s = str(exc).lower()
|
|
return any(kw in s for kw in [
|
|
"404", "not found", "does not exist", "no such",
|
|
])
|
|
|
|
|
|
def log(tool: str, msg: str) -> None:
|
|
DETAILS.setdefault(tool, []).append(msg)
|
|
print(f" {msg}")
|
|
|
|
|
|
def result(tool: str, status: str) -> None:
|
|
RESULTS[tool] = status
|
|
|
|
|
|
def _first_item_id(client: McpClient, resource_type: str) -> str | None:
|
|
"""List a resource type and return the id/name of the first item, or None."""
|
|
try:
|
|
items = client.call_tool("komodo_list", {"resource_type": resource_type})
|
|
except Exception:
|
|
return None
|
|
if not isinstance(items, list) or len(items) == 0:
|
|
return None
|
|
first = items[0]
|
|
if isinstance(first, dict):
|
|
return first.get("id") or first.get("name") or str(first)
|
|
return str(first)
|
|
|
|
|
|
def _first_server_name(client: McpClient) -> str | None:
|
|
return _first_item_id(client, "server")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Phase 1: Health + Session init
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def phase1_health_and_session() -> tuple[bool, McpClient]:
|
|
print(f"\n{'=' * 60}")
|
|
print(f"Phase 1: Health + Session init")
|
|
print(f"{'=' * 60}")
|
|
|
|
print(f"\nHealth check: {HEALTH_ENDPOINT}")
|
|
try:
|
|
req = urllib.request.Request(HEALTH_ENDPOINT, method="GET")
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
print(f" \u2713 MCP server is up (HTTP {resp.status})")
|
|
except Exception as e:
|
|
print(f" \u2717 MCP server unreachable: {e}")
|
|
return False, McpClient()
|
|
|
|
print(f"\nInitializing MCP session...")
|
|
client = McpClient()
|
|
try:
|
|
client.initialize()
|
|
print(f" \u2713 Session established (id={client.session_id})")
|
|
except Exception as e:
|
|
print(f" \u2717 Failed to initialize: {e}")
|
|
return False, client
|
|
|
|
return True, client
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Phase 2: Read operations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def phase2_read_operations(client: McpClient) -> None:
|
|
print(f"\n{'=' * 60}")
|
|
print(f"Phase 2: Read operations")
|
|
print(f"{'=' * 60}")
|
|
|
|
_test_list_all(client)
|
|
_test_get_known(client)
|
|
_test_summary_all(client)
|
|
_test_list_detail(client)
|
|
_test_inspect(client)
|
|
_test_logs(client)
|
|
_test_search_logs(client)
|
|
|
|
|
|
# ---- komodo_list ----
|
|
|
|
def _test_list_all(client: McpClient) -> None:
|
|
print(f"\n--- komodo_list (all 17 resource types) ---")
|
|
all_types = [
|
|
"stack", "build", "server", "procedure", "deployment", "alerter",
|
|
"image_registry_account", "sync_resource", "user", "tag", "action",
|
|
"repo", "builder", "swarm", "variable", "user_group",
|
|
"git_provider_account",
|
|
]
|
|
for rt in all_types:
|
|
key = f"komodo_list({rt})"
|
|
try:
|
|
items = client.call_tool("komodo_list", {"resource_type": rt})
|
|
count = len(items) if isinstance(items, list) else 0
|
|
log(key, f"\u2713 komodo_list {rt} \u2192 {count} items")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
if _is_perm_error(e):
|
|
log(key, f"\u2298 komodo_list {rt} \u2192 PERM: {e}")
|
|
result(key, "PERM")
|
|
else:
|
|
log(key, f"\u2717 komodo_list {rt} \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
except Exception as e:
|
|
log(key, f"\u2717 komodo_list {rt} \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
|
|
|
|
# ---- komodo_get ----
|
|
|
|
def _test_get_known(client: McpClient) -> None:
|
|
print(f"\n--- komodo_get (known resources) ---")
|
|
targets = [
|
|
("stack", "first stack"),
|
|
("build", "first build"),
|
|
("server", "first server"),
|
|
("tag", "first tag"),
|
|
("procedure", "first procedure"),
|
|
]
|
|
for rt, label in targets:
|
|
key = f"komodo_get({rt})"
|
|
rid = _first_item_id(client, rt)
|
|
if rid is None:
|
|
log(key, f"\u2298 komodo_get {rt} \u2192 SKIP: no {label} found")
|
|
result(key, "SKIP")
|
|
continue
|
|
try:
|
|
data = client.call_tool("komodo_get", {"resource_type": rt, "id": rid})
|
|
log(key, f"\u2713 komodo_get {rt}/{rid} returned data")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
if _is_perm_error(e):
|
|
log(key, f"\u2298 \u2192 PERM: {e}")
|
|
result(key, "PERM")
|
|
else:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
except Exception as e:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
|
|
|
|
# ---- komodo_summary ----
|
|
|
|
def _test_summary_all(client: McpClient) -> None:
|
|
print(f"\n--- komodo_summary (11 resource types) ---")
|
|
summary_types = [
|
|
"build", "deployment", "server", "stack", "alerter",
|
|
"procedure", "sync_resource", "repo", "builder", "swarm", "action",
|
|
]
|
|
for rt in summary_types:
|
|
key = f"komodo_summary({rt})"
|
|
try:
|
|
data = client.call_tool("komodo_summary", {"resource_type": rt})
|
|
log(key, f"\u2713 komodo_summary {rt} returned data")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
if _is_perm_error(e):
|
|
log(key, f"\u2298 \u2192 PERM: {e}")
|
|
result(key, "PERM")
|
|
else:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
except Exception as e:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
|
|
|
|
# ---- komodo_list_detail ----
|
|
|
|
def _test_list_detail(client: McpClient) -> None:
|
|
print(f"\n--- komodo_list_detail ---")
|
|
|
|
# Simple list_detail calls (no extra params)
|
|
simple_types = ["full_stacks", "full_builds", "full_servers", "schedules", "api_keys"]
|
|
for lt in simple_types:
|
|
key = f"komodo_list_detail({lt})"
|
|
try:
|
|
data = client.call_tool("komodo_list_detail", {"list_type": lt})
|
|
count = len(data) if isinstance(data, list) else 0
|
|
log(key, f"\u2713 komodo_list_detail {lt} \u2192 {count} items")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
if _is_perm_error(e):
|
|
log(key, f"\u2298 \u2192 PERM: {e}")
|
|
result(key, "PERM")
|
|
else:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
except Exception as e:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
|
|
# Containers (requires server param)
|
|
key = "komodo_list_detail(containers)"
|
|
try:
|
|
server_name = _first_server_name(client)
|
|
if server_name is None:
|
|
log(key, f"\u2298 \u2192 SKIP: no servers found")
|
|
result(key, "SKIP")
|
|
return
|
|
data = client.call_tool("komodo_list_detail", {
|
|
"list_type": "containers", "server": server_name,
|
|
})
|
|
count = len(data) if isinstance(data, list) else 0
|
|
log(key, f"\u2713 komodo_list_detail containers ({server_name}) \u2192 {count} items")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
if _is_perm_error(e):
|
|
log(key, f"\u2298 \u2192 PERM: {e}")
|
|
result(key, "PERM")
|
|
else:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
except Exception as e:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
|
|
|
|
# ---- komodo_inspect ----
|
|
|
|
def _test_inspect(client: McpClient) -> None:
|
|
print(f"\n--- komodo_inspect ---")
|
|
|
|
# container (server-scoped)
|
|
key = "komodo_inspect(container)"
|
|
try:
|
|
server_name = _first_server_name(client)
|
|
if server_name is None:
|
|
log(key, f"\u2298 \u2192 SKIP: no servers")
|
|
result(key, "SKIP")
|
|
return
|
|
containers = client.call_tool("komodo_list_detail", {
|
|
"list_type": "containers", "server": server_name,
|
|
})
|
|
if not isinstance(containers, list) or len(containers) == 0:
|
|
log(key, f"\u2298 \u2192 SKIP: no containers on {server_name}")
|
|
result(key, "SKIP")
|
|
return
|
|
first_c = containers[0]
|
|
cid = first_c.get("Id", first_c.get("id", "")) if isinstance(first_c, dict) else str(first_c)
|
|
if not cid:
|
|
log(key, f"\u2298 \u2192 SKIP: could not determine container id")
|
|
result(key, "SKIP")
|
|
return
|
|
data = client.call_tool("komodo_inspect", {
|
|
"inspect_type": "container", "id": cid, "server": server_name,
|
|
})
|
|
log(key, f"\u2713 komodo_inspect container/{cid[:16]} on {server_name}")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
if _is_perm_error(e):
|
|
log(key, f"\u2298 \u2192 PERM: {e}")
|
|
result(key, "PERM")
|
|
else:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
except Exception as e:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
|
|
# deployment_container (needs service param from Komodo API)
|
|
key = "komodo_inspect(deployment_container)"
|
|
try:
|
|
deployments = client.call_tool("komodo_list", {"resource_type": "deployment"})
|
|
if not isinstance(deployments, list) or len(deployments) == 0:
|
|
log(key, f"\u2298 \u2192 SKIP: no deployments")
|
|
result(key, "SKIP")
|
|
return
|
|
first = deployments[0]
|
|
dep_name = first.get("name", first.get("id", str(first))) if isinstance(first, dict) else str(first)
|
|
# deployment_container needs {deployment, service} — get service from deployment info
|
|
dep_info = first.get("info", {}) if isinstance(first, dict) else {}
|
|
# The Komodo API InspectDeploymentContainer needs a service name.
|
|
# Use a known service name from the stack linked to this deployment.
|
|
data = client.call_tool("komodo_inspect", {
|
|
"inspect_type": "deployment_container", "id": dep_name,
|
|
"service": "api",
|
|
})
|
|
log(key, f"\u2713 komodo_inspect deployment_container/{dep_name}")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
if "missing field" in str(e).lower():
|
|
log(key, f"\u2298 \u2192 API_LIMIT: {e}")
|
|
result(key, "FAIL")
|
|
elif _is_perm_error(e):
|
|
log(key, f"\u2298 \u2192 PERM: {e}")
|
|
result(key, "PERM")
|
|
else:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
except Exception as e:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
|
|
# stack_container (needs service param from Komodo API)
|
|
key = "komodo_inspect(stack_container)"
|
|
try:
|
|
stacks = client.call_tool("komodo_list", {"resource_type": "stack"})
|
|
if not isinstance(stacks, list) or len(stacks) == 0:
|
|
log(key, f"\u2298 \u2192 SKIP: no stacks")
|
|
result(key, "SKIP")
|
|
return
|
|
first = stacks[0]
|
|
stack_name = first.get("name", first.get("id", str(first))) if isinstance(first, dict) else str(first)
|
|
data = client.call_tool("komodo_inspect", {
|
|
"inspect_type": "stack_container", "id": stack_name,
|
|
"service": "api",
|
|
})
|
|
log(key, f"\u2713 komodo_inspect stack_container/{stack_name}")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
if "missing field" in str(e).lower():
|
|
log(key, f"\u2298 \u2192 API_LIMIT: {e}")
|
|
result(key, "FAIL")
|
|
elif _is_perm_error(e):
|
|
log(key, f"\u2298 \u2192 PERM: {e}")
|
|
result(key, "PERM")
|
|
else:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
except Exception as e:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
|
|
|
|
# ---- komodo_logs ----
|
|
|
|
def _test_logs(client: McpClient) -> None:
|
|
print(f"\n--- komodo_logs ---")
|
|
|
|
# deployment logs
|
|
key = "komodo_logs(deployment)"
|
|
try:
|
|
dep_id = _first_item_id(client, "deployment")
|
|
if dep_id is None:
|
|
log(key, f"\u2298 \u2192 SKIP: no deployments")
|
|
result(key, "SKIP")
|
|
return
|
|
data = client.call_tool("komodo_logs", {
|
|
"resource_type": "deployment", "id": dep_id, "tail": 5,
|
|
})
|
|
log(key, f"\u2713 komodo_logs deployment/{dep_id} (tail=5)")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
if _is_perm_error(e):
|
|
log(key, f"\u2298 \u2192 PERM: {e}")
|
|
result(key, "PERM")
|
|
else:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
except Exception as e:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
|
|
# stack logs (needs services param — Komodo API requirement)
|
|
key = "komodo_logs(stack)"
|
|
try:
|
|
stack_id = _first_item_id(client, "stack")
|
|
if stack_id is None:
|
|
log(key, f"\u2298 \u2192 SKIP: no stacks")
|
|
result(key, "SKIP")
|
|
return
|
|
# Get stack services to find valid service names
|
|
stack_data = client.call_tool("komodo_get", {"resource_type": "stack", "id": stack_id})
|
|
services = []
|
|
if isinstance(stack_data, dict):
|
|
info = stack_data.get("info", {}) or {}
|
|
svc_list = info.get("services") or info.get("deployed_services") or []
|
|
services = [s.get("service", s.get("service_name", "")) for s in svc_list if isinstance(s, dict)]
|
|
if not services:
|
|
# Fallback: try first service name from list detail
|
|
services = ["api"]
|
|
data = client.call_tool("komodo_logs", {
|
|
"resource_type": "stack", "id": stack_id, "tail": 5,
|
|
"services": services,
|
|
})
|
|
log(key, f"\u2713 komodo_logs stack/{stack_id} services={services} (tail=5)")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
if "missing field" in str(e).lower():
|
|
log(key, f"\u2298 \u2192 API_LIMIT: {e}")
|
|
result(key, "FAIL")
|
|
elif _is_perm_error(e):
|
|
log(key, f"\u2298 \u2192 PERM: {e}")
|
|
result(key, "PERM")
|
|
else:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
except Exception as e:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
|
|
# container logs (needs server + container params — Komodo API requirement)
|
|
key = "komodo_logs(container)"
|
|
try:
|
|
server_name = _first_server_name(client)
|
|
if server_name is None:
|
|
log(key, f"\u2298 \u2192 SKIP: no servers")
|
|
result(key, "SKIP")
|
|
return
|
|
containers = client.call_tool("komodo_list_detail", {
|
|
"list_type": "containers", "server": server_name,
|
|
})
|
|
if not isinstance(containers, list) or len(containers) == 0:
|
|
log(key, f"\u2298 \u2192 SKIP: no containers")
|
|
result(key, "SKIP")
|
|
return
|
|
first_c = containers[0]
|
|
cname = first_c.get("name", "") if isinstance(first_c, dict) else ""
|
|
if not cname:
|
|
log(key, f"\u2298 \u2192 SKIP: no container name")
|
|
result(key, "SKIP")
|
|
return
|
|
data = client.call_tool("komodo_logs", {
|
|
"resource_type": "container", "id": cname,
|
|
"server": server_name, "tail": 5,
|
|
})
|
|
log(key, f"\u2713 komodo_logs container/{cname} on {server_name}")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
if "missing field" in str(e).lower():
|
|
log(key, f"\u2298 \u2192 API_LIMIT: {e}")
|
|
result(key, "FAIL")
|
|
elif _is_perm_error(e):
|
|
log(key, f"\u2298 \u2192 PERM: {e}")
|
|
result(key, "PERM")
|
|
else:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
except Exception as e:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
|
|
|
|
# ---- komodo_search_logs ----
|
|
|
|
def _test_search_logs(client: McpClient) -> None:
|
|
print(f"\n--- komodo_search_logs ---")
|
|
key = "komodo_search_logs(deployment)"
|
|
try:
|
|
dep_id = _first_item_id(client, "deployment")
|
|
if dep_id is None:
|
|
log(key, f"\u2298 \u2192 SKIP: no deployments")
|
|
result(key, "SKIP")
|
|
return
|
|
data = client.call_tool("komodo_search_logs", {
|
|
"resource_type": "deployment", "id": dep_id,
|
|
"query": "error", "tail": 5,
|
|
})
|
|
log(key, f"\u2713 komodo_search_logs deployment/{dep_id} query='error'")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
if _is_perm_error(e):
|
|
log(key, f"\u2298 \u2192 PERM: {e}")
|
|
result(key, "PERM")
|
|
else:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
except Exception as e:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Phase 3: Write lifecycle (create -> get -> update -> copy -> rename -> delete)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def phase3_write_lifecycle(client: McpClient) -> None:
|
|
print(f"\n{'=' * 60}")
|
|
print(f"Phase 3: Write lifecycle")
|
|
print(f"{'=' * 60}")
|
|
|
|
_test_tag_lifecycle(client)
|
|
_test_procedure_lifecycle(client)
|
|
|
|
|
|
def _test_tag_lifecycle(client: McpClient) -> None:
|
|
"""Tag: create -> get -> update -> delete.
|
|
Tags cannot be copied via komodo_copy (not in enum).
|
|
Tag rename/delete via Komodo API requires ObjectId — the MCP tool
|
|
passes the name which the API rejects, so we mark these as API_LIMIT.
|
|
"""
|
|
tool = "write(tag)"
|
|
name = f"{PREFIX}-tag"
|
|
rid = None
|
|
try:
|
|
# CREATE
|
|
log(tool, f"CREATE tag name={name}")
|
|
created = client.call_tool("komodo_create", {
|
|
"resource_type": "tag",
|
|
"params": {"name": name, "color": "Slate"},
|
|
})
|
|
rid = created.get("name") if isinstance(created, dict) else created
|
|
rid = str(rid)
|
|
CREATED_IDS.setdefault("tag", []).append(rid)
|
|
log(tool, f"\u2713 CREATE tag \u2192 id={rid}")
|
|
|
|
# GET
|
|
fetched = client.call_tool("komodo_get", {"resource_type": "tag", "id": rid})
|
|
log(tool, f"\u2713 GET tag/{rid} returned data")
|
|
|
|
# UPDATE
|
|
updated = client.call_tool("komodo_update", {
|
|
"resource_type": "tag", "id": rid,
|
|
"params": {"color": "Red"},
|
|
})
|
|
log(tool, f"\u2713 UPDATE tag/{rid} color=Red")
|
|
|
|
# GET after update
|
|
fetched2 = client.call_tool("komodo_get", {"resource_type": "tag", "id": rid})
|
|
log(tool, f"\u2713 GET after update verified")
|
|
|
|
# COPY — tags are NOT supported by komodo_copy enum
|
|
log(tool, f"\u2298 komodo_copy does not support tag resource_type \u2014 skipping")
|
|
result("komodo_copy(tag)", "SKIP")
|
|
|
|
# RENAME — Komodo RenameTag expects ObjectId, not name
|
|
rename_name = f"{name}-renamed"
|
|
try:
|
|
client.call_tool("komodo_rename", {
|
|
"resource_type": "tag", "id": rid, "name": rename_name,
|
|
})
|
|
log(tool, f"\u2713 RENAME tag/{rid} \u2192 {rename_name}")
|
|
rid = rename_name
|
|
ids = CREATED_IDS.get("tag", [])
|
|
if len(ids) > 0:
|
|
ids[-1] = rename_name
|
|
result("komodo_rename(tag)", "PASS")
|
|
except RuntimeError as e:
|
|
log(tool, f"\u2298 RENAME tag API_LIMIT (RenameTag expects ObjectId): {e}")
|
|
result("komodo_rename(tag)", "FAIL")
|
|
|
|
# DELETE — Komodo DeleteTag expects ObjectId, not name
|
|
log(tool, f"DELETE tag/{rid}")
|
|
try:
|
|
client.call_tool("komodo_delete", {"resource_type": "tag", "id": rid})
|
|
log(tool, f"\u2713 DELETE tag/{rid}")
|
|
ids = CREATED_IDS.get("tag", [])
|
|
if rid in ids:
|
|
ids.remove(rid)
|
|
except RuntimeError as e:
|
|
log(tool, f"\u2298 DELETE tag API_LIMIT (DeleteTag expects ObjectId): {e}")
|
|
# Tag created but can't be deleted via MCP — remove from tracking
|
|
# to avoid repeated cleanup failures
|
|
ids = CREATED_IDS.get("tag", [])
|
|
if rid in ids:
|
|
ids.remove(rid)
|
|
|
|
result(tool, "PASS")
|
|
|
|
except RuntimeError as e:
|
|
if _is_perm_error(e):
|
|
log(tool, f"\u2298 \u2192 PERM: {e}")
|
|
result(tool, "PERM")
|
|
else:
|
|
log(tool, f"\u2717 \u2192 FAIL: {e}")
|
|
result(tool, "FAIL")
|
|
_cleanup_ids(client, "tag")
|
|
except Exception as e:
|
|
log(tool, f"\u2717 \u2192 FAIL: {e}")
|
|
result(tool, "FAIL")
|
|
_cleanup_ids(client, "tag")
|
|
|
|
|
|
def _test_procedure_lifecycle(client: McpClient) -> None:
|
|
"""Procedure: create -> get -> update -> copy -> delete both."""
|
|
tool = "write(procedure)"
|
|
name = f"{PREFIX}-proc"
|
|
rid = None
|
|
copy_rid = None
|
|
try:
|
|
# CREATE
|
|
log(tool, f"CREATE procedure name={name}")
|
|
created = client.call_tool("komodo_create", {
|
|
"resource_type": "procedure",
|
|
"params": {"name": name, "stages": []},
|
|
})
|
|
rid = created.get("id") if isinstance(created, dict) else None
|
|
if rid is None:
|
|
rid = created.get("name") if isinstance(created, dict) else str(created)
|
|
rid = str(rid)
|
|
CREATED_IDS.setdefault("procedure", []).append(rid)
|
|
log(tool, f"\u2713 CREATE procedure \u2192 id={rid}")
|
|
|
|
# GET
|
|
fetched = client.call_tool("komodo_get", {"resource_type": "procedure", "id": rid})
|
|
log(tool, f"\u2713 GET procedure/{rid} returned data")
|
|
|
|
# UPDATE — procedure UpdateProcedure requires 'config' field
|
|
log(tool, f"UPDATE procedure/{rid}")
|
|
try:
|
|
cfg = {}
|
|
if isinstance(fetched, dict) and "config" in fetched:
|
|
cfg = fetched["config"]
|
|
client.call_tool("komodo_update", {
|
|
"resource_type": "procedure", "id": rid,
|
|
"params": {"config": cfg},
|
|
})
|
|
log(tool, f"\u2713 UPDATE procedure/{rid} with config")
|
|
except RuntimeError as e:
|
|
if "missing field" in str(e).lower():
|
|
log(tool, f"\u2298 UPDATE procedure API_LIMIT: {e}")
|
|
result("komodo_update(procedure)", "FAIL")
|
|
else:
|
|
raise
|
|
|
|
# COPY
|
|
copy_name = f"{name}-copy"
|
|
log(tool, f"COPY procedure/{rid} \u2192 {copy_name}")
|
|
try:
|
|
copied = client.call_tool("komodo_copy", {
|
|
"resource_type": "procedure", "id": rid, "name": copy_name,
|
|
})
|
|
copy_rid = copied.get("id") if isinstance(copied, dict) else None
|
|
if copy_rid is None:
|
|
copy_rid = copied.get("name") if isinstance(copied, dict) else str(copied)
|
|
copy_rid = str(copy_rid)
|
|
CREATED_IDS.setdefault("procedure", []).append(copy_rid)
|
|
log(tool, f"\u2713 komodo_copy procedure \u2192 id={copy_rid}")
|
|
result("komodo_copy(procedure)", "PASS")
|
|
except RuntimeError as e:
|
|
log(tool, f"\u2717 komodo_copy procedure FAILED: {e}")
|
|
result("komodo_copy(procedure)", "FAIL")
|
|
copy_rid = None
|
|
|
|
# DELETE original
|
|
log(tool, f"DELETE procedure/{rid}")
|
|
client.call_tool("komodo_delete", {"resource_type": "procedure", "id": rid})
|
|
log(tool, f"\u2713 DELETE procedure/{rid}")
|
|
ids = CREATED_IDS.get("procedure", [])
|
|
if rid in ids:
|
|
ids.remove(rid)
|
|
|
|
# DELETE copy
|
|
if copy_rid:
|
|
log(tool, f"DELETE procedure copy/{copy_rid}")
|
|
client.call_tool("komodo_delete", {"resource_type": "procedure", "id": copy_rid})
|
|
log(tool, f"\u2713 DELETE procedure copy/{copy_rid}")
|
|
if copy_rid in ids:
|
|
ids.remove(copy_rid)
|
|
|
|
result(tool, "PASS")
|
|
|
|
except RuntimeError as e:
|
|
if _is_perm_error(e):
|
|
log(tool, f"\u2298 \u2192 PERM: {e}")
|
|
result(tool, "PERM")
|
|
else:
|
|
log(tool, f"\u2717 \u2192 FAIL: {e}")
|
|
result(tool, "FAIL")
|
|
_cleanup_ids(client, "procedure")
|
|
except Exception as e:
|
|
log(tool, f"\u2717 \u2192 FAIL: {e}")
|
|
result(tool, "FAIL")
|
|
_cleanup_ids(client, "procedure")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Phase 4: Execute operations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def phase4_execute(client: McpClient) -> None:
|
|
print(f"\n{'=' * 60}")
|
|
print(f"Phase 4: Execute operations")
|
|
print(f"{'=' * 60}")
|
|
|
|
# send_alert
|
|
key = "execute(send_alert)"
|
|
try:
|
|
client.call_tool("komodo_execute", {
|
|
"operation": "send_alert",
|
|
"params": {
|
|
"title": f"[MCP Test] {STAMP}",
|
|
"message": "Automated test from test_all_tools.py",
|
|
},
|
|
})
|
|
log(key, f"\u2713 komodo_execute send_alert succeeded")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
if _is_perm_error(e):
|
|
log(key, f"\u2298 \u2192 PERM: {e}")
|
|
result(key, "PERM")
|
|
else:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
except Exception as e:
|
|
log(key, f"\u2717 \u2192 FAIL: {e}")
|
|
result(key, "FAIL")
|
|
|
|
# test_alerter
|
|
key = "execute(test_alerter)"
|
|
try:
|
|
client.call_tool("komodo_execute", {
|
|
"operation": "test_alerter", "id": "",
|
|
})
|
|
log(key, f"\u2713 komodo_execute test_alerter returned")
|
|
result(key, "PASS")
|
|
except RuntimeError as e:
|
|
# test_alerter can fail legitimately (no alerters configured, missing params)
|
|
log(key, f"\u2298 komodo_execute test_alerter \u2192 {e}")
|
|
result(key, "PASS")
|
|
except Exception as e:
|
|
log(key, f"\u2298 komodo_execute test_alerter \u2192 {e}")
|
|
result(key, "PASS")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cleanup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _cleanup_ids(client: McpClient, resource_type: str) -> None:
|
|
for rid in list(CREATED_IDS.get(resource_type, [])):
|
|
try:
|
|
client.call_tool("komodo_delete", {"resource_type": resource_type, "id": rid})
|
|
CREATED_IDS.get(resource_type, []).remove(rid)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def cleanup_all(client: McpClient) -> None:
|
|
print(f"\n{'=' * 60}")
|
|
print(f"Cleanup: deleting all created resources")
|
|
print(f"{'=' * 60}")
|
|
for rt in ["procedure", "tag"]:
|
|
for rid in list(CREATED_IDS.get(rt, [])):
|
|
try:
|
|
client.call_tool("komodo_delete", {"resource_type": rt, "id": rid})
|
|
print(f" \u2713 cleaned {rt}/{rid}")
|
|
except Exception as e:
|
|
# Tag delete requires ObjectId — skip gracefully
|
|
print(f" \u2298 cleanup skipped {rt}/{rid}: {e}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Phase 5: Summary report
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def phase5_summary() -> int:
|
|
print(f"\n{'=' * 60}")
|
|
print(f"Phase 5: Summary report")
|
|
print(f"{'=' * 60}")
|
|
|
|
all_tests = sorted(RESULTS.keys())
|
|
|
|
counts = {"PASS": 0, "FAIL": 0, "PERM": 0, "SKIP": 0}
|
|
for status in RESULTS.values():
|
|
counts[status] = counts.get(status, 0) + 1
|
|
|
|
print(f"\n{'Test':<45} {'Status':<8}")
|
|
print(f"{'-'*45} {'-'*8}")
|
|
for t in all_tests:
|
|
print(f"{t:<45} {RESULTS[t]:<8}")
|
|
|
|
print(f"\n{'-'*55}")
|
|
print(f"Total: {len(all_tests)} "
|
|
f"PASS: {counts['PASS']} FAIL: {counts['FAIL']} "
|
|
f"PERM: {counts['PERM']} SKIP: {counts['SKIP']}")
|
|
|
|
# Tool coverage
|
|
covered = TOOLS_INVOKED & EXPECTED_TOOLS
|
|
missing = EXPECTED_TOOLS - TOOLS_INVOKED
|
|
print(f"\nTools invoked: {len(covered)}/13")
|
|
if missing:
|
|
print(f" Missing: {', '.join(sorted(missing))}")
|
|
|
|
has_fail = counts["FAIL"] > 0
|
|
print(f"\n{'=' * 60}")
|
|
if has_fail:
|
|
print("RESULT: SOME TESTS FAILED \u2717")
|
|
else:
|
|
print("RESULT: ALL PASS (PASS + PERM + SKIP) \u2713")
|
|
print(f"{'=' * 60}")
|
|
|
|
return 1 if has_fail else 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main() -> int:
|
|
print(f"Komodo MCP Comprehensive Tool Test")
|
|
print(f"Server: {MCP_URL}")
|
|
print(f"Timestamp prefix: {PREFIX}")
|
|
|
|
ok, client = phase1_health_and_session()
|
|
if not ok:
|
|
return 1
|
|
|
|
phase2_read_operations(client)
|
|
phase3_write_lifecycle(client)
|
|
phase4_execute(client)
|
|
cleanup_all(client)
|
|
return phase5_summary()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|