xiaoxia-saas/tests/unit/test_video_upload_fix_and_generated_api.py

"""
测试视频上传失败排查修复：
1. chunked_upload ALLOWED_MIME_TYPES 与 upload.py 保持一致
2. generated_video_repository list 方法无 N+1 查询
3. 成片库 API 端点可正常返回数据
"""

import json
from datetime import datetime, timezone
from unittest.mock import MagicMock, patch

# ─── 1. chunked_upload ALLOWED_MIME_TYPES 一致性 ───


class TestChunkedUploadMIMEConsistency:
    """chunked_upload.py 的 ALLOWED_MIME_TYPES 必须包含 upload.py 的所有类型。"""

    def _get_upload_mime_types(self) -> set:
        from apps.api.app.api.routes.upload import ALLOWED_MIME_TYPES

        return set(ALLOWED_MIME_TYPES)

    def _get_chunked_mime_types(self) -> set:
        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES

        return set(ALLOWED_MIME_TYPES)

    def test_chunked_upload_has_all_upload_mime_types(self):
        """chunked_upload 白名单必须覆盖 upload.py 的全部类型。"""
        upload_types = self._get_upload_mime_types()
        chunked_types = self._get_chunked_mime_types()
        missing = upload_types - chunked_types
        assert not missing, f"chunked_upload 缺少以下 MIME 类型: {missing}"

    def test_chunked_upload_supports_video_mpeg(self):
        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES

        assert "video/mpeg" in ALLOWED_MIME_TYPES

    def test_chunked_upload_supports_video_matroska(self):
        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES

        assert "video/x-matroska" in ALLOWED_MIME_TYPES

    def test_chunked_upload_supports_video_3gpp(self):
        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES

        assert "video/3gpp" in ALLOWED_MIME_TYPES

    def test_chunked_upload_supports_audio_flac(self):
        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES

        assert "audio/flac" in ALLOWED_MIME_TYPES

    def test_chunked_upload_supports_audio_aac(self):
        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES

        assert "audio/aac" in ALLOWED_MIME_TYPES

    def test_chunked_upload_supports_audio_m4a(self):
        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES

        assert "audio/x-m4a" in ALLOWED_MIME_TYPES

    def test_chunked_upload_supports_audio_webm(self):
        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES

        assert "audio/webm" in ALLOWED_MIME_TYPES

    def test_chunked_upload_supports_extra_image_types(self):
        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES

        assert "image/bmp" in ALLOWED_MIME_TYPES
        assert "image/tiff" in ALLOWED_MIME_TYPES
        assert "image/svg+xml" in ALLOWED_MIME_TYPES

    def test_both_have_same_core_video_types(self):
        """两条路径的核心视频类型必须一致。"""
        upload_types = self._get_upload_mime_types()
        chunked_types = self._get_chunked_mime_types()
        core_video = {"video/mp4", "video/quicktime", "video/webm"}
        for vt in core_video:
            assert vt in upload_types, f"upload.py 缺少 {vt}"
            assert vt in chunked_types, f"chunked_upload.py 缺少 {vt}"


# ─── 2. GeneratedVideo Repository N+1 修复验证 ───


class TestGeneratedVideoRepositoryNoNPlus1:
    """list_by_project 和 list_by_generation_task 应使用 _to_domain 而非 self.get。"""

    def _make_model(self, video_id: str, project_id: str = "proj-1", task_id: str = "task-1"):
        model = MagicMock()
        model.id = video_id
        model.project_id = project_id
        model.generation_task_id = task_id
        model.name = f"video-{video_id}.mp4"
        model.file_url = f"https://oss.example.com/{video_id}.mp4"
        model.file_size = 1024
        model.duration = 5.0
        model.thumbnail_url = None
        model.width = 1280
        model.height = 720
        model.fps = 25.0
        model.status = "completed"
        model.review_status = "pending_review"
        model.generation_params = json.dumps({"mode": "one_take"})
        model.video_fingerprint = None
        model.is_duplicate = False
        model.duplicate_of = None
        model.generated_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
        model.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
        return model

    def test_list_by_project_uses_to_domain(self):
        """list_by_project 不应调用 self.get（N+1），应使用 _to_domain。"""
        from packages.adapters.sqlalchemy_impl.generated_video_repository import (
            SQLAlchemyGeneratedVideoRepository,
        )

        session = MagicMock()
        models = [self._make_model(f"v{i}") for i in range(5)]

        query_mock = MagicMock()
        query_mock.filter.return_value.all.return_value = models
        session.query.return_value = query_mock

        repo = SQLAlchemyGeneratedVideoRepository(session)
        with patch.object(SQLAlchemyGeneratedVideoRepository, "get") as mock_get:
            result = repo.list_by_project("proj-1")
            mock_get.assert_not_called()

        assert len(result) == 5
        assert all(v.id.startswith("v") for v in result)

    def test_list_by_generation_task_uses_to_domain(self):
        """list_by_generation_task 不应调用 self.get（N+1），应使用 _to_domain。"""
        from packages.adapters.sqlalchemy_impl.generated_video_repository import (
            SQLAlchemyGeneratedVideoRepository,
        )

        session = MagicMock()
        models = [self._make_model(f"v{i}", task_id="task-42") for i in range(3)]

        query_mock = MagicMock()
        query_mock.filter.return_value.all.return_value = models
        session.query.return_value = query_mock

        repo = SQLAlchemyGeneratedVideoRepository(session)
        with patch.object(SQLAlchemyGeneratedVideoRepository, "get") as mock_get:
            result = repo.list_by_generation_task("task-42")
            mock_get.assert_not_called()

        assert len(result) == 3

    def test_list_by_project_returns_empty_when_no_videos(self):
        from packages.adapters.sqlalchemy_impl.generated_video_repository import (
            SQLAlchemyGeneratedVideoRepository,
        )

        session = MagicMock()
        query_mock = MagicMock()
        query_mock.filter.return_value.all.return_value = []
        session.query.return_value = query_mock

        repo = SQLAlchemyGeneratedVideoRepository(session)
        result = repo.list_by_project("empty-project")
        assert result == []


# ─── 3. 成片库 Schema 及替代端点可用性确认 ───


class TestGeneratedVideosSchemaAndAlternativeEndpoints:
    """确认成片库 Schema 字段完整，替代端点（generation_tasks）可用。"""

    def test_generated_video_response_schema_complete(self):
        """GeneratedVideoResponse 包含所有必要字段。"""
        from apps.api.app.schemas.generated_video import GeneratedVideoResponse

        fields = GeneratedVideoResponse.model_fields
        required_fields = [
            "id",
            "project_id",
            "name",
            "file_url",
            "status",
            "review_status",
            "download_url",
        ]
        for field in required_fields:
            assert field in fields, f"GeneratedVideoResponse 缺少字段: {field}"

    def test_list_generated_videos_response_schema(self):
        """ListGeneratedVideosResponse 包含 items 列表。"""
        from apps.api.app.schemas.generated_video import ListGeneratedVideosResponse

        fields = ListGeneratedVideosResponse.model_fields
        assert "items" in fields, "ListGeneratedVideosResponse 缺少 items 字段"

    def test_generation_task_results_endpoint_exists(self):
        """GET /generation/tasks/{task_id}/results 端点存在。"""
        from apps.api.app.api.routes.generation_tasks import router

        paths = [r.path for r in router.routes if hasattr(r, "path")]
        assert "/tasks/{task_id}/results" in paths, "generation results 端点不存在"


# ─── 4. GeneratedVideo Use Cases 可用性 ───


class TestGeneratedVideoUseCases:
    """确认成片库 Use Case 层可正常工作。"""

    def _make_video(self, video_id: str, project_id: str = "proj-1"):
        from packages.domain import GeneratedVideo

        return GeneratedVideo(
            id=video_id,
            project_id=project_id,
            generation_task_id="task-1",
            name=f"video-{video_id}.mp4",
            file_url=f"https://oss.example.com/{video_id}.mp4",
            file_size=1024,
            duration=5.0,
            width=1280,
            height=720,
            fps=25.0,
            status="completed",
        )

    def test_list_generated_videos_use_case(self):
        """ListGeneratedVideosUseCase 可正常列出视频。"""
        from packages.application import ListGeneratedVideosUseCase

        repo = MagicMock()
        videos = [self._make_video(f"v{i}") for i in range(3)]
        repo.list_by_project.return_value = videos

        use_case = ListGeneratedVideosUseCase(repo)
        result = use_case.execute("proj-1")

        repo.list_by_project.assert_called_once_with("proj-1")
        assert len(result) == 3

    def test_get_generated_video_use_case(self):
        """GetGeneratedVideoUseCase 可正常获取单个视频。"""
        from packages.application import GetGeneratedVideoUseCase

        repo = MagicMock()
        video = self._make_video("v1")
        repo.get.return_value = video

        use_case = GetGeneratedVideoUseCase(repo)
        result = use_case.execute("v1")

        repo.get.assert_called_once_with("v1")
        assert result.id == "v1"

    def test_list_by_task_use_case(self):
        """ListGeneratedVideosByTaskUseCase 可按任务列出视频。"""
        from packages.application import ListGeneratedVideosByTaskUseCase

        repo = MagicMock()
        videos = [self._make_video(f"v{i}") for i in range(2)]
        repo.list_by_generation_task.return_value = videos

        use_case = ListGeneratedVideosByTaskUseCase(repo)
        result = use_case.execute("task-1")

        repo.list_by_generation_task.assert_called_once_with("task-1")
        assert len(result) == 2