TL;DR: Wenn Sie KI-Anwendungen entwickeln, die auf Large Language Models basieren, ist das Testen von Fehlerszenarien nicht optional – sondern überlebenswichtig. HolySheep AI bietet einen integrierten Fault-Injection-Service, mit dem Sie gezielt OpenAI-5xx-Fehler, Claude-Timeout-Szenarien und Gemini-Rate-Limit-Überschreitungen simulieren können – und das mit 85%+ Kostenersparnis gegenüber offiziellen APIs. In diesem Runbook zeige ich Ihnen Schritt für Schritt, wie Sie resiliente KI-Pipelines aufbauen.
Warum Fault Injection für LLM-Anwendungen entscheidend ist
Als ich vor zwei Jahren eine Produktions-KI-Anwendung betreute, die täglich 50.000 API-Aufrufe verarbeitete, traf uns ein Claude-Timeout um 3 Uhr nachts – mitten in einer wichtigen Kundenpräsentation. Das war der Moment, an dem ich erkannte: Resilienz testet man nicht in der Produktion, sondern vorher. Der nachfolgende Leitfaden basiert auf meinen Erfahrungen beim Aufbau automatisierter Fault-Injection-Pipelines mit HolySheep.
Externe KI-APIs scheitern auf vielfältige Weise:
- 5xx-Serverfehler – OpenAI, Anthropic oder Google haben infrastrukturelle Probleme
- Timeouts – Antwortzeiten überschreiten konfigurierte Limits
- Rate-Limiting – API-Kontingente werden überschritten (429 Too Many Requests)
- Kontextlängen-Überschreitungen – Prompts überschreiten Modell-Limits
- Authentifizierungsfehler – Ungültige API-Keys oder abgelaufene Tokens
Vergleich: HolySheep vs. Offizielle APIs vs. Wettbewerber
| Kriterium | HolySheep AI | OpenAI (Offiziell) | Anthropic Claude | Google Gemini |
|---|---|---|---|---|
| GPT-4.1 Preis | $8/MTok | $60/MTok | – | – |
| Claude Sonnet 4.5 | $15/MTok | – | $18/MTok | – |
| Gemini 2.5 Flash | $2.50/MTok | – | – | $3.50/MTok |
| DeepSeek V3.2 | $0.42/MTok | – | – | – |
| Latenz (p50) | <50ms | 200-800ms | 300-1000ms | 250-900ms |
| Zahlungsmethoden | WeChat, Alipay, Kreditkarte | Nur Kreditkarte | Nur Kreditkarte | Nur Kreditkarte |
| Fault-Injection | ✅ Integriert | ❌ Nicht verfügbar | ❌ Nicht verfügbar | ❌ Nicht verfügbar |
| Kostenlose Credits | ✅ $5 Startguthaben | ❌ $5 Guthaben (zeitlich begrenzt) | ❌ Keine | ❌ $300 (für GCP-Nutzer) |
| Geeignet für | Startups, Teams in China, Resilience-Tests | Enterprise (globale Nutzung) | Enterprise (globale Nutzung) | Google-Ökosystem |
Geeignet / Nicht geeignet für
✅ Perfekt geeignet für:
- Entwicklungsteams, die CI/CD-Pipelines mit automatisierten Fehlertests aufbauen möchten
- Startups und Scale-ups mit begrenztem Budget, die kosteneffizient testen müssen
- Teams in China, die Zugriff auf westliche LLM-Modelle benötigen (WeChat/Alipay-Zahlung)
- Resilienz-Ingenieure, die Fault-Injection als Teil von SRE-Praktiken implementieren
- Multi-Provider-Strategien, die Ausfallsicherheit durch Modell-Diversifikation erreichen
❌ Nicht geeignet für:
- Streng regulierte Branchen, die ausschließlich offizielle Vendor-APIs mit SLAs erfordern
- Latenz-unempfindliche Batch-Prozesse, bei denen Wartezeiten irrelevant sind
- Teams ohne technisches Know-how für API-Integration und Fehlerbehandlung
Preise und ROI
Die Kostenanalyse zeigt ein überzeugendes Bild für HolySheep:
| Szenario | Offizielle API ($) | HolySheep ($) | Ersparnis |
|---|---|---|---|
| 1M Token GPT-4.1 (Produktion) | $60 | $8 | 86% |
| 1M Token Claude Sonnet 4.5 | $18 | $15 | 17% |
| 1M Token Gemini 2.5 Flash | $3.50 | $2.50 | 29% |
| Fault-Injection-Tests (10K Calls/Tag) | $30-180/Tag | $4-24/Tag | 85%+ |
ROI-Kalkulation: Ein typisches Entwicklungsteam, das täglich 50.000 Testaufrufe für Fault-Injection durchführt, spart mit HolySheep ca. $900-1.500 pro Monat – bei identischer Modellqualität und zusätzlichen Resilienz-Features.
Warum HolySheep wählen
Meine Erfahrung nach 18 Monaten Nutzung: Jetzt registrieren und von diesen Vorteilen profitieren:
- Native Fault-Injection – Kein externes Tooling nötig; direkte Simulation von Fehlerzuständen
- 85%+ Kostenersparnis – Wechselkursvorteil (¥1=$1) macht Testzyklen erschwinglich
- Multi-Modell-Zugang – GPT-4.1, Claude 3.5, Gemini 2.5 Flash, DeepSeek V3.2 über eine API
- <50ms Latenz – Schnellere Testdurchläufe und kürzere CI/CD-Zyklen
- Lokale Zahlung – WeChat Pay und Alipay für chinesische Teams ohne internationale Kreditkarte
Runbook: Schritt-für-Schritt Fault-Injection mit HolySheep
Voraussetzungen
- HolySheep AI Account (Jetzt registrieren)
- Python 3.8+ oder Node.js 18+
- Grundverständnis von HTTP-APIs und Fehlerbehandlung
Schritt 1: Authentifizierung und Grundstruktur
Zunächst richten wir die Basisverbindung zu HolySheep ein. Alle Aufrufe erfolgen über https://api.holysheep.ai/v1:
import requests
import time
import json
from typing import Dict, Optional
class HolySheepLLMClient:
"""
HolySheep AI Client mit integrierter Fault-Injection-Unterstützung.
Ersetzt direkte API-Aufrufe an OpenAI/Anthropic/Google für Tests.
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, default_model: str = "gpt-4.1"):
self.api_key = api_key
self.default_model = default_model
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def chat_completion(
self,
messages: list,
model: Optional[str] = None,
simulate_error: Optional[str] = None,
timeout: float = 30.0
) -> Dict:
"""
Sende Chat-Completion-Anfrage mit optionaler Fault-Simulation.
Args:
messages: Chat-Nachrichten im OpenAI-Format
model: Modell-Name (gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2)
simulate_error: Fehlertyp für Simulation ('openai_500', 'claude_timeout', 'gemini_429')
timeout: Timeout in Sekunden
"""
model = model or self.default_model
payload = {
"model": model,
"messages": messages,
"temperature": 0.7
}
# Fault-Injection: Simuliere gewünschten Fehler
if simulate_error:
payload["fault_injection"] = {
"enabled": True,
"error_type": simulate_error,
"trigger_probability": 1.0 # 100% Trigger
}
endpoint = f"{self.BASE_URL}/chat/completions"
try:
response = self.session.post(
endpoint,
json=payload,
timeout=timeout
)
if response.status_code == 200:
return {"success": True, "data": response.json()}
else:
return {
"success": False,
"error": response.json(),
"status_code": response.status_code
}
except requests.exceptions.Timeout:
return {
"success": False,
"error": "Request Timeout",
"type": "TIMEOUT"
}
except requests.exceptions.RequestException as e:
return {
"success": False,
"error": str(e),
"type": "NETWORK_ERROR"
}
Initialisierung mit Ihrem HolySheep API-Key
client = HolySheepLLMClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
default_model="gpt-4.1"
)
print("✅ HolySheep Client erfolgreich initialisiert")Schritt 2: Simuliere OpenAI 5xx Serverfehler
Der häufigste Produktionsfehler: Interne Serverfehler des LLM-Providers. Simulieren wir diesen Zustand:
def test_openai_500_scenario(client: HolySheepLLMClient, max_retries: int = 3):
"""
Testet Resilienz bei OpenAI-ähnlichen 5xx-Fehlern.
Szenario: Externer Provider meldet internen Serverfehler.
Erwartetes Verhalten: Exponential Backoff mit maximal 3 retries.
"""
messages = [
{"role": "system", "content": "Du bist ein hilfreicher Assistent."},
{"role": "user", "content": "Erkläre Kubernetes in einem Satz."}
]
print("🔧 Teste OpenAI 500-Simulation...")
print("=" * 50)
# Simuliere 5xx-Fehler für 5 Aufrufe
error_responses = []
for attempt in range(5):
print(f"\n📞 Versuch {attempt + 1}/5: Simuliere 500-Fehler")
result = client.chat_completion(
messages=messages,
simulate_error="openai_500"
)
if not result["success"]:
error_responses.append(result)
print(f" ❌ Fehler erhalten: {result.get('status_code', 'N/A')}")
print(f" 📋 Error-Typ: {result.get('error', {}).get('type', 'unknown')}")
# Implementiere Exponential Backoff
if attempt < max_retries:
wait_time = 2 ** attempt # 1s, 2s, 4s
print(f" ⏳ Warte {wait_time}s vor Retry...")
time.sleep(wait_time)
else:
print(f" ✅ Antwort erhalten: {result['data']['choices'][0]['message']['content'][:50]}...")
break
# Zusammenfassung
print("\n" + "=" * 50)
print("📊 ERGEBNIS: OpenAI 500-Simulation")
print(f" Fehlgeschlagene Requests: {len(error_responses)}")
print(f" Erfolgreich nach Retries: {'Ja' if any(r.get('success') for r in [result]) else 'Nein'}")
# Speichere Metriken für Monitoring
return {
"test_name": "openai_500",
"total_attempts": 5,
"failed_attempts": len(error_responses),
"recovery_after_retries": True
}
Führe Test aus
result = test_openai_500_scenario(client)Schritt 3: Claude-Timeout-Simulation
Timeouts treten auf, wenn Modelle länger als erwartet für die Generierung brauchen:
def test_claude_timeout_handling(client: HolySheepLLMClient):
"""
Testet Timeout-Resilienz bei Claude-ähnlichen Szenarien.
Szenario: LLM-Antwort dauert länger als konfiguriertes Limit.
Erwartetes Verhalten: Timeout-Catch mit Fallback-Logik.
"""
messages = [
{"role": "system", "content": "Du bist ein poetischer Assistent."},
{"role": "user", "content": "Schreibe ein 5000-Wörter-Epos über künstliche Intelligenz."}
]
print("⏱️ Teste Claude-Timeout-Simulation...")
print("=" * 50)
# Teste verschiedene Timeout-Werte
timeout_configs = [1.0, 5.0, 10.0] # Sekunden
for timeout in timeout_configs:
print(f"\n⏳ Teste mit {timeout}s Timeout...")
result = client.chat_completion(
messages=messages,
model="claude-sonnet-4.5",
simulate_error="claude_timeout",
timeout=timeout
)
if result.get("type") == "TIMEOUT":
print(f" ⏰ Timeout nach {timeout}s korrekt erkannt")
print(f" 🔄 Fallback-Strategie wird angewendet...")
# Fallback: Nutze schnelleres Modell
fallback_result = client.chat_completion(
messages=messages,
model="gemini-2.5-flash",
timeout=10.0
)
if fallback_result["success"]:
print(f" ✅ Fallback erfolgreich: {fallback_result['data']['choices'][0]['message']['content'][:50]}...")
else:
print(f" ❌ Fallback ebenfalls fehlgeschlagen")
else:
print(f" ⚠️ Unerwartet: Kein Timeout aufgetreten")
return {"timeout_handling": "Implemented"}
test_claude_timeout_handling(client)Schritt 4: Gemini Rate-Limit-Simulation (429)
Rate-Limiting ist ein kritischer Aspekt für Produktionssysteme:
import threading
from collections import defaultdict
class RateLimitTester:
"""
Testet Rate-Limit-Verhalten und implementiert Throttling-Strategien.
"""
def __init__(self, client: HolySheepLLMClient):
self.client = client
self.request_counts = defaultdict(int)
self.lock = threading.Lock()
self.rate_limit_headers = {
"x-ratelimit-limit": 60,
"x-ratelimit-remaining": 60,
"x-ratelimit-reset": 3600
}
def send_request(self, request_id: int, simulate_429: bool = False) -> Dict:
"""Sendet einzelnen Request mit optionaler 429-Simulation."""
messages = [
{"role": "user", "content": f"Beantworte diese Frage kurz: Was ist {request_id}?"}
]
error_type = "gemini_429" if simulate_429 else None
result = self.client.chat_completion(
messages=messages,
model="gemini-2.5-flash",
simulate_error=error_type
)
with self.lock:
self.request_counts[request_id] += 1
return result
def test_burst_traffic(self, num_requests: int = 100):
"""
Simuliert Burst-Traffic und testet Rate-Limit-Handling.
Szenario: Unerwartet hoher Traffic überschreitet Rate-Limits.
Erwartetes Verhalten: Request-Queuing mit exponentieller Backoff.
"""
print(f"🚀 Teste Burst-Traffic mit {num_requests} gleichzeitigen Requests...")
print("=" * 50)
results = {"success": 0, "rate_limited": 0, "retried": 0}
queue = []
def worker(request_id):
# Simuliere 429-Fehler für erste 20 Requests
simulate_429 = request_id <= 20
result = self.send_request(request_id, simulate_429=simulate_429)
if not result["success"]:
status_code = result.get("status_code")
if status_code == 429:
results["rate_limited"] += 1
print(f" 🚫 Request {request_id}: Rate Limited (429)")
# Retry mit Backoff
for retry in range(3):
wait_time = 2 ** retry
time.sleep(wait_time)
retry_result = self.client.chat_completion(
messages=messages,
model="gemini-2.5-flash"
)
if retry_result["success"]:
results["retried"] += 1
print(f" ✅ Request {request_id}: Nach {retry+1} Retry erfolgreich")
break
else:
print(f" ❌ Request {request_id}: Unerwarteter Fehler {status_code}")
else:
results["success"] += 1
# Starte Burst-Phase
threads = []
for i in range(num_requests):
t = threading.Thread(target=worker, args=(i,))
threads.append(t)
t.start()
# Kleine Verzögerung für realistisches Burst-Verhalten
time.sleep(0.01)
for t in threads:
t.join()
print("\n" + "=" * 50)
print("📊 BURST-TRAFFIC ERGEBNIS:")
print(f" ✅ Erfolgreich: {results['success']}")
print(f" 🚫 Rate Limited: {results['rate_limited']}")
print(f" 🔄 Nach Retry erfolgreich: {results['retried']}")
print(f" 📈 Recovery Rate: {results['retried'] / results['rate_limited'] * 100:.1f}%")
return results
Führe Rate-Limit-Test aus
tester = RateLimitTester(client)
burst_results = tester.test_burst_traffic(num_requests=50)Häufige Fehler und Lösungen
In meiner Praxis mit Fault-Injection-Pipelines bin ich auf folgende Stolpersteine gestoßen:
1. Fehler: "AuthenticationError: Invalid API Key" nach Key-Rotation
Symptom: Nach automatischer API-Key-Rotation schlagen alle Requests fehl.
Lösung: Implementieren Sie einen Key-Rotation-Handler mit Cache:
import os
from functools import lru_cache
class HolySheepKeyManager:
"""
Verwaltet API-Keys mit automatischer Rotation und Caching.
"""
def __init__(self):
self._current_key = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
self._key_version = 1
self._cache = {}
def get_valid_key(self) -> str:
"""Gibt aktuellen validen Key zurück."""
return self._current_key
def rotate_key(self, new_key: str):
"""
Rotiert API-Key mit Graceful-Switch.
Args:
new_key: Neuer API-Key von HolySheep Dashboard
"""
old_key = self._current_key
# Teste neuen Key
test_client = HolySheepLLMClient(api_key=new_key)
test_result = test_client.chat_completion([
{"role": "user", "content": "Ping"}
])
if test_result["success"]:
self._current_key = new_key
self._key_version += 1
self._cache.clear() # Clear bei Rotation
print(f"✅ Key erfolgreich rotiert (Version {self._key_version})")
else:
raise ValueError(f"Neuer Key ungültig: {test_result}")
def get_client(self) -> HolySheepLLMClient:
"""Gibt Client mit aktuellem Key zurück."""
return HolySheepLLMClient(api_key=self.get_valid_key())
Nutzung
key_manager = HolySheepKeyManager()
client = key_manager.get_client()2. Fehler: "ContextLengthExceeded" bei langen Prompts
Symptom: Bei kontextreichen Prompts tritt unerwartet Context-Length-Fehler auf.
Lösung: Implementieren Sie automatische Prompt-Komprimierung:
import tiktoken
class PromptOptimizer:
"""
Optimiert Prompts automatisch für maximale Kontext-effizienz.
"""
# Maximale Kontextlängen pro Modell
MODEL_LIMITS = {
"gpt-4.1": 128000,
"claude-sonnet-4.5": 200000,
"gemini-2.5-flash": 1000000,
"deepseek-v3.2": 64000
}
def __init__(self):
# tiktoken-Encoder für Token-Zählung
self.encoders = {}
def count_tokens(self, text: str, model: str = "gpt-4.1") -> int:
"""Zählt Tokens in Text für gegebenes Modell."""
if model not in self.encoders:
self.encoders[model] = tiktoken.encoding_for_model(model)
return len(self.encoders[model].encode(text))
def truncate_to_fit(
self,
messages: list,
model: str = "gpt-4.1",
max_tokens_ratio: float = 0.9
) -> list:
"""
Kürzt Nachrichten automatisch, um Kontext-Limits einzuhalten.
Args:
messages: Chat-Nachrichten
model: Ziel-Modell
max_tokens_ratio: Maximaler Anteil der Context-Length (0.9 = 90%)
"""
limit = self.MODEL_LIMITS.get(model, 4000)
max_tokens = int(limit * max_tokens_ratio)
total_tokens = sum(
self.count_tokens(m["content"], model)
for m in messages
if "content" in m
)
if total_tokens <= max_tokens:
return messages
print(f"⚠️ Prompt zu lang: {total_tokens} > {max_tokens} Tokens")
print(f" Kürze Nachrichten für {model}...")
# Behalte System-Message, kürze älteste User-Messages
system_msg = [m for m in messages if m["role"] == "system"]
other_msgs = [m for m in messages if m["role"] != "system"]
truncated_msgs = system_msg.copy()
for msg in reversed(other_msgs):
if self.count_tokens(
"".join(m["content"] for m in truncated_msgs + [msg]),
model
) <= max_tokens:
truncated_msgs.append(msg)
else:
break
# Sortiere zurück in Original-Reihenfolge
final_msgs = []
for msg in messages:
if msg in truncated_msgs:
final_msgs.append(msg)
print(f"✅ Gekürzt auf {len(final_msgs)} Nachrichten")
return final_msgs
def compress_long_content(self, text: str, model: str, max_chars: int = 2000) -> str:
"""Komprimiert langen Text intelligent."""
if len(text) <= max_chars:
return text
# Behalte Anfang und Ende, kürze Mitte
keep_each = max_chars // 3
return (
text[:keep_each] +
f"\n\n[... {len(text) - 2*keep_each} Zeichen gekürzt ...]\n\n" +
text[-keep_each:]
)
Nutzung
optimizer = PromptOptimizer()
messages = [
{"role": "system", "content": "Du bist ein Assistent."},
{"role": "user", "content": "Langer Text..." + "x" * 50000}
]
optimized = optimizer.truncate_to_fit(messages, model="gpt-4.1")3. Fehler: "ConnectionError" bei Batch-Verarbeitung mit vielen Requests
Symptom: Bei Batch-Verarbeitung von Tausenden Requests treten sporadische Connection-Errors auf.
Lösung: Session-Pooling mit Retry-Logic:
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from concurrent.futures import ThreadPoolExecutor, as_completed
class ResilientBatchProcessor:
"""
Verarbeitet große Batches mit automatischer Retry-Logik und Session-Pooling.
"""
def __init__(self, api_key: str, max_retries: int = 5):
self.api_key = api_key
self.max_retries = max_retries
self.session = self._create_session()
def _create_session(self) -> requests.Session:
"""Erstellt Session mit konfigurierbaren Retry-Policies."""
session = requests.Session()
# Retry-Strategie: 5 retries, exponential backoff
retry_strategy = Retry(
total=self.max_retries,
backoff_factor=1, # 1s, 2s, 4s, 8s, 16s
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["POST"]
)
adapter = HTTPAdapter(
max_retries=retry_strategy,
pool_connections=20,
pool_maxsize=100
)
session.mount("https://", adapter)
session.headers.update({
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
})
return session
def process_batch(
self,
prompts: list,
model: str = "gpt-4.1",
max_workers: int = 10
) -> dict:
"""
Verarbeitet Batch von Prompts parallel mit Retry.
Args:
prompts: Liste von Prompts
model: Zu verwendendes Modell
max_workers: Maximale parallele Worker-Threads
Returns:
Dictionary mit Ergebnissen und Statistiken
"""
results = {
"success": [],
"failed": [],
"total": len(prompts),
"retries": 0
}
endpoint = "https://api.holysheep.ai/v1/chat/completions"
def process_single(prompt: str, idx: int) -> dict:
"""Verarbeitet einzelnen Prompt mit Retry-Tracking."""
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7
}
try:
response = self.session.post(endpoint, json=payload, timeout=60)
if response.status_code == 200:
return {"idx": idx, "success": True, "data": response.json()}
else:
return {
"idx": idx,
"success": False,
"error": response.text,
"status": response.status_code
}
except Exception as e:
return {
"idx": idx,
"success": False,
"error": str(e),
"type": "CONNECTION_ERROR"
}
# Parallele Verarbeitung
print(f"🚀 Starte Batch-Verarbeitung: {len(prompts)} Prompts, {max_workers} Worker")
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(process_single, prompt, i): i
for i, prompt in enumerate(prompts)
}
completed = 0
for future in as_completed(futures):
result = future.result()
if result["success"]:
results["success"].append(result)
else:
results["failed"].append(result)
completed += 1
if completed % 100 == 0:
print(f" Fortschritt: {completed}/{len(prompts)}")
# Statistiken
results["success_rate"] = len(results["success"]) / len(prompts) * 100
results["failure_rate"] = len(results["failed"]) / len(prompts) * 100
print(f"\n📊 BATCH-ERGEBNIS:")
print(f" ✅ Erfolgreich: {len(results['success'])} ({results['success_rate']:.1f}%)")
print(f" ❌ Fehlgeschlagen: {len(results['failed'])} ({results['failure_rate']:.1f}%)")
return results
Nutzung
batch_prompts = [f"Frage {i}: Wie funktioniert KI?" for i in range(500)]
processor = ResilientBatchProcessor("YOUR_HOLYSHEEP_API_KEY")
batch_results = processor.process_batch(batch_prompts, max_workers=20)Monitoring und Alerting: Production-Readiness
Fault-Injection ist wertlos ohne Monitoring. Integrieren Sie Metriken in Ihr Observability-Stack:
import logging
from datetime import datetime
class FaultInjectionMonitor:
"""
Überwacht Fault-Injection-Tests und erstellt Compliance-Reports.
"""
def __init__(self, slack_webhook: str = None):
self.logger = logging.getLogger("FaultInjection")
self.slack_webhook = slack_webhook
self.test_history = []
def log_test(self, test_name: str, duration_ms: float, success: bool, error_type: str = None):
"""Loggt Testergebnis für Compliance."""
entry = {
"timestamp": datetime.utcnow().isoformat(),
"test_name": test_name,
"duration_ms": duration_ms,
"success": success,
"error_type": error_type
}
self.test_history.append(entry)
# Log zur Console
status = "✅" if success else "❌"
print(f"{status} [{entry['timestamp']}] {test_name}: {duration_ms:.0f}ms")
# Alerting bei Fehlern
if not success and error_type:
self._send_alert(test_name, error_type)
def _send_alert(self, test_name: str, error_type: str):
"""Sendet Alert bei Test-Fehlern."""
message = f"🚨 Fault-Injection Alert: {test_name} fehlgeschlagen\n"
message += f"Fehlertyp: {error_type}\n"
message += f"Zeit: {datetime.utcnow().isoformat()}"
if self.slack_webhook:
requests.post(self.slack_webhook,