xiaoxia
58ff565c48
CI/CD Pipeline / Validate Code Quality And Tests (push) Successful in 42s
CI/CD Pipeline / Unit Tests (push) Successful in 1m18s
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production Runtime Images (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
feat: 素材批量操作接口（软删除/打标签/改分类/智能视图标记）
289 lines
9.9 KiB
Python
Executable File
289 lines
9.9 KiB
Python
Executable File
"""素材批量操作单元测试：软删除、批量打标签、批量分类、批量智能视图标记。"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
|
|
|
|
from packages.adapters.in_memory.asset_repository import InMemoryAssetRepository
|
|
from packages.domain import Asset, AssetStatus
|
|
|
|
|
|
class TestBatchSoftDelete:
|
|
"""batch_delete 软删除测试。"""
|
|
|
|
def _make_assets(self, repo: InMemoryAssetRepository, count: int = 5) -> list[Asset]:
|
|
assets = []
|
|
for i in range(count):
|
|
asset = Asset.create(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name=f"asset_{i}.mp4",
|
|
storage_key=f"uploads/asset_{i}.mp4",
|
|
mime_type="video/mp4",
|
|
status=AssetStatus.READY,
|
|
)
|
|
repo.create(asset)
|
|
assets.append(asset)
|
|
return assets
|
|
|
|
def test_batch_soft_delete_marks_status_deleted(self):
|
|
"""软删除：status 变为 deleted，记录仍然存在。"""
|
|
repo = InMemoryAssetRepository()
|
|
assets = self._make_assets(repo, 3)
|
|
|
|
ids_to_delete = [assets[0].id, assets[2].id]
|
|
count = repo.batch_delete(ids_to_delete)
|
|
|
|
assert count == 2
|
|
# 记录仍在，只是 status 变了
|
|
assert repo.get(assets[0].id) is not None
|
|
assert repo.get(assets[0].id).status == AssetStatus.DELETED
|
|
assert repo.get(assets[2].id).status == AssetStatus.DELETED
|
|
# 未删除的保持原样
|
|
assert repo.get(assets[1].id).status == AssetStatus.READY
|
|
|
|
def test_batch_soft_delete_idempotent(self):
|
|
"""重复删除已删除的素材，计数不增加。"""
|
|
repo = InMemoryAssetRepository()
|
|
assets = self._make_assets(repo, 2)
|
|
|
|
count1 = repo.batch_delete([assets[0].id])
|
|
count2 = repo.batch_delete([assets[0].id])
|
|
|
|
assert count1 == 1
|
|
assert count2 == 0
|
|
assert repo.get(assets[0].id).status == AssetStatus.DELETED
|
|
|
|
def test_batch_soft_delete_empty_list(self):
|
|
repo = InMemoryAssetRepository()
|
|
self._make_assets(repo, 3)
|
|
assert repo.batch_delete([]) == 0
|
|
|
|
def test_batch_soft_delete_nonexistent_ids(self):
|
|
repo = InMemoryAssetRepository()
|
|
self._make_assets(repo, 3)
|
|
assert repo.batch_delete(["nonexistent-1", "nonexistent-2"]) == 0
|
|
|
|
|
|
class TestBatchUpdateMetadata:
|
|
"""batch_update_metadata 批量更新 metadata 测试。"""
|
|
|
|
def test_batch_update_category(self):
|
|
"""批量修改分类（metadata.category）。"""
|
|
repo = InMemoryAssetRepository()
|
|
assets = []
|
|
for i in range(3):
|
|
asset = Asset.create(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name=f"v{i}.mp4",
|
|
storage_key=f"v{i}.mp4",
|
|
mime_type="video/mp4",
|
|
metadata={"existing_key": "existing_value"},
|
|
status=AssetStatus.READY,
|
|
)
|
|
repo.create(asset)
|
|
assets.append(asset)
|
|
|
|
ids = [a.id for a in assets]
|
|
count = repo.batch_update_metadata(ids, {"category": "person"})
|
|
|
|
assert count == 3
|
|
for a in assets:
|
|
updated = repo.get(a.id)
|
|
assert updated.metadata["category"] == "person"
|
|
assert updated.metadata["existing_key"] == "existing_value" # 合并而非覆盖
|
|
|
|
def test_batch_update_smart_view(self):
|
|
"""批量设置智能视图标记。"""
|
|
repo = InMemoryAssetRepository()
|
|
assets = []
|
|
for i in range(4):
|
|
asset = Asset.create(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name=f"v{i}.mp4",
|
|
storage_key=f"v{i}.mp4",
|
|
mime_type="video/mp4",
|
|
status=AssetStatus.READY,
|
|
)
|
|
repo.create(asset)
|
|
assets.append(asset)
|
|
|
|
# 标记前2个为 recommended
|
|
count = repo.batch_update_metadata([assets[0].id, assets[1].id], {"smart_view": "recommended"})
|
|
assert count == 2
|
|
assert repo.get(assets[0].id).metadata["smart_view"] == "recommended"
|
|
assert repo.get(assets[1].id).metadata["smart_view"] == "recommended"
|
|
# 其余不变
|
|
assert "smart_view" not in repo.get(assets[2].id).metadata
|
|
|
|
# 再标记后2个为 high_risk
|
|
count2 = repo.batch_update_metadata([assets[2].id, assets[3].id], {"smart_view": "high_risk"})
|
|
assert count2 == 2
|
|
assert repo.get(assets[2].id).metadata["smart_view"] == "high_risk"
|
|
assert repo.get(assets[3].id).metadata["smart_view"] == "high_risk"
|
|
|
|
def test_batch_update_metadata_partial_existing(self):
|
|
"""部分素材存在时，只更新存在的。"""
|
|
repo = InMemoryAssetRepository()
|
|
asset = Asset.create(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name="v.mp4",
|
|
storage_key="v.mp4",
|
|
mime_type="video/mp4",
|
|
status=AssetStatus.READY,
|
|
)
|
|
repo.create(asset)
|
|
|
|
count = repo.batch_update_metadata([asset.id, "nonexistent"], {"category": "scenic"})
|
|
assert count == 1
|
|
assert repo.get(asset.id).metadata["category"] == "scenic"
|
|
|
|
def test_batch_update_metadata_empty_list(self):
|
|
repo = InMemoryAssetRepository()
|
|
assert repo.batch_update_metadata([], {"category": "x"}) == 0
|
|
|
|
|
|
class TestBatchAddTags:
|
|
"""batch_add_tags 批量添加标签测试。"""
|
|
|
|
def test_batch_add_tags_merges_and_dedups(self):
|
|
"""添加模式：合并去重，已有标签不重复添加。"""
|
|
repo = InMemoryAssetRepository()
|
|
assets = []
|
|
for i in range(3):
|
|
asset = Asset.create(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name=f"v{i}.mp4",
|
|
storage_key=f"v{i}.mp4",
|
|
mime_type="video/mp4",
|
|
status=AssetStatus.READY,
|
|
)
|
|
asset.add_tag("tag-existing")
|
|
repo.create(asset)
|
|
assets.append(asset)
|
|
|
|
ids = [a.id for a in assets]
|
|
count = repo.batch_add_tags(ids, ["tag-1", "tag-2", "tag-existing"])
|
|
|
|
assert count == 3 # 都有新增标签，所以都算变更
|
|
for a in assets:
|
|
updated = repo.get(a.id)
|
|
assert set(updated.tag_ids) == {"tag-existing", "tag-1", "tag-2"}
|
|
|
|
def test_batch_add_tags_no_change_when_all_exist(self):
|
|
"""所有标签都已存在时，返回0。"""
|
|
repo = InMemoryAssetRepository()
|
|
asset = Asset.create(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name="v.mp4",
|
|
storage_key="v.mp4",
|
|
mime_type="video/mp4",
|
|
status=AssetStatus.READY,
|
|
)
|
|
asset.add_tag("tag-a")
|
|
asset.add_tag("tag-b")
|
|
repo.create(asset)
|
|
|
|
count = repo.batch_add_tags([asset.id], ["tag-a", "tag-b"])
|
|
assert count == 0
|
|
|
|
def test_batch_add_tags_empty_input(self):
|
|
repo = InMemoryAssetRepository()
|
|
assert repo.batch_add_tags([], ["tag-1"]) == 0
|
|
assert repo.batch_add_tags(["aid"], []) == 0
|
|
|
|
|
|
class TestBatchReplaceTags:
|
|
"""batch_replace_tags 批量替换标签测试。"""
|
|
|
|
def test_batch_replace_tags_full_override(self):
|
|
"""替换模式：全量覆盖原有标签。"""
|
|
repo = InMemoryAssetRepository()
|
|
assets = []
|
|
for i in range(3):
|
|
asset = Asset.create(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name=f"v{i}.mp4",
|
|
storage_key=f"v{i}.mp4",
|
|
mime_type="video/mp4",
|
|
status=AssetStatus.READY,
|
|
)
|
|
asset.add_tag(f"old-{i}")
|
|
asset.add_tag("old-common")
|
|
repo.create(asset)
|
|
assets.append(asset)
|
|
|
|
ids = [a.id for a in assets]
|
|
count = repo.batch_replace_tags(ids, ["new-1", "new-2"])
|
|
|
|
assert count == 3
|
|
for a in assets:
|
|
updated = repo.get(a.id)
|
|
assert set(updated.tag_ids) == {"new-1", "new-2"}
|
|
|
|
def test_batch_replace_tags_empty_tags_clears_all(self):
|
|
"""替换为空列表：清空所有标签。"""
|
|
repo = InMemoryAssetRepository()
|
|
asset = Asset.create(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name="v.mp4",
|
|
storage_key="v.mp4",
|
|
mime_type="video/mp4",
|
|
status=AssetStatus.READY,
|
|
)
|
|
asset.add_tag("tag-a")
|
|
asset.add_tag("tag-b")
|
|
repo.create(asset)
|
|
|
|
count = repo.batch_replace_tags([asset.id], [])
|
|
assert count == 1
|
|
assert repo.get(asset.id).tag_ids == []
|
|
|
|
def test_batch_replace_tags_empty_assets(self):
|
|
repo = InMemoryAssetRepository()
|
|
assert repo.batch_replace_tags([], ["tag-1"]) == 0
|
|
|
|
|
|
class TestBatchOperationLimits:
|
|
"""批量操作上限与边界测试。"""
|
|
|
|
def test_large_batch_operations(self):
|
|
"""大量素材的批量操作（验证性能基本可用）。"""
|
|
repo = InMemoryAssetRepository()
|
|
assets = []
|
|
for i in range(50):
|
|
asset = Asset.create(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name=f"v{i}.mp4",
|
|
storage_key=f"v{i}.mp4",
|
|
mime_type="video/mp4",
|
|
status=AssetStatus.READY,
|
|
)
|
|
repo.create(asset)
|
|
assets.append(asset)
|
|
|
|
ids = [a.id for a in assets]
|
|
|
|
# 批量打标签
|
|
count = repo.batch_add_tags(ids, ["bulk-tag"])
|
|
assert count == 50
|
|
|
|
# 批量分类
|
|
count = repo.batch_update_metadata(ids, {"category": "scenic"})
|
|
assert count == 50
|
|
|
|
# 批量软删除
|
|
count = repo.batch_delete(ids)
|
|
assert count == 50
|
|
for a in assets:
|
|
assert repo.get(a.id).status == AssetStatus.DELETED
|