185 lines
6.1 KiB
Python
185 lines
6.1 KiB
Python
"""
|
|
Tests for Task API
|
|
"""
|
|
import uuid
|
|
import pytest
|
|
from models.task import TaskState, TaskType
|
|
|
|
|
|
@pytest.fixture
|
|
def session(client):
|
|
"""Create a test session with unique namespace to avoid session limit."""
|
|
unique_ns = f"test-ns-{uuid.uuid4().hex[:16]}"
|
|
response = client.post(
|
|
"/api/v1/sessions",
|
|
json={
|
|
"namespace_id": unique_ns,
|
|
"key_hash": "sha256:eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHg=",
|
|
},
|
|
)
|
|
session_data = response.json()
|
|
yield session_data
|
|
# Cleanup: delete session after test
|
|
if "id" in session_data:
|
|
client.delete(f"/api/v1/sessions/{session_data['id']}")
|
|
|
|
|
|
class TestTaskAPI:
|
|
"""Tests for task management."""
|
|
|
|
def test_create_task(self, client, session):
|
|
"""Test task creation."""
|
|
response = client.post(
|
|
"/api/v1/tasks",
|
|
json={
|
|
"session_id": session["id"],
|
|
"type": "student_observation",
|
|
"intent_text": "Notiz zu Max: heute wiederholt gestoert",
|
|
"parameters": {
|
|
"student_name": "Max",
|
|
"observation": "wiederholt gestoert",
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "id" in data
|
|
assert data["session_id"] == session["id"]
|
|
assert data["type"] == "student_observation"
|
|
# Task should be queued automatically for simple note types
|
|
assert data["state"] in ["draft", "queued", "ready"]
|
|
|
|
def test_create_task_invalid_session(self, client):
|
|
"""Test task creation with invalid session."""
|
|
response = client.post(
|
|
"/api/v1/tasks",
|
|
json={
|
|
"session_id": "nonexistent-session",
|
|
"type": "student_observation",
|
|
"intent_text": "Test",
|
|
},
|
|
)
|
|
assert response.status_code == 404
|
|
assert "Session not found" in response.json()["detail"]
|
|
|
|
def test_get_task(self, client, session):
|
|
"""Test getting task by ID."""
|
|
# Create task first
|
|
create_response = client.post(
|
|
"/api/v1/tasks",
|
|
json={
|
|
"session_id": session["id"],
|
|
"type": "reminder",
|
|
"intent_text": "Erinner mich morgen an Hausaufgaben",
|
|
},
|
|
)
|
|
task_id = create_response.json()["id"]
|
|
|
|
# Get task
|
|
response = client.get(f"/api/v1/tasks/{task_id}")
|
|
assert response.status_code == 200
|
|
assert response.json()["id"] == task_id
|
|
|
|
def test_get_task_not_found(self, client):
|
|
"""Test getting non-existent task."""
|
|
response = client.get("/api/v1/tasks/nonexistent-task")
|
|
assert response.status_code == 404
|
|
|
|
def test_task_transition_approve(self, client, session):
|
|
"""Test approving a task."""
|
|
# Create task
|
|
create_response = client.post(
|
|
"/api/v1/tasks",
|
|
json={
|
|
"session_id": session["id"],
|
|
"type": "student_observation",
|
|
"intent_text": "Notiz",
|
|
},
|
|
)
|
|
task_id = create_response.json()["id"]
|
|
|
|
# Get current state
|
|
task = client.get(f"/api/v1/tasks/{task_id}").json()
|
|
|
|
# Transition to approved if task is in ready state
|
|
if task["state"] == "ready":
|
|
response = client.put(
|
|
f"/api/v1/tasks/{task_id}/transition",
|
|
json={
|
|
"new_state": "approved",
|
|
"reason": "user_approved",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["state"] in ["approved", "completed"]
|
|
|
|
def test_task_transition_invalid(self, client, session):
|
|
"""Test invalid task transition."""
|
|
# Create task
|
|
create_response = client.post(
|
|
"/api/v1/tasks",
|
|
json={
|
|
"session_id": session["id"],
|
|
"type": "reminder",
|
|
"intent_text": "Test",
|
|
},
|
|
)
|
|
task_id = create_response.json()["id"]
|
|
|
|
# Try invalid transition (draft -> completed is not allowed)
|
|
response = client.put(
|
|
f"/api/v1/tasks/{task_id}/transition",
|
|
json={
|
|
"new_state": "completed",
|
|
"reason": "invalid",
|
|
},
|
|
)
|
|
# Should fail with 400 if state doesn't allow direct transition to completed
|
|
# or succeed if state machine allows it
|
|
assert response.status_code in [200, 400]
|
|
|
|
def test_delete_task(self, client, session):
|
|
"""Test deleting a task."""
|
|
# Create task
|
|
create_response = client.post(
|
|
"/api/v1/tasks",
|
|
json={
|
|
"session_id": session["id"],
|
|
"type": "student_observation",
|
|
"intent_text": "To delete",
|
|
},
|
|
)
|
|
task_id = create_response.json()["id"]
|
|
|
|
# Get task to check state
|
|
task = client.get(f"/api/v1/tasks/{task_id}").json()
|
|
|
|
# If task is in a deletable state, delete it
|
|
if task["state"] in ["draft", "completed", "expired", "rejected"]:
|
|
response = client.delete(f"/api/v1/tasks/{task_id}")
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "deleted"
|
|
|
|
# Verify task is gone
|
|
get_response = client.get(f"/api/v1/tasks/{task_id}")
|
|
assert get_response.status_code == 404
|
|
|
|
def test_session_tasks(self, client, session):
|
|
"""Test getting tasks for a session."""
|
|
# Create multiple tasks
|
|
for i in range(3):
|
|
client.post(
|
|
"/api/v1/tasks",
|
|
json={
|
|
"session_id": session["id"],
|
|
"type": "reminder",
|
|
"intent_text": f"Task {i}",
|
|
},
|
|
)
|
|
|
|
# Get session tasks
|
|
response = client.get(f"/api/v1/sessions/{session['id']}/tasks")
|
|
assert response.status_code == 200
|
|
tasks = response.json()
|
|
assert len(tasks) >= 3
|