xiaoxia-saas/tests/unit/test_cover_awaiting_fallback.py

"""#2028: generation_cover._get_task_video_url 兜底逻辑测试。"""

from __future__ import annotations

import os
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))

import pytest


class TestGetTaskVideoUrlAwaitingCoverFallback:
    def test_returns_url_from_rendered_output_when_awaiting_cover(self):
        from app.api.routes import generation_cover as cover_mod

        mock_usecase = MagicMock()
        mock_usecase.execute.return_value = []

        mock_task = MagicMock()
        mock_task.status.value = "awaiting_cover"
        mock_task.extra_meta = {"rendered_output": {"file_url": "oss://generated/awaiting.mp4"}}

        mock_task_repo_instance = MagicMock()
        mock_task_repo_instance.get.return_value = mock_task
        mock_db = MagicMock()

        with (
            patch.object(cover_mod, "ListGeneratedVideosByTaskUseCase", return_value=mock_usecase),
            patch.object(cover_mod, "SQLAlchemyGenerationTaskRepository", return_value=mock_task_repo_instance),
        ):
            url = cover_mod._get_task_video_url("task-aw1", mock_db)
        assert url == "oss://generated/awaiting.mp4"

    def test_returns_none_when_status_not_awaiting_cover(self):
        from app.api.routes import generation_cover as cover_mod

        mock_usecase = MagicMock()
        mock_usecase.execute.return_value = []
        mock_task = MagicMock()
        mock_task.status.value = "running"
        mock_task.extra_meta = {"rendered_output": {"file_url": "oss://x.mp4"}}
        mock_task_repo_instance = MagicMock()
        mock_task_repo_instance.get.return_value = mock_task
        mock_db = MagicMock()
        with (
            patch.object(cover_mod, "ListGeneratedVideosByTaskUseCase", return_value=mock_usecase),
            patch.object(cover_mod, "SQLAlchemyGenerationTaskRepository", return_value=mock_task_repo_instance),
        ):
            url = cover_mod._get_task_video_url("task-run", mock_db)
        assert url is None

    def test_returns_none_when_rendered_output_missing(self):
        from app.api.routes import generation_cover as cover_mod

        mock_usecase = MagicMock()
        mock_usecase.execute.return_value = []
        mock_task = MagicMock()
        mock_task.status.value = "awaiting_cover"
        mock_task.extra_meta = {}
        mock_task_repo_instance = MagicMock()
        mock_task_repo_instance.get.return_value = mock_task
        mock_db = MagicMock()
        with (
            patch.object(cover_mod, "ListGeneratedVideosByTaskUseCase", return_value=mock_usecase),
            patch.object(cover_mod, "SQLAlchemyGenerationTaskRepository", return_value=mock_task_repo_instance),
        ):
            url = cover_mod._get_task_video_url("task-empty", mock_db)
        assert url is None


class TestAwaitingCoverResultsSynthesis:
    """list_generation_results 在 awaiting_cover + 无 GeneratedVideo 时合成预览响应。"""

    def _invoke(self, task, storage_service):
        from app.api.routes import generation_tasks as gt_mod

        mock_auth = MagicMock()
        mock_auth.user.id = "u1"
        mock_task_repo = MagicMock()
        mock_task_repo.get.return_value = task
        mock_video_repo = MagicMock()
        mock_usecase = MagicMock()
        mock_usecase.execute.return_value = []
        mock_project_repo = MagicMock()

        with (
            patch.object(gt_mod, "check_project_access", return_value=None),
            patch.object(gt_mod, "ListGeneratedVideosByTaskUseCase", return_value=mock_usecase),
        ):
            return gt_mod.list_generation_results(
                task_id=task.id,
                authenticated_user=mock_auth,
                generation_task_repository=mock_task_repo,
                generated_video_repository=mock_video_repo,
                project_repository=mock_project_repo,
                storage_service=storage_service,
            )

    def test_synthesizes_preview_response_when_awaiting_cover(self):
        task = MagicMock()
        task.id = "task-syn-1"
        task.project_id = "proj1"
        task.status.value = "awaiting_cover"
        task.cover_url = ""
        task.extra_meta = {
            "rendered_output": {
                "file_url": "oss://bucket/v.mp4",
                "file_size": 2048,
                "duration": 10.5,
                "width": 1080,
                "height": 1920,
                "fps": 30.0,
                "name": "合成预览",
                "mode": "random",
                "thumbnail_url": "https://cdn/t.jpg",
            }
        }
        task.updated_at = None
        task.created_at = None
        storage = MagicMock()
        storage.get_download_url.return_value = "https://signed/v.mp4"
        resp = self._invoke(task, storage)
        assert len(resp.items) == 1
        item = resp.items[0]
        assert item.id == "preview-task-syn-1"
        assert item.name == "合成预览"
        assert item.file_size == 2048
        assert item.duration == 10.5
        assert item.width == 1080
        assert item.height == 1920
        assert item.fps == 30.0
        assert item.download_url == "https://signed/v.mp4"

    def test_http_file_url_used_directly(self):
        task = MagicMock()
        task.id = "task-httpx"
        task.project_id = "p"
        task.status.value = "awaiting_cover"
        task.cover_url = ""
        task.extra_meta = {"rendered_output": {"file_url": "https://cdn.example.com/v.mp4", "name": ""}}
        task.updated_at = None
        task.created_at = None
        storage = MagicMock()
        resp = self._invoke(task, storage)
        assert resp.items[0].download_url == "https://cdn.example.com/v.mp4"
        storage.get_download_url.assert_not_called()
        assert resp.items[0].name.startswith("generated-task-htt")

    def test_no_items_when_status_completed_without_videos(self):
        task = MagicMock()
        task.id = "task-done"
        task.project_id = "p"
        task.status.value = "completed"
        task.extra_meta = {"rendered_output": {"file_url": "oss://x.mp4"}}
        storage = MagicMock()
        resp = self._invoke(task, storage)
        assert resp.items == []


if __name__ == "__main__":
    sys.exit(pytest.main([__file__, "-v"]))