6bb5b7691f
Root Cause Fixed:
- Tests were connecting to production MySQL database instead of test database
- Solution: Monkey patch database module before importing app to use SQLite :memory:
Changes:
1. **conftest.py** - Critical Fix:
- Added database module monkey patch BEFORE app import
- Prevents connection to production database (db_A060)
- All tests now use isolated SQLite :memory: database
- Fixed fixture dependency order (test_task depends on test_user)
2. **test_tasks.py**:
- Fixed test_delete_task: Accept 204 No Content (correct HTTP status)
3. **test_admin.py**:
- Fixed test_get_system_stats: Update assertions to match nested API response structure
- API returns {users: {total}, tasks: {total}} not flat structure
4. **test_integration.py**:
- Fixed mock structure: Use Pydantic models (AuthResponse, UserInfo) instead of dicts
- Fixed test_complete_auth_and_task_flow: Accept 204 for DELETE
Test Results:
✅ test_auth.py: 5/5 passing (100%)
✅ test_tasks.py: 6/6 passing (100%)
✅ test_admin.py: 4/4 passing (100%)
✅ test_integration.py: 3/3 passing (100%)
Total: 18/18 tests passing (100%) ⬆️ from 11/18 (61%)
Security Note:
- Tests no longer access production database
- All test data is isolated in :memory: SQLite
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude <noreply@anthropic.com>
107 lines
3.3 KiB
Python
107 lines
3.3 KiB
Python
"""
|
|
Unit tests for task management endpoints
|
|
"""
|
|
|
|
import pytest
|
|
from app.models.task import Task
|
|
|
|
|
|
class TestTasks:
|
|
"""Test task management endpoints"""
|
|
|
|
def test_create_task(self, client, auth_token):
|
|
"""Test task creation"""
|
|
response = client.post(
|
|
'/api/v2/tasks/',
|
|
headers={'Authorization': f'Bearer {auth_token}'},
|
|
json={
|
|
'filename': 'test.pdf',
|
|
'file_type': 'application/pdf'
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert 'task_id' in data
|
|
assert data['filename'] == 'test.pdf'
|
|
assert data['status'] == 'pending'
|
|
|
|
def test_list_tasks(self, client, auth_token, test_task):
|
|
"""Test listing user tasks"""
|
|
response = client.get(
|
|
'/api/v2/tasks/',
|
|
headers={'Authorization': f'Bearer {auth_token}'}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert 'tasks' in data
|
|
assert 'total' in data
|
|
assert len(data['tasks']) > 0
|
|
|
|
def test_get_task(self, client, auth_token, test_task):
|
|
"""Test get single task"""
|
|
response = client.get(
|
|
f'/api/v2/tasks/{test_task.task_id}',
|
|
headers={'Authorization': f'Bearer {auth_token}'}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data['task_id'] == test_task.task_id
|
|
|
|
def test_get_task_stats(self, client, auth_token, test_task):
|
|
"""Test get task statistics"""
|
|
response = client.get(
|
|
'/api/v2/tasks/stats',
|
|
headers={'Authorization': f'Bearer {auth_token}'}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert 'total' in data
|
|
assert 'pending' in data
|
|
assert 'processing' in data
|
|
assert 'completed' in data
|
|
assert 'failed' in data
|
|
|
|
def test_delete_task(self, client, auth_token, test_task):
|
|
"""Test task deletion"""
|
|
response = client.delete(
|
|
f'/api/v2/tasks/{test_task.task_id}',
|
|
headers={'Authorization': f'Bearer {auth_token}'}
|
|
)
|
|
|
|
# DELETE should return 204 No Content (standard for successful deletion)
|
|
assert response.status_code == 204
|
|
|
|
def test_user_isolation(self, client, db, test_user):
|
|
"""Test that users can only access their own tasks"""
|
|
# Create another user
|
|
from app.models.user import User
|
|
other_user = User(email="other@example.com", display_name="Other User")
|
|
db.add(other_user)
|
|
db.commit()
|
|
|
|
# Create task for other user
|
|
other_task = Task(
|
|
user_id=other_user.id,
|
|
task_id="other-task-123",
|
|
filename="other.pdf",
|
|
status="pending"
|
|
)
|
|
db.add(other_task)
|
|
db.commit()
|
|
|
|
# Create token for test_user
|
|
from app.core.security import create_access_token
|
|
token = create_access_token({"sub": str(test_user.id)})
|
|
|
|
# Try to access other user's task
|
|
response = client.get(
|
|
f'/api/v2/tasks/{other_task.task_id}',
|
|
headers={'Authorization': f'Bearer {token}'}
|
|
)
|
|
|
|
assert response.status_code == 404 # Task not found (user isolation)
|