#!/usr/bin/env python3 """ Umfassende Tests für Tapo-Steckdosen-Integration und Drucker-Verwaltung Diese Tests stellen sicher, dass: 1. Alle 6 Drucker/Steckdosen immer angezeigt werden 2. Die 3 Status-Zustände korrekt funktionieren: an, aus, nicht erreichbar 3. Das automatische An-/Ausschalten basierend auf Reservierungen funktioniert 4. Die Kalender-Integration für Admins funktioniert """ import pytest import json import sys import os from datetime import datetime, timedelta from time import sleep # Pfad zum Backend hinzufügen sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) from app import app from models import init_database, User, Printer, Job, get_db_session from utils.logging_config import get_logger logger = get_logger("test_tapo_integration") # Test-Konfiguration TEST_USER = { "username": "testuser", "email": "test@mercedes-benz.com", "password": "TestPassword123!", "name": "Test User" } TEST_ADMIN = { "username": "testadmin", "email": "admin.test@mercedes-benz.com", "password": "AdminPassword123!", "name": "Test Admin" } # Die 6 Standard-Drucker mit Tapo-Steckdosen STANDARD_PRINTERS = [ { "name": "Drucker 1 - Halle A", "model": "Prusa MK3S+", "location": "Halle A - Arbeitsplatz 1", "ip_address": "192.168.1.101", "plug_ip": "192.168.1.201", "plug_username": "admin", "plug_password": "secure_password" }, { "name": "Drucker 2 - Halle A", "model": "Prusa MK3S+", "location": "Halle A - Arbeitsplatz 2", "ip_address": "192.168.1.102", "plug_ip": "192.168.1.202", "plug_username": "admin", "plug_password": "secure_password" }, { "name": "Drucker 3 - Halle B", "model": "Ultimaker S5", "location": "Halle B - Arbeitsplatz 1", "ip_address": "192.168.1.103", "plug_ip": "192.168.1.203", "plug_username": "admin", "plug_password": "secure_password" }, { "name": "Drucker 4 - Halle B", "model": "Ultimaker S5", "location": "Halle B - Arbeitsplatz 2", "ip_address": "192.168.1.104", "plug_ip": "192.168.1.204", "plug_username": "admin", "plug_password": "secure_password" }, { "name": "Drucker 5 - Labor", "model": "Formlabs Form 3", "location": "Labor - SLA-Bereich", "ip_address": "192.168.1.105", "plug_ip": "192.168.1.205", "plug_username": "admin", "plug_password": "secure_password" }, { "name": "Drucker 6 - Werkstatt", "model": "Markforged X7", "location": "Werkstatt - Spezialbereich", "ip_address": "192.168.1.106", "plug_ip": "192.168.1.206", "plug_username": "admin", "plug_password": "secure_password" } ] class TestTapoIntegration: """Test-Suite für Tapo-Steckdosen-Integration""" @pytest.fixture(autouse=True) def setup(self): """Setup für jeden Test""" self.app = app self.app.config['TESTING'] = True self.app.config['WTF_CSRF_ENABLED'] = False self.client = self.app.test_client() with self.app.app_context(): # Datenbank initialisieren init_database() # Test-Benutzer erstellen db_session = get_db_session() # Normaler Benutzer self.user = User() self.user.username = TEST_USER["username"] self.user.email = TEST_USER["email"] self.user.set_password(TEST_USER["password"]) self.user.name = TEST_USER["name"] self.user.role = "user" db_session.add(self.user) # Admin-Benutzer self.admin = User() self.admin.username = TEST_ADMIN["username"] self.admin.email = TEST_ADMIN["email"] self.admin.set_password(TEST_ADMIN["password"]) self.admin.name = TEST_ADMIN["name"] self.admin.role = "admin" db_session.add(self.admin) db_session.commit() # Standard-Drucker erstellen self._create_standard_printers() db_session.close() yield # Cleanup nach Test with self.app.app_context(): db_session = get_db_session() db_session.query(Job).delete() db_session.query(Printer).delete() db_session.query(User).delete() db_session.commit() db_session.close() def _create_standard_printers(self): """Erstellt die 6 Standard-Drucker""" db_session = get_db_session() for printer_data in STANDARD_PRINTERS: printer = Printer() printer.name = printer_data["name"] printer.model = printer_data["model"] printer.location = printer_data["location"] printer.ip_address = printer_data["ip_address"] printer.plug_ip = printer_data["plug_ip"] printer.plug_username = printer_data["plug_username"] printer.plug_password = printer_data["plug_password"] printer.status = "offline" # Initial-Status printer.active = True db_session.add(printer) db_session.commit() def _login_as_admin(self): """Login als Admin""" response = self.client.post('/api/login', json={ 'username': TEST_ADMIN['username'], 'password': TEST_ADMIN['password'] }) assert response.status_code == 200 return response.get_json() def _login_as_user(self): """Login als normaler Benutzer""" response = self.client.post('/api/login', json={ 'username': TEST_USER['username'], 'password': TEST_USER['password'] }) assert response.status_code == 200 return response.get_json() def test_all_printers_always_visible(self): """Test: Alle 6 Drucker werden immer angezeigt""" # Als Benutzer einloggen self._login_as_user() # Drucker abrufen response = self.client.get('/api/printers') assert response.status_code == 200 data = response.get_json() assert data['success'] is True assert data['count'] == 6 assert len(data['printers']) == 6 # Prüfen, dass alle Drucker vorhanden sind printer_names = [p['name'] for p in data['printers']] for expected_printer in STANDARD_PRINTERS: assert expected_printer['name'] in printer_names def test_printer_status_types(self): """Test: Die 3 Status-Typen (an, aus, nicht erreichbar)""" # Als Admin einloggen self._login_as_admin() # Status abrufen response = self.client.get('/api/printers/status') assert response.status_code == 200 data = response.get_json() assert data['success'] is True assert len(data['printers']) == 6 # Prüfen, dass alle Drucker einen gültigen Status haben for printer in data['printers']: assert printer['has_plug'] is True assert printer['plug_ip'] is not None # Status sollte einer der erwarteten sein assert printer['plug_status'] in ['on', 'off', 'unreachable', 'error', 'unavailable'] def test_crud_operations_printers(self): """Test: CRUD-Operationen für Drucker""" # Als Admin einloggen self._login_as_admin() # CREATE: Neuen Drucker erstellen new_printer_data = { "name": "Test-Drucker 7", "model": "Test Model", "location": "Test Location", "ip_address": "192.168.1.107", "plug_ip": "192.168.1.207", "plug_username": "test", "plug_password": "test123" } response = self.client.post('/api/admin/printers', json=new_printer_data) assert response.status_code == 201 data = response.get_json() assert data['success'] is True printer_id = data['printer']['id'] # READ: Einzelnen Drucker abrufen response = self.client.get(f'/api/admin/printers/{printer_id}') assert response.status_code == 200 data = response.get_json() assert data['printer']['name'] == new_printer_data['name'] # UPDATE: Drucker aktualisieren update_data = { "name": "Test-Drucker 7 Updated", "location": "Updated Location" } response = self.client.put(f'/api/admin/printers/{printer_id}', json=update_data) assert response.status_code == 200 data = response.get_json() assert data['printer']['name'] == update_data['name'] assert data['printer']['location'] == update_data['location'] # DELETE: Drucker löschen response = self.client.delete(f'/api/admin/printers/{printer_id}') assert response.status_code == 200 # Verifizieren, dass gelöscht wurde response = self.client.get(f'/api/admin/printers/{printer_id}') assert response.status_code == 404 def test_automatic_plug_control_with_jobs(self): """Test: Automatisches An-/Ausschalten der Steckdosen basierend auf Jobs""" # Als Benutzer einloggen self._login_as_user() # Job für in 5 Minuten erstellen start_time = datetime.now() + timedelta(minutes=5) end_time = start_time + timedelta(hours=2) job_data = { "name": "Test-Druckauftrag", "description": "Automatischer Steuerungstest", "printer_id": 1, "start_at": start_time.isoformat(), "end_at": end_time.isoformat(), "duration_minutes": 120 } response = self.client.post('/api/jobs', json=job_data) assert response.status_code in [200, 201] job_id = response.get_json()['id'] # Job-Details abrufen response = self.client.get(f'/api/jobs/{job_id}') assert response.status_code == 200 job = response.get_json() assert job['status'] == 'scheduled' # Simulieren: Zeit vergeht, Job sollte starten # (In der Realität würde der Scheduler dies automatisch tun) # Job manuell starten (simuliert Scheduler) response = self.client.post(f'/api/jobs/{job_id}/start') assert response.status_code == 200 # Prüfen, dass Job läuft response = self.client.get(f'/api/jobs/{job_id}') assert response.status_code == 200 job = response.get_json() assert job['status'] == 'running' def test_calendar_shows_printer_status_for_admin(self): """Test: Kalender zeigt Drucker-Status für Admin""" # Als Admin einloggen self._login_as_admin() # Jobs für verschiedene Zeiträume erstellen now = datetime.now() # Job 1: Läuft gerade running_job = { "name": "Laufender Druck", "printer_id": 1, "start_at": (now - timedelta(hours=1)).isoformat(), "end_at": (now + timedelta(hours=1)).isoformat(), "duration_minutes": 120, "status": "running" } # Job 2: Geplant für später scheduled_job = { "name": "Geplanter Druck", "printer_id": 2, "start_at": (now + timedelta(hours=2)).isoformat(), "end_at": (now + timedelta(hours=4)).isoformat(), "duration_minutes": 120, "status": "scheduled" } # Jobs erstellen for job_data in [running_job, scheduled_job]: response = self.client.post('/api/jobs', json=job_data) assert response.status_code in [200, 201] # Kalender-Events abrufen response = self.client.get('/api/calendar/events') assert response.status_code == 200 events = response.get_json() assert len(events) >= 2 # Prüfen, dass Events Drucker-Status enthalten for event in events: if 'extendedProps' in event: assert 'printer_id' in event['extendedProps'] assert 'status' in event['extendedProps'] def test_printer_status_persistence(self): """Test: Drucker-Status wird korrekt gespeichert und abgerufen""" # Als Admin einloggen self._login_as_admin() # Status für alle Drucker abrufen response = self.client.get('/api/printers/status') assert response.status_code == 200 initial_status = response.get_json() # Manuell einen Drucker "einschalten" (simuliert) response = self.client.post('/api/tapo/control', json={ 'printer_id': 1, 'action': 'on' }) # Kann fehlschlagen wenn Steckdose nicht erreichbar # assert response.status_code == 200 # Status erneut abrufen response = self.client.get('/api/printers/status') assert response.status_code == 200 updated_status = response.get_json() # Mindestens die Anzahl sollte gleich bleiben assert len(updated_status['printers']) == len(initial_status['printers']) def test_error_handling_unreachable_plugs(self): """Test: Fehlerbehandlung für nicht erreichbare Steckdosen""" # Als Admin einloggen self._login_as_admin() # Versuchen, eine nicht erreichbare Steckdose zu steuern response = self.client.post('/api/tapo/control', json={ 'printer_id': 1, 'action': 'on' }) # Sollte entweder erfolgreich sein oder einen kontrollierten Fehler zurückgeben assert response.status_code in [200, 400, 500] if response.status_code != 200: data = response.get_json() assert 'error' in data or 'message' in data def test_concurrent_job_scheduling(self): """Test: Mehrere Jobs gleichzeitig planen""" # Als Benutzer einloggen self._login_as_user() # Mehrere Jobs für verschiedene Drucker erstellen start_time = datetime.now() + timedelta(hours=1) jobs = [] for i in range(3): job_data = { "name": f"Concurrent Job {i+1}", "printer_id": i+1, "start_at": start_time.isoformat(), "end_at": (start_time + timedelta(hours=2)).isoformat(), "duration_minutes": 120 } response = self.client.post('/api/jobs', json=job_data) assert response.status_code in [200, 201] jobs.append(response.get_json()) # Prüfen, dass alle Jobs erstellt wurden assert len(jobs) == 3 # Alle Jobs abrufen response = self.client.get('/api/jobs') assert response.status_code == 200 all_jobs = response.get_json() assert len(all_jobs) >= 3 def test_admin_dashboard_printer_overview(self): """Test: Admin-Dashboard zeigt Drucker-Übersicht""" # Als Admin einloggen self._login_as_admin() # Dashboard-Daten abrufen response = self.client.get('/api/admin/dashboard') if response.status_code == 404: # Alternative: Stats abrufen response = self.client.get('/api/stats') if response.status_code == 200: data = response.get_json() # Prüfen auf relevante Daten assert isinstance(data, dict) def run_tests(): """Führt die Tests aus""" logger.info("Starte Tapo-Integrationstests...") # Pytest ausführen pytest_args = [ __file__, '-v', # Verbose '-s', # Keine Capture, zeige print-Ausgaben '--tb=short' # Kurze Traceback-Ausgabe ] result = pytest.main(pytest_args) if result == 0: logger.info("✅ Alle Tests erfolgreich bestanden!") else: logger.error("❌ Tests fehlgeschlagen!") return result if __name__ == "__main__": sys.exit(run_tests())