xiaoxia-saas/tests/unit/test_prepare_dedup_1714.py

"""#1714 prepare_direct_upload 去重 + 预建 asset 测试。

覆盖 4 类用例：
- 第一次上传：prepare 返回 duplicated=false + asset_id 非空
- 第二次同 hash：prepare 返回 duplicated=true, skip_transfer=true
- 同 client_upload_id 重试：prepare 也直接跳过
- file_hash 空：走老逻辑，duplicated=false，无 asset_id

以及：
- pre-create 的 PROCESSING 占位不被"文件名兜底去重"误命中
- _create_pending_asset find-or-create 复用现有记录
"""

from __future__ import annotations

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

import pytest

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

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from apps.api.app.api.routes import upload as upload_route  # noqa: E402
from packages.domain.entities import Asset, AssetLibrary, AssetLibraryKind, AssetStatus, Project  # noqa: E402

# ---------------------------------------------------------------------------
# Fake repository
# ---------------------------------------------------------------------------


class _FakeAssetRepo:
    """内存 asset 仓储：实现 prepare/complete 去重需要的所有方法。"""

    def __init__(self):
        self.assets = {}  # id -> Asset
        self.saved = 0
        self.updated = 0

    def create(self, asset):
        self.assets[asset.id] = asset
        self.saved += 1
        return asset

    def update(self, asset):
        self.assets[asset.id] = asset
        self.updated += 1
        return asset

    def find_by_id(self, asset_id):
        return self.assets.get(asset_id)

    def find_by_library_and_file_hash(self, library_id, file_hash):
        if not file_hash:
            return None
        for a in self.assets.values():
            if a.library_id == library_id and a.file_hash == file_hash:
                return a
        return None

    def find_by_library_and_client_upload_id(self, library_id, client_upload_id):
        if not client_upload_id:
            return None
        for a in self.assets.values():
            if a.library_id == library_id and a.client_upload_id == client_upload_id:
                return a
        return None

    def find_recent_active_by_library_and_name(self, library_id, name, within_minutes=30, file_size=0):
        return None


def _make_asset(**kw):
    defaults = dict(
        project_id="p-1",
        library_id="lib-1",
        name="existing.mp4",
        storage_key="uploads/old/existing.mp4",
        mime_type="video/mp4",
        status=AssetStatus.READY,
        file_hash="existinghash",
    )
    defaults.update(kw)
    return Asset(id=defaults.pop("id", "existing-asset"), **defaults)


def _make_pending(**kw):
    defaults = dict(
        project_id="p-1",
        library_id="lib-1",
        name="test.mp4",
        storage_key="uploads/abc/test.mp4",
        mime_type="video/mp4",
        status=AssetStatus.PROCESSING,
        file_hash="abc123",
    )
    defaults.update(kw)
    return Asset(id=defaults.pop("id", "pending-asset"), **defaults)


def _user():
    return SimpleNamespace(user=SimpleNamespace(id="user-1"), session_id="s", token_type="t")


class _StubProjectRepo:
    def __init__(self, project):
        self._p = project

    def get(self, pid):
        return self._p if self._p.id == pid else None

    def find_by_id(self, pid):
        return self._p if self._p.id == pid else None


class _StubLibraryRepo:
    def __init__(self, lib):
        self._lib = lib

    def find_by_project(self, pid, kind=None):
        if self._lib.project_id == pid:
            return [self._lib]
        return []


_FIXTURE_PROJECT = Project(id="p-1", owner_user_id="user-1", name="proj", description="")
_FIXTURE_LIBRARY = AssetLibrary(
    id="lib-1", project_id="p-1", name="videos", kind=AssetLibraryKind.VIDEO, asset_count=0, total_size=0
)


def _storage():
    s = MagicMock()
    s.create_direct_upload_post.return_value = {
        "url": "https://bucket.oss.example.com",
        "method": "POST",
        "storage_key": "uploads/abc/test.mp4",
        "expires_at": "2026-01-01T00:00:00Z",
        "fields": {"key": "uploads/abc/test.mp4"},
    }
    # Bug #2110: duplicated 命中时 _get_existing_asset_url 调用 get_url 返回公网 URL 字符串，
    # Mock 默认返回 MagicMock，会让 DirectUploadPrepareResponse.url: str 校验失败。
    s.get_url.return_value = ""
    return s


# ---------------------------------------------------------------------------
# 场景 1：第一次上传（无 file_hash）
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_prepare_first_upload_no_hash_returns_no_dedup():
    repo = _FakeAssetRepo()
    req = SimpleNamespace(
        project_id="p-1",
        library_id="lib-1",
        filename="test.mp4",
        content_type="video/mp4",
        file_size=1024,
        file_hash="",
        client_upload_id="",
    )
    resp = await upload_route.prepare_direct_upload(
        request=req,
        authenticated_user=_user(),
        project_repository=_StubProjectRepo(_FIXTURE_PROJECT),
        asset_library_repository=_StubLibraryRepo(_FIXTURE_LIBRARY),
        asset_repository=repo,
        storage_service=_storage(),
    )
    assert resp.duplicated is False
    assert resp.skip_transfer is False
    assert resp.asset_id == ""  # file_hash 空，不预建
    assert repo.saved == 0


# ---------------------------------------------------------------------------
# 场景 2：第一次上传带 file_hash → duplicated=false + asset_id 非空
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_prepare_first_upload_with_hash_creates_pending():
    repo = _FakeAssetRepo()
    req = SimpleNamespace(
        project_id="p-1",
        library_id="lib-1",
        filename="test.mp4",
        content_type="video/mp4",
        file_size=1024,
        file_hash="abc123",
        client_upload_id="",
    )
    resp = await upload_route.prepare_direct_upload(
        request=req,
        authenticated_user=_user(),
        project_repository=_StubProjectRepo(_FIXTURE_PROJECT),
        asset_library_repository=_StubLibraryRepo(_FIXTURE_LIBRARY),
        asset_repository=repo,
        storage_service=_storage(),
    )
    assert resp.duplicated is False
    assert resp.skip_transfer is False
    assert resp.asset_id != ""
    # 预建记录确实落库
    assert repo.saved == 1
    pending = repo.find_by_id(resp.asset_id)
    assert pending is not None
    assert pending.file_hash == "abc123"
    assert pending.status == AssetStatus.PROCESSING


# ---------------------------------------------------------------------------
# 场景 3：第二次同 hash → duplicated=true, skip_transfer=true
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_prepare_second_upload_same_hash_returns_duplicated():
    repo = _FakeAssetRepo()
    repo.create(_make_pending(file_hash="abc123", id="existing-asset"))
    req = SimpleNamespace(
        project_id="p-1",
        library_id="lib-1",
        filename="test.mp4",
        content_type="video/mp4",
        file_size=1024,
        file_hash="abc123",
        client_upload_id="",
    )
    resp = await upload_route.prepare_direct_upload(
        request=req,
        authenticated_user=_user(),
        project_repository=_StubProjectRepo(_FIXTURE_PROJECT),
        asset_library_repository=_StubLibraryRepo(_FIXTURE_LIBRARY),
        asset_repository=repo,
        storage_service=_storage(),
    )
    assert resp.duplicated is True
    assert resp.skip_transfer is True
    assert resp.asset_id == "existing-asset"
    assert resp.upload_url == ""  # 未签名 OSS
    # 未新增记录
    assert repo.saved == 1  # 只有初始那条


# ---------------------------------------------------------------------------
# 场景 4：同 client_upload_id 重试 → 直接跳过
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_prepare_retry_same_client_upload_id_skips():
    repo = _FakeAssetRepo()
    repo.create(
        _make_pending(
            file_hash="abc123",
            client_upload_id="cuid-xyz",
            id="existing-asset",
        )
    )
    # 即使 file_hash 不同（理论上不会），client_upload_id 命中也直接跳过
    req = SimpleNamespace(
        project_id="p-1",
        library_id="lib-1",
        filename="test.mp4",
        content_type="video/mp4",
        file_size=1024,
        file_hash="different-hash",
        client_upload_id="cuid-xyz",
    )
    resp = await upload_route.prepare_direct_upload(
        request=req,
        authenticated_user=_user(),
        project_repository=_StubProjectRepo(_FIXTURE_PROJECT),
        asset_library_repository=_StubLibraryRepo(_FIXTURE_LIBRARY),
        asset_repository=repo,
        storage_service=_storage(),
    )
    assert resp.duplicated is True
    assert resp.skip_transfer is True
    assert resp.asset_id == "existing-asset"


# ---------------------------------------------------------------------------
# 兜底：文件名兜底去重不误命中 PROCESSING 占位
# ---------------------------------------------------------------------------


def test_filename_fallback_does_not_match_processing_pending():
    """_find_duplicate_asset 按文件名兜底时，不能命中 pre-create 的 PROCESSING 记录。"""
    repo = _FakeAssetRepo()
    repo.create(_make_pending(id="p1"))
    result = upload_route._find_duplicate_asset(
        repo,
        library_id="lib-1",
        file_hash="",  # 无 hash
        client_upload_id="",  # 无 cuid
        filename="test.mp4",  # 同名
        file_size=1024,
    )
    assert result is None  # PROCESSING 占位不被兜底命中


def test_filename_fallback_matches_stable_ready_record():
    """READY 状态的已存在记录能被文件名兜底命中。"""
    repo = _FakeAssetRepo()
    repo.create(_make_asset(status=AssetStatus.READY, id="ready-asset"))
    # 伪造 find_recent_active_by_library_and_name 返回 READY 记录
    repo.find_recent_active_by_library_and_name = lambda **kw: repo.assets["ready-asset"]
    result = upload_route._find_duplicate_asset(
        repo,
        library_id="lib-1",
        file_hash="",
        client_upload_id="",
        filename="existing.mp4",
        file_size=1024,
    )
    assert result is not None
    assert result.id == "ready-asset"


# ---------------------------------------------------------------------------
# _create_pending_asset find-or-create
# ---------------------------------------------------------------------------


def test_create_pending_asset_reuses_existing_by_hash():
    """_create_pending_asset：file_hash 命中现有 PROCESSING 记录则复用，不新建。"""
    repo = _FakeAssetRepo()
    repo.create(_make_pending(file_hash="abc123", client_upload_id="", id="p1"))
    # 复用
    result = upload_route._create_pending_asset(
        asset_repository=repo,
        project_id="p-1",
        library_id="lib-1",
        storage_key="uploads/new/test.mp4",
        filename="test.mp4",
        mime_type="video/mp4",
        user_id="user-1",
        file_hash="abc123",
        client_upload_id="cuid-new",
    )
    assert result.id == "p1"
    assert repo.saved == 1  # 没新增
    assert repo.updated >= 1  # 字段补齐触发 update
    assert result.client_upload_id == "cuid-new"


def test_create_pending_asset_creates_when_no_match():
    """无匹配时正常新建。"""
    repo = _FakeAssetRepo()
    result = upload_route._create_pending_asset(
        asset_repository=repo,
        project_id="p-1",
        library_id="lib-1",
        storage_key="uploads/new/test.mp4",
        filename="test.mp4",
        mime_type="video/mp4",
        user_id="user-1",
        file_hash="newhash",
        client_upload_id="newcuid",
    )
    assert result.id != ""
    assert result.file_hash == "newhash"
    assert result.client_upload_id == "newcuid"
    assert repo.saved == 1


# ---------------------------------------------------------------------------
# 兜底去重：PROCESSING 占位 hash 不同时跳过
# ---------------------------------------------------------------------------


def test_filename_fallback_skips_processing_with_different_hash():
    """PROCESSING/UPLOADING 占位记录仅当 hash 一致（或占位无 hash）才命中；hash 不同跳过。"""
    repo = _FakeAssetRepo()
    repo.create(_make_pending(id="p1", file_hash="oldhash"))
    repo.find_recent_active_by_library_and_name = lambda **kw: repo.assets["p1"]
    result = upload_route._find_duplicate_asset(
        repo,
        library_id="lib-1",
        file_hash="differenthash",  # 新上传内容不同
        client_upload_id="",
        filename="test.mp4",
        file_size=1024,
    )
    assert result is None


def test_filename_fallback_matches_processing_with_same_hash():
    """PROCESSING 占位 hash 与请求一致时命中（重试场景）。"""
    repo = _FakeAssetRepo()
    repo.create(_make_pending(id="p1", file_hash="samehash"))
    repo.find_recent_active_by_library_and_name = lambda **kw: repo.assets["p1"]
    result = upload_route._find_duplicate_asset(
        repo,
        library_id="lib-1",
        file_hash="samehash",
        client_upload_id="",
        filename="test.mp4",
        file_size=1024,
    )
    assert result is not None
    assert result.id == "p1"


# ---------------------------------------------------------------------------
# prepare 预建失败降级：不阻塞签名
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_prepare_pending_asset_create_failure_degrades_gracefully():
    """预建 asset 抛异常时，prepare 仍正常返回签名（duplicated=False, asset_id 空）。"""

    class _BrokenRepo(_FakeAssetRepo):
        def create(self, asset):
            raise RuntimeError("db down")

    repo = _BrokenRepo()
    req = SimpleNamespace(
        project_id="p-1",
        library_id="lib-1",
        filename="test.mp4",
        content_type="video/mp4",
        file_size=1024,
        file_hash="abc123",
        client_upload_id="cuid-1",
    )
    resp = await upload_route.prepare_direct_upload(
        request=req,
        authenticated_user=_user(),
        project_repository=_StubProjectRepo(_FIXTURE_PROJECT),
        asset_library_repository=_StubLibraryRepo(_FIXTURE_LIBRARY),
        asset_repository=repo,
        storage_service=_storage(),
    )
    assert resp.duplicated is False
    assert resp.skip_transfer is False
    assert resp.asset_id == ""  # 预建失败，降级无 asset_id
    assert resp.upload_url != ""  # 签名仍正常返回


def test_create_pending_asset_update_failure_swallowed():
    """复用占位记录时字段补齐 update 抛异常被吞掉，不阻塞返回。"""

    class _UpdateBrokenRepo(_FakeAssetRepo):
        def update(self, asset):
            raise RuntimeError("db down")

    repo = _UpdateBrokenRepo()
    repo.create(_make_pending(file_hash="abc123", client_upload_id="", id="p1"))
    result = upload_route._create_pending_asset(
        asset_repository=repo,
        project_id="p-1",
        library_id="lib-1",
        storage_key="uploads/new/test.mp4",
        filename="test.mp4",
        mime_type="video/mp4",
        user_id="user-1",
        file_hash="abc123",
        client_upload_id="cuid-new",
        file_size=1024,
    )
    assert result.id == "p1"  # 仍复用，不抛异常
    assert repo.saved == 1