467 lines
19 KiB
Python
Executable File
467 lines
19 KiB
Python
Executable File
"""第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
|