灵应
4fee87c5e8
CI/CD Pipeline / Deploy Staging (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Failing after 45h56m35s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 45h56m48s
任务1: 素材重复上传检测 - 上传接口支持 file_hash 参数，通过 MD5+素材库ID 去重 - 命中去重直接返回已有 asset_id，不重复存 OSS - file_hash 透传: API → IngestJob → Asset 全链路 - 三条上传路径（表单/直传/分片）均支持去重 - Alembic 031: assets + ingest_jobs 加 file_hash 列+索引 - 6 个单元测试覆盖去重命中/未命中/空hash/透传 任务3: 批量生成视频 - POST /generations 支持 count 参数，一次创建多条生成任务 - 每条任务独立状态跟踪，响应返回 task_ids 列表 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
342 lines
11 KiB
Python
342 lines
11 KiB
Python
"""
|
||
素材重复上传检测 单元测试
|
||
|
||
覆盖：
|
||
- 表单上传（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")
|
||
|
||
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
|