xiaoxia-saas/tests/unit/test_generation_task.py

"""GenerationTask 生成任务领域模型单测."""

import pytest

from packages.domain.generation_task import (
    TERMINAL_STATUSES,
    GenerationTask,
    GenerationTaskStatus,
)


class TestGenerationTaskCreate:
    def test_create_with_template_id(self):
        task = GenerationTask.create(
            project_id="",
            asset_library_id="lib1",
            template_id="tmpl1",
        )
        assert task.id
        assert task.template_id == "tmpl1"
        assert task.asset_library_id == "lib1"
        assert task.status == GenerationTaskStatus.PENDING
        assert task.progress == 0.0
        assert task.retry_count == 0
        assert task.auto_retry_enabled is False

    def test_create_with_project_id(self):
        task = GenerationTask.create(
            project_id="proj1",
            asset_library_id="lib1",
        )
        assert task.project_id == "proj1"

    def test_create_both_empty_raises(self):
        with pytest.raises(ValueError, match="project_id 或 template_id 至少需要提供一个"):
            GenerationTask.create(
                project_id="",
                asset_library_id="lib1",
                template_id="",
            )

    def test_create_asset_library_and_assets_both_empty_raises(self):
        with pytest.raises(ValueError, match="asset_library_id 或 asset_ids/title_ids/voice_ids 至少需要提供一个"):
            GenerationTask.create(
                project_id="proj1",
                asset_library_id="",
                asset_ids=None,
                title_ids=None,
                voice_ids=None,
            )

    def test_create_with_asset_ids(self):
        task = GenerationTask.create(
            project_id="proj1",
            asset_library_id="",
            asset_ids=["a1", "a2"],
        )
        assert task.asset_ids == ["a1", "a2"]

    def test_create_strips_whitespace(self):
        task = GenerationTask.create(
            project_id="  proj1  ",
            asset_library_id="  lib1  ",
            template_id="  tmpl1  ",
            video_title="  测试视频  ",
            resolution="  1080p  ",
            created_by_user_id="  user1  ",
            source_edit_plan_id="  plan1  ",
        )
        assert task.project_id == "proj1"
        assert task.asset_library_id == "lib1"
        assert task.template_id == "tmpl1"
        assert task.video_title == "测试视频"
        assert task.resolution == "1080p"
        assert task.created_by_user_id == "user1"
        assert task.source_edit_plan_id == "plan1"

    def test_create_default_values(self):
        task = GenerationTask.create(
            project_id="proj1",
            asset_library_id="lib1",
        )
        assert task.title_ids == []
        assert task.voice_ids == []
        assert task.result_count == 0
        assert task.error_message == ""
        assert task.error_info == {}
        assert task.bgm_config == {}
        assert task.logs == "[]"

    def test_create_unique_ids(self):
        t1 = GenerationTask.create(project_id="p1", asset_library_id="l1")
        t2 = GenerationTask.create(project_id="p1", asset_library_id="l1")
        assert t1.id != t2.id


class TestGenerationTaskStatusQueries:
    def test_is_terminal_completed(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        task.mark_processing()
        task.mark_completed()
        assert task.is_terminal is True
        assert task.is_completed is True
        assert task.is_failed is False
        assert task.is_running is False

    def test_is_terminal_failed(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        task.mark_processing()
        task.mark_failed("error")
        assert task.is_terminal is True
        assert task.is_completed is False
        assert task.is_failed is True

    def test_is_terminal_cancelled(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        task.transition_to(GenerationTaskStatus.CANCELLED)
        assert task.is_terminal is True

    def test_is_running(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        assert task.is_running is False
        task.mark_processing()
        assert task.is_running is True

    def test_terminal_statuses_set(self):
        assert GenerationTaskStatus.COMPLETED in TERMINAL_STATUSES
        assert GenerationTaskStatus.FAILED in TERMINAL_STATUSES
        assert GenerationTaskStatus.CANCELLED in TERMINAL_STATUSES
        assert GenerationTaskStatus.PENDING not in TERMINAL_STATUSES
        assert GenerationTaskStatus.RUNNING not in TERMINAL_STATUSES


class TestGenerationTaskStateTransitions:
    def test_pending_to_running(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        task.mark_processing()
        assert task.status == GenerationTaskStatus.RUNNING
        assert task.started_at is not None
        assert task.error_message == ""

    def test_running_to_completed(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        task.mark_processing()
        task.mark_completed(result_count=3)
        assert task.status == GenerationTaskStatus.COMPLETED
        assert task.completed_at is not None
        assert task.progress == 100.0
        assert task.result_count == 3
        assert task.error_message == ""

    def test_running_to_failed_with_error_info(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        task.mark_processing()
        task.mark_failed("渲染失败", error_info={"stage": "render"})
        assert task.status == GenerationTaskStatus.FAILED
        assert task.completed_at is not None
        assert task.error_message == "渲染失败"
        assert task.error_info["stage"] == "render"

    def test_running_to_failed_without_error_info(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        task.mark_processing()
        task.mark_failed("未知错误")
        assert task.error_info is not None
        assert task.error_info["message"] == "未知错误"
        assert "error_type" in task.error_info
        assert "failed_at" in task.error_info

    def test_failed_to_pending_retry(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        task.mark_processing()
        task.mark_failed("error")
        assert task.retry_count == 0

        task.mark_pending_from_failed()
        assert task.status == GenerationTaskStatus.PENDING
        assert task.retry_count == 1
        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

    def test_mark_pending_from_failed_wrong_status_raises(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        with pytest.raises(ValueError, match="只有 failed 状态的任务可以重置为 pending"):
            task.mark_pending_from_failed()

    def test_invalid_transition_raises(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        # pending 不能直接到 completed
        with pytest.raises(ValueError, match="非法状态转换"):
            task.transition_to(GenerationTaskStatus.COMPLETED)

    def test_completed_cannot_transition(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        task.mark_processing()
        task.mark_completed()
        with pytest.raises(ValueError, match="非法状态转换"):
            task.mark_failed("test")

    def test_transition_to_with_string(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        task.transition_to("running")
        assert task.status == GenerationTaskStatus.RUNNING

    def test_transition_to_invalid_string_raises(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        with pytest.raises(ValueError, match="非法状态转换"):
            task.transition_to("invalid_status")


class TestGenerationTaskLogs:
    def test_append_log_single(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        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_multiple(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        for i in range(5):
            task.append_log(f"stage{i}", f"msg{i}", level="INFO")
        logs = task.get_logs()
        assert len(logs) == 5
        assert logs[0]["stage"] == "stage0"
        assert logs[4]["stage"] == "stage4"

    def test_append_log_with_extra_fields(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        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_log_error_level(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        task.append_log("渲染", "渲染失败", level="ERROR")
        logs = task.get_logs()
        assert logs[0]["level"] == "ERROR"

    def test_logs_max_limit(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        # _MAX_LOGS = 200
        for i in range(250):
            task.append_log("test", f"msg{i}")
        logs = task.get_logs()
        assert len(logs) == 200
        # 保留最新的200条
        assert logs[0]["message"] == "msg50"
        assert logs[-1]["message"] == "msg249"

    def test_get_logs_empty(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        assert task.get_logs() == []

    def test_get_logs_corrupted_json(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        task.logs = "not json"
        assert task.get_logs() == []

    def test_mark_failed_without_error_info(self):
        task = GenerationTask.create(project_id="p1", asset_library_id="l1")
        task.mark_processing()
        task.mark_failed("error msg")
        assert task.error_info is not None
        assert task.error_info["message"] == "error msg"
        assert "error_type" in task.error_info
        assert "failed_at" in task.error_info


class TestGenerationTaskCreateWithStrategy:
    def test_create_with_strategy_and_voice(self):
        task = GenerationTask.create(
            project_id="p1",
            asset_library_id="l1",
            strategy_id="s1",
            voice_library_id="v1",
            auto_retry_enabled=True,
            auto_retry_max=3,
        )
        assert task.strategy_id == "s1"
        assert task.voice_library_id == "v1"
        assert task.auto_retry_enabled is True
        assert task.auto_retry_max == 3

    def test_create_with_bgm_config(self):
        bgm = {"volume": 0.5, "track": "bgm1"}
        task = GenerationTask.create(
            project_id="p1",
            asset_library_id="l1",
            bgm_config=bgm,
        )
        assert task.bgm_config == bgm