xiaoxia-saas/tests/unit/test_form_upload_routes.py

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


# 确保 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)


class StubAssetRepository:
    """Minimal asset repository stub for upload tests."""
    def __init__(self):
        self._assets = {}

    def create(self, asset):
        self._assets[asset.id] = asset
        return asset

    def find_by_storage_key(self, storage_key):
        for a in self._assets.values():
            if a.storage_key == storage_key:
                return a
        return None

    def find_by_library_and_file_hash(self, library_id, file_hash):
        return None


def _build_app(
    project_repo: StubProjectRepository | None = None,
    library_repo: StubAssetLibraryRepository | None = None,
    storage: MagicMock | None = None,
    ingest_repo: StubIngestJobRepository | None = None,
    asset_repo: StubAssetRepository | 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_asset_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()
    asset_repo = asset_repo or StubAssetRepository()

    # Mock auth
    mock_user = MagicMock(spec=AuthenticatedUser)
    mock_user.id = "user-1"
    mock_user.email = "test@example.com"
    mock_user.user = MagicMock()
    mock_user.user.id = "user-1"

    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
    app.dependency_overrides[get_asset_repository] = lambda: asset_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"]