19855efacc
Some checks failed
Tests / Go Tests (push) Has been cancelled
Tests / Python Tests (push) Has been cancelled
Tests / Integration Tests (push) Has been cancelled
Tests / Go Lint (push) Has been cancelled
Tests / Python Lint (push) Has been cancelled
Tests / Security Scan (push) Has been cancelled
Tests / All Checks Passed (push) Has been cancelled
Security Scanning / Secret Scanning (push) Has been cancelled
Security Scanning / Dependency Vulnerability Scan (push) Has been cancelled
Security Scanning / Go Security Scan (push) Has been cancelled
Security Scanning / Python Security Scan (push) Has been cancelled
Security Scanning / Node.js Security Scan (push) Has been cancelled
Security Scanning / Docker Image Security (push) Has been cancelled
Security Scanning / Security Summary (push) Has been cancelled
CI/CD Pipeline / Go Tests (push) Has been cancelled
CI/CD Pipeline / Python Tests (push) Has been cancelled
CI/CD Pipeline / Website Tests (push) Has been cancelled
CI/CD Pipeline / Linting (push) Has been cancelled
CI/CD Pipeline / Security Scan (push) Has been cancelled
CI/CD Pipeline / Docker Build & Push (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Deploy to Staging (push) Has been cancelled
CI/CD Pipeline / Deploy to Production (push) Has been cancelled
CI/CD Pipeline / CI Summary (push) Has been cancelled
ci/woodpecker/manual/build-ci-image Pipeline was successful
ci/woodpecker/manual/main Pipeline failed
All services: admin-v2, studio-v2, website, ai-compliance-sdk, consent-service, klausur-service, voice-service, and infrastructure. Large PDFs and compiled binaries excluded via .gitignore.
1204 lines
42 KiB
Python
1204 lines
42 KiB
Python
"""
|
|
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")
|