524 lines
18 KiB
Python
Executable File
524 lines
18 KiB
Python
Executable File
"""Job 领域模型单元测试。"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from packages.domain.job import (
|
|
TERMINAL_STATUSES,
|
|
Job,
|
|
JobStatus,
|
|
JobType,
|
|
)
|
|
|
|
|
|
class TestJobTypeEnum:
|
|
def test_all_types_exist(self):
|
|
assert JobType.VIDEO_COMPOSE.value == "video_compose"
|
|
assert JobType.RENDER_EDIT_PLAN.value == "render_edit_plan"
|
|
assert JobType.ASSET_INGEST.value == "asset_ingest"
|
|
assert JobType.CLASSIFICATION.value == "classification"
|
|
assert JobType.VOICE_EXTRACTION.value == "voice_extraction"
|
|
assert JobType.GENERATION.value == "generation"
|
|
|
|
def test_from_string(self):
|
|
assert JobType("video_compose") == JobType.VIDEO_COMPOSE
|
|
assert JobType("generation") == JobType.GENERATION
|
|
|
|
def test_invalid_type_raises(self):
|
|
with pytest.raises(ValueError):
|
|
JobType("invalid_type")
|
|
|
|
|
|
class TestJobStatusEnum:
|
|
def test_all_statuses_exist(self):
|
|
assert JobStatus.PENDING.value == "pending"
|
|
assert JobStatus.RUNNING.value == "running"
|
|
assert JobStatus.SUCCESS.value == "success"
|
|
assert JobStatus.FAILED.value == "failed"
|
|
assert JobStatus.CANCELLED.value == "cancelled"
|
|
|
|
def test_from_string(self):
|
|
assert JobStatus("pending") == JobStatus.PENDING
|
|
assert JobStatus("success") == JobStatus.SUCCESS
|
|
|
|
|
|
class TestTerminalStatuses:
|
|
def test_success_is_terminal(self):
|
|
assert JobStatus.SUCCESS in TERMINAL_STATUSES
|
|
|
|
def test_failed_is_terminal(self):
|
|
assert JobStatus.FAILED in TERMINAL_STATUSES
|
|
|
|
def test_cancelled_is_terminal(self):
|
|
assert JobStatus.CANCELLED in TERMINAL_STATUSES
|
|
|
|
def test_pending_not_terminal(self):
|
|
assert JobStatus.PENDING not in TERMINAL_STATUSES
|
|
|
|
def test_running_not_terminal(self):
|
|
assert JobStatus.RUNNING not in TERMINAL_STATUSES
|
|
|
|
|
|
class TestJobCreate:
|
|
def test_create_minimal(self):
|
|
job = Job.create(project_id="proj1", job_type=JobType.VIDEO_COMPOSE)
|
|
assert job.id is not None
|
|
assert len(job.id) == 32
|
|
assert job.project_id == "proj1"
|
|
assert job.job_type == JobType.VIDEO_COMPOSE
|
|
assert job.status == JobStatus.PENDING
|
|
assert job.progress == 0.0
|
|
assert job.current_stage == ""
|
|
assert job.payload == {}
|
|
assert job.result == {}
|
|
assert job.error_message == ""
|
|
assert job.retry_count == 0
|
|
assert job.max_retries == 3
|
|
assert job.celery_task_id == ""
|
|
assert job.source_id == ""
|
|
assert job.created_by_user_id == ""
|
|
assert job.started_at is None
|
|
assert job.completed_at is None
|
|
assert isinstance(job.created_at, datetime)
|
|
assert isinstance(job.updated_at, datetime)
|
|
|
|
def test_create_with_enum_type(self):
|
|
job = Job.create("p1", JobType.GENERATION)
|
|
assert job.job_type == JobType.GENERATION
|
|
|
|
def test_create_with_string_type(self):
|
|
job = Job.create("p1", "video_compose")
|
|
assert job.job_type == JobType.VIDEO_COMPOSE
|
|
|
|
def test_create_with_payload(self):
|
|
payload = {"edit_plan_id": "plan123", "resolution": "1080p"}
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, payload=payload)
|
|
assert job.payload == payload
|
|
|
|
def test_create_with_none_payload(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, payload=None)
|
|
assert job.payload == {}
|
|
|
|
def test_create_with_source_id(self):
|
|
job = Job.create("p1", JobType.GENERATION, source_id="gen123")
|
|
assert job.source_id == "gen123"
|
|
|
|
def test_create_with_user_id(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, created_by_user_id="user1")
|
|
assert job.created_by_user_id == "user1"
|
|
|
|
def test_create_with_custom_max_retries(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=5)
|
|
assert job.max_retries == 5
|
|
|
|
def test_create_strips_project_id(self):
|
|
job = Job.create(" proj1 ", JobType.VIDEO_COMPOSE)
|
|
assert job.project_id == "proj1"
|
|
|
|
def test_create_strips_source_id(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, source_id=" src1 ")
|
|
assert job.source_id == "src1"
|
|
|
|
def test_create_strips_user_id(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, created_by_user_id=" u1 ")
|
|
assert job.created_by_user_id == "u1"
|
|
|
|
def test_create_empty_project_id(self):
|
|
with pytest.raises(ValueError, match="project_id 不能为空"):
|
|
Job.create("", JobType.VIDEO_COMPOSE)
|
|
|
|
def test_create_whitespace_project_id(self):
|
|
with pytest.raises(ValueError, match="project_id 不能为空"):
|
|
Job.create(" \t ", JobType.VIDEO_COMPOSE)
|
|
|
|
def test_create_invalid_job_type_string(self):
|
|
with pytest.raises(ValueError, match="不支持的任务类型"):
|
|
Job.create("p1", "invalid_type")
|
|
|
|
def test_create_unique_ids(self):
|
|
j1 = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
j2 = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
assert j1.id != j2.id
|
|
|
|
|
|
class TestIsTerminal:
|
|
def test_pending_not_terminal(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
assert job.is_terminal is False
|
|
|
|
def test_running_not_terminal(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.RUNNING)
|
|
assert job.is_terminal is False
|
|
|
|
def test_success_is_terminal(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.RUNNING)
|
|
job.transition_to(JobStatus.SUCCESS)
|
|
assert job.is_terminal is True
|
|
|
|
def test_failed_is_terminal(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.RUNNING)
|
|
job.transition_to(JobStatus.FAILED)
|
|
assert job.is_terminal is True
|
|
|
|
def test_cancelled_is_terminal(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.CANCELLED)
|
|
assert job.is_terminal is True
|
|
|
|
|
|
class TestIsRetryable:
|
|
def test_pending_not_retryable(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
assert job.is_retryable is False
|
|
|
|
def test_running_not_retryable(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.RUNNING)
|
|
assert job.is_retryable is False
|
|
|
|
def test_success_not_retryable(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.mark_running()
|
|
job.mark_success()
|
|
assert job.is_retryable is False
|
|
|
|
def test_failed_within_limit_is_retryable(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=3)
|
|
job.mark_running()
|
|
job.mark_failed("error")
|
|
assert job.is_retryable is True
|
|
|
|
def test_failed_at_limit_not_retryable(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=3)
|
|
job.mark_running()
|
|
job.mark_failed("error")
|
|
job.retry_count = 3 # 已达到上限
|
|
assert job.is_retryable is False
|
|
|
|
def test_failed_over_limit_not_retryable(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=3)
|
|
job.retry_count = 5
|
|
job.status = JobStatus.FAILED
|
|
assert job.is_retryable is False
|
|
|
|
def test_zero_max_retries_not_retryable(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=0)
|
|
job.status = JobStatus.FAILED
|
|
assert job.is_retryable is False
|
|
|
|
|
|
class TestTransitionTo:
|
|
def test_pending_to_running(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.RUNNING)
|
|
assert job.status == JobStatus.RUNNING
|
|
assert job.started_at is not None
|
|
|
|
def test_pending_to_success(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.SUCCESS)
|
|
assert job.status == JobStatus.SUCCESS
|
|
assert job.completed_at is not None
|
|
|
|
def test_pending_to_cancelled(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.CANCELLED)
|
|
assert job.status == JobStatus.CANCELLED
|
|
|
|
def test_pending_to_failed_invalid(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
job.transition_to(JobStatus.FAILED)
|
|
|
|
def test_running_to_success(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.RUNNING)
|
|
job.transition_to(JobStatus.SUCCESS)
|
|
assert job.status == JobStatus.SUCCESS
|
|
|
|
def test_running_to_failed(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.RUNNING)
|
|
job.transition_to(JobStatus.FAILED)
|
|
assert job.status == JobStatus.FAILED
|
|
|
|
def test_running_to_cancelled(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.RUNNING)
|
|
job.transition_to(JobStatus.CANCELLED)
|
|
assert job.status == JobStatus.CANCELLED
|
|
|
|
def test_running_to_pending_invalid(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.RUNNING)
|
|
with pytest.raises(ValueError, match="非法状态转换"):
|
|
job.transition_to(JobStatus.PENDING)
|
|
|
|
def test_failed_to_pending(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.RUNNING)
|
|
job.transition_to(JobStatus.FAILED)
|
|
# 注意：_VALID_TRANSITIONS 中 FAILED → PENDING 是允许的
|
|
job.transition_to(JobStatus.PENDING)
|
|
assert job.status == JobStatus.PENDING
|
|
|
|
def test_success_to_anything_invalid(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.RUNNING)
|
|
job.transition_to(JobStatus.SUCCESS)
|
|
with pytest.raises(ValueError):
|
|
job.transition_to(JobStatus.FAILED)
|
|
with pytest.raises(ValueError):
|
|
job.transition_to(JobStatus.RUNNING)
|
|
|
|
def test_transition_with_string_status(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to("running")
|
|
assert job.status == JobStatus.RUNNING
|
|
|
|
def test_transition_with_invalid_string(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
with pytest.raises(ValueError, match="无效状态"):
|
|
job.transition_to("invalid_status")
|
|
|
|
def test_transition_updates_updated_at(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
old_time = job.updated_at
|
|
job.transition_to(JobStatus.RUNNING)
|
|
assert job.updated_at >= old_time
|
|
|
|
def test_started_at_only_set_once(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.transition_to(JobStatus.RUNNING)
|
|
first_start = job.started_at
|
|
# 再次 RUNNING 不合法，但我们测试 started_at 在多次 running→success→retry→running 时的行为
|
|
# 先失败重试
|
|
job.transition_to(JobStatus.FAILED)
|
|
job.transition_to(JobStatus.PENDING)
|
|
job.started_at = None # 模拟 prepare_retry 的重置
|
|
job.transition_to(JobStatus.RUNNING)
|
|
assert job.started_at is not None
|
|
assert job.started_at != first_start
|
|
|
|
|
|
class TestMarkRunning:
|
|
def test_mark_running_basic(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.mark_running()
|
|
assert job.status == JobStatus.RUNNING
|
|
assert job.started_at is not None
|
|
|
|
def test_mark_running_with_stage(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.mark_running(stage="下载素材")
|
|
assert job.status == JobStatus.RUNNING
|
|
assert job.current_stage == "下载素材"
|
|
|
|
def test_mark_running_empty_stage_unchanged(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.current_stage = "已有阶段"
|
|
job.mark_running() # 不传 stage
|
|
assert job.current_stage == "已有阶段"
|
|
|
|
|
|
class TestMarkSuccess:
|
|
def test_mark_success_basic(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.mark_running()
|
|
job.mark_success()
|
|
assert job.status == JobStatus.SUCCESS
|
|
assert job.progress == 100.0
|
|
assert job.current_stage == "完成"
|
|
assert job.completed_at is not None
|
|
|
|
def test_mark_success_with_result(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.mark_running()
|
|
result = {"video_url": "https://...", "duration": 30}
|
|
job.mark_success(result=result)
|
|
assert job.result == result
|
|
|
|
def test_mark_success_without_result(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.mark_running()
|
|
original_result = job.result.copy()
|
|
job.mark_success()
|
|
assert job.result == original_result # 不变
|
|
|
|
|
|
class TestMarkFailed:
|
|
def test_mark_failed_basic(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.mark_running()
|
|
job.mark_failed("网络超时")
|
|
assert job.status == JobStatus.FAILED
|
|
assert job.error_message == "网络超时"
|
|
assert job.current_stage == "失败"
|
|
assert job.completed_at is not None
|
|
|
|
def test_mark_failed_empty_message(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.mark_running()
|
|
job.mark_failed("")
|
|
assert job.error_message == ""
|
|
|
|
|
|
class TestMarkCancelled:
|
|
def test_mark_cancelled_from_pending(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.mark_cancelled()
|
|
assert job.status == JobStatus.CANCELLED
|
|
assert job.current_stage == "已取消"
|
|
|
|
def test_mark_cancelled_from_running(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.mark_running()
|
|
job.mark_cancelled()
|
|
assert job.status == JobStatus.CANCELLED
|
|
|
|
|
|
class TestUpdateProgress:
|
|
def test_update_progress_valid(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.update_progress(50.0)
|
|
assert job.progress == 50.0
|
|
|
|
def test_update_progress_zero(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.update_progress(0.0)
|
|
assert job.progress == 0.0
|
|
|
|
def test_update_progress_hundred(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.update_progress(100.0)
|
|
assert job.progress == 100.0
|
|
|
|
def test_update_progress_negative(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
with pytest.raises(ValueError, match="进度必须在 0~100 之间"):
|
|
job.update_progress(-1.0)
|
|
|
|
def test_update_progress_over_100(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
with pytest.raises(ValueError, match="进度必须在 0~100 之间"):
|
|
job.update_progress(101.0)
|
|
|
|
def test_update_progress_with_stage(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.update_progress(30.0, stage="渲染中")
|
|
assert job.progress == 30.0
|
|
assert job.current_stage == "渲染中"
|
|
|
|
def test_update_progress_without_stage_unchanged(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.current_stage = "原阶段"
|
|
job.update_progress(50.0)
|
|
assert job.current_stage == "原阶段"
|
|
|
|
def test_update_progress_updates_timestamp(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
old_time = job.updated_at
|
|
job.update_progress(25.0)
|
|
assert job.updated_at >= old_time
|
|
|
|
|
|
class TestPrepareRetry:
|
|
def test_prepare_retry_success(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=3)
|
|
job.mark_running()
|
|
job.mark_failed("网络错误")
|
|
|
|
job.prepare_retry()
|
|
|
|
assert job.status == JobStatus.PENDING
|
|
assert job.retry_count == 1
|
|
assert job.progress == 0.0
|
|
assert "第 1 次重试" in job.current_stage
|
|
assert job.error_message == ""
|
|
assert job.started_at is None
|
|
assert job.completed_at is None
|
|
assert job.celery_task_id == ""
|
|
|
|
def test_prepare_retry_increments_count(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=5)
|
|
job.mark_running()
|
|
job.mark_failed("err")
|
|
|
|
job.prepare_retry()
|
|
assert job.retry_count == 1
|
|
|
|
# 再次失败重试
|
|
job.mark_running()
|
|
job.mark_failed("err2")
|
|
job.prepare_retry()
|
|
assert job.retry_count == 2
|
|
|
|
def test_prepare_retry_not_retryable_raises(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=0)
|
|
job.mark_running()
|
|
job.mark_failed("err")
|
|
with pytest.raises(ValueError, match="任务不可重试"):
|
|
job.prepare_retry()
|
|
|
|
def test_prepare_retry_wrong_status_raises(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
with pytest.raises(ValueError, match="任务不可重试"):
|
|
job.prepare_retry()
|
|
|
|
|
|
class TestToDict:
|
|
def test_to_dict_structure(self):
|
|
job = Job.create(
|
|
"p1",
|
|
JobType.VIDEO_COMPOSE,
|
|
payload={"key": "val"},
|
|
source_id="src1",
|
|
created_by_user_id="u1",
|
|
)
|
|
d = job.to_dict()
|
|
assert d["id"] == job.id
|
|
assert d["project_id"] == "p1"
|
|
assert d["job_type"] == "video_compose"
|
|
assert d["status"] == "pending"
|
|
assert d["progress"] == 0.0
|
|
assert d["current_stage"] == ""
|
|
assert d["payload"] == {"key": "val"}
|
|
assert d["result"] == {}
|
|
assert d["error_message"] == ""
|
|
assert d["retry_count"] == 0
|
|
assert d["max_retries"] == 3
|
|
assert d["celery_task_id"] == ""
|
|
assert d["source_id"] == "src1"
|
|
assert d["created_by_user_id"] == "u1"
|
|
assert d["is_retryable"] is False
|
|
assert d["started_at"] is None
|
|
assert d["completed_at"] is None
|
|
assert d["created_at"] is not None
|
|
assert d["updated_at"] is not None
|
|
|
|
def test_to_dict_after_success(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE)
|
|
job.mark_running("渲染")
|
|
job.mark_success({"url": "https://..."})
|
|
d = job.to_dict()
|
|
assert d["status"] == "success"
|
|
assert d["progress"] == 100.0
|
|
assert d["is_retryable"] is False
|
|
assert d["started_at"] is not None
|
|
assert d["completed_at"] is not None
|
|
assert isinstance(d["started_at"], str)
|
|
assert isinstance(d["completed_at"], str)
|
|
|
|
def test_to_dict_after_failed(self):
|
|
job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=3)
|
|
job.mark_running()
|
|
job.mark_failed("timeout")
|
|
d = job.to_dict()
|
|
assert d["status"] == "failed"
|
|
assert d["error_message"] == "timeout"
|
|
assert d["is_retryable"] is True
|