From 615637d1a4a213a0745f1cb39365dac7f13b5259 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sat, 25 Jul 2026 12:58:33 +0800 Subject: [PATCH] =?UTF-8?q?test(unit):=20=E7=AC=AC76=E6=B3=A2=20-=20TTSJob?= =?UTF-8?q?+Job=E7=8A=B6=E6=80=81=E6=9C=BA=20+=20Tag=E9=A2=86=E5=9F=9F?= =?UTF-8?q?=E5=AE=9E=E4=BD=93=20(+101)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - TTSJob: create参数校验/状态转换/业务方法/to_dict，约50个 - Job: create参数校验/状态转换/进度更新/重试机制/to_dict，约45个 - Tag: create验证逻辑，6个 - 全部本地全绿，black格式化通过 --- tests/unit/test_wave76_tts_job_tag.py | 839 ++++++++++++++++++++++++++ 1 file changed, 839 insertions(+) create mode 100755 tests/unit/test_wave76_tts_job_tag.py diff --git a/tests/unit/test_wave76_tts_job_tag.py b/tests/unit/test_wave76_tts_job_tag.py new file mode 100755 index 000000000..be7f6a0f7 --- /dev/null +++ b/tests/unit/test_wave76_tts_job_tag.py @@ -0,0 +1,839 @@ +"""第76波：TTSJob + Job + Tag 领域纯逻辑单测。""" + +from datetime import datetime, timezone + +import pytest + +from packages.domain.tag import Tag +from packages.domain.tts_job import TTSJob, TTSJobStatus +from packages.domain.job import Job, JobStatus, JobType + +# ============================================================ +# Tag.create 测试 +# ============================================================ + + +class TestTagCreate: + def test_create_basic(self): + tag = Tag.create(user_id="u1", name="风景") + assert tag.id + assert tag.user_id == "u1" + assert tag.name == "风景" + assert isinstance(tag.created_at, datetime) + + def test_create_empty_name_raises(self): + with pytest.raises(ValueError, match="标签名称不能为空"): + Tag.create(user_id="u1", name="") + + def test_create_whitespace_name_raises(self): + with pytest.raises(ValueError, match="标签名称不能为空"): + Tag.create(user_id="u1", name=" ") + + def test_create_name_stripped(self): + tag = Tag.create(user_id="u1", name=" 美食 ") + assert tag.name == "美食" + + def test_create_id_is_unique(self): + tag1 = Tag.create(user_id="u1", name="a") + tag2 = Tag.create(user_id="u1", name="b") + assert tag1.id != tag2.id + + def test_create_uses_utc_timezone(self): + before = datetime.now(timezone.utc) + tag = Tag.create(user_id="u1", name="t") + after = datetime.now(timezone.utc) + assert before <= tag.created_at <= after + + +# ============================================================ +# TTSJob.create 测试 +# ============================================================ + + +class TestTTSJobCreate: + def test_create_minimal(self): + job = TTSJob.create(user_id="u1", input_text="你好世界") + assert job.id + assert job.user_id == "u1" + assert job.input_text == "你好世界" + assert job.status == TTSJobStatus.PENDING + assert job.retry_count == 0 + assert job.max_retries == 3 + assert job.format == "mp3" + assert job.sample_rate == 22050 + assert job.error_message == "" + assert isinstance(job.created_at, datetime) + assert isinstance(job.updated_at, datetime) + + def test_create_empty_user_id_raises(self): + with pytest.raises(ValueError, match="user_id 不能为空"): + TTSJob.create(user_id=" ", input_text="hi") + + def test_create_empty_input_text_raises(self): + with pytest.raises(ValueError, match="input_text 不能为空"): + TTSJob.create(user_id="u1", input_text="") + + def test_create_whitespace_input_raises(self): + with pytest.raises(ValueError, match="input_text 不能为空"): + TTSJob.create(user_id="u1", input_text=" \n\t ") + + def test_create_too_long_input_raises(self): + long_text = "a" * 10001 + with pytest.raises(ValueError, match="长度不能超过"): + TTSJob.create(user_id="u1", input_text=long_text) + + def test_create_boundary_length_ok(self): + text = "a" * 10000 + job = TTSJob.create(user_id="u1", input_text=text) + assert len(job.input_text) == 10000 + + def test_create_invalid_format_raises(self): + with pytest.raises(ValueError, match="不支持的输出格式"): + TTSJob.create(user_id="u1", input_text="hi", format="flac") + + def test_create_valid_formats(self): + for fmt in ("mp3", "wav", "pcm"): + job = TTSJob.create(user_id="u1", input_text="hi", format=fmt) + assert job.format == fmt + + def test_create_strips_whitespace(self): + job = TTSJob.create( + user_id=" u1 ", + input_text=" hello ", + voice_id=" v1 ", + voice_model=" cosyvoice ", + project_id=" p1 ", + voice_clone_profile_id=" vp1 ", + ) + assert job.user_id == "u1" + assert job.input_text == "hello" + assert job.voice_id == "v1" + assert job.voice_model == "cosyvoice" + assert job.project_id == "p1" + assert job.voice_clone_profile_id == "vp1" + + def test_create_custom_params(self): + job = TTSJob.create( + user_id="u1", + input_text="hi", + voice_id="voice_001", + voice_model="cosyvoice", + project_id="proj_001", + voice_clone_profile_id="vcp_001", + sample_rate=16000, + format="wav", + max_retries=5, + metadata={"key": "val"}, + ) + assert job.voice_id == "voice_001" + assert job.voice_model == "cosyvoice" + assert job.project_id == "proj_001" + assert job.voice_clone_profile_id == "vcp_001" + assert job.sample_rate == 16000 + assert job.format == "wav" + assert job.max_retries == 5 + assert job.metadata == {"key": "val"} + + def test_create_metadata_default_empty_dict(self): + job = TTSJob.create(user_id="u1", input_text="hi") + assert job.metadata == {} + + def test_create_id_unique(self): + j1 = TTSJob.create(user_id="u1", input_text="a") + j2 = TTSJob.create(user_id="u1", input_text="b") + assert j1.id != j2.id + + +# ============================================================ +# TTSJob 状态属性测试 +# ============================================================ + + +class TestTTSJobStatusProperties: + def test_is_terminal_pending_false(self): + job = TTSJob.create(user_id="u1", input_text="hi") + assert not job.is_terminal + + def test_is_terminal_processing_false(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.mark_processing() + assert not job.is_terminal + + def test_is_terminal_completed_true(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.mark_processing() + job.mark_completed("http://out.mp3") + assert job.is_terminal + + def test_is_terminal_failed_true(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.mark_processing() + job.mark_failed("error") + assert job.is_terminal + + def test_is_terminal_cancelled_true(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.mark_cancelled() + assert job.is_terminal + + def test_is_retryable_failed_within_limit(self): + job = TTSJob.create(user_id="u1", input_text="hi", max_retries=3) + job.mark_processing() + job.mark_failed("err") + assert job.is_retryable + + def test_is_retryable_failed_at_limit(self): + job = TTSJob.create(user_id="u1", input_text="hi", max_retries=2) + job.mark_processing() + job.mark_failed("err") + job.retry_count = 2 + assert not job.is_retryable + + def test_is_retryable_pending_false(self): + job = TTSJob.create(user_id="u1", input_text="hi") + assert not job.is_retryable + + def test_is_completed_needs_url(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.status = TTSJobStatus.COMPLETED + job.output_audio_url = "" + assert not job.is_completed + job.output_audio_url = "http://x.mp3" + assert job.is_completed + + +# ============================================================ +# TTSJob 状态转换测试 +# ============================================================ + + +class TestTTSJobTransitions: + def test_pending_to_processing(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.transition_to(TTSJobStatus.PROCESSING) + assert job.status == TTSJobStatus.PROCESSING + + def test_pending_to_failed(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.transition_to(TTSJobStatus.FAILED) + assert job.status == TTSJobStatus.FAILED + + def test_pending_to_cancelled(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.transition_to(TTSJobStatus.CANCELLED) + assert job.status == TTSJobStatus.CANCELLED + + def test_pending_to_completed_invalid(self): + job = TTSJob.create(user_id="u1", input_text="hi") + with pytest.raises(ValueError, match="非法状态转换"): + job.transition_to(TTSJobStatus.COMPLETED) + + def test_processing_to_completed(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.transition_to(TTSJobStatus.PROCESSING) + job.transition_to(TTSJobStatus.COMPLETED) + assert job.status == TTSJobStatus.COMPLETED + + def test_processing_to_failed(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.transition_to(TTSJobStatus.PROCESSING) + job.transition_to(TTSJobStatus.FAILED) + assert job.status == TTSJobStatus.FAILED + + def test_failed_to_pending_retry(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.transition_to(TTSJobStatus.PROCESSING) + job.transition_to(TTSJobStatus.FAILED) + job.retry_count = 0 + job.transition_to(TTSJobStatus.PENDING) + assert job.status == TTSJobStatus.PENDING + + def test_completed_to_anything_invalid(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.transition_to(TTSJobStatus.PROCESSING) + job.transition_to(TTSJobStatus.COMPLETED) + with pytest.raises(ValueError): + job.transition_to(TTSJobStatus.PENDING) + with pytest.raises(ValueError): + job.transition_to(TTSJobStatus.FAILED) + + def test_transition_to_accepts_string(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.transition_to("processing") + assert job.status == TTSJobStatus.PROCESSING + + def test_transition_to_invalid_string_raises(self): + job = TTSJob.create(user_id="u1", input_text="hi") + with pytest.raises(ValueError, match="无效状态"): + job.transition_to("invalid_state") + + def test_transition_updates_updated_at(self): + job = TTSJob.create(user_id="u1", input_text="hi") + before = job.updated_at + import time + + time.sleep(0.001) + job.transition_to(TTSJobStatus.PROCESSING) + assert job.updated_at >= before + + +# ============================================================ +# TTSJob 业务方法测试 +# ============================================================ + + +class TestTTSJobBusinessMethods: + def test_mark_processing_sets_started_at(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.mark_processing() + assert job.started_at is not None + assert job.status == TTSJobStatus.PROCESSING + assert job.error_message == "" + + def test_mark_processing_clears_error(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.error_message = "old error" + job.mark_processing() + assert job.error_message == "" + + def test_mark_completed_sets_fields(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.mark_processing() + job.mark_completed( + "http://out.mp3", + output_audio_key="oss://key", + duration=10.5, + file_size=102400, + ) + assert job.status == TTSJobStatus.COMPLETED + assert job.output_audio_url == "http://out.mp3" + assert job.output_audio_key == "oss://key" + assert job.duration == 10.5 + assert job.file_size == 102400 + assert job.completed_at is not None + assert job.is_completed + + def test_mark_completed_empty_url_raises(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.mark_processing() + with pytest.raises(ValueError, match="output_audio_url 不能为空"): + job.mark_completed(" ") + + def test_mark_failed_sets_error(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.mark_processing() + job.mark_failed("网络超时") + assert job.status == TTSJobStatus.FAILED + assert job.error_message == "网络超时" + + def test_mark_cancelled_from_pending(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.mark_cancelled() + assert job.status == TTSJobStatus.CANCELLED + + def test_mark_cancelled_from_processing(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.mark_processing() + job.mark_cancelled() + assert job.status == TTSJobStatus.CANCELLED + + def test_prepare_retry_success(self): + job = TTSJob.create(user_id="u1", input_text="hi", max_retries=3) + job.mark_processing() + job.mark_failed("err") + assert job.retry_count == 0 + job.prepare_retry() + assert job.status == TTSJobStatus.PENDING + assert job.retry_count == 1 + assert job.error_message == "" + assert job.started_at is None + assert job.completed_at is None + + def test_prepare_retry_multiple_times(self): + job = TTSJob.create(user_id="u1", input_text="hi", max_retries=3) + for i in range(3): + job.mark_processing() + job.mark_failed(f"err{i}") + job.prepare_retry() + assert job.retry_count == i + 1 + assert job.status == TTSJobStatus.PENDING + + def test_prepare_retry_exceed_max_raises(self): + job = TTSJob.create(user_id="u1", input_text="hi", max_retries=2) + job.mark_processing() + job.mark_failed("err1") + job.prepare_retry() + job.mark_processing() + job.mark_failed("err2") + job.prepare_retry() + # 第3次失败后 retry_count=2 == max_retries=2，不可重试 + job.mark_processing() + job.mark_failed("err3") + assert not job.is_retryable + with pytest.raises(ValueError, match="不可重试"): + job.prepare_retry() + + def test_prepare_retry_pending_raises(self): + job = TTSJob.create(user_id="u1", input_text="hi") + with pytest.raises(ValueError, match="不可重试"): + job.prepare_retry() + + def test_to_dict_contains_all_fields(self): + job = TTSJob.create( + user_id="u1", + input_text="hi", + voice_id="v1", + project_id="p1", + metadata={"k": "v"}, + ) + d = job.to_dict() + expected_keys = { + "id", + "user_id", + "project_id", + "voice_clone_profile_id", + "status", + "input_text", + "voice_id", + "voice_model", + "output_audio_url", + "output_audio_key", + "duration", + "file_size", + "sample_rate", + "format", + "error_message", + "retry_count", + "max_retries", + "is_retryable", + "is_completed", + "metadata", + "started_at", + "completed_at", + "created_at", + "updated_at", + } + assert set(d.keys()) == expected_keys + assert d["status"] == "pending" + assert d["is_retryable"] is False + assert d["is_completed"] is False + assert d["metadata"] == {"k": "v"} + # 时间字段应为 ISO 字符串或 None + assert isinstance(d["created_at"], str) + assert d["started_at"] is None + + def test_to_dict_completed_state(self): + job = TTSJob.create(user_id="u1", input_text="hi") + job.mark_processing() + job.mark_completed("http://out.mp3") + d = job.to_dict() + assert d["status"] == "completed" + assert d["is_completed"] is True + assert d["started_at"] is not None + assert d["completed_at"] is not None + + +# ============================================================ +# Job.create 测试 +# ============================================================ + + +class TestJobCreate: + def test_create_minimal(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + assert job.id + assert job.project_id == "p1" + assert job.job_type == JobType.VIDEO_COMPOSE + assert job.status == JobStatus.PENDING + assert job.progress == 0.0 + assert job.retry_count == 0 + assert job.max_retries == 3 + assert job.payload == {} + assert job.result == {} + + def test_create_empty_project_id_raises(self): + with pytest.raises(ValueError, match="project_id 不能为空"): + Job.create(project_id=" ", job_type=JobType.VIDEO_COMPOSE) + + def test_create_string_job_type(self): + job = Job.create(project_id="p1", job_type="video_compose") + assert job.job_type == JobType.VIDEO_COMPOSE + + def test_create_invalid_job_type_string_raises(self): + with pytest.raises(ValueError, match="不支持的任务类型"): + Job.create(project_id="p1", job_type="unknown_type") + + def test_create_strips_project_id(self): + job = Job.create(project_id=" p1 ", job_type=JobType.VIDEO_COMPOSE) + assert job.project_id == "p1" + + def test_create_full_params(self): + job = Job.create( + project_id="p1", + job_type=JobType.RENDER_EDIT_PLAN, + payload={"edit_plan_id": "ep1"}, + source_id="src_001", + created_by_user_id="u1", + max_retries=5, + ) + assert job.job_type == JobType.RENDER_EDIT_PLAN + assert job.payload == {"edit_plan_id": "ep1"} + assert job.source_id == "src_001" + assert job.created_by_user_id == "u1" + assert job.max_retries == 5 + + def test_create_all_job_types(self): + for jt in JobType: + job = Job.create(project_id="p1", job_type=jt) + assert job.job_type == jt + + def test_create_id_unique(self): + j1 = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + j2 = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + assert j1.id != j2.id + + def test_create_strips_source_and_user(self): + job = Job.create( + project_id="p1", + job_type=JobType.VIDEO_COMPOSE, + source_id=" s1 ", + created_by_user_id=" u1 ", + ) + assert job.source_id == "s1" + assert job.created_by_user_id == "u1" + + +# ============================================================ +# Job 状态属性测试 +# ============================================================ + + +class TestJobStatusProperties: + def test_is_terminal_pending_false(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + assert not job.is_terminal + + def test_is_terminal_running_false(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.mark_running() + assert not job.is_terminal + + def test_is_terminal_success_true(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.mark_running() + job.mark_success() + assert job.is_terminal + + def test_is_terminal_failed_true(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.mark_running() + job.mark_failed("err") + assert job.is_terminal + + def test_is_terminal_cancelled_true(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.mark_cancelled() + assert job.is_terminal + + 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("err") + assert job.is_retryable + + def test_is_retryable_failed_at_limit(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=2) + job.mark_running() + job.mark_failed("err") + job.retry_count = 2 + assert not job.is_retryable + + 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 not job.is_retryable + + +# ============================================================ +# Job 状态转换测试 +# ============================================================ + + +class TestJobTransitions: + def test_pending_to_running(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.transition_to(JobStatus.RUNNING) + assert job.status == JobStatus.RUNNING + + def test_pending_to_success(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.transition_to(JobStatus.SUCCESS) + assert job.status == JobStatus.SUCCESS + + def test_pending_to_cancelled(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.transition_to(JobStatus.CANCELLED) + assert job.status == JobStatus.CANCELLED + + def test_pending_to_failed_invalid(self): + """pending 不能直接到 failed，必须经过 running 或直接 success/cancelled。""" + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + with pytest.raises(ValueError, match="非法状态转换"): + job.transition_to(JobStatus.FAILED) + + 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 + + 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 + + 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): + 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_success_to_pending_invalid(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.transition_to(JobStatus.RUNNING) + job.transition_to(JobStatus.SUCCESS) + with pytest.raises(ValueError): + job.transition_to(JobStatus.PENDING) + + def test_transition_to_accepts_string(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.transition_to("running") + assert job.status == JobStatus.RUNNING + + def test_transition_to_invalid_string_raises(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + with pytest.raises(ValueError, match="无效状态"): + job.transition_to("bogus") + + def test_transition_running_sets_started_at(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + assert job.started_at is None + job.transition_to(JobStatus.RUNNING) + assert job.started_at is not None + + def test_transition_success_sets_completed_at(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.transition_to(JobStatus.RUNNING) + assert job.completed_at is None + job.transition_to(JobStatus.SUCCESS) + assert job.completed_at is not None + + def test_transition_failed_sets_completed_at(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.transition_to(JobStatus.RUNNING) + job.transition_to(JobStatus.FAILED) + assert job.completed_at is not None + + def test_started_at_not_overwritten_on_second_running_transition(self): + """通过 transition_to 再次到 RUNNING 时，如果 started_at 已有值不覆盖。""" + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.transition_to(JobStatus.RUNNING) + first = job.started_at + # 走个 retry 流程再回来 + job.transition_to(JobStatus.FAILED) + job.transition_to(JobStatus.PENDING) + job.started_at = None # prepare_retry 会清掉，这里模拟 + job.transition_to(JobStatus.RUNNING) + second = job.started_at + # started_at 被重置后应该重新设置 + assert second is not None + # 时间可能相同（精度问题），但逻辑上第二次 running 会重新设置 + assert isinstance(second, datetime) + + +# ============================================================ +# Job 业务方法测试 +# ============================================================ + + +class TestJobBusinessMethods: + def test_mark_running_with_stage(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_without_stage(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.current_stage = "之前" + job.mark_running() + assert job.status == JobStatus.RUNNING + assert job.current_stage == "之前" # 不传 stage 不修改 + + def test_mark_success_with_result(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.mark_running() + job.mark_success({"output_url": "http://x.mp4"}) + assert job.status == JobStatus.SUCCESS + assert job.progress == 100.0 + assert job.current_stage == "完成" + assert job.result == {"output_url": "http://x.mp4"} + + def test_mark_success_without_result(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.mark_running() + job.mark_success() + assert job.status == JobStatus.SUCCESS + assert job.progress == 100.0 + assert job.result == {} + + def test_mark_failed(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.mark_running() + job.mark_failed("渲染失败") + assert job.status == JobStatus.FAILED + assert job.error_message == "渲染失败" + assert job.current_stage == "失败" + + def test_mark_cancelled(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.mark_cancelled() + assert job.status == JobStatus.CANCELLED + assert job.current_stage == "已取消" + + def test_update_progress_normal(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.update_progress(50.0, "合成中") + assert job.progress == 50.0 + assert job.current_stage == "合成中" + + def test_update_progress_boundary_zero(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.update_progress(0.0) + assert job.progress == 0.0 + + def test_update_progress_boundary_hundred(self): + job = Job.create(project_id="p1", job_type=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) + 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) + with pytest.raises(ValueError, match="进度必须在 0~100 之间"): + job.update_progress(100.1) + + def test_update_progress_updates_updated_at(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + before = job.updated_at + import time + + time.sleep(0.001) + job.update_progress(30.0) + assert job.updated_at >= before + + def test_prepare_retry_success(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=3) + job.mark_running("阶段1") + job.mark_failed("错误1") + job.celery_task_id = "celery-123" + 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_multiple_times(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=3) + for i in range(3): + job.mark_running() + job.mark_failed(f"err{i}") + job.prepare_retry() + assert job.retry_count == i + 1 + assert job.status == JobStatus.PENDING + + def test_prepare_retry_exceed_max_raises(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=2) + job.mark_running() + job.mark_failed("err1") + job.prepare_retry() + job.mark_running() + job.mark_failed("err2") + job.prepare_retry() + job.mark_running() + job.mark_failed("err3") + assert not job.is_retryable + with pytest.raises(ValueError, match="不可重试"): + job.prepare_retry() + + def test_prepare_retry_pending_raises(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + with pytest.raises(ValueError, match="不可重试"): + job.prepare_retry() + + def test_to_dict_contains_all_fields(self): + job = Job.create( + project_id="p1", + job_type=JobType.VIDEO_COMPOSE, + payload={"k": "v"}, + source_id="s1", + created_by_user_id="u1", + ) + d = job.to_dict() + expected_keys = { + "id", + "project_id", + "job_type", + "status", + "progress", + "current_stage", + "payload", + "result", + "error_message", + "retry_count", + "max_retries", + "celery_task_id", + "source_id", + "created_by_user_id", + "is_retryable", + "started_at", + "completed_at", + "created_at", + "updated_at", + } + assert set(d.keys()) == expected_keys + assert d["status"] == "pending" + assert d["job_type"] == "video_compose" + assert d["progress"] == 0.0 + assert d["payload"] == {"k": "v"} + assert d["is_retryable"] is False + assert d["started_at"] is None + assert isinstance(d["created_at"], str) + + def test_to_dict_success_state(self): + job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job.mark_running() + job.mark_success({"output": "ok"}) + d = job.to_dict() + assert d["status"] == "success" + assert d["progress"] == 100.0 + assert d["result"] == {"output": "ok"} + assert d["started_at"] is not None + assert d["completed_at"] is not None -- 2.54.0