275 lines
9.9 KiB
Python
275 lines
9.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Test-Script für den DatabaseCleanupManager
|
||
Validiert die robuste Datenbank-Cleanup-Funktionalität
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
import sqlite3
|
||
import threading
|
||
from datetime import datetime
|
||
|
||
# Pfad zur App hinzufügen
|
||
app_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
sys.path.insert(0, app_dir)
|
||
|
||
from utils.database_cleanup import DatabaseCleanupManager, safe_database_cleanup
|
||
from config.settings import DATABASE_PATH
|
||
from utils.logging_config import get_logger
|
||
|
||
logger = get_logger("database_cleanup_test")
|
||
|
||
def test_basic_cleanup():
|
||
"""Test der grundlegenden Cleanup-Funktionalität"""
|
||
print("🧪 Test 1: Grundlegende Cleanup-Funktionalität")
|
||
|
||
try:
|
||
# Erstelle Test-DatabaseCleanupManager
|
||
cleanup_manager = DatabaseCleanupManager()
|
||
|
||
# Teste WAL-Checkpoint
|
||
checkpoint_success, checkpoint_error = cleanup_manager.safe_wal_checkpoint(retry_attempts=3)
|
||
|
||
if checkpoint_success:
|
||
print("✅ WAL-Checkpoint erfolgreich")
|
||
else:
|
||
print(f"❌ WAL-Checkpoint fehlgeschlagen: {checkpoint_error}")
|
||
|
||
# Teste umfassendes Cleanup
|
||
cleanup_result = cleanup_manager.comprehensive_cleanup(force_mode_switch=False) # Kein Mode-Switch für Test
|
||
|
||
if cleanup_result["success"]:
|
||
print(f"✅ Umfassendes Cleanup erfolgreich: {', '.join(cleanup_result['operations'])}")
|
||
else:
|
||
print(f"❌ Umfassendes Cleanup fehlgeschlagen: {', '.join(cleanup_result['errors'])}")
|
||
|
||
return cleanup_result["success"]
|
||
|
||
except Exception as e:
|
||
print(f"❌ Test 1 fehlgeschlagen: {e}")
|
||
return False
|
||
|
||
def test_concurrent_access():
|
||
"""Test des Cleanup-Verhaltens bei gleichzeitigen Datenbankzugriffen"""
|
||
print("\n🧪 Test 2: Cleanup bei gleichzeitigen Datenbankzugriffen")
|
||
|
||
try:
|
||
# Worker-Thread der Datenbankoperationen ausführt
|
||
def database_worker():
|
||
try:
|
||
for i in range(5):
|
||
conn = sqlite3.connect(DATABASE_PATH, timeout=2)
|
||
conn.execute("SELECT COUNT(*) FROM users")
|
||
time.sleep(0.5)
|
||
conn.close()
|
||
print(f" Worker: Datenbankoperation {i+1} abgeschlossen")
|
||
except Exception as e:
|
||
print(f" Worker-Fehler: {e}")
|
||
|
||
# Starte Worker-Thread
|
||
worker_thread = threading.Thread(target=database_worker, daemon=True)
|
||
worker_thread.start()
|
||
|
||
# Kurz warten damit Worker startet
|
||
time.sleep(1)
|
||
|
||
# Teste Cleanup während Worker läuft
|
||
cleanup_manager = DatabaseCleanupManager()
|
||
cleanup_result = cleanup_manager.comprehensive_cleanup(force_mode_switch=False)
|
||
|
||
if cleanup_result["success"]:
|
||
print("✅ Cleanup erfolgreich trotz gleichzeitiger Datenbankzugriffe")
|
||
else:
|
||
print(f"❌ Cleanup fehlgeschlagen: {', '.join(cleanup_result['errors'])}")
|
||
|
||
# Warte auf Worker
|
||
worker_thread.join(timeout=10)
|
||
|
||
return cleanup_result["success"]
|
||
|
||
except Exception as e:
|
||
print(f"❌ Test 2 fehlgeschlagen: {e}")
|
||
return False
|
||
|
||
def test_error_recovery():
|
||
"""Test der Fehlerbehandlung und Recovery-Mechanismen"""
|
||
print("\n🧪 Test 3: Fehlerbehandlung und Recovery")
|
||
|
||
try:
|
||
cleanup_manager = DatabaseCleanupManager()
|
||
|
||
# Teste mit verschiedenen Retry-Parametern
|
||
for retry_attempts in [1, 3, 5]:
|
||
print(f" Teste mit {retry_attempts} Retry-Versuchen...")
|
||
|
||
checkpoint_success, checkpoint_error = cleanup_manager.safe_wal_checkpoint(retry_attempts=retry_attempts)
|
||
|
||
if checkpoint_success:
|
||
print(f" ✅ WAL-Checkpoint mit {retry_attempts} Versuchen erfolgreich")
|
||
else:
|
||
print(f" ⚠️ WAL-Checkpoint mit {retry_attempts} Versuchen: {checkpoint_error}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ Test 3 fehlgeschlagen: {e}")
|
||
return False
|
||
|
||
def test_journal_mode_operations():
|
||
"""Test der Journal-Mode-Operationen"""
|
||
print("\n🧪 Test 4: Journal-Mode-Operationen")
|
||
|
||
try:
|
||
cleanup_manager = DatabaseCleanupManager()
|
||
|
||
# Teste aktuellen Journal-Mode
|
||
conn = sqlite3.connect(DATABASE_PATH, timeout=5)
|
||
current_mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
|
||
print(f" Aktueller Journal-Mode: {current_mode}")
|
||
conn.close()
|
||
|
||
# Teste Journal-Mode-Switch (nur wenn bereits WAL-Mode)
|
||
if current_mode.upper() == "WAL":
|
||
print(" Teste Journal-Mode-Switch...")
|
||
|
||
# Teste Switch zu WAL (sollte bereits WAL sein)
|
||
mode_success, mode_error = cleanup_manager.safe_journal_mode_switch("WAL", retry_attempts=2)
|
||
|
||
if mode_success:
|
||
print(" ✅ Journal-Mode-Switch zu WAL erfolgreich")
|
||
else:
|
||
print(f" ❌ Journal-Mode-Switch fehlgeschlagen: {mode_error}")
|
||
|
||
return mode_success
|
||
else:
|
||
print(f" ℹ️ Database bereits im {current_mode}-Mode, kein Switch-Test nötig")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ Test 4 fehlgeschlagen: {e}")
|
||
return False
|
||
|
||
def test_convenience_function():
|
||
"""Test der Convenience-Funktion safe_database_cleanup"""
|
||
print("\n🧪 Test 5: Convenience-Funktion safe_database_cleanup")
|
||
|
||
try:
|
||
# Teste die einfache Convenience-Funktion
|
||
cleanup_result = safe_database_cleanup(force_mode_switch=False)
|
||
|
||
if cleanup_result["success"]:
|
||
print(f"✅ safe_database_cleanup erfolgreich: {', '.join(cleanup_result['operations'])}")
|
||
|
||
# Prüfe Cleanup-Details
|
||
if "timestamp" in cleanup_result:
|
||
print(f" Zeitstempel: {cleanup_result['timestamp']}")
|
||
|
||
if "wal_files_removed" in cleanup_result:
|
||
print(f" WAL-Dateien entfernt: {cleanup_result['wal_files_removed']}")
|
||
|
||
else:
|
||
print(f"❌ safe_database_cleanup fehlgeschlagen: {', '.join(cleanup_result['errors'])}")
|
||
|
||
return cleanup_result["success"]
|
||
|
||
except Exception as e:
|
||
print(f"❌ Test 5 fehlgeschlagen: {e}")
|
||
return False
|
||
|
||
def test_performance():
|
||
"""Test der Performance von Cleanup-Operationen"""
|
||
print("\n🧪 Test 6: Performance-Test")
|
||
|
||
try:
|
||
cleanup_manager = DatabaseCleanupManager()
|
||
|
||
# Messe Zeit für verschiedene Operationen
|
||
operations = [
|
||
("WAL-Checkpoint", lambda: cleanup_manager.safe_wal_checkpoint(retry_attempts=1)),
|
||
("Verbindungsschließung", lambda: cleanup_manager.force_close_all_connections(max_wait_seconds=5)),
|
||
("Umfassendes Cleanup", lambda: cleanup_manager.comprehensive_cleanup(force_mode_switch=False))
|
||
]
|
||
|
||
for operation_name, operation_func in operations:
|
||
start_time = time.time()
|
||
|
||
try:
|
||
result = operation_func()
|
||
duration = time.time() - start_time
|
||
|
||
success = result if isinstance(result, bool) else result[0] if isinstance(result, tuple) else result.get("success", False)
|
||
|
||
if success:
|
||
print(f" ✅ {operation_name}: {duration:.3f}s")
|
||
else:
|
||
print(f" ⚠️ {operation_name}: {duration:.3f}s (mit Problemen)")
|
||
|
||
except Exception as e:
|
||
duration = time.time() - start_time
|
||
print(f" ❌ {operation_name}: {duration:.3f}s (Fehler: {e})")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ Test 6 fehlgeschlagen: {e}")
|
||
return False
|
||
|
||
def main():
|
||
"""Hauptfunktion für alle Tests"""
|
||
print("🚀 Starte DatabaseCleanupManager Tests")
|
||
print(f"Database-Pfad: {DATABASE_PATH}")
|
||
print(f"Zeitstempel: {datetime.now().isoformat()}")
|
||
print("=" * 60)
|
||
|
||
# Prüfe ob Datenbankdatei existiert
|
||
if not os.path.exists(DATABASE_PATH):
|
||
print(f"❌ Datenbankdatei nicht gefunden: {DATABASE_PATH}")
|
||
return False
|
||
|
||
# Führe alle Tests aus
|
||
tests = [
|
||
("Grundlegende Cleanup-Funktionalität", test_basic_cleanup),
|
||
("Cleanup bei gleichzeitigen Zugriffen", test_concurrent_access),
|
||
("Fehlerbehandlung und Recovery", test_error_recovery),
|
||
("Journal-Mode-Operationen", test_journal_mode_operations),
|
||
("Convenience-Funktion", test_convenience_function),
|
||
("Performance-Test", test_performance)
|
||
]
|
||
|
||
passed_tests = 0
|
||
failed_tests = 0
|
||
|
||
for test_name, test_func in tests:
|
||
try:
|
||
if test_func():
|
||
passed_tests += 1
|
||
print(f"✅ {test_name}: BESTANDEN")
|
||
else:
|
||
failed_tests += 1
|
||
print(f"❌ {test_name}: FEHLGESCHLAGEN")
|
||
except Exception as e:
|
||
failed_tests += 1
|
||
print(f"❌ {test_name}: EXCEPTION - {e}")
|
||
|
||
print("\n" + "=" * 60)
|
||
print(f"📊 Test-Ergebnis: {passed_tests} bestanden, {failed_tests} fehlgeschlagen")
|
||
|
||
if failed_tests == 0:
|
||
print("🎉 Alle Tests bestanden! DatabaseCleanupManager funktioniert korrekt.")
|
||
return True
|
||
else:
|
||
print(f"⚠️ {failed_tests} Test(s) fehlgeschlagen. Überprüfung erforderlich.")
|
||
return False
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
success = main()
|
||
sys.exit(0 if success else 1)
|
||
except KeyboardInterrupt:
|
||
print("\n🔄 Test durch Benutzer unterbrochen")
|
||
sys.exit(130)
|
||
except Exception as e:
|
||
print(f"💥 Kritischer Fehler beim Testen: {e}")
|
||
sys.exit(1) |