""" GenerationTask 领域模型单元测试 """ import json import time from datetime import datetime, timezone import pytest from packages.domain.generation_task import ( TERMINAL_STATUSES, GenerationTask, GenerationTaskStatus, ) class TestGenerationTaskStatus: """GenerationTaskStatus 枚举测试""" def test_status_values(self): assert GenerationTaskStatus.PENDING == "pending" assert GenerationTaskStatus.RUNNING == "running" assert GenerationTaskStatus.COMPLETED == "completed" assert GenerationTaskStatus.FAILED == "failed" assert GenerationTaskStatus.CANCELLED == "cancelled" def test_terminal_statuses(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 TestGenerationTaskCreate: """GenerationTask 创建测试""" def test_create_basic(self): """测试基本创建""" task = GenerationTask.create( project_id="proj-123", asset_library_id="lib-456", ) assert task.id is not None assert task.project_id == "proj-123" assert task.asset_library_id == "lib-456" 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.logs == "[]" assert task.created_at is not None assert task.started_at is None assert task.completed_at is None def test_create_with_all_params(self): """测试创建带所有参数""" task = GenerationTask.create( project_id="proj-123", asset_library_id="lib-456", strategy_id="strat-789", voice_library_id="voice-lib-001", template_id="tmpl-001", asset_ids=["asset-1", "asset-2"], title_ids=["title-1"], voice_ids=["voice-1"], created_by_user_id="user-001", source_edit_plan_id="plan-001", asset_select_mode="smart", batch_id="batch-001", video_title="测试视频", resolution="1080x1920", auto_retry_enabled=True, auto_retry_max=3, ) assert task.strategy_id == "strat-789" assert task.voice_library_id == "voice-lib-001" assert task.template_id == "tmpl-001" assert task.asset_ids == ["asset-1", "asset-2"] assert task.title_ids == ["title-1"] assert task.voice_ids == ["voice-1"] assert task.created_by_user_id == "user-001" assert task.source_edit_plan_id == "plan-001" assert task.asset_select_mode == "smart" assert task.batch_id == "batch-001" assert task.video_title == "测试视频" assert task.resolution == "1080x1920" assert task.auto_retry_enabled is True assert task.auto_retry_max == 3 def test_create_with_bgm_config(self): """测试创建带自定义BGM配置""" bgm_cfg = { "enabled": True, "source": "asset", "asset_id": "bgm-asset-001", "volume": 0.6, } task = GenerationTask.create( project_id="", asset_library_id="", template_id="tmpl-001", asset_ids=["asset-1"], bgm_config=bgm_cfg, ) assert task.bgm_config == bgm_cfg # 确保是副本,不是引用 bgm_cfg["volume"] = 0.8 assert task.bgm_config["volume"] == 0.6 def test_create_default_bgm_config_empty(self): """测试默认BGM配置为空dict""" task = GenerationTask.create( project_id="", asset_library_id="", template_id="tmpl-001", asset_ids=["asset-1"], ) assert task.bgm_config == {} assert isinstance(task.bgm_config, dict) def test_create_with_template_id_only(self): """测试只提供 template_id 不提供 project_id(应通过校验)""" task = GenerationTask.create( project_id="", asset_library_id="lib-456", template_id="tmpl-001", ) assert task.project_id == "" assert task.template_id == "tmpl-001" def test_create_with_asset_ids_only(self): """测试只提供 asset_ids 不提供 asset_library_id(应通过校验)""" task = GenerationTask.create( project_id="proj-123", asset_library_id="", asset_ids=["asset-1", "asset-2"], ) assert task.asset_library_id == "" assert task.asset_ids == ["asset-1", "asset-2"] def test_create_missing_project_and_template(self): """测试 project_id 和 template_id 都为空""" with pytest.raises(ValueError, match="至少需要提供一个"): GenerationTask.create( project_id="", asset_library_id="lib-456", ) def test_create_missing_library_and_assets(self): """测试 asset_library_id 和素材列表都为空""" with pytest.raises(ValueError, match="至少需要提供一个"): GenerationTask.create( project_id="proj-123", asset_library_id="", ) def test_create_strips_strings(self): """测试字符串字段被 strip""" task = GenerationTask.create( project_id=" proj-123 ", asset_library_id=" lib-456 ", strategy_id=" strat-789 ", template_id=" tmpl-001 ", video_title=" 测试视频 ", resolution=" 1080x1920 ", ) assert task.project_id == "proj-123" assert task.asset_library_id == "lib-456" assert task.strategy_id == "strat-789" assert task.template_id == "tmpl-001" assert task.video_title == "测试视频" assert task.resolution == "1080x1920" def test_create_default_empty_lists(self): """测试 None 列表默认化为空列表""" task = GenerationTask.create( project_id="proj-123", asset_library_id="lib-456", asset_ids=None, title_ids=None, voice_ids=None, ) assert task.asset_ids == [] assert task.title_ids == [] assert task.voice_ids == [] def test_create_copies_list(self): """测试列表被复制(不共享引用)""" assets = ["a", "b"] task = GenerationTask.create( project_id="proj-123", asset_library_id="lib-456", asset_ids=assets, ) assets.append("c") assert task.asset_ids == ["a", "b"] class TestGenerationTaskStatusProperties: """状态属性测试""" @pytest.fixture def new_task(self): return GenerationTask.create( project_id="proj-123", asset_library_id="lib-456", ) def test_is_terminal_pending(self, new_task): assert not new_task.is_terminal def test_is_terminal_running(self, new_task): new_task.mark_processing() assert not new_task.is_terminal def test_is_terminal_completed(self, new_task): new_task.mark_processing() new_task.mark_completed() assert new_task.is_terminal def test_is_terminal_failed(self, new_task): new_task.mark_processing() new_task.mark_failed("error") assert new_task.is_terminal def test_is_terminal_cancelled(self, new_task): new_task.mark_cancelled() assert new_task.is_terminal def test_is_completed(self, new_task): assert not new_task.is_completed new_task.mark_processing() assert not new_task.is_completed new_task.mark_completed() assert new_task.is_completed def test_is_failed(self, new_task): assert not new_task.is_failed new_task.mark_processing() new_task.mark_failed("error") assert new_task.is_failed def test_is_running(self, new_task): assert not new_task.is_running new_task.mark_processing() assert new_task.is_running class TestGenerationTaskTransitions: """状态转换测试""" @pytest.fixture def new_task(self): return GenerationTask.create( project_id="proj-123", asset_library_id="lib-456", ) # ===== Pending → Running ===== def test_pending_to_running(self, new_task): """测试 pending → running""" new_task.mark_processing() assert new_task.status == GenerationTaskStatus.RUNNING assert new_task.started_at is not None assert new_task.error_message == "" def test_pending_to_running_clears_error(self, new_task): """测试 processing 清除错误信息""" # pending 可以直接 failed(跟 job 不同) new_task.mark_failed("some error") # 然后重试回 pending new_task.mark_pending_from_failed() new_task.mark_processing() assert new_task.error_message == "" # ===== Pending → Failed ===== def test_pending_to_failed(self, new_task): """测试 pending → failed(可以直接失败)""" new_task.mark_failed("task failed before start") assert new_task.status == GenerationTaskStatus.FAILED assert new_task.error_message == "task failed before start" assert new_task.completed_at is not None def test_pending_to_failed_with_error_info(self, new_task): """测试 pending → failed 带 error_info""" error_info = {"error_type": "ValidationError", "stage": "init"} new_task.mark_failed("validation failed", error_info=error_info) assert new_task.error_info == error_info def test_pending_to_failed_default_error_info(self, new_task): """测试 pending → failed 默认 error_info""" new_task.mark_failed("some error") assert new_task.error_info["error_type"] == "UnknownError" assert new_task.error_info["message"] == "some error" assert "failed_at" in new_task.error_info # ===== Pending → Cancelled ===== def test_pending_to_cancelled(self, new_task): """测试 pending → cancelled""" new_task.mark_cancelled() assert new_task.status == GenerationTaskStatus.CANCELLED assert new_task.completed_at is not None # ===== Running → Completed ===== def test_running_to_completed(self, new_task): """测试 running → completed""" new_task.mark_processing() new_task.mark_completed(result_count=3) assert new_task.status == GenerationTaskStatus.COMPLETED assert new_task.progress == 100.0 assert new_task.result_count == 3 assert new_task.error_message == "" assert new_task.completed_at is not None def test_running_to_completed_default_count(self, new_task): """测试 running → completed 默认 result_count=1""" new_task.mark_processing() new_task.mark_completed() assert new_task.result_count == 1 # ===== Running → Failed ===== def test_running_to_failed(self, new_task): """测试 running → failed""" new_task.mark_processing() new_task.mark_failed("render timeout") assert new_task.status == GenerationTaskStatus.FAILED assert new_task.error_message == "render timeout" assert new_task.completed_at is not None # ===== Running → Cancelled ===== def test_running_to_cancelled(self, new_task): """测试 running → cancelled""" new_task.mark_processing() new_task.mark_cancelled() assert new_task.status == GenerationTaskStatus.CANCELLED assert new_task.completed_at is not None # ===== Failed → Pending (重试) ===== def test_failed_to_pending_retry(self, new_task): """测试 failed → pending(重试)""" new_task.mark_processing() new_task.mark_failed("error", error_info={"error_type": "TimeoutError"}) assert new_task.retry_count == 0 new_task.mark_pending_from_failed() assert new_task.status == GenerationTaskStatus.PENDING assert new_task.retry_count == 1 assert new_task.error_message == "" assert new_task.error_info == {} assert new_task.started_at is None assert new_task.completed_at is None assert new_task.progress == 0.0 assert new_task.result_count == 0 def test_failed_to_pending_multiple_retries(self, new_task): """测试多次重试""" for i in range(3): new_task.mark_processing() new_task.mark_failed(f"error {i}") new_task.mark_pending_from_failed() assert new_task.retry_count == 3 assert new_task.status == GenerationTaskStatus.PENDING def test_mark_pending_from_failed_wrong_status(self, new_task): """测试非 failed 状态调用 mark_pending_from_failed 报错""" with pytest.raises(ValueError, match="只有 failed 状态"): new_task.mark_pending_from_failed() # pending 状态 # ===== 非法状态转换 ===== def test_invalid_completed_to_running(self, new_task): """测试 completed → running 非法""" new_task.mark_processing() new_task.mark_completed() with pytest.raises(ValueError, match="非法状态转换"): new_task.mark_processing() def test_invalid_cancelled_to_running(self, new_task): """测试 cancelled → running 非法""" new_task.mark_cancelled() with pytest.raises(ValueError, match="非法状态转换"): new_task.mark_processing() def test_invalid_completed_to_failed(self, new_task): """测试 completed → failed 非法""" new_task.mark_processing() new_task.mark_completed() with pytest.raises(ValueError, match="非法状态转换"): new_task.mark_failed("error") def test_invalid_status_string(self, new_task): """测试无效状态字符串""" with pytest.raises(ValueError, match="非法状态转换"): new_task.transition_to("invalid_status") class TestGenerationTaskLogs: """日志系统测试""" @pytest.fixture def new_task(self): return GenerationTask.create( project_id="proj-123", asset_library_id="lib-456", ) def test_initial_logs_empty(self, new_task): """测试初始日志为空""" assert new_task.get_logs() == [] assert new_task.logs == "[]" def test_append_log(self, new_task): """测试追加日志""" new_task.append_log(stage="下载素材", message="开始下载") logs = new_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_custom_level(self, new_task): """测试带自定义日志级别""" new_task.append_log(stage="渲染", message="渲染失败", level="ERROR") logs = new_task.get_logs() assert logs[0]["level"] == "ERROR" def test_append_log_with_extra_fields(self, new_task): """测试带额外字段的日志""" new_task.append_log( stage="下载", message="下载完成", asset_id="asset-001", duration=30.5, ) logs = new_task.get_logs() assert logs[0]["asset_id"] == "asset-001" assert logs[0]["duration"] == 30.5 def test_multiple_logs(self, new_task): """测试多条日志""" new_task.append_log(stage="步骤1", message="开始") new_task.append_log(stage="步骤2", message="进行中") new_task.append_log(stage="步骤3", message="完成") logs = new_task.get_logs() assert len(logs) == 3 assert logs[0]["stage"] == "步骤1" assert logs[2]["stage"] == "步骤3" def test_logs_corrupted_json(self, new_task): """测试 logs 字段损坏时仍能正常工作""" new_task.logs = "not valid json {{{" logs = new_task.get_logs() assert logs == [] # 追加新日志应该能正常工作 new_task.append_log(stage="test", message="after corruption") logs = new_task.get_logs() assert len(logs) == 1 assert logs[0]["message"] == "after corruption" def test_logs_max_limit(self, new_task): """测试日志数量上限""" for i in range(250): new_task.append_log(stage="loop", message=f"log {i}") logs = new_task.get_logs() assert len(logs) == 200 # _MAX_LOGS # 应该保留最近的 200 条 assert logs[0]["message"] == "log 50" assert logs[-1]["message"] == "log 249" def test_logs_empty_string(self, new_task): """测试 logs 为空字符串时返回空列表""" new_task.logs = "" assert new_task.get_logs() == [] def test_append_log_preserves_existing(self, new_task): """测试追加日志保留已有日志""" new_task.append_log(stage="first", message="first message") new_task.append_log(stage="second", message="second message") logs = new_task.get_logs() assert len(logs) == 2 assert logs[0]["message"] == "first message" assert logs[1]["message"] == "second message" class TestGenerationTaskTimestamps: """时间戳测试""" @pytest.fixture def new_task(self): return GenerationTask.create( project_id="proj-123", asset_library_id="lib-456", ) def test_created_at_set(self, new_task): assert new_task.created_at is not None assert isinstance(new_task.created_at, datetime) assert new_task.created_at.tzinfo is not None def test_mark_processing_sets_started_at(self, new_task): assert new_task.started_at is None before = datetime.now(timezone.utc) time.sleep(0.01) new_task.mark_processing() time.sleep(0.01) after = datetime.now(timezone.utc) assert before < new_task.started_at < after def test_mark_completed_sets_completed_at(self, new_task): new_task.mark_processing() assert new_task.completed_at is None new_task.mark_completed() assert new_task.completed_at is not None def test_mark_failed_sets_completed_at(self, new_task): new_task.mark_processing() assert new_task.completed_at is None new_task.mark_failed("error") assert new_task.completed_at is not None def test_retry_clears_timestamps(self, new_task): new_task.mark_processing() new_task.mark_failed("error") new_task.mark_pending_from_failed() assert new_task.started_at is None assert new_task.completed_at is None