From cf077ffcb87606b07d349a5370476d5f747c7830 Mon Sep 17 00:00:00 2001 From: Till Tomczak Date: Mon, 26 May 2025 12:58:23 +0200 Subject: [PATCH] "feat: Implement printer test suite in backend" --- backend/app/app.py | 52 +++++++++++++---- backend/app/templates/printers.html | 26 ++++++--- backend/app/test_printers.py | 86 +++++++++++++++++++++++++++++ 3 files changed, 146 insertions(+), 18 deletions(-) create mode 100644 backend/app/test_printers.py diff --git a/backend/app/app.py b/backend/app/app.py index 83edc717b..e1aedc387 100644 --- a/backend/app/app.py +++ b/backend/app/app.py @@ -136,28 +136,43 @@ def check_printer_status(ip_address: str, timeout: int = 7) -> Tuple[str, bool]: Returns: Tuple[str, bool]: (Status, Aktiv) - Status ist "online" oder "offline", Aktiv ist True/False """ - if not ip_address: + if not ip_address or ip_address.strip() == "": + printers_logger.debug(f"Keine IP-Adresse angegeben") return "offline", False try: + # IP-Adresse validieren + import ipaddress + try: + ipaddress.ip_address(ip_address.strip()) + except ValueError: + printers_logger.warning(f"Ungültige IP-Adresse: {ip_address}") + return "offline", False + # Windows-spezifischer Ping-Befehl mit Timeout if os.name == 'nt': # Windows - cmd = ['ping', '-n', '1', '-w', str(timeout * 1000), ip_address] + cmd = ['ping', '-n', '1', '-w', str(timeout * 1000), ip_address.strip()] else: # Unix/Linux/macOS - cmd = ['ping', '-c', '1', '-W', str(timeout), ip_address] + cmd = ['ping', '-c', '1', '-W', str(timeout), ip_address.strip()] + + printers_logger.debug(f"Ping-Befehl für {ip_address}: {' '.join(cmd)}") # Ping ausführen mit Timeout result = subprocess.run( cmd, capture_output=True, text=True, - timeout=timeout + 1 # Zusätzlicher Timeout für subprocess + encoding='utf-8', + errors='ignore', # Ignoriere Unicode-Fehler + timeout=timeout + 2 # Zusätzlicher Timeout für subprocess ) # Erfolgreicher Ping (Return Code 0) if result.returncode == 0: + printers_logger.debug(f"Ping erfolgreich für {ip_address}") return "online", True else: + printers_logger.debug(f"Ping fehlgeschlagen für {ip_address} (Return Code: {result.returncode})") return "offline", False except subprocess.TimeoutExpired: @@ -869,36 +884,50 @@ def get_printers_status(): # Drucker-Daten für Status-Check vorbereiten printer_data = [] for printer in printers: + # Verwende plug_ip als primäre IP-Adresse, fallback auf ip_address + ip_to_check = printer.plug_ip if printer.plug_ip else printer.ip_address printer_data.append({ 'id': printer.id, 'name': printer.name, - 'ip_address': printer.ip_address, + 'ip_address': ip_to_check, 'location': printer.location }) # Status aller Drucker parallel überprüfen mit 7-Sekunden-Timeout printers_logger.info(f"Starte Status-Check für {len(printer_data)} Drucker mit 7-Sekunden-Timeout") - status_results = check_multiple_printers_status(printer_data, timeout=7) + + # Fallback: Wenn keine IP-Adressen vorhanden sind, alle als offline markieren + if not any(p['ip_address'] for p in printer_data): + printers_logger.warning("Keine IP-Adressen für Drucker gefunden - alle als offline markiert") + status_results = {p['id']: ("offline", False) for p in printer_data} + else: + status_results = check_multiple_printers_status(printer_data, timeout=7) # Ergebnisse zusammenstellen und Datenbank aktualisieren status_data = [] for printer in printers: if printer.id in status_results: status, active = status_results[printer.id] + # Mapping für Frontend-Kompatibilität + if status == "online": + frontend_status = "available" + else: + frontend_status = "offline" else: # Fallback falls kein Ergebnis vorliegt - status, active = "offline", False + frontend_status = "offline" + active = False # Status in der Datenbank aktualisieren - printer.status = status + printer.status = frontend_status printer.active = active status_data.append({ "id": printer.id, "name": printer.name, - "status": status, + "status": frontend_status, "active": active, - "ip_address": printer.ip_address, + "ip_address": printer.plug_ip if printer.plug_ip else printer.ip_address, "location": printer.location, "last_checked": datetime.now().isoformat() }) @@ -907,7 +936,8 @@ def get_printers_status(): db_session.commit() db_session.close() - printers_logger.info(f"Status-Check abgeschlossen: {len([s for s in status_data if s['status'] == 'online'])} von {len(status_data)} Drucker online") + online_count = len([s for s in status_data if s['status'] == 'available']) + printers_logger.info(f"Status-Check abgeschlossen: {online_count} von {len(status_data)} Drucker online") return jsonify(status_data) except Exception as e: diff --git a/backend/app/templates/printers.html b/backend/app/templates/printers.html index 86503a7c9..2e9c44efe 100644 --- a/backend/app/templates/printers.html +++ b/backend/app/templates/printers.html @@ -221,7 +221,7 @@ document.getElementById('printerDetailModal').classList.add('hidden'); } - // Load printers (verwendet jetzt auch Status-Check) + // Load printers (schnelles Laden ohne Status-Check) async function loadPrinters() { try { const response = await fetch('/api/printers'); @@ -575,10 +575,16 @@ } const statusData = await response.json(); + // Prüfe ob statusData ein Array ist + if (!Array.isArray(statusData)) { + throw new Error('Ungültige Antwort vom Server'); + } + // Drucker-Daten mit Status-Informationen anreichern printers = statusData.map(printer => ({ ...printer, - status: printer.status === 'online' ? 'available' : 'offline' + // Status ist bereits korrekt gemappt vom Backend + status: printer.status || 'offline' })); renderPrinters(); @@ -587,14 +593,20 @@ const onlineCount = printers.filter(p => p.status === 'available').length; const totalCount = printers.length; - showStatusMessage( - `Status-Check abgeschlossen: ${onlineCount} von ${totalCount} Drucker online`, - onlineCount === totalCount ? 'success' : 'info' - ); + if (totalCount > 0) { + showStatusMessage( + `Status-Check abgeschlossen: ${onlineCount} von ${totalCount} Drucker verfügbar`, + onlineCount > 0 ? 'success' : 'warning' + ); + } else { + showStatusMessage('Keine Drucker gefunden', 'info'); + } } catch (error) { console.error('Error loading printer status:', error); - showError('Fehler beim Überprüfen der Drucker-Status'); + showStatusMessage('Fehler beim Überprüfen der Drucker-Status: ' + error.message, 'error'); + // Fallback: Lade normale Drucker-Liste + loadPrinters(); } } diff --git a/backend/app/test_printers.py b/backend/app/test_printers.py new file mode 100644 index 000000000..0dac91928 --- /dev/null +++ b/backend/app/test_printers.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 +""" +Test-Skript für die Drucker-Funktionen +""" + +import sys +import os +sys.path.append(os.path.dirname(os.path.abspath(__file__))) + +from app import check_printer_status, check_multiple_printers_status +from models import get_db_session, Printer + +def test_ping_function(): + """Teste die Ping-Funktion mit localhost""" + print("Teste Ping-Funktion...") + + # Test mit localhost + status, active = check_printer_status("127.0.0.1", timeout=3) + print(f"Localhost Test: Status={status}, Active={active}") + + # Test mit ungültiger IP + status, active = check_printer_status("192.168.999.999", timeout=2) + print(f"Ungültige IP Test: Status={status}, Active={active}") + + # Test mit leerer IP + status, active = check_printer_status("", timeout=1) + print(f"Leere IP Test: Status={status}, Active={active}") + +def test_multiple_printers(): + """Teste die Mehrfach-Drucker-Funktion""" + print("\nTeste Mehrfach-Drucker-Funktion...") + + test_printers = [ + {'id': 1, 'name': 'Test1', 'ip_address': '127.0.0.1', 'location': 'Test'}, + {'id': 2, 'name': 'Test2', 'ip_address': '192.168.1.999', 'location': 'Test'}, + {'id': 3, 'name': 'Test3', 'ip_address': '', 'location': 'Test'} + ] + + results = check_multiple_printers_status(test_printers, timeout=3) + + for printer in test_printers: + printer_id = printer['id'] + if printer_id in results: + status, active = results[printer_id] + print(f"Drucker {printer['name']} ({printer['ip_address']}): {status}, {active}") + else: + print(f"Drucker {printer['name']}: Kein Ergebnis") + +def test_database_printers(): + """Teste mit echten Druckern aus der Datenbank""" + print("\nTeste echte Drucker aus der Datenbank...") + + try: + db_session = get_db_session() + printers = db_session.query(Printer).all() + + if not printers: + print("Keine Drucker in der Datenbank gefunden") + db_session.close() + return + + print(f"Gefunden: {len(printers)} Drucker") + + for printer in printers: + ip_to_check = printer.plug_ip if printer.plug_ip else printer.ip_address + print(f"Drucker: {printer.name}, IP: {ip_to_check}") + + if ip_to_check: + status, active = check_printer_status(ip_to_check, timeout=3) + print(f" -> Status: {status}, Active: {active}") + else: + print(f" -> Keine IP-Adresse verfügbar") + + db_session.close() + + except Exception as e: + print(f"Fehler beim Testen der Datenbank-Drucker: {e}") + +if __name__ == "__main__": + print("=== Drucker-Funktionen Test ===") + + test_ping_function() + test_multiple_printers() + test_database_printers() + + print("\n=== Test abgeschlossen ===") \ No newline at end of file