yingqing/core/backend/apps/assets/tests.py

from io import BytesIO, StringIO
from types import SimpleNamespace
from unittest.mock import patch

from django.core.management import call_command
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import SimpleTestCase, TestCase
from rest_framework.test import APIClient

from apps.accounts.models import Team, TeamMember, User

from . import assets_client
from .models import Asset, AssetFile, FreeAsset, FreeAssetGroup, Model


def _mk_team(username, team_name):
    user = User.objects.create_user(username=username, password="x")
    team = Team.objects.create(name=team_name, owner=user)
    TeamMember.objects.create(team=team, user=user, role=TeamMember.Role.OWNER)
    return user, team


class ModelLibraryApiTests(TestCase):
    """模特库:本团队模特 ∪ 官方模板;官方跨团队可见;软删;官方不可删;团队隔离。"""

    def setUp(self):
        self.user, self.teamA = _mk_team("ua", "TeamA")
        self.other, self.teamB = _mk_team("ub", "TeamB")
        self.client = APIClient()
        self.client.force_authenticate(self.user)

        self.mine = Model.objects.create(team=self.teamA, name="我的模特")
        self.official = Model.objects.create(team=self.teamB, name="官方小美", is_official=True)
        self.bs_other = Model.objects.create(team=self.teamB, name="别队私有")  # B 的私有,A 看不到

    def test_list_returns_mine_plus_official(self):
        ids = {m["id"] for m in self.client.get("/api/models/").json()["results"]}
        self.assertIn(str(self.mine.id), ids)
        self.assertIn(str(self.official.id), ids)  # 官方跨团队可见
        self.assertNotIn(str(self.bs_other.id), ids)  # 别队私有不可见

    def test_tab_official_and_mine(self):
        off = {m["id"] for m in self.client.get("/api/models/?tab=official").json()["results"]}
        self.assertEqual(off, {str(self.official.id)})
        mine = {m["id"] for m in self.client.get("/api/models/?tab=mine").json()["results"]}
        self.assertEqual(mine, {str(self.mine.id)})

    def test_filter_by_portrait_asset(self):
        portrait = Asset.objects.create(team=self.teamA, name="角色立绘", asset_type="image", source="ai_generated")
        self.mine.portrait_asset = portrait
        self.mine.save(update_fields=["portrait_asset"])

        rows = self.client.get(f"/api/models/?portrait_asset={portrait.id}").json()["results"]
        self.assertEqual([row["id"] for row in rows], [str(self.mine.id)])

        missing = self.client.get("/api/models/?portrait_asset=00000000-0000-0000-0000-000000000000").json()["results"]
        self.assertEqual(missing, [])

    def test_enroll_asset_creates_model_once_and_reuses_it(self):
        portrait = Asset.objects.create(
            team=self.teamA, name="AI 模特候选", asset_type="image", source="ai_generated", category=Asset.Category.PERSON,
        )

        first = self.client.post("/api/models/enroll-asset/", {"asset_id": str(portrait.id), "name": "通勤模特"}, format="json")
        self.assertEqual(first.status_code, 201)
        self.assertEqual(first.json()["name"], "通勤模特")
        self.assertEqual(first.json()["portrait_asset"], str(portrait.id))

        second = self.client.post("/api/models/enroll-asset/", {"asset_id": str(portrait.id), "name": "重复名称不应覆盖"}, format="json")
        self.assertEqual(second.status_code, 200)
        self.assertEqual(second.json()["id"], first.json()["id"])
        self.assertEqual(Model.objects.filter(team=self.teamA, portrait_asset=portrait, is_deleted=False).count(), 1)

    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_update_own_model_profile_does_not_rewrite_portrait_asset(self):
        portrait = Asset.objects.create(team=self.teamA, name="原始形象资产名", asset_type="image", source="upload")
        self.mine.portrait_asset = portrait
        self.mine.save(update_fields=["portrait_asset"])

        res = self.client.patch(
            f"/api/models/{self.mine.id}/",
            {"name": "更新后的模特名", "description": "通勤风格"},
            format="json",
        )

        self.assertEqual(res.status_code, 200)
        self.mine.refresh_from_db()
        portrait.refresh_from_db()
        self.assertEqual(self.mine.name, "更新后的模特名")
        self.assertEqual(self.mine.description, "通勤风格")
        self.assertEqual(portrait.name, "原始形象资产名")

    def test_official_model_cannot_be_edited(self):
        res = self.client.patch(f"/api/models/{self.official.id}/", {"name": "不应更新"}, format="json")
        self.assertEqual(res.status_code, 403)
        self.official.refresh_from_db()
        self.assertEqual(self.official.name, "官方小美")

    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_trash_restore_and_purge_model_keeps_record_and_asset(self):
        portrait = Asset.objects.create(team=self.teamA, name="模特形象", asset_type="image", source="upload")
        portrait_file = AssetFile.objects.create(
            asset=portrait,
            object_key="models/portrait.png",
            bucket="test",
            content_type="image/png",
            is_primary=True,
        )
        self.mine.portrait_asset = portrait
        self.mine.save(update_fields=["portrait_asset"])

        self.client.delete(f"/api/models/{self.mine.id}/")
        trash_ids = {m["id"] for m in self.client.get("/api/models/trash/").json()["results"]}
        self.assertIn(str(self.mine.id), trash_ids)
        portrait.refresh_from_db()
        self.assertFalse(portrait.is_deleted)
        self.assertIsNone(portrait.purged_at)

        restore = self.client.post(f"/api/models/{self.mine.id}/restore/")
        self.assertEqual(restore.status_code, 200)
        self.mine.refresh_from_db()
        self.assertFalse(self.mine.is_deleted)
        self.assertIsNone(self.mine.purged_at)
        ids = {m["id"] for m in self.client.get("/api/models/?tab=mine").json()["results"]}
        self.assertIn(str(self.mine.id), ids)

        self.client.delete(f"/api/models/{self.mine.id}/")
        purge = self.client.delete(f"/api/models/{self.mine.id}/purge/")
        self.assertEqual(purge.status_code, 204)
        self.mine.refresh_from_db()
        self.assertTrue(self.mine.is_deleted)
        self.assertIsNotNone(self.mine.purged_at)
        self.assertTrue(Model.objects.filter(id=self.mine.id).exists())
        self.assertTrue(Asset.objects.filter(id=portrait.id).exists())
        portrait.refresh_from_db()
        self.assertFalse(portrait.is_deleted)
        self.assertIsNone(portrait.purged_at)
        self.assertTrue(AssetFile.objects.filter(id=portrait_file.id).exists())
        trash_ids = {m["id"] for m in self.client.get("/api/models/trash/").json()["results"]}
        self.assertNotIn(str(self.mine.id), trash_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)

    def test_upload_portrait_replaces_current_model_only_and_stays_out_of_library(self):
        old_portrait = Asset.objects.create(
            team=self.teamA, name="旧形象", asset_type="image", source="upload", category=Asset.Category.MODEL_PORTRAIT,
        )
        self.mine.portrait_asset = old_portrait
        self.mine.save(update_fields=["portrait_asset"])
        stored = SimpleNamespace(object_key="teams/x/models/new.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
            image = SimpleUploadedFile("new.png", b"img", content_type="image/png")
            res = self.client.post(
                f"/api/models/{self.mine.id}/upload-portrait/",
                {"file": image, "name": "一次保存的新模特名", "description": "一次保存的新描述"},
                format="multipart",
            )

        self.assertEqual(res.status_code, 200)
        self.mine.refresh_from_db()
        self.assertNotEqual(self.mine.portrait_asset_id, old_portrait.id)
        self.assertEqual(self.mine.name, "一次保存的新模特名")
        self.assertEqual(self.mine.description, "一次保存的新描述")
        new_portrait = self.mine.portrait_asset
        self.assertEqual(new_portrait.category, Asset.Category.MODEL_PORTRAIT)
        self.assertFalse(new_portrait.in_library)
        self.assertTrue(Asset.objects.filter(id=old_portrait.id).exists())
        self.assertEqual(Model.objects.filter(team=self.teamA).count(), 1)
        self.assertEqual(
            self.mine.metadata["portrait_versions"],
            [str(old_portrait.id), str(new_portrait.id)],
        )
        library_ids = {row["id"] for row in self.client.get("/api/assets/?page_size=200").json()["results"]}
        self.assertNotIn(str(new_portrait.id), library_ids)

    def test_official_model_cannot_replace_portrait(self):
        with patch("apps.assets.views.TosStorage") as Tos:
            image = SimpleUploadedFile("new.png", b"img", content_type="image/png")
            res = self.client.post(f"/api/models/{self.official.id}/upload-portrait/", {"file": image}, format="multipart")
        self.assertEqual(res.status_code, 403)
        Tos.assert_not_called()


class ActorModelMigrationAuditTests(TestCase):
    """审计命令只能分类和输出，绝不能改写旧资产、项目或模特记录。"""

    def setUp(self):
        from apps.products.models import Product
        from apps.projects.models import BaseAssetGroup, Project

        self.user, self.team = _mk_team("audit-u", "Audit")
        self.eligible = Asset.objects.create(
            team=self.team, name="旧保存演员", asset_type="image", source="upload", category=Asset.Category.PERSON, in_library=True,
        )
        existing = Asset.objects.create(
            team=self.team, name="已有模特形象", asset_type="image", source="upload", category=Asset.Category.PERSON, in_library=True,
        )
        Model.objects.create(team=self.team, name="已有模特", portrait_asset=existing)
        role = Asset.objects.create(
            team=self.team, name="项目角色", asset_type="image", source="ai_generated", category=Asset.Category.PERSON, in_library=True,
        )
        product = Product.objects.create(team=self.team, title="测试商品")
        project = Project.objects.create(team=self.team, name="测试项目", product=product)
        BaseAssetGroup.objects.create(project=project, kind=BaseAssetGroup.Kind.PERSON, adopted_asset=role)
        Asset.objects.create(
            team=self.team, name="垃圾桶演员", asset_type="image", source="upload", category=Asset.Category.PERSON, is_deleted=True,
        )

    def test_audit_is_read_only_and_excludes_project_and_existing_models(self):
        before_assets = Asset.objects.count()
        before_models = Model.objects.count()
        out = StringIO()

        call_command("audit_actor_model_migration", "--team-id", str(self.team.id), stdout=out)

        report = out.getvalue()
        self.assertIn("旧已保存人物候选: 1", report)
        self.assertIn(str(self.eligible.id), report)
        self.assertIn("垃圾桶 person（不迁移）: 1", report)
        self.assertIn("未创建、更新、删除任何记录或文件", report)
        self.assertEqual(Asset.objects.count(), before_assets)
        self.assertEqual(Model.objects.count(), before_models)

    def test_apply_migrates_only_eligible_saved_actor_and_is_idempotent(self):
        call_command("migrate_saved_actors_to_models", "--team-id", str(self.team.id), "--apply")

        model = Model.objects.get(team=self.team, portrait_asset=self.eligible)
        self.assertEqual(model.name, self.eligible.name)
        self.assertEqual(model.metadata.get("migrated_from_legacy_actor"), True)
        self.eligible.refresh_from_db()
        self.assertEqual(self.eligible.category, Asset.Category.PERSON)  # 历史分类不改
        self.assertFalse(self.eligible.is_deleted)
        self.assertEqual(Model.objects.filter(team=self.team, portrait_asset=self.eligible).count(), 1)
        self.assertFalse(Model.objects.filter(team=self.team, portrait_asset__name="项目角色").exists())

        call_command("migrate_saved_actors_to_models", "--team-id", str(self.team.id), "--apply")
        self.assertEqual(Model.objects.filter(team=self.team, portrait_asset=self.eligible).count(), 1)


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)

    def test_functional_model_and_project_assets_do_not_appear_in_others(self):
        model_portrait = Asset.objects.create(
            team=self.team, name="模特形象", asset_type="image", source="upload", category=Asset.Category.MODEL_PORTRAIT,
        )
        tri_view = Asset.objects.create(
            team=self.team, name="角色三视图", asset_type="image", source="ai_generated", category=Asset.Category.TRI_VIEW,
        )
        other = Asset.objects.create(
            team=self.team, name="其他图片", asset_type="image", source="upload", category=Asset.Category.UNCATEGORIZED,
        )

        ids = {row["id"] for row in self.client.get("/api/assets/?tab=others&page_size=50").json()["results"]}
        self.assertIn(str(other.id), ids)
        self.assertNotIn(str(model_portrait.id), ids)
        self.assertNotIn(str(tri_view.id), ids)


class AssetTrashApiTests(TestCase):
    """R108 资产垃圾桶:trash 只列本团队软删资产;restore 后资产库重新可见;purge 二级软删。"""

    def setUp(self):
        self.user, self.team = _mk_team("utr", "TeamTR")
        _, self.other_team = _mk_team("utr2", "TeamTR2")
        self.client = APIClient()
        self.client.force_authenticate(self.user)
        self.live = Asset.objects.create(team=self.team, name="活图", asset_type="image", source="ai_generated", category=Asset.Category.MODEL_TRYON)
        self.dead = Asset.objects.create(team=self.team, name="删掉的图", asset_type="image", source="ai_generated", category=Asset.Category.MODEL_TRYON, is_deleted=True)
        self.dead_file = AssetFile.objects.create(
            asset=self.dead,
            object_key="assets/deleted.png",
            bucket="test",
            content_type="image/png",
            is_primary=True,
        )
        # 别队软删资产:不得出现在本队垃圾桶(team 隔离)
        self.other_dead = Asset.objects.create(team=self.other_team, name="别队删图", asset_type="image", source="ai_generated", is_deleted=True)

    def test_trash_lists_only_own_deleted(self):
        ids = {a["id"] for a in self.client.get("/api/assets/trash/").json()["results"]}
        self.assertEqual(ids, {str(self.dead.id)})  # 只含本队软删;不含活图 / 别队删图

    def test_restore_makes_asset_visible_in_library_again(self):
        res = self.client.post(f"/api/assets/{self.dead.id}/restore/")
        self.assertEqual(res.status_code, 200)
        self.dead.refresh_from_db()
        self.assertFalse(self.dead.is_deleted)
        ids = {a["id"] for a in self.client.get("/api/assets/?page_size=50").json()["results"]}
        self.assertIn(str(self.dead.id), ids)  # 资产库重新可见
        # 恢复后不再在垃圾桶
        self.assertEqual(self.client.get("/api/assets/trash/").json()["count"], 0)

    def test_purge_soft_hides_from_trash(self):
        res = self.client.delete(f"/api/assets/{self.dead.id}/purge/")
        self.assertEqual(res.status_code, 204)
        self.dead.refresh_from_db()
        self.assertIsNotNone(self.dead.purged_at)
        self.assertTrue(Asset.objects.filter(id=self.dead.id).exists())
        self.assertTrue(AssetFile.objects.filter(id=self.dead_file.id).exists())
        self.assertEqual(self.client.get("/api/assets/trash/").json()["count"], 0)

    def test_purge_referenced_asset_keeps_record(self):
        from apps.products.models import Product, ProductImage

        product = Product.objects.create(team=self.team, created_by=self.user, title="P")
        ProductImage.objects.create(product=product, asset=self.dead)  # PROTECT 引用
        res = self.client.delete(f"/api/assets/{self.dead.id}/purge/")
        self.assertEqual(res.status_code, 204)
        self.dead.refresh_from_db()
        self.assertTrue(self.dead.is_deleted)
        self.assertIsNotNone(self.dead.purged_at)
        self.assertTrue(Asset.objects.filter(id=self.dead.id).exists())

    def test_restore_rejects_live_asset(self):
        # 未删资产不在 trash 态 queryset → 404(与商品垃圾桶口径一致)
        self.assertEqual(self.client.post(f"/api/assets/{self.live.id}/restore/").status_code, 404)


class AssetsClientErrorCodeTests(SimpleTestCase):
    def test_not_found_helpers_cover_volcano_variants(self):
        self.assertTrue(assets_client.is_not_found_code("NotFound"))
        self.assertTrue(assets_client.is_not_found_code("NotFound.asset_id"))
        self.assertTrue(assets_client.is_not_found_code("NotFound.group_id"))
        self.assertFalse(assets_client.is_not_found_code("Forbidden"))
        self.assertFalse(assets_client.is_not_found_code(None))

        self.assertTrue(assets_client.is_asset_not_found_code("NotFound"))
        self.assertTrue(assets_client.is_asset_not_found_code("NotFound.asset_id"))
        self.assertFalse(assets_client.is_asset_not_found_code("NotFound.group_id"))

        self.assertTrue(assets_client.is_group_not_found_code("NotFound"))
        self.assertTrue(assets_client.is_group_not_found_code("NotFound.group_id"))
        self.assertFalse(assets_client.is_group_not_found_code("NotFound.asset_id"))

    def test_not_found_variant_messages_are_not_generic(self):
        asset_error = assets_client.AssetsAPIError("NotFound.asset_id", "remote asset missing")
        group_error = assets_client.AssetsAPIError("NotFound.group_id", "remote group missing")
        unknown_error = assets_client.AssetsAPIError("Unknown", "unknown")

        self.assertNotEqual(asset_error.user_message, unknown_error.user_message)
        self.assertNotEqual(group_error.user_message, unknown_error.user_message)
        self.assertNotEqual(asset_error.user_message, group_error.user_message)


class FreeAssetQuickUploadTests(TestCase):
    """人物素材库快捷上传:自动默认组,但底层仍走 FreeAssetGroup / FreeAsset。"""

    def setUp(self):
        self.user, self.team = _mk_team("ufq", "TeamFQ")
        self.client = APIClient()
        self.client.force_authenticate(self.user)
        self.stored = SimpleNamespace(object_key="teams/x/free-create/library/a.png", bucket="b", content_type="image/png", size_bytes=10)

    def _image(self, name="a.png"):
        return SimpleUploadedFile(name, b"img", content_type="image/png")

    @patch("apps.assets.free_library.assets_client.is_enabled", return_value=True)
    @patch("apps.assets.free_library.assets_client.create_asset_group", return_value="Group-1")
    @patch("apps.assets.free_library.assets_client.create_asset", return_value="Asset-1")
    @patch("apps.assets.free_library.TosStorage")
    def test_quick_upload_creates_default_group(self, storage_cls, create_asset, create_group, _enabled):
        storage = storage_cls.return_value
        storage.upload_fileobj.return_value = self.stored
        storage.public_url.return_value = "http://tos/a.png"

        res = self.client.post("/api/assets/free-assets/quick-upload/", {"file": self._image()}, format="multipart")
        self.assertEqual(res.status_code, 201)
        group = FreeAssetGroup.objects.get(team=self.team)
        self.assertEqual(group.name, "默认素材")
        self.assertEqual(group.remote_group_id, "Group-1")
        self.assertEqual(group.assets.count(), 1)
        asset = group.assets.get()
        self.assertEqual(asset.asset_type, FreeAsset.Type.IMAGE)
        self.assertEqual(asset.status, FreeAsset.Status.PROCESSING)
        create_group.assert_called_once()
        create_asset.assert_called_once_with("Group-1", "http://tos/a.png", name="a.png", asset_type=FreeAsset.Type.IMAGE)
        self.assertEqual(res.json()["group"]["id"], str(group.id))

    @patch("apps.assets.free_library.assets_client.is_enabled", return_value=True)
    @patch("apps.assets.free_library.assets_client.get_asset_group", return_value={"Id": "Group-1"})
    @patch("apps.assets.free_library.assets_client.create_asset_group", return_value="Group-1")
    @patch("apps.assets.free_library.assets_client.create_asset", side_effect=["Asset-1", "Asset-2"])
    @patch("apps.assets.free_library.TosStorage")
    def test_quick_upload_reuses_default_group(self, storage_cls, create_asset, create_group, get_group, _enabled):
        storage = storage_cls.return_value
        storage.upload_fileobj.return_value = self.stored
        storage.public_url.return_value = "http://tos/a.png"

        first = self.client.post("/api/assets/free-assets/quick-upload/", {"file": self._image("a.png")}, format="multipart")
        second = self.client.post("/api/assets/free-assets/quick-upload/", {"file": self._image("b.png")}, format="multipart")
        self.assertEqual(first.status_code, 201)
        self.assertEqual(second.status_code, 201)
        self.assertEqual(FreeAssetGroup.objects.filter(team=self.team).count(), 1)
        self.assertEqual(FreeAsset.objects.filter(group__team=self.team).count(), 2)
        create_group.assert_called_once()
        get_group.assert_called_once_with("Group-1")
        self.assertEqual(create_asset.call_count, 2)

    @patch("apps.assets.free_library.assets_client.is_enabled", return_value=True)
    @patch(
        "apps.assets.free_library.assets_client.get_asset_group",
        side_effect=assets_client.AssetsAPIError("NotFound.group_id", "old group not visible"),
    )
    @patch("apps.assets.free_library.assets_client.create_asset_group", return_value="Group-2")
    @patch("apps.assets.free_library.assets_client.create_asset", return_value="Asset-2")
    @patch("apps.assets.free_library.TosStorage")
    def test_quick_upload_replaces_old_default_group(self, storage_cls, create_asset, create_group, get_group, _enabled):
        old_group = FreeAssetGroup.objects.create(
            team=self.team,
            created_by=self.user,
            name="默认素材",
            description="__airshelf_free_asset_default__",
            remote_group_id="Old-Group",
        )
        storage = storage_cls.return_value
        storage.upload_fileobj.return_value = self.stored
        storage.public_url.return_value = "http://tos/a.png"

        res = self.client.post("/api/assets/free-assets/quick-upload/", {"file": self._image("new.png")}, format="multipart")

        self.assertEqual(res.status_code, 201)
        self.assertFalse(FreeAsset.objects.filter(group=old_group).exists())
        new_group = FreeAssetGroup.objects.exclude(id=old_group.id).get(team=self.team)
        self.assertEqual(new_group.remote_group_id, "Group-2")
        self.assertEqual(new_group.assets.count(), 1)
        get_group.assert_called_once_with("Old-Group")
        create_group.assert_called_once()
        create_asset.assert_called_once_with("Group-2", "http://tos/a.png", name="new.png", asset_type=FreeAsset.Type.IMAGE)

    @patch("apps.assets.free_library.assets_client.is_enabled", return_value=True)
    @patch("apps.assets.free_library.assets_client.create_asset_group", return_value="Group-1")
    @patch("apps.assets.free_library.TosStorage")
    def test_quick_upload_rejects_non_image(self, storage_cls, create_group, _enabled):
        res = self.client.post(
            "/api/assets/free-assets/quick-upload/",
            {"file": SimpleUploadedFile("v.mp4", b"video", content_type="video/mp4")},
            format="multipart",
        )
        self.assertEqual(res.status_code, 400)
        self.assertIn("仅支持图片", res.json()["detail"])
        self.assertFalse(storage_cls.return_value.upload_fileobj.called)
        create_group.assert_called_once()


class FreeAssetGroupUploadTests(TestCase):
    def setUp(self):
        self.user, self.team = _mk_team("fgu", "TeamFGU")
        self.client = APIClient()
        self.client.force_authenticate(self.user)
        self.group = FreeAssetGroup.objects.create(
            team=self.team,
            created_by=self.user,
            name="旧素材组",
            description="",
            remote_group_id="Old-Group",
        )
        self.stored = SimpleNamespace(object_key="teams/x/free-create/library/a.png", bucket="b", content_type="image/png", size_bytes=10)

    def _image(self, name="a.png"):
        return SimpleUploadedFile(name, b"img", content_type="image/png")

    @patch("apps.assets.free_library.assets_client.is_enabled", return_value=True)
    @patch(
        "apps.assets.free_library.assets_client.create_asset",
        side_effect=assets_client.AssetsAPIError("NotFound.group_id", "old group not visible"),
    )
    @patch("apps.assets.free_library.TosStorage")
    def test_upload_to_old_remote_group_returns_clear_message(self, storage_cls, create_asset, _enabled):
        storage = storage_cls.return_value
        storage.upload_fileobj.return_value = self.stored
        storage.public_url.return_value = "http://tos/a.png"

        res = self.client.post(f"/api/assets/free-groups/{self.group.id}/assets/", {"file": self._image()}, format="multipart")

        self.assertEqual(res.status_code, 400)
        self.assertEqual(res.json()["detail"], "该素材组属于其他火山账号，请新建素材组后重新上传")
        self.assertFalse(FreeAsset.objects.filter(group=self.group).exists())
        create_asset.assert_called_once_with("Old-Group", "http://tos/a.png", name="a.png", asset_type=FreeAsset.Type.IMAGE)


class FreeAssetDeleteCompatibilityTests(TestCase):
    def setUp(self):
        self.user, self.team = _mk_team("fdel", "TeamFDEL")
        self.client = APIClient()
        self.client.force_authenticate(self.user)
        self.group = FreeAssetGroup.objects.create(
            team=self.team,
            created_by=self.user,
            name="旧素材组",
            description="",
            remote_group_id="Old-Group",
        )
        self.asset = FreeAsset.objects.create(
            group=self.group,
            remote_asset_id="Old-Asset",
            name="旧素材.png",
            url="http://tos/a.png",
            asset_type=FreeAsset.Type.IMAGE,
            thumbnail_url="http://tos/a.png",
            status=FreeAsset.Status.ACTIVE,
        )

    @patch(
        "apps.assets.free_library.assets_client.delete_asset",
        side_effect=assets_client.AssetsAPIError("NotFound.asset_id", "old asset not visible"),
    )
    def test_delete_old_remote_asset_clears_local_asset(self, delete_asset):
        res = self.client.delete(f"/api/assets/free-assets/{self.asset.id}/")

        self.assertEqual(res.status_code, 204)
        self.assertFalse(FreeAsset.objects.filter(id=self.asset.id).exists())
        delete_asset.assert_called_once_with("Old-Asset")

    @patch(
        "apps.assets.free_library.assets_client.delete_asset_group",
        side_effect=assets_client.AssetsAPIError("NotFound.group_id", "old group not visible"),
    )
    def test_delete_old_remote_group_marks_local_group_deleted(self, delete_group):
        res = self.client.delete(f"/api/assets/free-groups/{self.group.id}/")

        self.assertEqual(res.status_code, 204)
        self.group.refresh_from_db()
        self.assertTrue(self.group.is_deleted)
        self.assertTrue(FreeAsset.objects.filter(id=self.asset.id).exists())
        delete_group.assert_called_once_with("Old-Group")


class FreeAssetRenameCompatibilityTests(TestCase):
    def setUp(self):
        self.user, self.team = _mk_team("fren", "TeamFREN")
        self.client = APIClient()
        self.client.force_authenticate(self.user)
        self.group = FreeAssetGroup.objects.create(
            team=self.team,
            created_by=self.user,
            name="旧素材组",
            description="old description",
            remote_group_id="Old-Group",
        )
        self.asset = FreeAsset.objects.create(
            group=self.group,
            remote_asset_id="Old-Asset",
            name="旧素材.png",
            url="http://tos/a.png",
            asset_type=FreeAsset.Type.IMAGE,
            thumbnail_url="http://tos/a.png",
            status=FreeAsset.Status.ACTIVE,
        )

    @patch(
        "apps.assets.free_library.assets_client.update_asset",
        side_effect=assets_client.AssetsAPIError("NotFound.asset_id", "old asset not visible"),
    )
    def test_rename_old_remote_asset_returns_clear_message_without_saving(self, update_asset):
        res = self.client.patch(f"/api/assets/free-assets/{self.asset.id}/", {"name": "新名字.png"}, format="json")

        self.assertEqual(res.status_code, 400)
        self.assertEqual(res.json()["detail"], "该素材属于其他火山账号，请重新上传后再编辑")
        self.asset.refresh_from_db()
        self.assertEqual(self.asset.name, "旧素材.png")
        update_asset.assert_called_once_with("Old-Asset", name="新名字.png")

    @patch(
        "apps.assets.free_library.assets_client.update_asset_group",
        side_effect=assets_client.AssetsAPIError("NotFound.group_id", "old group not visible"),
    )
    def test_rename_old_remote_group_returns_clear_message_without_saving(self, update_group):
        res = self.client.patch(
            f"/api/assets/free-groups/{self.group.id}/",
            {"name": "新素材组", "description": "new description"},
            format="json",
        )

        self.assertEqual(res.status_code, 400)
        self.assertEqual(res.json()["detail"], "该素材组属于其他火山账号，请新建素材组后重新上传")
        self.group.refresh_from_db()
        self.assertEqual(self.group.name, "旧素材组")
        self.assertEqual(self.group.description, "old description")
        update_group.assert_called_once_with("Old-Group", name="新素材组", description="new description")


class VideoPacksTests(TestCase):
    """视频成品按项目素材包分组(与导出同源:取 VideoSegment.adopted_version 的真实视频资产)。"""

    def setUp(self):
        from apps.products.models import Product
        from apps.projects.models import Project, VideoSegment, VideoSegmentVersion
        from .models import AssetFile

        self.user, self.team = _mk_team("uvp", "TeamVP")
        self.client = APIClient()
        self.client.force_authenticate(self.user)
        self.product = Product.objects.create(team=self.team, created_by=self.user, title="P")
        self.proj = Project.objects.create(team=self.team, name="项目甲", product=self.product, created_by=self.user)
        self.assets = []
        for i in range(2):
            a = Asset.objects.create(team=self.team, name=f"clip{i}", asset_type="video", source="ai_generated", category=Asset.Category.VIDEO_CLIP)
            AssetFile.objects.create(asset=a, object_key=f"k{i}.mp4", bucket="b", content_type="video/mp4", is_primary=True)
            seg = VideoSegment.objects.create(project=self.proj, sort_order=i)
            ver = VideoSegmentVersion.objects.create(video_segment=seg, asset=a, is_adopted=True)
            seg.adopted_version = ver
            seg.save(update_fields=["adopted_version"])
            self.assets.append(a)

    def test_packs_grouped_by_project(self):
        data = self.client.get("/api/assets/video-packs/").json()
        self.assertEqual(len(data), 1)
        self.assertEqual(data[0]["project_name"], "项目甲")
        self.assertEqual(len(data[0]["clips"]), 2)
        self.assertIn("url", data[0]["clips"][0])

    def test_soft_deleted_clip_excluded(self):
        Asset.objects.filter(id=self.assets[0].id).update(is_deleted=True)
        data = self.client.get("/api/assets/video-packs/").json()
        self.assertEqual(len(data[0]["clips"]), 1)


class AssetBatchesTests(TestCase):
    """图片趴(模特上身图)按生成批次成组:同 metadata.batch_id 的多张图 = 一批;
    无 batch_id 的旧图各自成单图批。summary 也按批次计数。"""

    def setUp(self):
        self.user, self.team = _mk_team("ubt", "TeamBT")
        self.client = APIClient()
        self.client.force_authenticate(self.user)
        # 一批 3 张(同 batch_id)
        for i in range(3):
            Asset.objects.create(
                team=self.team, name=f"AI 生成 · 模特上身图 · {i + 1}", asset_type="image",
                source="ai_generated", category=Asset.Category.MODEL_TRYON, metadata={"batch_id": "B1"},
            )
        # 一张旧图(无 batch_id)→ 自己成一批
        Asset.objects.create(
            team=self.team, name="AI 生成 · 模特上身图 · 1", asset_type="image",
            source="ai_generated", category=Asset.Category.MODEL_TRYON,
        )

    def test_batches_grouped(self):
        data = self.client.get("/api/assets/batches/?tab=tryon&page_size=10").json()
        self.assertEqual(data["count"], 2)  # 一批 3 张 + 一张单图批
        by_count = sorted(b["count"] for b in data["results"])
        self.assertEqual(by_count, [1, 3])
        big = next(b for b in data["results"] if b["count"] == 3)
        self.assertEqual(big["name"], "AI 生成 · 模特上身图")  # 去掉「· 序号」
        self.assertEqual(len(big["items"]), 3)

    def test_summary_counts_batches(self):
        self.assertEqual(self.client.get("/api/assets/summary/").json()["tryon"], 2)


class FreeCreateSplitTabTests(TestCase):
    """Free create assets share category=free_create; tabs split by asset_type."""

    def setUp(self):
        self.user, self.team = _mk_team("ufc", "TeamFC")
        self.client = APIClient()
        self.client.force_authenticate(self.user)
        self.image1 = Asset.objects.create(
            team=self.team,
            name="free image 1",
            asset_type=Asset.Type.IMAGE,
            source=Asset.Source.AI_GENERATED,
            category=Asset.Category.FREE_CREATE,
            metadata={"batch_id": "IMG-B1"},
        )
        self.image2 = Asset.objects.create(
            team=self.team,
            name="free image 2",
            asset_type=Asset.Type.IMAGE,
            source=Asset.Source.AI_GENERATED,
            category=Asset.Category.FREE_CREATE,
            metadata={"batch_id": "IMG-B1"},
        )
        self.video = Asset.objects.create(
            team=self.team,
            name="free video",
            asset_type=Asset.Type.VIDEO,
            source=Asset.Source.AI_GENERATED,
            category=Asset.Category.FREE_CREATE,
            metadata={"feature": "free_video"},
        )

    def test_image_creations_batches_exclude_videos(self):
        data = self.client.get("/api/assets/batches/?tab=image_creations&page_size=10").json()
        self.assertEqual(data["count"], 1)
        self.assertEqual(data["results"][0]["count"], 2)
        self.assertEqual({a["asset_type"] for a in data["results"][0]["items"]}, {"image"})

    def test_video_creations_list_excludes_images(self):
        data = self.client.get("/api/assets/?tab=video_creations&page_size=10").json()
        self.assertEqual(data["count"], 1)
        self.assertEqual(data["results"][0]["id"], str(self.video.id))
        self.assertEqual(data["results"][0]["asset_type"], "video")

    def test_summary_exposes_split_counts_and_keeps_legacy_creations(self):
        data = self.client.get("/api/assets/summary/").json()
        self.assertEqual(data["image_creations"], 1)
        self.assertEqual(data["video_creations"], 1)
        self.assertEqual(data["creations"], 2)

    def test_delete_and_restore_video_returns_to_video_creations_tab(self):
        res = self.client.delete(f"/api/assets/{self.video.id}/")
        self.assertEqual(res.status_code, 204)
        self.video.refresh_from_db()
        self.assertTrue(self.video.is_deleted)

        video_tab = self.client.get("/api/assets/?tab=video_creations&page_size=10").json()
        self.assertEqual(video_tab["count"], 0)
        summary = self.client.get("/api/assets/summary/").json()
        self.assertEqual(summary["video_creations"], 0)
        self.assertEqual(summary["image_creations"], 1)
        trash_ids = {a["id"] for a in self.client.get("/api/assets/trash/").json()["results"]}
        self.assertIn(str(self.video.id), trash_ids)

        restore = self.client.post(f"/api/assets/{self.video.id}/restore/")
        self.assertEqual(restore.status_code, 200)
        self.video.refresh_from_db()
        self.assertFalse(self.video.is_deleted)
        self.assertEqual(self.video.category, Asset.Category.FREE_CREATE)
        self.assertEqual(self.video.asset_type, Asset.Type.VIDEO)

        video_tab = self.client.get("/api/assets/?tab=video_creations&page_size=10").json()
        self.assertEqual(video_tab["count"], 1)
        self.assertEqual(video_tab["results"][0]["id"], str(self.video.id))
        image_batches = self.client.get("/api/assets/batches/?tab=image_creations&page_size=10").json()
        self.assertEqual({a["asset_type"] for a in image_batches["results"][0]["items"]}, {"image"})


class FacetsProductsTests(TestCase):
    """facets 返回本 tab 下真实关联的商品(?product= 三路之一:metadata.product_id 等),
    并按 tab 隔离、team 隔离、只回 id/title。"""

    def setUp(self):
        from apps.products.models import Product

        self.user, self.team = _mk_team("ufp", "TeamFP")
        self.client = APIClient()
        self.client.force_authenticate(self.user)
        self.p1 = Product.objects.create(team=self.team, created_by=self.user, title="商品甲")
        self.p2 = Product.objects.create(team=self.team, created_by=self.user, title="商品乙")
        # tryon 图经 metadata.product_id 关联 p1
        Asset.objects.create(
            team=self.team, name="上身图", asset_type="image", source="ai_generated",
            category=Asset.Category.MODEL_TRYON, metadata={"product_id": str(self.p1.id)},
        )
        # kits 图关联 p2(验证 tab 隔离:tryon 下不应出现 p2)
        Asset.objects.create(
            team=self.team, name="套图", asset_type="image", source="ai_generated",
            category=Asset.Category.PLATFORM_KIT, metadata={"product_id": str(self.p2.id)},
        )

    def test_facets_products_scoped_by_tab(self):
        data = self.client.get("/api/assets/facets/?tab=tryon").json()
        self.assertIn("products", data)
        self.assertEqual({p["id"] for p in data["products"]}, {str(self.p1.id)})
        self.assertEqual({p["title"] for p in data["products"]}, {"商品甲"})
        # kits tab 只见 p2
        kits = self.client.get("/api/assets/facets/?tab=kits").json()
        self.assertEqual({p["id"] for p in kits["products"]}, {str(self.p2.id)})

    def test_facets_products_empty_when_none(self):
        data = self.client.get("/api/assets/facets/?tab=others").json()
        self.assertEqual(data["products"], [])

    def test_facets_products_newest_first(self):
        from datetime import timedelta

        from django.utils import timezone
        from apps.products.models import Product

        Asset.objects.create(
            team=self.team, name="上身图2", asset_type="image", source="ai_generated",
            category=Asset.Category.MODEL_TRYON, metadata={"product_id": str(self.p2.id)},
        )
        Product.objects.filter(id=self.p1.id).update(created_at=timezone.now() - timedelta(days=1))
        data = self.client.get("/api/assets/facets/?tab=tryon").json()
        self.assertEqual([p["id"] for p in data["products"]], [str(self.p2.id), str(self.p1.id)])


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, *, force=False):
            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_only_covers_assets_that_really_reached_volcano(self):
        """轮询的判据是「拿到了火山远端 Id」，不是类别。

        force 送审的上传视频、临时图也会是 processing，worker 必须盯到绿/红；
        反过来，没有远端 Id 的 processing 是没真正送出去的，轮它只会白打请求。
        送审范围仍然只挑 REVIEW_CATEGORIES（见下一条）。
        """
        from apps.assets import review

        _, team = _mk_team("uc", "TeamC")
        polled = {}
        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",
                                     review_remote_id=f"remote-{cat}")
            polled[cat] = str(a.id)
        # 图片趴不在送审范围里，但已经 force 送出去的同样要盯到终态。
        forced = Asset.objects.create(
            team=team, name="force", asset_type="image", source="ai_generated",
            category="free_create", review_status="processing", review_remote_id="remote-force",
        )
        # 没有远端 Id = 没真送出去，不该被轮询。
        Asset.objects.create(team=team, name="没送出去", asset_type="image", source="ai_generated",
                             category="person", 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(polled.values()) | {str(forced.id)})

    def test_submission_still_only_covers_review_categories(self):
        from apps.assets import review

        _, team = _mk_team("uc-sub", "TeamCSub")
        for cat in ("person", "product_image"):
            Asset.objects.create(team=team, name=cat, asset_type="image", source="ai_generated",
                                 category=cat, review_status="")
        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="")

        with patch("apps.assets.review.submit_asset_for_review", return_value=True) as sub:
            review.submit_unsubmitted_reviews(team=team)

        submitted = {asset.category for (asset,), _ in (call for call in sub.call_args_list)}
        self.assertEqual(submitted, {"person", "product_image"})

    def test_poll_team_reviews_auto_submits_unsubmitted(self):
        from apps.assets import review

        _, team = _mk_team("usub", "TeamUSub")
        empty = Asset.objects.create(
            team=team, name="待审角色", asset_type="image", source="ai_generated",
            category=Asset.Category.PERSON, review_status="",
        )
        with patch("apps.assets.review.submit_asset_for_review", return_value=True) as sub, patch(
            "apps.assets.review.ensure_review_drain_loop"
        ):
            review.poll_team_reviews(team)
        sub.assert_called()
        self.assertEqual(sub.call_args.args[0].id, empty.id)

    def test_drain_platform_reviews_submits_then_polls(self):
        from apps.assets import review

        _, team = _mk_team("drain", "TeamDrain")
        Asset.objects.create(
            team=team, name="未送", asset_type="image", source="ai_generated",
            category=Asset.Category.PERSON, review_status="",
        )
        proc = Asset.objects.create(
            team=team, name="审中", asset_type="image", source="ai_generated",
            category=Asset.Category.PERSON, review_status="processing",
            review_remote_id="remote-proc",  # 只有真的送到火山、拿到 Id 的才会被轮询
        )
        with patch("apps.assets.review.submit_asset_for_review", return_value=True) as sub, patch(
            "apps.assets.review.poll_asset_review", return_value="active"
        ) as polled:
            out = review.drain_platform_reviews()
        self.assertEqual(out["submitted"], 1)
        self.assertEqual(out["polled"], 1)
        self.assertEqual(out["statuses"][str(proc.id)], "active")
        sub.assert_called_once()
        polled.assert_called_once()


class ImageSizeGuardTests(TestCase):
    """火山素材审核要求图片 300–6000 px。小图以前能一路存进商品库,
    直到生成视频送审那一刻才炸([InvalidParameter.WidthTooSmall]) —— 报错离上传很远,
    用户对不上号。改成入库前就挡住。"""

    def _png(self, width, height):
        from io import BytesIO

        from PIL import Image

        buf = BytesIO()
        Image.new("RGB", (width, height), "white").save(buf, format="PNG")
        return buf.getvalue()

    def test_rejects_image_below_300px(self):
        from apps.assets.image_guard import ImageTooSmallError, ensure_reviewable_image

        with self.assertRaises(ImageTooSmallError) as ctx:
            ensure_reviewable_image(self._png(120, 800), label="商品图")
        self.assertIn("商品图", str(ctx.exception))
        self.assertIn("300", str(ctx.exception))
        self.assertIn("120×800", str(ctx.exception))   # 把实际尺寸写给用户,免得反复试

    def test_rejects_image_above_6000px(self):
        from apps.assets.image_guard import ImageTooSmallError, ensure_reviewable_image

        with self.assertRaises(ImageTooSmallError):
            ensure_reviewable_image(self._png(6400, 900))

    def test_accepts_image_in_range_and_returns_size(self):
        from apps.assets.image_guard import ensure_reviewable_image

        self.assertEqual(ensure_reviewable_image(self._png(800, 1200)), (800, 1200))

    def test_broken_file_is_rejected(self):
        from apps.assets.image_guard import ImageTooSmallError, ensure_reviewable_image

        with self.assertRaises(ImageTooSmallError):
            ensure_reviewable_image(b"not-an-image")

    def test_volcano_error_codes_are_humanized(self):
        from apps.assets.review import humanize_submit_error

        self.assertIn(
            "图片太小",
            humanize_submit_error("[InvalidParameter.WidthTooSmall] Width must be between 300px and 6000px."),
        )
        self.assertIn("格式不支持", humanize_submit_error("[InvalidParameter.InvalidImageFormat] bad"))
        # 认不出的原样带出,不要吞掉
        self.assertEqual(humanize_submit_error("Some brand new error"), "Some brand new error")