119 lines
3.7 KiB
Python
119 lines
3.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""WeCom-to-Grok context and send-commit regression tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
import ai_chat
|
|
import ai_config
|
|
import wechat_bot
|
|
from wechat_bot import AUTO_REPLY_TEXT, WeChatBot
|
|
|
|
|
|
class WeChatGrokContextTest(unittest.TestCase):
|
|
@staticmethod
|
|
def _bot() -> WeChatBot:
|
|
bot = WeChatBot.__new__(WeChatBot)
|
|
bot.get_session_history = mock.Mock(
|
|
return_value=[
|
|
{"role": "user", "content": "之前的问题"},
|
|
{"role": "assistant", "content": "之前实际发出的回复"},
|
|
]
|
|
)
|
|
bot.extract_context_for = mock.Mock(return_value="客户的新消息")
|
|
bot.remember_exchange = mock.Mock()
|
|
return bot
|
|
|
|
def test_generate_uses_stable_scope_but_does_not_write_before_send(self) -> None:
|
|
bot = self._bot()
|
|
fp = bytes.fromhex("11" * 16)
|
|
with (
|
|
mock.patch.multiple(
|
|
ai_config,
|
|
AI_ENABLED=True,
|
|
AI_USE_VISION=False,
|
|
AI_CONTEXT_ENABLED=True,
|
|
),
|
|
mock.patch.object(wechat_bot.time, "sleep"),
|
|
mock.patch.object(
|
|
ai_chat,
|
|
"call_ai_text",
|
|
return_value="Grok 生成的合规回复",
|
|
) as call,
|
|
):
|
|
reply = bot._generate_ai_reply(fp, chat_text="当前屏幕")
|
|
|
|
self.assertEqual("Grok 生成的合规回复", reply)
|
|
self.assertEqual(fp.hex(), call.call_args.kwargs["session_id"])
|
|
self.assertEqual(
|
|
bot.get_session_history.return_value,
|
|
call.call_args.kwargs["history"],
|
|
)
|
|
bot.remember_exchange.assert_not_called()
|
|
|
|
bot._commit_generated_exchange(fp, reply, sent=True)
|
|
bot.remember_exchange.assert_called_once_with(
|
|
fp,
|
|
"客户的新消息",
|
|
"Grok 生成的合规回复",
|
|
)
|
|
|
|
def test_failed_wecom_send_never_records_assistant_reply(self) -> None:
|
|
bot = self._bot()
|
|
fp = bytes.fromhex("22" * 16)
|
|
with (
|
|
mock.patch.multiple(
|
|
ai_config,
|
|
AI_ENABLED=True,
|
|
AI_USE_VISION=False,
|
|
AI_CONTEXT_ENABLED=True,
|
|
),
|
|
mock.patch.object(wechat_bot.time, "sleep"),
|
|
mock.patch.object(
|
|
ai_chat,
|
|
"call_ai_text",
|
|
return_value="尚未真正发送的回复",
|
|
),
|
|
):
|
|
reply = bot._generate_ai_reply(fp, chat_text="当前屏幕")
|
|
|
|
bot._commit_generated_exchange(fp, reply, sent=False)
|
|
bot.remember_exchange.assert_not_called()
|
|
|
|
def test_blocked_agent_claim_falls_back_without_persisting_claim(self) -> None:
|
|
bot = self._bot()
|
|
fp = bytes.fromhex("33" * 16)
|
|
with (
|
|
mock.patch.multiple(
|
|
ai_config,
|
|
AI_ENABLED=True,
|
|
AI_USE_VISION=False,
|
|
AI_CONTEXT_ENABLED=True,
|
|
),
|
|
mock.patch.object(wechat_bot.time, "sleep"),
|
|
mock.patch.object(
|
|
ai_chat,
|
|
"call_ai_text",
|
|
return_value="已经帮您预约成功了。",
|
|
),
|
|
):
|
|
reply = bot._generate_ai_reply(fp, chat_text="当前屏幕")
|
|
|
|
self.assertIsNone(reply)
|
|
bot._commit_generated_exchange(fp, reply, sent=True)
|
|
bot.remember_exchange.assert_called_once_with(
|
|
fp,
|
|
"客户的新消息",
|
|
AUTO_REPLY_TEXT,
|
|
)
|
|
self.assertNotIn(
|
|
"预约成功",
|
|
bot.remember_exchange.call_args.args[2],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|