from io import BytesIO from types import SimpleNamespace from unittest.mock import patch from django.test import TestCase from rest_framework.test import APIClient from apps.accounts.models import Team, TeamMember, User from .models import Asset, 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 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 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()))