From f7a945d417f693c5daf06e86b0756ea21e55d1b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=81=B5=E5=BA=94?= Date: Tue, 7 Jul 2026 19:05:07 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E8=A7=86=E9=A2=91=E4=B8=8A=E4=BC=A0MIME?= =?UTF-8?q?=E7=99=BD=E5=90=8D=E5=8D=95=E8=A1=A5=E5=85=A8=20+=20=E6=88=90?= =?UTF-8?q?=E7=89=87=E5=BA=93N+1=E6=9F=A5=E8=AF=A2=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - chunked_upload.py ALLOWED_MIME_TYPES 从12种扩展到22种，与 upload.py 保持一致 新增: video/mpeg, video/x-matroska, video/3gpp, audio/flac, audio/aac, audio/x-m4a, audio/webm, image/bmp, image/tiff, image/svg+xml - generated_video_repository list_by_project/list_by_generation_task 改用 _to_domain() 替代 self.get()，消除 N+1 查询 - 新增 24 个单元测试覆盖以上修复及成片库 API 可用性确认 Closes #172 #173 Co-Authored-By: Claude Fable 5 --- apps/api/app/api/routes/chunked_upload.py | 15 +- .../generated_video_repository.py | 4 +- ...test_video_upload_fix_and_generated_api.py | 310 ++++++++++++++++++ 3 files changed, 326 insertions(+), 3 deletions(-) create mode 100644 tests/unit/test_video_upload_fix_and_generated_api.py diff --git a/apps/api/app/api/routes/chunked_upload.py b/apps/api/app/api/routes/chunked_upload.py index cf8aaae4c..b9537c6e6 100644 --- a/apps/api/app/api/routes/chunked_upload.py +++ b/apps/api/app/api/routes/chunked_upload.py @@ -43,20 +43,33 @@ DEFAULT_CHUNK_SIZE = 5 * 1024 * 1024 # 5MB MAX_FILE_SIZE = 2 * 1024 * 1024 * 1024 # 2GB CHUNK_EXPIRY_HOURS = 24 -# Allowed file types (consistent with existing upload.py) +# Allowed file types — must stay in sync with upload.py ALLOWED_MIME_TYPES ALLOWED_MIME_TYPES = { + # Images "image/jpeg", "image/png", "image/gif", "image/webp", + "image/bmp", + "image/tiff", + "image/svg+xml", + # Video "video/mp4", "video/quicktime", + "video/mpeg", "video/x-msvideo", "video/webm", + "video/x-matroska", + "video/3gpp", + # Audio "audio/mpeg", "audio/wav", "audio/ogg", "audio/mp3", + "audio/flac", + "audio/aac", + "audio/x-m4a", + "audio/webm", } # Chunk storage root directory diff --git a/packages/adapters/sqlalchemy_impl/generated_video_repository.py b/packages/adapters/sqlalchemy_impl/generated_video_repository.py index 3730d04bf..31620e6f6 100644 --- a/packages/adapters/sqlalchemy_impl/generated_video_repository.py +++ b/packages/adapters/sqlalchemy_impl/generated_video_repository.py @@ -78,7 +78,7 @@ class SQLAlchemyGeneratedVideoRepository: def list_by_project(self, project_id: str) -> list[GeneratedVideo]: models = self.session.query(GeneratedVideoModel).filter(GeneratedVideoModel.project_id == project_id).all() - return [self.get(model.id) for model in models if self.get(model.id) is not None] + return [self._to_domain(model) for model in models] def list_by_generation_task(self, generation_task_id: str) -> list[GeneratedVideo]: models = ( @@ -86,7 +86,7 @@ class SQLAlchemyGeneratedVideoRepository: .filter(GeneratedVideoModel.generation_task_id == generation_task_id) .all() ) - return [self.get(model.id) for model in models if self.get(model.id) is not None] + return [self._to_domain(model) for model in models] def list_by_batch(self, batch_id: str) -> list[GeneratedVideo]: """通过 batch_id 查找同批次生成的所有视频（跨 generation_task 关联查询）。""" diff --git a/tests/unit/test_video_upload_fix_and_generated_api.py b/tests/unit/test_video_upload_fix_and_generated_api.py new file mode 100644 index 000000000..972aef13c --- /dev/null +++ b/tests/unit/test_video_upload_fix_and_generated_api.py @@ -0,0 +1,310 @@ +""" +测试视频上传失败排查修复： +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. 成片库 API 端点可用性确认 ─── + + +class TestGeneratedVideosAPIAvailability: + """确认成片库 API 路由注册正确，端点可正常返回数据。""" + + def test_generated_videos_routes_registered(self): + """成片库路由已注册到 router。""" + from apps.api.app.api.router import api_router + + # 检查 router 包含 generated-videos 路径 + routes = [r for r in api_router.routes if hasattr(r, "path")] + gv_routes = [r for r in routes if "generated-videos" in r.path] + assert len(gv_routes) > 0, "generated-videos 路由未注册" + + def test_generated_videos_list_endpoint_exists(self): + """GET /generated-videos 端点存在。""" + from apps.api.app.api.routes.generated_videos import router + + paths = [r.path for r in router.routes if hasattr(r, "path")] + assert "" in paths, "GET /generated-videos 列表端点不存在" + + def test_generated_videos_detail_endpoint_exists(self): + """GET /generated-videos/{video_id} 端点存在。""" + from apps.api.app.api.routes.generated_videos import router + + paths = [r.path for r in router.routes if hasattr(r, "path")] + assert "/{video_id}" in paths, "GET /generated-videos/{{video_id}} 详情端点不存在" + + def test_generated_videos_review_endpoint_exists(self): + """PATCH /generated-videos/{video_id}/review 端点存在。""" + from apps.api.app.api.routes.generated_videos import router + + paths = [r.path for r in router.routes if hasattr(r, "path")] + assert "/{video_id}/review" in paths, "PATCH review 端点不存在" + + def test_generated_videos_download_url_endpoint_exists(self): + """GET /generated-videos/{video_id}/download-url 端点存在。""" + from apps.api.app.api.routes.generated_videos import router + + paths = [r.path for r in router.routes if hasattr(r, "path")] + assert "/{video_id}/download-url" in paths, "download-url 端点不存在" + + 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