Testing und Debugging: Wenn's mal wieder nicht klappt
Ich sage es immer wieder: Es klappt NIE beim ersten Mal. Und weißt Du was? Das ist völlig okay. Mit den richtigen Tests findest Du jeden Fehler. Und mit der Zeit entwickelst Du ein Gespür dafür, wo die Probleme lauern.
Die Test-Pyramide
Wenn man sich ein paar Jahrzehnte mit Debugging beschäftigt, kristallisiert sich eine Wahrheit heraus: Strukturierte Tests sparen mehr Zeit als sie kosten.
# test_structure.py
"""
Test-Pyramide für MCP Server:
/\ Integration Tests (20%)
/ \ - Server läuft?
/ \ - Tools erreichbar?
/ \
/ Unit \ Unit Tests (70%)
/ Tests \ - Jede Funktion einzeln
/____________\ System Tests (10%)
- End-to-End mit Claude
"""
Unit Tests - Das Fundament
Fange klein an. Teste jede Funktion einzeln:
# test_security.py
#!/usr/bin/env python3
import unittest
import sys
sys.path.append('.')
from modules.security import SecurityValidator
class TestSecurityValidator(unittest.TestCase):
"""Tests für das Security-Modul"""
def setUp(self):
"""Vor jedem Test ausführen"""
self.validator = SecurityValidator('./config')
def test_sql_injection_detection(self):
"""SQL-Injection muss erkannt werden"""
dangerous_queries = [
"SELECT * FROM users; DROP TABLE users",
"' OR '1'='1",
"admin'--",
"1 UNION SELECT * FROM passwords"
]
for query in dangerous_queries:
with self.subTest(query=query):
valid, msg = self.validator.validate_query(query)
self.assertFalse(valid, f"Sollte blockiert werden: {query}")
print(f"✓ Blockiert: {query[:30]}...")
def test_valid_queries(self):
"""Normale Queries müssen durchkommen"""
valid_queries = [
"SELECT * FROM content_metadata WHERE id = ?",
"UPDATE content_metadata SET title = ? WHERE id = ?",
"INSERT INTO content_metadata (title) VALUES (?)"
]
for query in valid_queries:
with self.subTest(query=query):
valid, msg = self.validator.validate_query(query, [1, 2])
self.assertTrue(valid, f"Sollte erlaubt sein: {query}")
print(f"✓ Erlaubt: {query[:40]}...")
def test_rate_limiting(self):
"""Rate-Limiting muss greifen"""
user = "test_user"
# 50 Queries sollten okay sein
for i in range(50):
valid, msg = self.validator.check_rate_limit(user)
self.assertTrue(valid)
# Query 51 sollte blockiert werden
valid, msg = self.validator.check_rate_limit(user)
self.assertFalse(valid)
self.assertIn("Rate-Limit", msg)
print("✓ Rate-Limiting funktioniert")
if __name__ == "__main__":
unittest.main(verbosity=2)
Pro-Tipp: Schreibe Tests BEVOR Du den Code fixst. Test-Driven Development hat mir schon oft den Hintern gerettet.
Database Tests - Mit echten Daten
Teste mit einer Test-Datenbank, nicht mit Production-Daten:
# test_database.py
import unittest
import pymysql
from modules.database import DatabaseManager
class TestDatabaseOperations(unittest.TestCase):
"""Tests für Database-Modul"""
@classmethod
def setUpClass(cls):
"""Einmal vor allen Tests: Test-DB vorbereiten"""
cls.test_db = "test_mcp_db"
cls.test_table = "test_content"
# Test-Datenbank erstellen
conn = pymysql.connect(
host='localhost',
user='root',
password='root_password'
)
with conn.cursor() as cursor:
cursor.execute(f"CREATE DATABASE IF NOT EXISTS {cls.test_db}")
cursor.execute(f"USE {cls.test_db}")
cursor.execute(f"""
CREATE TABLE IF NOT EXISTS {cls.test_table} (
id INT PRIMARY KEY AUTO_INCREMENT,
title VARCHAR(255),
status VARCHAR(20),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.close()
def setUp(self):
"""Vor jedem Test"""
self.db = DatabaseManager('./config')
self.db.connect('content_writer')
# Tabelle leeren
with self.db.connection.cursor() as cursor:
cursor.execute(f"TRUNCATE TABLE {self.test_table}")
self.db.connection.commit()
def test_insert(self):
"""INSERT muss funktionieren"""
data = {
'title': 'Test Artikel',
'status': 'draft'
}
success, result = self.db.insert(data)
self.assertTrue(success)
self.assertIn('insert_id', result)
self.assertGreater(result['insert_id'], 0)
print(f"✓ INSERT: ID {result['insert_id']}")
def test_select(self):
"""SELECT muss Daten finden"""
# Erst was einfügen
self.db.insert({'title': 'Findmich', 'status': 'published'})
# Dann suchen
success, results = self.db.select(
columns=['id', 'title'],
where="title = ?",
params=['Findmich']
)
self.assertTrue(success)
self.assertEqual(len(results), 1)
self.assertEqual(results[0]['title'], 'Findmich')
print("✓ SELECT: Daten gefunden")
def test_update(self):
"""UPDATE muss ändern"""
# Einfügen
self.db.insert({'title': 'Alt', 'status': 'draft'})
# Ändern
success, result = self.db.update(
{'title': 'Neu'},
where="title = ?",
params=['Alt']
)
self.assertTrue(success)
self.assertEqual(result['affected_rows'], 1)
# Prüfen
success, results = self.db.select(
where="title = ?",
params=['Neu']
)
self.assertEqual(len(results), 1)
print("✓ UPDATE: Änderung durchgeführt")
def test_transaction_rollback(self):
"""Rollback muss funktionieren"""
self.db.begin_transaction()
# Einfügen
self.db.insert({'title': 'Sollte nicht bleiben'})
# Rollback
self.db.rollback_transaction()
# Prüfen dass nichts da ist
success, results = self.db.select()
self.assertEqual(len(results), 0)
print("✓ ROLLBACK: Transaktion zurückgerollt")
@classmethod
def tearDownClass(cls):
"""Nach allen Tests: Aufräumen"""
conn = pymysql.connect(host='localhost', user='root', password='root_password')
with conn.cursor() as cursor:
cursor.execute(f"DROP DATABASE IF EXISTS {cls.test_db}")
conn.close()
Integration Tests - Alles zusammen
Jetzt testen wir den kompletten Server:
# test_integration.py
import asyncio
import json
import subprocess
import time
import requests
class TestMCPIntegration:
"""Integration Tests für den MCP Server"""
def setup_method(self):
"""Server starten"""
self.server_process = subprocess.Popen(
['python', 'server.py'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
time.sleep(2) # Server hochfahren lassen
def teardown_method(self):
"""Server stoppen"""
self.server_process.terminate()
self.server_process.wait()
def test_server_responds(self):
"""Server muss antworten"""
# MCP verwendet JSON-RPC
payload = {
"jsonrpc": "2.0",
"method": "list_tables",
"params": {},
"id": 1
}
response = requests.post(
'http://localhost:3000/rpc',
json=payload
)
assert response.status_code == 200
data = response.json()
assert data['success'] == True
print("✓ Server antwortet")
def test_tool_execution(self):
"""Tools müssen ausführbar sein"""
tools = [
'list_tables',
'describe_table',
'count_records'
]
for tool in tools:
payload = {
"jsonrpc": "2.0",
"method": tool,
"params": {},
"id": 1
}
response = requests.post(
'http://localhost:3000/rpc',
json=payload
)
assert response.status_code == 200
print(f"✓ Tool {tool} funktioniert")
Debugging - Wenn's brennt
Hier meine bewährten Debugging-Techniken:
1. Der Print-Debug (Old but Gold)
def debug_query(query, params):
"""Alte Schule, aber funktioniert"""
print("="*50)
print(f"QUERY: {query}")
print(f"PARAMS: {params}")
print(f"TYPE: {type(params)}")
# Query mit Params zusammenbauen zum Testen
test_query = query
if params:
for p in params:
test_query = test_query.replace('?', f"'{p}'", 1)
print(f"FINAL: {test_query}")
print("="*50)
return query, params # Durchreichen
2. Der Logger-Debug
import logging
logging.basicConfig(level=logging.DEBUG)
# An kritischen Stellen
logger.debug(f"Vor Database-Connect: {locals()}")
# Bei Exceptions
try:
# Code
except Exception as e:
logger.exception("Vollständiger Stack-Trace:")
raise
3. Der PDB-Debug (Python Debugger)
import pdb
def problematic_function(data):
# Hier will ich reinschauen
pdb.set_trace() # <-- Breakpoint
# Ab hier kannst Du im Debugger:
# n - next line
# s - step into
# c - continue
# p variable - print variable
# ll - list code
result = process_data(data)
return result
Vergiss nicht, pdb.set_trace() wieder zu entfernen bevor Du commitest. Frag mich nicht, wie oft ich das vergessen habe...
Performance Testing
Wie schnell ist schnell genug?
# test_performance.py
import time
import statistics
from concurrent.futures import ThreadPoolExecutor
def test_query_performance():
"""Misst Query-Performance"""
times = []
for _ in range(100):
start = time.perf_counter()
# Deine Query hier
db.select(limit=10)
end = time.perf_counter()
times.append(end - start)
print(f"Min: {min(times)*1000:.2f}ms")
print(f"Max: {max(times)*1000:.2f}ms")
print(f"Avg: {statistics.mean(times)*1000:.2f}ms")
print(f"Median: {statistics.median(times)*1000:.2f}ms")
# Alles unter 100ms ist okay
assert statistics.median(times) < 0.1
def test_concurrent_load():
"""Testet unter Last"""
def single_request():
return db.select(limit=1)
with ThreadPoolExecutor(max_workers=10) as executor:
start = time.perf_counter()
# 100 Requests parallel
futures = [executor.submit(single_request) for _ in range(100)]
results = [f.result() for f in futures]
end = time.perf_counter()
print(f"100 Requests in {end-start:.2f}s")
print(f"Requests/sec: {100/(end-start):.0f}")
# Mindestens 20 req/s sollten drin sein
assert (100/(end-start)) > 20
Test Coverage - Wie viel ist genug?
Installiere coverage und schaue, was Du testest:
# Coverage installieren
pip install coverage
# Tests mit Coverage laufen lassen
coverage run -m pytest tests/
coverage report -m
# HTML Report generieren
coverage html
# Öffne htmlcov/index.html im Browser
Meine Coverage-Ziele:
- Security-Module: 100% (keine Kompromisse!)
- Database-Module: 90%+
- Logging: 70% reicht
- Utilities: 50% ist okay
Continuous Testing
Automatisiere Deine Tests:
# run_tests.sh
#!/bin/bash
echo "MCP Server Test Suite"
echo "========================"
# Umgebung vorbereiten
source venv/bin/activate
# Unit Tests
echo -e "\nUnit Tests:"
python -m pytest tests/unit/ -v
# Integration Tests
echo -e "\nIntegration Tests:"
python -m pytest tests/integration/ -v
# Security Tests
echo -e "\nSecurity Tests:"
python test_security.py
# Performance Tests
echo -e "\nPerformance Tests:"
python test_performance.py
# Coverage Report
echo -e "\nCoverage Report:"
coverage run -m pytest tests/
coverage report
echo -e "\n[OK] Alle Tests abgeschlossen!"
Debug-Checkliste
Wenn mal wieder nichts geht:
Die systematische Fehlersuche:
- Fehlermeldung lesen: Wirklich lesen, nicht überfliegen
- Stelle lokalisieren: Wo genau knallt's?
- Kontext prüfen: Was kam davor?
- Logs checken: Alle Logs, nicht nur einen
- Isoliert testen: Nur die kaputte Funktion
- Hypothese bilden: Was könnte es sein?
- Fix testen: Behebt es das Problem?
- Regression testen: Habe ich was anderes kaputt gemacht?
Typische Fehler und Lösungen
# debug_common_issues.py
"""
Häufige Probleme und ihre Lösungen
"""
COMMON_ISSUES = {
"Access denied for user": {
"ursache": "Falsches Passwort oder User existiert nicht",
"lösung": "credentials.json prüfen, MySQL-User checken",
"befehl": "mysql -u content_reader -p"
},
"No module named 'fastmcp'": {
"ursache": "Dependencies nicht installiert",
"lösung": "Virtual Environment aktivieren, pip install",
"befehl": "source venv/bin/activate && pip install fastmcp"
},
"Rate limit exceeded": {
"ursache": "Zu viele Requests",
"lösung": "Warten oder Limit in security.json erhöhen",
"befehl": "# 1 Minute warten"
},
"Connection refused": {
"ursache": "Server läuft nicht",
"lösung": "Server starten",
"befehl": "python server.py"
},
"Table doesn't exist": {
"ursache": "Falsche Datenbank oder Tabelle fehlt",
"lösung": "Datenbank prüfen, Tabelle erstellen",
"befehl": "SHOW TABLES IN karlkratz_de;"
}
}
Was ich gelernt habe
Nach unzähligen Debugging-Sessions:
- Tests sind keine Zeitverschwendung - sie sparen Zeit
- Der Fehler ist immer dümmer als Du denkst
- Wenn Du nicht weiterkommst, mach eine Pause
- Dokumentiere Deine Fehler - Du machst sie wieder
- Ein guter Test ist besser als 10 Stunden debuggen
Merke: Ein Bug, den Dein Test findet, findet nicht Dein Kunde.
Im nächsten Kapitel zeige ich Dir, wie Du Deinen fertigen und getesteten Server in die Production bringst. Du lernst, wie Du mit Systemd einen automatischen Start einrichtest, Health Checks implementierst und ein professionelles Monitoring aufbaust. Außerdem schauen wir uns Backup-Strategien an und wie Du Deinen Server gegen Angriffe härtest. Weiter zum Deployment und zur Production-Konfiguration