""" 空运 BOT 出站：没有当场回话地址时，打桥主动往群里说。 """ from __future__ import annotations import os import sys import unittest from unittest.mock import MagicMock, patch ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) if ROOT not in sys.path: sys.path.insert(0, ROOT) from agent.channel.aibot import normalize_aibot_payload from agent.channel.aibot.reply import AibotReplyClient class AibotInboundNormalizeTests(unittest.TestCase): def test_official_wecom_body(self) -> None: parsed = normalize_aibot_payload( { "msgid": "m1", "chatid": "wr_air_1", "chattype": "group", "from": {"userid": "WuJiLin"}, "msgtype": "text", "text": {"content": "@询价小助手 WO202609160009"}, "aibot_direct": True, } ) self.assertEqual(parsed.sender_id, "WuJiLin") self.assertEqual(parsed.message_id, "m1") self.assertEqual(parsed.chat_id, "wr_air_1") self.assertEqual(parsed.chat_type, "group") self.assertIn("WO202609160009", parsed.content) self.assertEqual(parsed.raw.get("source"), "aibot") def test_keeps_aibot_req_id_for_quote_reply(self) -> None: parsed = normalize_aibot_payload( { "sender_id": "air1", "message_id": "m2", "content": "锁舱", "chat_id": "wr_air_1", "aibot_req_id": "cb_lock_1", } ) self.assertEqual(parsed.raw.get("aibot_req_id"), "cb_lock_1") class AibotReplyTests(unittest.TestCase): def test_empty_content_skips(self) -> None: client = AibotReplyClient("") out = client.send_group(chat_id="wr_1", content=" ") self.assertFalse(out["ok"]) self.assertEqual(out["error"], "empty") def test_proactive_posts_send_group(self) -> None: client = AibotReplyClient("") fake_http = MagicMock() fake_http.post.return_value.status_code = 200 fake_http.post.return_value.text = '{"status":"sent"}' settings = MagicMock() settings.wecom_aibot_bridge_url = "http://127.0.0.1:8913" settings.wecom_aibot_bridge_token = "tok" with patch("agent.channel.aibot.reply.httpx", fake_http), patch( "agent.channel.aibot.reply.get_settings", return_value=settings ): out = client.send_group(chat_id="wr_air_1", content="附件已保存") self.assertTrue(out["ok"]) url, = fake_http.post.call_args.args self.assertEqual(url, "http://127.0.0.1:8913/send") payload = fake_http.post.call_args.kwargs["json"] self.assertEqual(payload["chatid"], "wr_air_1") self.assertEqual(payload["operation"], "text") self.assertEqual(payload["content"], "附件已保存") self.assertEqual(payload.get("aibot_req_id"), "") self.assertEqual( fake_http.post.call_args.kwargs["headers"]["X-Aibot-Bridge-Token"], "tok", ) def test_send_group_file_posts_send_file(self) -> None: client = AibotReplyClient("") fake_http = MagicMock() fake_http.post.return_value.status_code = 200 fake_http.post.return_value.text = '{"status":"sent"}' settings = MagicMock() settings.wecom_aibot_bridge_url = "http://127.0.0.1:8913" settings.wecom_aibot_bridge_token = "tok" with patch("agent.channel.aibot.reply.httpx", fake_http), patch( "agent.channel.aibot.reply.get_settings", return_value=settings ): out = client.send_group_file( chat_id="wr_air_1", filename="WO1_quote.xlsx", file_b64="QQ==", ) self.assertTrue(out["ok"]) url, = fake_http.post.call_args.args self.assertEqual(url, "http://127.0.0.1:8913/send-file") payload = fake_http.post.call_args.kwargs["json"] self.assertEqual(payload["chatid"], "wr_air_1") self.assertEqual(payload["filename"], "WO1_quote.xlsx") def test_at_reply_does_not_post_stream_payload(self) -> None: """ 企微 stream 回复会自带「内容由AI生成」和赞踩。 @ 当场也只走桥主动发言，不往 response_url 推 stream。 """ client = AibotReplyClient("https://example.com/reply") fake_http = MagicMock() fake_http.post.return_value.status_code = 200 fake_http.post.return_value.text = '{"status":"sent"}' settings = MagicMock() settings.wecom_aibot_bridge_url = "http://127.0.0.1:8913" settings.wecom_aibot_bridge_token = "tok" with patch("agent.channel.aibot.reply.httpx", fake_http), patch( "agent.channel.aibot.reply.get_settings", return_value=settings ): out = client.send_group(chat_id="wr_air_1", content="工单号：WO202609210001") self.assertTrue(out["ok"]) payloads = [c.kwargs.get("json") or {} for c in fake_http.post.call_args_list] self.assertFalse(any(p.get("msgtype") == "stream" for p in payloads)) self.assertFalse( any(isinstance(p.get("stream"), dict) for p in payloads) ) urls = [c.args[0] for c in fake_http.post.call_args_list] self.assertFalse(any("example.com/reply" in str(u) for u in urls)) def test_at_reply_passes_callback_req_id(self) -> None: client = AibotReplyClient("", callback_req_id="cb_lock_1") fake_http = MagicMock() fake_http.post.return_value.status_code = 200 fake_http.post.return_value.text = '{"status":"sent"}' settings = MagicMock() settings.wecom_aibot_bridge_url = "http://127.0.0.1:8913" settings.wecom_aibot_bridge_token = "tok" with patch("agent.channel.aibot.reply.httpx", fake_http), patch( "agent.channel.aibot.reply.get_settings", return_value=settings ): client.send_group(chat_id="wr_air_1", content="工单WO1 空运锁舱失败。") payload = fake_http.post.call_args.kwargs["json"] self.assertEqual(payload["aibot_req_id"], "cb_lock_1") def test_at_reply_puts_brief_on_same_channel_as_file(self) -> None: """ @ 当场流式回话群里经常只看得见后面的文件。 工单信息必须再走桥主动发言，和模板同一条通道。 """ client = AibotReplyClient("https://example.com/reply") fake_http = MagicMock() fake_http.post.return_value.status_code = 200 fake_http.post.return_value.text = '{"status":"sent"}' settings = MagicMock() settings.wecom_aibot_bridge_url = "http://127.0.0.1:8913" settings.wecom_aibot_bridge_token = "tok" brief = "工单号：WO202609160014\n询价字段：\n起运地：ZUH" with patch("agent.channel.aibot.reply.httpx", fake_http), patch( "agent.channel.aibot.reply.get_settings", return_value=settings ): out = client.send_group(chat_id="wr_air_1", content=brief) self.assertTrue(out["ok"]) urls = [c.args[0] for c in fake_http.post.call_args_list] self.assertTrue(any(u.endswith("/send") or u.endswith("/send-group") for u in urls)) payloads = [c.kwargs.get("json") or {} for c in fake_http.post.call_args_list] self.assertTrue(any(p.get("content") == brief for p in payloads))