fix: 修复换火山账号导致素材库操作失败问题

2026-07-09 17:54:05 +08:00
parent 8c50096888
commit 3e52942547
5 changed files with 247 additions and 15 deletions
@@ -3,11 +3,12 @@ from types import SimpleNamespace
 from unittest.mock import patch

 from django.core.files.uploadedfile import SimpleUploadedFile
-from django.test import TestCase
+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, FreeAsset, FreeAssetGroup, Model


@@ -152,6 +153,32 @@ class AssetTrashApiTests(TestCase):
        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。"""

@@ -187,10 +214,11 @@ class FreeAssetQuickUploadTests(TestCase):
        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, _enabled):
+    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"
@@ -202,8 +230,40 @@ class FreeAssetQuickUploadTests(TestCase):
        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")
@@ -219,6 +279,143 @@ class FreeAssetQuickUploadTests(TestCase):
        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 的真实视频资产)。"""