#!/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)