"feat: Implement printer test suite in backend"

This commit is contained in:
2025-05-26 12:58:23 +02:00
parent 2643ef1814
commit cf077ffcb8
3 changed files with 146 additions and 18 deletions

View File

@@ -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:

View File

@@ -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();
}
}

View File

@@ -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 ===")