237 lines
9.0 KiB
Python
237 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
任務管理 API 測試
|
|
|
|
Author: PANJIT IT Team
|
|
Created: 2024-01-28
|
|
Modified: 2024-01-28
|
|
"""
|
|
|
|
import pytest
|
|
from app.models.job import TranslationJob
|
|
|
|
|
|
class TestJobsAPI:
|
|
"""任務管理 API 測試類別"""
|
|
|
|
def test_get_user_jobs_success(self, authenticated_client, sample_job):
|
|
"""測試取得使用者任務列表"""
|
|
response = authenticated_client.get('/api/v1/jobs')
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['success'] is True
|
|
assert 'jobs' in data['data']
|
|
assert 'pagination' in data['data']
|
|
assert len(data['data']['jobs']) > 0
|
|
|
|
def test_get_user_jobs_with_status_filter(self, authenticated_client, sample_job):
|
|
"""測試按狀態篩選任務"""
|
|
response = authenticated_client.get('/api/v1/jobs?status=PENDING')
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['success'] is True
|
|
|
|
# 所有返回的任務都應該是 PENDING 狀態
|
|
for job in data['data']['jobs']:
|
|
assert job['status'] == 'PENDING'
|
|
|
|
def test_get_user_jobs_with_pagination(self, authenticated_client, sample_job):
|
|
"""測試分頁"""
|
|
response = authenticated_client.get('/api/v1/jobs?page=1&per_page=5')
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['success'] is True
|
|
assert data['data']['pagination']['page'] == 1
|
|
assert data['data']['pagination']['per_page'] == 5
|
|
|
|
def test_get_user_jobs_without_auth(self, client):
|
|
"""測試未認證取得任務列表"""
|
|
response = client.get('/api/v1/jobs')
|
|
|
|
assert response.status_code == 401
|
|
data = response.get_json()
|
|
assert data['success'] is False
|
|
assert data['error'] == 'AUTHENTICATION_REQUIRED'
|
|
|
|
def test_get_job_detail_success(self, authenticated_client, sample_job):
|
|
"""測試取得任務詳細資訊"""
|
|
response = authenticated_client.get(f'/api/v1/jobs/{sample_job.job_uuid}')
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['success'] is True
|
|
assert 'job' in data['data']
|
|
assert data['data']['job']['job_uuid'] == sample_job.job_uuid
|
|
|
|
def test_get_job_detail_not_found(self, authenticated_client):
|
|
"""測試取得不存在的任務"""
|
|
fake_uuid = '00000000-0000-0000-0000-000000000000'
|
|
response = authenticated_client.get(f'/api/v1/jobs/{fake_uuid}')
|
|
|
|
assert response.status_code == 404
|
|
data = response.get_json()
|
|
assert data['success'] is False
|
|
assert data['error'] == 'JOB_NOT_FOUND'
|
|
|
|
def test_get_job_detail_invalid_uuid(self, authenticated_client):
|
|
"""測試無效的UUID格式"""
|
|
invalid_uuid = 'invalid-uuid'
|
|
response = authenticated_client.get(f'/api/v1/jobs/{invalid_uuid}')
|
|
|
|
assert response.status_code == 400
|
|
data = response.get_json()
|
|
assert data['success'] is False
|
|
assert data['error'] == 'INVALID_UUID'
|
|
|
|
def test_get_job_detail_permission_denied(self, authenticated_client, app):
|
|
"""測試存取他人任務"""
|
|
from app.models.user import User
|
|
from app import db
|
|
|
|
with app.app_context():
|
|
# 建立另一個使用者和任務
|
|
other_user = User(
|
|
username='otheruser',
|
|
display_name='Other User',
|
|
email='other@panjit.com.tw',
|
|
department='IT',
|
|
is_admin=False
|
|
)
|
|
db.session.add(other_user)
|
|
db.session.commit()
|
|
|
|
other_job = TranslationJob(
|
|
user_id=other_user.id,
|
|
original_filename='other.docx',
|
|
file_extension='.docx',
|
|
file_size=1024,
|
|
file_path='/tmp/other.docx',
|
|
source_language='auto',
|
|
target_languages=['en'],
|
|
status='PENDING'
|
|
)
|
|
db.session.add(other_job)
|
|
db.session.commit()
|
|
|
|
response = authenticated_client.get(f'/api/v1/jobs/{other_job.job_uuid}')
|
|
|
|
assert response.status_code == 403
|
|
data = response.get_json()
|
|
assert data['success'] is False
|
|
assert data['error'] == 'PERMISSION_DENIED'
|
|
|
|
def test_retry_job_success(self, authenticated_client, sample_job):
|
|
"""測試重試失敗任務"""
|
|
# 設定任務為失敗狀態
|
|
sample_job.update_status('FAILED', error_message='Test error')
|
|
|
|
response = authenticated_client.post(f'/api/v1/jobs/{sample_job.job_uuid}/retry')
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['success'] is True
|
|
assert data['data']['status'] == 'PENDING'
|
|
assert data['data']['retry_count'] == 1
|
|
|
|
def test_retry_job_cannot_retry(self, authenticated_client, sample_job):
|
|
"""測試無法重試的任務"""
|
|
# 設定任務為完成狀態
|
|
sample_job.update_status('COMPLETED')
|
|
|
|
response = authenticated_client.post(f'/api/v1/jobs/{sample_job.job_uuid}/retry')
|
|
|
|
assert response.status_code == 400
|
|
data = response.get_json()
|
|
assert data['success'] is False
|
|
assert data['error'] == 'CANNOT_RETRY'
|
|
|
|
def test_retry_job_max_retries(self, authenticated_client, sample_job):
|
|
"""測試達到最大重試次數"""
|
|
# 設定任務為失敗且重試次數已達上限
|
|
sample_job.update_status('FAILED', error_message='Test error')
|
|
sample_job.retry_count = 3
|
|
|
|
from app import db
|
|
db.session.commit()
|
|
|
|
response = authenticated_client.post(f'/api/v1/jobs/{sample_job.job_uuid}/retry')
|
|
|
|
assert response.status_code == 400
|
|
data = response.get_json()
|
|
assert data['success'] is False
|
|
assert data['error'] == 'CANNOT_RETRY'
|
|
|
|
def test_get_user_statistics(self, authenticated_client, sample_job):
|
|
"""測試取得使用者統計資料"""
|
|
response = authenticated_client.get('/api/v1/jobs/statistics')
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['success'] is True
|
|
assert 'job_statistics' in data['data']
|
|
assert 'api_statistics' in data['data']
|
|
|
|
def test_get_user_statistics_with_date_range(self, authenticated_client):
|
|
"""測試指定日期範圍的統計"""
|
|
response = authenticated_client.get('/api/v1/jobs/statistics?start_date=2024-01-01T00:00:00Z&end_date=2024-12-31T23:59:59Z')
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['success'] is True
|
|
|
|
def test_get_user_statistics_invalid_date(self, authenticated_client):
|
|
"""測試無效的日期格式"""
|
|
response = authenticated_client.get('/api/v1/jobs/statistics?start_date=invalid-date')
|
|
|
|
assert response.status_code == 400
|
|
data = response.get_json()
|
|
assert data['success'] is False
|
|
assert data['error'] == 'INVALID_START_DATE'
|
|
|
|
def test_get_queue_status(self, client, sample_job):
|
|
"""測試取得佇列狀態（不需認證）"""
|
|
response = client.get('/api/v1/jobs/queue/status')
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['success'] is True
|
|
assert 'queue_status' in data['data']
|
|
assert 'processing_jobs' in data['data']
|
|
|
|
def test_cancel_job_success(self, authenticated_client, sample_job):
|
|
"""測試取消等待中的任務"""
|
|
# 確保任務是 PENDING 狀態
|
|
assert sample_job.status == 'PENDING'
|
|
|
|
response = authenticated_client.post(f'/api/v1/jobs/{sample_job.job_uuid}/cancel')
|
|
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['success'] is True
|
|
assert data['data']['status'] == 'FAILED'
|
|
|
|
def test_cancel_job_cannot_cancel(self, authenticated_client, sample_job):
|
|
"""測試取消非等待狀態的任務"""
|
|
# 設定任務為處理中
|
|
sample_job.update_status('PROCESSING')
|
|
|
|
response = authenticated_client.post(f'/api/v1/jobs/{sample_job.job_uuid}/cancel')
|
|
|
|
assert response.status_code == 400
|
|
data = response.get_json()
|
|
assert data['success'] is False
|
|
assert data['error'] == 'CANNOT_CANCEL'
|
|
|
|
def test_cancel_job_not_found(self, authenticated_client):
|
|
"""測試取消不存在的任務"""
|
|
fake_uuid = '00000000-0000-0000-0000-000000000000'
|
|
response = authenticated_client.post(f'/api/v1/jobs/{fake_uuid}/cancel')
|
|
|
|
assert response.status_code == 404
|
|
data = response.get_json()
|
|
assert data['success'] is False
|
|
assert data['error'] == 'JOB_NOT_FOUND' |