#!/usr/bin/env python3 """MCP Server Integration Tests. Tests all tools on all servers via HTTP, using the 'test' OAuth client. 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 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"] def get_token(port): """Get access token via OAuth client_credentials.""" 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 request failed: {r.text}" return r.json()["access_token"] def get_token_pkce(port): """Get access token via OAuth authorization_code + PKCE.""" 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/callback", "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, f"Token exchange failed: {r2.text}" return r2.json()["access_token"] def mcp_call(port, token, method, params=None): """Call an MCP method and return the result.""" 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=30) assert r.status_code == 200, f"MCP request failed: {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): """Call an MCP tool and return the text content.""" result = mcp_call(port, token, "tools/call", { "name": tool_name, "arguments": arguments or {}, }) assert result is not None, "No SSE response" assert "error" not in result, f"RPC error: {result.get('error')}" content = result["result"]["content"] assert len(content) > 0, "Empty content" return content[0].get("text", "") # --- OAuth Tests --- class TestOAuth: @pytest.mark.parametrize("service,port", SERVERS.items()) def test_metadata(self, service, 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("service,port", SERVERS.items()) def test_client_credentials(self, service, port): token = get_token(port) assert len(token) > 20 @pytest.mark.parametrize("service,port", SERVERS.items()) def test_pkce_flow(self, service, port): token = get_token_pkce(port) assert len(token) > 20 @pytest.mark.parametrize("service,port", SERVERS.items()) def test_invalid_secret(self, service, 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("service,port", SERVERS.items()) def test_no_auth(self, service, 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 # --- Mail Tests --- class TestMail: @pytest.fixture(autouse=True) def setup(self): self.token = get_token(SERVERS["mail"]) def test_list_accounts(self): text = tool_call(SERVERS["mail"], self.token, "list_accounts") assert "d370128_0-slohmaier" in text assert "folders" in text def test_list_folders(self): text = tool_call(SERVERS["mail"], self.token, "list_folders", {"account": "d370128_0-slohmaier"}) assert "INBOX" in text def test_search_mail(self): text = tool_call(SERVERS["mail"], self.token, "search_mail", {"query": "stefan", "limit": 2}) assert "Subject:" in text or "No results" in text def test_search_nonexistent(self): text = tool_call(SERVERS["mail"], self.token, "search_mail", {"query": "xyzzy_nonexistent_12345", "limit": 1, "account": "d370128_0-gitea"}) assert "No results" in text # --- Calendar Tests --- class TestCalendar: @pytest.fixture(autouse=True) def setup(self): self.token = get_token(SERVERS["calendar"]) def test_list_calendars(self): text = tool_call(SERVERS["calendar"], self.token, "list_calendars") assert "calendar-stefan" in text or "Stefan" in text def test_list_task_lists(self): text = tool_call(SERVERS["calendar"], self.token, "list_task_lists") assert "reminders" in text.lower() or "erinnerungen" in text.lower() or "calendar" in text.lower() def test_get_events(self): text = tool_call(SERVERS["calendar"], self.token, "get_events", { "calendar": "stefan", "date_from": "2026-06-01", "date_to": "2026-06-30" }) assert "Wann:" in text or "Keine Termine" in text def test_search_events(self): text = tool_call(SERVERS["calendar"], self.token, "search_events", {"query": "Training"}) assert "UID:" in text or "Keine Treffer" in text # --- Contacts Tests --- class TestContacts: @pytest.fixture(autouse=True) def setup(self): self.token = get_token(SERVERS["contacts"]) def test_search_contacts(self): text = tool_call(SERVERS["contacts"], self.token, "search_contacts", {"query": "Lohmaier", "limit": 3}) assert "Lohmaier" in text or "Keine Kontakte" in text def test_search_empty(self): text = tool_call(SERVERS["contacts"], self.token, "search_contacts", {"query": "xyzzy_nonexistent_99999"}) assert "Keine Kontakte" in text # --- Files Tests --- class TestFiles: @pytest.fixture(autouse=True) def setup(self): self.token = get_token(SERVERS["files"]) def test_list_root(self): text = tool_call(SERVERS["files"], self.token, "list_files", {"path": "/"}) assert "DIR" in text or "Empty" in text or "Leeres" in text def test_file_info(self): text = tool_call(SERVERS["files"], self.token, "file_info", {"path": "/"}) assert "Verzeichnis" in text or "Directory" in text or "Name:" in text # --- Notes Tests --- class TestNotes: @pytest.fixture(autouse=True) def setup(self): self.token = get_token(SERVERS["notes"]) def test_list_notebooks(self): text = tool_call(SERVERS["notes"], self.token, "list_notebooks") # Notes likely not configured, so accept error message too assert "Kein" in text or "Fehler" in text or "id:" in text.lower()