feat(avatar): honor square interaction authorization

2026-09-08 13:19:50 +08:00
parent 2a01a9946a
commit 434caac056
4 changed files with 432 additions and 18 deletions
@@ -0,0 +1,126 @@
+import json
+import os
+import sqlite3
+import tempfile
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from app.services import avatar_service
+
+
+class AvatarSquareAuthorizationTests(unittest.TestCase):
+    def setUp(self):
+        fd, self.db_path = tempfile.mkstemp(suffix=".db")
+        os.close(fd)
+        connection = sqlite3.connect(self.db_path)
+        connection.executescript("""
+            CREATE TABLE users (
+                huihui_user_id TEXT,
+                nickname TEXT,
+                avatar_url TEXT,
+                huihui_token TEXT
+            );
+            CREATE TABLE avatars (
+                id TEXT,
+                owner_id TEXT,
+                name TEXT,
+                display_name TEXT,
+                description TEXT,
+                photo_url TEXT,
+                config TEXT,
+                status TEXT
+            );
+        """)
+        connection.execute(
+            "INSERT INTO users VALUES (?, ?, ?, ?)",
+            ("huihui-7", "主人", "/owner.jpg", "huihui-token"),
+        )
+        connection.commit()
+        connection.close()
+        avatar_service._engine = None
+        avatar_service._SessionLocal = None
+        self.path_patch = patch.object(avatar_service.settings, "AVATAR_DB_PATH", self.db_path)
+        self.path_patch.start()
+
+    def tearDown(self):
+        self.path_patch.stop()
+        if avatar_service._engine is not None:
+            avatar_service._engine.dispose()
+        avatar_service._engine = None
+        avatar_service._SessionLocal = None
+        os.unlink(self.db_path)
+
+    def _insert_avatar(self, permissions, *, status="active", token=None):
+        connection = sqlite3.connect(self.db_path)
+        connection.execute(
+            "INSERT INTO avatars VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                "avatar-7",
+                "huihui-7",
+                "avatar",
+                "小会",
+                "语气友好，表达简洁",
+                "/avatar.jpg",
+                json.dumps({
+                    "authorizationPermissions": permissions,
+                    "replyStyle": "warm",
+                    "responseLength": "short",
+                }),
+                status,
+            ),
+        )
+        if token is not None:
+            connection.execute(
+                "UPDATE users SET huihui_token = ? WHERE huihui_user_id = ?",
+                (token, "huihui-7"),
+            )
+        connection.commit()
+        connection.close()
+
+    def test_interact_permission_exposes_only_requested_square_actions(self):
+        self._insert_avatar(["chat", "interact"])
+
+        permissions = avatar_service.get_square_interaction_permissions("avatar-7")
+
+        self.assertEqual(
+            permissions,
+            frozenset({"like", "collect", "comment", "reply"}),
+        )
+        self.assertNotIn("forward", permissions)
+
+    def test_missing_permission_inactive_avatar_or_missing_token_denies_execution(self):
+        scenarios = [
+            (["chat"], "active", "huihui-token"),
+            (["interact"], "inactive", "huihui-token"),
+            (["interact"], "active", ""),
+        ]
+        for permissions, status, token in scenarios:
+            with self.subTest(permissions=permissions, status=status, token=token):
+                connection = sqlite3.connect(self.db_path)
+                connection.execute("DELETE FROM avatars")
+                connection.commit()
+                connection.close()
+                self._insert_avatar(permissions, status=status, token=token)
+                self.assertEqual(
+                    avatar_service.get_square_interaction_permissions("avatar-7"),
+                    frozenset(),
+                )
+
+    def test_delegated_avatar_identity_is_recognized_without_matching_normal_users(self):
+        delegated = SimpleNamespace(account="__avatar__:avatar-7")
+        normal = SimpleNamespace(account="13800000000")
+
+        self.assertTrue(avatar_service.is_delegated_avatar_user(delegated))
+        self.assertEqual(avatar_service.delegated_avatar_id(delegated), "avatar-7")
+        self.assertFalse(avatar_service.is_delegated_avatar_user(normal))
+        self.assertEqual(avatar_service.delegated_avatar_id(normal), "")
+
+    def test_response_length_maps_to_scheduler_comment_limits(self):
+        self.assertEqual(avatar_service._word_count_range({"responseLength": "short"}), (10, 35))
+        self.assertEqual(avatar_service._word_count_range({"responseLength": "long"}), (30, 80))
+        self.assertEqual(avatar_service._word_count_range({"responseLength": "unknown"}), (20, 60))
+
+
+if __name__ == "__main__":
+    unittest.main()