Spaces:

Hwandji
/

SAAP

Sleeping

SAAP / backend /scripts /evaluation_tests.py

Add full SAAP platform with Docker SDK

63bfc55 7 months ago

22.2 kB

#!/usr/bin/env python3
"""
SAAP Evaluation Test Suite - Cross-Platform (Linux/Windows)
===========================================================
Dieses Skript führt alle Tests für Kapitel 4.4 der Masterarbeit aus.
Es ist plattformunabhängig und speichert Ergebnisse in JSON-Format.
Verwendung:
    python evaluation_tests.py [--base-url http://localhost:8000] [--output results.json]
Voraussetzungen:
    - SAAP Docker-Container müssen laufen: docker compose up -d
    - Python 3.10+ mit requests-Bibliothek
Installation:
    pip install requests
Autor: Hanan W.
Version: 1.0
Datum: 20. Januar 2026
"""

import argparse
import json
import platform
import statistics
import sys
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any

try:
    import requests
except ImportError:
    print("ERROR: 'requests' library not found. Install with: pip install requests")
    sys.exit(1)


# ============================================================================
# DATA CLASSES
# ============================================================================

@dataclass
class TestResult:
    """Einzelnes Testergebnis"""
    test_id: str
    test_name: str
    category: str
    input_data: str
    expected_result: str
    actual_result: str
    passed: bool
    latency_seconds: float
    provider: str = ""
    privacy_level: str = ""
    error_message: str = ""
    raw_response: dict = field(default_factory=dict)


@dataclass
class TestSuiteResults:
    """Gesamtergebnisse der Testsuite"""
    platform: str
    os_version: str
    python_version: str
    test_timestamp: str
    base_url: str
    total_tests: int = 0
    passed_tests: int = 0
    failed_tests: int = 0
    test_results: list = field(default_factory=list)
    statistics: dict = field(default_factory=dict)
    
    def add_result(self, result: TestResult):
        self.test_results.append(asdict(result))
        self.total_tests += 1
        if result.passed:
            self.passed_tests += 1
        else:
            self.failed_tests += 1


# ============================================================================
# TEST DEFINITIONS
# ============================================================================

# Test T1.1: Privacy Detection Test Cases
PRIVACY_DETECTION_TESTS = [
    # Kategorie 1: IBAN (sollte PRIVATE sein, Colossus)
    {
        "id": "T1.1-01",
        "category": "IBAN",
        "input": "Meine IBAN ist DE89370400440532013000",
        "expected_privacy": "private",
        "expected_provider": "colossus"
    },
    {
        "id": "T1.1-02",
        "category": "IBAN",
        "input": "Überweise auf DE12500105170648489890",
        "expected_privacy": "private",
        "expected_provider": "colossus"
    },
    # Kategorie 2: Kreditkarten (sollte PRIVATE sein, Colossus)
    {
        "id": "T1.1-03",
        "category": "Kreditkarte",
        "input": "Meine Kreditkartennummer ist 4111111111111111",
        "expected_privacy": "private",
        "expected_provider": "colossus"
    },
    {
        "id": "T1.1-04",
        "category": "Kreditkarte",
        "input": "Zahlung mit Karte 5555555555554444",
        "expected_privacy": "private",
        "expected_provider": "colossus"
    },
    # Kategorie 3: Medizinische Daten (sollte PRIVATE sein, Colossus)
    {
        "id": "T1.1-05",
        "category": "Medizin",
        "input": "Der Patient hat Diabetes Typ 2 und nimmt Metformin",
        "expected_privacy": "private",
        "expected_provider": "colossus"
    },
    {
        "id": "T1.1-06",
        "category": "Medizin",
        "input": "Diagnose: Hypertonie Grad II, Blutdruck 160/95",
        "expected_privacy": "private",
        "expected_provider": "colossus"
    },
    {
        "id": "T1.1-07",
        "category": "Medizin",
        "input": "Was sind die Nebenwirkungen von Ibuprofen?",
        "expected_privacy": "private",
        "expected_provider": "colossus"
    },
    # Kategorie 4: Finanzdaten (sollte PRIVATE/CONFIDENTIAL sein, Colossus)
    {
        "id": "T1.1-08",
        "category": "Finanzen",
        "input": "Mein Gehalt beträgt 75000 Euro pro Jahr",
        "expected_privacy": "private",
        "expected_provider": "colossus"
    },
    {
        "id": "T1.1-09",
        "category": "Finanzen",
        "input": "Der Kontostand ist 12500 Euro",
        "expected_privacy": "private",
        "expected_provider": "colossus"
    },
    # Kategorie 5: Nicht-sensible Daten (sollte PUBLIC sein, OpenRouter)
    {
        "id": "T1.1-10",
        "category": "Nicht-sensibel",
        "input": "Was ist REST?",
        "expected_privacy": "public",
        "expected_provider": "openrouter"
    },
    {
        "id": "T1.1-11",
        "category": "Nicht-sensibel",
        "input": "Erkläre mir Docker in einem Satz",
        "expected_privacy": "public",
        "expected_provider": "openrouter"
    },
    {
        "id": "T1.1-12",
        "category": "Nicht-sensibel",
        "input": "Was sind die Vorteile von Python?",
        "expected_privacy": "public",
        "expected_provider": "openrouter"
    },
    {
        "id": "T1.1-13",
        "category": "Nicht-sensibel",
        "input": "Das Wetter ist heute schön",
        "expected_privacy": "public",
        "expected_provider": "openrouter"
    },
    {
        "id": "T1.1-14",
        "category": "Nicht-sensibel",
        "input": "Wie funktioniert Machine Learning?",
        "expected_privacy": "public",
        "expected_provider": "openrouter"
    },
]

# Test T4.1: Simple Query Latency Tests
LATENCY_TESTS = [
    {"id": "T4.1-01", "input": "Was ist REST?"},
    {"id": "T4.1-02", "input": "Erkläre Docker in einem Satz"},
    {"id": "T4.1-03", "input": "Was sind die Vorteile von Python?"},
    {"id": "T4.1-04", "input": "Beschreibe API-Design"},
    {"id": "T4.1-05", "input": "Was ist Machine Learning?"},
]

# Test T4.3: Multi-Agent Coordination Tests
MULTI_AGENT_TESTS = [
    {
        "id": "T4.3-01",
        "category": "Medizin",
        "input": "Was sind die Nebenwirkungen von Ibuprofen?",
        "expected_specialist": "lara_alesi"
    },
    {
        "id": "T4.3-02",
        "category": "Entwicklung",
        "input": "Schreibe eine Python-Funktion für Fibonacci",
        "expected_specialist": "john_alesi"
    },
    {
        "id": "T4.3-03",
        "category": "Finanzen",
        "input": "Wie berechne ich den ROI einer Investition?",
        "expected_specialist": "theo_alesi"
    },
]


# ============================================================================
# TEST FUNCTIONS
# ============================================================================

def check_api_health(base_url: str) -> bool:
    """Prüft, ob die SAAP API erreichbar ist"""
    try:
        response = requests.get(f"{base_url}/health", timeout=5)
        return response.status_code == 200
    except requests.exceptions.RequestException:
        return False


def run_privacy_detection_test(base_url: str, test_case: dict) -> TestResult:
    """Führt einen einzelnen Privacy Detection Test durch"""
    start_time = time.time()
    
    try:
        response = requests.post(
            f"{base_url}/api/v1/multi-agent/chat",
            json={
                "user_message": test_case["input"],
                "privacy_mode": "auto"
            },
            timeout=60
        )
        latency = time.time() - start_time
        
        if response.status_code == 200:
            data = response.json()
            
            # Extrahiere Privacy-Informationen
            privacy_info = data.get("privacy_protection", {})
            actual_privacy = privacy_info.get("privacy_level", "unknown")
            actual_provider = privacy_info.get("selected_provider", "unknown")
            
            # Prüfe ob Provider korrekt ist (Hauptkriterium für DSGVO)
            provider_correct = actual_provider.lower() == test_case["expected_provider"].lower()
            
            return TestResult(
                test_id=test_case["id"],
                test_name=f"Privacy Detection: {test_case['category']}",
                category=test_case["category"],
                input_data=test_case["input"],
                expected_result=f"privacy={test_case['expected_privacy']}, provider={test_case['expected_provider']}",
                actual_result=f"privacy={actual_privacy}, provider={actual_provider}",
                passed=provider_correct,
                latency_seconds=round(latency, 3),
                provider=actual_provider,
                privacy_level=actual_privacy,
                raw_response=data
            )
        else:
            return TestResult(
                test_id=test_case["id"],
                test_name=f"Privacy Detection: {test_case['category']}",
                category=test_case["category"],
                input_data=test_case["input"],
                expected_result=test_case["expected_provider"],
                actual_result="API Error",
                passed=False,
                latency_seconds=round(latency, 3),
                error_message=f"HTTP {response.status_code}: {response.text[:200]}"
            )
            
    except requests.exceptions.Timeout:
        return TestResult(
            test_id=test_case["id"],
            test_name=f"Privacy Detection: {test_case['category']}",
            category=test_case["category"],
            input_data=test_case["input"],
            expected_result=test_case["expected_provider"],
            actual_result="Timeout",
            passed=False,
            latency_seconds=60.0,
            error_message="Request timeout after 60s"
        )
    except Exception as e:
        return TestResult(
            test_id=test_case["id"],
            test_name=f"Privacy Detection: {test_case['category']}",
            category=test_case["category"],
            input_data=test_case["input"],
            expected_result=test_case["expected_provider"],
            actual_result="Error",
            passed=False,
            latency_seconds=0,
            error_message=str(e)
        )


def run_latency_test(base_url: str, test_case: dict) -> TestResult:
    """Führt einen einzelnen Latenz-Test durch"""
    start_time = time.time()
    
    try:
        response = requests.post(
            f"{base_url}/api/v1/multi-agent/chat",
            json={
                "user_message": test_case["input"],
                "privacy_mode": "auto"
            },
            timeout=60
        )
        latency = time.time() - start_time
        
        if response.status_code == 200:
            data = response.json()
            provider = data.get("response", {}).get("provider", "unknown")
            
            # Zielwert: < 5s für einfache Anfragen
            target_latency = 5.0
            passed = latency < target_latency
            
            return TestResult(
                test_id=test_case["id"],
                test_name=f"Simple Query Latency",
                category="Latency",
                input_data=test_case["input"],
                expected_result=f"< {target_latency}s",
                actual_result=f"{round(latency, 2)}s",
                passed=passed,
                latency_seconds=round(latency, 3),
                provider=provider,
                raw_response=data
            )
        else:
            return TestResult(
                test_id=test_case["id"],
                test_name=f"Simple Query Latency",
                category="Latency",
                input_data=test_case["input"],
                expected_result="< 5s",
                actual_result="API Error",
                passed=False,
                latency_seconds=round(time.time() - start_time, 3),
                error_message=f"HTTP {response.status_code}"
            )
            
    except Exception as e:
        return TestResult(
            test_id=test_case["id"],
            test_name=f"Simple Query Latency",
            category="Latency",
            input_data=test_case["input"],
            expected_result="< 5s",
            actual_result="Error",
            passed=False,
            latency_seconds=0,
            error_message=str(e)
        )


def run_multi_agent_test(base_url: str, test_case: dict) -> TestResult:
    """Führt einen Multi-Agent Coordination Test durch"""
    start_time = time.time()
    
    try:
        response = requests.post(
            f"{base_url}/api/v1/multi-agent/chat",
            json={
                "user_message": test_case["input"],
                "privacy_mode": "auto"
            },
            timeout=60
        )
        latency = time.time() - start_time
        
        if response.status_code == 200:
            data = response.json()
            
            specialist = data.get("specialist_agent", "unknown")
            delegation_used = data.get("delegation_used", False)
            provider = data.get("response", {}).get("provider", "unknown")
            
            # Zielwert: < 30s für Multi-Agent
            target_latency = 30.0
            latency_ok = latency < target_latency
            
            # Prüfe ob korrekter Spezialist delegiert wurde
            specialist_correct = specialist == test_case["expected_specialist"]
            
            return TestResult(
                test_id=test_case["id"],
                test_name=f"Multi-Agent Coordination: {test_case['category']}",
                category=test_case["category"],
                input_data=test_case["input"],
                expected_result=f"specialist={test_case['expected_specialist']}, latency<{target_latency}s",
                actual_result=f"specialist={specialist}, latency={round(latency, 2)}s",
                passed=latency_ok,  # Hauptkriterium ist Latenz
                latency_seconds=round(latency, 3),
                provider=provider,
                raw_response=data
            )
        else:
            return TestResult(
                test_id=test_case["id"],
                test_name=f"Multi-Agent Coordination: {test_case['category']}",
                category=test_case["category"],
                input_data=test_case["input"],
                expected_result=test_case["expected_specialist"],
                actual_result="API Error",
                passed=False,
                latency_seconds=round(latency, 3),
                error_message=f"HTTP {response.status_code}"
            )
            
    except Exception as e:
        return TestResult(
            test_id=test_case["id"],
            test_name=f"Multi-Agent Coordination: {test_case['category']}",
            category=test_case["category"],
            input_data=test_case["input"],
            expected_result=test_case["expected_specialist"],
            actual_result="Error",
            passed=False,
            latency_seconds=0,
            error_message=str(e)
        )


def calculate_statistics(results: TestSuiteResults) -> dict:
    """Berechnet Statistiken aus den Testergebnissen"""
    stats = {
        "total": results.total_tests,
        "passed": results.passed_tests,
        "failed": results.failed_tests,
        "pass_rate": round(results.passed_tests / max(results.total_tests, 1) * 100, 2)
    }
    
    # Latenz-Statistiken
    latencies = [r["latency_seconds"] for r in results.test_results if r["latency_seconds"] > 0]
    if latencies:
        stats["latency"] = {
            "min": round(min(latencies), 3),
            "max": round(max(latencies), 3),
            "avg": round(statistics.mean(latencies), 3),
            "median": round(statistics.median(latencies), 3),
            "stdev": round(statistics.stdev(latencies), 3) if len(latencies) > 1 else 0
        }
    
    # Privacy Detection Statistiken
    privacy_tests = [r for r in results.test_results if "Privacy Detection" in r["test_name"]]
    if privacy_tests:
        tp = sum(1 for r in privacy_tests if r["passed"] and r["category"] != "Nicht-sensibel")
        tn = sum(1 for r in privacy_tests if r["passed"] and r["category"] == "Nicht-sensibel")
        fp = sum(1 for r in privacy_tests if not r["passed"] and r["category"] == "Nicht-sensibel")
        fn = sum(1 for r in privacy_tests if not r["passed"] and r["category"] != "Nicht-sensibel")
        
        precision = tp / max(tp + fp, 1)
        recall = tp / max(tp + fn, 1)
        f1 = 2 * precision * recall / max(precision + recall, 0.001)
        
        stats["privacy_detection"] = {
            "true_positives": tp,
            "true_negatives": tn,
            "false_positives": fp,
            "false_negatives": fn,
            "precision": round(precision, 4),
            "recall": round(recall, 4),
            "f1_score": round(f1, 4)
        }
    
    return stats


# ============================================================================
# MAIN EXECUTION
# ============================================================================

def run_all_tests(base_url: str, verbose: bool = True) -> TestSuiteResults:
    """Führt alle Tests aus und gibt die Ergebnisse zurück"""
    
    # Initialisiere Ergebnisse
    results = TestSuiteResults(
        platform=platform.system(),
        os_version=platform.version(),
        python_version=platform.python_version(),
        test_timestamp=datetime.now().isoformat(),
        base_url=base_url
    )
    
    print("=" * 70)
    print("SAAP EVALUATION TEST SUITE")
    print("=" * 70)
    print(f"Platform: {results.platform} ({results.os_version})")
    print(f"Python: {results.python_version}")
    print(f"Base URL: {base_url}")
    print(f"Timestamp: {results.test_timestamp}")
    print("=" * 70)
    
    # Health Check
    print("\n[1/4] Prüfe API-Verfügbarkeit...")
    if not check_api_health(base_url):
        print("ERROR: SAAP API nicht erreichbar!")
        print(f"       Stellen Sie sicher, dass 'docker compose up -d' ausgeführt wurde.")
        print(f"       URL: {base_url}/health")
        return results
    print("✓ API ist erreichbar\n")
    
    # Test T1.1: Privacy Detection
    print("[2/4] Führe Privacy Detection Tests durch (T1.1)...")
    print("-" * 50)
    for i, test_case in enumerate(PRIVACY_DETECTION_TESTS, 1):
        result = run_privacy_detection_test(base_url, test_case)
        results.add_result(result)
        status = "✓" if result.passed else "✗"
        if verbose:
            print(f"  [{i}/{len(PRIVACY_DETECTION_TESTS)}] {test_case['id']}: {status} "
                  f"({result.provider}, {result.latency_seconds}s)")
    
    # Test T4.1: Simple Query Latency
    print(f"\n[3/4] Führe Latenz-Tests durch (T4.1)...")
    print("-" * 50)
    for i, test_case in enumerate(LATENCY_TESTS, 1):
        result = run_latency_test(base_url, test_case)
        results.add_result(result)
        status = "✓" if result.passed else "✗"
        if verbose:
            print(f"  [{i}/{len(LATENCY_TESTS)}] {test_case['id']}: {status} "
                  f"({result.latency_seconds}s)")
    
    # Test T4.3: Multi-Agent Coordination
    print(f"\n[4/4] Führe Multi-Agent Tests durch (T4.3)...")
    print("-" * 50)
    for i, test_case in enumerate(MULTI_AGENT_TESTS, 1):
        result = run_multi_agent_test(base_url, test_case)
        results.add_result(result)
        status = "✓" if result.passed else "✗"
        if verbose:
            print(f"  [{i}/{len(MULTI_AGENT_TESTS)}] {test_case['id']}: {status} "
                  f"({result.latency_seconds}s)")
    
    # Berechne Statistiken
    results.statistics = calculate_statistics(results)
    
    # Ausgabe Zusammenfassung
    print("\n" + "=" * 70)
    print("ZUSAMMENFASSUNG")
    print("=" * 70)
    print(f"Gesamt: {results.total_tests} Tests")
    print(f"Bestanden: {results.passed_tests} ({results.statistics['pass_rate']}%)")
    print(f"Fehlgeschlagen: {results.failed_tests}")
    
    if "latency" in results.statistics:
        lat = results.statistics["latency"]
        print(f"\nLatenz-Statistiken:")
        print(f"  Min: {lat['min']}s | Max: {lat['max']}s | Avg: {lat['avg']}s | Median: {lat['median']}s")
    
    if "privacy_detection" in results.statistics:
        pd = results.statistics["privacy_detection"]
        print(f"\nPrivacy Detection:")
        print(f"  F1-Score: {pd['f1_score']} | Precision: {pd['precision']} | Recall: {pd['recall']}")
    
    print("=" * 70)
    
    return results


def main():
    parser = argparse.ArgumentParser(
        description="SAAP Evaluation Test Suite - Cross-Platform",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Beispiele:
  python evaluation_tests.py                    # Standard (localhost:8000)
  python evaluation_tests.py --base-url http://192.168.1.100:8000
  python evaluation_tests.py --output linux_results.json
  python evaluation_tests.py --quiet
        """
    )
    parser.add_argument(
        "--base-url", "-u",
        default="http://localhost:8000",
        help="SAAP API Base URL (default: http://localhost:8000)"
    )
    parser.add_argument(
        "--output", "-o",
        default=None,
        help="Output JSON file (default: results_<platform>_<timestamp>.json)"
    )
    parser.add_argument(
        "--quiet", "-q",
        action="store_true",
        help="Weniger Ausgabe"
    )
    
    args = parser.parse_args()
    
    # Tests ausführen
    results = run_all_tests(args.base_url, verbose=not args.quiet)
    
    # Ergebnisse speichern
    if args.output:
        output_file = args.output
    else:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_file = f"results_{platform.system().lower()}_{timestamp}.json"
    
    output_path = Path(output_file)
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(asdict(results), f, indent=2, ensure_ascii=False)
    
    print(f"\nErgebnisse gespeichert in: {output_path.absolute()}")
    
    # Exit-Code basierend auf Testergebnissen
    if results.failed_tests > 0:
        sys.exit(1)
    sys.exit(0)


if __name__ == "__main__":
    main()

	#!/usr/bin/env python3
	"""
	SAAP Evaluation Test Suite - Cross-Platform (Linux/Windows)
	===========================================================

	Dieses Skript führt alle Tests für Kapitel 4.4 der Masterarbeit aus.
	Es ist plattformunabhängig und speichert Ergebnisse in JSON-Format.

	Verwendung:
	python evaluation_tests.py [--base-url http://localhost:8000] [--output results.json]

	Voraussetzungen:
	- SAAP Docker-Container müssen laufen: docker compose up -d
	- Python 3.10+ mit requests-Bibliothek

	Installation:
	pip install requests

	Autor: Hanan W.
	Version: 1.0
	Datum: 20. Januar 2026
	"""

	import argparse
	import json
	import platform
	import statistics
	import sys
	import time
	from dataclasses import asdict, dataclass, field
	from datetime import datetime
	from pathlib import Path
	from typing import Any

	try:
	import requests
	except ImportError:
	print("ERROR: 'requests' library not found. Install with: pip install requests")
	sys.exit(1)


	# ============================================================================
	# DATA CLASSES
	# ============================================================================

	@dataclass
	class TestResult:
	"""Einzelnes Testergebnis"""
	test_id: str
	test_name: str
	category: str
	input_data: str
	expected_result: str
	actual_result: str
	passed: bool
	latency_seconds: float
	provider: str = ""
	privacy_level: str = ""
	error_message: str = ""
	raw_response: dict = field(default_factory=dict)


	@dataclass
	class TestSuiteResults:
	"""Gesamtergebnisse der Testsuite"""
	platform: str
	os_version: str
	python_version: str
	test_timestamp: str
	base_url: str
	total_tests: int = 0
	passed_tests: int = 0
	failed_tests: int = 0
	test_results: list = field(default_factory=list)
	statistics: dict = field(default_factory=dict)

	def add_result(self, result: TestResult):
	self.test_results.append(asdict(result))
	self.total_tests += 1
	if result.passed:
	self.passed_tests += 1
	else:
	self.failed_tests += 1


	# ============================================================================
	# TEST DEFINITIONS
	# ============================================================================

	# Test T1.1: Privacy Detection Test Cases
	PRIVACY_DETECTION_TESTS = [
	# Kategorie 1: IBAN (sollte PRIVATE sein, Colossus)
	{
	"id": "T1.1-01",
	"category": "IBAN",
	"input": "Meine IBAN ist DE89370400440532013000",
	"expected_privacy": "private",
	"expected_provider": "colossus"
	},
	{
	"id": "T1.1-02",
	"category": "IBAN",
	"input": "Überweise auf DE12500105170648489890",
	"expected_privacy": "private",
	"expected_provider": "colossus"
	},
	# Kategorie 2: Kreditkarten (sollte PRIVATE sein, Colossus)
	{
	"id": "T1.1-03",
	"category": "Kreditkarte",
	"input": "Meine Kreditkartennummer ist 4111111111111111",
	"expected_privacy": "private",
	"expected_provider": "colossus"
	},
	{
	"id": "T1.1-04",
	"category": "Kreditkarte",
	"input": "Zahlung mit Karte 5555555555554444",
	"expected_privacy": "private",
	"expected_provider": "colossus"
	},
	# Kategorie 3: Medizinische Daten (sollte PRIVATE sein, Colossus)
	{
	"id": "T1.1-05",
	"category": "Medizin",
	"input": "Der Patient hat Diabetes Typ 2 und nimmt Metformin",
	"expected_privacy": "private",
	"expected_provider": "colossus"
	},
	{
	"id": "T1.1-06",
	"category": "Medizin",
	"input": "Diagnose: Hypertonie Grad II, Blutdruck 160/95",
	"expected_privacy": "private",
	"expected_provider": "colossus"
	},
	{
	"id": "T1.1-07",
	"category": "Medizin",
	"input": "Was sind die Nebenwirkungen von Ibuprofen?",
	"expected_privacy": "private",
	"expected_provider": "colossus"
	},
	# Kategorie 4: Finanzdaten (sollte PRIVATE/CONFIDENTIAL sein, Colossus)
	{
	"id": "T1.1-08",
	"category": "Finanzen",
	"input": "Mein Gehalt beträgt 75000 Euro pro Jahr",
	"expected_privacy": "private",
	"expected_provider": "colossus"
	},
	{
	"id": "T1.1-09",
	"category": "Finanzen",
	"input": "Der Kontostand ist 12500 Euro",
	"expected_privacy": "private",
	"expected_provider": "colossus"
	},
	# Kategorie 5: Nicht-sensible Daten (sollte PUBLIC sein, OpenRouter)
	{
	"id": "T1.1-10",
	"category": "Nicht-sensibel",
	"input": "Was ist REST?",
	"expected_privacy": "public",
	"expected_provider": "openrouter"
	},
	{
	"id": "T1.1-11",
	"category": "Nicht-sensibel",
	"input": "Erkläre mir Docker in einem Satz",
	"expected_privacy": "public",
	"expected_provider": "openrouter"
	},
	{
	"id": "T1.1-12",
	"category": "Nicht-sensibel",
	"input": "Was sind die Vorteile von Python?",
	"expected_privacy": "public",
	"expected_provider": "openrouter"
	},
	{
	"id": "T1.1-13",
	"category": "Nicht-sensibel",
	"input": "Das Wetter ist heute schön",
	"expected_privacy": "public",
	"expected_provider": "openrouter"
	},
	{
	"id": "T1.1-14",
	"category": "Nicht-sensibel",
	"input": "Wie funktioniert Machine Learning?",
	"expected_privacy": "public",
	"expected_provider": "openrouter"
	},
	]

	# Test T4.1: Simple Query Latency Tests
	LATENCY_TESTS = [
	{"id": "T4.1-01", "input": "Was ist REST?"},
	{"id": "T4.1-02", "input": "Erkläre Docker in einem Satz"},
	{"id": "T4.1-03", "input": "Was sind die Vorteile von Python?"},
	{"id": "T4.1-04", "input": "Beschreibe API-Design"},
	{"id": "T4.1-05", "input": "Was ist Machine Learning?"},
	]

	# Test T4.3: Multi-Agent Coordination Tests
	MULTI_AGENT_TESTS = [
	{
	"id": "T4.3-01",
	"category": "Medizin",
	"input": "Was sind die Nebenwirkungen von Ibuprofen?",
	"expected_specialist": "lara_alesi"
	},
	{
	"id": "T4.3-02",
	"category": "Entwicklung",
	"input": "Schreibe eine Python-Funktion für Fibonacci",
	"expected_specialist": "john_alesi"
	},
	{
	"id": "T4.3-03",
	"category": "Finanzen",
	"input": "Wie berechne ich den ROI einer Investition?",
	"expected_specialist": "theo_alesi"
	},
	]


	# ============================================================================
	# TEST FUNCTIONS
	# ============================================================================

	def check_api_health(base_url: str) -> bool:
	"""Prüft, ob die SAAP API erreichbar ist"""
	try:
	response = requests.get(f"{base_url}/health", timeout=5)
	return response.status_code == 200
	except requests.exceptions.RequestException:
	return False


	def run_privacy_detection_test(base_url: str, test_case: dict) -> TestResult:
	"""Führt einen einzelnen Privacy Detection Test durch"""
	start_time = time.time()

	try:
	response = requests.post(
	f"{base_url}/api/v1/multi-agent/chat",
	json={
	"user_message": test_case["input"],
	"privacy_mode": "auto"
	},
	timeout=60
	)
	latency = time.time() - start_time

	if response.status_code == 200:
	data = response.json()

	# Extrahiere Privacy-Informationen
	privacy_info = data.get("privacy_protection", {})
	actual_privacy = privacy_info.get("privacy_level", "unknown")
	actual_provider = privacy_info.get("selected_provider", "unknown")

	# Prüfe ob Provider korrekt ist (Hauptkriterium für DSGVO)
	provider_correct = actual_provider.lower() == test_case["expected_provider"].lower()

	return TestResult(
	test_id=test_case["id"],
	test_name=f"Privacy Detection: {test_case['category']}",
	category=test_case["category"],
	input_data=test_case["input"],
	expected_result=f"privacy={test_case['expected_privacy']}, provider={test_case['expected_provider']}",
	actual_result=f"privacy={actual_privacy}, provider={actual_provider}",
	passed=provider_correct,
	latency_seconds=round(latency, 3),
	provider=actual_provider,
	privacy_level=actual_privacy,
	raw_response=data
	)
	else:
	return TestResult(
	test_id=test_case["id"],
	test_name=f"Privacy Detection: {test_case['category']}",
	category=test_case["category"],
	input_data=test_case["input"],
	expected_result=test_case["expected_provider"],
	actual_result="API Error",
	passed=False,
	latency_seconds=round(latency, 3),
	error_message=f"HTTP {response.status_code}: {response.text[:200]}"
	)

	except requests.exceptions.Timeout:
	return TestResult(
	test_id=test_case["id"],
	test_name=f"Privacy Detection: {test_case['category']}",
	category=test_case["category"],
	input_data=test_case["input"],
	expected_result=test_case["expected_provider"],
	actual_result="Timeout",
	passed=False,
	latency_seconds=60.0,
	error_message="Request timeout after 60s"
	)
	except Exception as e:
	return TestResult(
	test_id=test_case["id"],
	test_name=f"Privacy Detection: {test_case['category']}",
	category=test_case["category"],
	input_data=test_case["input"],
	expected_result=test_case["expected_provider"],
	actual_result="Error",
	passed=False,
	latency_seconds=0,
	error_message=str(e)
	)


	def run_latency_test(base_url: str, test_case: dict) -> TestResult:
	"""Führt einen einzelnen Latenz-Test durch"""
	start_time = time.time()

	try:
	response = requests.post(
	f"{base_url}/api/v1/multi-agent/chat",
	json={
	"user_message": test_case["input"],
	"privacy_mode": "auto"
	},
	timeout=60
	)
	latency = time.time() - start_time

	if response.status_code == 200:
	data = response.json()
	provider = data.get("response", {}).get("provider", "unknown")

	# Zielwert: < 5s für einfache Anfragen
	target_latency = 5.0
	passed = latency < target_latency

	return TestResult(
	test_id=test_case["id"],
	test_name=f"Simple Query Latency",
	category="Latency",
	input_data=test_case["input"],
	expected_result=f"< {target_latency}s",
	actual_result=f"{round(latency, 2)}s",
	passed=passed,
	latency_seconds=round(latency, 3),
	provider=provider,
	raw_response=data
	)
	else:
	return TestResult(
	test_id=test_case["id"],
	test_name=f"Simple Query Latency",
	category="Latency",
	input_data=test_case["input"],
	expected_result="< 5s",
	actual_result="API Error",
	passed=False,
	latency_seconds=round(time.time() - start_time, 3),
	error_message=f"HTTP {response.status_code}"
	)

	except Exception as e:
	return TestResult(
	test_id=test_case["id"],
	test_name=f"Simple Query Latency",
	category="Latency",
	input_data=test_case["input"],
	expected_result="< 5s",
	actual_result="Error",
	passed=False,
	latency_seconds=0,
	error_message=str(e)
	)


	def run_multi_agent_test(base_url: str, test_case: dict) -> TestResult:
	"""Führt einen Multi-Agent Coordination Test durch"""
	start_time = time.time()

	try:
	response = requests.post(
	f"{base_url}/api/v1/multi-agent/chat",
	json={
	"user_message": test_case["input"],
	"privacy_mode": "auto"
	},
	timeout=60
	)
	latency = time.time() - start_time

	if response.status_code == 200:
	data = response.json()

	specialist = data.get("specialist_agent", "unknown")
	delegation_used = data.get("delegation_used", False)
	provider = data.get("response", {}).get("provider", "unknown")

	# Zielwert: < 30s für Multi-Agent
	target_latency = 30.0
	latency_ok = latency < target_latency

	# Prüfe ob korrekter Spezialist delegiert wurde
	specialist_correct = specialist == test_case["expected_specialist"]

	return TestResult(
	test_id=test_case["id"],
	test_name=f"Multi-Agent Coordination: {test_case['category']}",
	category=test_case["category"],
	input_data=test_case["input"],
	expected_result=f"specialist={test_case['expected_specialist']}, latency<{target_latency}s",
	actual_result=f"specialist={specialist}, latency={round(latency, 2)}s",
	passed=latency_ok, # Hauptkriterium ist Latenz
	latency_seconds=round(latency, 3),
	provider=provider,
	raw_response=data
	)
	else:
	return TestResult(
	test_id=test_case["id"],
	test_name=f"Multi-Agent Coordination: {test_case['category']}",
	category=test_case["category"],
	input_data=test_case["input"],
	expected_result=test_case["expected_specialist"],
	actual_result="API Error",
	passed=False,
	latency_seconds=round(latency, 3),
	error_message=f"HTTP {response.status_code}"
	)

	except Exception as e:
	return TestResult(
	test_id=test_case["id"],
	test_name=f"Multi-Agent Coordination: {test_case['category']}",
	category=test_case["category"],
	input_data=test_case["input"],
	expected_result=test_case["expected_specialist"],
	actual_result="Error",
	passed=False,
	latency_seconds=0,
	error_message=str(e)
	)


	def calculate_statistics(results: TestSuiteResults) -> dict:
	"""Berechnet Statistiken aus den Testergebnissen"""
	stats = {
	"total": results.total_tests,
	"passed": results.passed_tests,
	"failed": results.failed_tests,
	"pass_rate": round(results.passed_tests / max(results.total_tests, 1) * 100, 2)
	}

	# Latenz-Statistiken
	latencies = [r["latency_seconds"] for r in results.test_results if r["latency_seconds"] > 0]
	if latencies:
	stats["latency"] = {
	"min": round(min(latencies), 3),
	"max": round(max(latencies), 3),
	"avg": round(statistics.mean(latencies), 3),
	"median": round(statistics.median(latencies), 3),
	"stdev": round(statistics.stdev(latencies), 3) if len(latencies) > 1 else 0
	}

	# Privacy Detection Statistiken
	privacy_tests = [r for r in results.test_results if "Privacy Detection" in r["test_name"]]
	if privacy_tests:
	tp = sum(1 for r in privacy_tests if r["passed"] and r["category"] != "Nicht-sensibel")
	tn = sum(1 for r in privacy_tests if r["passed"] and r["category"] == "Nicht-sensibel")
	fp = sum(1 for r in privacy_tests if not r["passed"] and r["category"] == "Nicht-sensibel")
	fn = sum(1 for r in privacy_tests if not r["passed"] and r["category"] != "Nicht-sensibel")

	precision = tp / max(tp + fp, 1)
	recall = tp / max(tp + fn, 1)
	f1 = 2 * precision * recall / max(precision + recall, 0.001)

	stats["privacy_detection"] = {
	"true_positives": tp,
	"true_negatives": tn,
	"false_positives": fp,
	"false_negatives": fn,
	"precision": round(precision, 4),
	"recall": round(recall, 4),
	"f1_score": round(f1, 4)
	}

	return stats


	# ============================================================================
	# MAIN EXECUTION
	# ============================================================================

	def run_all_tests(base_url: str, verbose: bool = True) -> TestSuiteResults:
	"""Führt alle Tests aus und gibt die Ergebnisse zurück"""

	# Initialisiere Ergebnisse
	results = TestSuiteResults(
	platform=platform.system(),
	os_version=platform.version(),
	python_version=platform.python_version(),
	test_timestamp=datetime.now().isoformat(),
	base_url=base_url
	)

	print("=" * 70)
	print("SAAP EVALUATION TEST SUITE")
	print("=" * 70)
	print(f"Platform: {results.platform} ({results.os_version})")
	print(f"Python: {results.python_version}")
	print(f"Base URL: {base_url}")
	print(f"Timestamp: {results.test_timestamp}")
	print("=" * 70)

	# Health Check
	print("\n[1/4] Prüfe API-Verfügbarkeit...")
	if not check_api_health(base_url):
	print("ERROR: SAAP API nicht erreichbar!")
	print(f" Stellen Sie sicher, dass 'docker compose up -d' ausgeführt wurde.")
	print(f" URL: {base_url}/health")
	return results
	print("✓ API ist erreichbar\n")

	# Test T1.1: Privacy Detection
	print("[2/4] Führe Privacy Detection Tests durch (T1.1)...")
	print("-" * 50)
	for i, test_case in enumerate(PRIVACY_DETECTION_TESTS, 1):
	result = run_privacy_detection_test(base_url, test_case)
	results.add_result(result)
	status = "✓" if result.passed else "✗"
	if verbose:
	print(f" [{i}/{len(PRIVACY_DETECTION_TESTS)}] {test_case['id']}: {status} "
	f"({result.provider}, {result.latency_seconds}s)")

	# Test T4.1: Simple Query Latency
	print(f"\n[3/4] Führe Latenz-Tests durch (T4.1)...")
	print("-" * 50)
	for i, test_case in enumerate(LATENCY_TESTS, 1):
	result = run_latency_test(base_url, test_case)
	results.add_result(result)
	status = "✓" if result.passed else "✗"
	if verbose:
	print(f" [{i}/{len(LATENCY_TESTS)}] {test_case['id']}: {status} "
	f"({result.latency_seconds}s)")

	# Test T4.3: Multi-Agent Coordination
	print(f"\n[4/4] Führe Multi-Agent Tests durch (T4.3)...")
	print("-" * 50)
	for i, test_case in enumerate(MULTI_AGENT_TESTS, 1):
	result = run_multi_agent_test(base_url, test_case)
	results.add_result(result)
	status = "✓" if result.passed else "✗"
	if verbose:
	print(f" [{i}/{len(MULTI_AGENT_TESTS)}] {test_case['id']}: {status} "
	f"({result.latency_seconds}s)")

	# Berechne Statistiken
	results.statistics = calculate_statistics(results)

	# Ausgabe Zusammenfassung
	print("\n" + "=" * 70)
	print("ZUSAMMENFASSUNG")
	print("=" * 70)
	print(f"Gesamt: {results.total_tests} Tests")
	print(f"Bestanden: {results.passed_tests} ({results.statistics['pass_rate']}%)")
	print(f"Fehlgeschlagen: {results.failed_tests}")

	if "latency" in results.statistics:
	lat = results.statistics["latency"]
	print(f"\nLatenz-Statistiken:")
	print(f" Min: {lat['min']}s \| Max: {lat['max']}s \| Avg: {lat['avg']}s \| Median: {lat['median']}s")

	if "privacy_detection" in results.statistics:
	pd = results.statistics["privacy_detection"]
	print(f"\nPrivacy Detection:")
	print(f" F1-Score: {pd['f1_score']} \| Precision: {pd['precision']} \| Recall: {pd['recall']}")

	print("=" * 70)

	return results


	def main():
	parser = argparse.ArgumentParser(
	description="SAAP Evaluation Test Suite - Cross-Platform",
	formatter_class=argparse.RawDescriptionHelpFormatter,
	epilog="""
	Beispiele:
	python evaluation_tests.py # Standard (localhost:8000)
	python evaluation_tests.py --base-url http://192.168.1.100:8000
	python evaluation_tests.py --output linux_results.json
	python evaluation_tests.py --quiet
	"""
	)
	parser.add_argument(
	"--base-url", "-u",
	default="http://localhost:8000",
	help="SAAP API Base URL (default: http://localhost:8000)"
	)
	parser.add_argument(
	"--output", "-o",
	default=None,
	help="Output JSON file (default: results_<platform>_<timestamp>.json)"
	)
	parser.add_argument(
	"--quiet", "-q",
	action="store_true",
	help="Weniger Ausgabe"
	)

	args = parser.parse_args()

	# Tests ausführen
	results = run_all_tests(args.base_url, verbose=not args.quiet)

	# Ergebnisse speichern
	if args.output:
	output_file = args.output
	else:
	timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
	output_file = f"results_{platform.system().lower()}_{timestamp}.json"

	output_path = Path(output_file)
	with open(output_path, "w", encoding="utf-8") as f:
	json.dump(asdict(results), f, indent=2, ensure_ascii=False)

	print(f"\nErgebnisse gespeichert in: {output_path.absolute()}")

	# Exit-Code basierend auf Testergebnissen
	if results.failed_tests > 0:
	sys.exit(1)
	sys.exit(0)


	if __name__ == "__main__":
	main()