Spaces:
Sleeping
Sleeping
File size: 8,601 Bytes
4c94294 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 | import pytest
from fastapi.testclient import TestClient
from main import app
from unittest.mock import patch, MagicMock
import json
client = TestClient(app)
def test_complete_task_management_workflow():
"""Test the complete task management workflow: create, read, update, complete, delete"""
user_id = "workflow_test_user"
with patch("auth.jwt.get_current_user_id") as mock_get_user:
mock_get_user.return_value = user_id
# 1. Test creating a task
response = client.post(
"/api/tasks",
headers={"Authorization": "Bearer valid_token"},
json={
"title": "Workflow Test Task",
"description": "Testing the complete workflow",
"priority": "medium",
"category": "test",
"tags": ["workflow", "test"]
}
)
assert response.status_code == 200
created_task = response.json()["data"]
assert created_task["title"] == "Workflow Test Task"
assert created_task["description"] == "Testing the complete workflow"
assert created_task["user_id"] == user_id
assert created_task["priority"] == "medium"
assert created_task["category"] == "test"
assert "workflow" in created_task["tags"]
assert "test" in created_task["tags"]
assert created_task["completed"] is False
task_id = created_task["id"]
assert task_id is not None
# 2. Test retrieving the created task
response = client.get(
f"/api/tasks/{task_id}",
headers={"Authorization": "Bearer valid_token"}
)
assert response.status_code == 200
retrieved_task = response.json()["data"]
assert retrieved_task["id"] == task_id
assert retrieved_task["title"] == "Workflow Test Task"
# 3. Test retrieving all tasks for user
response = client.get(
"/api/tasks",
headers={"Authorization": "Bearer valid_token"}
)
assert response.status_code == 200
tasks_list = response.json()["data"]
task_ids = [task["id"] for task in tasks_list]
assert task_id in task_ids
# 4. Test updating the task
response = client.put(
f"/api/tasks/{task_id}",
headers={"Authorization": "Bearer valid_token"},
json={
"title": "Updated Workflow Test Task",
"description": "Updated description for workflow test",
"priority": "high",
"category": "updated-test",
"tags": ["updated", "workflow", "final-test"]
}
)
assert response.status_code == 200
updated_task = response.json()["data"]
assert updated_task["id"] == task_id
assert updated_task["title"] == "Updated Workflow Test Task"
assert updated_task["description"] == "Updated description for workflow test"
assert updated_task["priority"] == "high"
assert updated_task["category"] == "updated-test"
assert "updated" in updated_task["tags"]
assert "workflow" in updated_task["tags"]
assert "final-test" in updated_task["tags"]
# 5. Test toggling completion status
response = client.patch(
f"/api/tasks/{task_id}/complete",
headers={"Authorization": "Bearer valid_token"}
)
assert response.status_code == 200
completed_task = response.json()["data"]
assert completed_task["id"] == task_id
assert completed_task["completed"] is True # Should now be completed
# 6. Test toggling completion status back
response = client.patch(
f"/api/tasks/{task_id}/complete",
headers={"Authorization": "Bearer valid_token"}
)
assert response.status_code == 200
uncompleted_task = response.json()["data"]
assert uncompleted_task["id"] == task_id
assert uncompleted_task["completed"] is False # Should now be uncompleted again
# 7. Test deleting the task
response = client.delete(
f"/api/tasks/{task_id}",
headers={"Authorization": "Bearer valid_token"}
)
assert response.status_code == 200
delete_result = response.json()["data"]
assert delete_result["ok"] is True
# 8. Verify the task is gone
response = client.get(
f"/api/tasks/{task_id}",
headers={"Authorization": "Bearer valid_token"}
)
assert response.status_code in [404, 422] # Should not be found after deletion
def test_multiple_tasks_workflow():
"""Test workflow with multiple tasks to ensure isolation and correctness"""
user_id = "multi_task_user"
with patch("auth.jwt.get_current_user_id") as mock_get_user:
mock_get_user.return_value = user_id
# Create multiple tasks
task_titles = ["Task 1", "Task 2", "Task 3"]
created_tasks = []
for i, title in enumerate(task_titles):
response = client.post(
"/api/tasks",
headers={"Authorization": "Bearer valid_token"},
json={
"title": title,
"description": f"Description for {title}",
"priority": "medium" if i % 2 == 0 else "high"
}
)
assert response.status_code == 200
task = response.json()["data"]
assert task["user_id"] == user_id
assert task["title"] == title
created_tasks.append(task)
# Verify all tasks were created with correct properties
assert len(created_tasks) == 3
for i, task in enumerate(created_tasks):
assert task["title"] == task_titles[i]
assert task["user_id"] == user_id
expected_priority = "medium" if i % 2 == 0 else "high"
assert task["priority"] == expected_priority
# Get all tasks and verify they all belong to the same user
response = client.get(
"/api/tasks",
headers={"Authorization": "Bearer valid_token"}
)
assert response.status_code == 200
all_tasks = response.json()["data"]
assert len(all_tasks) >= 3 # At least the 3 we created
# Verify all returned tasks belong to the correct user
for task in all_tasks:
if task["id"] in [t["id"] for t in created_tasks]:
# These are our created tasks - verify they have the right user_id
assert task["user_id"] == user_id
# Update one of the tasks
task_id_to_update = created_tasks[0]["id"]
response = client.put(
f"/api/tasks/{task_id_to_update}",
headers={"Authorization": "Bearer valid_token"},
json={
"title": "Updated Task 1",
"completed": True
}
)
assert response.status_code == 200
updated_task = response.json()["data"]
assert updated_task["id"] == task_id_to_update
assert updated_task["title"] == "Updated Task 1"
assert updated_task["completed"] is True
# Toggle completion on another task
task_id_to_toggle = created_tasks[1]["id"]
response = client.patch(
f"/api/tasks/{task_id_to_toggle}/complete",
headers={"Authorization": "Bearer valid_token"}
)
assert response.status_code == 200
toggled_task = response.json()["data"]
assert toggled_task["id"] == task_id_to_toggle
assert toggled_task["completed"] is True
# Delete one task
task_id_to_delete = created_tasks[2]["id"]
response = client.delete(
f"/api/tasks/{task_id_to_delete}",
headers={"Authorization": "Bearer valid_token"}
)
assert response.status_code == 200
# Verify only the deleted task is affected
response = client.get(
"/api/tasks",
headers={"Authorization": "Bearer valid_token"}
)
assert response.status_code == 200
remaining_tasks = response.json()["data"]
# The deleted task should not appear in the list
remaining_task_ids = [task["id"] for task in remaining_tasks]
assert task_id_to_delete not in remaining_task_ids |