209 lines
7.3 KiB
Python
209 lines
7.3 KiB
Python
"""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 == []
|