CI Bot
0d98d1ecca
- AssetStatus 新增 DELETED 枚举值 - batch_delete 从硬删除改为软删除（标记 status=deleted） - Repository 新增 batch_update_metadata / batch_add_tags / batch_replace_tags - 新增4个API接口： - POST /assets/batch-delete（软删除，字段从ids改为asset_ids） - POST /assets/batch-tag（add/replace两种模式） - POST /assets/batch-classify（设置metadata.category） - POST /assets/batch-mark（设置metadata.smart_view） - 单操作上限200个，返回成功/失败明细 - 素材库删除改为硬删除（循环调用单条delete） - 20个新增单测 + 更新旧单测，共88个素材相关测试通过 - 集成测试MockRepository同步更新
101 lines
3.5 KiB
Python
Executable File
101 lines
3.5 KiB
Python
Executable File
"""批量删除素材 + 分页优化 单元测试。
|
|
|
|
注意：batch_delete 现在是软删除（标记 status=deleted），不是硬删除。
|
|
"""
|
|
|
|
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 TestBatchDelete:
|
|
"""batch_delete 仓储方法测试（软删除）。"""
|
|
|
|
def _make_repo_with_assets(self):
|
|
repo = InMemoryAssetRepository()
|
|
for i in range(5):
|
|
asset = Asset.create(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name=f"voice_{i}.mp3",
|
|
storage_key=f"uploads/voice_{i}.mp3",
|
|
mime_type="audio/mpeg",
|
|
status=AssetStatus.READY,
|
|
)
|
|
repo.create(asset)
|
|
return repo
|
|
|
|
def test_batch_delete_marks_deleted_status(self):
|
|
"""软删除：status 变为 deleted，记录仍然存在。"""
|
|
repo = InMemoryAssetRepository()
|
|
assets = []
|
|
for i in range(5):
|
|
asset = Asset.create(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name=f"voice_{i}.mp3",
|
|
storage_key=f"uploads/voice_{i}.mp3",
|
|
mime_type="audio/mpeg",
|
|
status=AssetStatus.READY,
|
|
)
|
|
repo.create(asset)
|
|
assets.append(asset)
|
|
|
|
ids_to_delete = [assets[0].id, assets[2].id, assets[4].id]
|
|
deleted_count = repo.batch_delete(ids_to_delete)
|
|
|
|
assert deleted_count == 3
|
|
# 软删除：记录仍在，status 变为 deleted
|
|
assert repo.get(assets[0].id).status == AssetStatus.DELETED
|
|
assert repo.get(assets[2].id).status == AssetStatus.DELETED
|
|
assert repo.get(assets[4].id).status == AssetStatus.DELETED
|
|
# 未删除的保持 ready
|
|
assert repo.get(assets[1].id).status == AssetStatus.READY
|
|
assert repo.get(assets[3].id).status == AssetStatus.READY
|
|
|
|
def test_batch_delete_empty_list(self):
|
|
repo = self._make_repo_with_assets()
|
|
assert repo.batch_delete([]) == 0
|
|
|
|
def test_batch_delete_nonexistent_ids(self):
|
|
repo = self._make_repo_with_assets()
|
|
deleted = repo.batch_delete(["nonexistent-1", "nonexistent-2"])
|
|
assert deleted == 0
|
|
|
|
def test_batch_delete_mixed_existing_and_nonexistent(self):
|
|
repo = InMemoryAssetRepository()
|
|
asset = Asset.create(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name="voice.mp3",
|
|
storage_key="uploads/voice.mp3",
|
|
mime_type="audio/mpeg",
|
|
status=AssetStatus.READY,
|
|
)
|
|
repo.create(asset)
|
|
|
|
deleted = repo.batch_delete([asset.id, "nonexistent"])
|
|
assert deleted == 1
|
|
assert repo.get(asset.id).status == AssetStatus.DELETED
|
|
|
|
def test_batch_delete_idempotent(self):
|
|
"""重复删除已删除的素材不重复计数。"""
|
|
repo = InMemoryAssetRepository()
|
|
asset = Asset.create(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name="voice.mp3",
|
|
storage_key="uploads/voice.mp3",
|
|
mime_type="audio/mpeg",
|
|
status=AssetStatus.READY,
|
|
)
|
|
repo.create(asset)
|
|
|
|
assert repo.batch_delete([asset.id]) == 1
|
|
assert repo.batch_delete([asset.id]) == 0
|
|
assert repo.get(asset.id).status == AssetStatus.DELETED
|