diff --git a/tests/unit/domain/test_generation_task.py b/tests/unit/domain/test_generation_task.py new file mode 100644 index 000000000..edca75f5c --- /dev/null +++ b/tests/unit/domain/test_generation_task.py @@ -0,0 +1,208 @@ +"""generation_task 单测. + +domain 层生成任务实体纯逻辑模块,0 外部依赖。 +覆盖:枚举、create工厂/校验、列表拷贝。 +""" + +from __future__ import annotations + +from packages.domain.generation_task import GenerationTask, GenerationTaskStatus + + +class TestGenerationTaskStatus: + """GenerationTaskStatus 枚举测试.""" + + def test_five_statuses(self): + """五种状态.""" + assert len(GenerationTaskStatus) == 5 + + def test_pending(self): + assert GenerationTaskStatus.PENDING == "pending" + + def test_running(self): + assert GenerationTaskStatus.RUNNING == "running" + + def test_completed(self): + assert GenerationTaskStatus.COMPLETED == "completed" + + def test_failed(self): + assert GenerationTaskStatus.FAILED == "failed" + + def test_cancelled(self): + assert GenerationTaskStatus.CANCELLED == "cancelled" + + +class TestGenerationTaskCreate: + """GenerationTask.create 工厂测试.""" + + def test_create_minimal(self): + """最简创建(project_id + asset_library_id).""" + task = GenerationTask.create(project_id="proj1", asset_library_id="lib1") + assert task.project_id == "proj1" + assert task.asset_library_id == "lib1" + assert task.status == GenerationTaskStatus.PENDING + assert task.progress == 0.0 + assert task.result_count == 0 + assert task.error_message == "" + assert task.asset_ids == [] + assert task.title_ids == [] + assert task.voice_ids == [] + assert isinstance(task.id, str) + assert len(task.id) > 0 + + def test_create_with_all_fields(self): + """带全部字段创建.""" + task = GenerationTask.create( + project_id=" proj1 ", + asset_library_id=" lib1 ", + strategy_id=" strat1 ", + voice_library_id=" vlib1 ", + template_id=" tpl1 ", + asset_ids=["a1", "a2", "a3"], + title_ids=["t1", "t2"], + voice_ids=["v1"], + created_by_user_id=" user1 ", + source_edit_plan_id=" plan1 ", + asset_select_mode="random", + batch_id="batch1", + ) + assert task.project_id == "proj1" + assert task.asset_library_id == "lib1" + assert task.strategy_id == "strat1" + assert task.voice_library_id == "vlib1" + assert task.template_id == "tpl1" + assert task.asset_ids == ["a1", "a2", "a3"] + assert task.title_ids == ["t1", "t2"] + assert task.voice_ids == ["v1"] + assert task.created_by_user_id == "user1" + assert task.source_edit_plan_id == "plan1" + assert task.asset_select_mode == "random" + assert task.batch_id == "batch1" + + def test_create_with_template_instead_of_project(self): + """有 template_id 但 project_id 为空也可以.""" + task = GenerationTask.create( + project_id="", + asset_library_id="lib1", + template_id="tpl1", + ) + assert task.template_id == "tpl1" + assert task.project_id == "" + + def test_create_neither_project_nor_template(self): + """project_id 和 template_id 都为空,抛错.""" + try: + GenerationTask.create(project_id="", asset_library_id="lib1") + assert False + except ValueError as e: + assert "project_id" in str(e) and "template_id" in str(e) + + def test_create_whitespace_project_and_template(self): + """都是空白也抛错.""" + try: + GenerationTask.create(project_id=" ", asset_library_id="lib1", template_id=" ") + assert False + except ValueError as e: + assert "project_id" in str(e) and "template_id" in str(e) + + def test_create_no_asset_library_and_no_ids(self): + """asset_library_id 为空且没有素材列表,抛错.""" + try: + GenerationTask.create(project_id="p1", asset_library_id="") + assert False + except ValueError as e: + assert "asset_library_id" in str(e) + + def test_create_whitespace_asset_library_and_no_ids(self): + """空白 asset_library 且无素材列表,抛错.""" + try: + GenerationTask.create(project_id="p1", asset_library_id=" ") + assert False + except ValueError as e: + assert "asset_library_id" in str(e) + + def test_create_with_asset_ids_instead_of_library(self): + """用 asset_ids 替代 asset_library_id.""" + task = GenerationTask.create( + project_id="p1", + asset_library_id="", + asset_ids=["a1", "a2"], + ) + assert task.asset_library_id == "" + assert task.asset_ids == ["a1", "a2"] + + def test_create_with_title_ids_instead_of_library(self): + """用 title_ids 替代 asset_library_id.""" + task = GenerationTask.create( + project_id="p1", + asset_library_id="", + title_ids=["t1"], + ) + assert task.title_ids == ["t1"] + + def test_create_with_voice_ids_instead_of_library(self): + """用 voice_ids 替代 asset_library_id.""" + task = GenerationTask.create( + project_id="p1", + asset_library_id="", + voice_ids=["v1"], + ) + assert task.voice_ids == ["v1"] + + def test_create_asset_ids_copied(self): + """asset_ids 是拷贝不是引用.""" + original = ["a1", "a2"] + task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=original) + original.append("a3") + assert task.asset_ids == ["a1", "a2"] + + def test_create_title_ids_copied(self): + """title_ids 是拷贝不是引用.""" + original = ["t1"] + task = GenerationTask.create(project_id="p1", asset_library_id="lib1", title_ids=original) + original.append("t2") + assert task.title_ids == ["t1"] + + def test_create_voice_ids_copied(self): + """voice_ids 是拷贝不是引用.""" + original = ["v1"] + task = GenerationTask.create(project_id="p1", asset_library_id="lib1", voice_ids=original) + original.append("v2") + assert task.voice_ids == ["v1"] + + def test_create_none_lists_default_empty(self): + """None 列表默认为空.""" + task = GenerationTask.create( + project_id="p1", + asset_library_id="lib1", + asset_ids=None, + title_ids=None, + voice_ids=None, + ) + assert task.asset_ids == [] + assert task.title_ids == [] + assert task.voice_ids == [] + + def test_create_unique_id(self): + """不同任务 id 不同.""" + t1 = GenerationTask.create("p", "l") + t2 = GenerationTask.create("p", "l") + assert t1.id != t2.id + + def test_create_has_created_at(self): + """有创建时间.""" + task = GenerationTask.create("p", "l") + assert task.created_at is not None + + def test_create_defaults_started_completed_none(self): + """started_at 和 completed_at 默认 None.""" + task = GenerationTask.create("p", "l") + assert task.started_at is None + assert task.completed_at is None + + def test_empty_lists_independent(self): + """不同任务的空列表互不影响.""" + t1 = GenerationTask.create("p", "l") + t2 = GenerationTask.create("p", "l") + t1.asset_ids.append("x") + assert t2.asset_ids == []