Agent Zero
abbe7a18fc
- P0: hooks.py 3-tuple fix, trigger_dispatcher Contract, contacts/plugin unregister_actions_by_owner - P0: 5 test files — check_permission mocks removed, hardcoded DB credential → env var - P1: attachment_service DmsFile via Contract helper, restore_registry/history_hooks dedup - P1: mail/plugin restore unregister, mcp_client datetime.now(UTC), saved_views/filters patterns - P1: ProtectedRoute fail-closed, 13 test assertion fixes (bcrypt, DB-URLs, SECRET_KEYs) - P2: deprecated notifications → post_system_message (3 files), forgejo Base, report_generator lazy import - P2: webhooks permissions, deps.py/roles.py plugin perms removed, import_export default - P2: address/tags/entity_links patterns removed, worker.py Contract-Umgehungen fixed - P2: 28 frontend TODOs (hardcoded constants, deprecated notification API) - P3: dead code, duplicates, deprecated imports, private attr, __import__ inline - P3: 8 frontend TODOs (LucideIcons, inline styles, XSS, i18n) - ruff: 838 → 0 (612 auto-fix + 246 manual + 27 F821 regression fix) - F821: 30 → 0 (AutomationDefinition, DmsFile, user_id, Path, Any, String) - Contract-Umgehungen: 2 neue gefunden (worker.py:169, worker.py:280) und gefixt
129 lines
4.2 KiB
Python
129 lines
4.2 KiB
Python
"""Tests for MCP Server plugin — tool listing, execution, config.
|
|
|
|
Tests Task 5.16 acceptance criteria.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from tests.conftest import ORIGIN_HEADER
|
|
|
|
|
|
# ─── AC1: List MCP tools ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac1_list_mcp_tools(mcp_authed_client):
|
|
"""AC1: GET /api/v1/mcp/tools → 200 + tool list with 9 tools."""
|
|
client, _ = mcp_authed_client
|
|
resp = await client.get("/api/v1/mcp/tools", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["count"] >= 1
|
|
assert len(data["tools"]) == data["count"]
|
|
tool_names = [t["name"] for t in data["tools"]]
|
|
assert "call_crm_api" in tool_names
|
|
|
|
|
|
# ─── AC2: Get MCP config ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac2_get_mcp_config(mcp_authed_client):
|
|
"""AC2: GET /api/v1/mcp/config → 200 + server config."""
|
|
client, _ = mcp_authed_client
|
|
resp = await client.get("/api/v1/mcp/config", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["server_name"] == "LeoCRM"
|
|
assert data["server_version"] == "1.0.0"
|
|
assert data["protocol_version"] == "2024-11-05"
|
|
assert data["auth_method"] == "api-token"
|
|
assert "call_crm_api" in data["available_tools"]
|
|
assert len(data["available_tools"]) >= 1
|
|
|
|
|
|
# ─── AC3: Execute search_contacts tool ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac3_execute_search_contacts(mcp_authed_client):
|
|
"""AC3: POST /api/v1/mcp/tools/call_crm_api/execute → 200 + results."""
|
|
client, _ = mcp_authed_client
|
|
resp = await client.post(
|
|
"/api/v1/mcp/tools/call_crm_api/execute",
|
|
json={"arguments": {"method": "GET", "path": "/api/v1/contacts"}},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["tool"] == "call_crm_api"
|
|
assert data["success"] is True
|
|
assert "result" in data
|
|
|
|
|
|
# ─── AC4: Execute non-existent tool returns 404 ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac4_execute_nonexistent_tool(mcp_authed_client):
|
|
"""AC4: POST /api/v1/mcp/tools/nonexistent/execute → 404."""
|
|
client, _ = mcp_authed_client
|
|
resp = await client.post(
|
|
"/api/v1/mcp/tools/nonexistent_tool/execute",
|
|
json={"arguments": {}},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 404
|
|
assert "tool_not_found" in resp.text
|
|
|
|
|
|
# ─── AC5: Tool definitions have correct schema ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac5_tool_definitions_schema(mcp_authed_client):
|
|
"""AC5: GET /api/v1/mcp/tools → tools have proper schema with parameters."""
|
|
client, _ = mcp_authed_client
|
|
resp = await client.get("/api/v1/mcp/tools", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
tools = resp.json()["tools"]
|
|
|
|
# Check call_crm_api has method, path, body params
|
|
api_tool = next(t for t in tools if t["name"] == "call_crm_api")
|
|
param_names = [p["name"] for p in api_tool["parameters"]]
|
|
assert "method" in param_names
|
|
assert "path" in param_names
|
|
method_param = next(p for p in api_tool["parameters"] if p["name"] == "method")
|
|
assert method_param["required"] is True
|
|
|
|
|
|
# ─── AC6: Unauthorized access is rejected ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac6_unauthorized_access(mcp_client_fixture):
|
|
"""AC6: GET /api/v1/mcp/tools without auth → 401."""
|
|
resp = await mcp_client_fixture.get("/api/v1/mcp/tools", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 401
|
|
|
|
|
|
# ─── AC7: Execute create_contact tool ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac7_execute_create_contact(mcp_authed_client):
|
|
"""AC7: POST /api/v1/mcp/tools/call_crm_api/execute → 200 + created contact."""
|
|
client, _ = mcp_authed_client
|
|
resp = await client.post(
|
|
"/api/v1/mcp/tools/call_crm_api/execute",
|
|
json={"arguments": {"method": "POST", "path": "/api/v1/contacts", "body": {"firstname": "MCP", "surname": "Test", "email_1": "mcp@test.com"}}},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["tool"] == "call_crm_api"
|
|
assert data["success"] is True
|
|
assert "result" in data
|