fix(P1): 素材列表默认status排除deleted,uploading立即可见 #534
@@ -96,8 +96,8 @@ def list_assets(
|
||||
description="按内容分类筛选:scenic=风景、product=产品、person=人物、animal=动物、food=美食、tech=科技、sport=运动、music=音乐、other=其他",
|
||||
),
|
||||
status: Optional[str] = Query(
|
||||
"ready",
|
||||
description="按状态筛选,逗号分隔多值;默认仅返回 ready;传 all 返回所有状态(含 deleted)",
|
||||
"default",
|
||||
description="按状态筛选,逗号分隔多值;默认返回除deleted外的所有状态;传deleted查看回收站;传all返回所有状态",
|
||||
),
|
||||
page: Optional[int] = Query(None, ge=1, description="页码,从1开始;与 page_size 配对使用,优先于 skip/limit"),
|
||||
page_size: Optional[int] = Query(None, ge=1, le=500, description="每页数量;与 page 配对使用"),
|
||||
@@ -119,12 +119,16 @@ def list_assets(
|
||||
status_list: list[str] | None
|
||||
if status and status.lower() == "all":
|
||||
status_list = None # None = 不过滤,返回所有状态
|
||||
elif status and status.lower() == "deleted":
|
||||
status_list = ["deleted"] # 仅查回收站
|
||||
elif status and status.lower() == "default":
|
||||
status_list = ["ready", "uploading", "processing", "error"] # 默认排除deleted
|
||||
elif status:
|
||||
status_list = [s.strip() for s in status.split(",") if s.strip()]
|
||||
if not status_list:
|
||||
status_list = ["ready"]
|
||||
status_list = ["ready", "uploading", "processing", "error"]
|
||||
else:
|
||||
status_list = ["ready"]
|
||||
status_list = ["ready", "uploading", "processing", "error"]
|
||||
|
||||
# kind → file_type 映射(voice 对应 audio)
|
||||
kind_to_file_type = {"video": "video", "voice": "audio", "image": "image"}
|
||||
|
||||
@@ -139,6 +139,7 @@ class StubAssetRepository:
|
||||
return asset
|
||||
|
||||
def delete(self, asset_id: str) -> bool:
|
||||
"""硬删除(与真实实现一致)。"""
|
||||
if asset_id in self._assets:
|
||||
del self._assets[asset_id]
|
||||
return True
|
||||
@@ -470,6 +471,70 @@ class TestListAssets:
|
||||
assert data["skip"] == 0
|
||||
assert data["limit"] == 2
|
||||
|
||||
def test_list_page_page_size(self, client):
|
||||
"""page/page_size 分页参数生效。"""
|
||||
self._create_test_assets(client, 5)
|
||||
|
||||
resp = client.get("/api/v1/assets?library_id=lib-1&page=1&page_size=2")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data["items"]) == 2
|
||||
assert data["skip"] == 0
|
||||
assert data["limit"] == 2
|
||||
|
||||
resp = client.get("/api/v1/assets?library_id=lib-1&page=3&page_size=2")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data["items"]) == 1 # 第3页只有1条
|
||||
|
||||
def test_list_status_filter_default_excludes_deleted(self, client):
|
||||
"""默认列表排除deleted状态素材(批量删除走软删除)。"""
|
||||
self._create_test_assets(client, 3)
|
||||
|
||||
# 批量删除2个
|
||||
items = client.get("/api/v1/assets?library_id=lib-1").json()["items"]
|
||||
delete_ids = [items[0]["id"], items[1]["id"]]
|
||||
client.post("/api/v1/assets/batch-delete", json={"asset_ids": delete_ids})
|
||||
|
||||
# 默认查询应该只剩1个(排除deleted)
|
||||
resp = client.get("/api/v1/assets?library_id=lib-1")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data["items"]) == 1
|
||||
assert data["total"] == 1
|
||||
|
||||
# status=deleted 查回收站
|
||||
resp = client.get("/api/v1/assets?library_id=lib-1&status=deleted")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data["items"]) == 2
|
||||
|
||||
# status=all 查全部
|
||||
resp = client.get("/api/v1/assets?library_id=lib-1&status=all")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data["items"]) == 3
|
||||
|
||||
def test_list_status_filter_uploading_visible(self, client):
|
||||
"""uploading状态的素材默认能看到(上传后立即显示处理中)。"""
|
||||
client.post(
|
||||
"/api/v1/assets",
|
||||
json={
|
||||
"project_id": "proj-1",
|
||||
"library_id": "lib-1",
|
||||
"name": "uploading-test.mp4",
|
||||
"storage_key": "uploads/uploading-test.mp4",
|
||||
"mime_type": "video/mp4",
|
||||
"status": "uploading",
|
||||
},
|
||||
)
|
||||
|
||||
resp = client.get("/api/v1/assets?library_id=lib-1")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data["items"]) == 1
|
||||
assert data["items"][0]["status"] == "uploading"
|
||||
|
||||
def test_list_with_keyword_filter(self, client):
|
||||
"""按名称关键词过滤。"""
|
||||
client.post(
|
||||
|
||||
Reference in New Issue
Block a user