127 lines
4.6 KiB
Python
127 lines
4.6 KiB
Python
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()
|