7658939790
- Set up project structure and dependencies - Create task model and schema - Implement Alembic migrations - Add CRUD API endpoints for task management - Add health endpoint with database connectivity check - Add comprehensive error handling - Add tests for API endpoints - Update README with API documentation
152 lines
3.6 KiB
Python
152 lines
3.6 KiB
Python
from fastapi import status
|
|
|
|
from app.models.task import Task
|
|
|
|
|
|
def test_create_task(client, db):
|
|
"""
|
|
Test creating a new task.
|
|
"""
|
|
task_data = {
|
|
"title": "Test Task",
|
|
"description": "This is a test task",
|
|
"priority": 2
|
|
}
|
|
|
|
response = client.post("/api/v1/tasks/", json=task_data)
|
|
data = response.json()
|
|
|
|
assert response.status_code == status.HTTP_201_CREATED
|
|
assert data["title"] == task_data["title"]
|
|
assert data["description"] == task_data["description"]
|
|
assert data["priority"] == task_data["priority"]
|
|
assert "id" in data
|
|
assert "created_at" in data
|
|
|
|
|
|
def test_read_tasks(client, db):
|
|
"""
|
|
Test getting a list of tasks.
|
|
"""
|
|
# Create test tasks
|
|
task1 = Task(
|
|
title="Task 1",
|
|
description="Description 1",
|
|
priority=1
|
|
)
|
|
task2 = Task(
|
|
title="Task 2",
|
|
description="Description 2",
|
|
priority=2
|
|
)
|
|
db.add(task1)
|
|
db.add(task2)
|
|
db.commit()
|
|
|
|
response = client.get("/api/v1/tasks/")
|
|
data = response.json()
|
|
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert len(data) == 2
|
|
assert data[0]["title"] == "Task 1"
|
|
assert data[1]["title"] == "Task 2"
|
|
|
|
|
|
def test_read_task(client, db):
|
|
"""
|
|
Test getting a specific task.
|
|
"""
|
|
# Create test task
|
|
task = Task(
|
|
title="Test Task",
|
|
description="Test Description",
|
|
priority=2
|
|
)
|
|
db.add(task)
|
|
db.commit()
|
|
|
|
response = client.get(f"/api/v1/tasks/{task.id}")
|
|
data = response.json()
|
|
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert data["title"] == task.title
|
|
assert data["description"] == task.description
|
|
assert data["priority"] == task.priority
|
|
|
|
|
|
def test_update_task(client, db):
|
|
"""
|
|
Test updating a task.
|
|
"""
|
|
# Create test task
|
|
task = Task(
|
|
title="Old Title",
|
|
description="Old Description",
|
|
priority=1
|
|
)
|
|
db.add(task)
|
|
db.commit()
|
|
|
|
update_data = {
|
|
"title": "New Title",
|
|
"description": "New Description",
|
|
"priority": 3
|
|
}
|
|
|
|
response = client.put(f"/api/v1/tasks/{task.id}", json=update_data)
|
|
data = response.json()
|
|
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert data["title"] == update_data["title"]
|
|
assert data["description"] == update_data["description"]
|
|
assert data["priority"] == update_data["priority"]
|
|
|
|
|
|
def test_delete_task(client, db):
|
|
"""
|
|
Test deleting a task.
|
|
"""
|
|
# Create test task
|
|
task = Task(
|
|
title="Task to Delete",
|
|
description="This task will be deleted",
|
|
priority=2
|
|
)
|
|
db.add(task)
|
|
db.commit()
|
|
|
|
response = client.delete(f"/api/v1/tasks/{task.id}")
|
|
|
|
assert response.status_code == status.HTTP_204_NO_CONTENT
|
|
|
|
# Verify task was deleted
|
|
task_in_db = db.query(Task).filter(Task.id == task.id).first()
|
|
assert task_in_db is None
|
|
|
|
|
|
def test_filter_tasks_by_completed(client, db):
|
|
"""
|
|
Test filtering tasks by completed status.
|
|
"""
|
|
# Create test tasks
|
|
task1 = Task(
|
|
title="Task 1",
|
|
description="Description 1",
|
|
completed=True
|
|
)
|
|
task2 = Task(
|
|
title="Task 2",
|
|
description="Description 2",
|
|
completed=False
|
|
)
|
|
db.add(task1)
|
|
db.add(task2)
|
|
db.commit()
|
|
|
|
response = client.get("/api/v1/tasks/?completed=true")
|
|
data = response.json()
|
|
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert len(data) == 1
|
|
assert data[0]["title"] == "Task 1"
|
|
assert data[0]["completed"] is True |