xiaoxia-saas/tests/unit/test_1294_preview_voice_injection.py

"""测试 #1294 修复：预览视频配音注入（#1749 后重写）。

#1749 起冗余 voice_ids 字段已从 task_info 移除（DB 列保留只读），
配音一律以 voice_library_id 为准（批量独立配音每变体各自绑定）。
本测试断言：
1. _load_task_info 正确加载 voice_library_id；
2. task_info 不再包含 voice_ids 键；
3. 配音解析 effective_voice_id 直接取 voice_library_id，不再有 [0] 兜底。
"""

from unittest.mock import MagicMock, patch


def _mock_task(voice_library_id: str = "voice_lib_1"):
    mock_task = MagicMock()
    mock_task.project_id = "proj_1"
    mock_task.asset_library_id = "lib_1"
    mock_task.voice_library_id = voice_library_id
    mock_task.template_id = "tmpl_1"
    mock_task.strategy_id = "one_take"
    mock_task.asset_ids = ["a1", "a2"]
    mock_task.batch_id = "batch_1"
    mock_task.created_by_user_id = "user_1"
    mock_task.video_title = "test"
    mock_task.resolution = "854x480"
    mock_task.bgm_config = {}
    mock_task.is_preview = True
    mock_task.voice_ids = ["voice_1", "voice_2"]  # DB 列仍存在但不再读
    mock_task.source_task_id = ""
    mock_task.output_width = 854
    mock_task.output_height = 480
    mock_task.cover_url = ""
    mock_task.title_config = {}
    mock_task.source_edit_plan_id = "plan_1"
    return mock_task


def test_voice_library_id_loaded_from_task():
    with patch(
        "packages.adapters.sqlalchemy_impl.generation_task_repository.SQLAlchemyGenerationTaskRepository"
    ) as MockRepo:
        mock_repo = MagicMock()
        mock_repo.get.return_value = _mock_task("voice_lib_abc")
        MockRepo.return_value = mock_repo

        from worker_app.tasks.generation import _load_task_info

        result = _load_task_info("test_task_id")

        assert result is not None
        assert result["voice_library_id"] == "voice_lib_abc"


def test_task_info_has_no_voice_ids_key():
    """#1749：冗余 voice_ids 已从 task_info 移除。"""
    with patch(
        "packages.adapters.sqlalchemy_impl.generation_task_repository.SQLAlchemyGenerationTaskRepository"
    ) as MockRepo:
        mock_repo = MagicMock()
        mock_repo.get.return_value = _mock_task()
        MockRepo.return_value = mock_repo

        from worker_app.tasks.generation import _load_task_info

        result = _load_task_info("test_task_id")
        assert "voice_ids" not in result


def test_effective_voice_uses_voice_library_id_only():
    """配音解析：voice_library_id 即最终配音 ID，无 voice_ids[0] 兜底。"""
    # 模拟 _sync_task_config_to_plan 中的解析逻辑
    task_info = {"voice_library_id": "ANDT_voice"}
    voice_library_id = task_info.get("voice_library_id", "")
    effective_voice_id = voice_library_id or ""
    assert effective_voice_id == "ANDT_voice"

    # 空配音
    task_info2 = {"voice_library_id": ""}
    assert (task_info2.get("voice_library_id", "") or "") == ""


def test_empty_voice_library_id():
    with patch(
        "packages.adapters.sqlalchemy_impl.generation_task_repository.SQLAlchemyGenerationTaskRepository"
    ) as MockRepo:
        mock_repo = MagicMock()
        mock_repo.get.return_value = _mock_task("")
        MockRepo.return_value = mock_repo

        from worker_app.tasks.generation import _load_task_info

        result = _load_task_info("test_task_id")
        assert result["voice_library_id"] == ""