mcp-home/tests/test_stress.py

#!/usr/bin/env python3
"""MCP Stress-/Concurrency-Tests — ON-DEMAND (nicht naechtlich).

Feuert viele gleichzeitige Requests pro Connector und prueft: alle kommen sauber
durch (kein Deadlock/Haenger) und der Server ist danach sofort wieder responsiv.
Genau diese Art Test haette den Mail-Server-Haenger (single-threaded, 24h blockiert)
gefangen.

Lauf:  /opt/mcp-servers/venv/bin/python -m pytest test_stress.py -q
"""

import time
import random
import concurrent.futures as cf

import pytest

from test_all import SERVERS, get_token, tool_call

pytestmark = pytest.mark.stress

N = 40            # gleichzeitige Requests pro Welle
WORKERS = 16      # parallele Clients
RESPONSIVE_S = 15  # Server muss nach der Last so schnell wieder antworten


def _hammer(fn, n=N):
    """fn() n-mal nebenlaeufig ausfuehren; gibt Liste der Fehlertexte zurueck."""
    errors = []
    with cf.ThreadPoolExecutor(max_workers=WORKERS) as ex:
        futs = [ex.submit(fn) for _ in range(n)]
        for f in cf.as_completed(futs):
            try:
                f.result()
            except Exception as e:  # noqa: BLE001
                errors.append(repr(e)[:200])
    return errors


def _assert_responsive(port, probe):
    t = time.time()
    probe()
    dt = time.time() - t
    assert dt < RESPONSIVE_S, f"Server {port} nach Last traege: {dt:.1f}s"


class TestStress:

    def test_mail_concurrent_search(self):
        port = SERVERS["mail"]; token = get_token(port)
        queries = ["mcptest", "Rechnung", "stefan", "Willkommen",
                   "XUNKNOWNTOKEN", "URALTMAIL2009", "BADHEADERTOKEN", "Newsletter"]
        errors = _hammer(lambda: tool_call(port, token, "search_mail",
                         {"query": random.choice(queries), "limit": 5, "account": "mcp-test-mail"}))
        assert not errors, f"{len(errors)} Fehler unter Last: {errors[:3]}"
        _assert_responsive(port, lambda: tool_call(port, token, "list_accounts"))

    def test_calendar_concurrent_reads(self):
        port = SERVERS["calendar"]; token = get_token(port)
        def call():
            if random.random() < 0.5:
                tool_call(port, token, "list_calendars")
            else:
                tool_call(port, token, "get_events",
                          {"calendar": "calendar-test", "date_from": "2026-01-01", "date_to": "2026-12-31"})
        errors = _hammer(call)
        assert not errors, f"{len(errors)} Fehler: {errors[:3]}"
        _assert_responsive(port, lambda: tool_call(port, token, "list_calendars"))

    def test_contacts_concurrent_search(self):
        port = SERVERS["contacts"]; token = get_token(port)
        qs = ["Mustermann", "Mcp", "example", "xyzzy_nobody", "GmbH"]
        errors = _hammer(lambda: tool_call(port, token, "search_contacts",
                         {"query": random.choice(qs), "limit": 5}))
        assert not errors, f"{len(errors)} Fehler: {errors[:3]}"
        _assert_responsive(port, lambda: tool_call(port, token, "search_contacts", {"query": "Mustermann", "limit": 1}))

    def test_files_concurrent_reads(self):
        port = SERVERS["files"]; token = get_token(port)
        def call():
            if random.random() < 0.5:
                tool_call(port, token, "list_files", {"path": "/"})
            else:
                tool_call(port, token, "file_info", {"path": "/"})
        errors = _hammer(call)
        assert not errors, f"{len(errors)} Fehler: {errors[:3]}"
        _assert_responsive(port, lambda: tool_call(port, token, "list_files", {"path": "/"}))

    def test_notes_concurrent_reads(self):
        port = SERVERS["notes"]; token = get_token(port)
        def call():
            if random.random() < 0.5:
                tool_call(port, token, "list_notebooks")
            else:
                tool_call(port, token, "list_notes", {"limit": 5})
        errors = _hammer(call)
        assert not errors, f"{len(errors)} Fehler: {errors[:3]}"
        _assert_responsive(port, lambda: tool_call(port, token, "list_notebooks"))

    def test_all_servers_mixed_load(self):
        """Alle 5 Connector gleichzeitig unter Last — Gesamtsystem haelt."""
        probes = {
            "mail": lambda p, t: tool_call(p, t, "search_mail", {"query": "mcptest", "limit": 3, "account": "mcp-test-mail"}),
            "calendar": lambda p, t: tool_call(p, t, "list_calendars"),
            "contacts": lambda p, t: tool_call(p, t, "search_contacts", {"query": "Mustermann", "limit": 2}),
            "files": lambda p, t: tool_call(p, t, "list_files", {"path": "/"}),
            "notes": lambda p, t: tool_call(p, t, "list_notebooks"),
        }
        tokens = {svc: get_token(SERVERS[svc]) for svc in probes}
        def call():
            svc = random.choice(list(probes))
            probes[svc](SERVERS[svc], tokens[svc])
        errors = _hammer(call, n=60)
        assert not errors, f"{len(errors)} Fehler im Mischbetrieb: {errors[:3]}"