""" Unit Tests for Unit API Tests for contextual learning unit endpoints """ import pytest from unittest.mock import patch, AsyncMock, MagicMock from fastapi.testclient import TestClient from datetime import datetime, timedelta import uuid # Import the app and router import sys sys.path.insert(0, '..') from unit_api import ( router, create_session_token, verify_session_token, CreateSessionRequest, TelemetryPayload, TelemetryEvent, CompleteSessionRequest, PostcheckAnswer, CreateUnitRequest, UpdateUnitRequest, ValidationResult, validate_unit_definition, ) from fastapi import FastAPI # Create test app app = FastAPI() app.include_router(router) client = TestClient(app) class TestSessionTokens: """Test session token functions""" def test_create_session_token(self): """Test session token creation""" session_id = str(uuid.uuid4()) student_id = str(uuid.uuid4()) token = create_session_token(session_id, student_id) assert token is not None assert isinstance(token, str) assert len(token) > 50 # JWT tokens are typically longer def test_verify_session_token_valid(self): """Test valid session token verification""" session_id = str(uuid.uuid4()) student_id = str(uuid.uuid4()) token = create_session_token(session_id, student_id) payload = verify_session_token(token) assert payload is not None assert payload["session_id"] == session_id assert payload["student_id"] == student_id def test_verify_session_token_expired(self): """Test expired session token verification""" session_id = str(uuid.uuid4()) student_id = str(uuid.uuid4()) # Create token with 0 hours expiry (already expired) token = create_session_token(session_id, student_id, expires_hours=0) # Wait a moment for expiration import time time.sleep(0.1) payload = verify_session_token(token) # Token should be invalid/expired # Note: With expires_hours=0, the token expires immediately def test_verify_session_token_invalid(self): """Test invalid session token verification""" payload = verify_session_token("invalid-token") assert payload is None def test_verify_session_token_tampered(self): """Test tampered session token verification""" session_id = str(uuid.uuid4()) student_id = str(uuid.uuid4()) token = create_session_token(session_id, student_id) # Tamper with the token tampered_token = token[:-5] + "xxxxx" payload = verify_session_token(tampered_token) assert payload is None class TestUnitDefinitionsEndpoints: """Test unit definitions endpoints""" def test_list_unit_definitions(self): """Test listing unit definitions""" response = client.get("/api/units/definitions") assert response.status_code == 200 data = response.json() assert isinstance(data, list) # Should have at least the demo unit assert len(data) >= 1 def test_list_unit_definitions_with_template_filter(self): """Test listing with template filter""" response = client.get("/api/units/definitions?template=flight_path") assert response.status_code == 200 data = response.json() assert isinstance(data, list) def test_list_unit_definitions_with_locale_filter(self): """Test listing with locale filter""" response = client.get("/api/units/definitions?locale=de-DE") assert response.status_code == 200 data = response.json() assert isinstance(data, list) def test_get_unit_definition_demo(self): """Test getting demo unit definition""" response = client.get("/api/units/definitions/demo_unit_v1") assert response.status_code == 200 data = response.json() assert data["unit_id"] == "demo_unit_v1" assert data["template"] == "flight_path" assert data["version"] == "1.0.0" assert "definition" in data assert "stops" in data["definition"] def test_get_unit_definition_not_found(self): """Test getting non-existent unit definition""" response = client.get("/api/units/definitions/nonexistent_unit") assert response.status_code == 404 class TestUnitDefinitionsCRUD: """Test CRUD operations for unit definitions""" def test_create_unit_definition(self): """Test creating a new unit definition""" unit_data = { "unit_id": f"test_unit_{uuid.uuid4().hex[:8]}", "template": "flight_path", "version": "1.0.0", "locale": ["de-DE"], "grade_band": ["5", "6"], "duration_minutes": 8, "difficulty": "base", "subject": "Biologie", "topic": "Test Topic", "learning_objectives": ["Lernziel 1", "Lernziel 2"], "stops": [ { "stop_id": "stop_1", "order": 0, "label": {"de-DE": "Stop 1"}, "narration": {"de-DE": "Narration für Stop 1"}, "interaction": { "type": "toggle_switch", "params": {} } }, { "stop_id": "stop_2", "order": 1, "label": {"de-DE": "Stop 2"}, "narration": {"de-DE": "Narration für Stop 2"}, "interaction": { "type": "slider_adjust", "params": {"min": 0, "max": 100, "correct": 50} } } ], "status": "draft" } response = client.post("/api/units/definitions", json=unit_data) assert response.status_code == 200 data = response.json() assert data["unit_id"] == unit_data["unit_id"] assert data["template"] == "flight_path" assert "definition" in data def test_create_unit_definition_minimal(self): """Test creating unit with minimal required fields""" unit_data = { "unit_id": f"minimal_unit_{uuid.uuid4().hex[:8]}", "template": "station_loop", "stops": [ { "stop_id": "stop_1", "order": 0, "label": {"de-DE": "Einziger Stop"}, "narration": {"de-DE": "Narration"}, "interaction": {"type": "toggle_switch", "params": {}} } ] } response = client.post("/api/units/definitions", json=unit_data) assert response.status_code == 200 data = response.json() assert data["unit_id"] == unit_data["unit_id"] def test_create_unit_definition_duplicate_id(self): """Test creating unit with duplicate ID""" # First create a unit unit_id = f"dup_test_{uuid.uuid4().hex[:8]}" unit_data = { "unit_id": unit_id, "template": "flight_path", "stops": [ { "stop_id": "stop_1", "order": 0, "label": {"de-DE": "Stop"}, "narration": {"de-DE": "Narration"}, "interaction": {"type": "toggle_switch", "params": {}} } ] } # Create first time - should succeed response1 = client.post("/api/units/definitions", json=unit_data) assert response1.status_code == 200 # Try to create again with same ID - should fail with conflict response2 = client.post("/api/units/definitions", json=unit_data) # Without database, may return 200 (overwrites file) or 409 assert response2.status_code in [200, 409] def test_update_unit_definition(self): """Test updating an existing unit definition""" # First create a unit unit_id = f"update_test_{uuid.uuid4().hex[:8]}" create_data = { "unit_id": unit_id, "template": "flight_path", "subject": "Original Subject", "stops": [ { "stop_id": "stop_1", "order": 0, "label": {"de-DE": "Stop 1"}, "narration": {"de-DE": "Original"}, "interaction": {"type": "toggle_switch", "params": {}} } ] } client.post("/api/units/definitions", json=create_data) # Update the unit update_data = { "subject": "Updated Subject", "topic": "New Topic", "duration_minutes": 10 } response = client.put(f"/api/units/definitions/{unit_id}", json=update_data) assert response.status_code == 200 data = response.json() # Values should be in the definition object assert data["duration_minutes"] == 10 def test_update_unit_definition_not_found(self): """Test updating non-existent unit""" update_data = {"subject": "New Subject"} response = client.put("/api/units/definitions/nonexistent_unit_xyz", json=update_data) assert response.status_code == 404 def test_delete_unit_definition(self): """Test deleting a unit definition""" # First create a unit unit_id = f"delete_test_{uuid.uuid4().hex[:8]}" create_data = { "unit_id": unit_id, "template": "flight_path", "stops": [ { "stop_id": "stop_1", "order": 0, "label": {"de-DE": "Stop"}, "narration": {"de-DE": "Narration"}, "interaction": {"type": "toggle_switch", "params": {}} } ] } client.post("/api/units/definitions", json=create_data) # Delete the unit response = client.delete(f"/api/units/definitions/{unit_id}") assert response.status_code == 200 data = response.json() assert data["success"] is True # Verify it's deleted get_response = client.get(f"/api/units/definitions/{unit_id}") assert get_response.status_code == 404 def test_delete_unit_definition_not_found(self): """Test deleting non-existent unit""" response = client.delete("/api/units/definitions/nonexistent_unit_xyz") assert response.status_code == 404 class TestUnitValidation: """Test unit validation endpoints and functions""" def test_validate_unit_valid(self): """Test validating a valid unit definition""" unit_data = { "unit_id": "valid_unit", "template": "flight_path", "stops": [ { "stop_id": "stop_1", "order": 0, "label": {"de-DE": "Stop 1"}, "narration": {"de-DE": "Narration"}, "interaction": {"type": "toggle_switch", "params": {}} }, { "stop_id": "stop_2", "order": 1, "label": {"de-DE": "Stop 2"}, "narration": {"de-DE": "Narration"}, "interaction": {"type": "slider_adjust", "params": {}} }, { "stop_id": "stop_3", "order": 2, "label": {"de-DE": "Stop 3"}, "narration": {"de-DE": "Narration"}, "interaction": {"type": "aim_and_pass", "params": {}} } ], "duration_minutes": 8 } response = client.post("/api/units/definitions/validate", json=unit_data) assert response.status_code == 200 data = response.json() assert data["valid"] is True assert len(data["errors"]) == 0 def test_validate_unit_missing_unit_id(self): """Test validation fails without unit_id""" unit_data = { "template": "flight_path", "stops": [ { "stop_id": "stop_1", "order": 0, "label": {"de-DE": "Stop"}, "narration": {"de-DE": "Narration"}, "interaction": {"type": "toggle_switch", "params": {}} } ] } response = client.post("/api/units/definitions/validate", json=unit_data) assert response.status_code == 200 data = response.json() assert data["valid"] is False assert any("unit_id" in err["message"] for err in data["errors"]) def test_validate_unit_missing_template(self): """Test validation fails without template""" unit_data = { "unit_id": "test_unit", "stops": [ { "stop_id": "stop_1", "order": 0, "label": {"de-DE": "Stop"}, "narration": {"de-DE": "Narration"}, "interaction": {"type": "toggle_switch", "params": {}} } ] } response = client.post("/api/units/definitions/validate", json=unit_data) assert response.status_code == 200 data = response.json() assert data["valid"] is False assert any("template" in err["message"] for err in data["errors"]) def test_validate_unit_no_stops(self): """Test validation fails without stops""" unit_data = { "unit_id": "test_unit", "template": "flight_path", "stops": [] } response = client.post("/api/units/definitions/validate", json=unit_data) assert response.status_code == 200 data = response.json() assert data["valid"] is False assert any("Stop" in err["message"] for err in data["errors"]) def test_validate_unit_missing_interaction_type(self): """Test validation fails for stop without interaction type""" unit_data = { "unit_id": "test_unit", "template": "flight_path", "stops": [ { "stop_id": "stop_1", "order": 0, "label": {"de-DE": "Stop"}, "narration": {"de-DE": "Narration"}, "interaction": {"params": {}} # Missing type } ] } response = client.post("/api/units/definitions/validate", json=unit_data) assert response.status_code == 200 data = response.json() assert data["valid"] is False assert any("Interaktionstyp" in err["message"] for err in data["errors"]) def test_validate_unit_flight_path_warning_few_stops(self): """Test validation warns for flight_path with <3 stops""" unit_data = { "unit_id": "test_unit", "template": "flight_path", "stops": [ { "stop_id": "stop_1", "order": 0, "label": {"de-DE": "Stop"}, "narration": {"de-DE": "Narration"}, "interaction": {"type": "toggle_switch", "params": {}} } ] } response = client.post("/api/units/definitions/validate", json=unit_data) assert response.status_code == 200 data = response.json() # Should have warning, not error assert len(data["warnings"]) >= 1 assert any("3" in warn["message"] for warn in data["warnings"]) def test_validate_unit_invalid_duration(self): """Test validation warns for invalid duration""" unit_data = { "unit_id": "test_unit", "template": "flight_path", "duration_minutes": 25, # Too long (max 20) "stops": [ { "stop_id": "stop_1", "order": 0, "label": {"de-DE": "Stop"}, "narration": {"de-DE": "Narration"}, "interaction": {"type": "toggle_switch", "params": {}} } ] } response = client.post("/api/units/definitions/validate", json=unit_data) assert response.status_code == 200 data = response.json() # Duration warnings don't invalidate - it's just a warning assert any("Dauer" in warn["message"] or "duration" in warn["message"].lower() for warn in data["warnings"]) def test_validate_unit_invalid_difficulty(self): """Test validation warns for invalid difficulty""" unit_data = { "unit_id": "test_unit", "template": "flight_path", "difficulty": "extreme", # Invalid "stops": [ { "stop_id": "stop_1", "order": 0, "label": {"de-DE": "Stop"}, "narration": {"de-DE": "Narration"}, "interaction": {"type": "toggle_switch", "params": {}} } ] } response = client.post("/api/units/definitions/validate", json=unit_data) assert response.status_code == 200 data = response.json() # Difficulty warning doesn't invalidate - it's just a warning assert any("difficulty" in warn["message"].lower() for warn in data["warnings"]) def test_validate_function_directly(self): """Test validate_unit_definition function directly""" unit_data = { "unit_id": "direct_test", "template": "station_loop", "stops": [ { "stop_id": "s1", "interaction": {"type": "drag_match", "params": {}} } ] } result = validate_unit_definition(unit_data) assert isinstance(result, ValidationResult) assert result.valid is True assert len(result.errors) == 0 class TestUnitCRUDPydanticModels: """Test Pydantic models for CRUD operations""" def test_create_unit_request_defaults(self): """Test CreateUnitRequest with defaults""" request = CreateUnitRequest( unit_id="test_unit", template="flight_path", stops=[] ) assert request.version == "1.0.0" assert request.locale == ["de-DE"] assert request.grade_band == ["5", "6", "7"] assert request.duration_minutes == 8 assert request.difficulty == "base" assert request.status == "draft" def test_create_unit_request_custom_values(self): """Test CreateUnitRequest with custom values""" request = CreateUnitRequest( unit_id="custom_unit", template="station_loop", version="2.0.0", locale=["en-US", "de-DE"], grade_band=["8", "9"], duration_minutes=15, difficulty="advanced", subject="Physics", topic="Light", learning_objectives=["Understand refraction"], status="published", stops=[] ) assert request.unit_id == "custom_unit" assert request.template == "station_loop" assert request.version == "2.0.0" assert "en-US" in request.locale assert request.difficulty == "advanced" assert request.status == "published" def test_update_unit_request_partial(self): """Test UpdateUnitRequest allows partial updates""" request = UpdateUnitRequest( subject="New Subject" ) assert request.subject == "New Subject" assert request.topic is None assert request.stops is None def test_validation_result_model(self): """Test ValidationResult model""" from unit_api import ValidationError as VE result = ValidationResult( valid=False, errors=[VE(field="unit_id", message="Required")], warnings=[VE(field="stops", message="Consider adding more")] ) assert result.valid is False assert len(result.errors) == 1 assert len(result.warnings) == 1 class TestSessionEndpoints: """Test session endpoints""" def test_create_session(self): """Test creating a new session""" request_data = { "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()), "locale": "de-DE", "difficulty": "base" } response = client.post("/api/units/sessions", json=request_data) assert response.status_code == 200 data = response.json() assert "session_id" in data assert "session_token" in data assert "unit_definition_url" in data assert "telemetry_endpoint" in data assert "expires_at" in data def test_create_session_with_defaults(self): """Test creating session with default values""" request_data = { "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) } response = client.post("/api/units/sessions", json=request_data) assert response.status_code == 200 data = response.json() assert "session_id" in data def test_create_session_invalid_unit(self): """Test creating session with invalid unit""" request_data = { "unit_id": "nonexistent_unit", "student_id": str(uuid.uuid4()) } # Without database, this falls through to demo unit # With database, would return 404 response = client.post("/api/units/sessions", json=request_data) # In dev mode without DB, it still creates a session assert response.status_code in [200, 404] class TestTelemetryEndpoints: """Test telemetry endpoints""" def test_receive_telemetry_without_auth(self): """Test receiving telemetry without authentication (dev mode)""" # First create a session to get valid session_id session_request = { "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) } session_response = client.post("/api/units/sessions", json=session_request) session_data = session_response.json() # Send telemetry telemetry_data = { "session_id": session_data["session_id"], "events": [ { "type": "stop_completed", "stop_id": "stop_1", "metrics": {"success": True, "attempts": 1, "time_sec": 30} } ] } response = client.post("/api/units/telemetry", json=telemetry_data) assert response.status_code == 200 data = response.json() assert data["accepted"] == 1 def test_receive_telemetry_multiple_events(self): """Test receiving multiple telemetry events""" session_request = { "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) } session_response = client.post("/api/units/sessions", json=session_request) session_data = session_response.json() telemetry_data = { "session_id": session_data["session_id"], "events": [ {"type": "stop_completed", "stop_id": "stop_1", "metrics": {"success": True}}, {"type": "hint_used", "stop_id": "stop_1", "metrics": {"hint_id": "hint_1"}}, {"type": "state_change", "metrics": {"from_state": "StopActive", "to_state": "StopTransition"}}, ] } response = client.post("/api/units/telemetry", json=telemetry_data) assert response.status_code == 200 data = response.json() assert data["accepted"] == 3 def test_receive_telemetry_empty_events(self): """Test receiving empty events list""" telemetry_data = { "session_id": str(uuid.uuid4()), "events": [] } response = client.post("/api/units/telemetry", json=telemetry_data) assert response.status_code == 200 data = response.json() assert data["accepted"] == 0 def test_receive_telemetry_with_auth_header(self): """Test receiving telemetry with auth header""" session_request = { "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) } session_response = client.post("/api/units/sessions", json=session_request) session_data = session_response.json() telemetry_data = { "session_id": session_data["session_id"], "events": [ {"type": "stop_completed", "stop_id": "stop_1", "metrics": {}} ] } headers = {"Authorization": f"Bearer {session_data['session_token']}"} response = client.post("/api/units/telemetry", json=telemetry_data, headers=headers) assert response.status_code == 200 class TestCompleteSessionEndpoints: """Test complete session endpoints""" def test_complete_session(self): """Test completing a session""" # Create session first session_request = { "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) } session_response = client.post("/api/units/sessions", json=session_request) session_data = session_response.json() # Complete session complete_data = { "postcheck_answers": [] } headers = {"Authorization": f"Bearer {session_data['session_token']}"} response = client.post( f"/api/units/sessions/{session_data['session_id']}/complete", json=complete_data, headers=headers ) assert response.status_code == 200 data = response.json() assert "summary" in data assert "next_recommendations" in data def test_complete_session_with_postcheck(self): """Test completing session with postcheck answers""" session_request = { "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) } session_response = client.post("/api/units/sessions", json=session_request) session_data = session_response.json() complete_data = { "postcheck_answers": [ {"question_id": "q1", "answer": "B"}, {"question_id": "q2", "answer": "A"}, ] } headers = {"Authorization": f"Bearer {session_data['session_token']}"} response = client.post( f"/api/units/sessions/{session_data['session_id']}/complete", json=complete_data, headers=headers ) assert response.status_code == 200 class TestRecommendationsEndpoints: """Test recommendations endpoints""" def test_get_recommendations(self): """Test getting recommendations for a student""" student_id = str(uuid.uuid4()) response = client.get(f"/api/units/recommendations/{student_id}") assert response.status_code == 200 data = response.json() assert isinstance(data, list) # Should have at least the demo unit as recommendation assert len(data) >= 1 def test_get_recommendations_with_filters(self): """Test getting recommendations with filters""" student_id = str(uuid.uuid4()) response = client.get( f"/api/units/recommendations/{student_id}?grade=5&locale=de-DE&limit=3" ) assert response.status_code == 200 data = response.json() assert isinstance(data, list) assert len(data) <= 3 class TestAnalyticsEndpoints: """Test analytics endpoints""" def test_get_student_analytics(self): """Test getting student analytics""" student_id = str(uuid.uuid4()) response = client.get(f"/api/units/analytics/student/{student_id}") assert response.status_code == 200 data = response.json() assert "student_id" in data assert "units_attempted" in data assert "units_completed" in data def test_get_unit_analytics(self): """Test getting unit analytics""" response = client.get("/api/units/analytics/unit/demo_unit_v1") assert response.status_code == 200 data = response.json() assert "unit_id" in data assert "total_sessions" in data class TestHealthEndpoint: """Test health endpoint""" def test_health_check(self): """Test health check endpoint""" response = client.get("/api/units/health") assert response.status_code == 200 data = response.json() assert data["status"] == "healthy" assert data["service"] == "breakpilot-units" assert "database" in data assert "auth_required" in data class TestPydanticModels: """Test Pydantic model validation""" def test_create_session_request_validation(self): """Test CreateSessionRequest validation""" # Valid request request = CreateSessionRequest( unit_id="test_unit", student_id="student-123", locale="de-DE", difficulty="base" ) assert request.unit_id == "test_unit" assert request.locale == "de-DE" def test_telemetry_event_validation(self): """Test TelemetryEvent validation""" event = TelemetryEvent( type="stop_completed", stop_id="stop_1", metrics={"success": True} ) assert event.type == "stop_completed" assert event.stop_id == "stop_1" def test_telemetry_payload_validation(self): """Test TelemetryPayload validation""" payload = TelemetryPayload( session_id="session-123", events=[ TelemetryEvent(type="stop_completed", stop_id="stop_1") ] ) assert payload.session_id == "session-123" assert len(payload.events) == 1 def test_complete_session_request_validation(self): """Test CompleteSessionRequest validation""" request = CompleteSessionRequest( postcheck_answers=[ PostcheckAnswer(question_id="q1", answer="A") ] ) assert len(request.postcheck_answers) == 1 def test_complete_session_request_empty_answers(self): """Test CompleteSessionRequest with no answers""" request = CompleteSessionRequest(postcheck_answers=None) assert request.postcheck_answers is None class TestEdgeCases: """Test edge cases and error handling""" def test_telemetry_with_missing_metrics(self): """Test telemetry event without metrics""" session_request = { "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) } session_response = client.post("/api/units/sessions", json=session_request) session_data = session_response.json() telemetry_data = { "session_id": session_data["session_id"], "events": [ {"type": "state_change"} # No metrics ] } response = client.post("/api/units/telemetry", json=telemetry_data) assert response.status_code == 200 def test_telemetry_with_timestamp(self): """Test telemetry event with custom timestamp""" session_request = { "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) } session_response = client.post("/api/units/sessions", json=session_request) session_data = session_response.json() telemetry_data = { "session_id": session_data["session_id"], "events": [ { "ts": "2026-01-13T10:00:00Z", "type": "stop_completed", "stop_id": "stop_1" } ] } response = client.post("/api/units/telemetry", json=telemetry_data) assert response.status_code == 200 def test_concurrent_sessions_same_student(self): """Test creating multiple sessions for same student""" student_id = str(uuid.uuid4()) # Create first session session1_request = { "unit_id": "demo_unit_v1", "student_id": student_id } response1 = client.post("/api/units/sessions", json=session1_request) assert response1.status_code == 200 session1_id = response1.json()["session_id"] # Create second session session2_request = { "unit_id": "demo_unit_v1", "student_id": student_id } response2 = client.post("/api/units/sessions", json=session2_request) assert response2.status_code == 200 session2_id = response2.json()["session_id"] # Sessions should be different assert session1_id != session2_id def test_session_token_mismatch(self): """Test telemetry with mismatched session token""" # Create two sessions session1 = client.post("/api/units/sessions", json={ "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) }).json() session2 = client.post("/api/units/sessions", json={ "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) }).json() # Try to use session1's token for session2's telemetry telemetry_data = { "session_id": session2["session_id"], "events": [{"type": "test"}] } headers = {"Authorization": f"Bearer {session1['session_token']}"} response = client.post("/api/units/telemetry", json=telemetry_data, headers=headers) # Should reject due to session_id mismatch assert response.status_code == 403 def test_large_telemetry_batch(self): """Test receiving a large batch of telemetry events""" session = client.post("/api/units/sessions", json={ "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) }).json() # Create 50 events events = [ {"type": f"event_{i}", "stop_id": f"stop_{i % 3}"} for i in range(50) ] telemetry_data = { "session_id": session["session_id"], "events": events } response = client.post("/api/units/telemetry", json=telemetry_data) assert response.status_code == 200 data = response.json() assert data["accepted"] == 50 def test_complete_session_not_found(self): """Test completing non-existent session""" fake_session_id = str(uuid.uuid4()) response = client.post( f"/api/units/sessions/{fake_session_id}/complete", json={} ) # Without database, returns fallback summary assert response.status_code == 200 def test_get_session_not_found(self): """Test getting non-existent session""" fake_session_id = str(uuid.uuid4()) response = client.get(f"/api/units/sessions/{fake_session_id}") assert response.status_code == 404 def test_telemetry_event_with_complex_metrics(self): """Test telemetry with complex nested metrics""" session = client.post("/api/units/sessions", json={ "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) }).json() telemetry_data = { "session_id": session["session_id"], "events": [ { "type": "interaction_attempt", "stop_id": "lens", "metrics": { "success": True, "attempts": 3, "time_sec": 45.5, "interaction_type": "slider_adjust", "details": { "target_value": 50, "actual_value": 48, "tolerance": 5 }, "attempts_log": [ {"value": 30, "time": 10}, {"value": 45, "time": 25}, {"value": 48, "time": 45} ] } } ] } response = client.post("/api/units/telemetry", json=telemetry_data) assert response.status_code == 200 class TestContentGenerationEndpoints: """Test content generation endpoints""" def test_h5p_endpoint_for_nonexistent_unit(self): """Test H5P generation for non-existent unit""" response = client.get("/api/units/content/nonexistent_unit_xyz/h5p") assert response.status_code == 404 def test_worksheet_endpoint_for_nonexistent_unit(self): """Test worksheet generation for non-existent unit""" response = client.get("/api/units/content/nonexistent_unit_xyz/worksheet") assert response.status_code == 404 def test_pdf_endpoint_for_nonexistent_unit(self): """Test PDF download for non-existent unit""" response = client.get("/api/units/content/nonexistent_unit_xyz/worksheet.pdf") assert response.status_code == 404 def test_h5p_endpoint_locale_parameter(self): """Test H5P endpoint accepts locale parameter""" response = client.get("/api/units/content/demo_unit_v1/h5p?locale=en-US") # Without database and content generators, returns 404 assert response.status_code in [200, 404] def test_worksheet_endpoint_locale_parameter(self): """Test worksheet endpoint accepts locale parameter""" response = client.get("/api/units/content/demo_unit_v1/worksheet?locale=en-US") assert response.status_code in [200, 404] class TestSessionWorkflow: """Test complete session workflow""" def test_full_session_workflow(self): """Test complete session lifecycle""" student_id = str(uuid.uuid4()) # 1. Create session session_response = client.post("/api/units/sessions", json={ "unit_id": "demo_unit_v1", "student_id": student_id, "locale": "de-DE", "difficulty": "base" }) assert session_response.status_code == 200 session = session_response.json() session_id = session["session_id"] token = session["session_token"] # 2. Send telemetry events telemetry_events = [ {"type": "state_change", "metrics": {"from_state": "BOOT", "to_state": "PRECHECK"}}, {"type": "precheck_answer", "metrics": {"question_id": "q1", "answer": "B", "correct": True}}, {"type": "state_change", "metrics": {"from_state": "PRECHECK", "to_state": "STOP_ACTIVE"}}, {"type": "stop_completed", "stop_id": "stop_1", "metrics": {"success": True, "time_sec": 20, "attempts": 1}}, {"type": "stop_completed", "stop_id": "stop_2", "metrics": {"success": True, "time_sec": 25, "attempts": 2}}, {"type": "stop_completed", "stop_id": "stop_3", "metrics": {"success": True, "time_sec": 30, "attempts": 1}}, {"type": "state_change", "metrics": {"from_state": "STOP_ACTIVE", "to_state": "POSTCHECK"}}, ] telemetry_response = client.post( "/api/units/telemetry", json={"session_id": session_id, "events": telemetry_events}, headers={"Authorization": f"Bearer {token}"} ) assert telemetry_response.status_code == 200 assert telemetry_response.json()["accepted"] == len(telemetry_events) # 3. Complete session with postcheck complete_response = client.post( f"/api/units/sessions/{session_id}/complete", json={ "postcheck_answers": [ {"question_id": "q1", "answer": "B"}, {"question_id": "q2", "answer": "A"}, {"question_id": "q3", "answer": "C"} ] }, headers={"Authorization": f"Bearer {token}"} ) assert complete_response.status_code == 200 summary = complete_response.json() assert "summary" in summary assert "next_recommendations" in summary def test_session_workflow_with_hints(self): """Test session workflow with hint usage""" session = client.post("/api/units/sessions", json={ "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) }).json() # Send telemetry with hint events events = [ {"type": "hint_used", "stop_id": "stop_1", "metrics": {"hint_id": "hint_1"}}, {"type": "hint_used", "stop_id": "stop_1", "metrics": {"hint_id": "hint_2"}}, {"type": "stop_completed", "stop_id": "stop_1", "metrics": {"success": True, "hints_used": 2}}, ] response = client.post( "/api/units/telemetry", json={"session_id": session["session_id"], "events": events} ) assert response.status_code == 200 def test_session_workflow_with_misconception(self): """Test session workflow with misconception detection""" session = client.post("/api/units/sessions", json={ "unit_id": "demo_unit_v1", "student_id": str(uuid.uuid4()) }).json() events = [ { "type": "misconception_detected", "stop_id": "iris", "metrics": { "concept_id": "pupil_focus", "misconception_type": "wrong_function", "detected_via": "interaction" } }, {"type": "stop_completed", "stop_id": "iris", "metrics": {"success": True}}, ] response = client.post( "/api/units/telemetry", json={"session_id": session["session_id"], "events": events} ) assert response.status_code == 200