71 lines
2.6 KiB
Python
71 lines
2.6 KiB
Python
import os
|
|
import unittest
|
|
from unittest.mock import AsyncMock
|
|
|
|
from core.websocket_client_v2 import QingjianAPIClient
|
|
|
|
|
|
class SystemInquiryRulesTest(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
self.rules = {
|
|
"enabled": True,
|
|
"default_action": "silent",
|
|
"default_reply": "已收到",
|
|
"sender_keywords": ["系统客服", "官方客服"],
|
|
"message_keywords": ["系统询单", "代客咨询"],
|
|
"shops": {
|
|
"shop_reply": {
|
|
"enabled": True,
|
|
"action": "reply",
|
|
"reply": "店铺回复模板",
|
|
"sender_keywords": ["机器人客服"],
|
|
"message_keywords": ["询单"],
|
|
}
|
|
},
|
|
}
|
|
os.environ["SYSTEM_INQUIRY_ENABLED"] = "true"
|
|
os.environ["SYSTEM_INQUIRY_SHOPS"] = ""
|
|
|
|
async def test_detect_by_sender_keyword(self):
|
|
client = QingjianAPIClient(enable_agent=False)
|
|
client._system_inquiry_rules = self.rules
|
|
policy = client._resolve_system_inquiry_policy("shop_a")
|
|
data = {"acc_id": "shop_a", "from_name": "平台系统客服", "from_id": "kefu001", "msg": "你好"}
|
|
self.assertTrue(client._match_system_inquiry(data, policy))
|
|
|
|
async def test_shop_rule_reply_action(self):
|
|
client = QingjianAPIClient(enable_agent=False)
|
|
client._system_inquiry_rules = self.rules
|
|
client.send_reply = AsyncMock()
|
|
client.transfer_to_human = AsyncMock()
|
|
|
|
data = {
|
|
"acc_id": "shop_reply",
|
|
"from_name": "机器人客服A",
|
|
"from_id": "robot_01",
|
|
"msg": "有个询单请处理",
|
|
"acc_type": "AliWorkbench",
|
|
}
|
|
handled = await client._handle_system_inquiry(data)
|
|
self.assertTrue(handled)
|
|
client.send_reply.assert_awaited_once()
|
|
client.transfer_to_human.assert_not_awaited()
|
|
|
|
async def test_shop_whitelist_blocks_other_shops(self):
|
|
os.environ["SYSTEM_INQUIRY_SHOPS"] = "shop_only"
|
|
client = QingjianAPIClient(enable_agent=False)
|
|
client._system_inquiry_rules = self.rules
|
|
client.send_reply = AsyncMock()
|
|
data = {"acc_id": "shop_other", "from_name": "系统客服", "from_id": "sys_1", "msg": "系统询单"}
|
|
handled = await client._handle_system_inquiry(data)
|
|
self.assertFalse(handled)
|
|
client.send_reply.assert_not_awaited()
|
|
|
|
def tearDown(self):
|
|
for k in ("SYSTEM_INQUIRY_ENABLED", "SYSTEM_INQUIRY_SHOPS"):
|
|
os.environ.pop(k, None)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|