feat(#1714): prepare_direct_upload 去重 + 预建 PROCESSING asset 占位 #1730
@@ -165,15 +165,44 @@ def _find_duplicate_asset(
|
||||
within_minutes=FALLBACK_DEDUP_WINDOW_MINUTES,
|
||||
file_size=file_size or 0,
|
||||
)
|
||||
if existing is not None and getattr(existing, "status", None) in ACTIVE_ASSET_STATUSES:
|
||||
logger.info(
|
||||
"素材幂等兜底命中(近期活动同名记录): library=%s name=%s asset=%s status=%s",
|
||||
library_id,
|
||||
filename,
|
||||
getattr(existing, "id", "?"),
|
||||
getattr(existing, "status", "?"),
|
||||
)
|
||||
return existing
|
||||
# 兜底去重：按状态区分处理
|
||||
# - READY/ERROR：稳定素材，总命中（避免重复创建）
|
||||
# - PROCESSING/UPLOADING：预建或 complete 占位，仅当 hash 一致才命中
|
||||
# - 占位无 hash（旧客户端 complete 建的）→ 命中
|
||||
# - 占位有 hash 且与当前请求 hash 一致 → 命中
|
||||
# - 占位有 hash 且与当前请求 hash 不同 → 跳过（内容不同）
|
||||
if existing is not None:
|
||||
status = getattr(existing, "status", None)
|
||||
existing_hash = getattr(existing, "file_hash", "") or ""
|
||||
if status in (AssetStatus.READY, AssetStatus.ERROR):
|
||||
logger.info(
|
||||
"素材幂等兜底命中(近期同名稳定记录): library=%s name=%s asset=%s status=%s",
|
||||
library_id,
|
||||
filename,
|
||||
getattr(existing, "id", "?"),
|
||||
status,
|
||||
)
|
||||
return existing
|
||||
elif status in ACTIVE_ASSET_STATUSES:
|
||||
if existing_hash and file_hash and existing_hash != file_hash:
|
||||
logger.debug(
|
||||
"素材兜底去重跳过(占位 hash 不同): library=%s name=%s asset=%s hash=%s req_hash=%s",
|
||||
library_id,
|
||||
filename,
|
||||
getattr(existing, "id", "?"),
|
||||
existing_hash,
|
||||
file_hash,
|
||||
)
|
||||
existing = None
|
||||
else:
|
||||
logger.info(
|
||||
"素材幂等兜底命中(近期同名活动记录): library=%s name=%s asset=%s status=%s",
|
||||
library_id,
|
||||
filename,
|
||||
getattr(existing, "id", "?"),
|
||||
status,
|
||||
)
|
||||
return existing
|
||||
return None
|
||||
|
||||
|
||||
@@ -187,8 +216,44 @@ def _create_pending_asset(
|
||||
user_id,
|
||||
file_hash="",
|
||||
client_upload_id="",
|
||||
file_size: int = 0,
|
||||
):
|
||||
"""立即创建一条 PROCESSING 状态的 Asset 记录，使前端能马上看到新素材。"""
|
||||
"""立即创建或复用一条 PROCESSING 状态的 Asset 记录。
|
||||
|
||||
find-or-create：prepare 阶段已按 file_hash/client_upload_id 预建的占位记录
|
||||
会被 find_by_library_and_file_hash/find_by_library_and_client_upload_id 命中，
|
||||
直接复用并补齐字段（避免 pre-create + complete 重复建两条）。
|
||||
"""
|
||||
# 1. 按 client_upload_id / file_hash 查找现有记录
|
||||
existing = None
|
||||
if client_upload_id:
|
||||
find_by_cuid = getattr(asset_repository, "find_by_library_and_client_upload_id", None)
|
||||
if callable(find_by_cuid):
|
||||
existing = find_by_cuid(library_id=library_id, client_upload_id=client_upload_id)
|
||||
if existing is None and file_hash:
|
||||
existing = asset_repository.find_by_library_and_file_hash(library_id=library_id, file_hash=file_hash)
|
||||
if existing is not None:
|
||||
# 补齐字段（幂等：避免重复建记录，前端已拿到 asset_id）
|
||||
changed = False
|
||||
if file_hash and not existing.file_hash:
|
||||
existing.file_hash = file_hash
|
||||
changed = True
|
||||
if client_upload_id and not existing.client_upload_id:
|
||||
existing.client_upload_id = client_upload_id
|
||||
changed = True
|
||||
if file_size and not existing.file_size:
|
||||
existing.file_size = file_size
|
||||
changed = True
|
||||
if existing.status not in (AssetStatus.PROCESSING, AssetStatus.UPLOADING):
|
||||
existing.status = AssetStatus.PROCESSING
|
||||
changed = True
|
||||
if changed:
|
||||
try:
|
||||
asset_repository.update(existing)
|
||||
except Exception: # noqa: BLE001 — 字段补齐失败不阻塞主流程
|
||||
pass
|
||||
return existing
|
||||
|
||||
asset = Asset.create(
|
||||
project_id=project_id,
|
||||
library_id=library_id,
|
||||
@@ -199,6 +264,7 @@ def _create_pending_asset(
|
||||
uploaded_by_user_id=user_id,
|
||||
file_hash=file_hash,
|
||||
client_upload_id=client_upload_id,
|
||||
file_size=file_size,
|
||||
)
|
||||
return asset_repository.create(asset)
|
||||
|
||||
@@ -243,9 +309,15 @@ async def prepare_direct_upload(
|
||||
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
||||
project_repository: Any = Depends(get_project_repository),
|
||||
asset_library_repository: Any = Depends(get_asset_library_repository),
|
||||
asset_repository: Any = Depends(get_asset_repository),
|
||||
storage_service: OSSStorageService = Depends(get_storage_service),
|
||||
) -> DirectUploadPrepareResponse:
|
||||
"""创建浏览器直传 OSS 的短期表单签名。"""
|
||||
"""创建浏览器直传 OSS 的短期表单签名，并在签名前按 file_hash/client_upload_id 去重。
|
||||
|
||||
命中去重：直接返回 duplicated=True + skip_transfer=True（前端跳过 OSS 直传），
|
||||
未命中：正常签名 OSS 并立即预建一条 PROCESSING 状态的 asset 记录占住
|
||||
file_hash 闸门，响应带 asset_id 供前端/后续 complete 关联。
|
||||
"""
|
||||
settings = get_settings()
|
||||
max_size_bytes = settings.OSS_DIRECT_UPLOAD_MAX_MB * 1024 * 1024
|
||||
if request.file_size > max_size_bytes:
|
||||
@@ -264,8 +336,39 @@ async def prepare_direct_upload(
|
||||
asset_library_repository,
|
||||
)
|
||||
|
||||
file_id = uuid4().hex[:8]
|
||||
safe_filename = request.filename.replace("/", "_").replace("\\", "_")
|
||||
|
||||
# ── prepare 阶段去重：OSS 签名之前先查已存在素材 ──
|
||||
if request.file_hash or request.client_upload_id:
|
||||
existing = _find_duplicate_asset(
|
||||
asset_repository,
|
||||
library_id=request.library_id,
|
||||
file_hash=request.file_hash,
|
||||
client_upload_id=request.client_upload_id,
|
||||
filename=request.filename,
|
||||
file_size=request.file_size,
|
||||
)
|
||||
if existing is not None:
|
||||
logger.info(
|
||||
"prepare 命中去重: library=%s hash=%s cuid=%s existing_asset=%s",
|
||||
request.library_id,
|
||||
request.file_hash,
|
||||
request.client_upload_id,
|
||||
existing.id,
|
||||
)
|
||||
return DirectUploadPrepareResponse(
|
||||
upload_url="",
|
||||
method="",
|
||||
storage_key=existing.storage_key,
|
||||
expires_at="",
|
||||
fields={},
|
||||
max_size_bytes=0,
|
||||
duplicated=True,
|
||||
skip_transfer=True,
|
||||
asset_id=existing.id,
|
||||
)
|
||||
|
||||
file_id = uuid4().hex[:8]
|
||||
storage_key = f"uploads/{file_id}/{safe_filename}"
|
||||
try:
|
||||
payload = storage_service.create_direct_upload_post(
|
||||
@@ -284,6 +387,27 @@ async def prepare_direct_upload(
|
||||
detail=f"Failed to prepare upload: {type(error).__name__}",
|
||||
) from error
|
||||
|
||||
# ── 预建 asset 占位：占住 file_hash/client_upload_id 闸门，避免并发重复上传 ──
|
||||
pending_asset_id = ""
|
||||
if request.file_hash or request.client_upload_id:
|
||||
try:
|
||||
pending = _create_pending_asset(
|
||||
asset_repository=asset_repository,
|
||||
project_id=request.project_id,
|
||||
library_id=request.library_id,
|
||||
storage_key=storage_key,
|
||||
filename=safe_filename,
|
||||
mime_type=validated_content_type,
|
||||
user_id=authenticated_user.user.id,
|
||||
file_hash=request.file_hash,
|
||||
client_upload_id=request.client_upload_id,
|
||||
file_size=request.file_size,
|
||||
)
|
||||
pending_asset_id = pending.id
|
||||
except Exception as error:
|
||||
# 预建失败不阻塞签名：complete 仍可按 OSS 文件 + hash 兜底去重
|
||||
logger.warning("预建 asset 占位失败，降级走 old flow: %s", error)
|
||||
|
||||
return DirectUploadPrepareResponse(
|
||||
upload_url=str(payload["url"]),
|
||||
method=str(payload["method"]),
|
||||
@@ -291,6 +415,9 @@ async def prepare_direct_upload(
|
||||
expires_at=str(payload["expires_at"]),
|
||||
fields={str(key): str(value) for key, value in dict(payload["fields"]).items()},
|
||||
max_size_bytes=max_size_bytes,
|
||||
duplicated=False,
|
||||
skip_transfer=False,
|
||||
asset_id=pending_asset_id,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -16,6 +16,7 @@ class DirectUploadPrepareRequest(BaseModel):
|
||||
content_type: str = Field(default="application/octet-stream", min_length=1, max_length=100)
|
||||
file_size: int = Field(..., gt=0)
|
||||
file_hash: str = Field(default="", max_length=64, description="文件 MD5 哈希，用于去重检测")
|
||||
client_upload_id: str = Field(default="", max_length=64, description="客户端幂等 token（同一次上传的重试保持一致）")
|
||||
|
||||
|
||||
class DirectUploadPrepareResponse(BaseModel):
|
||||
@@ -25,6 +26,9 @@ class DirectUploadPrepareResponse(BaseModel):
|
||||
expires_at: str
|
||||
fields: dict[str, str]
|
||||
max_size_bytes: int
|
||||
duplicated: bool = False
|
||||
skip_transfer: bool = False
|
||||
asset_id: str = ""
|
||||
|
||||
|
||||
class DirectUploadCompleteRequest(BaseModel):
|
||||
|
||||
@@ -0,0 +1,471 @@
|
||||
"""#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"},
|
||||
}
|
||||
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
|
||||
Reference in New Issue
Block a user