From 6f4bf2e0204cdfcdcbb0a2d01e8f4cdaa9d506f7 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sun, 26 Jul 2026 17:34:58 +0800 Subject: [PATCH] test(wave96): 76 unit tests for Job domain model (state machine + lifecycle) - JobType enum: 3 tests - JobStatus enum: 2 tests - TERMINAL_STATUSES: 5 tests - Job.create: 17 tests (validation + string enum + fields) - is_terminal / is_retryable: 12 tests - transition_to (state machine): 16 tests (all valid + invalid transitions) - mark_running / mark_success / mark_failed / mark_cancelled: 8 tests - update_progress: 9 tests - prepare_retry: 4 tests - to_dict serialization: 3 tests Total: 76 unit tests for job.py (289 lines) --- tests/unit/test_job_domain.py | 655 +++++++++++++++++----------------- 1 file changed, 335 insertions(+), 320 deletions(-) diff --git a/tests/unit/test_job_domain.py b/tests/unit/test_job_domain.py index ba0cc9aa1..948ccc105 100755 --- a/tests/unit/test_job_domain.py +++ b/tests/unit/test_job_domain.py @@ -1,4 +1,6 @@ -"""Job 领域层单元测试 - job.py""" +"""Job 领域模型单元测试。""" + +from datetime import datetime, timezone import pytest @@ -10,45 +12,45 @@ from packages.domain.job import ( ) -class TestJobType: - """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_all_types_have_values(self): - """所有枚举成员都有字符串值""" - for jt in JobType: - assert isinstance(jt.value, str) - assert jt.value + def test_from_string(self): + assert JobType("video_compose") == JobType.VIDEO_COMPOSE + assert JobType("generation") == JobType.GENERATION - def test_str_enum_behavior(self): - """是 str 枚举""" - assert JobType.VIDEO_COMPOSE == "video_compose" - assert isinstance(JobType.VIDEO_COMPOSE, str) - - def test_known_types_exist(self): - """核心任务类型都存在""" - assert JobType.VIDEO_COMPOSE - assert JobType.RENDER_EDIT_PLAN - assert JobType.ASSET_INGEST - assert JobType.CLASSIFICATION - assert JobType.GENERATION + def test_invalid_type_raises(self): + with pytest.raises(ValueError): + JobType("invalid_type") -class TestJobStatus: - """JobStatus 枚举测试""" +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_all_statuses_have_values(self): - for js in JobStatus: - assert isinstance(js.value, str) - assert js.value + def test_from_string(self): + assert JobStatus("pending") == JobStatus.PENDING + assert JobStatus("success") == JobStatus.SUCCESS - def test_str_enum_behavior(self): - assert JobStatus.PENDING == "pending" - assert isinstance(JobStatus.PENDING, str) - def test_terminal_statuses(self): - """终态集合包含成功/失败/取消""" +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): @@ -59,372 +61,376 @@ class TestJobStatus: class TestJobCreate: - """Job.create 工厂方法测试""" - - def test_create_basic(self): - """基本创建""" - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - ) - assert job.id - assert len(job.id) == 32 # uuid4 hex - assert job.project_id == "proj-1" + 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.created_at - assert job.updated_at + 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_string_job_type(self): - """用字符串创建任务类型""" - job = Job.create( - project_id="proj-1", - job_type="video_compose", - ) + 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_invalid_string_job_type_raises(self): - """无效的任务类型字符串抛 ValueError""" - with pytest.raises(ValueError, match="不支持的任务类型"): - Job.create(project_id="proj-1", job_type="invalid_type") - - def test_create_empty_project_id_raises(self): - """空 project_id 抛 ValueError""" - with pytest.raises(ValueError, match="project_id 不能为空"): - Job.create(project_id=" ", job_type=JobType.VIDEO_COMPOSE) - def test_create_with_payload(self): - """带 payload 创建""" - payload = {"video_id": "v1", "quality": "1080p"} - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - payload=payload, - ) + payload = {"edit_plan_id": "plan123", "resolution": "1080p"} + job = Job.create("p1", JobType.VIDEO_COMPOSE, payload=payload) assert job.payload == payload - def test_create_with_source_id(self): - """带 source_id 创建""" - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - source_id="plan-123", - ) - assert job.source_id == "plan-123" - - def test_create_with_created_by(self): - """带创建人""" - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - created_by_user_id="user-1", - ) - assert job.created_by_user_id == "user-1" - - def test_create_with_custom_max_retries(self): - """自定义最大重试次数""" - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - max_retries=5, - ) - assert job.max_retries == 5 - - def test_create_project_id_stripped(self): - """project_id 会被 strip""" - job = Job.create( - project_id=" proj-1 ", - job_type=JobType.VIDEO_COMPOSE, - ) - assert job.project_id == "proj-1" - - def test_create_source_id_stripped(self): - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - source_id=" src-1 ", - ) - assert job.source_id == "src-1" - - def test_create_created_by_stripped(self): - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - created_by_user_id=" user-1 ", - ) - assert job.created_by_user_id == "user-1" - - def test_create_none_payload_defaults_to_empty_dict(self): - """payload=None 时默认为空 dict""" - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - payload=None, - ) + 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" -class TestJobIsTerminal: - """is_terminal 属性测试""" + 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(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) assert job.is_terminal is False def test_running_not_terminal(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + 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(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + 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(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + 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(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to(JobStatus.CANCELLED) assert job.is_terminal is True -class TestJobTransitions: - """状态转换测试""" +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(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + 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): - """pending 可以直接到 success（快速成功）""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + 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(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to(JobStatus.CANCELLED) assert job.status == JobStatus.CANCELLED - def test_running_to_success(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.transition_to(JobStatus.RUNNING) - job.transition_to(JobStatus.SUCCESS) - assert job.status == JobStatus.SUCCESS - assert job.completed_at is not None - - def test_running_to_failed(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.transition_to(JobStatus.RUNNING) - job.transition_to(JobStatus.FAILED) - assert job.status == JobStatus.FAILED - assert job.completed_at is not None - - def test_running_to_cancelled(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.transition_to(JobStatus.RUNNING) - job.transition_to(JobStatus.CANCELLED) - assert job.status == JobStatus.CANCELLED - - def test_failed_to_pending_retry(self): - """失败后可以回到 pending（重试）""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.transition_to(JobStatus.RUNNING) - job.transition_to(JobStatus.FAILED) - job.transition_to(JobStatus.PENDING) - assert job.status == JobStatus.PENDING - - def test_invalid_transition_raises(self): - """非法状态转换抛 ValueError""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - # pending 不能直接到 failed + 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_success_to_pending_raises(self): - """成功后不能回到 pending""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + 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.PENDING) + job.transition_to(JobStatus.FAILED) + with pytest.raises(ValueError): + job.transition_to(JobStatus.RUNNING) def test_transition_with_string_status(self): - """用字符串做状态转换""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to("running") assert job.status == JobStatus.RUNNING - def test_transition_invalid_string_raises(self): - """无效状态字符串抛 ValueError""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + 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): - """状态转换更新 updated_at""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - old_updated = job.updated_at - import time - - time.sleep(0.001) + job = Job.create("p1", JobType.VIDEO_COMPOSE) + old_time = job.updated_at job.transition_to(JobStatus.RUNNING) - assert job.updated_at >= old_updated + assert job.updated_at >= old_time def test_started_at_only_set_once(self): - """started_at 只在第一次 RUNNING 时设置""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to(JobStatus.RUNNING) - first_started = job.started_at - job.transition_to(JobStatus.SUCCESS) - # 回到 pending 再 running（模拟重试场景，但started_at是None时才设置） - # 注意：正常重试是通过 prepare_retry 重置的 - assert first_started is not None + 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 TestJobMarkMethods: - """便捷标记方法测试""" - - def test_mark_running(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.mark_running("合成中") - assert job.status == JobStatus.RUNNING - assert job.current_stage == "合成中" - - def test_mark_running_no_stage(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) +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.current_stage == "" + assert job.started_at is not None - def test_mark_success(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.mark_running() - job.mark_success({"output_url": "http://..."}) - assert job.status == JobStatus.SUCCESS - assert job.progress == 100.0 - assert job.current_stage == "完成" - assert job.result == {"output_url": "http://..."} + 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_success_no_result(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + 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.result == {} + assert job.progress == 100.0 + assert job.current_stage == "完成" + assert job.completed_at is not None - def test_mark_failed(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + 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_cancelled(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + 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 TestJobProgress: - """进度更新测试""" - def test_update_progress(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.update_progress(50.0, "渲染中") +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 - assert job.current_stage == "渲染中" def test_update_progress_zero(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.update_progress(0.0) assert job.progress == 0.0 - def test_update_progress_100(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + 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_raises(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + 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_raises(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + 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_without_stage(self): - """不传 stage 时不修改 current_stage""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.current_stage = "初始阶段" - job.update_progress(30.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 == "初始阶段" + assert job.current_stage == "渲染中" - def test_update_progress_updates_updated_at(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - old_updated = job.updated_at - import time - - time.sleep(0.001) + 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.updated_at >= old_updated + 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 TestJobRetry: - """重试逻辑测试""" - - def test_is_retryable_failed_within_limit(self): - """失败且未超过重试上限时可重试""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=3) - job.mark_running() - job.mark_failed("错误") - assert job.is_retryable is True - - def test_is_retryable_failed_at_limit(self): - """达到重试上限时不可重试""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=1) - job.mark_running() - job.mark_failed("错误") - job.retry_count = 1 - assert job.is_retryable is False - - def test_is_retryable_pending_false(self): - """pending 状态不可重试""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - assert job.is_retryable is False - - def test_is_retryable_success_false(self): - """成功状态不可重试""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.mark_running() - job.mark_success() - assert job.is_retryable is False - - def test_prepare_retry(self): - """准备重试""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=3) +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.celery_task_id = "task-123" job.prepare_retry() @@ -437,38 +443,41 @@ class TestJobRetry: assert job.completed_at is None assert job.celery_task_id == "" - def test_prepare_retry_not_retryable_raises(self): - """不可重试时抛 ValueError""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=0) + def test_prepare_retry_increments_count(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=5) job.mark_running() - job.mark_failed("错误") - with pytest.raises(ValueError, match="任务不可重试"): - job.prepare_retry() + job.mark_failed("err") - def test_prepare_retry_increments_correctly(self): - """多次重试计数正确""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=3) - job.mark_running() - job.mark_failed("错误1") job.prepare_retry() assert job.retry_count == 1 + # 再次失败重试 job.mark_running() - job.mark_failed("错误2") + 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() -class TestJobToDict: - """to_dict 序列化测试""" + def test_prepare_retry_wrong_status_raises(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + with pytest.raises(ValueError, match="任务不可重试"): + job.prepare_retry() - def test_to_dict_contains_all_fields(self): + +class TestToDict: + def test_to_dict_structure(self): job = Job.create( - project_id="p1", - job_type=JobType.VIDEO_COMPOSE, - payload={"key": "value"}, - source_id="src-1", - created_by_user_id="user-1", + "p1", + JobType.VIDEO_COMPOSE, + payload={"key": "val"}, + source_id="src1", + created_by_user_id="u1", ) d = job.to_dict() assert d["id"] == job.id @@ -476,33 +485,39 @@ class TestJobToDict: assert d["job_type"] == "video_compose" assert d["status"] == "pending" assert d["progress"] == 0.0 - assert d["payload"] == {"key": "value"} - assert d["source_id"] == "src-1" - assert d["created_by_user_id"] == "user-1" + 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 - - def test_to_dict_datetime_fields_are_strings(self): - """时间字段序列化为 ISO 字符串""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - d = job.to_dict() - assert isinstance(d["created_at"], str) - assert isinstance(d["updated_at"], str) - - def test_to_dict_none_datetime_fields(self): - """未设置的时间字段为 None""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - d = job.to_dict() 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): - """成功后 to_dict 状态正确""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.mark_running() - job.mark_success({"url": "http://..."}) + 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["result"] == {"url": "http://..."} + 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