harden: per-item robustness + deadlines/caps across connectors; layered tests
Mail: _safe_decode + _iter_messages (per-mail try/except) -> eine kaputte Mail (charset x-unknown) bricht die Suche nicht mehr ab, findet so auch uralte Mails. Vollscan + Datum-Sort + MAX_SCAN/DEADLINE_S (Selbstheilung, kein 24h-Haenger mehr). Files: search_files mit SEARCH_DEADLINE_S + SEARCH_MAX_DIRS begrenzt (war unbegrenzte rekursive PROPFIND -> real 35s/449 Calls). Calendar/Contacts: PARSE_DEADLINE_S fuer client-seitiges vobject-Parsen; MAX_RESULTS-Output-Cap in get_events/get_tasks (war 10939 Zeilen bei 6000 Events); search Top-N + Hinweise. Tests: in Schichten getrennt -> Smoke (nightly), Edge + Stress (on-demand). Neu: test_smoke/test_edge/test_stress + conftest-Marker + run_full_tests.sh, nightly laeuft nur Smoke. Kontakt-Foto-Roundtrip-Test, Mail-Edge-Mails (x-unknown, kaputter Header, 2009er Alt-Mail). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Stefan Lohmaier
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
936ebc2f56
commit
c9ccb4392b
@@ -0,0 +1,183 @@
|
||||
#!/usr/bin/env python3
|
||||
"""MCP Edge-Case-Tests — ON-DEMAND (nicht naechtlich).
|
||||
|
||||
Pro Connector: ungewoehnliche/boese Eingaben (Unicode/Emoji, Sonderzeichen,
|
||||
riesige Limits/Bodies, kaputte Datumsangaben, Path-Traversal). Ziel: der Server
|
||||
faengt das sauber ab (graceful) statt zu crashen oder Daten zu leaken.
|
||||
|
||||
Lauf: /opt/mcp-servers/venv/bin/python -m pytest test_edge.py -q
|
||||
"""
|
||||
|
||||
import time
|
||||
import pytest
|
||||
|
||||
from test_all import SERVERS, get_token, mcp_call, tool_call
|
||||
|
||||
pytestmark = pytest.mark.edge
|
||||
|
||||
EMOJI = "Zürdäß 🚀😀 漢字 مرحبا"
|
||||
|
||||
|
||||
def _alive(port):
|
||||
"""Server beantwortet nach einer boesen Eingabe noch normal."""
|
||||
res = mcp_call(port, get_token(port), "tools/list")
|
||||
assert res and res.get("result", {}).get("tools"), f"Server {port} nach Edge-Call tot"
|
||||
|
||||
|
||||
# ============================================================
|
||||
# Mail
|
||||
# ============================================================
|
||||
|
||||
class TestMailEdge:
|
||||
PORT = SERVERS["mail"]
|
||||
|
||||
def test_huge_limit(self):
|
||||
tool_call(self.PORT, get_token(self.PORT), "search_mail",
|
||||
{"query": "mcptest", "limit": 100000, "account": "mcp-test-mail"})
|
||||
_alive(self.PORT)
|
||||
|
||||
def test_regex_special_chars_literal(self):
|
||||
for q in [".*", "[a-z]+", "%27 OR 1=1", "a\\b", "()|"]:
|
||||
tool_call(self.PORT, get_token(self.PORT), "search_mail",
|
||||
{"query": q, "limit": 3, "account": "mcp-test-mail"})
|
||||
_alive(self.PORT)
|
||||
|
||||
def test_very_long_query(self):
|
||||
tool_call(self.PORT, get_token(self.PORT), "search_mail",
|
||||
{"query": "x" * 2000, "limit": 3, "account": "mcp-test-mail"})
|
||||
_alive(self.PORT)
|
||||
|
||||
def test_nonexistent_account_filter(self):
|
||||
text = tool_call(self.PORT, get_token(self.PORT), "search_mail",
|
||||
{"query": "mcptest", "account": "does-not-exist-xyz"})
|
||||
assert "No results" in text or text.strip() == "" or "Subject:" not in text
|
||||
_alive(self.PORT)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# Calendar
|
||||
# ============================================================
|
||||
|
||||
class TestCalendarEdge:
|
||||
PORT = SERVERS["calendar"]
|
||||
CAL = "calendar-test"
|
||||
|
||||
def test_invalid_date_graceful(self):
|
||||
# kaputtes Datum darf den Server nicht killen
|
||||
mcp_call(self.PORT, get_token(self.PORT), "tools/call", {
|
||||
"name": "create_event",
|
||||
"arguments": {"calendar": self.CAL, "title": "BadDate", "start": "kein-datum", "end": "32.13.2026"},
|
||||
})
|
||||
_alive(self.PORT)
|
||||
|
||||
def test_unicode_emoji_event(self):
|
||||
token = get_token(self.PORT)
|
||||
tag = f"edge-uni-{int(time.time())}"
|
||||
tool_call(self.PORT, token, "create_event", {
|
||||
"calendar": self.CAL, "title": f"{EMOJI} {tag}",
|
||||
"start": "2027-01-02T10:00", "end": "2027-01-02T11:00",
|
||||
})
|
||||
found = tool_call(self.PORT, token, "search_events", {"query": tag})
|
||||
assert tag in found
|
||||
|
||||
def test_reversed_date_range(self):
|
||||
tool_call(self.PORT, get_token(self.PORT), "get_events",
|
||||
{"calendar": self.CAL, "date_from": "2027-12-31", "date_to": "2027-01-01"})
|
||||
_alive(self.PORT)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# Contacts
|
||||
# ============================================================
|
||||
|
||||
class TestContactsEdge:
|
||||
PORT = SERVERS["contacts"]
|
||||
|
||||
def test_unicode_emoji_contact(self):
|
||||
token = get_token(self.PORT)
|
||||
tag = f"EdgeUni{int(time.time())}"
|
||||
tool_call(self.PORT, token, "create_contact",
|
||||
{"name": f"{tag} {EMOJI}", "email": f"{tag.lower()}@example.com"})
|
||||
found = tool_call(self.PORT, token, "search_contacts", {"query": tag})
|
||||
assert tag in found
|
||||
|
||||
def test_special_char_search(self):
|
||||
for q in ["O'Brien", "müller & söhne", "100% sicher", "<script>"]:
|
||||
tool_call(self.PORT, get_token(self.PORT), "search_contacts", {"query": q})
|
||||
_alive(self.PORT)
|
||||
|
||||
def test_very_long_name(self):
|
||||
token = get_token(self.PORT)
|
||||
tag = f"EdgeLong{int(time.time())}"
|
||||
tool_call(self.PORT, token, "create_contact",
|
||||
{"name": f"{tag} " + ("Lang" * 200), "email": f"{tag.lower()}@example.com"})
|
||||
_alive(self.PORT)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# Files
|
||||
# ============================================================
|
||||
|
||||
class TestFilesEdge:
|
||||
PORT = SERVERS["files"]
|
||||
DIR = "/.mcp-tests-edge"
|
||||
|
||||
@pytest.fixture(autouse=True, scope="class")
|
||||
def _dir(self):
|
||||
token = get_token(self.PORT)
|
||||
tool_call(self.PORT, token, "create_folder", {"path": self.DIR})
|
||||
yield
|
||||
tool_call(self.PORT, token, "delete_file", {"path": self.DIR})
|
||||
|
||||
def test_path_traversal_blocked(self):
|
||||
# darf NIEMALS /etc/passwd ausliefern
|
||||
for p in ["/../../../../etc/passwd", "/.mcp-tests-edge/../../../etc/passwd"]:
|
||||
res = mcp_call(self.PORT, get_token(self.PORT), "tools/call",
|
||||
{"name": "read_file", "arguments": {"path": p}})
|
||||
text = ""
|
||||
if res and "result" in res:
|
||||
text = res["result"]["content"][0].get("text", "")
|
||||
assert "root:x:0:0" not in text, f"Path-Traversal LEAK bei {p}"
|
||||
_alive(self.PORT)
|
||||
|
||||
def test_unicode_emoji_roundtrip(self):
|
||||
token = get_token(self.PORT)
|
||||
path = f"{self.DIR}/edge-{int(time.time())}-{EMOJI[:4]}.txt"
|
||||
tool_call(self.PORT, token, "write_file", {"path": path, "content": EMOJI})
|
||||
read = mcp_call(self.PORT, token, "tools/call", {"name": "read_file", "arguments": {"path": path}})
|
||||
assert "🚀" in read["result"]["content"][0]["text"]
|
||||
tool_call(self.PORT, token, "delete_file", {"path": path})
|
||||
|
||||
def test_nonexistent_deep_path(self):
|
||||
text = tool_call(self.PORT, get_token(self.PORT), "file_info",
|
||||
{"path": "/.mcp-tests-edge/nope/nope/nope.txt"})
|
||||
assert "404" in text or "gefunden" in text.lower() or "not found" in text.lower()
|
||||
_alive(self.PORT)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# Notes
|
||||
# ============================================================
|
||||
|
||||
class TestNotesEdge:
|
||||
PORT = SERVERS["notes"]
|
||||
NB = "Inbox"
|
||||
|
||||
def test_unicode_emoji_markdown_note(self):
|
||||
token = get_token(self.PORT)
|
||||
tag = f"edge-uni-{int(time.time())}"
|
||||
body = f"# {EMOJI} {tag}\n\n- [ ] Aufgabe\n```py\nprint('x')\n```\n| a | b |\n|---|---|\n"
|
||||
result = tool_call(self.PORT, token, "create_note", {"notebook": self.NB, "title": f"Edge {tag}", "body": body})
|
||||
note_id = result.split("id: ")[1].rstrip(")").strip()
|
||||
read = tool_call(self.PORT, token, "read_note", {"note_id": note_id})
|
||||
assert tag in read and "🚀" in read
|
||||
|
||||
def test_large_body(self):
|
||||
token = get_token(self.PORT)
|
||||
tag = f"edge-big-{int(time.time())}"
|
||||
big = (f"Zeile {tag} " + "lorem ipsum " * 8 + "\n") * 800 # ~ hunderte KB
|
||||
result = tool_call(self.PORT, token, "create_note", {"notebook": self.NB, "title": f"Big {tag}", "body": big})
|
||||
note_id = result.split("id: ")[1].rstrip(")").strip()
|
||||
read = tool_call(self.PORT, token, "read_note", {"note_id": note_id})
|
||||
assert tag in read
|
||||
_alive(self.PORT)
|
||||
Reference in New Issue
Block a user