RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
96 lines
4.3 KiB
Python
96 lines
4.3 KiB
Python
"""Komari notifications and natural-style prompt integration tests."""
|
|
|
|
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 for_admin_komari_agent as komari
|
|
import help_ui
|
|
import notification_render
|
|
import notifications
|
|
from chat import engine
|
|
|
|
|
|
class KomariNotificationTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_komari_alert_enters_structured_notification_pipeline(self):
|
|
with patch("identity.admin_uids", return_value=[1]), \
|
|
patch("identity.push_targets", return_value=[("tg", 10001)]), \
|
|
patch.object(notifications.center, "publish", return_value="event-1") as publish:
|
|
accepted = await komari._notify_admins(
|
|
"[Komari] 服务器「Ama-10」当前处于离线状态。",
|
|
node_uuid="node-1",
|
|
server="Ama-10",
|
|
status="offline",
|
|
)
|
|
self.assertTrue(accepted)
|
|
event = publish.call_args.args[0]
|
|
self.assertEqual(event.kind, "komari")
|
|
self.assertEqual(event.uid, 1)
|
|
self.assertEqual(event.payload["server"], "Ama-10")
|
|
self.assertEqual(event.payload["status"], "offline")
|
|
self.assertIn("当前处于离线状态", event.legacy_text)
|
|
|
|
async def test_komari_alert_without_targets_does_not_advance(self):
|
|
with patch("identity.admin_uids", return_value=[1]), \
|
|
patch("identity.push_targets", return_value=[]), \
|
|
patch.object(notifications.center, "publish") as publish:
|
|
accepted = await komari._notify_admins(
|
|
"offline", node_uuid="node-1", server="Ama-10", status="offline"
|
|
)
|
|
self.assertFalse(accepted)
|
|
publish.assert_not_called()
|
|
|
|
async def test_chat_renderer_can_naturalize_komari_event(self):
|
|
event = notifications.Event(
|
|
kind="komari",
|
|
dedupe_key="komari:test",
|
|
legacy_text="[Komari] 服务器「Ama-10」当前处于离线状态。",
|
|
uid=1,
|
|
payload={"server": "Ama-10", "status": "offline", "source": "Komari"},
|
|
)
|
|
with patch.object(engine, "call_deepseek", new=AsyncMock(return_value="Ama-10 掉线了。")) as model:
|
|
rendered = await notification_render.render_event(event, "chat", 10001)
|
|
self.assertEqual(rendered, "Ama-10 掉线了。")
|
|
self.assertIn("自然表达补充", model.await_args.args[0])
|
|
|
|
|
|
class StylePromptTests(unittest.IsolatedAsyncioTestCase):
|
|
def test_base_prompt_includes_natural_style_supplement(self):
|
|
self.assertIn("自然表达补充", engine.BASE_PROMPT)
|
|
self.assertIn("话不算多，先把", engine.BASE_PROMPT)
|
|
self.assertIn("我先跟你说一声", engine.BASE_PROMPT)
|
|
self.assertIn("不过还不能", engine.BASE_PROMPT)
|
|
self.assertIn("已知且与当前回答直接相关的客观限制", engine.BASE_PROMPT)
|
|
self.assertIn("不影响 JSON", engine.BASE_PROMPT)
|
|
|
|
def test_active_prompt_has_varied_events_without_waiting_for_user(self):
|
|
prompt = engine.load_instruction("notification_instructions.txt")
|
|
self.assertIn("随机发生的生活感", prompt)
|
|
self.assertIn("不是从不出门", prompt)
|
|
self.assertIn("生活只围着用户转", prompt)
|
|
self.assertIn("不必每条都关心用户", prompt)
|
|
|
|
async def test_help_no_longer_requires_self_introduction(self):
|
|
with patch.object(engine, "call_deepseek", new=AsyncMock(return_value="能做的事如下")) as model:
|
|
result = await help_ui.chat_help(1)
|
|
self.assertEqual(result, "能做的事如下")
|
|
prompt = model.await_args.args[0]
|
|
self.assertIn("不必自我介绍", prompt)
|
|
self.assertNotIn("必须先简短介绍自己", prompt)
|
|
|
|
async def test_help_fallback_also_avoids_self_introduction_and_preview(self):
|
|
with patch.object(engine, "call_deepseek", new=AsyncMock(return_value="")):
|
|
result = await help_ui.chat_help(1)
|
|
self.assertNotIn("我是白花", result)
|
|
self.assertNotIn("话不算多", result)
|
|
self.assertNotIn("先理清顺序", result)
|
|
self.assertIn("直接说你想做什么就行", result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|