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 类型

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"]