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.
95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
"""
|
|
Tests for Session API
|
|
"""
|
|
import pytest
|
|
|
|
|
|
class TestSessionAPI:
|
|
"""Tests for session management."""
|
|
|
|
def test_health_check(self, client):
|
|
"""Test health endpoint returns healthy status."""
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "healthy"
|
|
assert data["service"] == "voice-service"
|
|
assert data["dsgvo_compliance"]["audio_persistence"] is False
|
|
|
|
def test_root_endpoint(self, client):
|
|
"""Test root endpoint returns service info."""
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["service"] == "Breakpilot Voice Service"
|
|
assert "endpoints" in data
|
|
assert data["privacy"]["audio_stored"] is False
|
|
|
|
def test_create_session(self, client):
|
|
"""Test session creation."""
|
|
response = client.post(
|
|
"/api/v1/sessions",
|
|
json={
|
|
"namespace_id": "test-ns-12345678",
|
|
"key_hash": "sha256:eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHg=", # 32 bytes base64
|
|
"device_type": "pwa",
|
|
"client_version": "1.0.0",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "id" in data
|
|
assert data["namespace_id"] == "test-ns-12345678"
|
|
assert data["status"] == "created"
|
|
assert "websocket_url" in data
|
|
|
|
def test_create_session_invalid_key_hash(self, client):
|
|
"""Test session creation with invalid key hash."""
|
|
response = client.post(
|
|
"/api/v1/sessions",
|
|
json={
|
|
"namespace_id": "test-ns-12345678",
|
|
"key_hash": "invalid",
|
|
"device_type": "pwa",
|
|
},
|
|
)
|
|
assert response.status_code == 401
|
|
assert "Invalid encryption key hash" in response.json()["detail"]
|
|
|
|
def test_get_session_not_found(self, client):
|
|
"""Test getting non-existent session."""
|
|
response = client.get("/api/v1/sessions/nonexistent-session")
|
|
assert response.status_code == 404
|
|
|
|
def test_session_lifecycle(self, client):
|
|
"""Test full session lifecycle."""
|
|
# Create session
|
|
create_response = client.post(
|
|
"/api/v1/sessions",
|
|
json={
|
|
"namespace_id": "test-ns-lifecycle",
|
|
"key_hash": "sha256:eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHg=",
|
|
},
|
|
)
|
|
assert create_response.status_code == 200
|
|
session_id = create_response.json()["id"]
|
|
|
|
# Get session
|
|
get_response = client.get(f"/api/v1/sessions/{session_id}")
|
|
assert get_response.status_code == 200
|
|
assert get_response.json()["id"] == session_id
|
|
|
|
# Get session stats
|
|
stats_response = client.get(f"/api/v1/sessions/{session_id}/stats")
|
|
assert stats_response.status_code == 200
|
|
assert "message_count" in stats_response.json()
|
|
|
|
# Delete session
|
|
delete_response = client.delete(f"/api/v1/sessions/{session_id}")
|
|
assert delete_response.status_code == 200
|
|
assert delete_response.json()["status"] == "closed"
|
|
|
|
# Verify session is gone
|
|
get_again = client.get(f"/api/v1/sessions/{session_id}")
|
|
assert get_again.status_code == 404
|