import json from pathlib import Path from types import SimpleNamespace from django.conf import settings from django.test import SimpleTestCase from apps.ai.script_agent import normalize_draft class SkillPromptBundlingTests(SimpleTestCase): """守护「skills 必须随后端打进镜像」这条线 —— 历史上 skills 放仓库根、不在 Docker 构建上下文 (`./core/backend`)→ 镜像里没有 → 提示词加载为空 → 提取模型收不到「只输出 JSON」铁律 → 吐散文 → 「提取结果解析失败」。这两条断言锁死:① skills 落在 BASE_DIR 内(随镜像走);② 提示词非空且带铁律。""" def test_skills_dir_bundled_under_base_dir(self): """skills 必须在 BASE_DIR(=core/backend)内,才会被 Dockerfile 的 `COPY . .` 打进镜像。""" for name in ("ecommerce-entity-extract", "ecommerce-video-script"): self.assertTrue( (Path(settings.BASE_DIR) / "skills" / name / "SKILL.md").exists(), f"skills/{name}/SKILL.md 不在 BASE_DIR 内 —— 镜像将丢失该提示词(详见 services._skills_root)", ) def test_entity_extract_prompt_loads_nonempty_with_json_rule(self): """提取系统提示词必须加载到非空内容,且含「只输出 JSON」铁律(空 → 模型不吐 JSON → 解析失败)。""" from apps.ai.services import _load_skill_system_prompt prompt = _load_skill_system_prompt("ecommerce-entity-extract") self.assertGreater(len(prompt), 1000, "提取提示词为空/过短 —— skills 没被正确加载") self.assertIn("JSON", prompt) self.assertIn("entities", prompt) def test_extract_output_contract_is_hardcoded_safety_net(self): """写死的提取输出契约(随代码进镜像)必须存在且含 JSON 铁律 —— 这是 skill 丢失时的兜底, 保证提取永远收到「只输出 JSON」指令(对齐脚本 agent 的 _OUTPUT_PROTOCOL,根治"系统提示词为空"事故)。""" from apps.ai.services import _EXTRACT_OUTPUT_CONTRACT self.assertGreater(len(_EXTRACT_OUTPUT_CONTRACT), 200) for marker in ("entities", "segments", "character", "scene", "JSON"): self.assertIn(marker, _EXTRACT_OUTPUT_CONTRACT) class NormalizeDraftTests(SimpleTestCase): """normalize_draft 对模型不按契约输出的容错(防「旁白/画面全空」回归)。""" def test_scriptdraft_shots_variant_fills_narration_and_visual(self): """模型常见变体:{"ScriptDraft":{"basicInfo":..,"shots":[{scene,dialogue,subtitle}]}}。 必须解开外壳 + 把 shots→segments、scene→画面、dialogue/subtitle→旁白,而非全填空占位镜。""" raw = json.dumps({ "ScriptDraft": { "basicInfo": {"totalDuration": 60, "aspectRatio": "9:16", "totalShots": 4}, "shots": [ {"shotNo": 1, "duration": 15, "scene": "更衣室扯卡裆旧裤", "dialogue": "卡裆太社死！", "subtitle": "还在卡裆？"}, {"shotNo": 2, "duration": 15, "scene": "特写拉扯面料回弹", "dialogue": "高弹不变形！", "subtitle": "裸感面料"}, {"shotNo": 3, "duration": 15, "scene": "健身切通勤", "dialogue": "都能穿！", "subtitle": "一裤多穿"}, {"shotNo": 4, "duration": 15, "scene": "对镜弹小黄车", "dialogue": "点小黄车抢！", "subtitle": "点击入手"}, ], } }, ensure_ascii=False) draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=60) self.assertEqual(len(draft["segments"]), 4) self.assertEqual([s["role"] for s in draft["segments"]], ["钩子", "痛点", "卖点", "CTA"]) for seg in draft["segments"]: self.assertTrue(seg["narration"], "旁白不应为空") self.assertTrue(seg["visual"], "画面不应为空") self.assertEqual(draft["segments"][0]["narration"], "卡裆太社死！") self.assertEqual(draft["segments"][0]["visual"], "更衣室扯卡裆旧裤") def test_lines_and_caption_variant_fills_narration(self): """另一种变体(Doubao-Seed-2.0-P):shots 用 lines:[{role,content}] 放口播、caption 放字幕。 必须把 lines 的 content 当对白/旁白,caption 兜底,而不是只填画面留旁白空。""" raw = json.dumps({ "ScriptDraft": { "basicInfo": {"totalDuration": 30, "aspectRatio": "9:16"}, "shots": [ {"shotNo": 1, "scene": "化妆台两闺蜜", "lines": [{"role": "女主", "content": "防晒泛白太尴尬！"}, {"role": "闺蜜", "content": "试试这个！"}], "caption": "防晒踩雷？"}, {"shotNo": 2, "scene": "手背挤膏体", "lines": [{"role": "闺蜜", "content": "质地清透不泛白"}], "caption": "清透质地"}, ], } }, ensure_ascii=False) draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=30) self.assertEqual(len(draft["segments"]), 2) for seg in draft["segments"]: self.assertTrue(seg["narration"], "旁白不应为空") self.assertTrue(seg["visual"], "画面不应为空") self.assertIn("防晒泛白太尴尬", draft["segments"][0]["narration"]) self.assertEqual(len(draft["segments"][0]["dialogue"]), 2) # lines→结构化对白 def test_generic_resolver_covers_unseen_field_names(self): """模型每次换字段名(scene/screenDescription/画面…、dialogue/lines/caption…)。 通用解析应「优先键 + 关键词模糊匹配」都能填,且跳过 bgMusic/note 等非内容键。""" variants = { "canonical": {"total_duration": 15, "segments": [{"role": "钩子", "narration": "口播A", "visual": "画面A"}]}, "scene+dialogue_str+subtitle": {"total_duration": 15, "shots": [{"scene": "画面B", "dialogue": "口播B", "subtitle": "字幕B"}]}, "scene+lines+caption": {"total_duration": 15, "shots": [{"scene": "画面C", "lines": [{"role": "女主", "content": "口播C"}], "caption": "字幕C"}]}, "screenDescription+dialogue+bgMusic+note": {"total_duration": 15, "shots": [{"screenDescription": "画面D", "dialogue": "口播D", "bgMusic": "音乐D", "note": "备注D"}]}, } for name, raw in variants.items(): draft = normalize_draft(json.dumps(raw, ensure_ascii=False), aspect_ratio="9:16", total_duration=15) seg = draft["segments"][0] self.assertTrue(seg["narration"], f"{name}: 旁白为空") self.assertTrue(seg["visual"], f"{name}: 画面为空") # bgMusic/note 不得被误当画面/旁白 d = normalize_draft(json.dumps(variants["screenDescription+dialogue+bgMusic+note"], ensure_ascii=False), aspect_ratio="9:16", total_duration=15) self.assertEqual(d["segments"][0]["visual"], "画面D") self.assertEqual(d["segments"][0]["narration"], "口播D") def test_string_dialogue_not_iterated_as_chars(self): """dialogue 为整句字符串时,要当作旁白而不是逐字符遍历。""" raw = json.dumps({ "segments": [{"role": "钩子", "dialogue": "一句完整口播", "visual": "画面"}], "total_duration": 15, }, ensure_ascii=False) draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=15) self.assertEqual(draft["segments"][0]["narration"], "一句完整口播") self.assertEqual(draft["segments"][0]["dialogue"], []) # 字符串不进结构化对白 def test_empty_segments_skeleton_yields_to_richer_scenes(self): """真实回归(全空根因):模型把内容放 scenes/voiceover/visual,却另给一个**空 segments 骨架**。 必须挑内容最丰富的数组(scenes),而不是见 segments 是 list 就用→落 4 个空镜。""" raw = json.dumps({ "scenes": [ {"sceneIndex": 1, "sceneTitle": "通勤", "voiceover": "手机一卡效率掉线", "visual": "地铁口拿出亮银色机身"}, {"sceneIndex": 2, "sceneTitle": "办公", "voiceover": "A19多任务很跟手", "visual": "办公桌俯拍切换应用"}, ], "segments": [{"index": 0, "role": "钩子", "narration": "", "visual": ""}, {"index": 1, "role": "痛点", "narration": "", "visual": ""}], "total_duration": 30, }, ensure_ascii=False) draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=30) self.assertEqual(draft["segments"][0]["narration"], "手机一卡效率掉线") self.assertEqual(draft["segments"][0]["visual"], "地铁口拿出亮银色机身") def test_script_key_with_visual_object_is_flattened(self): """真实回归:数组键叫 script、visual 写成 {setting,camera,key_shots} 对象。 必须认出 script 数组并把 visual 对象拍平成一句,而非留空。""" raw = json.dumps({ "script": [ {"scene_id": 1, "scene_title": "通勤", "voiceover": "选手机看重流畅好看", "visual": {"setting": "地铁口", "camera": "竖屏手持跟拍", "key_shots": ["拿出亮银色机身"]}}, ], "total_duration": 15, }, ensure_ascii=False) draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=15) seg = draft["segments"][0] self.assertEqual(seg["narration"], "选手机看重流畅好看") self.assertIn("地铁口", seg["visual"]) self.assertIn("竖屏手持跟拍", seg["visual"]) def test_audio_field_fills_narration(self): """真实回归:shots 用 audio 放口播(GPT 变体),旁白曾因 audio 不在词典而留空。""" raw = json.dumps({ "shots": [{"scene": 1, "visual": "咖啡厅办公把玩机身", "audio": "这款亮银色是我的高光决定"}], "total_duration": 15, }, ensure_ascii=False) draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=15) self.assertEqual(draft["segments"][0]["narration"], "这款亮银色是我的高光决定") self.assertEqual(draft["segments"][0]["visual"], "咖啡厅办公把玩机身") def test_canonical_flat_schema_still_works(self): """契约内的扁平 schema(segments/narration/visual)不受兼容改动影响。""" raw = json.dumps({ "total_duration": 30, "segments": [ {"role": "钩子", "narration": "口播1", "visual": "画面1"}, {"role": "CTA", "narration": "口播2", "visual": "画面2"}, ], }, ensure_ascii=False) draft = normalize_draft(raw, aspect_ratio="9:16", total_duration=30) self.assertEqual(len(draft["segments"]), 2) self.assertEqual(draft["segments"][0]["narration"], "口播1") self.assertEqual(draft["segments"][1]["visual"], "画面2") from io import BytesIO from unittest.mock import MagicMock, patch from django.test import TestCase, override_settings from rest_framework.test import APIClient from apps.accounts.models import Team, TeamMember, User from apps.ai.models import AITask, ImageConversation, ModelConfig, ModelProvider from apps.ai.providers.volcano import VolcanoArkProvider from apps.ai.services import enqueue_standalone_images from apps.assets.models import Asset, AssetFile from apps.billing.models import CreditAccount, CreditLedger from apps.products.models import Product class FreeVideoAssetDeletionTests(TestCase): """自由创作视频资产删除后,任务流应降级为无可播成品,不能回退到旧 fallback URL。""" def setUp(self): self.user = User.objects.create_user(username="fvdel", password="pass") self.team = Team.objects.create(name="FreeVideoDel", owner=self.user) TeamMember.objects.create(team=self.team, user=self.user, role="owner", status="active") provider = ModelProvider.objects.create(name="fv-provider", display_name="FV Provider") self.model = ModelConfig.objects.create( provider=provider, name="fv-model", display_name="FV Model", capability=ModelConfig.Capability.VIDEO, ) self.task = AITask.objects.create( team=self.team, created_by=self.user, task_type=AITask.Type.FREE_VIDEO, status=AITask.Status.SUCCEEDED, model_config=self.model, idempotency_key="free-video-delete-test", request_payload={ "prompt": "生成一个广告视频", "fallback_video_url": "http://fallback.example/video.mp4", "fallback_note": "fallback", }, ) self.asset = Asset.objects.create( team=self.team, created_by=self.user, name="自由创作视频", asset_type=Asset.Type.VIDEO, source=Asset.Source.AI_GENERATED, category=Asset.Category.FREE_CREATE, origin_task=self.task, ) AssetFile.objects.create( asset=self.asset, object_key="free-video.mp4", bucket="b", content_type="video/mp4", preview_url="http://tos.example/video.mp4", is_primary=True, ) AssetFile.objects.create( asset=self.asset, object_key="free-video-poster.jpg", bucket="b", content_type="image/jpeg", preview_url="http://tos.example/free-video-poster.jpg", is_primary=False, ) def test_deleted_persisted_asset_does_not_fall_back_to_raw_video_url(self): from apps.ai.free_video import serialize_free_video_task self.assertEqual(serialize_free_video_task(self.task)["video_url"], "http://tos.example/video.mp4") self.asset.is_deleted = True self.asset.save(update_fields=["is_deleted"]) task = AITask.objects.prefetch_related("generated_assets", "generated_assets__files").get(id=self.task.id) data = serialize_free_video_task(task) self.assertEqual(data["video_url"], "") self.assertEqual(data["thumbnail_url"], "") self.assertEqual(data["fallback_note"], "fallback") trash_data = serialize_free_video_task(task, include_deleted_assets=True) self.assertEqual(trash_data["video_url"], "http://tos.example/video.mp4") self.assertEqual(trash_data["thumbnail_url"], "http://tos.example/free-video-poster.jpg") def test_asset_library_uses_full_free_video_title_and_searches_it(self): """资产库应从关联任务恢复历史截断标题，并支持搜索完整标题的后半段。""" full_prompt = "@O1CN01anEq9u245Jn8Ccedn_!!4611686018427385691-0-s-完整后缀" self.task.request_payload["prompt"] = full_prompt self.task.save(update_fields=["request_payload"]) self.asset.name = full_prompt[:50] # 模拟旧版本写库的截断数据 self.asset.save(update_fields=["name"]) client = APIClient() client.force_authenticate(self.user) listing = client.get("/api/assets/?tab=video_creations&page_size=10").json() self.assertEqual(listing["count"], 1) self.assertEqual(listing["results"][0]["display_name"], full_prompt) search = client.get(f"/api/assets/?tab=video_creations&q={full_prompt[-4:]}&page_size=10").json() self.assertEqual([row["id"] for row in search["results"]], [str(self.asset.id)]) def test_direct_asset_delete_stays_in_asset_trash(self): client = APIClient() client.force_authenticate(self.user) delete = client.delete(f"/api/assets/{self.asset.id}/") self.assertEqual(delete.status_code, 204) self.task.refresh_from_db() self.asset.refresh_from_db() self.assertFalse(self.task.is_deleted) self.assertTrue(self.asset.is_deleted) trash_ids = {a["id"] for a in client.get("/api/assets/trash/").json()["results"]} self.assertIn(str(self.asset.id), trash_ids) self.assertEqual(client.get("/api/ai/free-video/").json()["total"], 0) restore = client.post(f"/api/assets/{self.asset.id}/restore/") self.assertEqual(restore.status_code, 200) live_ids = {t["id"] for t in client.get("/api/ai/free-video/").json()["results"]} self.assertIn(str(self.task.id), live_ids) def test_trash_restore_and_purge_free_video_keeps_record(self): self.client = APIClient() self.client.force_authenticate(self.user) delete = self.client.delete(f"/api/ai/free-video/{self.task.id}/") self.assertEqual(delete.status_code, 204) self.task.refresh_from_db() self.asset.refresh_from_db() self.assertTrue(self.task.is_deleted) self.assertTrue(self.asset.is_deleted) self.assertIsNone(self.task.purged_at) self.assertEqual(self.client.get("/api/ai/free-video/").json()["total"], 0) trash_items = self.client.get("/api/ai/free-video/trash/").json()["results"] trash_ids = {t["id"] for t in trash_items} self.assertIn(str(self.task.id), trash_ids) asset_trash_ids = {a["id"] for a in self.client.get("/api/assets/trash/").json()["results"]} self.assertNotIn(str(self.asset.id), asset_trash_ids) trash_item = next(t for t in trash_items if t["id"] == str(self.task.id)) self.assertEqual(trash_item["thumbnail_url"], "http://tos.example/free-video-poster.jpg") restore = self.client.post(f"/api/ai/free-video/{self.task.id}/restore/") self.assertEqual(restore.status_code, 200) self.task.refresh_from_db() self.asset.refresh_from_db() self.assertFalse(self.task.is_deleted) self.assertFalse(self.asset.is_deleted) self.assertIsNone(self.task.purged_at) live_ids = {t["id"] for t in self.client.get("/api/ai/free-video/").json()["results"]} self.assertIn(str(self.task.id), live_ids) self.client.delete(f"/api/ai/free-video/{self.task.id}/") purge = self.client.delete(f"/api/ai/free-video/{self.task.id}/purge/") self.assertEqual(purge.status_code, 204) self.task.refresh_from_db() self.asset.refresh_from_db() self.assertTrue(self.task.is_deleted) self.assertTrue(self.asset.is_deleted) self.assertIsNotNone(self.task.purged_at) self.assertIsNotNone(self.asset.purged_at) self.assertTrue(AITask.objects.filter(id=self.task.id).exists()) self.assertTrue(Asset.objects.filter(id=self.asset.id).exists()) self.assertEqual(self.asset.files.count(), 2) trash_ids = {t["id"] for t in self.client.get("/api/ai/free-video/trash/").json()["results"]} self.assertNotIn(str(self.task.id), trash_ids) class StandaloneImageReferenceTests(TestCase): """独立生图(平台套图 / 模特上身图)必须把商品真实主图(+ 模特图)作为参考图走 image_edit, 而不是纯文生图——回归保护 #图片生成没参考商品主图# 这个 bug。 (图像默认模型由迁移 seed 的 tokenssr:gpt-image-2 提供,get_image_provider 全程 mock。)""" def setUp(self): self.user = User.objects.create_user(username="owner", password="pass") self.team = Team.objects.create(name="T", owner=self.user) CreditAccount.objects.create(team=self.team, balance="100.0000") # 商品 + 主图(带可访问 preview_url) self.product = Product.objects.create(team=self.team, created_by=self.user, title="南卡 Lite Pro") cover = Asset.objects.create( team=self.team, created_by=self.user, name="主图", asset_type=Asset.Type.IMAGE, source=Asset.Source.UPLOAD, category=Asset.Category.PRODUCT_IMAGE, ) AssetFile.objects.create(asset=cover, object_key="c.png", bucket="b", content_type="image/png", preview_url="http://x/cover.png", is_primary=True) self.product.cover_asset = cover self.product.save(update_fields=["cover_asset"]) def _patch_provider(self): # image_edit 返回值 + 媒体落库链路全部 mock,聚焦验证「传了哪些参考图」 provider = patch("apps.ai.services.get_image_provider").start() prov = provider.return_value prov.image_edit.return_value = {"data": [{"url": "http://x/out.png"}]} prov.image_generation.return_value = {"data": [{"url": "http://x/out.png"}]} prov.extract_first_media_url.return_value = "http://x/out.png" media = patch("apps.ai.services.VolcanoArkProvider.media_to_bytes").start() media.return_value = (BytesIO(b"img"), "image/png") store = patch("apps.ai.services.TosStorage").start() stored = store.return_value.upload_fileobj.return_value stored.object_key, stored.bucket, stored.content_type, stored.size_bytes = "o.png", "b", "image/png", 3 self.addCleanup(patch.stopall) return prov def test_cover_mode_references_product_main_image(self): prov = self._patch_provider() enqueue_standalone_images(team=self.team, user=self.user, prompt="平台套图", mode="cover", count=1, product_id=str(self.product.id), ratio="4:5") prov.image_edit.assert_called_once() self.assertEqual(prov.image_edit.call_args.kwargs["images"], ["http://x/cover.png"]) prov.image_generation.assert_not_called() def test_model_tryon_combines_product_and_model_images(self): prov = self._patch_provider() # 模特资产(PERSON)带 preview_url model_asset = Asset.objects.create( team=self.team, created_by=self.user, name="模特", asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PERSON, ) AssetFile.objects.create(asset=model_asset, object_key="m.png", bucket="b", content_type="image/png", preview_url="http://x/model.png", is_primary=True) enqueue_standalone_images(team=self.team, user=self.user, prompt="上身图", mode="model", count=1, product_id=str(self.product.id), model_id=str(model_asset.id), ratio="4:5") prov.image_edit.assert_called_once() self.assertEqual(prov.image_edit.call_args.kwargs["images"], ["http://x/cover.png", "http://x/model.png"]) prov.image_generation.assert_not_called() @override_settings(MODEL_TRYON_PROMPT_V2_ENABLED=True) def test_model_tryon_v2_uses_structured_prompt_and_persists_trace(self): prov = self._patch_provider() self.product.title = "中蓝色高腰宽腿牛仔裤" self.product.category = "服饰内衣" self.product.save(update_fields=["title", "category"]) self.product.selling_points.create(title="显高显瘦", sort_order=0) model_asset = Asset.objects.create( team=self.team, created_by=self.user, name="模特", asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PERSON, ) AssetFile.objects.create( asset=model_asset, object_key="m-v2.png", bucket="b", content_type="image/png", preview_url="http://x/model-v2.png", is_primary=True, ) user_prompt = "参考图1的女生穿上图2的牛仔裤，全身站姿，双腿完整露出，浅灰纯色背景" submitted = enqueue_standalone_images( team=self.team, user=self.user, prompt=user_prompt, mode="model", count=2, product_id=str(self.product.id), model_id=str(model_asset.id), ratio="7:10", ) tasks = list(AITask.objects.filter(id__in=[task.id for task in submitted]).order_by("request_payload__index")) self.assertEqual(len(prov.image_edit.call_args_list), 2) used_prompts = [call.kwargs["prompt"] for call in prov.image_edit.call_args_list] used_sizes = [call.kwargs["size"] for call in prov.image_edit.call_args_list] self.assertTrue(all(user_prompt in prompt for prompt in used_prompts)) self.assertTrue(all("参考图角色（固定语义）" in prompt for prompt in used_prompts)) self.assertTrue(all("严格使用 7:10 比例" in prompt for prompt in used_prompts)) self.assertTrue(all("半身近景" not in prompt and "手持或局部" not in prompt for prompt in used_prompts)) self.assertNotEqual(used_prompts[0], used_prompts[1]) self.assertEqual(used_sizes, ["1024x1536", "1024x1536"]) for index, task in enumerate(tasks): trace = task.request_payload["tryon_prompt"] self.assertTrue(trace["applied"]) self.assertEqual(trace["version"], "v2.6") self.assertEqual(trace["rollout_source"], "global") self.assertEqual(trace["effective_prompt"], used_prompts[index]) self.assertEqual(trace["shot_index"], index) self.assertEqual(trace["batch_count"], 2) self.assertEqual(trace["requested_ratio"], "7:10") self.assertEqual(trace["resolved_ratio"], "7:10") self.assertEqual(trace["reference_roles"]["product_numbers"], [1]) self.assertEqual(trace["reference_roles"]["model_portrait_number"], 2) self.assertIsNone(trace["reference_roles"]["model_triview_number"]) self.assertEqual(task.request_payload["tryon_classification"]["kind"], "lower") self.assertEqual(trace["trouser_facts"], task.request_payload["tryon_trouser_facts"]) @override_settings(MODEL_TRYON_PROMPT_V2_ENABLED=True) def test_model_tryon_v26_zero_cost_model_count_ratio_matrix(self): """Mock 全矩阵：两个模型都覆盖 1/2/4 张与预设/自定义比例，不调用真实供应商。""" CreditAccount.objects.filter(team=self.team).update(balance="5000.0000") self.product.title = "舒适柔软面料女裤" self.product.category = "服饰内衣" self.product.save(update_fields=["title", "category"]) model_asset = Asset.objects.create( team=self.team, created_by=self.user, name="矩阵模特", asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PERSON, ) AssetFile.objects.create( asset=model_asset, object_key="matrix-model.png", bucket="b", content_type="image/png", preview_url="http://x/matrix-model.png", is_primary=True, ) volcano, _ = ModelProvider.objects.get_or_create( name="volcengine", defaults={"display_name": "火山"}, ) seedream_config = ModelConfig.objects.create( provider=volcano, name="seedream-v25-matrix", display_name="Seedream V2.5 matrix", capability=ModelConfig.Capability.IMAGE, unit_price="1.0000", ) gpt_config = ModelConfig.objects.filter( capability=ModelConfig.Capability.IMAGE, name__icontains="gpt-image", ).first() self.assertIsNotNone(gpt_config) seedream = MagicMock() del seedream.image_edit seedream.image_generation.return_value = {"data": [{"url": "http://x/seedream-matrix.png"}]} seedream.extract_first_media_url.return_value = "http://x/seedream-matrix.png" gpt = MagicMock() gpt.image_edit.return_value = {"data": [{"url": "http://x/gpt-matrix.png"}]} gpt.extract_first_media_url.return_value = "http://x/gpt-matrix.png" def provider_for(model_config, *args, **kwargs): del args, kwargs return seedream if model_config.id == seedream_config.id else gpt stored = MagicMock(object_key="matrix.png", bucket="b", content_type="image/png", size_bytes=3) ratios = ("1:1", "3:4", "9:16", "7:10") models = ( (seedream_config, seedream.image_generation, "image"), (gpt_config, gpt.image_edit, "images"), ) with ( patch("apps.ai.services.get_image_provider", side_effect=provider_for), patch( "apps.ai.services.VolcanoArkProvider.media_to_bytes", return_value=(BytesIO(b"img"), "image/png"), ), patch("apps.ai.services.TosStorage") as storage, ): storage.return_value.upload_fileobj.return_value = stored for model_config, provider_method, reference_key in models: for count in (1, 2, 4): for ratio in ratios: with self.subTest(model=model_config.name, count=count, ratio=ratio): provider_method.reset_mock() tasks = enqueue_standalone_images( team=self.team, user=self.user, prompt="模特上身展示，自然光，真实质感，电商主图", mode="model", count=count, product_id=str(self.product.id), model_id=str(model_asset.id), ratio=ratio, image_model=f"{model_config.provider.name}:{model_config.name}", ) self.assertEqual(len(tasks), count) self.assertEqual(len(provider_method.call_args_list), count) for index, (task, call) in enumerate(zip(tasks, provider_method.call_args_list)): task.refresh_from_db() trace = task.request_payload["tryon_prompt"] effective_prompt = call.kwargs["prompt"] self.assertEqual(task.model_config_id, model_config.id) self.assertEqual(trace["version"], "v2.6") self.assertTrue(trace["applied"]) self.assertEqual(trace["rollout_source"], "global") self.assertEqual(trace["shot_index"], index) self.assertEqual(trace["batch_count"], count) self.assertEqual(trace["requested_ratio"], ratio) self.assertEqual(trace["resolved_ratio"], ratio) self.assertEqual(trace["ratio_source"], "user") self.assertEqual(trace["effective_prompt"], effective_prompt) self.assertEqual(trace["reference_roles"]["product_numbers"], [1]) self.assertEqual(trace["reference_roles"]["model_portrait_number"], 2) self.assertEqual(trace["trouser_facts"], task.request_payload["tryon_trouser_facts"]) self.assertEqual(trace["trouser_facts"]["length"], "unknown") self.assertEqual(trace["trouser_facts"]["silhouette"], "unknown") self.assertIn(f"严格使用 {ratio} 比例", effective_prompt) self.assertIn("裤装几何保真", effective_prompt) self.assertNotIn("不得在九分裤、长裤、拖地裤之间变化", effective_prompt) self.assertNotIn("背面", effective_prompt) if count == 1: self.assertIn("正面标准全身", effective_prompt) elif count == 2: expected_role = "正面标准全身" if index == 0 else "正面人物中景" self.assertIn(expected_role, effective_prompt) else: expected_role = "全身" if index < 2 else "人物中景" self.assertIn(expected_role, effective_prompt) self.assertEqual( call.kwargs[reference_key], ["http://x/cover.png", "http://x/matrix-model.png"], ) for model_config, provider_method, _reference_key in models: for count in (1, 2, 4): with self.subTest( model=model_config.name, count=count, default_ratio=True, ): provider_method.reset_mock() tasks = enqueue_standalone_images( team=self.team, user=self.user, prompt="模特上身展示，自然光，真实质感，电商主图", mode="model", count=count, product_id=str(self.product.id), model_id=str(model_asset.id), image_model=f"{model_config.provider.name}:{model_config.name}", ) self.assertEqual(len(tasks), count) for task, call in zip(tasks, provider_method.call_args_list): task.refresh_from_db() trace = task.request_payload["tryon_prompt"] self.assertIsNone(trace["requested_ratio"]) self.assertEqual(trace["resolved_ratio"], "3:4") self.assertEqual(trace["ratio_source"], "default") self.assertIn("严格使用 3:4 比例", call.kwargs["prompt"]) for model_config, provider_method, _reference_key in models: for count in (1, 2, 4): for ratio in ratios: with self.subTest( model=model_config.name, count=count, ratio=ratio, explicit_full_body=True, ): provider_method.reset_mock() tasks = enqueue_standalone_images( team=self.team, user=self.user, prompt="每张全身站姿，从头到脚完整展示，双腿完整露出", mode="model", count=count, product_id=str(self.product.id), model_id=str(model_asset.id), ratio=ratio, image_model=f"{model_config.provider.name}:{model_config.name}", ) prompts = [call.kwargs["prompt"] for call in provider_method.call_args_list] self.assertEqual(len(tasks), count) self.assertEqual(len(prompts), count) for task, effective_prompt in zip(tasks, prompts): task.refresh_from_db() trace = task.request_payload["tryon_prompt"] self.assertEqual(task.model_config_id, model_config.id) self.assertEqual(trace["version"], "v2.6") self.assertEqual(trace["resolved_ratio"], ratio) self.assertIn(f"严格使用 {ratio} 比例", effective_prompt) self.assertIn("全身站姿", effective_prompt) self.assertIn("从头到脚完整入镜", effective_prompt) self.assertNotIn("人物中景", effective_prompt) self.assertNotIn("背面", effective_prompt) self.assertNotIn("不得在九分裤、长裤、拖地裤之间变化", effective_prompt) for model_config, provider_method, _reference_key in models: for count in (1, 2, 4): for ratio in ratios: with self.subTest( model=model_config.name, count=count, ratio=ratio, explicit_back=True, ): provider_method.reset_mock() tasks = enqueue_standalone_images( team=self.team, user=self.user, prompt="模特上身展示，完整展示后袋", mode="model", count=count, product_id=str(self.product.id), model_id=str(model_asset.id), ratio=ratio, image_model=f"{model_config.provider.name}:{model_config.name}", ) prompts = [call.kwargs["prompt"] for call in provider_method.call_args_list] self.assertEqual(len(tasks), count) self.assertEqual(len(prompts), count) self.assertTrue(all("背面全身站姿" not in prompt for prompt in prompts[:-1])) self.assertIn("背面全身站姿", prompts[-1]) for task, effective_prompt in zip(tasks, prompts): task.refresh_from_db() trace = task.request_payload["tryon_prompt"] self.assertEqual(task.model_config_id, model_config.id) self.assertEqual(trace["version"], "v2.6") self.assertEqual(trace["resolved_ratio"], ratio) self.assertIn(f"严格使用 {ratio} 比例", effective_prompt) if count == 2: self.assertTrue(all("全身" in prompt for prompt in prompts)) if count == 4: self.assertTrue(all("全身" in prompt for prompt in prompts[:2])) self.assertIn("人物中景", prompts[2]) cross_category_cases = ( ("高腰半身裙", "服饰内衣", "lower"), ("宽松针织上衣", "服饰内衣", "upper"), ("法式收腰连衣裙", "女装", "full_body"), ("女式运动鞋", "鞋靴", "footwear"), ("智能运动手表", "数码配件", "wrist"), ("便携无线小风扇", "家电", "handheld"), ) for title, category, expected_kind in cross_category_cases: self.product.title = title self.product.category = category self.product.save(update_fields=["title", "category"]) for model_config, provider_method, _reference_key in models: with self.subTest( model=model_config.name, product=title, cross_category=True, ): provider_method.reset_mock() tasks = enqueue_standalone_images( team=self.team, user=self.user, prompt="模特上身展示，自然光，真实质感，电商主图", mode="model", count=4, product_id=str(self.product.id), model_id=str(model_asset.id), ratio="3:4", image_model=f"{model_config.provider.name}:{model_config.name}", ) prompts = [call.kwargs["prompt"] for call in provider_method.call_args_list] self.assertEqual(len(tasks), 4) self.assertEqual(len(prompts), 4) self.assertTrue(all(task.model_config_id == model_config.id for task in tasks)) self.assertTrue( all( task.request_payload["tryon_classification"]["kind"] == expected_kind for task in tasks ) ) self.assertTrue( all("tryon_trouser_facts" not in task.request_payload for task in tasks) ) self.assertNotIn("裤装几何保真", "\n".join(prompts)) self.assertNotIn("人物中景", "\n".join(prompts)) @override_settings( MODEL_TRYON_PROMPT_V2_ENABLED=False, MODEL_TRYON_PROMPT_V2_CANARY_TEAM_IDS=frozenset(), ) def test_model_tryon_flag_off_keeps_legacy_prompt(self): prov = self._patch_provider() self.product.title = "女装裤子" self.product.save(update_fields=["title"]) model_asset = Asset.objects.create( team=self.team, created_by=self.user, name="模特", asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PERSON, ) AssetFile.objects.create( asset=model_asset, object_key="m-legacy.png", bucket="b", content_type="image/png", preview_url="http://x/model-legacy.png", is_primary=True, ) submitted = enqueue_standalone_images( team=self.team, user=self.user, prompt="模特上身图", mode="model", count=1, product_id=str(self.product.id), model_id=str(model_asset.id), ratio="3:4", ) used_prompt = prov.image_edit.call_args.kwargs["prompt"] self.assertIn("半身近景", used_prompt) task = AITask.objects.get(id=submitted[0].id) self.assertNotIn("tryon_prompt", task.request_payload) @override_settings(MODEL_TRYON_PROMPT_V2_ENABLED=False) def test_model_tryon_canary_team_uses_v22_and_records_source(self): prov = self._patch_provider() self.product.title = "女装裤子" self.product.save(update_fields=["title"]) with self.settings(MODEL_TRYON_PROMPT_V2_CANARY_TEAM_IDS={str(self.team.id).upper()}): submitted = enqueue_standalone_images( team=self.team, user=self.user, prompt="全身站姿，双腿完整露出", mode="model", count=1, product_id=str(self.product.id), ratio="3:4", ) used_prompt = prov.image_edit.call_args.kwargs["prompt"] self.assertIn("商品关键结构优先", used_prompt) task = AITask.objects.get(id=submitted[0].id) trace = task.request_payload["tryon_prompt"] self.assertTrue(trace["applied"]) self.assertEqual(trace["version"], "v2.6") self.assertEqual(trace["rollout_source"], "canary") @override_settings( MODEL_TRYON_PROMPT_V2_ENABLED=False, MODEL_TRYON_PROMPT_V2_CANARY_TEAM_IDS={"not-a-team-uuid"}, ) def test_model_tryon_non_canary_team_fails_closed_to_legacy_prompt(self): prov = self._patch_provider() self.product.title = "女装裤子" self.product.save(update_fields=["title"]) submitted = enqueue_standalone_images( team=self.team, user=self.user, prompt="模特上身图", mode="model", count=1, product_id=str(self.product.id), ratio="3:4", ) used_prompt = prov.image_edit.call_args.kwargs["prompt"] self.assertIn("半身近景", used_prompt) task = AITask.objects.get(id=submitted[0].id) self.assertNotIn("tryon_prompt", task.request_payload) @override_settings(MODEL_TRYON_PROMPT_V2_ENABLED=False) def test_model_tryon_canary_without_product_reference_records_safe_fallback(self): prov = self._patch_provider() self.product.cover_asset = None self.product.save(update_fields=["cover_asset"]) with self.settings(MODEL_TRYON_PROMPT_V2_CANARY_TEAM_IDS={str(self.team.id)}): submitted = enqueue_standalone_images( team=self.team, user=self.user, prompt="模特上身图", mode="model", count=1, product_id=str(self.product.id), ratio="3:4", ) prov.image_generation.assert_called_once() task = AITask.objects.get(id=submitted[0].id) trace = task.request_payload["tryon_prompt"] self.assertFalse(trace["applied"]) self.assertEqual(trace["rollout_source"], "canary") self.assertEqual(trace["fallback_reason"], "no_product_reference") @override_settings(MODEL_TRYON_PROMPT_V2_ENABLED=True) def test_model_tryon_v22_uses_category_default_ratio_when_user_omits_ratio(self): prov = self._patch_provider() self.product.title = "女装裤子" self.product.save(update_fields=["title"]) submitted = enqueue_standalone_images( team=self.team, user=self.user, prompt="模特上身图", mode="model", count=1, product_id=str(self.product.id), ) self.assertEqual(prov.image_edit.call_args.kwargs["size"], "1024x1536") task = AITask.objects.get(id=submitted[0].id) trace = task.request_payload["tryon_prompt"] self.assertEqual(trace["requested_ratio"], None) self.assertEqual(trace["resolved_ratio"], "3:4") self.assertEqual(trace["ratio_source"], "default") self.assertEqual(trace["rollout_source"], "global") @override_settings(MODEL_TRYON_PROMPT_V2_ENABLED=True) def test_model_tryon_old_queued_task_without_v2_metadata_falls_back_safely(self): from apps.ai.services import run_standalone_image_task prov = self._patch_provider() self.product.title = "女装裤子" self.product.save(update_fields=["title"]) with patch("apps.ai.tasks.generate_standalone_image_task.delay"): task = enqueue_standalone_images( team=self.team, user=self.user, prompt="模特上身图", mode="model", count=1, product_id=str(self.product.id), ratio="3:4", )[0] legacy_payload = dict(task.request_payload) legacy_payload.pop("tryon_batch_count", None) legacy_payload.pop("tryon_classification", None) task.request_payload = legacy_payload task.save(update_fields=["request_payload", "updated_at"]) run_standalone_image_task(task_id=str(task.id)) task.refresh_from_db() used_prompt = prov.image_edit.call_args.kwargs["prompt"] self.assertIn("半身近景", used_prompt) self.assertEqual(task.status, AITask.Status.SUCCEEDED) trace = task.request_payload["tryon_prompt"] self.assertFalse(trace["applied"]) self.assertEqual(trace["rollout_source"], "global") self.assertEqual(trace["fallback_reason"], "unsupported_tryon_batch_count") def test_cover_mode_falls_back_to_t2i_without_main_image(self): prov = self._patch_provider() self.product.cover_asset = None self.product.save(update_fields=["cover_asset"]) enqueue_standalone_images(team=self.team, user=self.user, prompt="平台套图", mode="cover", count=1, product_id=str(self.product.id), ratio="4:5") prov.image_generation.assert_called_once() prov.image_edit.assert_not_called() def test_image_mode_uses_uploaded_reference_images(self): """图片创作自由模式:用户上传的参考图必须作为 image_edit 的参考图传进去, 而不是被忽略走纯文生图——回归保护 #图片创作没参考我上传的素材# 这个最致命的 bug。""" prov = self._patch_provider() ref = Asset.objects.create( team=self.team, created_by=self.user, name="背心参考", asset_type=Asset.Type.IMAGE, source=Asset.Source.UPLOAD, category=Asset.Category.UPLOAD, ) AssetFile.objects.create(asset=ref, object_key="r.png", bucket="b", content_type="image/png", preview_url="http://x/ref.png", is_primary=True) enqueue_standalone_images( team=self.team, user=self.user, prompt="按这件背心生成不同场景的穿搭", mode="image", count=1, ratio="1:1", reference_image_ids=[str(ref.id)], ) prov.image_edit.assert_called_once() self.assertEqual(prov.image_edit.call_args.kwargs["images"], ["http://x/ref.png"]) # 提示词必须显式钉住「参考图=主体唯一依据」+ 带上用户原话(否则图生图不真的还原上传素材); # 但不能再无差别「严格保留款式」——用户要求(如换现代服装)必须是最高优先级,否则同批一半不换装 used_prompt = prov.image_edit.call_args.kwargs["prompt"] self.assertIn("按这件背心生成不同场景的穿搭", used_prompt) self.assertIn("唯一依据", used_prompt) self.assertIn("用户要求是最高优先级", used_prompt) prov.image_generation.assert_not_called() def test_refs_keep_user_selected_volcano_model_as_primary(self): """图片创作上传参考图后仍必须尊重用户选择的 Seedream；它可走带 image 的图生图， 只有真实调用失败后才允许统一路由按配置 Fallback，不能在建任务阶段静默换成 gpt-image。""" from apps.ai.models import ModelProvider vp, _ = ModelProvider.objects.get_or_create(name="volcengine", defaults={"display_name": "火山"}) ModelConfig.objects.create(provider=vp, name="seedream-4", display_name="Seedream", capability=ModelConfig.Capability.IMAGE) self._patch_provider() ref = Asset.objects.create( team=self.team, created_by=self.user, name="背心", asset_type=Asset.Type.IMAGE, source=Asset.Source.UPLOAD, category=Asset.Category.UPLOAD, ) AssetFile.objects.create(asset=ref, object_key="r.png", bucket="b", content_type="image/png", preview_url="http://x/ref.png", is_primary=True) tasks = enqueue_standalone_images( team=self.team, user=self.user, prompt="不同场景", mode="image", count=1, image_model="volcano", reference_image_ids=[str(ref.id)], ) self.assertEqual(tasks[0].model_config.provider_id, vp.id) self.assertIn("seedream", tasks[0].model_config.name) def test_model_tryon_keeps_each_user_selected_model_even_with_extra_refs(self): """模特上身图由专用 Worker 向两类模型传商品/人物参考图，不能沿用自由创作的自动换模逻辑。""" vp, _ = ModelProvider.objects.get_or_create( name="volcengine", defaults={"display_name": "火山"}, ) seedream = ModelConfig.objects.create( provider=vp, name="seedream-v22-user-choice", display_name="Seedream V2.2 user choice", capability=ModelConfig.Capability.IMAGE, ) gpt = ModelConfig.objects.filter( capability=ModelConfig.Capability.IMAGE, name__icontains="gpt-image", ).first() self.assertIsNotNone(gpt) extra_ref = Asset.objects.create( team=self.team, created_by=self.user, name="额外参考", asset_type=Asset.Type.IMAGE, source=Asset.Source.UPLOAD, category=Asset.Category.UPLOAD, ) AssetFile.objects.create( asset=extra_ref, object_key="extra.png", bucket="b", content_type="image/png", preview_url="http://x/extra.png", is_primary=True, ) seedream_task = enqueue_standalone_images( team=self.team, user=self.user, prompt="模特上身图", mode="model", count=1, product_id=str(self.product.id), image_model=f"{vp.name}:{seedream.name}", reference_image_ids=[str(extra_ref.id)], dispatch=False, )[0] gpt_task = enqueue_standalone_images( team=self.team, user=self.user, prompt="模特上身图", mode="model", count=1, product_id=str(self.product.id), image_model=f"{gpt.provider.name}:{gpt.name}", reference_image_ids=[str(extra_ref.id)], dispatch=False, )[0] self.assertEqual(seedream_task.model_config_id, seedream.id) self.assertEqual(gpt_task.model_config_id, gpt.id) class ImageConversationTests(TestCase): """图片创作「对话」实体:CRUD + 团队隔离 + 软删 + 生图自动归属对话 + 任务回填。""" def setUp(self): self.user = User.objects.create_user(username="convowner", password="pass") self.team = Team.objects.create(name="ConvT", owner=self.user) TeamMember.objects.create(team=self.team, user=self.user, role="owner", status="active") self.client = APIClient() self.client.force_authenticate(self.user) def test_crud_and_listing(self): # 新建 r = self.client.post("/api/ai/image-conversations/", {"mode": "image", "title": "测试创作"}, format="json") self.assertEqual(r.status_code, 201, r.content) conv_id = r.json()["id"] # 列出(只 image 模式) r = self.client.get("/api/ai/image-conversations/?mode=image") self.assertEqual(r.status_code, 200) self.assertEqual(len(r.json()["results"]), 1) # 重命名 r = self.client.patch(f"/api/ai/image-conversations/{conv_id}/", {"title": "改后"}, format="json") self.assertEqual(r.status_code, 200) self.assertEqual(r.json()["title"], "改后") # 软删:列表消失,但 DB 记录仍在(is_deleted=True) r = self.client.delete(f"/api/ai/image-conversations/{conv_id}/") self.assertIn(r.status_code, (204, 200)) self.assertEqual(self.client.get("/api/ai/image-conversations/?mode=image").json()["results"], []) self.assertTrue(ImageConversation.objects.get(id=conv_id).is_deleted) def test_listing_filters_product_and_unbound_scopes(self): product = Product.objects.create(team=self.team, created_by=self.user, title="范围商品") unbound = ImageConversation.objects.create(team=self.team, created_by=self.user, title="通用会话") bound = ImageConversation.objects.create(team=self.team, created_by=self.user, title="商品会话", product=product) unbound_rows = self.client.get( "/api/ai/image-conversations/?mode=image&scope=unbound" ).json()["results"] product_rows = self.client.get( f"/api/ai/image-conversations/?mode=image&product_id={product.id}" ).json()["results"] self.assertEqual([row["id"] for row in unbound_rows], [str(unbound.id)]) self.assertEqual([row["id"] for row in product_rows], [str(bound.id)]) conflict = self.client.get( f"/api/ai/image-conversations/?mode=image&scope=unbound&product_id={product.id}" ) self.assertEqual(conflict.status_code, 400) def test_conversation_scope_rejects_cross_team_product(self): other = User.objects.create_user(username="conv-other", password="pass") other_team = Team.objects.create(name="Conv Other", owner=other) other_product = Product.objects.create(team=other_team, created_by=other, title="其他团队商品") listed = self.client.get( f"/api/ai/image-conversations/?mode=image&product_id={other_product.id}" ) created = self.client.post( "/api/ai/image-conversations/", {"mode": "image", "title": "越权会话", "product": str(other_product.id)}, format="json", ) self.assertEqual(listed.status_code, 400) self.assertEqual(created.status_code, 400) def test_trash_restore_and_purge_conversation_keeps_record(self): conv = ImageConversation.objects.create(team=self.team, created_by=self.user, title="待删", mode=ImageConversation.Mode.IMAGE) provider = ModelProvider.objects.create(name="conv-trash-provider", display_name="Conv Trash Provider") model = ModelConfig.objects.create( provider=provider, name="conv-trash-model", display_name="Conv Trash Model", capability=ModelConfig.Capability.IMAGE, ) task = AITask.objects.create( team=self.team, created_by=self.user, conversation=conv, task_type=AITask.Type.PRODUCT_IMAGE, status=AITask.Status.SUCCEEDED, model_config=model, idempotency_key="conv-trash-task", ) asset = Asset.objects.create( team=self.team, created_by=self.user, name="会话生成图", asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.FREE_CREATE, origin_task=task, ) asset_file = AssetFile.objects.create( asset=asset, object_key="free-create/image.png", bucket="test", content_type="image/png", is_primary=True, ) delete = self.client.delete(f"/api/ai/image-conversations/{conv.id}/") self.assertIn(delete.status_code, (204, 200)) conv.refresh_from_db() task.refresh_from_db() asset.refresh_from_db() self.assertTrue(conv.is_deleted) self.assertTrue(task.is_deleted) self.assertTrue(asset.is_deleted) self.assertIsNone(conv.purged_at) trash_ids = {c["id"] for c in self.client.get("/api/ai/image-conversations/trash/?mode=image").json()["results"]} self.assertIn(str(conv.id), trash_ids) asset_trash_ids = {a["id"] for a in self.client.get("/api/assets/trash/").json()["results"]} self.assertNotIn(str(asset.id), asset_trash_ids) restore = self.client.post(f"/api/ai/image-conversations/{conv.id}/restore/") self.assertEqual(restore.status_code, 200) conv.refresh_from_db() task.refresh_from_db() asset.refresh_from_db() self.assertFalse(conv.is_deleted) self.assertFalse(task.is_deleted) self.assertFalse(asset.is_deleted) self.assertIsNone(conv.purged_at) live_ids = {c["id"] for c in self.client.get("/api/ai/image-conversations/?mode=image").json()["results"]} self.assertIn(str(conv.id), live_ids) self.client.delete(f"/api/ai/image-conversations/{conv.id}/") purge = self.client.delete(f"/api/ai/image-conversations/{conv.id}/purge/") self.assertEqual(purge.status_code, 204) conv.refresh_from_db() task.refresh_from_db() asset.refresh_from_db() self.assertTrue(conv.is_deleted) self.assertTrue(task.is_deleted) self.assertTrue(asset.is_deleted) self.assertIsNotNone(conv.purged_at) self.assertIsNotNone(task.purged_at) self.assertIsNotNone(asset.purged_at) self.assertTrue(ImageConversation.objects.filter(id=conv.id).exists()) self.assertTrue(AITask.objects.filter(id=task.id).exists()) self.assertTrue(Asset.objects.filter(id=asset.id).exists()) self.assertTrue(AssetFile.objects.filter(id=asset_file.id).exists()) trash_ids = {c["id"] for c in self.client.get("/api/ai/image-conversations/trash/?mode=image").json()["results"]} self.assertNotIn(str(conv.id), trash_ids) def test_conversation_trash_uses_latest_generated_image_as_preview(self): conv = ImageConversation.objects.create( team=self.team, created_by=self.user, title="多资产对话", mode=ImageConversation.Mode.IMAGE, ) provider = ModelProvider.objects.create(name="conv-cover-provider", display_name="Conv Cover Provider") model = ModelConfig.objects.create( provider=provider, name="conv-cover-model", display_name="Conv Cover Model", capability=ModelConfig.Capability.IMAGE, ) assets = [] for index, preview_url in enumerate(("http://x/older.png", "http://x/latest.png")): task = AITask.objects.create( team=self.team, created_by=self.user, conversation=conv, task_type=AITask.Type.PRODUCT_IMAGE, status=AITask.Status.SUCCEEDED, model_config=model, idempotency_key=f"conv-cover-task-{index}", ) asset = Asset.objects.create( team=self.team, created_by=self.user, name=f"对话生成图 {index}", asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.FREE_CREATE, origin_task=task, ) AssetFile.objects.create( asset=asset, object_key=f"free-create/{index}.png", bucket="test", content_type="image/png", preview_url=preview_url, is_primary=True, ) assets.append(asset) delete = self.client.delete(f"/api/ai/image-conversations/{conv.id}/") self.assertIn(delete.status_code, (204, 200)) rows = self.client.get("/api/ai/image-conversations/trash/?mode=image").json()["results"] self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["cover_preview_url"], "http://x/latest.png") self.assertEqual(Asset.objects.filter(id__in=[asset.id for asset in assets]).count(), 2) def test_conversation_trash_without_generated_image_has_empty_preview(self): conv = ImageConversation.objects.create( team=self.team, created_by=self.user, title="无图对话", mode=ImageConversation.Mode.IMAGE, ) delete = self.client.delete(f"/api/ai/image-conversations/{conv.id}/") self.assertIn(delete.status_code, (204, 200)) rows = self.client.get("/api/ai/image-conversations/trash/?mode=image").json()["results"] self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["cover_preview_url"], "") def test_direct_asset_delete_hides_image_but_keeps_conversation(self): conv = ImageConversation.objects.create( team=self.team, created_by=self.user, title="资产独立删除", mode=ImageConversation.Mode.IMAGE, ) provider = ModelProvider.objects.create(name="asset-delete-provider", display_name="Asset Delete Provider") model = ModelConfig.objects.create( provider=provider, name="asset-delete-model", display_name="Asset Delete Model", capability=ModelConfig.Capability.IMAGE, ) task = AITask.objects.create( team=self.team, created_by=self.user, conversation=conv, task_type=AITask.Type.PRODUCT_IMAGE, status=AITask.Status.SUCCEEDED, model_config=model, idempotency_key="direct-asset-delete-image", ) asset = Asset.objects.create( team=self.team, created_by=self.user, name="自由创作图片", asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.FREE_CREATE, origin_task=task, ) delete = self.client.delete(f"/api/assets/{asset.id}/") self.assertEqual(delete.status_code, 204) conv.refresh_from_db() task.refresh_from_db() self.assertFalse(conv.is_deleted) self.assertFalse(task.is_deleted) conversation_ids = { c["id"] for c in self.client.get("/api/ai/image-conversations/?mode=image").json()["results"] } self.assertIn(str(conv.id), conversation_ids) deleted_tasks = self.client.get(f"/api/ai/image-conversations/{conv.id}/tasks/").json()["tasks"] self.assertEqual(deleted_tasks[0]["assets"], []) restore = self.client.post(f"/api/assets/{asset.id}/restore/") self.assertEqual(restore.status_code, 200) restored_tasks = self.client.get(f"/api/ai/image-conversations/{conv.id}/tasks/").json()["tasks"] self.assertEqual([a["id"] for a in restored_tasks[0]["assets"]], [str(asset.id)]) def test_team_isolation(self): other = User.objects.create_user(username="other", password="pass") other_team = Team.objects.create(name="Other", owner=other) ImageConversation.objects.create(team=other_team, created_by=other, title="别人的") r = self.client.get("/api/ai/image-conversations/?mode=image") self.assertEqual(r.json()["results"], []) def test_generate_auto_creates_and_binds_conversation(self): # 余额 + 默认图像模型(迁移已 seed),provider/worker 全 mock,聚焦验证「任务挂到对话」 CreditAccount.objects.create(team=self.team, balance="100.0000") patch("apps.ai.tasks.generate_standalone_image_task.delay").start() self.addCleanup(patch.stopall) conv = ImageConversation.objects.create(team=self.team, created_by=self.user, title="目标对话") tasks = enqueue_standalone_images(team=self.team, user=self.user, prompt="一只猫", mode="image", count=2, conversation=conv) self.assertEqual(len(tasks), 2) self.assertTrue(all(t.conversation_id == conv.id for t in tasks)) self.assertEqual(conv.tasks.count(), 2) def test_image_generate_without_product_keeps_conversation_and_task_unbound(self): """图片生成首页进入的图片创作不应静默补入当前/首个商品。""" CreditAccount.objects.create(team=self.team, balance="100.0000") Product.objects.create(team=self.team, created_by=self.user, title="不应自动关联的商品") with patch("apps.ai.tasks.generate_standalone_image_task.delay"): response = self.client.post( "/api/ai/generate-image/", {"prompt": "一只宇航猫", "mode": "image", "count": 1}, format="json", ) self.assertEqual(response.status_code, 202, response.content) conversation = ImageConversation.objects.get(id=response.json()["conversation_id"]) task = AITask.objects.get(conversation=conversation) self.assertIsNone(conversation.product_id) self.assertIsNone(task.request_payload.get("product_id")) def test_image_generate_with_explicit_product_keeps_product_binding(self): """商品详情显式进入图片创作时，仍保留商品归属。""" CreditAccount.objects.create(team=self.team, balance="100.0000") product = Product.objects.create(team=self.team, created_by=self.user, title="显式关联商品") with patch("apps.ai.tasks.generate_standalone_image_task.delay"): response = self.client.post( "/api/ai/generate-image/", {"prompt": "商品海报", "mode": "image", "count": 1, "product_id": str(product.id)}, format="json", ) self.assertEqual(response.status_code, 202, response.content) conversation = ImageConversation.objects.get(id=response.json()["conversation_id"]) task = AITask.objects.get(conversation=conversation) self.assertEqual(conversation.product_id, product.id) self.assertEqual(task.request_payload.get("product_id"), str(product.id)) def test_new_generation_with_mismatched_conversation_creates_correct_scope(self): CreditAccount.objects.create(team=self.team, balance="100.0000") original_product = Product.objects.create(team=self.team, created_by=self.user, title="原商品") target_product = Product.objects.create(team=self.team, created_by=self.user, title="目标商品") original_conversation = ImageConversation.objects.create( team=self.team, created_by=self.user, title="原商品会话", product=original_product, ) with patch("apps.ai.tasks.generate_standalone_image_task.delay"): response = self.client.post( "/api/ai/generate-image/", { "prompt": "目标商品海报", "mode": "image", "count": 1, "product_id": str(target_product.id), "conversation_id": str(original_conversation.id), }, format="json", ) self.assertEqual(response.status_code, 202, response.content) self.assertNotEqual(response.json()["conversation_id"], str(original_conversation.id)) created_conversation = ImageConversation.objects.get(id=response.json()["conversation_id"]) self.assertEqual(created_conversation.product_id, target_product.id) self.assertEqual(original_conversation.tasks.count(), 0) self.assertEqual(created_conversation.tasks.get().request_payload.get("product_id"), str(target_product.id)) def test_tasks_endpoint_returns_grouped_history(self): product = Product.objects.create(team=self.team, created_by=self.user, title="历史商品") conv = ImageConversation.objects.create(team=self.team, created_by=self.user, title="历史", product=product) mc = ModelConfig.objects.filter(capability=ModelConfig.Capability.IMAGE).first() AITask.objects.create( team=self.team, created_by=self.user, conversation=conv, task_type=AITask.Type.PRODUCT_IMAGE, status=AITask.Status.SUCCEEDED, model_config=mc, idempotency_key="conv-test-1", request_payload={"prompt": "猫", "batch_id": "bx", "ratio": "1:1", "product_id": str(product.id)}, ) r = self.client.get(f"/api/ai/image-conversations/{conv.id}/tasks/") self.assertEqual(r.status_code, 200, r.content) data = r.json()["tasks"] self.assertEqual(len(data), 1) self.assertEqual(data[0]["prompt"], "猫") self.assertEqual(data[0]["batch_id"], "bx") self.assertEqual(data[0]["product_id"], str(product.id)) def test_rerun_uses_original_batch_product_instead_of_request_product(self): CreditAccount.objects.create(team=self.team, balance="100.0000") original_product = Product.objects.create(team=self.team, created_by=self.user, title="原批次商品") other_product = Product.objects.create(team=self.team, created_by=self.user, title="当前路由商品") with patch("apps.ai.tasks.generate_standalone_image_task.delay"): first = self.client.post( "/api/ai/generate-image/", {"prompt": "原商品海报", "mode": "image", "count": 1, "product_id": str(original_product.id)}, format="json", ) self.assertEqual(first.status_code, 202, first.content) original_task = AITask.objects.get(conversation_id=first.json()["conversation_id"]) original_task.status = AITask.Status.FAILED original_task.save(update_fields=["status"]) rerun = self.client.post( "/api/ai/generate-image/", { "prompt": "原商品海报", "mode": "image", "count": 1, "product_id": str(other_product.id), "conversation_id": first.json()["conversation_id"], "batch_id": first.json()["batch_id"], "retry_of_task_id": str(original_task.id), }, format="json", ) self.assertEqual(rerun.status_code, 202, rerun.content) self.assertEqual(rerun.json()["conversation_id"], first.json()["conversation_id"]) newest = AITask.objects.filter(conversation_id=first.json()["conversation_id"]).order_by("created_at").last() self.assertEqual(newest.request_payload.get("product_id"), str(original_product.id)) def test_generate_with_refs_persists_and_tasks_endpoint_returns_them(self): """HTTP 全链路:带 reference_image_ids 提交 → 任务 payload 落 ids → tasks 接口把参考图解析回 {name,url}。 守护「上传的参考图被带进生成 + 切换/刷新后批次头仍能回显参考图」。""" CreditAccount.objects.create(team=self.team, balance="100.0000") patch("apps.ai.tasks.generate_standalone_image_task.delay").start() self.addCleanup(patch.stopall) ref = Asset.objects.create( team=self.team, created_by=self.user, name="背心参考", asset_type=Asset.Type.IMAGE, source=Asset.Source.UPLOAD, category=Asset.Category.UPLOAD, ) AssetFile.objects.create(asset=ref, object_key="r.png", bucket="b", content_type="image/png", preview_url="http://x/ref.png", is_primary=True) r = self.client.post( "/api/ai/generate-image/", {"prompt": "按这件背心生成不同场景", "mode": "image", "count": 1, "reference_image_ids": [str(ref.id)]}, format="json", ) self.assertEqual(r.status_code, 202, r.content) conv_id = r.json()["conversation_id"] # 任务 payload 落了参考图 id task = AITask.objects.filter(conversation_id=conv_id).first() self.assertEqual(task.request_payload.get("reference_image_ids"), [str(ref.id)]) # tasks 接口把参考图解析成 {id,name,url} 回显 —— id 供前端重跑时原样复用(否则重跑丢参考图) data = self.client.get(f"/api/ai/image-conversations/{conv_id}/tasks/").json()["tasks"] self.assertEqual(data[0]["reference_images"], [{"id": str(ref.id), "name": "背心参考", "url": "http://x/ref.png"}]) def test_rerun_with_batch_id_reuses_batch_and_marks_append(self): """重跑/补图带原批次 batch_id → 新任务沿用同一 batch_id 并打 batch_append 标记(tasks 接口回 rerun=True)—— 守护「重跑不裂新批次卡 + 失败格计数不膨胀」;非法 batch_id 则铸新批次兜底。""" CreditAccount.objects.create(team=self.team, balance="100.0000") patch("apps.ai.tasks.generate_standalone_image_task.delay").start() self.addCleanup(patch.stopall) first = self.client.post( "/api/ai/generate-image/", {"prompt": "一只猫", "mode": "image", "count": 2}, format="json" ) self.assertEqual(first.status_code, 202, first.content) conv_id = first.json()["conversation_id"] batch_id = first.json()["batch_id"] self.assertTrue(batch_id) failed_task = AITask.objects.filter(conversation_id=conv_id).order_by("created_at").first() failed_task.status = AITask.Status.FAILED failed_task.save(update_fields=["status"]) # 重跑一张:带原对话 + 原 batch_id → 归回原批次 second = self.client.post( "/api/ai/generate-image/", {"prompt": "一只猫", "mode": "image", "count": 1, "conversation_id": conv_id, "batch_id": batch_id, "retry_of_task_id": str(failed_task.id)}, format="json", ) self.assertEqual(second.status_code, 202, second.content) self.assertEqual(second.json()["batch_id"], batch_id) tasks = self.client.get(f"/api/ai/image-conversations/{conv_id}/tasks/").json()["tasks"] self.assertEqual(len(tasks), 3) self.assertTrue(all(t["batch_id"] == batch_id for t in tasks)) self.assertEqual([t["rerun"] for t in tasks], [False, False, True]) self.assertEqual(tasks[-1]["retry_of_task_id"], str(failed_task.id)) # 非法 batch_id(不是 UUID):不沿用,铸新批次,不打 append 标记 third = self.client.post( "/api/ai/generate-image/", {"prompt": "一只猫", "mode": "image", "count": 1, "conversation_id": conv_id, "batch_id": "not-a-uuid"}, format="json", ) self.assertEqual(third.status_code, 202, third.content) self.assertNotEqual(third.json()["batch_id"], batch_id) newest = self.client.get(f"/api/ai/image-conversations/{conv_id}/tasks/").json()["tasks"][-1] self.assertFalse(newest["rerun"]) class StandaloneCategoryTests(TestCase): """图片趴三类归类(期2):模特上身图→model_tryon / 平台套图→platform_kit / 自由创作→free_create; 生成演员(model 无 product)仍→person(视频角色)。直接跑 worker 函数避开异步 .delay。""" def setUp(self): self.user = User.objects.create_user(username="catowner", password="pass") self.team = Team.objects.create(name="CT", owner=self.user) CreditAccount.objects.create(team=self.team, balance="100.0000") self.product = Product.objects.create(team=self.team, created_by=self.user, title="测试商品") cover = Asset.objects.create( team=self.team, created_by=self.user, name="主图", asset_type=Asset.Type.IMAGE, source=Asset.Source.UPLOAD, category=Asset.Category.PRODUCT_IMAGE, ) AssetFile.objects.create(asset=cover, object_key="c.png", bucket="b", content_type="image/png", preview_url="http://x/cover.png", is_primary=True) self.product.cover_asset = cover self.product.save(update_fields=["cover_asset"]) def _patch_provider(self): provider = patch("apps.ai.services.get_image_provider").start() prov = provider.return_value prov.image_edit.return_value = {"data": [{"url": "http://x/out.png"}]} prov.image_generation.return_value = {"data": [{"url": "http://x/out.png"}]} prov.extract_first_media_url.return_value = "http://x/out.png" media = patch("apps.ai.services.VolcanoArkProvider.media_to_bytes").start() media.return_value = (BytesIO(b"img"), "image/png") store = patch("apps.ai.services.TosStorage").start() stored = store.return_value.upload_fileobj.return_value stored.object_key, stored.bucket, stored.content_type, stored.size_bytes = "o.png", "b", "image/png", 3 self.addCleanup(patch.stopall) return prov def _run(self, mode, *, product_id=None, model_id=None): from apps.ai.services import run_standalone_image_task self._patch_provider() tasks = enqueue_standalone_images( team=self.team, user=self.user, prompt="x", mode=mode, count=1, product_id=product_id, model_id=model_id, model_entity_id="ent-1", ratio="4:5", ) run_standalone_image_task(task_id=str(tasks[0].id)) return Asset.objects.filter(origin_task=tasks[0]).first() def test_model_tryon_category(self): a = self._run("model", product_id=str(self.product.id)) self.assertEqual(a.category, Asset.Category.MODEL_TRYON) self.assertEqual(a.metadata.get("mode"), "model") self.assertTrue(a.metadata.get("batch_id")) # 成组 self.assertEqual(a.metadata.get("model_entity_id"), "ent-1") # 溯源模特库 self.assertTrue(a.in_library) # R109:图片生成成图自动入资产库 def test_platform_kit_category(self): a = self._run("cover", product_id=str(self.product.id)) self.assertEqual(a.category, Asset.Category.PLATFORM_KIT) self.assertTrue(a.in_library) # R109 def test_free_create_category(self): a = self._run("image") self.assertEqual(a.category, Asset.Category.FREE_CREATE) self.assertTrue(a.in_library) # R109 def test_generate_model_candidate_is_model_portrait_and_not_in_asset_library(self): # 图片生成 mode=model 且无商品 = 新建模特候选，不再生成“演员”person 资产。 a = self._run("model") self.assertEqual(a.category, Asset.Category.MODEL_PORTRAIT) self.assertFalse(a.in_library) @override_settings(MODEL_TRYON_PROMPT_V2_ENABLED=True) def test_model_tryon_v2_provider_failure_releases_reserved_credit(self): from apps.ai.services import run_standalone_image_task self.product.title = "女装裤子" self.product.save(update_fields=["title"]) prov = self._patch_provider() prov.image_edit.side_effect = ValueError("生成失败") with patch("apps.ai.tasks.generate_standalone_image_task.delay"): task = enqueue_standalone_images( team=self.team, user=self.user, prompt="全身站姿，双腿完整露出", mode="model", count=1, product_id=str(self.product.id), ratio="3:4", )[0] run_standalone_image_task(task_id=str(task.id)) task.refresh_from_db() account = CreditAccount.objects.get(team=self.team) self.assertEqual(task.status, AITask.Status.FAILED) self.assertEqual(float(task.actual_cost), 0.0) self.assertEqual(float(account.balance), 100.0) self.assertEqual(float(account.reserved_balance), 0.0) self.assertTrue(task.request_payload["tryon_prompt"]["applied"]) self.assertTrue( CreditLedger.objects.filter(task=task, ledger_type=CreditLedger.Type.RELEASE).exists() ) @override_settings(CACHES={"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}}) class TriviewModelDecouplingTests(TestCase): """项目角色三视图只归项目，不自动创建或更新团队模特。""" def setUp(self): from apps.projects.models import Project self.user = User.objects.create_user(username="trio", password="p") self.team = Team.objects.create(name="TR", owner=self.user) CreditAccount.objects.create(team=self.team, balance="100.0000") self.product = Product.objects.create(team=self.team, created_by=self.user, title="P") self.project = Project.objects.create(team=self.team, name="项目甲", product=self.product, created_by=self.user) self.portrait = Asset.objects.create( team=self.team, created_by=self.user, name="项目甲-person", asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PERSON, ) AssetFile.objects.create(asset=self.portrait, object_key="p.png", bucket="b", content_type="image/png", preview_url="http://x/p.png", is_primary=True) def _patch_provider(self): provider = patch("apps.ai.services.get_image_provider").start() prov = provider.return_value prov.image_generation.return_value = {"data": [{"url": "http://x/person.png"}]} prov.image_edit.return_value = {"data": [{"url": "http://x/tri.png"}]} prov.extract_first_media_url.return_value = "http://x/tri.png" media = patch("apps.ai.services.VolcanoArkProvider.media_to_bytes").start() media.return_value = (BytesIO(b"img"), "image/png") store = patch("apps.ai.services.TosStorage").start() stored = store.return_value.upload_fileobj.return_value stored.object_key, stored.bucket, stored.content_type, stored.size_bytes = "o.png", "b", "image/png", 3 self.addCleanup(patch.stopall) def _gen_triview(self): from apps.ai.services import generate_person_triview, run_triview_task task = generate_person_triview(project=self.project, user=self.user, portrait_asset=self.portrait) run_triview_task(task_id=str(task.id)) def test_triview_stays_on_project_and_does_not_enroll_model(self): from apps.assets.models import Model self._patch_provider() self._gen_triview() self.assertFalse(Model.objects.filter(portrait_asset=self.portrait).exists()) group = next(g for g in self.project.base_asset_groups.all() if (g.metadata or {}).get("triview_of") == str(self.portrait.id)) self.assertIsNotNone(group.adopted_asset) self.assertEqual(group.adopted_asset.category, Asset.Category.TRI_VIEW) self.assertIn(group.adopted_asset.category, Asset.REVIEW_CATEGORIES) def test_triview_does_not_overwrite_existing_model_triview(self): from apps.assets.models import Model existing_tri = Asset.objects.create( team=self.team, created_by=self.user, name="模特原三视图", asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.TRI_VIEW, ) model = Model.objects.create( team=self.team, created_by=self.user, name="已有模特", portrait_asset=self.portrait, triview_asset=existing_tri, ) self._patch_provider() self._gen_triview() model.refresh_from_db() self.assertEqual(model.triview_asset_id, existing_tri.id) @patch("apps.ai.tasks.generate_base_asset_task.delay") def test_person_base_asset_marks_auto_triview_for_worker(self, _enqueue_base_asset): """角色立绘任务需标记自动接力三视图，实际接力只在立绘成功落库后发生。""" from apps.ai.services import generate_base_asset from apps.projects.models import BaseAssetGroup ModelConfig.objects.create( provider=ModelProvider.objects.create(name="auto-tri-provider", display_name="Auto Tri Provider"), name="auto-tri-img", display_name="Auto Tri Img", capability=ModelConfig.Capability.IMAGE, endpoint="images/generations", unit_price="1.0000", is_default=True, ) self._patch_provider() task = generate_base_asset( project=self.project, user=self.user, kind=BaseAssetGroup.Kind.PERSON, prompt="项目角色立绘", label="女主", auto_triview=True, ) task.refresh_from_db() self.assertTrue(task.request_payload.get("auto_triview")) @patch("apps.assets.review.submit_asset_for_review") @patch("apps.ai.services.generate_person_triview") @patch("apps.ai.services._store_generated_media") @patch("apps.ai.services.execute_routed_image_request") @patch("apps.ai.tasks.generate_base_asset_task.delay") def test_completed_auto_person_generation_queues_triview( self, _enqueue_base_asset, routed, store_media, kickoff_triview, _review ): """角色立绘成功落库后，worker 必须自动创建绑定该立绘的三视图任务。""" from apps.ai.services import generate_base_asset, run_base_asset_task from apps.projects.models import BaseAssetGroup ModelConfig.objects.create( provider=ModelProvider.objects.create(name="auto-tri-success-provider", display_name="Auto Tri Success Provider"), name="auto-tri-success-img", display_name="Auto Tri Success Img", capability=ModelConfig.Capability.IMAGE, endpoint="images/generations", unit_price="1.0000", is_default=True, ) routed.return_value = SimpleNamespace(value=({"data": [{"url": "http://x/person.png"}]}, "http://x/person.png")) store_media.return_value = self.portrait task = generate_base_asset( project=self.project, user=self.user, kind=BaseAssetGroup.Kind.PERSON, prompt="项目角色立绘", label="女主", # 专业模式显式要求时才接力三视图；极速成片传 False 会跳过。 auto_triview=True, ) with self.captureOnCommitCallbacks(execute=True): run_base_asset_task(task_id=str(task.id)) kickoff_triview.assert_called_once_with( project=self.project, user=self.user, portrait_asset=self.portrait ) class ModelLibraryTriviewTaskTests(TestCase): """团队模特三视图：无项目归属、20 积分闭环、幂等防重复、只更新当前 Model。""" def setUp(self): from apps.assets.models import Model self.user = User.objects.create_user(username="model-tri", password="p") self.team = Team.objects.create(name="Model Tri", owner=self.user) TeamMember.objects.create(team=self.team, user=self.user, role="owner", status="active") self.client = APIClient() self.client.force_authenticate(self.user) self.account = CreditAccount.objects.create(team=self.team, balance="100.0000") provider = ModelProvider.objects.create( name="model-tri-provider", display_name="Model Tri Provider", base_url="http://x.test/v1", api_key="test-key", ) self.config = ModelConfig.objects.create( provider=provider, name="model-tri-image", display_name="Model Tri Image", capability=ModelConfig.Capability.IMAGE, endpoint="images/edits", unit_price="20.0000", is_default=True, ) self.portrait = Asset.objects.create( team=self.team, created_by=self.user, name="模特形象", asset_type=Asset.Type.IMAGE, source=Asset.Source.UPLOAD, category=Asset.Category.MODEL_PORTRAIT, in_library=False, ) AssetFile.objects.create( asset=self.portrait, object_key="portrait.png", bucket="b", content_type="image/png", preview_url="http://x/portrait.png", is_primary=True, ) self.model = Model.objects.create( team=self.team, created_by=self.user, name="团队模特", portrait_asset=self.portrait, ) def _submit(self): from apps.ai.services import generate_model_triview with patch("apps.ai.tasks.generate_model_triview_task.delay") as delay: result = generate_model_triview(model=self.model, user=self.user) return result, delay def _provider_success(self): provider = patch("apps.ai.services.get_image_provider").start().return_value provider.image_edit.return_value = {"data": [{"url": "http://x/tri.png"}]} provider.extract_first_media_url.return_value = "http://x/tri.png" patch("apps.ai.services.VolcanoArkProvider.media_to_bytes", return_value=(BytesIO(b"img"), "image/png")).start() storage = patch("apps.ai.services.TosStorage").start() stored = storage.return_value.upload_fileobj.return_value stored.object_key, stored.bucket, stored.content_type, stored.size_bytes = "tri.png", "b", "image/png", 3 self.addCleanup(patch.stopall) def test_submit_is_projectless_reserves_20_and_reuses_inflight(self): (task, created), delay = self._submit() self.assertTrue(created) self.assertIsNone(task.project_id) self.assertEqual(task.task_type, AITask.Type.MODEL_TRIVIEW) self.assertEqual(float(task.estimated_cost), 20.0) self.assertEqual(task.request_payload["model_id"], str(self.model.id)) delay.assert_called_once_with(str(task.id)) self.account.refresh_from_db() self.assertEqual(float(self.account.reserved_balance), 20.0) reserve = CreditLedger.objects.get(task=task, ledger_type=CreditLedger.Type.RESERVE) self.assertEqual(reserve.reason, "AI 任务预扣额度") (same_task, created_again), second_delay = self._submit() self.assertFalse(created_again) self.assertEqual(same_task.id, task.id) second_delay.assert_not_called() self.assertEqual(AITask.objects.filter(task_type=AITask.Type.MODEL_TRIVIEW).count(), 1) def test_success_charges_once_and_updates_only_model_triview(self): from apps.ai.services import run_model_triview_task (task, _created), _delay = self._submit() self._provider_success() run_model_triview_task(task_id=str(task.id)) task.refresh_from_db() self.model.refresh_from_db() self.account.refresh_from_db() self.assertEqual(task.status, AITask.Status.SUCCEEDED) self.assertEqual(float(task.actual_cost), 20.0) self.assertEqual(float(self.account.balance), 80.0) self.assertEqual(float(self.account.reserved_balance), 0.0) self.assertIsNotNone(self.model.triview_asset_id) self.assertEqual(self.model.triview_asset.category, Asset.Category.TRI_VIEW) self.assertFalse(self.model.triview_asset.in_library) self.assertEqual(self.model.triview_asset.metadata["model_id"], str(self.model.id)) self.assertEqual(self.model.metadata["triview_versions"], [str(self.model.triview_asset_id)]) charge = CreditLedger.objects.get(task=task, ledger_type=CreditLedger.Type.CHARGE) self.assertEqual(charge.reason, "AI 任务扣费") run_model_triview_task(task_id=str(task.id)) self.assertEqual(CreditLedger.objects.filter(task=task, ledger_type=CreditLedger.Type.CHARGE).count(), 1) def test_failure_releases_credit_and_keeps_old_triview(self): from apps.ai.services import run_model_triview_task old_tri = Asset.objects.create( team=self.team, created_by=self.user, name="旧三视图", asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.TRI_VIEW, in_library=False, ) self.model.triview_asset = old_tri self.model.save(update_fields=["triview_asset"]) (task, _created), _delay = self._submit() provider = patch("apps.ai.services.get_image_provider").start().return_value provider.image_edit.side_effect = ValueError("生成失败") self.addCleanup(patch.stopall) run_model_triview_task(task_id=str(task.id)) task.refresh_from_db() self.model.refresh_from_db() self.account.refresh_from_db() self.assertEqual(task.status, AITask.Status.FAILED) self.assertEqual(self.model.triview_asset_id, old_tri.id) self.assertEqual(float(self.account.balance), 100.0) self.assertEqual(float(self.account.reserved_balance), 0.0) self.assertFalse(CreditLedger.objects.filter(task=task, ledger_type=CreditLedger.Type.CHARGE).exists()) self.assertTrue(CreditLedger.objects.filter(task=task, ledger_type=CreditLedger.Type.RELEASE).exists()) def test_api_quote_submit_and_status_contract(self): quote = self.client.get(f"/api/models/{self.model.id}/triview-quote/") self.assertEqual(quote.status_code, 200) self.assertEqual(float(quote.json()["points"]), 20.0) # 测试环境可能从本机配置继承开关；此处显式锁定关闭分支，避免环境依赖。 with self.settings(MODEL_TRIVIEW_GENERATION_ENABLED=False): disabled = self.client.post(f"/api/models/{self.model.id}/generate-triview/") self.assertEqual(disabled.status_code, 409) self.assertFalse(AITask.objects.filter(task_type=AITask.Type.MODEL_TRIVIEW).exists()) with self.settings(MODEL_TRIVIEW_GENERATION_ENABLED=True): with patch("apps.ai.tasks.generate_model_triview_task.delay"): submit = self.client.post(f"/api/models/{self.model.id}/generate-triview/") self.assertEqual(submit.status_code, 202) task_id = submit.json()["task_id"] self.assertEqual(float(submit.json()["estimated_cost"]), 20.0) status_res = self.client.get(f"/api/models/{self.model.id}/triview-status/?task_id={task_id}") self.assertEqual(status_res.status_code, 200) self.assertEqual(status_res.json()["status"], AITask.Status.RESERVED) self.assertIsNone(status_res.json()["model"]) task = AITask.objects.get(id=task_id) raw_error = ( "400 moderation_blocked Your request was rejected by the safety system. " "request ID 47ffb351-2307-4c91-bd7e-76749729eef7" ) task.status = AITask.Status.FAILED task.error_message = raw_error task.save(update_fields=["status", "error_message"]) failed_status = self.client.get(f"/api/models/{self.model.id}/triview-status/?task_id={task_id}") self.assertEqual(failed_status.status_code, 200) payload = failed_status.json() self.assertEqual(payload["error"]["code"], "content_rejected") self.assertEqual(payload["error"]["operation"], "triview_generate") self.assertEqual( payload["error_message"], "三视图内容需要调整：内容未通过生成审核，请调整描述或素材后重试。", ) self.assertNotIn("moderation_blocked", payload["error_message"]) self.assertNotIn("Azure", str(payload)) self.assertNotIn("47ffb351-2307-4c91-bd7e-76749729eef7", str(payload)) class _FakeStreamResp: """模拟 requests 流式响应:支持 with、raise_for_status、可写 encoding、iter_lines。""" status_code = 200 encoding = None def __init__(self, lines): self._lines = lines def __enter__(self): return self def __exit__(self, *exc): return False def raise_for_status(self): return None def iter_lines(self, decode_unicode=True): # noqa: ARG002 yield from self._lines class ChatStreamReasoningTests(SimpleTestCase): """推理模型(豆包 seed-pro 等)思考期只发 reasoning_content、不发 content。 provider 必须把它作为独立 `reasoning` 事件转发——否则脚本 agent 思考期零输出 = 前端「卡在生成分镜」假死。本测试锁住该转发,防回归。""" def test_reasoning_content_forwarded_as_reasoning_event(self): def _chunk(delta): return "data: " + json.dumps({"choices": [{"delta": delta}]}, ensure_ascii=False) lines = [ _chunk({"reasoning_content": "先想想"}), _chunk({"reasoning_content": "用户要4镜"}), _chunk({"content": "正在生成"}), _chunk({"content": "脚本…"}), "data: [DONE]", ] prov = VolcanoArkProvider(api_key="k", base_url="http://x") with patch("apps.ai.providers.volcano.requests.post", return_value=_FakeStreamResp(lines)): events = list(prov.chat_completion_stream(model="m", messages=[{"role": "user", "content": "hi"}])) self.assertEqual([e["type"] for e in events], ["reasoning", "reasoning", "delta", "delta", "done"]) self.assertEqual([e["text"] for e in events if e["type"] == "reasoning"], ["先想想", "用户要4镜"]) self.assertEqual("".join(e["text"] for e in events if e["type"] == "delta"), "正在生成脚本…") def test_length_finish_reason_is_forwarded(self): """撞 max_tokens 的那一刀必须能被上层看见,否则截断的 tool 参数会被当成模型偷懒反复重试。""" lines = [ "data: " + json.dumps({"choices": [{"delta": {"content": "前半段"}}]}, ensure_ascii=False), "data: " + json.dumps({"choices": [{"delta": {}, "finish_reason": "length"}]}, ensure_ascii=False), "data: [DONE]", ] prov = VolcanoArkProvider(api_key="k", base_url="http://x") with patch("apps.ai.providers.volcano.requests.post", return_value=_FakeStreamResp(lines)): events = list(prov.chat_completion_stream(model="m", messages=[{"role": "user", "content": "hi"}])) self.assertEqual([e["type"] for e in events], ["delta", "finish", "done"]) self.assertEqual(events[1]["reason"], "length") class VolcanoVideoSeedTests(SimpleTestCase): """Seedance 2.5 r2v 的 seed 上限是 int32，超了会 InvalidParameter。""" def test_create_video_task_clamps_seed_to_int32(self): captured = {} class _Resp: ok = True status_code = 200 text = "{}" def json(self): return {"id": "ark-1"} def _post(*_args, **kwargs): captured["body"] = kwargs["json"] return _Resp() prov = VolcanoArkProvider(api_key="k", base_url="http://x") with patch("apps.ai.providers.volcano.requests.post", side_effect=_post): prov.create_video_task( model="doubao-seedance-2-5-260628", endpoint="contents/generations/tasks", prompt="test", seed=3857139456, ) self.assertEqual(captured["body"]["seed"], 3857139456 & 2147483647) self.assertLessEqual(captured["body"]["seed"], 2147483647) class WorkbenchAndUnreadTests(TestCase): """R100(工作台记录后端持久化)+ R96(未读生成任务角标)+ R109(删除联动)端到端: · /api/ai/tasks/workbench/ 按 mode+product 回放批次流(与任务中心同源),软删的图不回显; · /api/ai/tasks/unread/ 只统计 mode∈model/cover/image 的未读任务,按团队隔离; · /api/ai/tasks/mark-read/ 按商品清零。""" def setUp(self): self.user = User.objects.create_user(username="wbowner", password="pass") self.team = Team.objects.create(name="WB", owner=self.user) TeamMember.objects.create(team=self.team, user=self.user, role="owner", status="active") CreditAccount.objects.create(team=self.team, balance="100.0000") self.product = Product.objects.create(team=self.team, created_by=self.user, title="工作台商品") cover = Asset.objects.create( team=self.team, created_by=self.user, name="主图", asset_type=Asset.Type.IMAGE, source=Asset.Source.UPLOAD, category=Asset.Category.PRODUCT_IMAGE, ) AssetFile.objects.create(asset=cover, object_key="c.png", bucket="b", content_type="image/png", preview_url="http://x/cover.png", is_primary=True) self.product.cover_asset = cover self.product.save(update_fields=["cover_asset"]) self.client = APIClient() self.client.force_authenticate(self.user) def _patch_provider(self): provider = patch("apps.ai.services.get_image_provider").start() prov = provider.return_value prov.image_edit.return_value = {"data": [{"url": "http://x/out.png"}]} prov.image_generation.return_value = {"data": [{"url": "http://x/out.png"}]} prov.extract_first_media_url.return_value = "http://x/out.png" media = patch("apps.ai.services.VolcanoArkProvider.media_to_bytes").start() media.return_value = (BytesIO(b"img"), "image/png") store = patch("apps.ai.services.TosStorage").start() stored = store.return_value.upload_fileobj.return_value stored.object_key, stored.bucket, stored.content_type, stored.size_bytes = "o.png", "b", "image/png", 3 self.addCleanup(patch.stopall) return prov def _generate(self, mode="cover", count=2, platform_id=None): from apps.ai.services import run_standalone_image_task self._patch_provider() tasks = enqueue_standalone_images( team=self.team, user=self.user, prompt="平台套图测试", mode=mode, count=count, product_id=str(self.product.id), ratio="1:1", platform_id=platform_id, ) for t in tasks: run_standalone_image_task(task_id=str(t.id)) return tasks def test_workbench_returns_batches_and_hides_deleted_assets(self): tasks = self._generate(mode="cover", count=2, platform_id="douyin") r = self.client.get(f"/api/ai/tasks/workbench/?mode=cover&product={self.product.id}") self.assertEqual(r.status_code, 200, r.content) rows = r.json()["tasks"] self.assertEqual(len(rows), 2) # 同一批共享 batch_id;prompt/ratio/platform/product 从 payload 抽出;每任务带成图 self.assertEqual(len({row["batch_id"] for row in rows}), 1) self.assertEqual(rows[0]["prompt"], "平台套图测试") self.assertEqual(rows[0]["ratio"], "1:1") self.assertEqual(rows[0]["platform_id"], "douyin") self.assertEqual(rows[0]["product_id"], str(self.product.id)) self.assertEqual(len(rows[0]["assets"]), 1) # 其他 mode / 其他商品查不到这批 self.assertEqual(self.client.get("/api/ai/tasks/workbench/?mode=model").json()["tasks"], []) # R108/R109:软删其中一张 → 该任务的 assets 随之为空(删除联动),资产不再回显 asset = Asset.objects.filter(origin_task=tasks[0]).first() d = self.client.delete(f"/api/assets/{asset.id}/") self.assertEqual(d.status_code, 204, d.content) asset.refresh_from_db() self.assertTrue(asset.is_deleted) # 回收站软删,不是物理删除 rows = self.client.get(f"/api/ai/tasks/workbench/?mode=cover&product={self.product.id}").json()["tasks"] by_id = {row["id"]: row for row in rows} self.assertEqual(by_id[str(tasks[0].id)]["assets"], []) self.assertEqual(len(by_id[str(tasks[1].id)]["assets"]), 1) def test_workbench_rejects_unknown_mode(self): # mode 白名单:脚本 agent 复用的 auto/theme/revise 不能混进来 self.assertEqual(self.client.get("/api/ai/tasks/workbench/?mode=auto").status_code, 400) self.assertEqual(self.client.get("/api/ai/tasks/workbench/").status_code, 400) def test_unread_counts_and_mark_read(self): self._generate(mode="cover", count=2, platform_id="douyin") r = self.client.get("/api/ai/tasks/unread/") self.assertEqual(r.status_code, 200, r.content) body = r.json() self.assertEqual(body["total"], 2) self.assertEqual(body["by_product"], {str(self.product.id): 2}) # 按商品标记已读 → 清零 m = self.client.post("/api/ai/tasks/mark-read/", {"product_id": str(self.product.id)}, format="json") self.assertEqual(m.status_code, 200, m.content) self.assertEqual(m.json()["updated"], 2) body = self.client.get("/api/ai/tasks/unread/").json() self.assertEqual(body["total"], 0) self.assertEqual(body["by_product"], {}) def test_exception_batch_delete_restore_and_purge_are_atomic(self): """部分失败批次进入“图片异常批次”，而非与资产垃圾桶重复；恢复后整批回放。""" tasks = self._generate(mode="cover", count=2, platform_id="douyin") failed, succeeded = tasks # 模拟其中一张供应商失败：失败任务无成图，另一张保持成功，构成真实的混合批次。 Asset.objects.filter(origin_task=failed).delete() failed.status = AITask.Status.FAILED failed.save(update_fields=["status"]) succeeded_asset = Asset.objects.get(origin_task=succeeded) succeeded_file = succeeded_asset.files.get(is_primary=True) succeeded_file.preview_url = "http://x/generated-success.png" succeeded_file.save(update_fields=["preview_url"]) deleted = self.client.delete(f"/api/ai/tasks/{failed.id}/workbench-batch/") self.assertEqual(deleted.status_code, 200, deleted.content) self.assertEqual(deleted.json()["storage"], "exception_batch") failed.refresh_from_db() succeeded.refresh_from_db() succeeded_asset.refresh_from_db() self.assertTrue(failed.is_deleted) self.assertTrue(succeeded.is_deleted) self.assertTrue(succeeded_asset.is_deleted) self.assertEqual( self.client.get(f"/api/ai/tasks/workbench/?mode=cover&product={self.product.id}").json()["tasks"], [] ) trash = self.client.get("/api/ai/tasks/workbench-exception-batches-trash/") self.assertEqual(trash.status_code, 200, trash.content) rows = trash.json()["results"] self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["id"], str(failed.id)) self.assertEqual(rows[0]["cover_preview_url"], "http://x/generated-success.png") self.assertEqual(rows[0]["product_title"], self.product.title) asset_trash_ids = {row["id"] for row in self.client.get("/api/assets/trash/").json()["results"]} self.assertNotIn(str(succeeded_asset.id), asset_trash_ids) restored = self.client.post(f"/api/ai/tasks/{failed.id}/restore-workbench-batch/") self.assertEqual(restored.status_code, 200, restored.content) failed.refresh_from_db() succeeded.refresh_from_db() succeeded_asset.refresh_from_db() self.assertFalse(failed.is_deleted) self.assertFalse(succeeded.is_deleted) self.assertFalse(succeeded_asset.is_deleted) rows = self.client.get(f"/api/ai/tasks/workbench/?mode=cover&product={self.product.id}").json()["tasks"] self.assertEqual({row["id"] for row in rows}, {str(failed.id), str(succeeded.id)}) self.client.delete(f"/api/ai/tasks/{failed.id}/workbench-batch/") purged = self.client.delete(f"/api/ai/tasks/{failed.id}/purge-workbench-batch/") self.assertEqual(purged.status_code, 204, purged.content) failed.refresh_from_db() succeeded.refresh_from_db() succeeded_asset.refresh_from_db() self.assertIsNotNone(failed.purged_at) self.assertIsNotNone(succeeded.purged_at) self.assertIsNotNone(succeeded_asset.purged_at) self.assertEqual(self.client.get("/api/ai/tasks/workbench-exception-batches-trash/").json()["results"], []) def test_exception_batch_without_successful_image_uses_no_image_even_with_product_cover(self): """纯失败批次即使商品有封面也必须返回空缩略图，由前端显示“无图”。""" task = self._generate(mode="cover", count=1, platform_id="douyin")[0] Asset.objects.filter(origin_task=task).delete() task.status = AITask.Status.FAILED task.save(update_fields=["status"]) deleted = self.client.delete(f"/api/ai/tasks/{task.id}/workbench-batch/") self.assertEqual(deleted.status_code, 200, deleted.content) rows = self.client.get("/api/ai/tasks/workbench-exception-batches-trash/").json()["results"] self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["cover_preview_url"], "") self.assertEqual(rows[0]["product_title"], self.product.title) def test_exception_batch_uses_first_successful_image_in_original_task_order(self): """多张成功图时跳过前序失败任务，稳定选择原始批次中第一张成功图。""" tasks = self._generate(mode="cover", count=3, platform_id="douyin") failed, first_succeeded, later_succeeded = tasks Asset.objects.filter(origin_task=failed).delete() failed.status = AITask.Status.FAILED failed.save(update_fields=["status"]) AssetFile.objects.filter(asset__origin_task=first_succeeded).update(preview_url="http://x/first-success.png") AssetFile.objects.filter(asset__origin_task=later_succeeded).update(preview_url="http://x/later-success.png") deleted = self.client.delete(f"/api/ai/tasks/{failed.id}/workbench-batch/") self.assertEqual(deleted.status_code, 200, deleted.content) rows = self.client.get("/api/ai/tasks/workbench-exception-batches-trash/").json()["results"] self.assertEqual(rows[0]["cover_preview_url"], "http://x/first-success.png") def test_exception_batch_uses_successful_append_when_original_tasks_all_failed(self): """原始任务均失败但同批补图成功时，使用最先成功的补图作为缩略图。""" original_tasks = self._generate(mode="cover", count=2, platform_id="douyin") for task in original_tasks: Asset.objects.filter(origin_task=task).delete() task.status = AITask.Status.FAILED task.save(update_fields=["status"]) payload = dict(original_tasks[0].request_payload) payload.update({ "index": 0, "batch_append": True, "retry_of_task_id": str(original_tasks[0].id), }) appended = AITask.objects.create( team=self.team, created_by=self.user, task_type=original_tasks[0].task_type, status=AITask.Status.SUCCEEDED, model_config=original_tasks[0].model_config, idempotency_key="exception-batch-successful-append", request_payload=payload, ) appended_asset = Asset.objects.create( team=self.team, created_by=self.user, name="成功补图", asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=Asset.Category.PLATFORM_KIT, origin_task=appended, ) AssetFile.objects.create( asset=appended_asset, object_key="append.png", bucket="b", content_type="image/png", preview_url="http://x/successful-append.png", is_primary=True, ) deleted = self.client.delete(f"/api/ai/tasks/{original_tasks[0].id}/workbench-batch/") self.assertEqual(deleted.status_code, 200, deleted.content) rows = self.client.get("/api/ai/tasks/workbench-exception-batches-trash/").json()["results"] self.assertEqual(rows[0]["cover_preview_url"], "http://x/successful-append.png") def test_successful_batch_keeps_existing_asset_trash_behavior(self): tasks = self._generate(mode="cover", count=2, platform_id="douyin") assets = list(Asset.objects.filter(origin_task__in=tasks)) deleted = self.client.delete(f"/api/ai/tasks/{tasks[0].id}/workbench-batch/") self.assertEqual(deleted.status_code, 200, deleted.content) self.assertEqual(deleted.json()["storage"], "asset") self.assertFalse(AITask.objects.get(id=tasks[0].id).is_deleted) self.assertTrue(all(Asset.objects.get(id=asset.id).is_deleted for asset in assets)) asset_trash_ids = {row["id"] for row in self.client.get("/api/assets/trash/").json()["results"]} self.assertTrue({str(asset.id) for asset in assets}.issubset(asset_trash_ids)) self.assertEqual(self.client.get("/api/ai/tasks/workbench-exception-batches-trash/").json()["results"], []) def test_image_conversation_exception_batch_hides_and_restores_only_its_tasks(self): """图片创作走对话任务读取，也必须在异常批次删除/恢复后保持一致。""" conversation = ImageConversation.objects.create( team=self.team, created_by=self.user, mode=ImageConversation.Mode.IMAGE, title="失败图片创作", product=self.product ) task = AITask.objects.create( team=self.team, created_by=self.user, conversation=conversation, task_type=AITask.Type.PRODUCT_IMAGE, status=AITask.Status.FAILED, model_config=ModelConfig.objects.filter(capability=ModelConfig.Capability.IMAGE).first(), idempotency_key="conversation-exception-batch", request_payload={"mode": "image", "batch_id": "conversation-exception-batch", "prompt": "失败的图片创作", "product_id": str(self.product.id)}, ) task_rows = self.client.get(f"/api/ai/image-conversations/{conversation.id}/tasks/").json()["tasks"] self.assertEqual([row["id"] for row in task_rows], [str(task.id)]) self.assertEqual(self.client.delete(f"/api/ai/tasks/{task.id}/workbench-batch/").status_code, 200) self.assertEqual(self.client.get(f"/api/ai/image-conversations/{conversation.id}/tasks/").json()["tasks"], []) self.assertEqual(self.client.post(f"/api/ai/tasks/{task.id}/restore-workbench-batch/").status_code, 200) task_rows = self.client.get(f"/api/ai/image-conversations/{conversation.id}/tasks/").json()["tasks"] self.assertEqual([row["id"] for row in task_rows], [str(task.id)]) def test_inflight_batch_cannot_be_deleted(self): tasks = self._generate(mode="cover", count=1, platform_id="douyin") task = tasks[0] task.status = AITask.Status.POLLING task.save(update_fields=["status"]) deleted = self.client.delete(f"/api/ai/tasks/{task.id}/workbench-batch/") self.assertEqual(deleted.status_code, 409, deleted.content) task.refresh_from_db() self.assertFalse(task.is_deleted) def test_other_team_cannot_delete_image_batch(self): task = self._generate(mode="cover", count=1, platform_id="douyin")[0] other = User.objects.create_user(username="other-batch-owner", password="pass") other_team = Team.objects.create(name="Other batch team", owner=other) TeamMember.objects.create(team=other_team, user=other, role="owner", status="active") other_client = APIClient() other_client.force_authenticate(other) deleted = other_client.delete(f"/api/ai/tasks/{task.id}/workbench-batch/") self.assertEqual(deleted.status_code, 404, deleted.content) task.refresh_from_db() self.assertFalse(task.is_deleted) def test_unread_is_team_scoped_and_mode_whitelisted(self): self._generate(mode="cover", count=1) # 非生成任务(脚本 agent 复用 mode=auto)不计入未读 AITask.objects.create( team=self.team, created_by=self.user, task_type=AITask.Type.SCRIPT_GENERATION, model_config=ModelConfig.objects.first(), idempotency_key="script-x", request_payload={"mode": "auto"}, ) self.assertEqual(self.client.get("/api/ai/tasks/unread/").json()["total"], 1) # 另一团队的用户看不到本团队的未读 other = User.objects.create_user(username="wbother", password="pass") other_team = Team.objects.create(name="WB2", owner=other) TeamMember.objects.create(team=other_team, user=other, role="owner", status="active") c2 = APIClient() c2.force_authenticate(other) self.assertEqual(c2.get("/api/ai/tasks/unread/").json()["total"], 0)