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:

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:

  1. Fehlermeldung lesen: Wirklich lesen, nicht überfliegen
  2. Stelle lokalisieren: Wo genau knallt's?
  3. Kontext prüfen: Was kam davor?
  4. Logs checken: Alle Logs, nicht nur einen
  5. Isoliert testen: Nur die kaputte Funktion
  6. Hypothese bilden: Was könnte es sein?
  7. Fix testen: Behebt es das Problem?
  8. 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:

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