"""Per-user persona overlays must be shared by chat, help, and notifications.""" from pathlib import Path import sys import unittest from unittest.mock import AsyncMock, patch BASE = Path(__file__).resolve().parents[1] sys.path[:0] = [str(BASE / "bot-utility"), str(BASE)] import help_ui import notification_render import notifications from chat import engine class UserPersonaTests(unittest.IsolatedAsyncioTestCase): def test_overlay_uses_long_term_preferences_with_latest_priority(self): rows = [ (2, "称呼用户为主人", 5, "个人设定,表达偏好", 2, 0), (1, "用户不喜欢书面腔", 4, "表达偏好", 1, 0), ] with patch.object(engine.memory_store, "list_facts", return_value=rows) as facts: prompt = engine.user_persona_prompt(42) facts.assert_called_once_with(42, limit=30) self.assertIn("当前用户的个人设定", prompt) self.assertIn("称呼用户为主人", prompt) self.assertIn("最新、最明确", prompt) self.assertTrue(prompt.startswith(engine.BASE_PROMPT)) self.assertIs(engine.user_persona_prompt(None), engine.BASE_PROMPT) def test_private_chat_prompt_uses_same_user_overlay(self): custom = engine.BASE_PROMPT + "\n【当前用户的个人设定】称呼用户为主人" with patch.object(engine, "user_persona_prompt", return_value=custom) as persona, \ patch.object(engine, "load_user_status", return_value={"pause": False, "sleep": False}), \ patch.object(engine.memory_store, "get_relation", return_value=None), \ patch.object(engine.memory_store, "list_entities", return_value=[]), \ patch.object(engine.memory_store, "recall_facts", return_value=[]), \ patch.object(engine.memory_store, "migrate_legacy"), \ patch.object(engine, "get_history_within_budget", return_value=[]), \ patch.object(engine.notebook.notebook, "context_for_chat", return_value=""), \ patch.object(engine, "_sticker_prompt_summary", return_value=""): prompt = engine.build_system_prompt(42, query="你好") persona.assert_called_once_with(42) self.assertTrue(prompt.startswith(custom)) async def test_notification_renderer_uses_same_user_overlay(self): event = notifications.Event( kind="komari", dedupe_key="persona-test", uid=42, legacy_text="raw", payload={"server": "Ama-10", "status": "offline"}, ) custom = engine.BASE_PROMPT + "\n【当前用户的个人设定】称呼用户为主人" with patch.object(engine, "user_persona_prompt", return_value=custom) as persona, \ patch.object(engine, "call_deepseek", new=AsyncMock(return_value="主人,Ama-10 掉线了。")) as model: rendered = await notification_render.render_event(event, "chat", 10001) persona.assert_called_once_with(42) self.assertTrue(model.await_args.args[0].startswith(custom)) self.assertEqual(rendered, "主人,Ama-10 掉线了。") async def test_help_renderer_uses_same_user_overlay(self): custom = engine.BASE_PROMPT + "\n【当前用户的个人设定】称呼用户为主人" with patch.object(engine, "user_persona_prompt", return_value=custom) as persona, \ patch.object(engine, "call_deepseek", new=AsyncMock(return_value="主人,可以用这些功能。")): await help_ui.chat_help(42) persona.assert_called_once_with(42) def test_deterministic_notification_fallback_has_no_hardcoded_address(self): mail = notifications.Event( kind="mail", dedupe_key="mail-persona", uid=42, legacy_text="raw", payload={"mid": 1, "subject": "test", "sender": "a@example.test"}, ) ginkgo = notifications.Event( kind="ginkgo:course", dedupe_key="ginkgo-persona", uid=42, legacy_text="raw", ) self.assertNotIn("主人", notification_render._fallback(mail)) self.assertNotIn("主人", notification_render._fallback(ginkgo)) def test_memory_instruction_preserves_expression_preferences(self): prompt = engine.load_instruction("agent_instructions.txt") self.assertIn("个人设定", prompt) self.assertIn("表达偏好", prompt) self.assertIn("importance=5", prompt) if __name__ == "__main__": unittest.main()