19855efacc
Some checks failed
Tests / Go Tests (push) Has been cancelled
Tests / Python Tests (push) Has been cancelled
Tests / Integration Tests (push) Has been cancelled
Tests / Go Lint (push) Has been cancelled
Tests / Python Lint (push) Has been cancelled
Tests / Security Scan (push) Has been cancelled
Tests / All Checks Passed (push) Has been cancelled
Security Scanning / Secret Scanning (push) Has been cancelled
Security Scanning / Dependency Vulnerability Scan (push) Has been cancelled
Security Scanning / Go Security Scan (push) Has been cancelled
Security Scanning / Python Security Scan (push) Has been cancelled
Security Scanning / Node.js Security Scan (push) Has been cancelled
Security Scanning / Docker Image Security (push) Has been cancelled
Security Scanning / Security Summary (push) Has been cancelled
CI/CD Pipeline / Go Tests (push) Has been cancelled
CI/CD Pipeline / Python Tests (push) Has been cancelled
CI/CD Pipeline / Website Tests (push) Has been cancelled
CI/CD Pipeline / Linting (push) Has been cancelled
CI/CD Pipeline / Security Scan (push) Has been cancelled
CI/CD Pipeline / Docker Build & Push (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Deploy to Staging (push) Has been cancelled
CI/CD Pipeline / Deploy to Production (push) Has been cancelled
CI/CD Pipeline / CI Summary (push) Has been cancelled
ci/woodpecker/manual/build-ci-image Pipeline was successful
ci/woodpecker/manual/main Pipeline failed
All services: admin-v2, studio-v2, website, ai-compliance-sdk, consent-service, klausur-service, voice-service, and infrastructure. Large PDFs and compiled binaries excluded via .gitignore.
739 lines
27 KiB
Python
739 lines
27 KiB
Python
"""
|
|
UI Test API
|
|
|
|
Provides endpoints for the interactive UI Test Wizard.
|
|
Allows testing of middleware components with educational feedback.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
|
|
# ==============================================
|
|
# Data Models
|
|
# ==============================================
|
|
|
|
|
|
class TestResult(BaseModel):
|
|
"""Result of a single test."""
|
|
|
|
name: str
|
|
description: str
|
|
expected: str
|
|
actual: str
|
|
status: str # "passed" | "failed" | "pending" | "skipped"
|
|
duration_ms: float = 0.0
|
|
error_message: Optional[str] = None
|
|
|
|
|
|
class TestCategoryResult(BaseModel):
|
|
"""Result of a test category."""
|
|
|
|
category: str
|
|
display_name: str
|
|
description: str
|
|
why_important: str
|
|
tests: List[TestResult]
|
|
passed: int
|
|
failed: int
|
|
total: int
|
|
duration_ms: float
|
|
|
|
|
|
class FullTestResults(BaseModel):
|
|
"""Full test results for all categories."""
|
|
|
|
timestamp: str
|
|
categories: List[TestCategoryResult]
|
|
total_passed: int
|
|
total_failed: int
|
|
total_tests: int
|
|
duration_ms: float
|
|
|
|
|
|
# ==============================================
|
|
# Educational Content
|
|
# ==============================================
|
|
|
|
EDUCATION_CONTENT = {
|
|
"request-id": {
|
|
"display_name": "Request-ID & Tracing",
|
|
"description": "Request-ID Middleware fuer Distributed Tracing",
|
|
"why_important": """
|
|
Stellen Sie sich vor, ein Benutzer meldet einen Fehler. Ohne Request-ID muessen
|
|
Sie Tausende von Log-Eintraegen durchsuchen. Mit Request-ID finden Sie den genauen
|
|
Pfad der Anfrage durch alle Microservices in Sekunden.
|
|
|
|
Request-IDs sind essentiell fuer:
|
|
- Fehlersuche in verteilten Systemen
|
|
- Performance-Analyse
|
|
- Audit-Trails fuer Compliance
|
|
""",
|
|
},
|
|
"security-headers": {
|
|
"display_name": "Security Headers",
|
|
"description": "HTTP Security Headers zum Schutz vor Web-Angriffen",
|
|
"why_important": """
|
|
Security Headers sind Anweisungen an den Browser, wie er Ihre Seite schuetzen soll:
|
|
|
|
- X-Content-Type-Options: nosniff - Verhindert MIME-Sniffing Angriffe
|
|
- X-Frame-Options: DENY - Blockiert Clickjacking-Angriffe
|
|
- Content-Security-Policy - Stoppt XSS durch Whitelist erlaubter Quellen
|
|
- Strict-Transport-Security - Erzwingt HTTPS
|
|
|
|
OWASP empfiehlt diese Headers als Mindeststandard. Sie sind Pflicht fuer
|
|
DSGVO-Konformitaet und schuetzen Ihre Benutzer vor gaengigen Angriffen.
|
|
""",
|
|
},
|
|
"rate-limiter": {
|
|
"display_name": "Rate Limiting",
|
|
"description": "Schutz vor Brute-Force und DDoS Angriffen",
|
|
"why_important": """
|
|
Ohne Rate Limiting kann ein Angreifer:
|
|
- Passwort-Brute-Force durchfuehren (Millionen Versuche/Minute)
|
|
- Ihre Server mit Anfragen ueberfluten (DDoS)
|
|
- Teure API-Aufrufe missbrauchen
|
|
|
|
BreakPilot limitiert:
|
|
- 100 Anfragen/Minute pro IP (allgemein)
|
|
- 20 Anfragen/Minute fuer Auth-Endpoints
|
|
- 500 Anfragen/Minute pro authentifiziertem Benutzer
|
|
""",
|
|
},
|
|
"pii-redactor": {
|
|
"display_name": "PII Redaktion",
|
|
"description": "Automatische Entfernung personenbezogener Daten aus Logs",
|
|
"why_important": """
|
|
Personenbezogene Daten in Logs sind ein DSGVO-Verstoss:
|
|
|
|
- Email-Adressen: Bussgelder bis 20 Mio. EUR
|
|
- IP-Adressen: Gelten als personenbezogen (EuGH-Urteil)
|
|
- Telefonnummern: Direkter Personenbezug
|
|
|
|
Der PII Redactor erkennt automatisch:
|
|
- Email-Adressen → [EMAIL_REDACTED]
|
|
- IP-Adressen → [IP_REDACTED]
|
|
- Deutsche Telefonnummern → [PHONE_REDACTED]
|
|
- IBAN-Nummern → [IBAN_REDACTED]
|
|
""",
|
|
},
|
|
"input-gate": {
|
|
"display_name": "Input Validierung",
|
|
"description": "Validierung eingehender Anfragen und Uploads",
|
|
"why_important": """
|
|
Das Input Gate prueft jede Anfrage bevor sie Ihren Code erreicht:
|
|
|
|
- Groessenlimit: Blockiert ueberdimensionierte Payloads (DoS-Schutz)
|
|
- Content-Type: Erlaubt nur erwartete Formate
|
|
- Dateiendungen: Blockiert .exe, .bat, .sh Uploads
|
|
|
|
Ein Angreifer, der an Ihrem Code vorbeikommt, wird hier gestoppt.
|
|
Dies ist Ihre erste Verteidigungslinie gegen Injection-Angriffe.
|
|
""",
|
|
},
|
|
"cors": {
|
|
"display_name": "CORS",
|
|
"description": "Cross-Origin Resource Sharing Konfiguration",
|
|
"why_important": """
|
|
CORS bestimmt, welche Websites Ihre API aufrufen duerfen:
|
|
|
|
- Zu offen (*): Jede Website kann Ihre API missbrauchen
|
|
- Zu streng: Ihre eigene Frontend-App wird blockiert
|
|
|
|
BreakPilot erlaubt nur:
|
|
- https://breakpilot.app (Produktion)
|
|
- http://localhost:3000 (Development)
|
|
|
|
Falsch konfiguriertes CORS ist eine der haeufigsten Sicherheitsluecken
|
|
in Web-Anwendungen.
|
|
""",
|
|
},
|
|
}
|
|
|
|
|
|
# ==============================================
|
|
# Test Runner
|
|
# ==============================================
|
|
|
|
|
|
class MiddlewareTestRunner:
|
|
"""Runs middleware tests against the backend."""
|
|
|
|
def __init__(self, base_url: str = "http://localhost:8000"):
|
|
self.base_url = base_url
|
|
|
|
async def test_request_id(self) -> TestCategoryResult:
|
|
"""Test Request-ID middleware."""
|
|
tests: List[TestResult] = []
|
|
start_time = time.time()
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
# Test 1: New request should get a Request-ID
|
|
test_start = time.time()
|
|
try:
|
|
response = await client.get(f"{self.base_url}/api/health")
|
|
request_id = response.headers.get("X-Request-ID", "")
|
|
|
|
tests.append(
|
|
TestResult(
|
|
name="Generiert neue Request-ID",
|
|
description="Eine neue UUID wird generiert wenn keine vorhanden ist",
|
|
expected="UUID im X-Request-ID Header",
|
|
actual=request_id[:36] if len(request_id) >= 36 else request_id,
|
|
status="passed" if len(request_id) >= 36 else "failed",
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
)
|
|
)
|
|
except Exception as e:
|
|
tests.append(
|
|
TestResult(
|
|
name="Generiert neue Request-ID",
|
|
description="Eine neue UUID wird generiert wenn keine vorhanden ist",
|
|
expected="UUID im X-Request-ID Header",
|
|
actual="Fehler",
|
|
status="failed",
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
error_message=str(e),
|
|
)
|
|
)
|
|
|
|
# Test 2: Existing Request-ID should be propagated
|
|
test_start = time.time()
|
|
custom_id = "test-request-id-12345"
|
|
try:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/health",
|
|
headers={"X-Request-ID": custom_id},
|
|
)
|
|
returned_id = response.headers.get("X-Request-ID", "")
|
|
|
|
tests.append(
|
|
TestResult(
|
|
name="Propagiert vorhandene Request-ID",
|
|
description="Existierende X-Request-ID wird weitergeleitet",
|
|
expected=custom_id,
|
|
actual=returned_id,
|
|
status="passed" if returned_id == custom_id else "failed",
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
)
|
|
)
|
|
except Exception as e:
|
|
tests.append(
|
|
TestResult(
|
|
name="Propagiert vorhandene Request-ID",
|
|
description="Existierende X-Request-ID wird weitergeleitet",
|
|
expected=custom_id,
|
|
actual="Fehler",
|
|
status="failed",
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
error_message=str(e),
|
|
)
|
|
)
|
|
|
|
# Test 3: Correlation-ID should work
|
|
test_start = time.time()
|
|
correlation_id = "correlation-12345"
|
|
try:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/health",
|
|
headers={"X-Correlation-ID": correlation_id},
|
|
)
|
|
returned_id = response.headers.get("X-Request-ID", "")
|
|
|
|
tests.append(
|
|
TestResult(
|
|
name="Unterstuetzt X-Correlation-ID",
|
|
description="X-Correlation-ID wird als Request-ID verwendet",
|
|
expected=correlation_id,
|
|
actual=returned_id,
|
|
status="passed" if returned_id == correlation_id else "failed",
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
)
|
|
)
|
|
except Exception as e:
|
|
tests.append(
|
|
TestResult(
|
|
name="Unterstuetzt X-Correlation-ID",
|
|
description="X-Correlation-ID wird als Request-ID verwendet",
|
|
expected=correlation_id,
|
|
actual="Fehler",
|
|
status="failed",
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
error_message=str(e),
|
|
)
|
|
)
|
|
|
|
passed = sum(1 for t in tests if t.status == "passed")
|
|
return TestCategoryResult(
|
|
category="request-id",
|
|
display_name=EDUCATION_CONTENT["request-id"]["display_name"],
|
|
description=EDUCATION_CONTENT["request-id"]["description"],
|
|
why_important=EDUCATION_CONTENT["request-id"]["why_important"],
|
|
tests=tests,
|
|
passed=passed,
|
|
failed=len(tests) - passed,
|
|
total=len(tests),
|
|
duration_ms=(time.time() - start_time) * 1000,
|
|
)
|
|
|
|
async def test_security_headers(self) -> TestCategoryResult:
|
|
"""Test Security Headers middleware."""
|
|
tests: List[TestResult] = []
|
|
start_time = time.time()
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
try:
|
|
response = await client.get(f"{self.base_url}/api/health")
|
|
headers = response.headers
|
|
|
|
# Required headers
|
|
header_tests = [
|
|
("X-Content-Type-Options", "nosniff", "Verhindert MIME-Sniffing"),
|
|
("X-Frame-Options", "DENY", "Verhindert Clickjacking"),
|
|
("X-XSS-Protection", "1; mode=block", "XSS Filter aktiviert"),
|
|
("Referrer-Policy", None, "Kontrolliert Referrer-Informationen"),
|
|
]
|
|
|
|
for header_name, expected_value, description in header_tests:
|
|
test_start = time.time()
|
|
actual = headers.get(header_name, "")
|
|
|
|
if expected_value:
|
|
status = "passed" if actual == expected_value else "failed"
|
|
else:
|
|
status = "passed" if actual else "failed"
|
|
|
|
tests.append(
|
|
TestResult(
|
|
name=f"{header_name} Header",
|
|
description=description,
|
|
expected=expected_value or "Beliebiger Wert",
|
|
actual=actual or "(nicht gesetzt)",
|
|
status=status,
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
)
|
|
)
|
|
|
|
except Exception as e:
|
|
tests.append(
|
|
TestResult(
|
|
name="Security Headers",
|
|
description="Ueberpruefen der Security Headers",
|
|
expected="Headers vorhanden",
|
|
actual="Fehler",
|
|
status="failed",
|
|
error_message=str(e),
|
|
)
|
|
)
|
|
|
|
passed = sum(1 for t in tests if t.status == "passed")
|
|
return TestCategoryResult(
|
|
category="security-headers",
|
|
display_name=EDUCATION_CONTENT["security-headers"]["display_name"],
|
|
description=EDUCATION_CONTENT["security-headers"]["description"],
|
|
why_important=EDUCATION_CONTENT["security-headers"]["why_important"],
|
|
tests=tests,
|
|
passed=passed,
|
|
failed=len(tests) - passed,
|
|
total=len(tests),
|
|
duration_ms=(time.time() - start_time) * 1000,
|
|
)
|
|
|
|
async def test_pii_redactor(self) -> TestCategoryResult:
|
|
"""Test PII Redactor."""
|
|
tests: List[TestResult] = []
|
|
start_time = time.time()
|
|
|
|
try:
|
|
from middleware.pii_redactor import PIIRedactor, redact_pii
|
|
|
|
redactor = PIIRedactor()
|
|
|
|
# Test cases
|
|
test_cases = [
|
|
(
|
|
"Email Redaktion",
|
|
"User test@example.com hat sich angemeldet",
|
|
"test@example.com",
|
|
"[EMAIL_REDACTED]",
|
|
),
|
|
(
|
|
"IPv4 Redaktion",
|
|
"Request von 192.168.1.100",
|
|
"192.168.1.100",
|
|
"[IP_REDACTED]",
|
|
),
|
|
(
|
|
"Telefon Redaktion",
|
|
"Anruf von +49 30 12345678",
|
|
"+49 30 12345678",
|
|
"[PHONE_REDACTED]",
|
|
),
|
|
]
|
|
|
|
for name, text, pii, expected_replacement in test_cases:
|
|
test_start = time.time()
|
|
result = redact_pii(text)
|
|
status = "passed" if pii not in result and expected_replacement in result else "failed"
|
|
|
|
tests.append(
|
|
TestResult(
|
|
name=name,
|
|
description=f"Prueft ob {pii} korrekt entfernt wird",
|
|
expected=expected_replacement,
|
|
actual=result.replace(text.replace(pii, ""), "").strip(),
|
|
status=status,
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
)
|
|
)
|
|
|
|
# Test non-PII preservation
|
|
test_start = time.time()
|
|
clean_text = "Normale Nachricht ohne PII"
|
|
result = redact_pii(clean_text)
|
|
tests.append(
|
|
TestResult(
|
|
name="Nicht-PII bleibt erhalten",
|
|
description="Text ohne PII wird nicht veraendert",
|
|
expected=clean_text,
|
|
actual=result,
|
|
status="passed" if result == clean_text else "failed",
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
)
|
|
)
|
|
|
|
except Exception as e:
|
|
tests.append(
|
|
TestResult(
|
|
name="PII Redactor",
|
|
description="PII Redaction testen",
|
|
expected="Redaktion funktioniert",
|
|
actual="Fehler",
|
|
status="failed",
|
|
error_message=str(e),
|
|
)
|
|
)
|
|
|
|
passed = sum(1 for t in tests if t.status == "passed")
|
|
return TestCategoryResult(
|
|
category="pii-redactor",
|
|
display_name=EDUCATION_CONTENT["pii-redactor"]["display_name"],
|
|
description=EDUCATION_CONTENT["pii-redactor"]["description"],
|
|
why_important=EDUCATION_CONTENT["pii-redactor"]["why_important"],
|
|
tests=tests,
|
|
passed=passed,
|
|
failed=len(tests) - passed,
|
|
total=len(tests),
|
|
duration_ms=(time.time() - start_time) * 1000,
|
|
)
|
|
|
|
async def test_input_gate(self) -> TestCategoryResult:
|
|
"""Test Input Gate middleware."""
|
|
tests: List[TestResult] = []
|
|
start_time = time.time()
|
|
|
|
try:
|
|
from middleware.input_gate import validate_file_upload, InputGateConfig
|
|
|
|
config = InputGateConfig()
|
|
|
|
# Test blocked extensions
|
|
blocked_files = [
|
|
("malware.exe", True),
|
|
("script.bat", True),
|
|
("document.pdf", False),
|
|
("image.jpg", False),
|
|
]
|
|
|
|
for filename, should_block in blocked_files:
|
|
test_start = time.time()
|
|
content_type = "application/pdf" if filename.endswith(".pdf") else "image/jpeg" if filename.endswith(".jpg") else "application/octet-stream"
|
|
valid, _ = validate_file_upload(filename, content_type, 1000, config)
|
|
|
|
if should_block:
|
|
status = "passed" if not valid else "failed"
|
|
expected = "Blockiert"
|
|
actual = "Blockiert" if not valid else "Erlaubt"
|
|
else:
|
|
status = "passed" if valid else "failed"
|
|
expected = "Erlaubt"
|
|
actual = "Erlaubt" if valid else "Blockiert"
|
|
|
|
tests.append(
|
|
TestResult(
|
|
name=f"Datei: {filename}",
|
|
description=f"Prueft ob {filename} korrekt behandelt wird",
|
|
expected=expected,
|
|
actual=actual,
|
|
status=status,
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
)
|
|
)
|
|
|
|
except Exception as e:
|
|
tests.append(
|
|
TestResult(
|
|
name="Input Gate",
|
|
description="Input Validierung testen",
|
|
expected="Validierung funktioniert",
|
|
actual="Fehler",
|
|
status="failed",
|
|
error_message=str(e),
|
|
)
|
|
)
|
|
|
|
passed = sum(1 for t in tests if t.status == "passed")
|
|
return TestCategoryResult(
|
|
category="input-gate",
|
|
display_name=EDUCATION_CONTENT["input-gate"]["display_name"],
|
|
description=EDUCATION_CONTENT["input-gate"]["description"],
|
|
why_important=EDUCATION_CONTENT["input-gate"]["why_important"],
|
|
tests=tests,
|
|
passed=passed,
|
|
failed=len(tests) - passed,
|
|
total=len(tests),
|
|
duration_ms=(time.time() - start_time) * 1000,
|
|
)
|
|
|
|
async def test_rate_limiter(self) -> TestCategoryResult:
|
|
"""Test Rate Limiter middleware."""
|
|
tests: List[TestResult] = []
|
|
start_time = time.time()
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
# Test 1: Rate limit headers present
|
|
test_start = time.time()
|
|
try:
|
|
response = await client.get(f"{self.base_url}/api/health")
|
|
has_headers = (
|
|
"X-RateLimit-Limit" in response.headers
|
|
or "X-RateLimit-Remaining" in response.headers
|
|
)
|
|
|
|
tests.append(
|
|
TestResult(
|
|
name="Rate Limit Headers",
|
|
description="Prueft ob Rate Limit Headers vorhanden sind",
|
|
expected="X-RateLimit-* Headers",
|
|
actual="Vorhanden" if has_headers else "Fehlen",
|
|
status="passed" if has_headers else "skipped",
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
)
|
|
)
|
|
except Exception as e:
|
|
tests.append(
|
|
TestResult(
|
|
name="Rate Limit Headers",
|
|
description="Prueft ob Rate Limit Headers vorhanden sind",
|
|
expected="X-RateLimit-* Headers",
|
|
actual="Fehler",
|
|
status="failed",
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
error_message=str(e),
|
|
)
|
|
)
|
|
|
|
# Test 2: Normal requests pass
|
|
test_start = time.time()
|
|
try:
|
|
response = await client.get(f"{self.base_url}/api/health")
|
|
tests.append(
|
|
TestResult(
|
|
name="Normale Anfragen erlaubt",
|
|
description="Einzelne Anfragen werden durchgelassen",
|
|
expected="Status 200",
|
|
actual=f"Status {response.status_code}",
|
|
status="passed" if response.status_code == 200 else "failed",
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
)
|
|
)
|
|
except Exception as e:
|
|
tests.append(
|
|
TestResult(
|
|
name="Normale Anfragen erlaubt",
|
|
description="Einzelne Anfragen werden durchgelassen",
|
|
expected="Status 200",
|
|
actual="Fehler",
|
|
status="failed",
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
error_message=str(e),
|
|
)
|
|
)
|
|
|
|
passed = sum(1 for t in tests if t.status == "passed")
|
|
return TestCategoryResult(
|
|
category="rate-limiter",
|
|
display_name=EDUCATION_CONTENT["rate-limiter"]["display_name"],
|
|
description=EDUCATION_CONTENT["rate-limiter"]["description"],
|
|
why_important=EDUCATION_CONTENT["rate-limiter"]["why_important"],
|
|
tests=tests,
|
|
passed=passed,
|
|
failed=len(tests) - passed,
|
|
total=len(tests),
|
|
duration_ms=(time.time() - start_time) * 1000,
|
|
)
|
|
|
|
async def test_cors(self) -> TestCategoryResult:
|
|
"""Test CORS middleware."""
|
|
tests: List[TestResult] = []
|
|
start_time = time.time()
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
# Test preflight request
|
|
test_start = time.time()
|
|
try:
|
|
response = await client.options(
|
|
f"{self.base_url}/api/health",
|
|
headers={
|
|
"Origin": "http://localhost:3000",
|
|
"Access-Control-Request-Method": "GET",
|
|
},
|
|
)
|
|
cors_header = response.headers.get("Access-Control-Allow-Origin", "")
|
|
|
|
tests.append(
|
|
TestResult(
|
|
name="CORS Preflight",
|
|
description="OPTIONS Anfrage fuer CORS",
|
|
expected="Access-Control-Allow-Origin Header",
|
|
actual=cors_header or "(nicht gesetzt)",
|
|
status="passed" if cors_header else "skipped",
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
)
|
|
)
|
|
except Exception as e:
|
|
tests.append(
|
|
TestResult(
|
|
name="CORS Preflight",
|
|
description="OPTIONS Anfrage fuer CORS",
|
|
expected="Access-Control-Allow-Origin Header",
|
|
actual="Fehler",
|
|
status="failed",
|
|
duration_ms=(time.time() - test_start) * 1000,
|
|
error_message=str(e),
|
|
)
|
|
)
|
|
|
|
passed = sum(1 for t in tests if t.status == "passed")
|
|
return TestCategoryResult(
|
|
category="cors",
|
|
display_name=EDUCATION_CONTENT["cors"]["display_name"],
|
|
description=EDUCATION_CONTENT["cors"]["description"],
|
|
why_important=EDUCATION_CONTENT["cors"]["why_important"],
|
|
tests=tests,
|
|
passed=passed,
|
|
failed=len(tests) - passed,
|
|
total=len(tests),
|
|
duration_ms=(time.time() - start_time) * 1000,
|
|
)
|
|
|
|
async def run_all(self) -> FullTestResults:
|
|
"""Run all middleware tests."""
|
|
start_time = time.time()
|
|
|
|
# Run all test categories
|
|
categories = await asyncio.gather(
|
|
self.test_request_id(),
|
|
self.test_security_headers(),
|
|
self.test_rate_limiter(),
|
|
self.test_pii_redactor(),
|
|
self.test_input_gate(),
|
|
self.test_cors(),
|
|
)
|
|
|
|
total_passed = sum(c.passed for c in categories)
|
|
total_failed = sum(c.failed for c in categories)
|
|
total_tests = sum(c.total for c in categories)
|
|
|
|
return FullTestResults(
|
|
timestamp=datetime.now().isoformat(),
|
|
categories=list(categories),
|
|
total_passed=total_passed,
|
|
total_failed=total_failed,
|
|
total_tests=total_tests,
|
|
duration_ms=(time.time() - start_time) * 1000,
|
|
)
|
|
|
|
|
|
# ==============================================
|
|
# API Router
|
|
# ==============================================
|
|
|
|
router = APIRouter(prefix="/api/admin/ui-tests", tags=["ui-tests"])
|
|
|
|
# Global test runner instance
|
|
test_runner = MiddlewareTestRunner()
|
|
|
|
|
|
@router.post("/request-id", response_model=TestCategoryResult)
|
|
async def test_request_id():
|
|
"""Run Request-ID middleware tests."""
|
|
return await test_runner.test_request_id()
|
|
|
|
|
|
@router.post("/security-headers", response_model=TestCategoryResult)
|
|
async def test_security_headers():
|
|
"""Run Security Headers middleware tests."""
|
|
return await test_runner.test_security_headers()
|
|
|
|
|
|
@router.post("/rate-limiter", response_model=TestCategoryResult)
|
|
async def test_rate_limiter():
|
|
"""Run Rate Limiter middleware tests."""
|
|
return await test_runner.test_rate_limiter()
|
|
|
|
|
|
@router.post("/pii-redactor", response_model=TestCategoryResult)
|
|
async def test_pii_redactor():
|
|
"""Run PII Redactor tests."""
|
|
return await test_runner.test_pii_redactor()
|
|
|
|
|
|
@router.post("/input-gate", response_model=TestCategoryResult)
|
|
async def test_input_gate():
|
|
"""Run Input Gate middleware tests."""
|
|
return await test_runner.test_input_gate()
|
|
|
|
|
|
@router.post("/cors", response_model=TestCategoryResult)
|
|
async def test_cors():
|
|
"""Run CORS middleware tests."""
|
|
return await test_runner.test_cors()
|
|
|
|
|
|
@router.post("/run-all", response_model=FullTestResults)
|
|
async def run_all_tests():
|
|
"""Run all middleware tests."""
|
|
return await test_runner.run_all()
|
|
|
|
|
|
@router.get("/education/{category}")
|
|
async def get_education_content(category: str):
|
|
"""Get educational content for a test category."""
|
|
if category not in EDUCATION_CONTENT:
|
|
raise HTTPException(status_code=404, detail=f"Category '{category}' not found")
|
|
|
|
return EDUCATION_CONTENT[category]
|
|
|
|
|
|
@router.get("/categories")
|
|
async def list_categories():
|
|
"""List all available test categories."""
|
|
return [
|
|
{
|
|
"id": key,
|
|
"display_name": value["display_name"],
|
|
"description": value["description"],
|
|
}
|
|
for key, value in EDUCATION_CONTENT.items()
|
|
]
|