slohmaier/mcp-home
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
bced937a24c2941ba217df98431ae70453d76e42
mcp-home/tests/test_all.py
T
Stefan Lohmaier c06e6d6b4c Joplin via Data API + Mail attachments 
Notes server rewritten to use Joplin CLI Data API (joplin-cli sync
clients on host, ports 41184/41185). Clean fast search, proper
resource handling. New tools: list_note_resources, read_resource
(attachments as inline image/document/text).

Mail server: read_mail now lists attachments; new read_attachment
tool returns images inline, PDFs/docs as EmbeddedResource, text directly.

Tests: 54 total. Notes now real (notebooks, list, create+read).
Mail attachment listing + fetch tested.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-06-12 12:26:26 +02:00

424 lines
17 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""MCP Server Integration Tests.
Uses a 'test' OAuth client (maps to stefan's data).
Creates, reads, updates, and deletes test data in dedicated test collections:
- Radicale: calendar-test, contacts-test
- oCIS: /mcp-tests/
- Mail: read-only (no test mailbox)
Run: /opt/mcp-servers/venv/bin/python -m pytest /opt/mcp-servers/tests/test_all.py -v
"""
import json
import hashlib
import base64
import secrets
import time
import httpx
import pytest
SERVERS = {
"mail": 5100,
"calendar": 5101,
"contacts": 5102,
"files": 5103,
"notes": 5104,
}
with open("/opt/mcp-servers/tokens.json") as f:
TEST_SECRET = json.load(f)["test"]["token"]
_tokens = {}
def get_token(port):
if port not in _tokens:
r = httpx.post(f"http://127.0.0.1:{port}/token", data={
"grant_type": "client_credentials", "client_id": "test", "client_secret": TEST_SECRET,
}, timeout=10)
assert r.status_code == 200, f"Token failed: {r.text}"
_tokens[port] = r.json()["access_token"]
return _tokens[port]
def get_token_pkce(port):
verifier = secrets.token_urlsafe(32)
challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
r = httpx.get(f"http://127.0.0.1:{port}/authorize", params={
"response_type": "code", "client_id": "test",
"redirect_uri": "http://localhost/cb", "code_challenge": challenge, "code_challenge_method": "S256",
}, follow_redirects=False, timeout=10)
assert r.status_code == 302
code = r.headers["location"].split("code=")[1].split("&")[0]
r2 = httpx.post(f"http://127.0.0.1:{port}/token", data={
"grant_type": "authorization_code", "code": code, "client_id": "test", "code_verifier": verifier,
}, timeout=10)
assert r2.status_code == 200
return r2.json()["access_token"]
def mcp_call(port, token, method, params=None):
r = httpx.post(f"http://127.0.0.1:{port}/mcp",
json={"jsonrpc": "2.0", "id": 1, "method": method, "params": params or {}},
headers={"Authorization": f"Bearer {token}", "Accept": "application/json, text/event-stream"},
timeout=60)
assert r.status_code == 200, f"MCP {r.status_code}: {r.text[:200]}"
for line in r.text.split("\n"):
if line.startswith("data: "):
return json.loads(line[6:])
return None
def tool_call(port, token, tool_name, arguments=None):
result = mcp_call(port, token, "tools/call", {"name": tool_name, "arguments": arguments or {}})
assert result and "error" not in result, f"Error: {result}"
content = result["result"]["content"]
assert len(content) > 0
return content[0].get("text", "")
# ============================================================
# OAuth Tests (all servers)
# ============================================================
class TestOAuth:
@pytest.mark.parametrize("svc,port", SERVERS.items())
def test_metadata(self, svc, port):
r = httpx.get(f"http://127.0.0.1:{port}/.well-known/oauth-authorization-server", timeout=10)
assert r.status_code == 200
d = r.json()
assert "authorization_endpoint" in d
assert "token_endpoint" in d
@pytest.mark.parametrize("svc,port", SERVERS.items())
def test_client_credentials(self, svc, port):
assert len(get_token(port)) > 20
@pytest.mark.parametrize("svc,port", SERVERS.items())
def test_pkce_flow(self, svc, port):
assert len(get_token_pkce(port)) > 20
@pytest.mark.parametrize("svc,port", SERVERS.items())
def test_invalid_secret_rejected(self, svc, port):
r = httpx.post(f"http://127.0.0.1:{port}/token", data={
"grant_type": "client_credentials", "client_id": "test", "client_secret": "WRONG",
}, timeout=10)
assert r.status_code == 401
@pytest.mark.parametrize("svc,port", SERVERS.items())
def test_no_auth_rejected(self, svc, port):
r = httpx.post(f"http://127.0.0.1:{port}/mcp",
json={"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}},
headers={"Accept": "application/json, text/event-stream"}, timeout=10)
assert r.status_code == 401
@pytest.mark.parametrize("svc,port", SERVERS.items())
def test_tools_list(self, svc, port):
result = mcp_call(port, get_token(port), "tools/list")
tools = result["result"]["tools"]
assert len(tools) > 0
for t in tools:
assert "name" in t
assert "description" in t
# ============================================================
# Mail Tests (read-only)
# ============================================================
class TestMail:
PORT = SERVERS["mail"]
def test_list_accounts(self):
text = tool_call(self.PORT, get_token(self.PORT), "list_accounts")
assert "d370128_0-slohmaier" in text
def test_list_folders(self):
text = tool_call(self.PORT, get_token(self.PORT), "list_folders", {"account": "d370128_0-slohmaier"})
assert "INBOX" in text
def test_search_finds_something(self):
text = tool_call(self.PORT, get_token(self.PORT), "search_mail", {"query": "stefan", "limit": 2})
assert "Subject:" in text
def test_search_empty_result(self):
text = tool_call(self.PORT, get_token(self.PORT), "search_mail",
{"query": "xyzzy_nonexistent_99", "limit": 1, "account": "d370128_0-gitea"})
assert "No results" in text
def test_read_mail(self):
# First search to get a key
text = tool_call(self.PORT, get_token(self.PORT), "search_mail",
{"query": "stefan", "limit": 1, "account": "d370128_0-slohmaier"})
if "No results" in text:
pytest.skip("No mails found")
# Parse account, folder, key from result
for line in text.split("\n"):
if "Account:" in line:
parts = line.strip().split(", ")
account = parts[0].split(": ")[1]
folder = parts[1].split(": ")[1]
key = parts[2].split(": ")[1]
break
body = tool_call(self.PORT, get_token(self.PORT), "read_mail",
{"account": account, "folder": folder, "key": key})
assert "From:" in body
assert "Subject:" in body
def test_attachment_listing_and_fetch(self):
token = get_token(self.PORT)
# Find a mail with an attachment by scanning Rechnung search results
text = tool_call(self.PORT, token, "search_mail",
{"query": "Rechnung", "limit": 10, "account": "d370128_0-slohmaier"})
if "No results" in text:
pytest.skip("No mails found")
# Read each result, look for one with attachments
found_att = None
for block in text.split("\n\n"):
acc = fld = k = None
for line in block.split("\n"):
if "Account:" in line:
parts = line.strip().split(", ")
acc = parts[0].split(": ")[1]
fld = parts[1].split(": ")[1]
k = parts[2].split(": ")[1]
if not k:
continue
body = tool_call(self.PORT, token, "read_mail", {"account": acc, "folder": fld, "key": k})
if "Anhaenge" in body:
found_att = (acc, fld, k)
break
if not found_att:
pytest.skip("No mail with attachment found")
acc, fld, k = found_att
result = mcp_call(self.PORT, token, "tools/call", {
"name": "read_attachment",
"arguments": {"account": acc, "folder": fld, "key": k, "attachment_index": 1},
})
content = result["result"]["content"][0]
assert content["type"] in ("image", "resource", "text")
# ============================================================
# Calendar Tests (CRUD on calendar-test)
# ============================================================
class TestCalendar:
PORT = SERVERS["calendar"]
TEST_CAL = "calendar-test"
def test_list_calendars(self):
text = tool_call(self.PORT, get_token(self.PORT), "list_calendars")
assert "calendar-stefan" in text or "Stefan" in text
def test_list_task_lists(self):
text = tool_call(self.PORT, get_token(self.PORT), "list_task_lists")
assert len(text) > 0
def test_create_and_find_event(self):
token = get_token(self.PORT)
tag = f"mcptest-{int(time.time())}"
# Create
result = tool_call(self.PORT, token, "create_event", {
"calendar": self.TEST_CAL,
"title": f"Test Event {tag}",
"start": "2026-12-25T10:00",
"end": "2026-12-25T11:00",
"location": "Testort",
"description": "Automatischer Test",
})
assert "erstellt" in result.lower() or "created" in result.lower(), f"Create failed: {result}"
# Search
found = tool_call(self.PORT, token, "search_events", {"query": tag})
assert tag in found, f"Event not found after create: {found[:200]}"
def test_create_and_find_task(self):
token = get_token(self.PORT)
tag = f"mcptask-{int(time.time())}"
result = tool_call(self.PORT, token, "create_task", {
"task_list": self.TEST_CAL,
"title": f"Test Task {tag}",
"due": "2026-12-31",
"priority": 5,
})
assert "erstellt" in result.lower() or "created" in result.lower(), f"Create failed: {result}"
tasks = tool_call(self.PORT, token, "get_tasks", {"task_list": self.TEST_CAL, "include_completed": True})
assert tag in tasks, f"Task not found: {tasks[:200]}"
def test_get_events_date_range(self):
text = tool_call(self.PORT, get_token(self.PORT), "get_events", {
"calendar": "stefan", "date_from": "2026-06-01", "date_to": "2026-06-30",
})
assert "Wann:" in text or "Keine Termine" in text
# ============================================================
# Contacts Tests (CRUD on contacts-test)
# ============================================================
class TestContacts:
PORT = SERVERS["contacts"]
def test_search_real_contacts(self):
text = tool_call(self.PORT, get_token(self.PORT), "search_contacts", {"query": "Lohmaier", "limit": 3})
assert "Lohmaier" in text
def test_search_empty(self):
text = tool_call(self.PORT, get_token(self.PORT), "search_contacts", {"query": "xyzzy_nobody_99"})
assert "Keine Kontakte" in text
def test_create_search_contact(self):
token = get_token(self.PORT)
tag = f"McpTest-{int(time.time())}"
result = tool_call(self.PORT, token, "create_contact", {
"name": f"{tag} Mustermann",
"email": f"{tag.lower()}@example.com",
"phone": "+49 170 0000000",
"organization": "MCP Tests GmbH",
"note": "Automatisch erstellt",
})
assert "erstellt" in result.lower() or "created" in result.lower(), f"Create failed: {result}"
found = tool_call(self.PORT, token, "search_contacts", {"query": tag})
assert tag in found, f"Contact not found: {found[:200]}"
def test_get_contact_details(self):
token = get_token(self.PORT)
search = tool_call(self.PORT, token, "search_contacts", {"query": "Lohmaier", "limit": 1})
if "Keine" in search:
pytest.skip("No contacts")
uid = [l for l in search.split("\n") if "UID:" in l][0].split("UID: ")[1].strip()
# get_contact returns list content, first element is text
result = mcp_call(self.PORT, token, "tools/call", {"name": "get_contact", "arguments": {"uid": uid}})
text = result["result"]["content"][0]["text"]
assert "Lohmaier" in text
assert "UID:" in text
# ============================================================
# Files Tests (CRUD on /mcp-tests/)
# ============================================================
class TestFiles:
PORT = SERVERS["files"]
TEST_DIR = "/.mcp-tests"
@pytest.fixture(autouse=True, scope="class")
def ensure_test_dir(self):
"""Create test dir at start, delete at end. Isolated from real data."""
token = get_token(self.PORT)
tool_call(self.PORT, token, "create_folder", {"path": self.TEST_DIR})
yield
tool_call(self.PORT, token, "delete_file", {"path": self.TEST_DIR})
def test_list_root(self):
text = tool_call(self.PORT, get_token(self.PORT), "list_files", {"path": "/"})
assert "DIR" in text
def test_file_info_root(self):
text = tool_call(self.PORT, get_token(self.PORT), "file_info", {"path": "/"})
assert "Verzeichnis" in text or "Directory" in text
def test_write_read_delete_file(self):
token = get_token(self.PORT)
tag = f"test-{int(time.time())}"
path = f"{self.TEST_DIR}/{tag}.txt"
result = tool_call(self.PORT, token, "write_file", {
"path": path, "content": f"Hello MCP Test {tag}\nZeile 2",
})
assert "geschrieben" in result.lower() or "written" in result.lower(), f"Write failed: {result}"
read_result = mcp_call(self.PORT, token, "tools/call", {"name": "read_file", "arguments": {"path": path}})
text = read_result["result"]["content"][0]["text"]
assert tag in text
info = tool_call(self.PORT, token, "file_info", {"path": path})
assert "bytes" in info.lower()
del_result = tool_call(self.PORT, token, "delete_file", {"path": path})
assert "eloescht" in del_result.lower() or "deleted" in del_result.lower()
info2 = tool_call(self.PORT, token, "file_info", {"path": path})
assert "404" in info2 or "Nicht gefunden" in info2 or "Not found" in info2
def test_create_delete_folder(self):
token = get_token(self.PORT)
tag = f"testdir-{int(time.time())}"
path = f"{self.TEST_DIR}/{tag}"
result = tool_call(self.PORT, token, "create_folder", {"path": path})
assert "erstellt" in result.lower() or "created" in result.lower()
listing = tool_call(self.PORT, token, "list_files", {"path": self.TEST_DIR})
assert tag in listing
tool_call(self.PORT, token, "delete_file", {"path": path})
def test_move_file(self):
token = get_token(self.PORT)
tag = f"move-{int(time.time())}"
src = f"{self.TEST_DIR}/{tag}-src.txt"
dst = f"{self.TEST_DIR}/{tag}-dst.txt"
tool_call(self.PORT, token, "write_file", {"path": src, "content": "move test"})
result = tool_call(self.PORT, token, "move_file", {"source": src, "destination": dst})
assert "erschoben" in result.lower() or "moved" in result.lower()
read = mcp_call(self.PORT, token, "tools/call", {"name": "read_file", "arguments": {"path": dst}})
assert "move test" in read["result"]["content"][0]["text"]
tool_call(self.PORT, token, "delete_file", {"path": dst})
def test_search_files(self):
token = get_token(self.PORT)
tag = f"search-{int(time.time())}"
path = f"{self.TEST_DIR}/{tag}.md"
tool_call(self.PORT, token, "write_file", {"path": path, "content": "searchable"})
result = tool_call(self.PORT, token, "search_files", {"query": tag, "path": self.TEST_DIR})
assert tag in result
tool_call(self.PORT, token, "delete_file", {"path": path})
# ============================================================
# Notes Tests (Joplin Data API)
# ============================================================
class TestNotes:
PORT = SERVERS["notes"]
TEST_NOTEBOOK = "Inbox"
def test_list_notebooks(self):
text = tool_call(self.PORT, get_token(self.PORT), "list_notebooks")
assert "id:" in text and "Keine" not in text
def test_list_notes(self):
text = tool_call(self.PORT, get_token(self.PORT), "list_notes", {"limit": 5})
assert "id:" in text or "Keine Notizen" in text
def test_create_search_read_note(self):
token = get_token(self.PORT)
tag = f"mcptest-{int(time.time())}"
# Create
result = tool_call(self.PORT, token, "create_note", {
"notebook": self.TEST_NOTEBOOK,
"title": f"Test Note {tag}",
"body": f"Automatischer Test {tag}\n\nInhalt.",
})
assert "erstellt" in result.lower(), f"Create failed: {result}"
note_id = result.split("id: ")[1].rstrip(")").strip()
# Read back (search FTS index may lag, so verify via read)
body = tool_call(self.PORT, token, "read_note", {"note_id": note_id})
assert tag in body, f"Note content mismatch: {body[:200]}"
# Verify it appears in the notebook listing
listing = tool_call(self.PORT, token, "list_notes", {"notebook": self.TEST_NOTEBOOK})
assert note_id in listing or f"Test Note {tag}" in listing
Reference in New Issue View Git Blame Copy Permalink