476 lines
24 KiB
Python
476 lines
24 KiB
Python
from io import BytesIO
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from django.test import TestCase
|
|
from rest_framework.test import APIClient
|
|
|
|
from apps.accounts.models import Team, TeamMember, User
|
|
|
|
from .models import Asset, FreeAsset, FreeAssetGroup, Model
|
|
|
|
|
|
def _mk_team(username, team_name):
|
|
user = User.objects.create_user(username=username, password="x")
|
|
team = Team.objects.create(name=team_name, owner=user)
|
|
TeamMember.objects.create(team=team, user=user, role=TeamMember.Role.OWNER)
|
|
return user, team
|
|
|
|
|
|
class ModelLibraryApiTests(TestCase):
|
|
"""模特库:本团队模特 ∪ 官方模板;官方跨团队可见;软删;官方不可删;团队隔离。"""
|
|
|
|
def setUp(self):
|
|
self.user, self.teamA = _mk_team("ua", "TeamA")
|
|
self.other, self.teamB = _mk_team("ub", "TeamB")
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(self.user)
|
|
|
|
self.mine = Model.objects.create(team=self.teamA, name="我的模特")
|
|
self.official = Model.objects.create(team=self.teamB, name="官方小美", is_official=True)
|
|
self.bs_other = Model.objects.create(team=self.teamB, name="别队私有") # B 的私有,A 看不到
|
|
|
|
def test_list_returns_mine_plus_official(self):
|
|
ids = {m["id"] for m in self.client.get("/api/models/").json()["results"]}
|
|
self.assertIn(str(self.mine.id), ids)
|
|
self.assertIn(str(self.official.id), ids) # 官方跨团队可见
|
|
self.assertNotIn(str(self.bs_other.id), ids) # 别队私有不可见
|
|
|
|
def test_tab_official_and_mine(self):
|
|
off = {m["id"] for m in self.client.get("/api/models/?tab=official").json()["results"]}
|
|
self.assertEqual(off, {str(self.official.id)})
|
|
mine = {m["id"] for m in self.client.get("/api/models/?tab=mine").json()["results"]}
|
|
self.assertEqual(mine, {str(self.mine.id)})
|
|
|
|
def test_official_flag_in_payload(self):
|
|
row = next(m for m in self.client.get("/api/models/").json()["results"] if m["id"] == str(self.official.id))
|
|
self.assertTrue(row["is_official"])
|
|
|
|
def test_delete_is_soft(self):
|
|
res = self.client.delete(f"/api/models/{self.mine.id}/")
|
|
self.assertEqual(res.status_code, 204)
|
|
self.mine.refresh_from_db()
|
|
self.assertTrue(self.mine.is_deleted)
|
|
ids = {m["id"] for m in self.client.get("/api/models/").json()["results"]}
|
|
self.assertNotIn(str(self.mine.id), ids)
|
|
|
|
def test_official_cannot_be_deleted(self):
|
|
res = self.client.delete(f"/api/models/{self.official.id}/")
|
|
self.assertEqual(res.status_code, 403)
|
|
self.official.refresh_from_db()
|
|
self.assertFalse(self.official.is_deleted)
|
|
|
|
def test_cannot_delete_other_team_model(self):
|
|
res = self.client.delete(f"/api/models/{self.bs_other.id}/")
|
|
self.assertIn(res.status_code, (403, 404)) # 别队私有:不在可见 qs → 404,或命中守卫 → 403
|
|
self.bs_other.refresh_from_db()
|
|
self.assertFalse(self.bs_other.is_deleted)
|
|
|
|
def test_create_rejects_cross_team_asset(self):
|
|
b_asset = Asset.objects.create(team=self.teamB, name="别队图", asset_type="image", source="upload")
|
|
res = self.client.post("/api/models/", {"name": "盗用", "portrait_asset": str(b_asset.id)}, format="json")
|
|
self.assertEqual(res.status_code, 400)
|
|
|
|
def test_upload_real_person_creates_asset_and_model(self):
|
|
stored = SimpleNamespace(object_key="teams/x/models/a.png", bucket="b", content_type="image/png", size_bytes=10)
|
|
with patch("apps.assets.views.TosStorage") as Tos:
|
|
Tos.return_value.upload_fileobj.return_value = stored
|
|
f = BytesIO(b"img")
|
|
f.name = "lady.png"
|
|
res = self.client.post("/api/models/upload/", {"name": "上传妹", "file": f}, format="multipart")
|
|
self.assertEqual(res.status_code, 201)
|
|
body = res.json()
|
|
self.assertEqual(body["source"], "upload")
|
|
model = Model.objects.get(id=body["id"])
|
|
self.assertIsNotNone(model.portrait_asset)
|
|
self.assertEqual(model.portrait_asset.category, Asset.Category.MODEL_PORTRAIT)
|
|
|
|
|
|
class AssetSoftDeleteTests(TestCase):
|
|
"""资产库不展示软删资产(is_deleted=True):列表 + tab 计数都排除。"""
|
|
|
|
def setUp(self):
|
|
self.user, self.team = _mk_team("usd", "TeamSD")
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(self.user)
|
|
self.live = Asset.objects.create(team=self.team, name="留存图", asset_type="image", source="ai_generated", category=Asset.Category.MODEL_TRYON)
|
|
self.dead = Asset.objects.create(team=self.team, name="废图", asset_type="image", source="ai_generated", category=Asset.Category.MODEL_TRYON, is_deleted=True)
|
|
|
|
def test_list_excludes_soft_deleted(self):
|
|
ids = {a["id"] for a in self.client.get("/api/assets/?page_size=50").json()["results"]}
|
|
self.assertIn(str(self.live.id), ids)
|
|
self.assertNotIn(str(self.dead.id), ids)
|
|
|
|
def test_summary_excludes_soft_deleted(self):
|
|
self.assertEqual(self.client.get("/api/assets/summary/").json()["tryon"], 1)
|
|
|
|
|
|
class AssetTrashApiTests(TestCase):
|
|
"""R108 资产垃圾桶:trash 只列本团队软删资产;restore 后资产库重新可见;purge 硬删且被引用时 409。"""
|
|
|
|
def setUp(self):
|
|
self.user, self.team = _mk_team("utr", "TeamTR")
|
|
_, self.other_team = _mk_team("utr2", "TeamTR2")
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(self.user)
|
|
self.live = Asset.objects.create(team=self.team, name="活图", asset_type="image", source="ai_generated", category=Asset.Category.MODEL_TRYON)
|
|
self.dead = Asset.objects.create(team=self.team, name="删掉的图", asset_type="image", source="ai_generated", category=Asset.Category.MODEL_TRYON, is_deleted=True)
|
|
# 别队软删资产:不得出现在本队垃圾桶(team 隔离)
|
|
self.other_dead = Asset.objects.create(team=self.other_team, name="别队删图", asset_type="image", source="ai_generated", is_deleted=True)
|
|
|
|
def test_trash_lists_only_own_deleted(self):
|
|
ids = {a["id"] for a in self.client.get("/api/assets/trash/").json()["results"]}
|
|
self.assertEqual(ids, {str(self.dead.id)}) # 只含本队软删;不含活图 / 别队删图
|
|
|
|
def test_restore_makes_asset_visible_in_library_again(self):
|
|
res = self.client.post(f"/api/assets/{self.dead.id}/restore/")
|
|
self.assertEqual(res.status_code, 200)
|
|
self.dead.refresh_from_db()
|
|
self.assertFalse(self.dead.is_deleted)
|
|
ids = {a["id"] for a in self.client.get("/api/assets/?page_size=50").json()["results"]}
|
|
self.assertIn(str(self.dead.id), ids) # 资产库重新可见
|
|
# 恢复后不再在垃圾桶
|
|
self.assertEqual(self.client.get("/api/assets/trash/").json()["count"], 0)
|
|
|
|
def test_purge_hard_deletes(self):
|
|
res = self.client.delete(f"/api/assets/{self.dead.id}/purge/")
|
|
self.assertEqual(res.status_code, 204)
|
|
self.assertFalse(Asset.objects.filter(id=self.dead.id).exists())
|
|
|
|
def test_purge_referenced_asset_returns_409(self):
|
|
from apps.products.models import Product, ProductImage
|
|
|
|
product = Product.objects.create(team=self.team, created_by=self.user, title="P")
|
|
ProductImage.objects.create(product=product, asset=self.dead) # PROTECT 引用
|
|
res = self.client.delete(f"/api/assets/{self.dead.id}/purge/")
|
|
self.assertEqual(res.status_code, 409)
|
|
self.assertTrue(Asset.objects.filter(id=self.dead.id, is_deleted=True).exists()) # 留在垃圾桶可恢复
|
|
|
|
def test_restore_rejects_live_asset(self):
|
|
# 未删资产不在 trash 态 queryset → 404(与商品垃圾桶口径一致)
|
|
self.assertEqual(self.client.post(f"/api/assets/{self.live.id}/restore/").status_code, 404)
|
|
|
|
|
|
class FreeAssetQuickUploadTests(TestCase):
|
|
"""人物素材库快捷上传:自动默认组,但底层仍走 FreeAssetGroup / FreeAsset。"""
|
|
|
|
def setUp(self):
|
|
self.user, self.team = _mk_team("ufq", "TeamFQ")
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(self.user)
|
|
self.stored = SimpleNamespace(object_key="teams/x/free-create/library/a.png", bucket="b", content_type="image/png", size_bytes=10)
|
|
|
|
def _image(self, name="a.png"):
|
|
return SimpleUploadedFile(name, b"img", content_type="image/png")
|
|
|
|
@patch("apps.assets.free_library.assets_client.is_enabled", return_value=True)
|
|
@patch("apps.assets.free_library.assets_client.create_asset_group", return_value="Group-1")
|
|
@patch("apps.assets.free_library.assets_client.create_asset", return_value="Asset-1")
|
|
@patch("apps.assets.free_library.TosStorage")
|
|
def test_quick_upload_creates_default_group(self, storage_cls, create_asset, create_group, _enabled):
|
|
storage = storage_cls.return_value
|
|
storage.upload_fileobj.return_value = self.stored
|
|
storage.public_url.return_value = "http://tos/a.png"
|
|
|
|
res = self.client.post("/api/assets/free-assets/quick-upload/", {"file": self._image()}, format="multipart")
|
|
self.assertEqual(res.status_code, 201)
|
|
group = FreeAssetGroup.objects.get(team=self.team)
|
|
self.assertEqual(group.name, "默认素材")
|
|
self.assertEqual(group.remote_group_id, "Group-1")
|
|
self.assertEqual(group.assets.count(), 1)
|
|
asset = group.assets.get()
|
|
self.assertEqual(asset.asset_type, FreeAsset.Type.IMAGE)
|
|
self.assertEqual(asset.status, FreeAsset.Status.PROCESSING)
|
|
create_group.assert_called_once()
|
|
create_asset.assert_called_once_with("Group-1", "http://tos/a.png", name="a.png", asset_type=FreeAsset.Type.IMAGE)
|
|
self.assertEqual(res.json()["group"]["id"], str(group.id))
|
|
|
|
@patch("apps.assets.free_library.assets_client.is_enabled", return_value=True)
|
|
@patch("apps.assets.free_library.assets_client.create_asset_group", return_value="Group-1")
|
|
@patch("apps.assets.free_library.assets_client.create_asset", side_effect=["Asset-1", "Asset-2"])
|
|
@patch("apps.assets.free_library.TosStorage")
|
|
def test_quick_upload_reuses_default_group(self, storage_cls, create_asset, create_group, _enabled):
|
|
storage = storage_cls.return_value
|
|
storage.upload_fileobj.return_value = self.stored
|
|
storage.public_url.return_value = "http://tos/a.png"
|
|
|
|
first = self.client.post("/api/assets/free-assets/quick-upload/", {"file": self._image("a.png")}, format="multipart")
|
|
second = self.client.post("/api/assets/free-assets/quick-upload/", {"file": self._image("b.png")}, format="multipart")
|
|
self.assertEqual(first.status_code, 201)
|
|
self.assertEqual(second.status_code, 201)
|
|
self.assertEqual(FreeAssetGroup.objects.filter(team=self.team).count(), 1)
|
|
self.assertEqual(FreeAsset.objects.filter(group__team=self.team).count(), 2)
|
|
create_group.assert_called_once()
|
|
self.assertEqual(create_asset.call_count, 2)
|
|
|
|
@patch("apps.assets.free_library.assets_client.is_enabled", return_value=True)
|
|
@patch("apps.assets.free_library.assets_client.create_asset_group", return_value="Group-1")
|
|
@patch("apps.assets.free_library.TosStorage")
|
|
def test_quick_upload_rejects_non_image(self, storage_cls, create_group, _enabled):
|
|
res = self.client.post(
|
|
"/api/assets/free-assets/quick-upload/",
|
|
{"file": SimpleUploadedFile("v.mp4", b"video", content_type="video/mp4")},
|
|
format="multipart",
|
|
)
|
|
self.assertEqual(res.status_code, 400)
|
|
self.assertIn("仅支持图片", res.json()["detail"])
|
|
self.assertFalse(storage_cls.return_value.upload_fileobj.called)
|
|
create_group.assert_called_once()
|
|
|
|
|
|
class VideoPacksTests(TestCase):
|
|
"""视频成品按项目素材包分组(与导出同源:取 VideoSegment.adopted_version 的真实视频资产)。"""
|
|
|
|
def setUp(self):
|
|
from apps.products.models import Product
|
|
from apps.projects.models import Project, VideoSegment, VideoSegmentVersion
|
|
from .models import AssetFile
|
|
|
|
self.user, self.team = _mk_team("uvp", "TeamVP")
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(self.user)
|
|
self.product = Product.objects.create(team=self.team, created_by=self.user, title="P")
|
|
self.proj = Project.objects.create(team=self.team, name="项目甲", product=self.product, created_by=self.user)
|
|
self.assets = []
|
|
for i in range(2):
|
|
a = Asset.objects.create(team=self.team, name=f"clip{i}", asset_type="video", source="ai_generated", category=Asset.Category.VIDEO_CLIP)
|
|
AssetFile.objects.create(asset=a, object_key=f"k{i}.mp4", bucket="b", content_type="video/mp4", is_primary=True)
|
|
seg = VideoSegment.objects.create(project=self.proj, sort_order=i)
|
|
ver = VideoSegmentVersion.objects.create(video_segment=seg, asset=a, is_adopted=True)
|
|
seg.adopted_version = ver
|
|
seg.save(update_fields=["adopted_version"])
|
|
self.assets.append(a)
|
|
|
|
def test_packs_grouped_by_project(self):
|
|
data = self.client.get("/api/assets/video-packs/").json()
|
|
self.assertEqual(len(data), 1)
|
|
self.assertEqual(data[0]["project_name"], "项目甲")
|
|
self.assertEqual(len(data[0]["clips"]), 2)
|
|
self.assertIn("url", data[0]["clips"][0])
|
|
|
|
def test_soft_deleted_clip_excluded(self):
|
|
Asset.objects.filter(id=self.assets[0].id).update(is_deleted=True)
|
|
data = self.client.get("/api/assets/video-packs/").json()
|
|
self.assertEqual(len(data[0]["clips"]), 1)
|
|
|
|
|
|
class AssetBatchesTests(TestCase):
|
|
"""图片趴(模特上身图)按生成批次成组:同 metadata.batch_id 的多张图 = 一批;
|
|
无 batch_id 的旧图各自成单图批。summary 也按批次计数。"""
|
|
|
|
def setUp(self):
|
|
self.user, self.team = _mk_team("ubt", "TeamBT")
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(self.user)
|
|
# 一批 3 张(同 batch_id)
|
|
for i in range(3):
|
|
Asset.objects.create(
|
|
team=self.team, name=f"AI 生成 · 模特上身图 · {i + 1}", asset_type="image",
|
|
source="ai_generated", category=Asset.Category.MODEL_TRYON, metadata={"batch_id": "B1"},
|
|
)
|
|
# 一张旧图(无 batch_id)→ 自己成一批
|
|
Asset.objects.create(
|
|
team=self.team, name="AI 生成 · 模特上身图 · 1", asset_type="image",
|
|
source="ai_generated", category=Asset.Category.MODEL_TRYON,
|
|
)
|
|
|
|
def test_batches_grouped(self):
|
|
data = self.client.get("/api/assets/batches/?tab=tryon&page_size=10").json()
|
|
self.assertEqual(data["count"], 2) # 一批 3 张 + 一张单图批
|
|
by_count = sorted(b["count"] for b in data["results"])
|
|
self.assertEqual(by_count, [1, 3])
|
|
big = next(b for b in data["results"] if b["count"] == 3)
|
|
self.assertEqual(big["name"], "AI 生成 · 模特上身图") # 去掉「· 序号」
|
|
self.assertEqual(len(big["items"]), 3)
|
|
|
|
def test_summary_counts_batches(self):
|
|
self.assertEqual(self.client.get("/api/assets/summary/").json()["tryon"], 2)
|
|
|
|
|
|
class FreeCreateSplitTabTests(TestCase):
|
|
"""Free create assets share category=free_create; tabs split by asset_type."""
|
|
|
|
def setUp(self):
|
|
self.user, self.team = _mk_team("ufc", "TeamFC")
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(self.user)
|
|
self.image1 = Asset.objects.create(
|
|
team=self.team,
|
|
name="free image 1",
|
|
asset_type=Asset.Type.IMAGE,
|
|
source=Asset.Source.AI_GENERATED,
|
|
category=Asset.Category.FREE_CREATE,
|
|
metadata={"batch_id": "IMG-B1"},
|
|
)
|
|
self.image2 = Asset.objects.create(
|
|
team=self.team,
|
|
name="free image 2",
|
|
asset_type=Asset.Type.IMAGE,
|
|
source=Asset.Source.AI_GENERATED,
|
|
category=Asset.Category.FREE_CREATE,
|
|
metadata={"batch_id": "IMG-B1"},
|
|
)
|
|
self.video = Asset.objects.create(
|
|
team=self.team,
|
|
name="free video",
|
|
asset_type=Asset.Type.VIDEO,
|
|
source=Asset.Source.AI_GENERATED,
|
|
category=Asset.Category.FREE_CREATE,
|
|
metadata={"feature": "free_video"},
|
|
)
|
|
|
|
def test_image_creations_batches_exclude_videos(self):
|
|
data = self.client.get("/api/assets/batches/?tab=image_creations&page_size=10").json()
|
|
self.assertEqual(data["count"], 1)
|
|
self.assertEqual(data["results"][0]["count"], 2)
|
|
self.assertEqual({a["asset_type"] for a in data["results"][0]["items"]}, {"image"})
|
|
|
|
def test_video_creations_list_excludes_images(self):
|
|
data = self.client.get("/api/assets/?tab=video_creations&page_size=10").json()
|
|
self.assertEqual(data["count"], 1)
|
|
self.assertEqual(data["results"][0]["id"], str(self.video.id))
|
|
self.assertEqual(data["results"][0]["asset_type"], "video")
|
|
|
|
def test_summary_exposes_split_counts_and_keeps_legacy_creations(self):
|
|
data = self.client.get("/api/assets/summary/").json()
|
|
self.assertEqual(data["image_creations"], 1)
|
|
self.assertEqual(data["video_creations"], 1)
|
|
self.assertEqual(data["creations"], 2)
|
|
|
|
def test_delete_and_restore_video_returns_to_video_creations_tab(self):
|
|
res = self.client.delete(f"/api/assets/{self.video.id}/")
|
|
self.assertEqual(res.status_code, 204)
|
|
self.video.refresh_from_db()
|
|
self.assertTrue(self.video.is_deleted)
|
|
|
|
video_tab = self.client.get("/api/assets/?tab=video_creations&page_size=10").json()
|
|
self.assertEqual(video_tab["count"], 0)
|
|
summary = self.client.get("/api/assets/summary/").json()
|
|
self.assertEqual(summary["video_creations"], 0)
|
|
self.assertEqual(summary["image_creations"], 1)
|
|
trash_ids = {a["id"] for a in self.client.get("/api/assets/trash/").json()["results"]}
|
|
self.assertIn(str(self.video.id), trash_ids)
|
|
|
|
restore = self.client.post(f"/api/assets/{self.video.id}/restore/")
|
|
self.assertEqual(restore.status_code, 200)
|
|
self.video.refresh_from_db()
|
|
self.assertFalse(self.video.is_deleted)
|
|
self.assertEqual(self.video.category, Asset.Category.FREE_CREATE)
|
|
self.assertEqual(self.video.asset_type, Asset.Type.VIDEO)
|
|
|
|
video_tab = self.client.get("/api/assets/?tab=video_creations&page_size=10").json()
|
|
self.assertEqual(video_tab["count"], 1)
|
|
self.assertEqual(video_tab["results"][0]["id"], str(self.video.id))
|
|
image_batches = self.client.get("/api/assets/batches/?tab=image_creations&page_size=10").json()
|
|
self.assertEqual({a["asset_type"] for a in image_batches["results"][0]["items"]}, {"image"})
|
|
|
|
|
|
class FacetsProductsTests(TestCase):
|
|
"""facets 返回本 tab 下真实关联的商品(?product= 三路之一:metadata.product_id 等),
|
|
并按 tab 隔离、team 隔离、只回 id/title。"""
|
|
|
|
def setUp(self):
|
|
from apps.products.models import Product
|
|
|
|
self.user, self.team = _mk_team("ufp", "TeamFP")
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(self.user)
|
|
self.p1 = Product.objects.create(team=self.team, created_by=self.user, title="商品甲")
|
|
self.p2 = Product.objects.create(team=self.team, created_by=self.user, title="商品乙")
|
|
# tryon 图经 metadata.product_id 关联 p1
|
|
Asset.objects.create(
|
|
team=self.team, name="上身图", asset_type="image", source="ai_generated",
|
|
category=Asset.Category.MODEL_TRYON, metadata={"product_id": str(self.p1.id)},
|
|
)
|
|
# kits 图关联 p2(验证 tab 隔离:tryon 下不应出现 p2)
|
|
Asset.objects.create(
|
|
team=self.team, name="套图", asset_type="image", source="ai_generated",
|
|
category=Asset.Category.PLATFORM_KIT, metadata={"product_id": str(self.p2.id)},
|
|
)
|
|
|
|
def test_facets_products_scoped_by_tab(self):
|
|
data = self.client.get("/api/assets/facets/?tab=tryon").json()
|
|
self.assertIn("products", data)
|
|
self.assertEqual({p["id"] for p in data["products"]}, {str(self.p1.id)})
|
|
self.assertEqual({p["title"] for p in data["products"]}, {"商品甲"})
|
|
# kits tab 只见 p2
|
|
kits = self.client.get("/api/assets/facets/?tab=kits").json()
|
|
self.assertEqual({p["id"] for p in kits["products"]}, {str(self.p2.id)})
|
|
|
|
def test_facets_products_empty_when_none(self):
|
|
data = self.client.get("/api/assets/facets/?tab=others").json()
|
|
self.assertEqual(data["products"], [])
|
|
|
|
|
|
class SubmitReviewTests(TestCase):
|
|
"""手动兜底:灰盾点击 → 提交审核;只对送审类放行,非送审类 400。"""
|
|
|
|
def setUp(self):
|
|
self.user, self.team = _mk_team("usr", "TeamSR")
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(self.user)
|
|
self.person = Asset.objects.create(team=self.team, name="角色", asset_type="image", source="ai_generated", category=Asset.Category.PERSON)
|
|
# 图片趴(模特上身图)= 非送审类,用于校验 submit-review 对其拒绝(product_image/scene 现已属送审类)
|
|
self.tryon = Asset.objects.create(team=self.team, name="上身图", asset_type="image", source="ai_generated", category=Asset.Category.MODEL_TRYON)
|
|
|
|
def test_submit_review_person_ok(self):
|
|
with patch("apps.assets.review.submit_asset_for_review") as sub:
|
|
res = self.client.post(f"/api/assets/{self.person.id}/submit-review/")
|
|
self.assertEqual(res.status_code, 200)
|
|
sub.assert_called_once()
|
|
self.assertIn("review_status", res.json())
|
|
|
|
def test_submit_review_rejects_non_review_category(self):
|
|
res = self.client.post(f"/api/assets/{self.tryon.id}/submit-review/")
|
|
self.assertEqual(res.status_code, 400)
|
|
|
|
def test_submit_review_503_when_submission_did_not_start(self):
|
|
"""送审没真正发出去(服务未启用/未配 SDK/火山未回 Id),且素材非终态 → 503,
|
|
review_status 仍为空。这样前端不会把「没送出去」误显示成「审核中」(刷新即打回原形)。"""
|
|
with patch("apps.assets.review.submit_asset_for_review", return_value=False) as sub:
|
|
res = self.client.post(f"/api/assets/{self.person.id}/submit-review/")
|
|
sub.assert_called_once()
|
|
self.assertEqual(res.status_code, 503)
|
|
self.assertEqual(res.json().get("review_status"), "")
|
|
|
|
def test_submit_review_200_when_started(self):
|
|
"""真正进入审核(processing)→ 200 + 真态,供前端落 processing。"""
|
|
def _mark_processing(asset):
|
|
asset.review_status = "processing"
|
|
asset.save(update_fields=["review_status"])
|
|
return True
|
|
|
|
with patch("apps.assets.review.submit_asset_for_review", side_effect=_mark_processing):
|
|
res = self.client.post(f"/api/assets/{self.person.id}/submit-review/")
|
|
self.assertEqual(res.status_code, 200)
|
|
self.assertEqual(res.json().get("review_status"), "processing")
|
|
|
|
|
|
class ReviewScopeTests(TestCase):
|
|
"""送审范围 = 含人脸资产:角色定妆照(person)/ 模特形象图(model_portrait)/ 三视图(tri_view)/
|
|
分镜图(storyboard)/ 场景(scene)/ 商品图(product_image);图片趴(上身图/套图/创作)不送审。"""
|
|
|
|
def test_review_categories_constant(self):
|
|
self.assertEqual(
|
|
Asset.REVIEW_CATEGORIES,
|
|
("person", "model_portrait", "tri_view", "storyboard", "scene", "product_image"),
|
|
)
|
|
|
|
def test_poll_team_reviews_only_scans_review_categories(self):
|
|
from apps.assets import review
|
|
|
|
_, team = _mk_team("uc", "TeamC")
|
|
keep = {}
|
|
for cat in ("person", "model_portrait", "tri_view", "storyboard", "scene", "product_image"):
|
|
a = Asset.objects.create(team=team, name=cat, asset_type="image", source="ai_generated",
|
|
category=cat, review_status="processing")
|
|
keep[cat] = str(a.id)
|
|
# 图片趴:不应进送审队列
|
|
for cat in ("model_tryon", "platform_kit", "free_create"):
|
|
Asset.objects.create(team=team, name=cat, asset_type="image", source="ai_generated",
|
|
category=cat, review_status="processing")
|
|
with patch("apps.assets.review.assets_client.is_enabled", return_value=False):
|
|
out = review.poll_team_reviews(team)
|
|
self.assertEqual(set(out.keys()), set(keep.values()))
|