diff --git a/tests/unit/test_wave77_generation_task.py b/tests/unit/test_wave77_generation_task.py new file mode 100755 index 000000000..5d9c24ebb --- /dev/null +++ b/tests/unit/test_wave77_generation_task.py @@ -0,0 +1,466 @@ +"""第77波:GenerationTask 领域模型纯逻辑单测。""" + +import json + +import pytest + +from packages.domain.generation_task import GenerationTask, GenerationTaskStatus + +# ============================================================ +# GenerationTaskStatus._missing_ 兼容枚举测试 +# ============================================================ + + +class TestGenerationTaskStatusMissing: + def test_completed_aliases(self): + for val in ("done", "success", "finished", "complete", "completed"): + assert GenerationTaskStatus(val) == GenerationTaskStatus.COMPLETED + assert GenerationTaskStatus(val.upper()) == GenerationTaskStatus.COMPLETED + + def test_failed_aliases(self): + for val in ("fail", "failed", "error", "err"): + assert GenerationTaskStatus(val) == GenerationTaskStatus.FAILED + + def test_running_aliases(self): + for val in ("process", "processing", "run", "running", "in_progress"): + assert GenerationTaskStatus(val) == GenerationTaskStatus.RUNNING + + def test_cancelled_aliases(self): + for val in ("cancel", "cancelled", "canceled"): + assert GenerationTaskStatus(val) == GenerationTaskStatus.CANCELLED + + def test_unknown_defaults_to_pending(self): + assert GenerationTaskStatus("unknown_state") == GenerationTaskStatus.PENDING + assert GenerationTaskStatus("") == GenerationTaskStatus.PENDING + + def test_whitespace_and_case_insensitive(self): + assert GenerationTaskStatus(" DONE ") == GenerationTaskStatus.COMPLETED + assert GenerationTaskStatus("Failed") == GenerationTaskStatus.FAILED + + def test_non_string_falls_back_to_pending(self): + assert GenerationTaskStatus(None) == GenerationTaskStatus.PENDING + assert GenerationTaskStatus(123) == GenerationTaskStatus.PENDING + + def test_normal_values_still_work(self): + assert GenerationTaskStatus("pending") == GenerationTaskStatus.PENDING + assert GenerationTaskStatus("running") == GenerationTaskStatus.RUNNING + assert GenerationTaskStatus("completed") == GenerationTaskStatus.COMPLETED + assert GenerationTaskStatus("failed") == GenerationTaskStatus.FAILED + assert GenerationTaskStatus("cancelled") == GenerationTaskStatus.CANCELLED + + +# ============================================================ +# GenerationTask.create 测试 +# ============================================================ + + +class TestGenerationTaskCreate: + def test_create_minimal(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + assert task.id + assert task.project_id == "p1" + assert task.asset_library_id == "lib1" + assert task.status == GenerationTaskStatus.PENDING + assert task.progress == 0.0 + assert task.result_count == 0 + assert task.retry_count == 0 + assert task.asset_ids == [] + assert task.title_ids == [] + assert task.voice_ids == [] + assert task.bgm_config == {} + assert task.logs == "[]" + + def test_create_requires_project_or_template(self): + """project_id 和 template_id 至少需要一个。""" + with pytest.raises(ValueError, match="至少需要提供一个"): + GenerationTask.create(project_id="", asset_library_id="lib1", template_id="") + + def test_create_template_id_only(self): + task = GenerationTask.create(project_id="", template_id="tpl1", asset_library_id="lib1") + assert task.template_id == "tpl1" + assert task.project_id == "" + + def test_create_requires_library_or_assets(self): + """asset_library_id 和 asset_ids/title_ids/voice_ids 至少需要一个。""" + with pytest.raises(ValueError, match="至少需要提供一个"): + GenerationTask.create( + project_id="p1", + asset_library_id="", + asset_ids=None, + title_ids=None, + voice_ids=None, + ) + + def test_create_asset_ids_only(self): + task = GenerationTask.create(project_id="p1", asset_library_id="", asset_ids=["a1", "a2"]) + assert task.asset_ids == ["a1", "a2"] + assert task.asset_library_id == "" + + def test_create_title_ids_only(self): + task = GenerationTask.create(project_id="p1", asset_library_id="", title_ids=["t1"]) + assert task.title_ids == ["t1"] + + def test_create_voice_ids_only(self): + task = GenerationTask.create(project_id="p1", asset_library_id="", voice_ids=["v1"]) + assert task.voice_ids == ["v1"] + + def test_create_full_params(self): + task = GenerationTask.create( + project_id="p1", + asset_library_id="lib1", + strategy_id="s1", + voice_library_id="vlib1", + template_id="tpl1", + asset_ids=["a1", "a2"], + title_ids=["t1"], + voice_ids=["v1"], + created_by_user_id="u1", + source_edit_plan_id="ep1", + asset_select_mode="random", + batch_id="batch_001", + video_title="测试视频", + resolution="1080p", + bgm_config={"volume": 0.5}, + auto_retry_enabled=True, + auto_retry_max=3, + ) + assert task.strategy_id == "s1" + assert task.voice_library_id == "vlib1" + assert task.template_id == "tpl1" + assert task.asset_ids == ["a1", "a2"] + assert task.title_ids == ["t1"] + assert task.voice_ids == ["v1"] + assert task.created_by_user_id == "u1" + assert task.source_edit_plan_id == "ep1" + assert task.asset_select_mode == "random" + assert task.batch_id == "batch_001" + assert task.video_title == "测试视频" + assert task.resolution == "1080p" + assert task.bgm_config == {"volume": 0.5} + assert task.auto_retry_enabled is True + assert task.auto_retry_max == 3 + + def test_create_strips_string_fields(self): + task = GenerationTask.create( + project_id=" p1 ", + asset_library_id=" lib1 ", + strategy_id=" s1 ", + template_id=" tpl1 ", + created_by_user_id=" u1 ", + video_title=" 测试 ", + resolution=" 1080p ", + ) + assert task.project_id == "p1" + assert task.asset_library_id == "lib1" + assert task.strategy_id == "s1" + assert task.template_id == "tpl1" + assert task.created_by_user_id == "u1" + assert task.video_title == "测试" + assert task.resolution == "1080p" + + def test_create_asset_ids_is_copy(self): + ids = ["a1", "a2"] + task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=ids) + ids.append("a3") + assert task.asset_ids == ["a1", "a2"] + + def test_create_bgm_config_is_copy(self): + cfg = {"vol": 0.5} + task = GenerationTask.create(project_id="p1", asset_library_id="lib1", bgm_config=cfg) + cfg["vol"] = 0.8 + assert task.bgm_config == {"vol": 0.5} + + def test_create_id_unique(self): + t1 = GenerationTask.create(project_id="p1", asset_library_id="lib1") + t2 = GenerationTask.create(project_id="p1", asset_library_id="lib1") + assert t1.id != t2.id + + +# ============================================================ +# GenerationTask 状态查询测试 +# ============================================================ + + +class TestGenerationTaskStatusQuery: + def test_is_terminal_pending_false(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + assert not task.is_terminal + + def test_is_terminal_running_false(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.mark_processing() + assert not task.is_terminal + + def test_is_terminal_completed_true(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.mark_processing() + task.mark_completed() + assert task.is_terminal + + def test_is_terminal_failed_true(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.mark_processing() + task.mark_failed("err") + assert task.is_terminal + + def test_is_terminal_cancelled_true(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.mark_cancelled() + assert task.is_terminal + + def test_is_completed(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + assert not task.is_completed + task.mark_processing() + task.mark_completed() + assert task.is_completed + + def test_is_failed(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + assert not task.is_failed + task.mark_processing() + task.mark_failed("err") + assert task.is_failed + + def test_is_running(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + assert not task.is_running + task.mark_processing() + assert task.is_running + + +# ============================================================ +# GenerationTask 状态转换测试 +# ============================================================ + + +class TestGenerationTaskTransitions: + def test_pending_to_running(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.transition_to(GenerationTaskStatus.RUNNING) + assert task.status == GenerationTaskStatus.RUNNING + + def test_pending_to_failed(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.transition_to(GenerationTaskStatus.FAILED) + assert task.status == GenerationTaskStatus.FAILED + + def test_pending_to_cancelled(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.transition_to(GenerationTaskStatus.CANCELLED) + assert task.status == GenerationTaskStatus.CANCELLED + + def test_pending_to_completed_invalid(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + with pytest.raises(ValueError, match="非法状态转换"): + task.transition_to(GenerationTaskStatus.COMPLETED) + + def test_running_to_completed(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.transition_to(GenerationTaskStatus.RUNNING) + task.transition_to(GenerationTaskStatus.COMPLETED) + assert task.status == GenerationTaskStatus.COMPLETED + + def test_running_to_failed(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.transition_to(GenerationTaskStatus.RUNNING) + task.transition_to(GenerationTaskStatus.FAILED) + assert task.status == GenerationTaskStatus.FAILED + + def test_running_to_cancelled(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.transition_to(GenerationTaskStatus.RUNNING) + task.transition_to(GenerationTaskStatus.CANCELLED) + assert task.status == GenerationTaskStatus.CANCELLED + + def test_failed_to_pending(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.transition_to(GenerationTaskStatus.RUNNING) + task.transition_to(GenerationTaskStatus.FAILED) + task.transition_to(GenerationTaskStatus.PENDING) + assert task.status == GenerationTaskStatus.PENDING + + def test_completed_to_pending_invalid(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.transition_to(GenerationTaskStatus.RUNNING) + task.transition_to(GenerationTaskStatus.COMPLETED) + with pytest.raises(ValueError): + task.transition_to(GenerationTaskStatus.PENDING) + + def test_transition_to_accepts_string(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.transition_to("running") + assert task.status == GenerationTaskStatus.RUNNING + + def test_transition_to_unknown_string_defaults_pending(self): + """_missing_ 兜底:未知字符串映射为 PENDING,再走状态机校验。""" + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + # 当前已经是 pending,pending→pending 不合法 + with pytest.raises(ValueError, match="非法状态转换"): + task.transition_to("bogus_status") + + +# ============================================================ +# GenerationTask 业务方法测试 +# ============================================================ + + +class TestGenerationTaskBusinessMethods: + def test_mark_processing_sets_started_at(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + assert task.started_at is None + task.mark_processing() + assert task.started_at is not None + assert task.status == GenerationTaskStatus.RUNNING + + def test_mark_processing_clears_error(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.error_message = "old error" + task.mark_processing() + assert task.error_message == "" + + def test_mark_completed_default_count(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.mark_processing() + task.mark_completed() + assert task.status == GenerationTaskStatus.COMPLETED + assert task.progress == 100.0 + assert task.result_count == 1 + assert task.completed_at is not None + assert task.error_message == "" + + def test_mark_completed_custom_count(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.mark_processing() + task.mark_completed(result_count=5) + assert task.result_count == 5 + + def test_mark_failed_with_message(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.mark_processing() + task.mark_failed("网络超时") + assert task.status == GenerationTaskStatus.FAILED + assert task.error_message == "网络超时" + assert task.completed_at is not None + + def test_mark_failed_with_error_info(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.mark_processing() + info = {"error_type": "NetworkError", "stage": "download"} + task.mark_failed("超时", error_info=info) + assert task.error_info == info + + def test_mark_failed_default_error_info(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.mark_processing() + task.mark_failed("未知错误") + assert "error_type" in task.error_info + assert task.error_info["message"] == "未知错误" + assert "failed_at" in task.error_info + + def test_mark_cancelled_from_pending(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.mark_cancelled() + assert task.status == GenerationTaskStatus.CANCELLED + assert task.completed_at is not None + + def test_mark_cancelled_from_running(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.mark_processing() + task.mark_cancelled() + assert task.status == GenerationTaskStatus.CANCELLED + + def test_mark_pending_from_failed(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.mark_processing() + task.mark_failed("err") + task.mark_pending_from_failed() + assert task.status == GenerationTaskStatus.PENDING + assert task.error_message == "" + assert task.error_info == {} + assert task.started_at is None + assert task.completed_at is None + assert task.progress == 0.0 + assert task.result_count == 0 + assert task.retry_count == 1 + + def test_mark_pending_from_failed_multiple_retries(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + for i in range(3): + task.mark_processing() + task.mark_failed(f"err{i}") + task.mark_pending_from_failed() + assert task.retry_count == i + 1 + assert task.status == GenerationTaskStatus.PENDING + + def test_mark_pending_from_failed_wrong_status_raises(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + with pytest.raises(ValueError, match="只有 failed 状态"): + task.mark_pending_from_failed() + + +# ============================================================ +# GenerationTask 日志系统测试 +# ============================================================ + + +class TestGenerationTaskLogs: + def test_append_log_basic(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.append_log("初始化", "任务创建成功") + logs = task.get_logs() + assert len(logs) == 1 + assert logs[0]["stage"] == "初始化" + assert logs[0]["message"] == "任务创建成功" + assert logs[0]["level"] == "INFO" + assert "ts" in logs[0] + + def test_append_log_with_level(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.append_log("渲染", "渲染失败", level="ERROR") + logs = task.get_logs() + assert logs[0]["level"] == "ERROR" + + def test_append_log_with_extra_fields(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.append_log("下载", "下载完成", asset_id="a1", duration=10.5) + logs = task.get_logs() + assert logs[0]["asset_id"] == "a1" + assert logs[0]["duration"] == 10.5 + + def test_append_multiple_logs(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + for i in range(5): + task.append_log(f"阶段{i}", f"消息{i}") + logs = task.get_logs() + assert len(logs) == 5 + assert logs[0]["stage"] == "阶段0" + assert logs[4]["stage"] == "阶段4" + + def test_get_logs_empty(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + assert task.get_logs() == [] + + def test_get_logs_corrupted_json(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.logs = "not valid json" + assert task.get_logs() == [] + + def test_append_log_respects_max_limit(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + # _MAX_LOGS = 200,超过后保留最新的 200 条 + for i in range(250): + task.append_log("阶段", f"消息{i}") + logs = task.get_logs() + assert len(logs) == 200 + assert logs[0]["message"] == "消息50" # 前50条被截掉 + assert logs[-1]["message"] == "消息249" + + def test_logs_persisted_as_json_string(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1") + task.append_log("阶段", "消息") + assert isinstance(task.logs, str) + # 可以被 json.loads 解析 + parsed = json.loads(task.logs) + assert isinstance(parsed, list) + assert len(parsed) == 1