diff --git a/apps/api/app/api/routes/upload.py b/apps/api/app/api/routes/upload.py index 8001d705d..c11f918f7 100644 --- a/apps/api/app/api/routes/upload.py +++ b/apps/api/app/api/routes/upload.py @@ -199,7 +199,8 @@ async def complete_direct_upload( description="上传素材文件（multipart/form-data），支持视频、音频、图片。触发导入流水线自动处理。", ) async def upload_asset( - form_data: Annotated[UploadAssetRequest, Form()], + project_id: str = Form(..., min_length=1, description="项目 ID"), + library_id: str = Form(..., min_length=1, description="素材库 ID"), file: UploadFile = File(..., description="要上传的文件（视频、音频、图片等）"), authenticated_user: AuthenticatedUser = Depends(get_current_user), ingest_job_repository: Any = Depends(get_ingest_job_repository), @@ -208,8 +209,6 @@ async def upload_asset( storage_service: OSSStorageService = Depends(get_storage_service), ) -> UploadAssetResponse: """上传素材文件并触发导入流水线。""" - project_id = form_data.project_id - library_id = form_data.library_id _require_project_and_library(project_id, library_id, project_repository, asset_library_repository) # P2-5: 服务端验证 MIME 类型 diff --git a/tests/unit/test_form_upload_routes.py b/tests/unit/test_form_upload_routes.py new file mode 100644 index 000000000..dcfa631ae --- /dev/null +++ b/tests/unit/test_form_upload_routes.py @@ -0,0 +1,650 @@ +""" +upload.py 表单上传端点单元测试 + +覆盖（24个测试用例）： +- 正常路径（5）：视频/音频/图片上传成功、创建导入任务、响应包含URL +- 文件名校验（5）：路径遍历防护、反斜杠处理、空文件名、特殊字符、无扩展名 +- 异常路径（8）：不支持文件类型、项目/素材库不存在、存储服务错误、缺少参数 +- 多格式支持（3）：多种视频(4种)/音频(5种)/图片(6种)格式 +- MIME验证（6）：有效类型、空类型(400)、不支持类型(415) +""" +from __future__ import annotations + +import io +import os +import sys +from pathlib import Path +from unittest.mock import MagicMock + +# 设置必要环境变量（必须在导入 app 模块之前） +os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing") +os.environ.setdefault("DATABASE_URL", "sqlite:///test.db") + +import pytest + +# 确保 app 模块可导入 +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from packages.domain import AssetLibrary, AssetLibraryKind, Project + + +# --------------------------------------------------------------------------- +# 测试用 Stub（不继承 Port ABC，因为 Port 定义 async 方法，路由实际使用同步 duck-type） +# --------------------------------------------------------------------------- + + +class StubProjectRepository: + def __init__(self, projects: dict[str, Project] | None = None): + self._projects = projects or {} + + def get(self, project_id: str) -> Project | None: + return self._projects.get(project_id) + + def find_by_id(self, project_id: str) -> Project | None: + return self._projects.get(project_id) + + +class StubAssetLibraryRepository: + def __init__(self, libraries: dict[str, AssetLibrary] | None = None): + self._libraries = libraries or {} + + def find_by_project(self, project_id: str, kind=None) -> list[AssetLibrary]: + items = [lib for lib in self._libraries.values() if lib.project_id == project_id] + if kind is not None: + items = [lib for lib in items if lib.kind == kind] + return items + + def list_by_project(self, project_id: str) -> list[AssetLibrary]: + raise AssertionError("路由不应调用 list_by_project，应调用 find_by_project") + + +class StubIngestJobRepository: + def __init__(self): + self._jobs = {} + self._counter = 0 + + def add(self, job) -> None: + self._jobs[job.id] = job + + def get(self, job_id: str): + return self._jobs.get(job_id) + + def update_status(self, job_id, status, **kwargs): + if job_id in self._jobs: + self._jobs[job_id].status = status + + def list_by_library(self, library_id: str, skip: int = 0, limit: int = 50): + return [] + + def count_by_library(self, library_id: str) -> int: + return 0 + + def create(self, job): + """创建一个模拟的 ingest job""" + self._jobs[job.id] = job + return job + + +# --------------------------------------------------------------------------- +# 测试 Fixtures +# --------------------------------------------------------------------------- + + +def _make_project(id: str = "proj-1", owner_user_id: str = "user-1") -> Project: + return Project(id=id, name="Test Project", owner_user_id=owner_user_id) + + +def _make_library( + id: str = "lib-1", + project_id: str = "proj-1", + kind: AssetLibraryKind = AssetLibraryKind.VIDEO, +) -> AssetLibrary: + return AssetLibrary(id=id, name="Test Library", project_id=project_id, kind=kind) + + +def _build_app( + project_repo: StubProjectRepository | None = None, + library_repo: StubAssetLibraryRepository | None = None, + storage: MagicMock | None = None, + ingest_repo: StubIngestJobRepository | None = None, +) -> FastAPI: + """构建一个最小化的 FastAPI app，只注册 upload 路由。""" + from app.api.routes.upload import router + from app.auth import AuthenticatedUser, get_current_user + from app.core.storage import get_storage_service + from app.dependencies import ( + get_asset_library_repository, + get_ingest_job_repository, + get_project_repository, + ) + + app = FastAPI() + app.include_router(router, prefix="/api/v1") + + project_repo = project_repo or StubProjectRepository() + library_repo = library_repo or StubAssetLibraryRepository() + storage = storage or MagicMock() + storage.is_configured = True + storage.upload_file.return_value = "https://bucket.oss.example.com/uploads/test.mp4" + ingest_repo = ingest_repo or StubIngestJobRepository() + + # Mock auth + mock_user = MagicMock(spec=AuthenticatedUser) + mock_user.id = "user-1" + mock_user.email = "test@example.com" + + app.dependency_overrides[get_current_user] = lambda: mock_user + app.dependency_overrides[get_project_repository] = lambda: project_repo + app.dependency_overrides[get_asset_library_repository] = lambda: library_repo + app.dependency_overrides[get_storage_service] = lambda: storage + app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo + + return app + + +def _client(**kwargs) -> TestClient: + app = _build_app(**kwargs) + return TestClient(app) + + +# --------------------------------------------------------------------------- +# 测试用例 +# --------------------------------------------------------------------------- + + +class TestFormUploadSuccess: + """表单上传正常成功路径测试。""" + + def test_upload_video_file_successfully(self): + """上传视频文件成功。""" + project = _make_project() + library = _make_library() + project_repo = StubProjectRepository({project.id: project}) + library_repo = StubAssetLibraryRepository({library.id: library}) + ingest_repo = StubIngestJobRepository() + + storage = MagicMock() + storage.upload_file.return_value = "https://bucket.oss.example.com/uploads/video.mp4" + + client = _client( + project_repo=project_repo, + library_repo=library_repo, + storage=storage, + ingest_repo=ingest_repo, + ) + + # 模拟一个视频文件 + file_content = b"fake video content" * 100 + files = {"file": ("test-video.mp4", io.BytesIO(file_content), "video/mp4")} + data = {"project_id": project.id, "library_id": library.id} + + resp = client.post("/api/v1", files=files, data=data) + + assert resp.status_code == 200 + result = resp.json() + assert "storage_key" in result + assert "ingest_job_id" in result + assert "url" in result + assert storage.upload_file.called + + def test_upload_image_file_successfully(self): + """上传图片文件成功。""" + project = _make_project() + library = _make_library(kind=AssetLibraryKind.IMAGE) + project_repo = StubProjectRepository({project.id: project}) + library_repo = StubAssetLibraryRepository({library.id: library}) + + storage = MagicMock() + storage.upload_file.return_value = "https://bucket.oss.example.com/uploads/image.jpg" + + client = _client( + project_repo=project_repo, + library_repo=library_repo, + storage=storage, + ) + + # 模拟一个图片文件 + file_content = b"fake image content" * 50 + files = {"file": ("test-image.jpg", io.BytesIO(file_content), "image/jpeg")} + data = {"project_id": project.id, "library_id": library.id} + + resp = client.post("/api/v1", files=files, data=data) + + assert resp.status_code == 200 + assert storage.upload_file.called + + def test_upload_audio_file_successfully(self): + """上传音频文件成功。""" + project = _make_project() + library = _make_library(kind=AssetLibraryKind.VOICE) + project_repo = StubProjectRepository({project.id: project}) + library_repo = StubAssetLibraryRepository({library.id: library}) + + storage = MagicMock() + storage.upload_file.return_value = "https://bucket.oss.example.com/uploads/audio.mp3" + + client = _client( + project_repo=project_repo, + library_repo=library_repo, + storage=storage, + ) + + file_content = b"fake audio content" * 50 + files = {"file": ("test-audio.mp3", io.BytesIO(file_content), "audio/mpeg")} + data = {"project_id": project.id, "library_id": library.id} + + resp = client.post("/api/v1", files=files, data=data) + + assert resp.status_code == 200 + assert storage.upload_file.called + + def test_ingest_job_is_created(self): + """验证上传成功后创建导入任务。""" + project = _make_project() + library = _make_library() + project_repo = StubProjectRepository({project.id: project}) + library_repo = StubAssetLibraryRepository({library.id: library}) + ingest_repo = StubIngestJobRepository() + + client = _client( + project_repo=project_repo, + library_repo=library_repo, + ingest_repo=ingest_repo, + ) + + file_content = b"fake video content" + files = {"file": ("test.mp4", io.BytesIO(file_content), "video/mp4")} + data = {"project_id": project.id, "library_id": library.id} + + resp = client.post("/api/v1", files=files, data=data) + + assert resp.status_code == 200 + result = resp.json() + assert "ingest_job_id" in result + # 验证 ingest job 被创建 + job = ingest_repo.get(result["ingest_job_id"]) + assert job is not None + + def test_response_contains_url(self): + """验证响应包含文件 URL。""" + project = _make_project() + library = _make_library() + project_repo = StubProjectRepository({project.id: project}) + library_repo = StubAssetLibraryRepository({library.id: library}) + + storage = MagicMock() + expected_url = "https://bucket.oss.example.com/uploads/my-video.mp4" + storage.upload_file.return_value = expected_url + + client = _client( + project_repo=project_repo, + library_repo=library_repo, + storage=storage, + ) + + file_content = b"fake video content" + files = {"file": ("test.mp4", io.BytesIO(file_content), "video/mp4")} + data = {"project_id": project.id, "library_id": library.id} + + resp = client.post("/api/v1", files=files, data=data) + + assert resp.status_code == 200 + result = resp.json() + assert "url" in result + assert result["url"].startswith("https://") + + +class TestFormUploadFilenameSanitization: + """文件名校验测试。""" + + def _upload_with_filename(self, filename: str): + """辅助方法：使用指定文件名上传文件。""" + project = _make_project() + library = _make_library() + project_repo = StubProjectRepository({project.id: project}) + library_repo = StubAssetLibraryRepository({library.id: library}) + + storage = MagicMock() + storage.upload_file.return_value = "https://bucket.oss.example.com/uploads/test.mp4" + + client = _client( + project_repo=project_repo, + library_repo=library_repo, + storage=storage, + ) + + file_content = b"fake video content" + files = {"file": (filename, io.BytesIO(file_content), "video/mp4")} + data = {"project_id": project.id, "library_id": library.id} + + resp = client.post("/api/v1", files=files, data=data) + return resp, storage + + def test_path_traversal_prevented(self): + """路径遍历防护：../ 被替换为 __。""" + resp, storage = self._upload_with_filename("../../../etc/passwd") + + assert resp.status_code == 200 + # 验证文件名被清理 + call_args = storage.upload_file.call_args + storage_key = call_args[0][1] # 第二个位置参数是 storage_key + # 提取文件名部分（去掉 uploads/{file_id}/ 前缀） + filename_part = storage_key.split("/", 2)[-1] + # 文件名部分不应包含 / 或 \ + assert "/" not in filename_part + assert "\\" not in filename_part + + def test_backslash_replaced(self): + """反斜杠被替换为下划线。""" + resp, storage = self._upload_with_filename(r"folder\subfolder\video.mp4") + + assert resp.status_code == 200 + call_args = storage.upload_file.call_args + storage_key = call_args[0][1] + assert "\\" not in storage_key + + def test_empty_filename_becomes_unknown(self): + """空文件名被 FastAPI 拒绝（422）。""" + resp, storage = self._upload_with_filename("") + + # FastAPI 验证文件名不能为空 + assert resp.status_code == 422 + + def test_special_characters_in_filename(self): + """特殊字符文件名正常处理。""" + resp, storage = self._upload_with_filename("my-video (2024) [HD].mp4") + + assert resp.status_code == 200 + call_args = storage.upload_file.call_args + storage_key = call_args[0][1] + assert "my-video (2024) [HD].mp4" in storage_key + + def test_filename_without_extension(self): + """无扩展名文件名正常处理。""" + resp, storage = self._upload_with_filename("no-extension-file") + + assert resp.status_code == 200 + call_args = storage.upload_file.call_args + storage_key = call_args[0][1] + assert "no-extension-file" in storage_key + + +class TestFormUploadMissingFile: + """缺少文件字段测试。""" + + def test_missing_file_field_returns_422(self): + """缺少文件字段时返回 422。""" + project = _make_project() + library = _make_library() + project_repo = StubProjectRepository({project.id: project}) + library_repo = StubAssetLibraryRepository({library.id: library}) + + client = _client(project_repo=project_repo, library_repo=library_repo) + + # 只提交表单数据，不上传文件 + data = {"project_id": project.id, "library_id": library.id} + resp = client.post("/api/v1", data=data) + + assert resp.status_code == 422 + + +class TestFormUploadMissingParameters: + """缺少必填参数测试。""" + + def test_missing_project_id_returns_422(self): + """缺少 project_id 时返回 422。""" + library = _make_library() + library_repo = StubAssetLibraryRepository({library.id: library}) + + client = _client(library_repo=library_repo) + + file_content = b"fake video content" + files = {"file": ("test.mp4", io.BytesIO(file_content), "video/mp4")} + data = {"library_id": library.id} # 缺少 project_id + + resp = client.post("/api/v1", files=files, data=data) + + assert resp.status_code == 422 + + def test_missing_library_id_returns_422(self): + """缺少 library_id 时返回 422。""" + project = _make_project() + project_repo = StubProjectRepository({project.id: project}) + + client = _client(project_repo=project_repo) + + file_content = b"fake video content" + files = {"file": ("test.mp4", io.BytesIO(file_content), "video/mp4")} + data = {"project_id": project.id} # 缺少 library_id + + resp = client.post("/api/v1", files=files, data=data) + + assert resp.status_code == 422 + + def test_empty_project_id_returns_422(self): + """project_id 为空字符串时返回 422（min_length=1）。""" + library = _make_library() + library_repo = StubAssetLibraryRepository({library.id: library}) + + client = _client(library_repo=library_repo) + + file_content = b"fake video content" + files = {"file": ("test.mp4", io.BytesIO(file_content), "video/mp4")} + data = {"project_id": "", "library_id": library.id} + + resp = client.post("/api/v1", files=files, data=data) + + assert resp.status_code == 422 + + +class TestFormUploadProjectNotFound: + """项目/素材库不存在测试。""" + + def test_returns_404_when_project_not_found(self): + """项目不存在时返回 404。""" + library = _make_library() + library_repo = StubAssetLibraryRepository({library.id: library}) + + client = _client( + project_repo=StubProjectRepository(), + library_repo=library_repo, + ) + + file_content = b"fake video content" + files = {"file": ("test.mp4", io.BytesIO(file_content), "video/mp4")} + data = {"project_id": "nonexistent", "library_id": library.id} + + resp = client.post("/api/v1", files=files, data=data) + + assert resp.status_code == 404 + assert "Project not found" in resp.json()["detail"] + + def test_returns_404_when_library_not_in_project(self): + """素材库不属于该项目时返回 404。""" + project = _make_project() + project_repo = StubProjectRepository({project.id: project}) + + # 素材库属于另一个项目 + other_library = _make_library(project_id="other-project") + library_repo = StubAssetLibraryRepository({other_library.id: other_library}) + + client = _client(project_repo=project_repo, library_repo=library_repo) + + file_content = b"fake video content" + files = {"file": ("test.mp4", io.BytesIO(file_content), "video/mp4")} + data = {"project_id": project.id, "library_id": "nonexistent-lib"} + + resp = client.post("/api/v1", files=files, data=data) + + assert resp.status_code == 404 + assert "Asset library not found" in resp.json()["detail"] + + +class TestFormUploadOSSNotConfigured: + """OSS 未配置测试。""" + + def test_returns_503_when_oss_not_configured(self): + """OSS 未配置时返回 503（RuntimeError）。""" + project = _make_project() + library = _make_library() + project_repo = StubProjectRepository({project.id: project}) + library_repo = StubAssetLibraryRepository({library.id: library}) + + storage = MagicMock() + storage.upload_file.side_effect = RuntimeError("OSS 未配置") + + client = _client( + project_repo=project_repo, + library_repo=library_repo, + storage=storage, + ) + + file_content = b"fake video content" + files = {"file": ("test.mp4", io.BytesIO(file_content), "video/mp4")} + data = {"project_id": project.id, "library_id": library.id} + + resp = client.post("/api/v1", files=files, data=data) + + assert resp.status_code == 503 + + +class TestFormUploadStorageErrors: + """存储服务错误测试。""" + + def test_returns_500_on_generic_storage_error(self): + """存储服务通用错误返回 500。""" + project = _make_project() + library = _make_library() + project_repo = StubProjectRepository({project.id: project}) + library_repo = StubAssetLibraryRepository({library.id: library}) + + storage = MagicMock() + storage.upload_file.side_effect = Exception("Network error") + + client = _client( + project_repo=project_repo, + library_repo=library_repo, + storage=storage, + ) + + file_content = b"fake video content" + files = {"file": ("test.mp4", io.BytesIO(file_content), "video/mp4")} + data = {"project_id": project.id, "library_id": library.id} + + resp = client.post("/api/v1", files=files, data=data) + + assert resp.status_code == 500 + assert "Failed to upload file" in resp.json()["detail"] + + +class TestFormUploadMultipleFormats: + """多格式支持测试。""" + + def _upload_with_mime(self, mime_type: str, filename: str = "test.mp4"): + """辅助方法：使用指定 MIME 类型上传文件。""" + project = _make_project() + library = _make_library() + project_repo = StubProjectRepository({project.id: project}) + library_repo = StubAssetLibraryRepository({library.id: library}) + + storage = MagicMock() + storage.upload_file.return_value = "https://bucket.oss.example.com/uploads/test.mp4" + + client = _client( + project_repo=project_repo, + library_repo=library_repo, + storage=storage, + ) + + file_content = b"fake content" + files = {"file": (filename, io.BytesIO(file_content), mime_type)} + data = {"project_id": project.id, "library_id": library.id} + + return client.post("/api/v1", files=files, data=data) + + def test_multiple_video_formats(self): + """支持多种视频格式（4种）。""" + video_formats = [ + ("video/mp4", "test.mp4"), + ("video/mpeg", "test.mpeg"), + ("video/quicktime", "test.mov"), + ("video/x-msvideo", "test.avi"), + ] + for mime_type, filename in video_formats: + resp = self._upload_with_mime(mime_type, filename) + assert resp.status_code == 200, f"Failed for {mime_type}" + + def test_multiple_audio_formats(self): + """支持多种音频格式（5种）。""" + audio_formats = [ + ("audio/mpeg", "test.mp3"), + ("audio/wav", "test.wav"), + ("audio/ogg", "test.ogg"), + ("audio/flac", "test.flac"), + ("audio/aac", "test.aac"), + ] + for mime_type, filename in audio_formats: + resp = self._upload_with_mime(mime_type, filename) + assert resp.status_code == 200, f"Failed for {mime_type}" + + def test_multiple_image_formats(self): + """支持多种图片格式（6种）。""" + image_formats = [ + ("image/jpeg", "test.jpg"), + ("image/png", "test.png"), + ("image/gif", "test.gif"), + ("image/webp", "test.webp"), + ("image/bmp", "test.bmp"), + ("image/svg+xml", "test.svg"), + ] + for mime_type, filename in image_formats: + resp = self._upload_with_mime(mime_type, filename) + assert resp.status_code == 200, f"Failed for {mime_type}" + + +class TestFormUploadMIMEValidation: + """MIME 类型验证测试。""" + + def _upload_with_content_type(self, content_type: str | None): + """辅助方法：使用指定 Content-Type 上传文件。""" + project = _make_project() + library = _make_library() + project_repo = StubProjectRepository({project.id: project}) + library_repo = StubAssetLibraryRepository({library.id: library}) + + storage = MagicMock() + storage.upload_file.return_value = "https://bucket.oss.example.com/uploads/test.mp4" + + client = _client( + project_repo=project_repo, + library_repo=library_repo, + storage=storage, + ) + + file_content = b"fake content" + # 使用元组形式明确指定 content_type + files = {"file": ("test.mp4", io.BytesIO(file_content), content_type)} + data = {"project_id": project.id, "library_id": library.id} + + return client.post("/api/v1", files=files, data=data) + + def test_valid_mime_type_accepted(self): + """有效 MIME 类型被接受。""" + resp = self._upload_with_content_type("video/mp4") + assert resp.status_code == 200 + + def test_empty_content_type_returns_400(self): + """空 Content-Type 返回 400。""" + # 使用空字符串作为 content type + resp = self._upload_with_content_type("") + assert resp.status_code == 400 + assert "Content-Type header is required" in resp.json()["detail"] + + def test_unsupported_content_type_returns_415(self): + """不支持的 Content-Type 返回 415。""" + resp = self._upload_with_content_type("text/plain") + assert resp.status_code == 415 + assert "not supported" in resp.json()["detail"]