"""Offline unit tests for the scope-isolated sticker store.""" import json from pathlib import Path import sys import tempfile import unittest from unittest.mock import patch # Locally this file and sticker_store.py share a directory. In deployment the # test lives in backend/tests and the module in bot-utility/chat. TEST_DIR = Path(__file__).resolve().parent BACKEND = TEST_DIR.parent sys.path[:0] = [ str(TEST_DIR), str(BACKEND / "bot-utility" / "chat"), str(BACKEND / "bot-utility"), ] from sticker_store import SCENES, StickerStore, scene_prompt # noqa: E402 class StickerStoreTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.base = Path(self.temporary.name) self.store = StickerStore(self.base / "store") self.assets = self.base / "assets" self.assets.mkdir() def tearDown(self): self.temporary.cleanup() def make_source(self, name="one.webp", unique_id="unique-1", body=b"sticker"): original = self.assets / name original.write_bytes(body) preview = self.assets / f"{original.stem}-preview.png" preview.write_bytes(b"preview-" + body) return { "original_path": str(original), "url": str(preview), "file_unique_id": unique_id, "emoji": "🙂", "set_name": "test-set", "source_uid": "sender-1", } def test_scene_prompt_lists_all_supported_scenes(self): prompt = scene_prompt() for scene, chinese_description in SCENES.items(): self.assertIn(scene, prompt) self.assertIn(chinese_description, prompt) def test_learn_saves_object_preview_index_and_metadata(self): source = self.make_source() record = self.store.learn( "user-1", source, {"accept": True, "scenes": ["happy", "daily_cute"], "description": "开心挥手"}, ) self.assertIsNotNone(record) self.assertEqual(record["scenes"], ["happy", "daily_cute"]) self.assertEqual(record["description"], "开心挥手") self.assertEqual(record["emoji"], "🙂") self.assertEqual(record["set_name"], "test-set") self.assertEqual(record["source_uid"], "sender-1") self.assertEqual(record["file_unique_id"], "unique-1") self.assertEqual(record["format"], "webp") self.assertTrue(record["created_at"]) self.assertEqual(len(record["id"]), 32) object_path = Path(record["path"]) preview_path = Path(record["preview_path"]) self.assertTrue(object_path.is_absolute()) self.assertTrue(preview_path.is_absolute()) self.assertEqual(object_path.read_bytes(), b"sticker") self.assertEqual(preview_path.read_bytes(), b"preview-sticker") self.assertEqual(object_path.parent, self.base / "store" / "user-1" / "objects") self.assertEqual(preview_path.parent, self.base / "store" / "user-1" / "previews") index_path = self.base / "store" / "user-1" / "index.json" saved_index = json.loads(index_path.read_text(encoding="utf-8")) self.assertIn(record["id"], saved_index["stickers"]) self.assertEqual(StickerStore(self.base / "store").get("user-1", record["id"]), record) def test_invalid_scenes_and_explicit_rejection_are_not_learned(self): invalid_classifications = [ {"accept": False, "scenes": ["happy"]}, {"scenes": []}, {"scenes": ["not-a-scene"]}, {"scenes": ["happy", "not-a-scene"]}, {"scenes": ["happy", "sad", "angry", "thanks"]}, {"scenes": "happy"}, ] for number, classification in enumerate(invalid_classifications): with self.subTest(classification=classification): source = self.make_source( name=f"invalid-{number}.webp", unique_id=f"invalid-{number}" ) self.assertIsNone(self.store.learn("user-1", source, classification)) self.assertEqual(self.store.prompt_summary("user-1"), "") def test_random_for_scene_chooses_only_same_scene_candidates(self): first = self.store.learn( "user-1", self.make_source("first.webp", "first"), {"scenes": ["happy"], "description": "第一张"}, ) second = self.store.learn( "user-1", self.make_source("second.webp", "second"), {"scenes": ["happy", "sad"], "description": "第二张"}, ) self.store.learn( "user-1", self.make_source("third.webp", "third"), {"scenes": ["thinking"], "description": "第三张"}, ) with patch("sticker_store.random.choice", side_effect=lambda items: items[0]) as choice: selected = self.store.random_for_scene("user-1", "happy") candidates = choice.call_args.args[0] self.assertEqual(selected["id"], candidates[0]["id"]) self.assertEqual({item["id"] for item in candidates}, {first["id"], second["id"]}) with patch("sticker_store.random.choice", side_effect=lambda items: items[-1]) as choice: selected = self.store.random_for_scene("user-1", "happy") self.assertEqual(selected["id"], choice.call_args.args[0][-1]["id"]) self.assertIsNone(self.store.random_for_scene("user-1", "angry")) def test_legacy_scopes_merge_into_shared_global_library(self): first = self.store.learn( 1001, self.make_source("first.webp", "first"), {"scenes": ["happy"], "description": "私聊学到"}, ) second = self.store.learn( 2002, self.make_source("second.webp", "second"), {"scenes": ["sad"], "description": "群聊学到"}, ) migrated = self.store.migrate_all_to_global() self.assertEqual(migrated, 2) self.assertIsNotNone(self.store.random_for_scene("global", "happy")) self.assertIsNotNone(self.store.random_for_scene("global", "sad")) self.assertIn("私聊学到", self.store.prompt_summary("global")) self.assertIn("群聊学到", self.store.prompt_summary("global")) self.assertFalse((self.base / "store" / "1001").exists()) self.assertTrue((self.base / "store" / ".migrated" / "1001").is_dir()) self.assertEqual(self.store.migrate_all_to_global(), 0) self.assertNotEqual(first["id"], second["id"]) def test_failed_record_prevents_scope_archive_until_retry_succeeds(self): valid = self.store.learn( 1001, self.make_source("valid.webp", "valid"), {"scenes": ["happy"], "description": "可迁移"}, ) broken = self.store.learn( 1001, self.make_source("broken.webp", "broken"), {"scenes": ["sad"], "description": "源文件丢失"}, ) broken_path = Path(broken["path"]) broken_path.unlink() self.assertEqual(self.store.migrate_all_to_global(), 1) self.assertTrue((self.base / "store" / "1001").is_dir()) self.assertFalse((self.base / "store" / ".migrated" / "1001").exists()) self.assertIsNotNone(self.store.random_for_scene("global", "happy")) self.assertIsNone(self.store.random_for_scene("global", "sad")) broken_path.write_bytes(b"repaired") self.assertEqual(self.store.migrate_all_to_global(), 2) self.assertFalse((self.base / "store" / "1001").exists()) self.assertTrue((self.base / "store" / ".migrated" / "1001").is_dir()) self.assertIsNotNone(self.store.random_for_scene("global", "sad")) self.assertEqual(valid["file_unique_id"], "valid") def test_duplicate_file_unique_id_merges_without_copying(self): first = self.store.learn( "user-1", self.make_source("original.webp", "same-id", b"original"), {"scenes": ["happy"], "description": "旧描述"}, ) second = self.store.learn( "user-1", self.make_source("replacement.webp", "same-id", b"replacement"), {"scenes": ["sad", "happy"], "description": "新描述"}, ) self.assertEqual(second["id"], first["id"]) self.assertEqual(second["scenes"], ["happy", "sad"]) self.assertEqual(second["description"], "新描述") self.assertEqual(second["created_at"], first["created_at"]) self.assertEqual(Path(second["path"]).read_bytes(), b"original") self.assertEqual(Path(second["preview_path"]).read_bytes(), b"preview-original") self.assertEqual(len(list((self.base / "store" / "user-1" / "objects").iterdir())), 1) self.assertEqual(len(list((self.base / "store" / "user-1" / "previews").iterdir())), 1) def test_scopes_have_independent_inventory_and_deduplication(self): private = self.store.learn( "private-uid", self.make_source(), {"scenes": ["happy"], "description": "私聊贴纸"}, ) group = self.store.learn( "group-id", self.make_source("group.webp", body=b"group"), {"scenes": ["sad"], "description": "群聊贴纸"}, ) self.assertNotEqual(private["id"], group["id"]) self.assertEqual(self.store.get("private-uid", private["id"])["description"], "私聊贴纸") self.assertIsNone(self.store.get("private-uid", group["id"])) self.assertIsNone(self.store.get("group-id", private["id"])) self.assertIsNone(self.store.random_for_scene("private-uid", "sad")) self.assertEqual(self.store.random_for_scene("group-id", "sad")["id"], group["id"]) self.assertTrue((self.base / "store" / "private-uid" / "index.json").is_file()) self.assertTrue((self.base / "store" / "group-id" / "index.json").is_file()) def test_prompt_summary_contains_only_scene_counts_and_briefs(self): first = self.store.learn( "user-1", self.make_source("first.webp", "first"), {"scenes": ["happy"], "description": "开心鼓掌"}, ) second = self.store.learn( "user-1", self.make_source("second.webp", "second"), {"scenes": ["happy", "comfort"], "description": "温柔拥抱"}, ) summary = self.store.prompt_summary("user-1") self.assertIn("happy", summary) self.assertIn("开心、愉快或表达喜悦", summary) self.assertIn("2 个", summary) self.assertIn("comfort", summary) self.assertIn("1 个", summary) self.assertIn("开心鼓掌", summary) self.assertIn("温柔拥抱", summary) for forbidden in ( first["id"], second["id"], first["file_unique_id"], first["path"], first["preview_path"], "objects/", "previews/", ): self.assertNotIn(forbidden, summary) def test_scope_cannot_escape_root(self): source = self.make_source() for scope in ("../outside", "/tmp/outside", "a/b", "a\\b"): with self.subTest(scope=scope), self.assertRaises(ValueError): self.store.learn(scope, source, {"scenes": ["happy"]}) self.assertFalse((self.base / "outside").exists()) if __name__ == "__main__": unittest.main()