RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
261 lines
11 KiB
Python
261 lines
11 KiB
Python
"""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()
|