1814 lines
73 KiB
Python
1814 lines
73 KiB
Python
"""
|
|
空运群内协同：激活当前工单、补字段、航线报价、锁舱/释放、成交。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
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)
|
|
|
|
from agent.channel.wecom.models import InboundMessage
|
|
from agent.handlers.air_group_collab import handle_air_group
|
|
from agent.ledger.memory_ledger import MemoryLedger
|
|
from agent.policy.air_text_flow import AirTextInquiryFlow
|
|
from agent.policy import inquiry_copy as copy
|
|
from agent.routing.air_group_intent import (
|
|
INTENT_LOCK,
|
|
INTENT_RELEASE,
|
|
classify_air_group_text,
|
|
)
|
|
from agent.routing.dispatch import dispatch_inbound
|
|
from agent.routing.decision import RouteDecision
|
|
|
|
|
|
class FakeBotGroup:
|
|
def __init__(self) -> None:
|
|
self.sent: list[dict] = []
|
|
|
|
def send_group(self, *, chat_id: str, content: str, mention_userids=None):
|
|
self.sent.append({"chat_id": chat_id, "content": content, "kind": "text"})
|
|
|
|
def send_group_file(self, *, chat_id: str, filename: str, filepath: str = "", file_b64: str = ""):
|
|
self.sent.append({"chat_id": chat_id, "filename": filename, "kind": "file"})
|
|
return {"ok": True}
|
|
|
|
|
|
class AirSheetQuoteTotalTests(unittest.TestCase):
|
|
"""群填表报价包必须带 TMS 总销售价，不能把附加费之和当预估费用。"""
|
|
|
|
def test_sheet_quote_replaces_fee_sum_total_with_tms_sale_price(self) -> None:
|
|
from agent.policy.air_group_ops import _sheet_quote_with_tms_sale_total
|
|
|
|
thin = {
|
|
"total": "200",
|
|
"currency": "CNY",
|
|
"fee_lines": [
|
|
{"name": "增点费", "amount": "100", "currency": "CNY"},
|
|
{"name": "拆板费", "amount": "100", "currency": "USD"},
|
|
],
|
|
}
|
|
tms = {
|
|
"source": "TMS",
|
|
"total": "7918.25",
|
|
"currency": "CNY",
|
|
"chargeableWeightKg": "833.5",
|
|
"airOptions": [
|
|
{
|
|
"optionNo": "AIR-1",
|
|
"salePrice": 7918.25,
|
|
"currency": "CNY",
|
|
"departureCharges": [
|
|
{"chargeName": "增点费", "unitPrice": 100, "currency": "CNY"},
|
|
{"chargeName": "拆板费", "unitPrice": 100, "currency": "USD"},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
out = _sheet_quote_with_tms_sale_total(thin, tms=tms, facts={"起运港": "南京", "目的港": "吉隆坡"})
|
|
self.assertIn("7918.25", str(out.get("total") or ""))
|
|
self.assertNotEqual(str(out.get("total") or "").strip(), "200")
|
|
self.assertTrue(out.get("airOptions") or out.get("selected_air_option"))
|
|
|
|
def test_sheet_quote_keeps_product_total_when_not_fee_sum(self) -> None:
|
|
from agent.policy.air_group_ops import _sheet_quote_with_tms_sale_total
|
|
|
|
product = {
|
|
"total": "9000",
|
|
"currency": "CNY",
|
|
"source_label": "航线报价",
|
|
"fee_lines": [
|
|
{"name": "增点费", "amount": "100", "currency": "CNY"},
|
|
{"name": "拆板费", "amount": "100", "currency": "USD"},
|
|
],
|
|
}
|
|
tms = {
|
|
"source": "TMS",
|
|
"total": "7918.25",
|
|
"currency": "CNY",
|
|
"airOptions": [{"optionNo": "AIR-1", "salePrice": 7918.25, "currency": "CNY"}],
|
|
}
|
|
out = _sheet_quote_with_tms_sale_total(product, tms=tms, facts={})
|
|
self.assertEqual(str(out.get("total") or "").strip(), "9000")
|
|
|
|
|
|
class AirGroupCollabTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
from agent.redis_coord.air_inbound import reset_air_inbound_for_tests
|
|
|
|
reset_air_inbound_for_tests()
|
|
self.ledger = MemoryLedger()
|
|
self.groups = FakeBotGroup()
|
|
self.flow = AirTextInquiryFlow(ledger=self.ledger, group_client=self.groups)
|
|
self.ledger.upsert_staff(
|
|
{
|
|
"name": "WuJiLin",
|
|
"wecomId": "sales1",
|
|
"roleCode": "sales",
|
|
"status": "active",
|
|
}
|
|
)
|
|
self.ledger.upsert_staff(
|
|
{
|
|
"name": "HangXian",
|
|
"wecomId": "air1",
|
|
"roleCode": "air",
|
|
"status": "active",
|
|
}
|
|
)
|
|
self.ledger.set_quote_templates(
|
|
[
|
|
{
|
|
"templateId": "air-default",
|
|
"name": "空运报价",
|
|
"bizType": "空运",
|
|
"isDefault": True,
|
|
},
|
|
{
|
|
"templateId": "sea-default",
|
|
"name": "海运报价",
|
|
"bizType": "海运",
|
|
"isDefault": True,
|
|
},
|
|
]
|
|
)
|
|
|
|
def _create_air(self, *, sender="sales1", quote=None, status="已报价", quote_date="2026-09-16") -> str:
|
|
created = self.ledger.create_ticket(
|
|
sender_id=sender,
|
|
business_line="AIR",
|
|
facts={
|
|
"起运地": "上海",
|
|
"目的地": "洛杉矶",
|
|
"品名": "配件",
|
|
"件数": "3",
|
|
"毛重": "100",
|
|
"体积": "2",
|
|
"包装类型": "纸箱",
|
|
"报价日期": quote_date,
|
|
},
|
|
)
|
|
wo = created["work_order_no"]
|
|
if quote is None:
|
|
quote = {
|
|
"total": "USD 800",
|
|
"source": "TMS",
|
|
"source_label": "TMS 标准报价",
|
|
"quoteId": "QT-AIR-1",
|
|
"airOptions": [
|
|
{"optionNo": "AIR-OPT-01", "airlineCode": "CZ", "airlineName": "南航", "flightNo": "CZ327"}
|
|
],
|
|
}
|
|
if quote:
|
|
self.ledger.upsert_quote(work_order_no=wo, quote=quote, to_status=status)
|
|
return wo
|
|
|
|
def _msg(self, *, sender: str, text: str, chat: str, mid: str) -> InboundMessage:
|
|
return InboundMessage(
|
|
sender_id=sender,
|
|
message_id=mid,
|
|
content=text,
|
|
chat_id=chat,
|
|
chat_type="group",
|
|
raw={"source": "aibot"},
|
|
)
|
|
|
|
def _texts(self) -> list[str]:
|
|
return [x.get("content") or "" for x in self.groups.sent]
|
|
|
|
def test_attach_bot_client_keeps_req_id_for_quote(self) -> None:
|
|
"""这次 @ 的 req_id 要带给桥，群里才能引用原消息。"""
|
|
from agent.channel.aibot.reply import AibotReplyClient
|
|
from agent.handlers.air_group_collab import _attach_bot_client
|
|
|
|
engine = AirTextInquiryFlow(ledger=self.ledger, group_client=None)
|
|
msg = InboundMessage(
|
|
sender_id="air1",
|
|
message_id="m_quote",
|
|
content="锁舱",
|
|
chat_id="wr_air_1",
|
|
chat_type="group",
|
|
raw={"source": "aibot", "aibot_req_id": "cb_lock_1"},
|
|
)
|
|
_attach_bot_client(engine, msg)
|
|
client = engine.group_client()
|
|
self.assertIsInstance(client, AibotReplyClient)
|
|
self.assertEqual(client._callback_req_id, "cb_lock_1")
|
|
|
|
def test_intent_lock_is_not_hardcoded(self) -> None:
|
|
self.assertEqual(classify_air_group_text("帮我锁舱 WO202609160001", allow_network=False), INTENT_LOCK)
|
|
self.assertEqual(classify_air_group_text("WO202609160001释放舱位", allow_network=False), INTENT_RELEASE)
|
|
|
|
def test_airline_fee_wording_is_quote_not_adjust(self) -> None:
|
|
"""航线写「空运费改为900」是手工报价，不能先判成调价。比例和改总价仍是调价。"""
|
|
from agent.routing.air_group_intent import INTENT_ADJUST, INTENT_QUOTE
|
|
|
|
self.assertEqual(
|
|
classify_air_group_text("空运费改为900USD", allow_network=False),
|
|
INTENT_QUOTE,
|
|
)
|
|
self.assertEqual(
|
|
classify_air_group_text("报关费改为300CNY", allow_network=False),
|
|
INTENT_QUOTE,
|
|
)
|
|
self.assertEqual(
|
|
classify_air_group_text("空运费 9900CNY", allow_network=False),
|
|
INTENT_QUOTE,
|
|
)
|
|
self.assertEqual(
|
|
classify_air_group_text("空运费加10%", allow_network=False),
|
|
INTENT_ADJUST,
|
|
)
|
|
self.assertEqual(
|
|
classify_air_group_text("总销售价改为180", allow_network=False),
|
|
INTENT_ADJUST,
|
|
)
|
|
|
|
def test_airline_rewritten_fee_does_not_say_only_sales(self) -> None:
|
|
"""已有报价后，航线再写费用金额，不能提示只有销售能调价。"""
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="air1", text=wo, chat="wr_air_1", mid="fee0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.groups.sent.clear()
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text="空运费改为900USD", chat="wr_air_1", mid="fee1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertNotEqual(phase, "adjust_not_sales")
|
|
self.assertNotIn("只有这张单的销售能调价", "\n".join(self._texts()))
|
|
self.assertIn(phase, {"wait_tms_choice", "quoted"})
|
|
|
|
def test_sales_percent_still_adjusts(self) -> None:
|
|
"""销售说加百分之几，仍是调价，不会被改成航线报价。"""
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="adj0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.groups.sent.clear()
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text="空运费加10%", chat="wr_air_1", mid="adj1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertNotEqual(phase, "not_airline")
|
|
self.assertNotIn("只有这张单的销售能调价", "\n".join(self._texts()))
|
|
self.assertIn(phase, {"group_adjusted", "adjust_need_fee", "adjust_none", "adjust_need_lines"})
|
|
|
|
def test_air_group_brief_matches_sea_sections(self) -> None:
|
|
text = copy.air_group_brief(
|
|
work_order_no="WO202609160009",
|
|
facts={
|
|
"起运地": "ZUH",
|
|
"目的地": "广州",
|
|
"品名": "服务器",
|
|
"件数": "1",
|
|
"毛重": "10",
|
|
"体积": "1",
|
|
"包装类型": "纸箱",
|
|
"报价日期": "2026-09-16",
|
|
},
|
|
quote=None,
|
|
collab_facts={},
|
|
has_price=False,
|
|
)
|
|
self.assertIn("工单号：WO202609160009", text)
|
|
self.assertIn("询价字段：\n运输方式：空运", text)
|
|
self.assertIn("起运地：ZUH", text)
|
|
self.assertIn("\n\n协同补充字段：\n是否含电：-", text)
|
|
self.assertIn("是否含磁：-", text)
|
|
general = copy.air_group_brief(
|
|
work_order_no="WO202609230017",
|
|
facts={
|
|
"起运地": "上海",
|
|
"目的地": "洛杉矶",
|
|
"品名": "普货",
|
|
},
|
|
quote=None,
|
|
collab_facts={},
|
|
has_price=False,
|
|
)
|
|
self.assertIn("是否含电：否", general)
|
|
self.assertIn("是否含磁：否", general)
|
|
self.assertNotIn("是否含电：-", general)
|
|
self.assertIn("\n\nTMS报价：\nTMS 暂无匹配报价", text)
|
|
self.assertIn("航线人员可通过“纯文字”告知新报价，或直接将新报价excel表发给智能体。", text)
|
|
self.assertIn("附件报价单仅作为文件存储于后台，智能体不读取且不将报价内容写进系统", text)
|
|
self.assertNotIn("产品人员可通过", text)
|
|
self.assertNotIn("当前工单：", text)
|
|
self.assertNotIn("贸易条款", text)
|
|
|
|
def test_air_group_brief_tms_total_only_once(self) -> None:
|
|
"""群摘要 TMS 段总销售价只一行，禁止重复。"""
|
|
tms = {
|
|
"source": "TMS",
|
|
"source_label": "TMS标准报价",
|
|
"total": "183370.0",
|
|
"currency": "CNY",
|
|
"eta": "-",
|
|
"valid_until": "2026-09-30",
|
|
"fee_lines": [
|
|
{"name": "运输费", "amount": "11.0", "currency": "CNY"},
|
|
],
|
|
}
|
|
text = copy.air_group_brief(
|
|
work_order_no="WO202609220018",
|
|
facts={"起运港": "珠海", "目的港": "克拉克"},
|
|
quote=tms,
|
|
collab_facts={},
|
|
has_price=True,
|
|
tms_quote=tms,
|
|
)
|
|
tms_block = text.split("TMS报价：", 1)[1]
|
|
self.assertEqual(tms_block.count("费用合计："), 1)
|
|
self.assertIn("费用合计：CNY 183370.0", tms_block)
|
|
self.assertNotIn("时效", tms_block)
|
|
self.assertIn("报价有效期：2026-09-30", tms_block)
|
|
self.assertIn("请航线人员回复“确认报价”确定采用TMS报价。", text)
|
|
self.assertIn("亦可通过“纯文字”告知新报价，或直接将新报价excel表发给智能体。", text)
|
|
self.assertIn("附件报价单仅作为文件存储于后台，智能体不读取且不将报价内容写进系统", text)
|
|
self.assertNotIn("请产品人员回复", text)
|
|
|
|
def test_air_group_brief_uses_route_fees_not_air_freight_total(self) -> None:
|
|
"""
|
|
群摘要必须跟单聊同一套线路分项。
|
|
主账快照若只有「空运费=总价」，仍要列出该线路运输费/提送货费/人工费。
|
|
"""
|
|
tms = {
|
|
"source": "TMS",
|
|
"source_label": "TMS标准报价",
|
|
"total": "183370.0",
|
|
"freight": "183370.0",
|
|
"currency": "CNY",
|
|
"eta": "-",
|
|
"valid_until": "2026-09-30",
|
|
"fee_lines": [
|
|
{
|
|
"name": "空运费",
|
|
"amount": "183370.00",
|
|
"currency": "CNY",
|
|
}
|
|
],
|
|
"airOptions": [
|
|
{
|
|
"originAirportCode": "ZUH",
|
|
"transferAirportCode": "CHG",
|
|
"destinationAirportCode": "CRK",
|
|
"salePrice": 183370,
|
|
"currency": "CNY",
|
|
"departureCharges": [
|
|
{"chargeName": "运输费", "currency": "CNY", "unitPrice": 11.0},
|
|
{"chargeName": "提送货费", "currency": "CNY", "unitPrice": 6.0},
|
|
{"chargeName": "人工费", "currency": "CNY", "unitPrice": 123.0},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
text = copy.air_group_brief(
|
|
work_order_no="WO202609220018",
|
|
facts={"起运港": "珠海", "目的港": "克拉克", "originCode": "ZUH"},
|
|
quote=tms,
|
|
collab_facts={},
|
|
has_price=True,
|
|
tms_quote=tms,
|
|
)
|
|
tms_block = text.split("TMS报价：", 1)[1]
|
|
self.assertIn("运输费：CNY 11.0", tms_block)
|
|
self.assertIn("提送货费：CNY 6.0", tms_block)
|
|
self.assertIn("人工费：CNY 123.0", tms_block)
|
|
self.assertIn("费用合计：CNY 183370", tms_block)
|
|
self.assertEqual(tms_block.count("费用合计："), 1)
|
|
self.assertNotIn("空运费", tms_block)
|
|
|
|
def test_air_group_brief_multi_route_does_not_borrow_other_air_freight(self) -> None:
|
|
"""多条线路：只出对上起运港那条的分项，禁止套整单空运费或另一条的运输费。"""
|
|
tms = {
|
|
"source": "TMS",
|
|
"total": "200040",
|
|
"freight": "183370.00",
|
|
"currency": "CNY",
|
|
"valid_until": "2026-09-30",
|
|
"fee_lines": [
|
|
{"name": "空运费", "amount": "183370.00", "currency": "CNY"},
|
|
],
|
|
"airOptions": [
|
|
{
|
|
"originAirportCode": "ZUH",
|
|
"transferAirportCode": "CHG",
|
|
"destinationAirportCode": "CRK",
|
|
"salePrice": 183370,
|
|
"currency": "CNY",
|
|
"departureCharges": [
|
|
{"chargeName": "运输费", "currency": "CNY", "unitPrice": 11.0},
|
|
],
|
|
},
|
|
{
|
|
"originAirportCode": "SZX",
|
|
"destinationAirportCode": "CRK",
|
|
"salePrice": 200040,
|
|
"currency": "CNY",
|
|
"departureCharges": [],
|
|
},
|
|
],
|
|
}
|
|
text = copy.air_group_brief(
|
|
work_order_no="WO202609220018",
|
|
facts={"起运港": "深圳", "目的港": "克拉克", "originCode": "SZX"},
|
|
quote=tms,
|
|
collab_facts={},
|
|
has_price=True,
|
|
tms_quote=tms,
|
|
)
|
|
tms_block = text.split("TMS报价：", 1)[1]
|
|
self.assertNotIn("空运费", tms_block)
|
|
self.assertNotIn("运输费", tms_block)
|
|
self.assertIn("费用合计：CNY 200040", tms_block)
|
|
self.assertEqual(tms_block.count("费用合计："), 1)
|
|
|
|
def test_air_group_brief_keeps_sole_air_freight_line(self) -> None:
|
|
"""TMS 只有空运费、金额等于总价时，群摘要必须写出这一行，不能只剩费用合计。"""
|
|
tms = {
|
|
"source": "TMS",
|
|
"source_label": "TMS标准报价",
|
|
"total": "1833.7",
|
|
"freight": "1833.70",
|
|
"currency": "CNY",
|
|
"valid_until": "2026-10-21",
|
|
"fee_lines": '[{"name":"空运费","amount":"CNY 1833.70"}]',
|
|
"feeItems": [
|
|
{
|
|
"feeName": "空运费",
|
|
"feeCode": "AIR_FREIGHT",
|
|
"amount": 1833.70,
|
|
"unitPrice": 11.00,
|
|
"quantity": 166.7,
|
|
"currency": "CNY",
|
|
}
|
|
],
|
|
"airOptions": [
|
|
{
|
|
"originAirportCode": "CGO",
|
|
"destinationAirportCode": "KUL",
|
|
"salePrice": 1833.70,
|
|
"currency": "CNY",
|
|
"departureCharges": [],
|
|
}
|
|
],
|
|
}
|
|
text = copy.air_group_brief(
|
|
work_order_no="WO202610080020",
|
|
facts={"起运地": "CGO", "目的地": "吉隆坡", "originCode": "CGO"},
|
|
quote=tms,
|
|
collab_facts={},
|
|
has_price=True,
|
|
tms_quote=tms,
|
|
)
|
|
tms_block = text.split("TMS报价：", 1)[1]
|
|
self.assertIn("空运费：CNY 1833.70", tms_block)
|
|
self.assertLess(tms_block.find("空运费："), tms_block.find("费用合计："))
|
|
self.assertIn("费用合计：CNY 1833.7", tms_block)
|
|
self.assertEqual(tms_block.count("费用合计："), 1)
|
|
self.assertEqual(tms_block.count("费用合计："), 1)
|
|
|
|
def test_work_order_plus_lost_reason_is_deal_not_activate(self) -> None:
|
|
"""WO@助手 未成交，原因… 必须当关单，不能只激活。"""
|
|
from agent.routing.air_group_intent import INTENT_ACTIVATE, INTENT_LOST
|
|
|
|
text = "WO202609220014@询价小助手 未成交，原因客户不想送货了"
|
|
self.assertEqual(classify_air_group_text(text, allow_network=False), INTENT_LOST)
|
|
from unittest.mock import patch
|
|
|
|
with patch(
|
|
"agent.routing.air_group_intent._classify_with_model",
|
|
return_value=INTENT_ACTIVATE,
|
|
):
|
|
self.assertEqual(
|
|
classify_air_group_text(text, allow_network=True),
|
|
INTENT_LOST,
|
|
)
|
|
|
|
def test_activate_lost_and_reason_same_message(self) -> None:
|
|
wo = self._create_air()
|
|
phase = handle_air_group(
|
|
self._msg(
|
|
sender="sales1",
|
|
text=f"{wo}@询价小助手 未成交，原因客户不想送货了",
|
|
chat="wr_air_1",
|
|
mid="lost-inline-wo",
|
|
),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "done")
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.status, "未成交")
|
|
self.assertEqual(ticket.lost_reason, "客户不想送货了")
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn("客户不想送货了", joined)
|
|
self.assertNotIn("请补充未成交原因", joined)
|
|
|
|
def test_work_order_only_is_activate_not_quote(self) -> None:
|
|
from agent.routing.air_group_intent import INTENT_ACTIVATE
|
|
|
|
self.assertEqual(
|
|
classify_air_group_text("WO202609160002", allow_network=False),
|
|
INTENT_ACTIVATE,
|
|
)
|
|
self.assertEqual(
|
|
classify_air_group_text("@询价小助手 WO202609160002", allow_network=False),
|
|
INTENT_ACTIVATE,
|
|
)
|
|
from unittest.mock import patch
|
|
|
|
with patch(
|
|
"agent.routing.air_group_intent._classify_with_model",
|
|
side_effect=AssertionError("纯工单号不应再打模型"),
|
|
):
|
|
self.assertEqual(
|
|
classify_air_group_text(
|
|
"@询价小助手 WO202609160002",
|
|
allow_network=True,
|
|
),
|
|
INTENT_ACTIVATE,
|
|
)
|
|
|
|
def test_dispatch_aibot_goes_air(self) -> None:
|
|
from agent.routing.dispatch import dispatch_inbound as _disp
|
|
|
|
wo = self._create_air()
|
|
msg = self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="d1")
|
|
# 单测注入 flow，避免打到进程单例
|
|
import agent.handlers.air_group_collab as air_mod
|
|
orig = air_mod.get_air_text_flow
|
|
|
|
def _fake():
|
|
return self.flow
|
|
|
|
air_mod.get_air_text_flow = _fake # type: ignore
|
|
try:
|
|
name = _disp(msg, RouteDecision(intent="ordinary_text_other"))
|
|
finally:
|
|
air_mod.get_air_text_flow = orig # type: ignore
|
|
self.assertEqual(name, "air_group_collab")
|
|
|
|
def test_activate_and_switch(self) -> None:
|
|
wo1 = self._create_air()
|
|
wo2 = self._create_air()
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text=wo1, chat="wr_air_1", mid="a1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "activated")
|
|
ticket = self.ledger.get_ticket(work_order_no=wo1)
|
|
self.assertEqual(ticket.event_wait_kind, "air_quote")
|
|
self.assertTrue(ticket.event_wait_started_at)
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn(f"工单号：{wo1}", joined)
|
|
self.assertIn("询价字段：", joined)
|
|
self.assertIn("运输方式：空运", joined)
|
|
self.assertIn("协同补充字段：", joined)
|
|
self.assertIn("TMS报价：", joined)
|
|
self.assertNotIn("当前工单：", joined)
|
|
self.assertNotIn("询价要点：", joined)
|
|
self.assertTrue(any("@WuJiLin" in t and "@HangXian" in t for t in self._texts()))
|
|
self.assertEqual(getattr(self.ledger, "last_render", {}).get("templateId"), "air-default")
|
|
kinds = [x.get("kind") for x in self.groups.sent]
|
|
self.assertLess(kinds.index("text"), kinds.index("file"))
|
|
self.assertEqual(sum(1 for x in self.groups.sent if x.get("kind") == "file"), 1)
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=wo2, chat="wr_air_1", mid="a2"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "activated")
|
|
self.assertEqual(self.ledger.find_by_collab_chat(chat_id="wr_air_1").work_order_no, wo2)
|
|
|
|
def test_activate_prefills_magnet(self) -> None:
|
|
created = self.ledger.create_ticket(
|
|
sender_id="sales1",
|
|
business_line="AIR",
|
|
facts={
|
|
"起运地": "上海",
|
|
"目的地": "洛杉矶",
|
|
"品名": "配件",
|
|
"件数": "3",
|
|
"毛重": "100",
|
|
"体积": "2",
|
|
"包装类型": "纸箱",
|
|
"报价日期": "2026-09-16",
|
|
"是否含磁": "是",
|
|
},
|
|
)
|
|
wo = created["work_order_no"]
|
|
self.ledger.upsert_quote(
|
|
work_order_no=wo,
|
|
quote={"total": "USD 800", "source": "TMS", "quoteId": "QT-AIR-1"},
|
|
to_status="已报价",
|
|
)
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="pre1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn("是否含磁：是", joined)
|
|
self.assertIn("是否含电：-", joined)
|
|
|
|
def test_activate_general_cargo_defaults_battery_and_magnet(self) -> None:
|
|
"""品名是普货：群内激活后含电、含磁都是否，报价单也带上。"""
|
|
created = self.ledger.create_ticket(
|
|
sender_id="sales1",
|
|
business_line="AIR",
|
|
facts={
|
|
"起运地": "上海",
|
|
"目的地": "洛杉矶",
|
|
"品名": "普货",
|
|
"件数": "3",
|
|
"毛重": "100",
|
|
"体积": "2",
|
|
"包装类型": "纸箱",
|
|
"报价日期": "2026-09-16",
|
|
},
|
|
)
|
|
wo = created["work_order_no"]
|
|
self.ledger.upsert_quote(
|
|
work_order_no=wo,
|
|
quote={"total": "USD 800", "source": "TMS", "quoteId": "QT-AIR-1"},
|
|
to_status="已报价",
|
|
)
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="gc1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.collab_facts.get("是否含电"), "否")
|
|
self.assertEqual(ticket.collab_facts.get("是否含磁"), "否")
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn("是否含电：否", joined)
|
|
self.assertIn("是否含磁：否", joined)
|
|
fill = dict(getattr(self.ledger, "last_render", {}).get("facts") or {})
|
|
self.assertEqual(fill.get("是否含电") or fill.get("has_battery"), "否")
|
|
self.assertEqual(fill.get("是否含磁") or fill.get("has_magnet"), "否")
|
|
|
|
def test_activate_general_cargo_keeps_explicit_battery(self) -> None:
|
|
"""普货只补空着的项；私聊已经说含电，激活不能改成否。"""
|
|
created = self.ledger.create_ticket(
|
|
sender_id="sales1",
|
|
business_line="AIR",
|
|
facts={
|
|
"起运地": "上海",
|
|
"目的地": "洛杉矶",
|
|
"品名": "普货",
|
|
"件数": "3",
|
|
"毛重": "100",
|
|
"体积": "2",
|
|
"包装类型": "纸箱",
|
|
"报价日期": "2026-09-16",
|
|
"是否含电": "是",
|
|
},
|
|
)
|
|
wo = created["work_order_no"]
|
|
self.ledger.upsert_quote(
|
|
work_order_no=wo,
|
|
quote={"total": "USD 800", "source": "TMS", "quoteId": "QT-AIR-1"},
|
|
to_status="已报价",
|
|
)
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="gc2"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.collab_facts.get("是否含电"), "是")
|
|
self.assertEqual(ticket.collab_facts.get("是否含磁"), "否")
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn("是否含电：是", joined)
|
|
self.assertIn("是否含磁：否", joined)
|
|
|
|
def test_second_group_occupied(self) -> None:
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="o1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_2", mid="o2"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "occupied")
|
|
self.assertTrue(any("另一个群" in t for t in self._texts()))
|
|
|
|
def test_need_work_order_when_empty(self) -> None:
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text="空运费100USD", chat="wr_air_1", mid="n1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="product_quote",
|
|
)
|
|
self.assertEqual(phase, "need_work_order")
|
|
|
|
def test_activated_multi_quote_follows_without_work_order(self) -> None:
|
|
"""多段激活后，报价不必再带工单号，也不要回「请先 @我」。"""
|
|
import json
|
|
|
|
created = self.ledger.create_ticket(
|
|
sender_id="sales1",
|
|
business_line="MULTI",
|
|
facts={
|
|
"segments_json": json.dumps(
|
|
[
|
|
{"index": 1, "mode": "LAND", "facts": {"起运港": "北京", "目的港": "上海"}},
|
|
{"index": 2, "mode": "AIR", "facts": {"起运港": "上海", "目的港": "北宁"}},
|
|
],
|
|
ensure_ascii=False,
|
|
)
|
|
},
|
|
)
|
|
wo = created["work_order_no"]
|
|
self.ledger.bind_collab_group(
|
|
work_order_no=wo,
|
|
chat_id="wr_multi",
|
|
member_ids=["sales1", "air1"],
|
|
product_ids=["air1"],
|
|
product_names=["HangXian"],
|
|
)
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text="空运费 100USD", chat="wr_multi", mid="mq1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="product_quote",
|
|
)
|
|
self.assertEqual(phase, "segment_quoted")
|
|
joined = "\n".join(self._texts())
|
|
self.assertNotIn("请先 @我", joined)
|
|
self.assertNotIn("并发送要跟的工单号", joined)
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
air_quote = (ticket.quote or {}).get("segment_quotes", {}).get("AIR") or {}
|
|
self.assertTrue(air_quote.get("total") or air_quote.get("fee_rows"))
|
|
|
|
def test_sales_can_fill_battery(self) -> None:
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="f0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text="含电 不含磁", chat="wr_air_1", mid="f1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="collab_fields",
|
|
)
|
|
self.assertEqual(phase, "fields_saved")
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.collab_facts.get("是否含电"), "是")
|
|
self.assertEqual(ticket.collab_facts.get("是否含磁"), "否")
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn("工单号：", joined)
|
|
self.assertIn("询价字段：", joined)
|
|
self.assertIn("是否含电：是", joined)
|
|
self.assertIn("是否含磁：否", joined)
|
|
self.assertNotIn("已记下", joined)
|
|
self.assertTrue(any("@WuJiLin" in t and "@HangXian" in t for t in self._texts()))
|
|
self.assertEqual(getattr(self.ledger, "last_render", {}).get("templateId"), "air-default")
|
|
fill = dict(getattr(self.ledger, "last_render", {}).get("facts") or {})
|
|
self.assertEqual(fill.get("是否含电") or fill.get("has_battery"), "是")
|
|
self.assertEqual(fill.get("是否含磁") or fill.get("has_magnet"), "否")
|
|
texts = [i for i, x in enumerate(self.groups.sent) if x.get("kind") == "text" and "工单号：" in (x.get("content") or "")]
|
|
files = [i for i, x in enumerate(self.groups.sent) if x.get("kind") == "file"]
|
|
self.assertTrue(texts and files)
|
|
self.assertLess(texts[-1], files[-1])
|
|
self.assertEqual(len(files), 2)
|
|
|
|
def test_collab_after_activate_does_not_use_old_land_ticket(self) -> None:
|
|
"""
|
|
测服 WO202609230023：空运进原群激活后补含电含磁，
|
|
不能写到同群更早的陆运单 WO202609230007，摘要也不能变成空字段。
|
|
"""
|
|
land = self.ledger.create_ticket(
|
|
sender_id="sales1",
|
|
business_line="LAND",
|
|
facts={
|
|
"运输类型": "国内运输整车",
|
|
"始发站": "深圳",
|
|
"目的站": "上海",
|
|
"品名": "配件",
|
|
},
|
|
)["work_order_no"]
|
|
self.ledger.bind_collab_group(
|
|
work_order_no=land,
|
|
chat_id="wr_air_1",
|
|
member_ids=["sales1"],
|
|
)
|
|
air = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text=air, chat="wr_air_1", mid="mix0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(self.ledger.find_by_collab_chat(chat_id="wr_air_1").work_order_no, air)
|
|
self.assertEqual(
|
|
str(self.ledger.get_ticket(work_order_no=land).collab_chat_id or ""),
|
|
"",
|
|
)
|
|
self.groups.sent.clear()
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text="含电 不含磁", chat="wr_air_1", mid="mix1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="collab_fields",
|
|
)
|
|
self.assertEqual(phase, "fields_saved")
|
|
air_ticket = self.ledger.get_ticket(work_order_no=air)
|
|
land_ticket = self.ledger.get_ticket(work_order_no=land)
|
|
self.assertEqual(air_ticket.collab_facts.get("是否含电"), "是")
|
|
self.assertEqual(air_ticket.collab_facts.get("是否含磁"), "否")
|
|
self.assertFalse(land_ticket.collab_facts.get("是否含电"))
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn(f"工单号：{air}", joined)
|
|
self.assertNotIn(f"工单号：{land}", joined)
|
|
self.assertIn("起运地：上海", joined)
|
|
self.assertIn("重量（KG）：100", joined)
|
|
self.assertNotIn("起运地：-", joined)
|
|
|
|
def test_sales_quote_rejected(self) -> None:
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="q0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text="空运费 100USD", chat="wr_air_1", mid="q1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="product_quote",
|
|
)
|
|
self.assertEqual(phase, "not_airline")
|
|
|
|
def test_airline_quote_asks_tms_choice(self) -> None:
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="air1", text=wo, chat="wr_air_1", mid="t0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text="空运费改为900USD", chat="wr_air_1", mid="t1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="product_quote",
|
|
injected_quote={"total": "USD 900", "source_label": "航线报价"},
|
|
)
|
|
self.assertEqual(phase, "wait_tms_choice")
|
|
self.assertTrue(any("在 TMS 基础上改" in t or "在TMS基础上改" in t for t in self._texts()))
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.quote.get("total"), "USD 800")
|
|
self.assertNotEqual(ticket.quote.get("source_label"), "航线报价")
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text="在TMS基础上改", chat="wr_air_1", mid="t2"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="quote_keep_tms",
|
|
)
|
|
self.assertEqual(phase, "quoted")
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.quote.get("total"), "USD 900")
|
|
self.assertEqual(ticket.quote.get("quoteId"), "QT-AIR-1")
|
|
self.assertEqual(ticket.quote.get("source_label"), "航线在TMS上改")
|
|
self.assertFalse(ticket.collab_facts.get("__pending_air_quote"))
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn("工单号：", joined)
|
|
self.assertIn("询价字段：", joined)
|
|
self.assertIn("人工报价：", joined)
|
|
self.assertIn("报价单已发送，请确认成交状态", joined)
|
|
self.assertEqual(getattr(self.ledger, "last_render", {}).get("templateId"), "air-default")
|
|
self.assertTrue(any(x.get("kind") == "file" for x in self.groups.sent))
|
|
|
|
def test_reply_2_means_drop_tms_not_price(self) -> None:
|
|
"""问 1/2 之后，航线回 @询价小助手 2 是选不用 TMS，不能把价格改成 2。"""
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="air1", text=wo, chat="wr_air_1", mid="d0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text="空运费 900USD", chat="wr_air_1", mid="d1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="product_quote",
|
|
injected_quote={"total": "USD 900", "currency": "USD", "source_label": "航线报价"},
|
|
)
|
|
self.assertEqual(phase, "wait_tms_choice")
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text="@询价小助手 2", chat="wr_air_1", mid="d2"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "quoted")
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.quote.get("total"), "USD 900")
|
|
self.assertEqual(ticket.quote.get("source_label"), "航线报价")
|
|
self.assertNotEqual(ticket.quote.get("quoteId"), "QT-AIR-1")
|
|
|
|
def test_airline_quote_without_tms_sends_brief_sheet_deal(self) -> None:
|
|
wo = self._create_air(quote={}, status="询价中")
|
|
handle_air_group(
|
|
self._msg(sender="air1", text=wo, chat="wr_air_1", mid="nq0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.groups.sent.clear()
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text="空运费 900USD", chat="wr_air_1", mid="nq1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="product_quote",
|
|
injected_quote={"total": "USD 900", "source_label": "航线报价"},
|
|
)
|
|
self.assertEqual(phase, "quoted")
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.event_wait_kind, "deal")
|
|
self.assertTrue(ticket.event_wait_started_at)
|
|
self.assertEqual(ticket.status, "已报价")
|
|
self.assertEqual(ticket.quote.get("total"), "USD 900")
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn("工单号：", joined)
|
|
self.assertIn("询价字段：", joined)
|
|
self.assertIn("TMS 暂无匹配报价", joined)
|
|
self.assertIn("人工报价：", joined)
|
|
self.assertIn("报价单已发送，请确认成交状态", joined)
|
|
self.assertNotIn("航线报价已记下", joined)
|
|
self.assertEqual(getattr(self.ledger, "last_render", {}).get("templateId"), "air-default")
|
|
self.assertTrue(any(x.get("kind") == "file" for x in self.groups.sent))
|
|
|
|
def test_quote_same_sentence_saves_battery(self) -> None:
|
|
wo = self._create_air(quote={}, status="询价中")
|
|
handle_air_group(
|
|
self._msg(sender="air1", text=wo, chat="wr_air_1", mid="qs0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text="含电 空运费900USD", chat="wr_air_1", mid="qs1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="product_quote",
|
|
injected_quote={"total": "USD 900", "source_label": "航线报价"},
|
|
)
|
|
self.assertEqual(phase, "quoted")
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.collab_facts.get("是否含电"), "是")
|
|
self.assertIn("是否含电：是", "\n".join(self._texts()))
|
|
|
|
def test_confirm_tms_saves_sheet_and_deal(self) -> None:
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="air1", text=wo, chat="wr_air_1", mid="ct0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.groups.sent.clear()
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text="确认报价", chat="wr_air_1", mid="ct1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="quote_confirm_tms",
|
|
)
|
|
self.assertEqual(phase, "quote_confirmed")
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.quote.get("source_label"), "航线确认TMS")
|
|
self.assertTrue(any(f"工单{wo}的1个附件已保存到后台工单" in t for t in self._texts()))
|
|
self.assertTrue(any("报价单已发送，请确认成交状态" in t for t in self._texts()))
|
|
|
|
def test_stranger_cannot_fill_fields(self) -> None:
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="sf0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
phase = handle_air_group(
|
|
self._msg(sender="other1", text="含电 不含磁", chat="wr_air_1", mid="sf1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="collab_fields",
|
|
)
|
|
self.assertEqual(phase, "fields_denied")
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertFalse(ticket.collab_facts.get("是否含电"))
|
|
|
|
def test_sales_deal(self) -> None:
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="c0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text="成交", chat="wr_air_1", mid="c1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="deal_won",
|
|
)
|
|
self.assertEqual(phase, "done")
|
|
self.assertEqual(self.ledger.get_ticket(work_order_no=wo).status, "已成交")
|
|
|
|
def test_lock_requires_work_order(self) -> None:
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text="锁舱", chat="wr_air_1", mid="l0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertEqual(phase, "cabin_need_work_order")
|
|
self.assertTrue(any("带上工单号" in t for t in self._texts()))
|
|
|
|
def test_lock_without_tms_quote(self) -> None:
|
|
wo = self._create_air(quote={}, status="询价中")
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_1", mid="l1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertEqual(phase, "no_quote")
|
|
self.assertTrue(any("还没有报价，不能锁舱" in t for t in self._texts()))
|
|
self.assertFalse(any("AIR-OPT-01" in t for t in self._texts()))
|
|
self.assertFalse(any("执行成功" in t for t in self._texts()))
|
|
|
|
def test_manual_quote_without_quote_id_can_lock(self) -> None:
|
|
wo = self._create_air(
|
|
quote={
|
|
"total": "100",
|
|
"source": "manual",
|
|
"source_label": "产品报价",
|
|
},
|
|
status="已报价",
|
|
)
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_lock_manual", mid="lm1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertEqual(phase, "wait_lock_params")
|
|
phase = handle_air_group(
|
|
self._msg(
|
|
sender="air1",
|
|
text="AIR-OPT01 CZ 航班CZ327",
|
|
chat="wr_air_lock_manual",
|
|
mid="lm2",
|
|
),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_pick_option",
|
|
)
|
|
self.assertEqual(phase, "lock_ok")
|
|
self.assertTrue(any("执行成功" in t for t in self._texts()))
|
|
|
|
def test_lock_missing_params_asks_and_does_not_call_tms(self) -> None:
|
|
"""缺舱位、航司、航班号时只请补，不锁舱、不报 TMS 失败、不通知 IT。"""
|
|
wo = self._create_air(
|
|
quote={
|
|
"total": "CNY 1800",
|
|
"source": "TMS",
|
|
"source_label": "TMS 标准报价",
|
|
"quoteId": "QT-AIR-23",
|
|
}
|
|
)
|
|
calls: list[dict] = []
|
|
orig = self.ledger.lock_cabin
|
|
|
|
def spy(**kwargs):
|
|
calls.append(kwargs)
|
|
return orig(**kwargs)
|
|
|
|
self.ledger.lock_cabin = spy # type: ignore
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_lock_ask", mid="ask1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertEqual(phase, "wait_lock_params")
|
|
self.assertEqual(calls, [])
|
|
asked = self._texts()[-1]
|
|
self.assertEqual(
|
|
asked,
|
|
"锁舱参数不完整，缺少：舱位编号、航司、航班号。请补充后再锁，例如：AIR-OPT01，南航，航班CZ327。",
|
|
)
|
|
self.assertNotIn("已通知", asked)
|
|
self.assertNotIn("锁舱失败", asked)
|
|
phase = handle_air_group(
|
|
self._msg(
|
|
sender="air1",
|
|
text="AIR-OPT01 CZ 航班CZ327",
|
|
chat="wr_air_lock_ask",
|
|
mid="ask2",
|
|
),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_pick_option",
|
|
)
|
|
self.assertEqual(phase, "lock_ok")
|
|
self.assertEqual(len(calls), 1)
|
|
self.assertEqual(calls[0].get("option_no"), "AIR-OPT01")
|
|
self.assertEqual(calls[0].get("airline"), "CZ")
|
|
self.assertEqual(calls[0].get("flight_no"), "CZ327")
|
|
|
|
def test_lock_accepts_free_wording_not_only_sample_format(self) -> None:
|
|
"""补舱位不限格式。点名舱位、航司、航班号，换顺序、用航司中文名，也要锁。"""
|
|
wo = self._create_air(
|
|
quote={
|
|
"total": "CNY 1800",
|
|
"source": "TMS",
|
|
"source_label": "TMS 标准报价",
|
|
"quoteId": "QT-AIR-24",
|
|
}
|
|
)
|
|
calls: list[dict] = []
|
|
orig = self.ledger.lock_cabin
|
|
|
|
def spy(**kwargs):
|
|
calls.append(kwargs)
|
|
return orig(**kwargs)
|
|
|
|
self.ledger.lock_cabin = spy # type: ignore
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_lock_free", mid="free1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertEqual(phase, "wait_lock_params")
|
|
self.assertEqual(calls, [])
|
|
phase = handle_air_group(
|
|
self._msg(
|
|
sender="air1",
|
|
text="航班号：CZ327，航司南航，舱位编号是AIR-OPT01",
|
|
chat="wr_air_lock_free",
|
|
mid="free2",
|
|
),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "lock_ok")
|
|
self.assertEqual(calls[0].get("option_no"), "AIR-OPT01")
|
|
self.assertEqual(calls[0].get("airline"), "CZ")
|
|
self.assertEqual(calls[0].get("flight_no"), "CZ327")
|
|
|
|
def test_lock_sentence_with_labeled_params_locks_immediately(self) -> None:
|
|
"""同一句里已经说齐三项，直接锁，不再要求按示例格式重说。"""
|
|
wo = self._create_air(
|
|
quote={
|
|
"total": "CNY 1800",
|
|
"source": "manual",
|
|
"source_label": "产品报价",
|
|
}
|
|
)
|
|
calls: list[dict] = []
|
|
orig = self.ledger.lock_cabin
|
|
|
|
def spy(**kwargs):
|
|
calls.append(kwargs)
|
|
return orig(**kwargs)
|
|
|
|
self.ledger.lock_cabin = spy # type: ignore
|
|
phase = handle_air_group(
|
|
self._msg(
|
|
sender="air1",
|
|
text=f"{wo}锁舱，舱位 AIR-OPT02，东方航空，航班号 MU5123",
|
|
chat="wr_air_lock_once",
|
|
mid="once1",
|
|
),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertEqual(phase, "lock_ok")
|
|
self.assertEqual(len(calls), 1)
|
|
self.assertEqual(calls[0].get("option_no"), "AIR-OPT02")
|
|
self.assertEqual(calls[0].get("airline"), "MU")
|
|
self.assertEqual(calls[0].get("flight_no"), "MU5123")
|
|
self.assertFalse(any("锁舱参数不完整" in t for t in self._texts()))
|
|
|
|
def test_lock_success_and_release_after_deal_blocked(self) -> None:
|
|
wo = self._create_air()
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_1", mid="l2"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertEqual(phase, "lock_ok")
|
|
self.assertTrue(any("执行成功" in t for t in self._texts()))
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text="成交", chat="wr_air_1", mid="l3"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="deal_won",
|
|
)
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}释放舱位", chat="wr_air_1", mid="l4"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_release",
|
|
)
|
|
self.assertEqual(phase, "release_deal_blocked")
|
|
self.assertTrue(any("已成交" in t for t in self._texts()))
|
|
|
|
def test_lost_can_release(self) -> None:
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_1", mid="r0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text="未成交", chat="wr_air_1", mid="r1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="deal_lost",
|
|
)
|
|
before = len(self._texts())
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}释放舱位", chat="wr_air_1", mid="r2"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_release",
|
|
)
|
|
self.assertEqual(phase, "release_ok")
|
|
added = "\n".join(self._texts()[before:])
|
|
self.assertIn("释放", added)
|
|
self.assertNotIn("请确认成交状态", added)
|
|
|
|
def test_sales_cannot_lock(self) -> None:
|
|
wo = self._create_air()
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text=f"{wo}锁舱", chat="wr_air_1", mid="s1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertEqual(phase, "not_airline")
|
|
|
|
def test_dual_role_can_lock(self) -> None:
|
|
self.ledger.upsert_staff(
|
|
{
|
|
"name": "WuJiLin",
|
|
"wecomId": "sales1",
|
|
"roleCode": "air",
|
|
"status": "active",
|
|
}
|
|
)
|
|
wo = self._create_air()
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text=f"{wo}锁舱", chat="wr_air_1", mid="s2"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertEqual(phase, "lock_ok")
|
|
|
|
def test_multi_option_asks(self) -> None:
|
|
wo = self._create_air(
|
|
quote={
|
|
"total": "USD 800",
|
|
"source": "TMS",
|
|
"quoteId": "QT-AIR-2",
|
|
"airOptions": [
|
|
{"optionNo": "AIR-OPT-01", "airlineCode": "CZ", "flightNo": "CZ327"},
|
|
{"optionNo": "AIR-OPT-02", "airlineCode": "MU", "flightNo": "MU5123"},
|
|
],
|
|
}
|
|
)
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_1", mid="m1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertEqual(phase, "wait_lock_params")
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text="AIR-OPT-02", chat="wr_air_1", mid="m2"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_pick_option",
|
|
)
|
|
self.assertEqual(phase, "lock_ok")
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.collab_facts.get("__air_lock_option"), "AIR-OPT-02")
|
|
|
|
def _file(self, *, sender: str, filename: str, chat: str, mid: str, text: str = "") -> InboundMessage:
|
|
return InboundMessage(
|
|
sender_id=sender,
|
|
message_id=mid,
|
|
content=text,
|
|
msg_type="file",
|
|
chat_id=chat,
|
|
chat_type="group",
|
|
media={"filename": filename, "media_id": f"mid-{mid}"},
|
|
raw={"source": "aibot"},
|
|
)
|
|
|
|
def test_file_needs_work_order(self) -> None:
|
|
phase = handle_air_group(
|
|
self._file(sender="air1", filename="航线报价.xlsx", chat="wr_air_1", mid="af0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "need_work_order")
|
|
|
|
def test_sales_excel_archived_not_quote(self) -> None:
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="af1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
phase = handle_air_group(
|
|
self._file(sender="sales1", filename="对账单.xlsx", chat="wr_air_1", mid="af2"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "file_archived")
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.attachments[0]["file_name"], "对账单.xlsx")
|
|
self.assertNotEqual(ticket.quote.get("source_label"), "航线附件报价")
|
|
self.assertTrue(any(f"工单{wo}的1个附件已保存到后台工单" in t for t in self._texts()))
|
|
|
|
def test_airline_excel_quotes_and_keeps_tms_id(self) -> None:
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="air1", text=wo, chat="wr_air_1", mid="af3"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
phase = handle_air_group(
|
|
self._file(sender="air1", filename="航线报价.pdf", chat="wr_air_1", mid="af4"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "airline_file_quote")
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.attachments[0]["file_name"], "航线报价.pdf")
|
|
self.assertEqual(ticket.quote.get("source_label"), "航线附件报价")
|
|
self.assertEqual(ticket.quote.get("quoteId"), "QT-AIR-1")
|
|
self.assertEqual(ticket.status, "已报价")
|
|
self.assertTrue(any(f"工单{wo}的1个附件已保存到后台工单" in t for t in self._texts()))
|
|
self.assertTrue(any("报价单已发送，请确认成交状态" in t for t in self._texts()))
|
|
|
|
def test_reactivate_after_airline_file_is_deal_not_blank_sheet(self) -> None:
|
|
"""航线附件已报价后，再发工单号回到成交跟进，不再要航线报价。"""
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="air1", text=wo, chat="wr_air_1", mid="af3b"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
handle_air_group(
|
|
self._file(sender="air1", filename="航线报价.pdf", chat="wr_air_1", mid="af4b"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
before = len(self.groups.sent)
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="af4c"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "activated_deal")
|
|
added = self.groups.sent[before:]
|
|
texts = "\n".join(x.get("content") or "" for x in added)
|
|
self.assertIn("请确认成交状态", texts)
|
|
self.assertFalse(any(x.get("kind") == "file" for x in added))
|
|
|
|
def test_reactivate_quoted_tms_file_is_deal_not_sheet(self) -> None:
|
|
"""TMS 价的报价单已经发出、停在成交跟进：再发工单号只回成交跟进。"""
|
|
wo = self._create_air(
|
|
quote={
|
|
"total": "CNY 3500.70",
|
|
"source": "TMS",
|
|
"source_label": "TMS 标准报价",
|
|
"quoteId": "QT-AIR-FILE",
|
|
"quoteFileIssued": True,
|
|
"airOptions": [
|
|
{"optionNo": "AIR-OPT-01", "airlineCode": "CZ", "flightNo": "CZ327"}
|
|
],
|
|
}
|
|
)
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="react-deal"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "activated_deal")
|
|
texts = "\n".join(self._texts())
|
|
self.assertIn("请确认成交状态", texts)
|
|
self.assertNotIn("询价字段：", texts)
|
|
self.assertFalse(any(x.get("kind") == "file" for x in self.groups.sent))
|
|
|
|
def test_image_rejected(self) -> None:
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="air1", text=wo, chat="wr_air_1", mid="af5"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
phase = handle_air_group(
|
|
InboundMessage(
|
|
sender_id="air1",
|
|
message_id="af6",
|
|
content="",
|
|
msg_type="image",
|
|
chat_id="wr_air_1",
|
|
chat_type="group",
|
|
media={"filename": "现场.jpg", "media_id": "mid-img"},
|
|
raw={"source": "aibot"},
|
|
),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "file_type_rejected")
|
|
self.assertFalse(self.ledger.get_ticket(work_order_no=wo).attachments)
|
|
|
|
def test_archive_file_without_at_quotes(self) -> None:
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="air1", text=wo, chat="wr_air_1", mid="ar0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
msg = self._file(sender="air1", filename="航线报价.xlsx", chat="wr_air_1", mid="ar1")
|
|
msg = InboundMessage(
|
|
sender_id=msg.sender_id,
|
|
message_id=msg.message_id,
|
|
content=msg.content,
|
|
msg_type="file",
|
|
chat_id=msg.chat_id,
|
|
chat_type="group",
|
|
media=dict(msg.media),
|
|
raw={"source": "wshoto_archive"},
|
|
)
|
|
import agent.handlers.air_group_collab as air_mod
|
|
orig = air_mod.get_air_text_flow
|
|
|
|
def _fake():
|
|
return self.flow
|
|
|
|
air_mod.get_air_text_flow = _fake # type: ignore
|
|
try:
|
|
name = dispatch_inbound(msg, RouteDecision(intent="ordinary_text_other"))
|
|
finally:
|
|
air_mod.get_air_text_flow = orig # type: ignore
|
|
self.assertEqual(name, "air_group_collab")
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.quote.get("source_label"), "航线附件报价")
|
|
self.assertTrue(any(f"工单{wo}的1个附件已保存到后台工单" in t for t in self._texts()))
|
|
self.assertTrue(any("成交" in t or "未成交" in t or "协商" in t for t in self._texts()))
|
|
|
|
def test_archive_text_ignored_until_at(self) -> None:
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="ar2"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
msg = self._msg(sender="sales1", text="成交", chat="wr_air_1", mid="ar3")
|
|
msg = InboundMessage(
|
|
sender_id=msg.sender_id,
|
|
message_id=msg.message_id,
|
|
content=msg.content,
|
|
chat_id=msg.chat_id,
|
|
chat_type="group",
|
|
raw={"source": "wshoto_archive"},
|
|
)
|
|
import agent.handlers.air_group_collab as air_mod
|
|
orig = air_mod.get_air_text_flow
|
|
|
|
def _fake():
|
|
return self.flow
|
|
|
|
air_mod.get_air_text_flow = _fake # type: ignore
|
|
try:
|
|
name = dispatch_inbound(msg, RouteDecision(intent="ordinary_text_other"))
|
|
finally:
|
|
air_mod.get_air_text_flow = orig # type: ignore
|
|
self.assertEqual(name, "air_archive_text_ignore")
|
|
self.assertEqual(self.ledger.get_ticket(work_order_no=wo).status, "已报价")
|
|
|
|
def test_archive_at_work_order_activates(self) -> None:
|
|
wo = self._create_air()
|
|
msg = InboundMessage(
|
|
sender_id="sales1",
|
|
message_id="ar-at-1",
|
|
content=f"@询价小助手 {wo}",
|
|
chat_id="wr_air_new",
|
|
chat_type="group",
|
|
raw={"source": "wshoto_archive"},
|
|
)
|
|
import agent.handlers.air_group_collab as air_mod
|
|
import agent.policy.air_text_flow as flow_mod
|
|
|
|
orig = air_mod.get_air_text_flow
|
|
orig2 = flow_mod.get_air_text_flow
|
|
|
|
def _fake():
|
|
return self.flow
|
|
|
|
air_mod.get_air_text_flow = _fake # type: ignore
|
|
flow_mod.get_air_text_flow = _fake # type: ignore
|
|
try:
|
|
name = dispatch_inbound(msg, RouteDecision(intent="ordinary_text_other"))
|
|
finally:
|
|
air_mod.get_air_text_flow = orig # type: ignore
|
|
flow_mod.get_air_text_flow = orig2 # type: ignore
|
|
self.assertEqual(name, "air_group_collab")
|
|
self.assertTrue(any(wo in t for t in self._texts()))
|
|
|
|
def test_file_with_work_order_activates(self) -> None:
|
|
wo = self._create_air()
|
|
phase = handle_air_group(
|
|
self._file(
|
|
sender="air1",
|
|
filename="航线报价.xlsx",
|
|
chat="wr_air_1",
|
|
mid="af7",
|
|
text=wo,
|
|
),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "airline_file_quote")
|
|
self.assertEqual(self.ledger.find_by_collab_chat(chat_id="wr_air_1").work_order_no, wo)
|
|
|
|
def test_archive_same_sentence_different_msgid_sends_one_sheet(self) -> None:
|
|
wo = self._create_air()
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_dup2", mid="bot-mid-012"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "activated")
|
|
self.assertEqual(sum(1 for x in self.groups.sent if x.get("kind") == "file"), 1)
|
|
phase = handle_air_group(
|
|
InboundMessage(
|
|
sender_id="sales1",
|
|
message_id="wshoto:archive-mid-999",
|
|
content=f"@询价小助手 {wo}",
|
|
chat_id="wr_air_dup2",
|
|
chat_type="group",
|
|
raw={"source": "wshoto_archive", "msgid": "archive-mid-999"},
|
|
),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "duplicate_inbound")
|
|
self.assertEqual(sum(1 for x in self.groups.sent if x.get("kind") == "file"), 1)
|
|
|
|
def test_aibot_and_archive_same_msgid_sends_one_sheet(self) -> None:
|
|
wo = self._create_air()
|
|
import agent.handlers.air_group_collab as air_mod
|
|
import agent.policy.air_text_flow as flow_mod
|
|
|
|
orig = air_mod.get_air_text_flow
|
|
orig2 = flow_mod.get_air_text_flow
|
|
|
|
def _fake():
|
|
return self.flow
|
|
|
|
air_mod.get_air_text_flow = _fake # type: ignore
|
|
flow_mod.get_air_text_flow = _fake # type: ignore
|
|
try:
|
|
aibot = self._msg(sender="sales1", text=wo, chat="wr_air_dup", mid="same-mid-011")
|
|
name = dispatch_inbound(aibot, RouteDecision(intent="ordinary_text_other"))
|
|
self.assertEqual(name, "air_group_collab")
|
|
files_after_bot = sum(1 for x in self.groups.sent if x.get("kind") == "file")
|
|
self.assertEqual(files_after_bot, 1)
|
|
archive = InboundMessage(
|
|
sender_id="sales1",
|
|
message_id="wshoto:same-mid-011",
|
|
content=f"@询价小助手 {wo}",
|
|
chat_id="wr_air_dup",
|
|
chat_type="group",
|
|
raw={"source": "wshoto_archive", "msgid": "same-mid-011"},
|
|
)
|
|
name = dispatch_inbound(archive, RouteDecision(intent="ordinary_text_other"))
|
|
self.assertEqual(name, "air_group_collab")
|
|
self.assertEqual(sum(1 for x in self.groups.sent if x.get("kind") == "file"), 1)
|
|
finally:
|
|
air_mod.get_air_text_flow = orig # type: ignore
|
|
flow_mod.get_air_text_flow = orig2 # type: ignore
|
|
|
|
def test_sea_ticket_rejected(self) -> None:
|
|
created = self.ledger.create_ticket(
|
|
sender_id="sales1",
|
|
business_line="SEA",
|
|
facts={"起运港": "上海", "目的港": "洛杉矶", "品名": "配件", "货量": "1x40HQ", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", "报价日期": "2026-09-16"},
|
|
)
|
|
wo = created["work_order_no"]
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_1", mid="sea1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.assertEqual(phase, "not_air")
|
|
|
|
def test_lock_fail_tells_operator_and_tms_reason(self) -> None:
|
|
"""锁舱失败群里要回操作人和 TMS 原因，不能被系统异常兜底盖掉。"""
|
|
wo = self._create_air()
|
|
self.ledger._cabin_lock_result = {
|
|
"ok": False,
|
|
"error": "lock_fail",
|
|
"failReason": "该航班日期不存在可锁定的配载记录",
|
|
}
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_lock_fail", mid="lf1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertEqual(phase, "system_exception")
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn(f"工单{wo} 空运锁舱失败。", joined)
|
|
self.assertIn("操作人：HangXian", joined)
|
|
self.assertIn("锁舱失败的原因：该航班日期不存在可锁定的配载记录", joined)
|
|
self.assertIn("已通知IT运维排查，请稍等。", joined)
|
|
self.assertNotIn("TMS接口异常，IT运维排查中", joined)
|
|
self.assertNotIn("报价选项不存在", joined)
|
|
self.assertNotIn("与工单不匹配", joined)
|
|
|
|
def test_lock_fail_does_not_block_airline_quote(self) -> None:
|
|
"""锁舱失败后，航线再报手工价必须收下，不能回 TMS 接口异常。"""
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="air1", text=wo, chat="wr_air_lock_quote", mid="lq0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
)
|
|
self.ledger._cabin_lock_result = {
|
|
"ok": False,
|
|
"error": "lock_fail",
|
|
"failReason": "该航班日期不存在可锁定的配载记录",
|
|
}
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_lock_quote", mid="lq1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertEqual(phase, "system_exception")
|
|
self.groups.sent.clear()
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text="空运费改为900USD", chat="wr_air_lock_quote", mid="lq2"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="product_quote",
|
|
)
|
|
self.assertNotEqual(phase, "system_exception_paused")
|
|
joined = "\n".join(self._texts())
|
|
self.assertNotIn("TMS接口异常", joined)
|
|
self.assertIn(phase, {"wait_tms_choice", "quoted"})
|
|
|
|
def test_lock_without_quote_still_shows_tms_reason(self) -> None:
|
|
"""TMS 拒绝锁舱时原样回失败原因，不要改写成选项不匹配。"""
|
|
wo = self._create_air(quote={}, status="询价中")
|
|
self.ledger._cabin_lock_result = {
|
|
"ok": False,
|
|
"error": "lock_fail",
|
|
"failReason": "该航班日期不存在可锁定的配载记录",
|
|
}
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_lock_no_quote", mid="lnq1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertEqual(phase, "system_exception")
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn("锁舱失败的原因：该航班日期不存在可锁定的配载记录", joined)
|
|
self.assertNotIn("报价选项不存在", joined)
|
|
self.assertNotIn("与工单不匹配", joined)
|
|
|
|
def test_paused_lock_retries_and_shows_tms_reason(self) -> None:
|
|
"""上次锁舱失败标了系统异常后，再锁仍要打 TMS，不能只回兜底。"""
|
|
wo = self._create_air()
|
|
self.ledger.mark_system_exception(
|
|
work_order_no=wo,
|
|
system_exception="TMS",
|
|
system_exception_reason="空运锁舱接口调用失败",
|
|
step="tms_lock",
|
|
payload={},
|
|
)
|
|
hits = {"n": 0}
|
|
orig = self.ledger.lock_cabin
|
|
|
|
def counted(**kwargs):
|
|
hits["n"] += 1
|
|
return orig(**kwargs)
|
|
|
|
self.ledger.lock_cabin = counted # type: ignore[method-assign]
|
|
self.ledger._cabin_lock_result = {
|
|
"ok": False,
|
|
"error": "lock_fail",
|
|
"failReason": "该航班日期不存在可锁定的配载记录",
|
|
}
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_lock_paused", mid="lp1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.assertNotEqual(phase, "system_exception_paused")
|
|
self.assertGreaterEqual(hits["n"], 1)
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn("锁舱失败的原因：该航班日期不存在可锁定的配载记录", joined)
|
|
self.assertNotIn("TMS接口异常，IT运维排查中", joined)
|
|
|
|
def test_paused_after_lock_still_records_lost_reason(self) -> None:
|
|
"""锁舱失败标了暂停后，未成交和原因仍要入账，不能回 TMS 异常。"""
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="sales1", text=wo, chat="wr_air_paused_lost", mid="pl0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_activate",
|
|
)
|
|
self.ledger.mark_system_exception(
|
|
work_order_no=wo,
|
|
system_exception="TMS",
|
|
system_exception_reason="空运锁舱接口调用失败",
|
|
step="tms_lock",
|
|
payload={},
|
|
)
|
|
self.groups.sent.clear()
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text="未成交", chat="wr_air_paused_lost", mid="pl1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="deal_lost",
|
|
)
|
|
self.assertEqual(phase, "wait_lost_reason")
|
|
self.assertNotIn("TMS接口异常", "\n".join(self._texts()))
|
|
self.groups.sent.clear()
|
|
phase = handle_air_group(
|
|
self._msg(sender="sales1", text="客户不需要了", chat="wr_air_paused_lost", mid="pl2"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="other",
|
|
)
|
|
self.assertEqual(phase, "done")
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn("已记下未成交原因：客户不需要了", joined)
|
|
self.assertNotIn("TMS接口异常", joined)
|
|
ticket = self.ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.lost_reason, "客户不需要了")
|
|
|
|
def test_release_fail_tells_operator_and_tms_reason(self) -> None:
|
|
"""释放失败同理：操作人、原因、已通知IT，不要系统异常兜底。"""
|
|
wo = self._create_air()
|
|
handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_rl_fail", mid="rf0"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_lock",
|
|
)
|
|
self.ledger._cabin_release_result = {
|
|
"ok": False,
|
|
"error": "release_fail",
|
|
"failReason": "舱位不存在",
|
|
}
|
|
phase = handle_air_group(
|
|
self._msg(sender="air1", text=f"{wo}释放舱位", chat="wr_air_rl_fail", mid="rf1"),
|
|
flow=self.flow,
|
|
ledger=self.ledger,
|
|
injected_intent="air_release",
|
|
)
|
|
self.assertEqual(phase, "system_exception")
|
|
joined = "\n".join(self._texts())
|
|
self.assertIn(f"工单{wo} 空运释放舱位失败。", joined)
|
|
self.assertIn("操作人：HangXian", joined)
|
|
self.assertIn("释放舱位失败的原因：舱位不存在", joined)
|
|
self.assertIn("已通知IT运维排查，请稍等。", joined)
|
|
self.assertNotIn("TMS接口异常，IT运维排查中", joined)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|