RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
125 lines
5.3 KiB
Python
125 lines
5.3 KiB
Python
"""Integration tests for learning and sending user-owned stickers."""
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
BASE = Path(__file__).resolve().parents[1]
|
|
sys.path[:0] = [str(BASE / "bot-utility"), str(BASE)]
|
|
|
|
import main
|
|
from chat import engine
|
|
|
|
|
|
class FakeStore:
|
|
def __init__(self, item=None):
|
|
self.item = item
|
|
self.learned = []
|
|
self.random_calls = []
|
|
def learn(self, scope, source, classification):
|
|
self.learned.append((scope, source, classification))
|
|
return {"id": "learned"}
|
|
def random_for_scene(self, scope, scene):
|
|
self.random_calls.append((scope, scene))
|
|
return self.item
|
|
|
|
|
|
class StickerLearningTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_only_marked_stickers_are_learned(self):
|
|
store = FakeStore()
|
|
images = [
|
|
{"kind": "sticker", "source_id": "s1", "original_path": "/tmp/a.webp"},
|
|
{"kind": "photo", "source_id": "s2", "original_path": "/tmp/b.jpg"},
|
|
]
|
|
parsed = {"learned_stickers": [
|
|
{"source_id": "s1", "accept": True, "scenes": ["sad"], "description": "委屈"},
|
|
{"source_id": "s2", "accept": True, "scenes": ["happy"], "description": "普通照片"},
|
|
]}
|
|
with patch.object(engine, "STICKER_STORE", store), \
|
|
patch.object(engine, "stickers_enabled", return_value=True):
|
|
engine._learn_stickers(1001, images, parsed)
|
|
self.assertEqual(len(store.learned), 1)
|
|
self.assertEqual(store.learned[0][0], engine.STICKER_GLOBAL_SCOPE)
|
|
self.assertEqual(store.learned[0][2]["scenes"], ["sad"])
|
|
|
|
async def test_source_marker_preserves_user_instruction(self):
|
|
text = engine._with_sticker_sources(
|
|
"这个适合什么时候用",
|
|
[{"kind": "sticker", "source_id": "s1"}, {"kind": "photo"}],
|
|
)
|
|
self.assertIn("source_id=s1", text)
|
|
self.assertTrue(text.endswith("这个适合什么时候用"))
|
|
|
|
async def test_random_candidate_requires_visual_confirmation(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
original = Path(directory, "a.webp")
|
|
preview = Path(directory, "a.png")
|
|
original.write_bytes(b"sticker")
|
|
preview.write_bytes(b"preview")
|
|
item = {"id": "a", "path": str(original), "preview_path": str(preview)}
|
|
store = FakeStore(item)
|
|
bot = SimpleNamespace(send_sticker=AsyncMock())
|
|
channel = SimpleNamespace(bot=bot, _to_raw_target=lambda raw: raw)
|
|
with patch.object(engine, "STICKER_STORE", store), \
|
|
patch.object(engine, "stickers_enabled", return_value=True), \
|
|
patch.object(engine, "_resolve_delivery", return_value=(channel, 10001)), \
|
|
patch.object(engine, "_confirm_sticker", new=AsyncMock(return_value=False)):
|
|
sent = await engine._maybe_send_sticker(channel, 1, "sad", "context")
|
|
self.assertFalse(sent)
|
|
self.assertEqual(store.random_calls, [(engine.STICKER_GLOBAL_SCOPE, "sad")])
|
|
bot.send_sticker.assert_not_awaited()
|
|
|
|
async def test_confirmation_uses_candidate_preview(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
preview = Path(directory, "preview.png")
|
|
preview.write_bytes(b"preview")
|
|
model = AsyncMock(return_value='{"send":true}')
|
|
with patch.object(engine, "call_deepseek", new=model):
|
|
approved = await engine._confirm_sticker(
|
|
{"preview_path": str(preview), "description": "轻微委屈"},
|
|
"sad", "用户有点失落",
|
|
)
|
|
self.assertTrue(approved)
|
|
self.assertEqual(model.await_args.kwargs["image_infos"][0]["url"], str(preview))
|
|
|
|
|
|
class TelegramStickerInputTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_tg_sticker_is_downloaded_and_forwarded_as_sticker(self):
|
|
sticker = SimpleNamespace(
|
|
is_animated=False,
|
|
is_video=False,
|
|
file_unique_id="unique-1",
|
|
emoji="🙂",
|
|
set_name="set-1",
|
|
)
|
|
message = SimpleNamespace(
|
|
from_user=SimpleNamespace(id=123, is_bot=False),
|
|
chat=SimpleNamespace(id=123, type="private"),
|
|
text=None,
|
|
caption=None,
|
|
photo=[],
|
|
sticker=sticker,
|
|
entities=[],
|
|
)
|
|
async def download(obj, destination):
|
|
Path(destination).write_bytes(b"sticker")
|
|
bot = SimpleNamespace(download=download, id=999)
|
|
dispatch = AsyncMock()
|
|
with patch.object(main, "handle_message", new=dispatch):
|
|
await main.make_tg_handler(SimpleNamespace(tg_username="bot"))(message, bot)
|
|
images = dispatch.await_args.args[5]
|
|
self.assertEqual(len(images), 1)
|
|
self.assertEqual(images[0]["kind"], "sticker")
|
|
self.assertEqual(images[0]["source_id"], "s1")
|
|
self.assertEqual(images[0]["file_unique_id"], "unique-1")
|
|
self.assertEqual(images[0]["format"], "static")
|
|
self.assertTrue(Path(images[0]["original_path"]).exists())
|
|
Path(images[0]["original_path"]).unlink()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|