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.
1223 lines
42 KiB
Python
1223 lines
42 KiB
Python
"""
|
|
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
|