""" Tests für Classroom Engine und API. Testet: - LessonSession Model - LessonStateMachine (FSM) - PhaseTimer (inkl. Pause) - SuggestionEngine - REST API Endpoints Note: Some tests (Analytics, Reflections) require a running PostgreSQL database. """ import pytest from datetime import datetime, timedelta from fastapi.testclient import TestClient from unittest.mock import patch import sys import os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # Marker for tests requiring PostgreSQL # These tests will be skipped in CI (via conftest.py) or when DB is unavailable requires_postgres = pytest.mark.requires_postgres from main import app from classroom_engine import ( LessonPhase, LessonSession, LessonStateMachine, PhaseTimer, SuggestionEngine, LESSON_PHASES, get_default_durations, ) client = TestClient(app) # ==================== MODEL TESTS ==================== class TestLessonPhase: """Tests für LessonPhase Enum.""" def test_all_phases_defined(self): """Testet dass alle 7 Phasen definiert sind.""" phases = list(LessonPhase) assert len(phases) == 7 assert LessonPhase.NOT_STARTED in phases assert LessonPhase.EINSTIEG in phases assert LessonPhase.ERARBEITUNG in phases assert LessonPhase.SICHERUNG in phases assert LessonPhase.TRANSFER in phases assert LessonPhase.REFLEXION in phases assert LessonPhase.ENDED in phases def test_phase_values(self): """Testet Phase-Werte.""" assert LessonPhase.NOT_STARTED.value == "not_started" assert LessonPhase.EINSTIEG.value == "einstieg" assert LessonPhase.ENDED.value == "ended" class TestLessonSession: """Tests für LessonSession Model.""" def test_create_session(self): """Testet Erstellung einer Session.""" session = LessonSession( session_id="test-123", teacher_id="teacher-1", class_id="7a", subject="Mathematik" ) assert session.session_id == "test-123" assert session.teacher_id == "teacher-1" assert session.current_phase == LessonPhase.NOT_STARTED assert session.is_paused == False assert session.total_paused_seconds == 0 def test_default_durations(self): """Testet Standard-Phasendauern.""" durations = get_default_durations() assert durations["einstieg"] == 8 assert durations["erarbeitung"] == 20 # 20 Minuten Hauptarbeitsphase assert durations["sicherung"] == 10 assert durations["transfer"] == 7 assert durations["reflexion"] == 5 def test_to_dict(self): """Testet Serialisierung zu Dictionary.""" session = LessonSession( session_id="test-123", teacher_id="teacher-1", class_id="7a", subject="Mathematik", topic="Bruchrechnung" ) data = session.to_dict() assert data["session_id"] == "test-123" assert data["subject"] == "Mathematik" assert data["topic"] == "Bruchrechnung" assert data["is_paused"] == False assert data["total_paused_seconds"] == 0 # ==================== STATE MACHINE TESTS ==================== class TestLessonStateMachine: """Tests für LessonStateMachine.""" def test_initial_phase_transition(self): """Testet Übergang von NOT_STARTED zu EINSTIEG.""" fsm = LessonStateMachine() session = LessonSession( session_id="test", teacher_id="t1", class_id="7a", subject="Deutsch" ) # Start lesson session = fsm.transition(session, LessonPhase.EINSTIEG) assert session.current_phase == LessonPhase.EINSTIEG assert session.phase_started_at is not None assert session.lesson_started_at is not None def test_full_lesson_flow(self): """Testet kompletten Unterrichtsablauf.""" fsm = LessonStateMachine() session = LessonSession( session_id="test", teacher_id="t1", class_id="7a", subject="Mathematik" ) # Durchlaufe alle Phasen phases = [ LessonPhase.EINSTIEG, LessonPhase.ERARBEITUNG, LessonPhase.SICHERUNG, LessonPhase.TRANSFER, LessonPhase.REFLEXION, LessonPhase.ENDED ] for phase in phases: session = fsm.transition(session, phase) assert session.current_phase == phase # Am Ende sollte Stunde beendet sein assert fsm.is_lesson_ended(session) assert not fsm.is_lesson_active(session) def test_next_phase(self): """Testet next_phase Funktion.""" fsm = LessonStateMachine() assert fsm.next_phase(LessonPhase.NOT_STARTED) == LessonPhase.EINSTIEG assert fsm.next_phase(LessonPhase.EINSTIEG) == LessonPhase.ERARBEITUNG assert fsm.next_phase(LessonPhase.REFLEXION) == LessonPhase.ENDED assert fsm.next_phase(LessonPhase.ENDED) is None def test_invalid_transition(self): """Testet dass ungültige Übergänge blockiert werden.""" fsm = LessonStateMachine() # Von NOT_STARTED kann man nicht direkt zu ERARBEITUNG assert not fsm.can_transition(LessonPhase.NOT_STARTED, LessonPhase.ERARBEITUNG) # Von EINSTIEG kann man nicht zu SICHERUNG springen (muss durch ERARBEITUNG) assert not fsm.can_transition(LessonPhase.EINSTIEG, LessonPhase.SICHERUNG) def test_phase_history(self): """Testet Phase-History.""" fsm = LessonStateMachine() session = LessonSession( session_id="test", teacher_id="t1", class_id="7a", subject="Deutsch" ) session = fsm.transition(session, LessonPhase.EINSTIEG) session = fsm.transition(session, LessonPhase.ERARBEITUNG) # History enthält 1 Eintrag: EINSTIEG wurde abgeschlossen assert len(session.phase_history) == 1 assert session.phase_history[0]["phase"] == "einstieg" # Eine weitere Phase durchlaufen session = fsm.transition(session, LessonPhase.SICHERUNG) assert len(session.phase_history) == 2 assert session.phase_history[1]["phase"] == "erarbeitung" # ==================== TIMER TESTS ==================== class TestPhaseTimer: """Tests für PhaseTimer.""" def test_remaining_seconds(self): """Testet verbleibende Zeit Berechnung.""" timer = PhaseTimer() session = LessonSession( session_id="test", teacher_id="t1", class_id="7a", subject="Deutsch" ) # Starte Phase session.current_phase = LessonPhase.EINSTIEG session.phase_started_at = datetime.utcnow() remaining = timer.get_remaining_seconds(session) # Sollte nahe an 8 Minuten (480 Sekunden) sein assert 475 <= remaining <= 480 def test_pause_stops_timer(self): """Testet dass Pause den Timer anhält.""" timer = PhaseTimer() session = LessonSession( session_id="test", teacher_id="t1", class_id="7a", subject="Deutsch" ) # Starte Phase vor 2 Minuten session.current_phase = LessonPhase.EINSTIEG session.phase_started_at = datetime.utcnow() - timedelta(minutes=2) # Ohne Pause: ca. 6 Minuten übrig (8 Min Gesamt - 2 Min verstrichene Zeit) remaining_no_pause = timer.get_remaining_seconds(session) assert 355 <= remaining_no_pause <= 365 # ca. 6 Minuten # Wenn wir vor 1 Minute pausiert haben und immer noch pausiert sind: session.is_paused = True session.pause_started_at = datetime.utcnow() - timedelta(minutes=1) # Jetzt sollten wir ca. 7 Minuten übrig haben (8 Min - 1 Min effektive Zeit) # Weil die Pause-Zeit (1 Min) von der verstrichenen Zeit abgezogen wird remaining_with_pause = timer.get_remaining_seconds(session) # Mit Pause sollte mehr Zeit übrig sein als ohne assert remaining_with_pause > remaining_no_pause def test_total_paused_seconds(self): """Testet kumulative Pausenzeit.""" timer = PhaseTimer() session = LessonSession( session_id="test", teacher_id="t1", class_id="7a", subject="Deutsch" ) # Starte Phase vor 5 Minuten session.current_phase = LessonPhase.EINSTIEG session.phase_started_at = datetime.utcnow() - timedelta(minutes=5) session.total_paused_seconds = 60 # 1 Minute Pause # Sollte 4 Minuten effektiv verstrichene Zeit haben elapsed = timer.get_elapsed_seconds(session) assert 235 <= elapsed <= 245 # ca. 4 Minuten def test_warning_threshold(self): """Testet Warnung bei 2 Minuten vor Ende.""" timer = PhaseTimer() session = LessonSession( session_id="test", teacher_id="t1", class_id="7a", subject="Deutsch" ) # Phase mit weniger als 2 Minuten übrig session.current_phase = LessonPhase.EINSTIEG session.phase_started_at = datetime.utcnow() - timedelta(minutes=7) assert timer.is_warning(session) def test_overtime(self): """Testet Overtime Erkennung.""" timer = PhaseTimer() session = LessonSession( session_id="test", teacher_id="t1", class_id="7a", subject="Deutsch" ) # Phase mit abgelaufener Zeit session.current_phase = LessonPhase.EINSTIEG session.phase_started_at = datetime.utcnow() - timedelta(minutes=10) assert timer.is_overtime(session) assert timer.get_overtime_seconds(session) > 0 def test_phase_status_includes_is_paused(self): """Testet dass phase_status is_paused enthält.""" timer = PhaseTimer() session = LessonSession( session_id="test", teacher_id="t1", class_id="7a", subject="Deutsch" ) session.current_phase = LessonPhase.EINSTIEG session.phase_started_at = datetime.utcnow() session.is_paused = True status = timer.get_phase_status(session) assert "is_paused" in status assert status["is_paused"] == True # ==================== SUGGESTION ENGINE TESTS ==================== class TestSuggestionEngine: """Tests für SuggestionEngine.""" def test_suggestions_for_einstieg(self): """Testet Vorschläge für Einstiegsphase.""" engine = SuggestionEngine() session = LessonSession( session_id="test", teacher_id="t1", class_id="7a", subject="Deutsch" ) session.current_phase = LessonPhase.EINSTIEG suggestions = engine.get_suggestions(session, limit=3) assert len(suggestions) <= 3 # Jeder Vorschlag hat einen title und description assert all(s.title and s.description for s in suggestions) def test_suggestions_for_each_phase(self): """Testet dass jede Phase Vorschläge hat.""" engine = SuggestionEngine() active_phases = [ LessonPhase.EINSTIEG, LessonPhase.ERARBEITUNG, LessonPhase.SICHERUNG, LessonPhase.TRANSFER, LessonPhase.REFLEXION ] for phase in active_phases: session = LessonSession( session_id="test", teacher_id="t1", class_id="7a", subject="Deutsch" ) session.current_phase = phase suggestions = engine.get_suggestions(session, limit=5) assert len(suggestions) > 0, f"Keine Vorschläge für {phase.value}" def test_subject_specific_suggestions(self): """Testet fachspezifische Vorschlaege (Feature f18).""" engine = SuggestionEngine() # Mathematik-Session math_session = LessonSession( session_id="test-math", teacher_id="t1", class_id="7a", subject="Mathematik" ) math_session.current_phase = LessonPhase.EINSTIEG math_suggestions = engine.get_suggestions(math_session, limit=5) assert len(math_suggestions) > 0 # Mindestens ein Mathe-Vorschlag sollte dabei sein has_math_specific = any( s.subjects and "mathematik" in s.subjects for s in math_suggestions ) assert has_math_specific, "Keine Mathe-spezifischen Vorschlaege gefunden" # Informatik-Session info_session = LessonSession( session_id="test-info", teacher_id="t1", class_id="10b", subject="Informatik" ) info_session.current_phase = LessonPhase.ERARBEITUNG info_suggestions = engine.get_suggestions(info_session, limit=5) has_info_specific = any( s.subjects and "informatik" in s.subjects for s in info_suggestions ) assert has_info_specific, "Keine Informatik-spezifischen Vorschlaege gefunden" def test_suggestions_response_includes_subject_info(self): """Testet dass Response Fach-Info enthaelt (Feature f18).""" engine = SuggestionEngine() session = LessonSession( session_id="test", teacher_id="t1", class_id="7a", subject="Deutsch" ) session.current_phase = LessonPhase.EINSTIEG response = engine.get_suggestions_response(session, limit=3) assert "subject" in response assert "subject_specific_available" in response assert response["subject"] == "Deutsch" # ==================== API TESTS ==================== class TestClassroomAPI: """Tests für Classroom REST API.""" def test_create_session(self): """Testet Session-Erstellung via API.""" response = client.post("/api/classroom/sessions", json={ "teacher_id": "test-teacher", "class_id": "7a", "subject": "Mathematik", "topic": "Bruchrechnung" }) assert response.status_code == 200 data = response.json() assert data["subject"] == "Mathematik" assert data["current_phase"] == "not_started" assert data["is_paused"] == False def test_start_session(self): """Testet Stunden-Start via API.""" # Session erstellen create_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "test-teacher", "class_id": "7b", "subject": "Deutsch" }) session_id = create_resp.json()["session_id"] # Stunde starten start_resp = client.post(f"/api/classroom/sessions/{session_id}/start") assert start_resp.status_code == 200 data = start_resp.json() assert data["current_phase"] == "einstieg" assert data["is_active"] == True def test_next_phase(self): """Testet Phasenwechsel via API.""" # Session erstellen und starten create_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "test-teacher", "class_id": "7c", "subject": "Englisch" }) session_id = create_resp.json()["session_id"] client.post(f"/api/classroom/sessions/{session_id}/start") # Nächste Phase next_resp = client.post(f"/api/classroom/sessions/{session_id}/next-phase") assert next_resp.status_code == 200 assert next_resp.json()["current_phase"] == "erarbeitung" def test_pause_toggle(self): """Testet Pause-Toggle via API.""" # Session erstellen und starten create_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "test-teacher", "class_id": "7d", "subject": "Geschichte" }) session_id = create_resp.json()["session_id"] client.post(f"/api/classroom/sessions/{session_id}/start") # Pausieren pause_resp = client.post(f"/api/classroom/sessions/{session_id}/pause") assert pause_resp.status_code == 200 assert pause_resp.json()["is_paused"] == True # Fortsetzen resume_resp = client.post(f"/api/classroom/sessions/{session_id}/pause") assert resume_resp.status_code == 200 assert resume_resp.json()["is_paused"] == False def test_extend_phase(self): """Testet Phase-Verlängerung via API.""" # Session erstellen und starten create_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "test-teacher", "class_id": "7e", "subject": "Biologie" }) session_id = create_resp.json()["session_id"] start_resp = client.post(f"/api/classroom/sessions/{session_id}/start") # Ursprüngliche Timer-Zeit merken initial_total = start_resp.json()["timer"]["total_seconds"] # Phase um 5 Minuten verlängern extend_resp = client.post( f"/api/classroom/sessions/{session_id}/extend", json={"minutes": 5} ) assert extend_resp.status_code == 200 new_total = extend_resp.json()["timer"]["total_seconds"] assert new_total == initial_total + 300 # +5 Minuten = +300 Sekunden def test_get_timer(self): """Testet Timer-Abruf via API.""" # Session erstellen und starten create_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "test-teacher", "class_id": "7f", "subject": "Physik" }) session_id = create_resp.json()["session_id"] client.post(f"/api/classroom/sessions/{session_id}/start") # Timer abrufen timer_resp = client.get(f"/api/classroom/sessions/{session_id}/timer") assert timer_resp.status_code == 200 data = timer_resp.json() assert "remaining_seconds" in data assert "percentage" in data assert "is_paused" in data assert data["is_paused"] == False def test_get_suggestions(self): """Testet Vorschläge-Abruf via API.""" # Session erstellen und starten create_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "test-teacher", "class_id": "7g", "subject": "Chemie" }) session_id = create_resp.json()["session_id"] client.post(f"/api/classroom/sessions/{session_id}/start") # Vorschläge abrufen suggestions_resp = client.get( f"/api/classroom/sessions/{session_id}/suggestions", params={"limit": 3} ) assert suggestions_resp.status_code == 200 data = suggestions_resp.json() assert "suggestions" in data assert "current_phase" in data assert len(data["suggestions"]) <= 3 def test_list_phases(self): """Testet Phasen-Liste via API.""" response = client.get("/api/classroom/phases") assert response.status_code == 200 data = response.json() assert "phases" in data assert len(data["phases"]) == 5 # 5 aktive Phasen def test_pause_inactive_session_fails(self): """Testet dass Pause bei nicht-aktiver Session fehlschlägt.""" # Session erstellen, aber nicht starten create_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "test-teacher", "class_id": "7h", "subject": "Kunst" }) session_id = create_resp.json()["session_id"] # Pausieren sollte fehlschlagen pause_resp = client.post(f"/api/classroom/sessions/{session_id}/pause") assert pause_resp.status_code == 400 def test_extend_inactive_session_fails(self): """Testet dass Extend bei nicht-aktiver Session fehlschlägt.""" # Session erstellen, aber nicht starten create_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "test-teacher", "class_id": "7i", "subject": "Musik" }) session_id = create_resp.json()["session_id"] # Extend sollte fehlschlagen extend_resp = client.post( f"/api/classroom/sessions/{session_id}/extend", json={"minutes": 5} ) assert extend_resp.status_code == 400 # ==================== HOMEWORK API TESTS (Feature f20) ==================== class TestHomeworkAPI: """Tests fuer Homework REST API (Feature f20).""" def test_create_homework(self): """Testet Hausaufgaben-Erstellung.""" response = client.post("/api/classroom/homework", json={ "teacher_id": "test-teacher", "class_id": "7a", "subject": "Mathematik", "title": "Aufgabe 5 auf Seite 42", "description": "Alle Teilaufgaben bearbeiten" }) assert response.status_code == 201 data = response.json() assert data["title"] == "Aufgabe 5 auf Seite 42" assert data["status"] == "assigned" assert data["class_id"] == "7a" def test_create_homework_with_due_date(self): """Testet Hausaufgaben mit Faelligkeitsdatum.""" response = client.post("/api/classroom/homework", json={ "teacher_id": "test-teacher", "class_id": "7a", "subject": "Deutsch", "title": "Aufsatz schreiben", "due_date": "2026-01-20T23:59:00" }) assert response.status_code == 201 data = response.json() assert data["due_date"] is not None assert "2026-01-20" in data["due_date"] def test_list_homework_by_teacher(self): """Testet Auflisten von Hausaufgaben nach Lehrer.""" # Erst Hausaufgabe erstellen client.post("/api/classroom/homework", json={ "teacher_id": "list-test-teacher", "class_id": "8b", "subject": "Englisch", "title": "Vocabulary Unit 5" }) # Dann Liste abrufen response = client.get("/api/classroom/homework?teacher_id=list-test-teacher") assert response.status_code == 200 data = response.json() assert "homework" in data assert data["total"] >= 1 def test_update_homework_status(self): """Testet Status-Update einer Hausaufgabe.""" # Hausaufgabe erstellen create_resp = client.post("/api/classroom/homework", json={ "teacher_id": "status-test-teacher", "class_id": "9c", "subject": "Physik", "title": "Experiment durchfuehren" }) homework_id = create_resp.json()["homework_id"] # Status aktualisieren response = client.patch(f"/api/classroom/homework/{homework_id}/status?status=completed") assert response.status_code == 200 data = response.json() assert data["status"] == "completed" def test_delete_homework(self): """Testet Loeschen einer Hausaufgabe.""" # Hausaufgabe erstellen create_resp = client.post("/api/classroom/homework", json={ "teacher_id": "delete-test-teacher", "class_id": "10d", "subject": "Chemie", "title": "Formel auswendig lernen" }) homework_id = create_resp.json()["homework_id"] # Loeschen response = client.delete(f"/api/classroom/homework/{homework_id}") assert response.status_code == 200 assert response.json()["status"] == "deleted" def test_homework_not_found(self): """Testet 404 bei nicht existierender Hausaufgabe.""" response = client.get("/api/classroom/homework/nonexistent-id") assert response.status_code == 404 # ==================== MATERIALS API TESTS (Feature f19) ==================== class TestMaterialsAPI: """Tests fuer Materials REST API (Feature f19).""" def test_create_material(self): """Testet Material-Erstellung.""" response = client.post("/api/classroom/materials", json={ "teacher_id": "test-teacher", "title": "Arbeitsblatt Bruchrechnung", "material_type": "worksheet", "url": "https://example.com/ab-bruch.pdf", "description": "Uebungsaufgaben zu Bruechen", "phase": "erarbeitung", "subject": "Mathematik", "tags": ["bruchrechnung", "uebung"] }) assert response.status_code == 201 data = response.json() assert data["title"] == "Arbeitsblatt Bruchrechnung" assert data["material_type"] == "worksheet" assert data["phase"] == "erarbeitung" def test_create_video_material(self): """Testet Video-Material-Erstellung.""" response = client.post("/api/classroom/materials", json={ "teacher_id": "test-teacher", "title": "Erklaervideo Photosynthese", "material_type": "video", "url": "https://youtube.com/watch?v=abc123", "phase": "einstieg", "subject": "Biologie" }) assert response.status_code == 201 data = response.json() assert data["material_type"] == "video" def test_list_materials_by_teacher(self): """Testet Auflisten von Materialien nach Lehrer.""" # Erst Material erstellen client.post("/api/classroom/materials", json={ "teacher_id": "list-test-teacher", "title": "Test-Praesentation", "material_type": "presentation", "phase": "sicherung" }) # Dann Liste abrufen response = client.get("/api/classroom/materials?teacher_id=list-test-teacher") assert response.status_code == 200 data = response.json() assert "materials" in data assert data["total"] >= 1 def test_get_materials_by_phase(self): """Testet Abrufen von Materialien nach Phase.""" # Material fuer Phase erstellen client.post("/api/classroom/materials", json={ "teacher_id": "phase-test-teacher", "title": "Einstiegs-Material", "material_type": "link", "phase": "einstieg" }) # Nach Phase filtern response = client.get("/api/classroom/materials/by-phase/einstieg?teacher_id=phase-test-teacher") assert response.status_code == 200 data = response.json() assert all(m["phase"] == "einstieg" for m in data["materials"]) def test_update_material(self): """Testet Material-Update.""" # Material erstellen create_resp = client.post("/api/classroom/materials", json={ "teacher_id": "update-test-teacher", "title": "Original Titel", "material_type": "document" }) material_id = create_resp.json()["material_id"] # Aktualisieren response = client.put(f"/api/classroom/materials/{material_id}", json={ "title": "Aktualisierter Titel", "is_public": True }) assert response.status_code == 200 data = response.json() assert data["title"] == "Aktualisierter Titel" assert data["is_public"] == True def test_attach_material_to_session(self): """Testet Material-Session-Verknuepfung.""" # Session erstellen session_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "attach-test-teacher", "class_id": "10a", "subject": "Geschichte" }) session_id = session_resp.json()["session_id"] # Material erstellen material_resp = client.post("/api/classroom/materials", json={ "teacher_id": "attach-test-teacher", "title": "Quellentext", "material_type": "document" }) material_id = material_resp.json()["material_id"] # Verknuepfen response = client.post(f"/api/classroom/materials/{material_id}/attach/{session_id}") assert response.status_code == 200 data = response.json() assert data["session_id"] == session_id assert data["usage_count"] == 1 def test_delete_material(self): """Testet Material-Loeschung.""" # Material erstellen create_resp = client.post("/api/classroom/materials", json={ "teacher_id": "delete-test-teacher", "title": "Zu loeschendes Material", "material_type": "other" }) material_id = create_resp.json()["material_id"] # Loeschen response = client.delete(f"/api/classroom/materials/{material_id}") assert response.status_code == 200 assert response.json()["status"] == "deleted" def test_material_not_found(self): """Testet 404 bei nicht existierendem Material.""" response = client.get("/api/classroom/materials/nonexistent-id") assert response.status_code == 404 # ==================== ANALYTICS TESTS (Phase 5) ==================== @requires_postgres class TestAnalyticsAPI: """Tests fuer Analytics API (Phase 5). Requires PostgreSQL for storing and querying analytics data. """ def test_get_teacher_analytics(self): """Testet Lehrer-Analytics Endpoint.""" response = client.get("/api/classroom/analytics/teacher/demo-teacher?days=30") # Kann 503 sein wenn DB nicht verfuegbar, oder 200 mit Daten assert response.status_code in [200, 503] if response.status_code == 200: data = response.json() assert "teacher_id" in data assert "total_sessions" in data assert "avg_phase_durations" in data def test_get_phase_trends(self): """Testet Phase-Trends Endpoint.""" response = client.get("/api/classroom/analytics/phase-trends/demo-teacher/erarbeitung?limit=10") assert response.status_code in [200, 503] if response.status_code == 200: data = response.json() assert "phase" in data assert data["phase"] == "erarbeitung" assert "data_points" in data def test_get_phase_trends_invalid_phase(self): """Testet Phase-Trends mit ungueltiger Phase.""" response = client.get("/api/classroom/analytics/phase-trends/demo-teacher/invalid_phase") assert response.status_code == 400 def test_get_overtime_analysis(self): """Testet Overtime-Analyse Endpoint.""" response = client.get("/api/classroom/analytics/overtime/demo-teacher?limit=20") assert response.status_code in [200, 503] if response.status_code == 200: data = response.json() assert "phases" in data # Alle 5 Phasen sollten vorhanden sein for phase in ["einstieg", "erarbeitung", "sicherung", "transfer", "reflexion"]: assert phase in data["phases"] @requires_postgres class TestReflectionAPI: """Tests fuer Reflection API (Phase 5). Requires PostgreSQL for persisting reflections. """ def test_create_reflection(self): """Testet Reflection-Erstellung.""" # Erst Session erstellen session_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "reflection-test-teacher", "class_id": "test-class", "subject": "Philosophie" }) session_id = session_resp.json()["session_id"] # Session starten und beenden fuer realistische Reflection client.post(f"/api/classroom/sessions/{session_id}/start") client.post(f"/api/classroom/sessions/{session_id}/end") # Reflection erstellen response = client.post("/api/classroom/reflections", json={ "session_id": session_id, "teacher_id": "reflection-test-teacher", "notes": "Die Stunde lief gut, Schueler waren aktiv.", "overall_rating": 4, "what_worked": ["Gruppendiskussion", "Visualisierung"], "improvements": ["Mehr Zeit fuer Einstieg"], "notes_for_next_lesson": "Wiederholung einplanen" }) # DB muss verfuegbar sein if response.status_code == 201: data = response.json() assert data["session_id"] == session_id assert data["notes"] == "Die Stunde lief gut, Schueler waren aktiv." assert data["overall_rating"] == 4 assert "Gruppendiskussion" in data["what_worked"] def test_get_reflection_by_session(self): """Testet Abruf einer Reflection nach Session.""" # Session und Reflection erstellen session_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "get-reflection-teacher", "class_id": "test-class", "subject": "Kunst" }) session_id = session_resp.json()["session_id"] client.post(f"/api/classroom/sessions/{session_id}/start") client.post(f"/api/classroom/sessions/{session_id}/end") create_resp = client.post("/api/classroom/reflections", json={ "session_id": session_id, "teacher_id": "get-reflection-teacher", "notes": "Kreative Stunde" }) if create_resp.status_code == 201: # Reflection abrufen response = client.get(f"/api/classroom/reflections/session/{session_id}") assert response.status_code == 200 assert response.json()["notes"] == "Kreative Stunde" def test_update_reflection(self): """Testet Reflection-Update.""" # Session und Reflection erstellen session_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "update-reflection-teacher", "class_id": "test-class", "subject": "Musik" }) session_id = session_resp.json()["session_id"] client.post(f"/api/classroom/sessions/{session_id}/start") client.post(f"/api/classroom/sessions/{session_id}/end") create_resp = client.post("/api/classroom/reflections", json={ "session_id": session_id, "teacher_id": "update-reflection-teacher", "notes": "Urspruengliche Notizen" }) if create_resp.status_code == 201: reflection_id = create_resp.json()["reflection_id"] # Update response = client.put( f"/api/classroom/reflections/{reflection_id}?teacher_id=update-reflection-teacher", json={ "notes": "Aktualisierte Notizen", "overall_rating": 5 } ) assert response.status_code == 200 assert response.json()["notes"] == "Aktualisierte Notizen" assert response.json()["overall_rating"] == 5 def test_delete_reflection(self): """Testet Reflection-Loeschung.""" # Session und Reflection erstellen session_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "delete-reflection-teacher", "class_id": "test-class", "subject": "Sport" }) session_id = session_resp.json()["session_id"] client.post(f"/api/classroom/sessions/{session_id}/start") client.post(f"/api/classroom/sessions/{session_id}/end") create_resp = client.post("/api/classroom/reflections", json={ "session_id": session_id, "teacher_id": "delete-reflection-teacher", "notes": "Zu loeschende Notizen" }) if create_resp.status_code == 201: reflection_id = create_resp.json()["reflection_id"] # Loeschen response = client.delete( f"/api/classroom/reflections/{reflection_id}?teacher_id=delete-reflection-teacher" ) assert response.status_code == 200 assert response.json()["success"] is True def test_reflection_not_found(self): """Testet 404 bei nicht existierender Reflection.""" response = client.get("/api/classroom/reflections/session/nonexistent-session-id") # 503 wenn DB nicht verfuegbar, 404 wenn nicht gefunden assert response.status_code in [404, 503] def test_duplicate_reflection_rejected(self): """Testet dass doppelte Reflections abgelehnt werden.""" # Session erstellen session_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "duplicate-reflection-teacher", "class_id": "test-class", "subject": "Ethik" }) session_id = session_resp.json()["session_id"] client.post(f"/api/classroom/sessions/{session_id}/start") client.post(f"/api/classroom/sessions/{session_id}/end") # Erste Reflection erstellen first_resp = client.post("/api/classroom/reflections", json={ "session_id": session_id, "teacher_id": "duplicate-reflection-teacher", "notes": "Erste Reflection" }) if first_resp.status_code == 201: # Zweite Reflection sollte abgelehnt werden second_resp = client.post("/api/classroom/reflections", json={ "session_id": session_id, "teacher_id": "duplicate-reflection-teacher", "notes": "Zweite Reflection" }) assert second_resp.status_code == 409 # Conflict # ==================== WEBSOCKET TESTS (Phase 6) ==================== class TestWebSocketStatus: """Tests fuer WebSocket Status Endpoint.""" def test_ws_status_endpoint(self): """Testet den WebSocket Status-Endpoint.""" response = client.get("/api/classroom/ws/status") assert response.status_code == 200 data = response.json() assert "active_sessions" in data assert "sessions" in data assert "broadcast_task_running" in data assert isinstance(data["active_sessions"], int) assert isinstance(data["sessions"], list) class TestWebSocketConnection: """Tests fuer WebSocket-Verbindungen. Hinweis: Vollstaendige WebSocket-Tests erfordern asyncio. Diese Tests pruefen die grundlegende Funktionalitaet. """ def test_ws_invalid_session_rejected(self): """Testet dass WebSocket mit ungültiger Session abgelehnt wird.""" # TestClient unterstuetzt WebSocket-Tests mit context manager with pytest.raises(Exception): # Sollte fehlschlagen da Session nicht existiert with client.websocket_connect("/api/classroom/ws/invalid-session-123"): pass def test_ws_ended_session_rejected(self): """Testet dass WebSocket bei beendeter Session abgelehnt wird.""" # Session erstellen und beenden session_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "ws-test-teacher", "class_id": "test-class", "subject": "WebSocket Test" }) session_id = session_resp.json()["session_id"] client.post(f"/api/classroom/sessions/{session_id}/start") client.post(f"/api/classroom/sessions/{session_id}/end") # WebSocket zu beendeter Session sollte fehlschlagen with pytest.raises(Exception): with client.websocket_connect(f"/api/classroom/ws/{session_id}"): pass def test_ws_connection_to_active_session(self): """Testet WebSocket-Verbindung zu aktiver Session.""" # Session erstellen und starten session_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "ws-active-teacher", "class_id": "test-class", "subject": "WebSocket Active Test" }) session_id = session_resp.json()["session_id"] client.post(f"/api/classroom/sessions/{session_id}/start") # WebSocket verbinden try: with client.websocket_connect(f"/api/classroom/ws/{session_id}") as websocket: # Initiale Nachricht empfangen data = websocket.receive_json() assert data["type"] == "connected" assert "data" in data assert data["data"]["session_id"] == session_id assert "timer" in data["data"] assert "client_count" in data["data"] # Ping senden websocket.send_json({"type": "ping"}) pong = websocket.receive_json() assert pong["type"] == "pong" except Exception as e: # WebSocket kann fehlschlagen wenn Event-Loop nicht verfuegbar pytest.skip(f"WebSocket test skipped: {e}") # Aufraeumen client.post(f"/api/classroom/sessions/{session_id}/end") def test_ws_timer_request(self): """Testet manuellen Timer-Request ueber WebSocket.""" # Session erstellen und starten session_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "ws-timer-teacher", "class_id": "test-class", "subject": "WebSocket Timer Test" }) session_id = session_resp.json()["session_id"] client.post(f"/api/classroom/sessions/{session_id}/start") try: with client.websocket_connect(f"/api/classroom/ws/{session_id}") as websocket: # Initiale Nachricht empfangen initial = websocket.receive_json() assert initial["type"] == "connected" # Timer anfordern websocket.send_json({"type": "get_timer"}) timer_data = websocket.receive_json() assert timer_data["type"] == "timer_update" assert "data" in timer_data assert "remaining_seconds" in timer_data["data"] except Exception as e: pytest.skip(f"WebSocket test skipped: {e}") # Aufraeumen client.post(f"/api/classroom/sessions/{session_id}/end") def test_ws_invalid_json_handled(self): """Testet dass ungültiges JSON korrekt behandelt wird.""" # Session erstellen und starten session_resp = client.post("/api/classroom/sessions", json={ "teacher_id": "ws-json-teacher", "class_id": "test-class", "subject": "WebSocket JSON Test" }) session_id = session_resp.json()["session_id"] client.post(f"/api/classroom/sessions/{session_id}/start") try: with client.websocket_connect(f"/api/classroom/ws/{session_id}") as websocket: # Initiale Nachricht empfangen websocket.receive_json() # Ungültiges JSON senden websocket.send_text("not valid json {{{") error = websocket.receive_json() assert error["type"] == "error" assert "Invalid JSON" in error["data"]["message"] except Exception as e: pytest.skip(f"WebSocket test skipped: {e}") # Aufraeumen client.post(f"/api/classroom/sessions/{session_id}/end")