"""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="smart", 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 == "smart" 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 == []