xiaoxia-saas/tests/unit/test_asset_dedup.py

"""
素材重复上传检测 单元测试

覆盖：
- 表单上传（multipart）命中去重 → 直接返回已有 asset_id，不上传 OSS
- 直传 OSS complete 命中去重 → 直接返回已有 asset_id，不创建 ingest job
- 未命中去重 → 正常创建 ingest job
- file_hash 为空 → 跳过去重检测
- IngestJob 透传 file_hash 到 Asset
"""

from __future__ import annotations

import os
import sys
from pathlib import Path
from unittest.mock import MagicMock

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

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 Asset, AssetLibrary, AssetLibraryKind, AssetStatus, IngestJob, Project

# ---------------------------------------------------------------------------
# Stub repositories
# ---------------------------------------------------------------------------


class StubProjectRepository:
    def __init__(self, projects: dict | None = None):
        self._projects = projects or {}

    def get(self, project_id: str):
        return self._projects.get(project_id)

    def find_by_id(self, project_id: str):
        return self._projects.get(project_id)


class StubAssetLibraryRepository:
    def __init__(self, libraries: dict | None = None):
        self._libraries = libraries or {}

    def find_by_project(self, project_id: str, kind=None) -> list:
        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


class StubAssetRepository:
    """支持 find_by_library_and_file_hash 去重检测。"""

    def __init__(self, assets: list[Asset] | None = None):
        self._assets = assets or []

    def find_by_library_and_file_hash(self, library_id: str, file_hash: str) -> Asset | None:
        for a in self._assets:
            if a.library_id == library_id and a.file_hash == file_hash:
                return a
        return None

    def create(self, asset: Asset) -> Asset:
        self._assets.append(asset)
        return asset


class StubIngestJobRepository:
    def __init__(self):
        self._jobs: dict[str, IngestJob] = {}

    def create(self, job: IngestJob) -> IngestJob:
        self._jobs[job.id] = job
        return job

    def get(self, job_id: str) -> IngestJob | None:
        return self._jobs.get(job_id)

    def update(self, job: IngestJob) -> IngestJob:
        self._jobs[job.id] = job
        return job


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

DUPE_HASH = "a" * 32


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") -> AssetLibrary:
    return AssetLibrary(id=id, name="Test Library", project_id=project_id, kind=AssetLibraryKind.VIDEO)


def _make_existing_asset(
    id: str = "existing-asset-1",
    library_id: str = "lib-1",
    file_hash: str = DUPE_HASH,
) -> Asset:
    return Asset(
        id=id,
        project_id="proj-1",
        library_id=library_id,
        name="existing.mp4",
        storage_key="uploads/existing/existing.mp4",
        mime_type="video/mp4",
        file_hash=file_hash,
        status=AssetStatus.READY,
    )


def _build_app(
    project_repo=None,
    library_repo=None,
    asset_repo=None,
    ingest_repo=None,
    storage=None,
):
    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()
    asset_repo = asset_repo or StubAssetRepository()
    ingest_repo = ingest_repo or StubIngestJobRepository()
    storage = storage or MagicMock()
    storage.is_configured = True
    storage._normalize_storage_key = lambda key: key
    storage.file_exists = lambda key: True
    storage.upload_file = MagicMock(return_value="https://oss.example.com/file.mp4")
    storage.get_url = MagicMock(return_value="https://oss.example.com/file.mp4")

    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_asset_repository] = lambda: asset_repo
    app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
    app.dependency_overrides[get_storage_service] = lambda: storage

    return app


def _client(**kwargs) -> TestClient:
    return TestClient(_build_app(**kwargs))


# ---------------------------------------------------------------------------
# 测试用例
# ---------------------------------------------------------------------------


class TestMultipartUploadDedup:
    """表单上传（POST /api/v1/assets）去重检测。"""

    def test_dedup_hit_returns_existing_asset(self):
        """file_hash 命中已有素材 → 返回 duplicated=true + asset_id，不上传 OSS。"""
        project = _make_project()
        library = _make_library()
        existing = _make_existing_asset()

        client = _client(
            project_repo=StubProjectRepository({project.id: project}),
            library_repo=StubAssetLibraryRepository({library.id: library}),
            asset_repo=StubAssetRepository([existing]),
        )

        resp = client.post(
            "/api/v1",
            data={
                "project_id": project.id,
                "library_id": library.id,
                "file_hash": DUPE_HASH,
            },
            files={"file": ("test.mp4", b"fake-video-data", "video/mp4")},
        )
        assert resp.status_code == 200
        body = resp.json()
        assert body["duplicated"] is True
        assert body["asset_id"] == existing.id
        assert body["ingest_job_id"] == ""

    def test_dedup_miss_creates_ingest_job(self):
        """file_hash 未命中 → 正常上传并创建 ingest job。"""
        project = _make_project()
        library = _make_library()

        client = _client(
            project_repo=StubProjectRepository({project.id: project}),
            library_repo=StubAssetLibraryRepository({library.id: library}),
            asset_repo=StubAssetRepository([]),  # 无已有素材
        )

        resp = client.post(
            "/api/v1",
            data={
                "project_id": project.id,
                "library_id": library.id,
                "file_hash": "b" * 32,  # 新的 hash
            },
            files={"file": ("test.mp4", b"fake-video-data", "video/mp4")},
        )
        assert resp.status_code == 200
        body = resp.json()
        assert body["duplicated"] is False
        assert body["ingest_job_id"] != ""

    def test_empty_hash_skips_dedup(self):
        """file_hash 为空 → 跳过去重检测，直接上传。"""
        project = _make_project()
        library = _make_library()
        existing = _make_existing_asset()

        client = _client(
            project_repo=StubProjectRepository({project.id: project}),
            library_repo=StubAssetLibraryRepository({library.id: library}),
            asset_repo=StubAssetRepository([existing]),
        )

        resp = client.post(
            "/api/v1",
            data={
                "project_id": project.id,
                "library_id": library.id,
                # 不传 file_hash
            },
            files={"file": ("test.mp4", b"fake-video-data", "video/mp4")},
        )
        assert resp.status_code == 200
        body = resp.json()
        assert body["duplicated"] is False


class TestDirectUploadDedup:
    """直传 OSS complete（POST /api/v1/direct/complete）去重检测。"""

    def test_dedup_hit_returns_existing_asset(self):
        """complete 阶段 file_hash 命中 → 返回 duplicated=true。"""
        project = _make_project()
        library = _make_library()
        existing = _make_existing_asset()

        client = _client(
            project_repo=StubProjectRepository({project.id: project}),
            library_repo=StubAssetLibraryRepository({library.id: library}),
            asset_repo=StubAssetRepository([existing]),
        )

        resp = client.post(
            "/api/v1/direct/complete",
            json={
                "project_id": project.id,
                "library_id": library.id,
                "storage_key": "uploads/abc/test.mp4",
                "file_hash": DUPE_HASH,
            },
        )
        assert resp.status_code == 200
        body = resp.json()
        assert body["duplicated"] is True
        assert body["asset_id"] == existing.id
        assert body["ingest_job_id"] == ""

    def test_dedup_miss_creates_ingest_job(self):
        """complete 阶段 file_hash 未命中 → 创建 ingest job。"""
        project = _make_project()
        library = _make_library()

        client = _client(
            project_repo=StubProjectRepository({project.id: project}),
            library_repo=StubAssetLibraryRepository({library.id: library}),
            asset_repo=StubAssetRepository([]),
        )

        resp = client.post(
            "/api/v1/direct/complete",
            json={
                "project_id": project.id,
                "library_id": library.id,
                "storage_key": "uploads/abc/test.mp4",
                "file_hash": "c" * 32,
            },
        )
        assert resp.status_code == 200
        body = resp.json()
        assert body["duplicated"] is False
        assert body["ingest_job_id"] != ""


class TestIngestJobFileHashPassthrough:
    """file_hash 从上传接口透传到 IngestJob。"""

    def test_ingest_job_stores_file_hash(self):
        """上传时传入的 file_hash 应保存到 IngestJob 实体。"""
        project = _make_project()
        library = _make_library()
        ingest_repo = StubIngestJobRepository()

        client = _client(
            project_repo=StubProjectRepository({project.id: project}),
            library_repo=StubAssetLibraryRepository({library.id: library}),
            asset_repo=StubAssetRepository([]),
            ingest_repo=ingest_repo,
        )

        new_hash = "d" * 32
        client.post(
            "/api/v1",
            data={
                "project_id": project.id,
                "library_id": library.id,
                "file_hash": new_hash,
            },
            files={"file": ("test.mp4", b"fake-video-data", "video/mp4")},
        )

        # 验证 IngestJob 存储了 file_hash
        assert len(ingest_repo._jobs) == 1
        job = list(ingest_repo._jobs.values())[0]
        assert job.file_hash == new_hash