3f24a7ea3e
- Implemented API and web task access assertions in the task status polling endpoint. - Added functions to remember and check task access in user sessions. - Updated task status tests to validate access control based on session data. - Enhanced download route tests to ensure proper access checks. - Improved SEO metadata handling with dynamic social preview images. - Updated sitemap generation to include blog posts and new tools. - Added a social preview SVG for better sharing on social media platforms.
86 lines
3.3 KiB
Python
86 lines
3.3 KiB
Python
"""Tests for task status polling route."""
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from app.utils.auth import TASK_ACCESS_SESSION_KEY
|
|
|
|
|
|
class TestTaskStatus:
|
|
def test_pending_task(self, client, monkeypatch):
|
|
"""Should return PENDING state for a queued task."""
|
|
mock_result = MagicMock()
|
|
mock_result.state = 'PENDING'
|
|
mock_result.info = None
|
|
|
|
with client.session_transaction() as session:
|
|
session[TASK_ACCESS_SESSION_KEY] = ['test-task-id']
|
|
|
|
with patch('app.routes.tasks.AsyncResult', return_value=mock_result):
|
|
response = client.get('/api/tasks/test-task-id/status')
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['task_id'] == 'test-task-id'
|
|
assert data['state'] == 'PENDING'
|
|
assert 'progress' in data
|
|
|
|
def test_processing_task(self, client, monkeypatch):
|
|
"""Should return PROCESSING state with step info."""
|
|
mock_result = MagicMock()
|
|
mock_result.state = 'PROCESSING'
|
|
mock_result.info = {'step': 'Converting page 3 of 10...'}
|
|
|
|
with client.session_transaction() as session:
|
|
session[TASK_ACCESS_SESSION_KEY] = ['processing-id']
|
|
|
|
with patch('app.routes.tasks.AsyncResult', return_value=mock_result):
|
|
response = client.get('/api/tasks/processing-id/status')
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['state'] == 'PROCESSING'
|
|
assert data['progress'] == 'Converting page 3 of 10...'
|
|
|
|
def test_success_task(self, client, monkeypatch):
|
|
"""Should return SUCCESS state with result data."""
|
|
mock_result = MagicMock()
|
|
mock_result.state = 'SUCCESS'
|
|
mock_result.result = {
|
|
'status': 'completed',
|
|
'download_url': '/api/download/task-id/output.pdf',
|
|
'filename': 'output.pdf',
|
|
}
|
|
|
|
with client.session_transaction() as session:
|
|
session[TASK_ACCESS_SESSION_KEY] = ['success-id']
|
|
|
|
with patch('app.routes.tasks.AsyncResult', return_value=mock_result):
|
|
response = client.get('/api/tasks/success-id/status')
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['state'] == 'SUCCESS'
|
|
assert data['result']['status'] == 'completed'
|
|
assert 'download_url' in data['result']
|
|
|
|
def test_failure_task(self, client, monkeypatch):
|
|
"""Should return FAILURE state with error message."""
|
|
mock_result = MagicMock()
|
|
mock_result.state = 'FAILURE'
|
|
mock_result.info = Exception('Conversion failed due to corrupt PDF.')
|
|
|
|
with client.session_transaction() as session:
|
|
session[TASK_ACCESS_SESSION_KEY] = ['failed-id']
|
|
|
|
with patch('app.routes.tasks.AsyncResult', return_value=mock_result):
|
|
response = client.get('/api/tasks/failed-id/status')
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['state'] == 'FAILURE'
|
|
assert 'error' in data
|
|
|
|
def test_unknown_task_without_access_returns_404(self, client):
|
|
"""Should not expose task state without session or API ownership."""
|
|
response = client.get('/api/tasks/unknown-task/status')
|
|
|
|
assert response.status_code == 404 |