jillion886andCursor
99da1faf30
查到价后把币种、单位、数量、单价、备注写入报价单，避免销售只看到光秃秃的金额。 Co-authored-by: Cursor <cursoragent@cursor.com>
996 lines
38 KiB
Python
996 lines
38 KiB
Python
"""
|
||
空运文字询价主路径单测（不连企微 / 真实 TMS / 8180）。
|
||
|
||
覆盖：无运输方式、缺项补问、建单卡片、有价/无价、跳过协同、采用 Excel/PDF、成交。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
import unittest
|
||
|
||
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
||
if ROOT not in sys.path:
|
||
sys.path.insert(0, ROOT)
|
||
|
||
os.environ.setdefault("YTD_ENV", "test")
|
||
os.environ.setdefault("REDIS_BACKEND", "memory")
|
||
os.environ.setdefault("CHECKPOINT_BACKEND", "memory")
|
||
os.environ.setdefault("MESSAGE_STORE_BACKEND", "memory")
|
||
os.environ.setdefault("LLM_DATA_USAGE_CONFIRMED", "false")
|
||
os.environ.setdefault("LLM_ALLOW_NETWORK", "false")
|
||
os.environ.setdefault("LEDGER_BACKEND", "memory")
|
||
|
||
from agent.channel.queue import MemoryMessageStore
|
||
from agent.channel.wecom.models import InboundMessage
|
||
from agent.handlers.text_inquiry import handle_text_inquiry
|
||
from agent.ledger.memory_ledger import MemoryLedger
|
||
from agent.policy import inquiry_copy as copy
|
||
from agent.policy.air_text_flow import AirTextInquiryFlow
|
||
from agent.schema.field_validate import validate_required_fields
|
||
|
||
|
||
COMPLETE_AIR = {
|
||
"起运港": "上海",
|
||
"目的港": "洛杉矶",
|
||
"品名": "普货",
|
||
"件数": "3",
|
||
"毛重": "180",
|
||
"体积": "2.5",
|
||
"包装方式": "纸箱",
|
||
}
|
||
|
||
|
||
class AirTextInquiryTests(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self.flow = AirTextInquiryFlow(MemoryLedger())
|
||
self.replies: list[str] = []
|
||
self.extras: list[dict] = []
|
||
|
||
def _reply(self, text: str, extra=None) -> None:
|
||
self.replies.append(text)
|
||
self.extras.append(dict(extra) if extra else {})
|
||
|
||
def test_clarify_shows_known_and_examples(self) -> None:
|
||
text = copy.ask_clarify(
|
||
facts={
|
||
"起运港": "珠海",
|
||
"目的港": "克拉克",
|
||
"originCode": "ZUH",
|
||
"destinationCode": "CRK",
|
||
"品名": "茶叶",
|
||
"件数": "10",
|
||
},
|
||
missing_keys=["毛重", "体积", "包装方式"],
|
||
transport_mode="空运",
|
||
)
|
||
self.assertEqual(
|
||
text,
|
||
"\n".join(
|
||
[
|
||
"请确认并补充以下信息：",
|
||
"运输方式：空运",
|
||
"起运地：珠海",
|
||
"目的地：克拉克",
|
||
"品名：茶叶",
|
||
"件数：10件",
|
||
"",
|
||
"待补充：",
|
||
"重量：(如：100KGS）",
|
||
"体积：(如：2CBM）",
|
||
"包装类型：(如：托盘）",
|
||
]
|
||
),
|
||
)
|
||
|
||
def test_inquiry_card_matches_sales_format(self) -> None:
|
||
text = copy.inquiry_card(
|
||
work_order_no="WO202609110080",
|
||
facts={
|
||
"起运港": "珠海",
|
||
"目的港": "克拉克",
|
||
"originCode": "ZUH",
|
||
"destinationCode": "CRK",
|
||
"品名": "茶叶",
|
||
"件数": "10",
|
||
"毛重": "1",
|
||
"体积": "8",
|
||
"包装方式": "托盘",
|
||
"报价日期": "2026-09-12",
|
||
},
|
||
)
|
||
self.assertEqual(
|
||
text,
|
||
"\n".join(
|
||
[
|
||
"工单号：WO202609110080",
|
||
"运输方式：空运",
|
||
"起运地：珠海",
|
||
"目的地：克拉克",
|
||
"品名：茶叶",
|
||
"件数：10件",
|
||
"重量：1",
|
||
"体积：8",
|
||
"包装类型：托盘",
|
||
"报价日期：2026-09-12",
|
||
"工单已创建，正在查询TMS报价。",
|
||
]
|
||
),
|
||
)
|
||
|
||
def test_quote_date_defaults_today_not_in_missing(self) -> None:
|
||
result = validate_required_fields(facts=COMPLETE_AIR, business_line="AIR")
|
||
self.assertTrue(result["ok"])
|
||
self.assertTrue(result["facts"]["报价日期"])
|
||
self.assertNotIn("报价日期", result["missing"])
|
||
|
||
def test_fly_is_air_and_followup_keeps_mode(self) -> None:
|
||
phase = self.flow.on_text(
|
||
sender_id="u_fly",
|
||
text="从南京飞吉隆坡，9月14日茶叶，托盘，4.2cbm，1.5KG",
|
||
reply=self._reply,
|
||
)
|
||
self.assertEqual(phase, "clarify")
|
||
self.assertIn("运输方式：空运", self.replies[-1])
|
||
self.assertNotIn("请补充运输方式", self.replies[-1])
|
||
self.replies.clear()
|
||
phase = self.flow.on_text(
|
||
sender_id="u_fly",
|
||
text="10件",
|
||
reply=self._reply,
|
||
)
|
||
self.assertNotEqual(phase, "need_mode")
|
||
joined = "\n".join(self.replies)
|
||
self.assertNotIn("请补充运输方式", joined)
|
||
self.assertEqual(self.flow.session_of("u_fly").business_line, "AIR")
|
||
|
||
def test_missing_mode_asks_transport(self) -> None:
|
||
phase = self.flow.on_text(
|
||
sender_id="u1",
|
||
text="帮我看看价格",
|
||
reply=self._reply,
|
||
injected_facts={},
|
||
injected_mode="",
|
||
)
|
||
self.assertEqual(phase, "need_mode")
|
||
self.assertIn("运输方式", self.replies[-1])
|
||
|
||
def test_pack_bulk_cargo_accepted_as_user_said(self) -> None:
|
||
facts = dict(COMPLETE_AIR)
|
||
facts["包装方式"] = "散货"
|
||
phase = self.flow.on_text(
|
||
sender_id="u_bulk",
|
||
text="空运",
|
||
reply=self._reply,
|
||
injected_facts=facts,
|
||
injected_mode="AIR",
|
||
)
|
||
self.assertNotEqual(phase, "clarify")
|
||
joined = "\n".join(self.replies)
|
||
self.assertIn("包装类型：散货", joined)
|
||
self.assertIn("工单已创建，正在查询TMS报价", joined)
|
||
self.assertEqual(self.flow.session_of("u_bulk").facts["包装方式"], "散货")
|
||
|
||
def test_oral_kgs_cbm_not_asked_again(self) -> None:
|
||
phase = self.flow.on_text(
|
||
sender_id="u_kgs",
|
||
text="空运 茶叶 10件 100KGS 2CBM",
|
||
reply=self._reply,
|
||
injected_facts={
|
||
"起运港": "珠海",
|
||
"目的港": "克拉克",
|
||
"品名": "茶叶",
|
||
"包装方式": "托盘",
|
||
},
|
||
injected_mode="AIR",
|
||
)
|
||
self.assertNotEqual(phase, "clarify")
|
||
sess = self.flow.session_of("u_kgs")
|
||
self.assertEqual(sess.facts["毛重"], "100KGS")
|
||
self.assertEqual(sess.facts["体积"], "2CBM")
|
||
self.assertEqual(sess.facts["originCode"], "ZUH")
|
||
self.assertEqual(sess.facts["destinationCode"], "CRK")
|
||
self.assertEqual(sess.facts["packageType"], "2")
|
||
|
||
def test_missing_fields_prompt(self) -> None:
|
||
phase = self.flow.on_text(
|
||
sender_id="u1",
|
||
text="空运询价",
|
||
reply=self._reply,
|
||
injected_facts={"起运港": "上海"},
|
||
injected_mode="AIR",
|
||
)
|
||
self.assertEqual(phase, "clarify")
|
||
last = self.replies[-1]
|
||
self.assertIn("请确认并补充以下信息", last)
|
||
self.assertIn("运输方式：空运", last)
|
||
self.assertIn("起运地：上海", last)
|
||
self.assertIn("待补充：", last)
|
||
self.assertIn("目的地", last)
|
||
self.assertIn("件数", last)
|
||
self.assertIn("如：", last)
|
||
self.assertIn("报价日期：", last)
|
||
self.assertNotIn("报价日期：(如", last)
|
||
|
||
def test_complete_air_hit_then_skip_adopt_excel_deal(self) -> None:
|
||
phase = self.flow.on_text(
|
||
sender_id="u1",
|
||
text="空运完整",
|
||
reply=self._reply,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
self.assertEqual(phase, "wait_collab")
|
||
joined = "\n".join(self.replies)
|
||
self.assertIn("工单号：", joined)
|
||
self.assertIn("运输方式：空运", joined)
|
||
self.assertIn("起运地：上海", joined)
|
||
self.assertIn("目的地：洛杉矶", joined)
|
||
self.assertIn("重量：", joined)
|
||
self.assertIn("正在查询TMS报价", joined)
|
||
self.assertNotIn("【询价卡片】", joined)
|
||
self.assertNotIn("[test]", joined)
|
||
self.assertIn("已报价", self.flow.session_of("u1").status)
|
||
self.assertIn(copy.BTN_SKIP_COLLAB, joined)
|
||
self.assertIn("价格明细", joined)
|
||
self.assertIn("总销售价合计", joined)
|
||
self.assertIn("时效：", joined)
|
||
self.assertIn("报价有效期：", joined)
|
||
self.assertIn("相同询价内容检索：首次询价", joined)
|
||
self.assertNotIn("工单状态：已报价", joined)
|
||
self.assertNotIn("【价格明细】", joined)
|
||
|
||
extras: list[dict] = []
|
||
|
||
def _reply_keep(text: str, extra=None) -> None:
|
||
self.replies.append(text)
|
||
extras.append(dict(extra or {}))
|
||
|
||
phase = self.flow.on_button(
|
||
sender_id="u1",
|
||
action=copy.BTN_SKIP_COLLAB,
|
||
reply=_reply_keep,
|
||
card_meta={"response_code": "rc-test-1"},
|
||
)
|
||
self.assertEqual(phase, "wait_adopt")
|
||
self.assertIn("已跳过内部协同", "\n".join(self.replies))
|
||
self.assertTrue(
|
||
any(x.get("msgtype") == "update_template_card" for x in extras)
|
||
)
|
||
wo = self.flow.session_of("u1").work_order_no
|
||
scheme_cards = [
|
||
x
|
||
for x in extras
|
||
if (x.get("template_card") or {}).get("main_title", {}).get("title")
|
||
== f"{wo}报价方案"
|
||
]
|
||
self.assertEqual(len(scheme_cards), 1)
|
||
card = scheme_cards[0]["template_card"]
|
||
self.assertIn("空运费：USD 1,200", card["quote_area"]["quote_text"])
|
||
self.assertEqual(len(card["button_list"]), 3)
|
||
self.assertEqual(
|
||
[x["text"] for x in card["button_list"]],
|
||
[copy.BTN_ADOPT_EXCEL_CARD, copy.BTN_ADOPT_PDF_CARD, copy.BTN_REJECT],
|
||
)
|
||
rows = {x["keyname"]: x["value"] for x in card["horizontal_content_list"]}
|
||
self.assertIn("2.5", rows["货物"])
|
||
self.assertEqual(rows["时效"], "3-5 天")
|
||
self.assertNotIn("sub_title_text", card)
|
||
n = len(self.replies)
|
||
again = self.flow.on_button(
|
||
sender_id="u1", action=copy.BTN_SKIP_COLLAB, reply=self._reply
|
||
)
|
||
self.assertEqual(again, "card_discard")
|
||
self.assertEqual(len(self.replies), n)
|
||
self.assertIn(copy.BTN_ADOPT_EXCEL_CARD, "\n".join(self.replies))
|
||
|
||
phase = self.flow.on_button(
|
||
sender_id="u1", action=copy.BTN_ADOPT_EXCEL, reply=self._reply
|
||
)
|
||
self.assertEqual(phase, "wait_deal")
|
||
self.assertIn("Excel", "\n".join(self.replies))
|
||
self.assertIn("咨询成交状态", "\n".join(self.replies))
|
||
|
||
phase = self.flow.on_button(sender_id="u1", action=copy.BTN_DEAL, reply=self._reply)
|
||
self.assertEqual(phase, "done")
|
||
ticket = self.flow.ledger.get_ticket(
|
||
work_order_no=self.flow.session_of("u1").work_order_no
|
||
)
|
||
self.assertEqual(ticket.status, "已成交")
|
||
|
||
def test_city_name_counts_as_origin_dest(self) -> None:
|
||
"""起运地/目的地说了地名就算齐，不对销售再要三字码。"""
|
||
facts = {
|
||
"起运港": "南京",
|
||
"目的港": "吉隆坡",
|
||
"品名": "茶叶",
|
||
"件数": "10",
|
||
"毛重": "1.5",
|
||
"体积": "4.2",
|
||
"包装方式": "散货",
|
||
"报价日期": "2026-09-16",
|
||
}
|
||
phase = self.flow.on_text(
|
||
sender_id="u_nkg",
|
||
text="南京飞吉隆坡，9月16日茶叶，散货，4.2cbm，1.5KG，10件",
|
||
reply=self._reply,
|
||
injected_facts=facts,
|
||
injected_mode="AIR",
|
||
)
|
||
self.assertNotEqual(phase, "clarify")
|
||
joined = "\n".join(self.replies)
|
||
self.assertIn("起运地：南京", joined)
|
||
self.assertIn("目的地：吉隆坡", joined)
|
||
self.assertIn("重量：", joined)
|
||
self.assertIn("体积：", joined)
|
||
self.assertIn("包装类型：散货", joined)
|
||
self.assertIn("报价日期：2026-09-16", joined)
|
||
self.assertNotIn("重量(KG)", joined)
|
||
self.assertNotIn("体积(CBM)", joined)
|
||
self.assertNotIn("机场名或三字码", joined)
|
||
|
||
def test_unknown_airport_still_queries_tms(self) -> None:
|
||
"""机场对不上也必须先打 query_tms，不能本地直接说暂无报价。"""
|
||
called: list[dict] = []
|
||
|
||
class _Spy:
|
||
def create_ticket(self, **kwargs):
|
||
return {
|
||
"ok": True,
|
||
"work_order_no": "WO202609140088",
|
||
"first_or_same": "first",
|
||
}
|
||
|
||
def query_tms(self, **kwargs):
|
||
called.append(dict(kwargs))
|
||
return {
|
||
"ok": True,
|
||
"has_price": False,
|
||
"classification": "TMS_NO_QUOTE_1002",
|
||
}
|
||
|
||
facts = dict(COMPLETE_AIR)
|
||
facts["起运港"] = "阿拉伯"
|
||
facts["目的港"] = "广州"
|
||
flow = AirTextInquiryFlow(_Spy())
|
||
phase = flow.on_text(
|
||
sender_id="u_arab",
|
||
text="空运",
|
||
reply=self._reply,
|
||
injected_facts=facts,
|
||
injected_mode="AIR",
|
||
)
|
||
self.assertEqual(len(called), 1, "对不上机场也必须去 TMS 查价")
|
||
self.assertEqual(called[0]["work_order_no"], "WO202609140088")
|
||
self.assertEqual(called[0]["facts"].get("起运港"), "阿拉伯")
|
||
self.assertFalse(called[0]["facts"].get("originCode"))
|
||
self.assertEqual(phase, "tms_miss")
|
||
self.assertIn("暂无报价", self.replies[-1])
|
||
self.assertNotIn("查价参数尚未齐全", "\n".join(self.replies))
|
||
miss = [x for x in self.extras if x.get("msgtype") == "template_card"]
|
||
self.assertTrue(miss)
|
||
self.assertEqual(miss[-1]["template_card"]["main_title"]["title"], "WO202609140088暂无报价")
|
||
self.assertEqual(
|
||
miss[-1]["template_card"]["quote_area"]["quote_text"],
|
||
"请务必在群聊发送工单编号WO202609140088",
|
||
)
|
||
|
||
def test_unknown_airport_no_card(self) -> None:
|
||
facts = dict(COMPLETE_AIR)
|
||
facts["目的港"] = "歧义港"
|
||
phase = self.flow.on_text(
|
||
sender_id="u2c",
|
||
text="空运",
|
||
reply=self._reply,
|
||
injected_facts=facts,
|
||
injected_mode="AIR",
|
||
)
|
||
# 目的地说了就算齐；对不上字典不补问，查价后按 TMS 分类
|
||
self.assertEqual(phase, "tms_miss")
|
||
self.assertIn("暂无报价", self.replies[-1])
|
||
|
||
def test_tms_miss_copy(self) -> None:
|
||
facts = dict(COMPLETE_AIR)
|
||
facts["tms_force"] = "1002"
|
||
phase = self.flow.on_text(
|
||
sender_id="u2",
|
||
text="空运",
|
||
reply=self._reply,
|
||
injected_facts=facts,
|
||
injected_mode="AIR",
|
||
)
|
||
self.assertEqual(phase, "tms_miss")
|
||
self.assertIn("暂无报价", self.replies[-1])
|
||
self.assertIn("工单编号", self.replies[-1])
|
||
miss = [x for x in self.extras if x.get("msgtype") == "template_card"]
|
||
self.assertTrue(miss)
|
||
card = miss[-1]["template_card"]
|
||
self.assertEqual(card["card_type"], "text_notice")
|
||
self.assertEqual(card["source"]["desc"], "TMS查价结果")
|
||
self.assertEqual(card["main_title"]["title"], self.flow.session_of("u2").work_order_no + "暂无报价")
|
||
self.assertNotIn("desc", card["main_title"])
|
||
self.assertEqual(card["sub_title_text"], copy.tms_miss_hint())
|
||
self.assertEqual(
|
||
card["quote_area"]["quote_text"],
|
||
copy.tms_miss_quote_text(work_order_no=self.flow.session_of("u2").work_order_no),
|
||
)
|
||
self.assertNotIn("horizontal_content_list", card)
|
||
|
||
def test_incomplete_skip_is_not_no_quote(self) -> None:
|
||
"""主账未出站不能对销售说暂无报价。"""
|
||
|
||
class _Skip:
|
||
def create_ticket(self, **kwargs):
|
||
return {
|
||
"ok": True,
|
||
"work_order_no": "WO202609140099",
|
||
"first_or_same": "first",
|
||
}
|
||
|
||
def query_tms(self, **kwargs):
|
||
return {
|
||
"ok": False,
|
||
"has_price": False,
|
||
"classification": "TMS_QUERY_INCOMPLETE",
|
||
}
|
||
|
||
flow = AirTextInquiryFlow(_Skip())
|
||
phase = flow.on_text(
|
||
sender_id="u_skip",
|
||
text="空运",
|
||
reply=self._reply,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
self.assertEqual(phase, "tms_tech")
|
||
joined = "\n".join(self.replies)
|
||
self.assertIn("TMS 查询失败", joined)
|
||
self.assertNotIn("暂无报价", joined)
|
||
self.assertNotIn("查价参数尚未齐全", joined)
|
||
|
||
def test_same_content_second_ticket(self) -> None:
|
||
self.flow.on_text(
|
||
sender_id="u3",
|
||
text="第一次",
|
||
reply=self._reply,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
no1 = self.flow.session_of("u3").work_order_no
|
||
self.replies.clear()
|
||
self.flow.on_text(
|
||
sender_id="u4",
|
||
text="第二次相同",
|
||
reply=self._reply,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
no2 = self.flow.session_of("u4").work_order_no
|
||
self.assertIn("工单已创建，正在查询TMS报价", "\n".join(self.replies))
|
||
self.assertIn(f"引用历史工单号{no1}", "\n".join(self.replies))
|
||
self.assertTrue(no1)
|
||
self.assertTrue(no2)
|
||
self.assertNotEqual(no1, no2)
|
||
|
||
def test_reject_then_adjust_prompt(self) -> None:
|
||
self.flow.on_text(
|
||
sender_id="u5",
|
||
text="空运",
|
||
reply=self._reply,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
self.flow.on_button(sender_id="u5", action=copy.BTN_SKIP_COLLAB, reply=self._reply)
|
||
phase = self.flow.on_button(sender_id="u5", action=copy.BTN_REJECT, reply=self._reply)
|
||
self.assertEqual(phase, "wait_adjust")
|
||
self.assertIn("报价加 10%", self.replies[-1])
|
||
|
||
def test_adopt_pdf(self) -> None:
|
||
self.flow.on_text(
|
||
sender_id="u6",
|
||
text="空运",
|
||
reply=self._reply,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
self.flow.on_button(sender_id="u6", action=copy.BTN_SKIP_COLLAB, reply=self._reply)
|
||
phase = self.flow.on_button(
|
||
sender_id="u6", action=copy.BTN_ADOPT_PDF, reply=self._reply
|
||
)
|
||
self.assertEqual(phase, "wait_deal")
|
||
self.assertIn("PDF", "\n".join(self.replies))
|
||
|
||
def test_handler_writes_outbox(self) -> None:
|
||
store = MemoryMessageStore()
|
||
msg = InboundMessage(
|
||
sender_id="u7",
|
||
message_id="m7",
|
||
content="空运",
|
||
msg_type="text",
|
||
)
|
||
phase = handle_text_inquiry(
|
||
msg,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
flow=self.flow,
|
||
store=store,
|
||
)
|
||
self.assertEqual(phase, "wait_collab")
|
||
pending = [i for i in store._outbox.values()]
|
||
self.assertGreaterEqual(len(pending), 2)
|
||
self.assertTrue(any("工单已创建，正在查询TMS报价" in i.content for i in pending))
|
||
self.assertFalse(any("[test]" in i.content for i in pending))
|
||
self.assertFalse(any("【询价卡片】" in i.content for i in pending))
|
||
|
||
def test_wrong_button_discarded(self) -> None:
|
||
self.flow.on_text(
|
||
sender_id="u8",
|
||
text="空运",
|
||
reply=self._reply,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
n = len(self.replies)
|
||
phase = self.flow.on_button(
|
||
sender_id="u8", action=copy.BTN_ADOPT_EXCEL, reply=self._reply
|
||
)
|
||
self.assertEqual(phase, "card_discard")
|
||
self.assertEqual(len(self.replies), n)
|
||
|
||
def test_create_fail_keeps_weight_and_replies(self) -> None:
|
||
class _FailCreate:
|
||
def create_ticket(self, **kwargs):
|
||
return {"ok": False, "work_order_no": "", "first_or_same": "first"}
|
||
|
||
flow = AirTextInquiryFlow(_FailCreate())
|
||
replies: list[str] = []
|
||
phase = flow.on_text(
|
||
sender_id="u9",
|
||
text="空运",
|
||
reply=replies.append,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
self.assertEqual(phase, "ledger_fail")
|
||
self.assertTrue(any("建单失败" in x for x in replies))
|
||
self.assertEqual(flow.session_of("u9").facts["毛重"], "180")
|
||
|
||
def test_tms_miss_card_matches_sales_layout(self) -> None:
|
||
payload = copy.tms_miss_wecom_payload(work_order_no="WO202606160008")
|
||
card = payload["template_card"]
|
||
self.assertEqual(card["card_type"], "text_notice")
|
||
self.assertEqual(card["source"]["desc"], "TMS查价结果")
|
||
self.assertEqual(card["main_title"]["title"], "WO202606160008暂无报价")
|
||
self.assertNotIn("desc", card["main_title"])
|
||
self.assertEqual(card["sub_title_text"], "TMS标准价库暂无匹配报价，请手动拉新人与智能体到原有群")
|
||
self.assertEqual(card["quote_area"]["quote_text"], "请务必在群聊发送工单编号WO202606160008")
|
||
self.assertNotIn("title", card["quote_area"])
|
||
self.assertNotIn("horizontal_content_list", card)
|
||
self.assertNotIn("card_action", card)
|
||
with_url = copy.tms_miss_wecom_payload(
|
||
work_order_no="WO202606160008",
|
||
action_url="https://example.test",
|
||
)["template_card"]
|
||
self.assertEqual(with_url["card_action"]["url"], "https://example.test")
|
||
|
||
def test_quote_card_payload_has_clickable_button(self) -> None:
|
||
payload = copy.tms_hit_wecom_payload(
|
||
work_order_no="WO202609140001",
|
||
quote={
|
||
"total": "USD 1,280",
|
||
"eta": "3-5 天",
|
||
"valid_until": "2026-09-20",
|
||
"source_label": "TMS 标准报价",
|
||
"fee_lines": [
|
||
{"name": "空运费", "amount": "USD 1,200"},
|
||
{"name": "其他附加费", "amount": "USD 80"},
|
||
],
|
||
},
|
||
)
|
||
card = payload["template_card"]
|
||
self.assertEqual(payload["msgtype"], "template_card")
|
||
self.assertEqual(card["card_type"], "button_interaction")
|
||
self.assertEqual(card["main_title"]["title"], "WO202609140001价格明细")
|
||
self.assertEqual(card["button_list"][0]["text"], copy.BTN_SKIP_COLLAB)
|
||
self.assertEqual(card["button_list"][0]["key"], "skip_collab:WO202609140001")
|
||
self.assertTrue(str(card["task_id"]).startswith("quote-WO202609140001-"))
|
||
self.assertEqual(
|
||
copy.work_order_from_card_meta({"task_id": card["task_id"]}),
|
||
"WO202609140001",
|
||
)
|
||
self.assertNotIn("horizontal_content_list", card)
|
||
quote_text = card["quote_area"]["quote_text"]
|
||
self.assertNotIn("title", card["quote_area"])
|
||
self.assertIn("空运费：USD 1,200", quote_text)
|
||
self.assertIn("总销售价合计：USD 1,280", quote_text)
|
||
self.assertNotIn("时效：", quote_text)
|
||
numeric = copy.tms_hit_wecom_payload(
|
||
work_order_no="WO202609140048",
|
||
quote={
|
||
"total": "14669.60",
|
||
"currency": "CNY",
|
||
"eta": "-",
|
||
"valid_until": "2026-09-15",
|
||
"fee_lines": [{"name": "空运费", "amount": "14669.60", "currency": "CNY"}],
|
||
},
|
||
)["template_card"]["quote_area"]["quote_text"]
|
||
self.assertIn("空运费：CNY 14669.60", numeric)
|
||
self.assertIn("总销售价合计：CNY 14669.60", numeric)
|
||
self.assertEqual(copy.money_text("14669.60", ""), "14669.60")
|
||
self.assertNotIn(copy.collab_hint("WO202609140001"), quote_text)
|
||
sub = card["sub_title_text"]
|
||
self.assertIn("时效：3-5 天", sub)
|
||
self.assertIn("报价有效期：2026-09-20", sub)
|
||
self.assertIn("相同询价内容检索：首次询价", sub)
|
||
self.assertNotIn("工单状态", sub)
|
||
self.assertTrue(sub.endswith(copy.collab_hint("WO202609140001")))
|
||
self.assertNotIn("内部协同", json.dumps(card, ensure_ascii=False))
|
||
|
||
def test_scheme_card_payload_has_quote_and_buttons(self) -> None:
|
||
quote = {
|
||
"total": "USD 1,280",
|
||
"eta": "3-5 天",
|
||
"valid_until": "2026-09-20",
|
||
"source_label": "TMS 标准报价",
|
||
"fee_lines": [
|
||
{"name": "空运费", "amount": "USD 1,200"},
|
||
{"name": "其他附加费", "amount": "USD 80"},
|
||
],
|
||
}
|
||
text = copy.scheme_card(
|
||
work_order_no="WO202609140001",
|
||
status="已报价",
|
||
facts=COMPLETE_AIR,
|
||
quote=quote,
|
||
)
|
||
self.assertIn("【WO202609140001报价方案】", text)
|
||
self.assertIn("工单号：WO202609140001", text)
|
||
self.assertIn("起运地：上海", text)
|
||
self.assertIn("体积：2.5", text)
|
||
self.assertIn("空运费：USD 1,200", text)
|
||
self.assertIn("时效：3-5 天", text)
|
||
self.assertIn("报价有效期：2026-09-20", text)
|
||
self.assertNotIn("报价日期：", text)
|
||
self.assertNotIn("工单状态", text)
|
||
payload = copy.scheme_wecom_payload(
|
||
work_order_no="WO202609140001", facts=COMPLETE_AIR, quote=quote
|
||
)
|
||
card = payload["template_card"]
|
||
self.assertEqual(card["main_title"]["title"], "WO202609140001报价方案")
|
||
self.assertTrue(str(card["task_id"]).startswith("scheme-WO202609140001-"))
|
||
again = copy.scheme_wecom_payload(
|
||
work_order_no="WO202609140001", facts=COMPLETE_AIR, quote=quote
|
||
)
|
||
self.assertNotEqual(card["task_id"], again["template_card"]["task_id"])
|
||
self.assertEqual(
|
||
copy.work_order_from_card_meta({"task_id": card["task_id"]}),
|
||
"WO202609140001",
|
||
)
|
||
quote_text = card["quote_area"]["quote_text"]
|
||
self.assertNotIn("title", card["quote_area"])
|
||
self.assertIn("空运费：USD 1,200", quote_text)
|
||
self.assertIn("总销售价合计：USD 1,280", quote_text)
|
||
self.assertNotIn("sub_title_text", card)
|
||
rows = {x["keyname"]: x["value"] for x in card["horizontal_content_list"]}
|
||
self.assertEqual(rows["工单号"], "WO202609140001")
|
||
self.assertIn("上海", rows["航线"])
|
||
self.assertIn("洛杉矶", rows["航线"])
|
||
self.assertIn("普货", rows["货物"])
|
||
self.assertIn("2.5", rows["货物"])
|
||
self.assertEqual(rows["时效"], "3-5 天")
|
||
self.assertEqual(rows["报价有效期"], "2026-09-20")
|
||
self.assertNotIn("报价日期", rows)
|
||
empty_eta = copy.scheme_horizontal_rows(
|
||
work_order_no="WO1", facts=COMPLETE_AIR, quote={"valid_until": "2026-09-20"}
|
||
)
|
||
self.assertEqual({x["keyname"]: x["value"] for x in empty_eta}["时效"], "-")
|
||
dashed = copy.scheme_horizontal_rows(
|
||
work_order_no="WO1",
|
||
facts=COMPLETE_AIR,
|
||
quote={"eta": "-3-5 天", "valid_until": "2026-09-20"},
|
||
)
|
||
dashed_rows = {x["keyname"]: x["value"] for x in dashed}
|
||
self.assertEqual(dashed_rows["时效"], "3-5 天")
|
||
self.assertEqual(len(card["button_list"]), 3)
|
||
self.assertEqual(
|
||
[x["text"] for x in card["button_list"]],
|
||
["生成Excel", "生成PDF", "不采用"],
|
||
)
|
||
self.assertEqual(
|
||
copy.work_order_from_card_meta(
|
||
{
|
||
"task_id": "scheme-WO202609140001",
|
||
"event_key": "adopt_excel:WO202609140001",
|
||
}
|
||
),
|
||
"WO202609140001",
|
||
)
|
||
self.assertEqual(copy.canonical_button("adopt_excel:WO202609140001"), copy.BTN_ADOPT_EXCEL)
|
||
self.assertEqual(copy.canonical_button(copy.BTN_ADOPT_EXCEL_CARD), copy.BTN_ADOPT_EXCEL)
|
||
|
||
def test_same_demand_cites_latest_even_if_closed(self) -> None:
|
||
ledger = MemoryLedger()
|
||
flow = AirTextInquiryFlow(ledger)
|
||
replies_a: list[str] = []
|
||
|
||
def _keep_a(text: str, extra=None) -> None:
|
||
replies_a.append(text)
|
||
_ = extra
|
||
|
||
flow.on_text(
|
||
sender_id="hist-a",
|
||
text="空运",
|
||
reply=_keep_a,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
wo1 = flow.session_of("hist-a").work_order_no
|
||
self.assertTrue(wo1)
|
||
ledger.transition(work_order_no=wo1, to_status="已关闭")
|
||
replies_b: list[str] = []
|
||
|
||
def _keep_b(text: str, extra=None) -> None:
|
||
replies_b.append(text)
|
||
_ = extra
|
||
|
||
flow.on_text(
|
||
sender_id="hist-b",
|
||
text="空运",
|
||
reply=_keep_b,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
wo2 = flow.session_of("hist-b").work_order_no
|
||
self.assertTrue(wo2)
|
||
self.assertNotEqual(wo2, wo1)
|
||
self.assertIn(f"引用历史工单号{wo1}", "\n".join(replies_b))
|
||
replies_c: list[str] = []
|
||
|
||
def _keep_c(text: str, extra=None) -> None:
|
||
replies_c.append(text)
|
||
_ = extra
|
||
|
||
flow.on_text(
|
||
sender_id="hist-c",
|
||
text="空运",
|
||
reply=_keep_c,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
wo3 = flow.session_of("hist-c").work_order_no
|
||
self.assertNotEqual(wo3, wo2)
|
||
self.assertIn(f"引用历史工单号{wo2}", "\n".join(replies_c))
|
||
self.assertNotIn(f"引用历史工单号{wo1}", "\n".join(replies_c))
|
||
|
||
def test_card_event_key_matches_skip_button(self) -> None:
|
||
from agent.channel.wecom.parse import parse_plain_xml
|
||
|
||
xml = (
|
||
"<xml><ToUserName>ww</ToUserName><FromUserName>WuJiLin</FromUserName>"
|
||
"<CreateTime>1</CreateTime><MsgType>event</MsgType>"
|
||
"<Event>template_card_event</Event>"
|
||
"<EventKey>skip_collab:WO202609140099</EventKey>"
|
||
"<TaskId>quote-WO202609140099</TaskId>"
|
||
"<ResponseCode>rc-wo-001</ResponseCode></xml>"
|
||
)
|
||
msg = parse_plain_xml(xml, default_agent_id="1000010")
|
||
self.assertEqual(msg.sender_id, "WuJiLin")
|
||
self.assertEqual(msg.content, "skip_collab:WO202609140099")
|
||
self.assertEqual(msg.raw.get("response_code"), "rc-wo-001")
|
||
self.assertEqual(msg.raw.get("task_id"), "quote-WO202609140099")
|
||
self.assertEqual(copy.canonical_button(msg.content), copy.BTN_SKIP_COLLAB)
|
||
self.assertEqual(
|
||
copy.work_order_from_card_meta(msg.raw, text=msg.content),
|
||
"WO202609140099",
|
||
)
|
||
self.flow.on_text(
|
||
sender_id="WuJiLin",
|
||
text="空运",
|
||
reply=self._reply,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
wo = self.flow.session_of("WuJiLin").work_order_no
|
||
xml_live = xml.replace("WO202609140099", wo)
|
||
live = parse_plain_xml(xml_live, default_agent_id="1000010")
|
||
phase = self.flow.on_text(
|
||
sender_id="WuJiLin",
|
||
text=live.content,
|
||
reply=self._reply,
|
||
inbound_raw=live.raw,
|
||
)
|
||
self.assertEqual(phase, "wait_adopt")
|
||
|
||
def test_old_price_card_uses_that_ticket_not_latest(self) -> None:
|
||
"""同一人连问两单，点第一张价格卡必须出第一单的方案，不能落到最新单。"""
|
||
facts_a = dict(COMPLETE_AIR)
|
||
facts_a["品名"] = "茶叶"
|
||
facts_b = dict(COMPLETE_AIR)
|
||
facts_b["品名"] = "服装"
|
||
self.flow.on_text(
|
||
sender_id="card-owner",
|
||
text="空运",
|
||
reply=self._reply,
|
||
injected_facts=facts_a,
|
||
injected_mode="AIR",
|
||
)
|
||
wo1 = self.flow.session_of("card-owner").work_order_no
|
||
self.assertTrue(wo1)
|
||
self.flow.on_text(
|
||
sender_id="card-owner",
|
||
text="空运",
|
||
reply=self._reply,
|
||
injected_facts=facts_b,
|
||
injected_mode="AIR",
|
||
)
|
||
wo2 = self.flow.session_of("card-owner").work_order_no
|
||
self.assertTrue(wo2)
|
||
self.assertNotEqual(wo1, wo2)
|
||
|
||
self.replies.clear()
|
||
phase = self.flow.on_text(
|
||
sender_id="card-owner",
|
||
text=f"skip_collab:{wo1}",
|
||
reply=self._reply,
|
||
inbound_raw={
|
||
"event": "template_card_event",
|
||
"event_key": f"skip_collab:{wo1}",
|
||
"task_id": f"quote-{wo1}",
|
||
"response_code": "rc-old-card",
|
||
},
|
||
)
|
||
self.assertEqual(phase, "wait_adopt")
|
||
scheme = "\n".join(x for x in self.replies if "报价方案" in x)
|
||
self.assertIn(wo1, scheme)
|
||
self.assertIn("茶叶", scheme)
|
||
self.assertNotIn(wo2, scheme)
|
||
self.assertNotIn("服装", scheme)
|
||
self.assertEqual(self.flow.session_of("card-owner").work_order_no, wo1)
|
||
|
||
def test_second_ticket_skip_still_greys_card(self) -> None:
|
||
"""两张价格卡都要点「不需要协同」并置灰，去重键不能只按按钮文案。"""
|
||
from agent.handlers.card_action import handle_card_action
|
||
|
||
store = MemoryMessageStore()
|
||
ledger = MemoryLedger()
|
||
flow = AirTextInquiryFlow(ledger)
|
||
|
||
def _open(uid: str) -> str:
|
||
flow.on_text(
|
||
sender_id=uid,
|
||
text="空运",
|
||
reply=lambda *_a, **_k: None,
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
return flow.session_of(uid).work_order_no
|
||
|
||
wo1 = _open("ua")
|
||
wo2 = _open("ua")
|
||
handle_card_action(
|
||
sender_id="ua",
|
||
thread_id="",
|
||
wait_version=0,
|
||
action=copy.BTN_SKIP_COLLAB,
|
||
payload={"response_code": "rc-aaa", "task_id": f"quote-{wo1}-x"},
|
||
flow=flow,
|
||
store=store,
|
||
)
|
||
handle_card_action(
|
||
sender_id="ua",
|
||
thread_id="",
|
||
wait_version=0,
|
||
action=copy.BTN_SKIP_COLLAB,
|
||
payload={"response_code": "rc-bbb", "task_id": f"quote-{wo2}-y"},
|
||
flow=flow,
|
||
store=store,
|
||
)
|
||
updates = [
|
||
item
|
||
for item in store._outbox.values()
|
||
if (item.payload or {}).get("msgtype") == "update_template_card"
|
||
]
|
||
self.assertEqual(len(updates), 2)
|
||
|
||
def test_adopt_excel_without_template_stays_wait_adopt(self) -> None:
|
||
ledger = MemoryLedger()
|
||
flow = AirTextInquiryFlow(ledger)
|
||
replies: list[str] = []
|
||
flow.on_text(
|
||
sender_id="u-tpl-0",
|
||
text="空运",
|
||
reply=lambda t, extra=None: replies.append(t),
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
flow.on_button(
|
||
sender_id="u-tpl-0",
|
||
action=copy.BTN_SKIP_COLLAB,
|
||
reply=lambda t, extra=None: replies.append(t),
|
||
)
|
||
ledger.set_quote_templates([])
|
||
phase = flow.on_button(
|
||
sender_id="u-tpl-0",
|
||
action=copy.BTN_ADOPT_EXCEL,
|
||
reply=lambda t, extra=None: replies.append(t),
|
||
)
|
||
self.assertEqual(phase, "wait_adopt")
|
||
self.assertIn(copy.QUOTE_TEMPLATE_MISSING, "\n".join(replies))
|
||
self.assertNotIn("咨询成交状态", "\n".join(replies))
|
||
self.assertEqual(flow.session_of("u-tpl-0").phase, "wait_adopt")
|
||
|
||
def test_adopt_excel_keyword_then_default(self) -> None:
|
||
ledger = MemoryLedger()
|
||
ledger.set_quote_templates(
|
||
[
|
||
{
|
||
"templateId": "d1",
|
||
"name": "默认空运",
|
||
"bizType": "空运",
|
||
"keywords": ["欧洲"],
|
||
"isDefault": True,
|
||
},
|
||
{
|
||
"templateId": "k1",
|
||
"name": "洛杉矶专线",
|
||
"bizType": "空运",
|
||
"keywords": ["洛杉矶"],
|
||
"isDefault": False,
|
||
},
|
||
]
|
||
)
|
||
flow = AirTextInquiryFlow(ledger)
|
||
replies: list[str] = []
|
||
flow.on_text(
|
||
sender_id="u-tpl-k",
|
||
text="空运",
|
||
reply=lambda t, extra=None: replies.append(t),
|
||
injected_facts=COMPLETE_AIR,
|
||
injected_mode="AIR",
|
||
)
|
||
flow.on_button(
|
||
sender_id="u-tpl-k",
|
||
action=copy.BTN_SKIP_COLLAB,
|
||
reply=lambda t, extra=None: replies.append(t),
|
||
)
|
||
phase = flow.on_button(
|
||
sender_id="u-tpl-k",
|
||
action=copy.BTN_ADOPT_EXCEL,
|
||
reply=lambda t, extra=None: replies.append(t),
|
||
)
|
||
self.assertEqual(phase, "wait_deal")
|
||
self.assertEqual(ledger.last_render.get("matchType"), "keyword")
|
||
self.assertEqual(ledger.last_render.get("templateName"), "洛杉矶专线")
|
||
self.assertIn("咨询成交状态", "\n".join(replies))
|
||
|
||
ny = dict(COMPLETE_AIR)
|
||
ny["目的港"] = "纽约"
|
||
flow2 = AirTextInquiryFlow(ledger)
|
||
replies2: list[str] = []
|
||
flow2.on_text(
|
||
sender_id="u-tpl-d",
|
||
text="空运",
|
||
reply=lambda t, extra=None: replies2.append(t),
|
||
injected_facts=ny,
|
||
injected_mode="AIR",
|
||
)
|
||
flow2.on_button(
|
||
sender_id="u-tpl-d",
|
||
action=copy.BTN_SKIP_COLLAB,
|
||
reply=lambda t, extra=None: replies2.append(t),
|
||
)
|
||
phase2 = flow2.on_button(
|
||
sender_id="u-tpl-d",
|
||
action=copy.BTN_ADOPT_EXCEL,
|
||
reply=lambda t, extra=None: replies2.append(t),
|
||
)
|
||
self.assertEqual(phase2, "wait_deal")
|
||
self.assertEqual(ledger.last_render.get("matchType"), "default")
|
||
self.assertEqual(ledger.last_render.get("templateName"), "默认空运")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|