573 lines
23 KiB
Python
573 lines
23 KiB
Python
"""
|
|
入站选路：陆运书签/本句陆运交给 LandTextInquiryFlow；海运立即建单不变。
|
|
"""
|
|
|
|
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)
|
|
|
|
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, get_message_store
|
|
from agent.channel.wecom.models import InboundMessage
|
|
from agent.handlers.text_inquiry import handle_text_inquiry
|
|
from agent.policy.air_text_flow import reset_air_text_flow_for_test
|
|
from agent.policy.land_text_flow import get_land_text_flow, reset_land_text_flow_for_test
|
|
from agent.policy.sea_text_flow import get_sea_text_flow, reset_sea_text_flow_for_test
|
|
from agent.routing.decision import RouteDecision
|
|
from agent.routing.dispatch import dispatch_inbound
|
|
from tests.test_sea_text_inquiry import COMPLETE_SEA
|
|
|
|
|
|
class LandDispatchTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
reset_sea_text_flow_for_test()
|
|
reset_land_text_flow_for_test()
|
|
reset_air_text_flow_for_test()
|
|
self.store = MemoryMessageStore()
|
|
|
|
def _outbox_text(self) -> str:
|
|
return "\n".join(i.content for i in self.store._outbox.values())
|
|
|
|
def test_handler_routes_land_to_option_sheet(self) -> None:
|
|
msg = InboundMessage(sender_id="ld1", message_id="m-land-1", content="陆运询价")
|
|
phase = handle_text_inquiry(
|
|
msg,
|
|
injected_facts={"起运港": "广州", "目的港": "深圳", "品名": "衣服"},
|
|
injected_mode="LAND",
|
|
store=self.store,
|
|
)
|
|
self.assertEqual(phase, "need_land_options")
|
|
body = self._outbox_text()
|
|
self.assertIn("运输类型", body)
|
|
self.assertIn("线路类别", body)
|
|
self.assertIn("运输分类", body)
|
|
self.assertNotIn("请补充陆运类型：", body)
|
|
|
|
def test_air_owner_forwards_land_to_option_sheet(self) -> None:
|
|
from agent.policy.air_text_flow import get_air_text_flow
|
|
|
|
replies: list[str] = []
|
|
phase = get_air_text_flow().on_text(
|
|
sender_id="ld-air",
|
|
text="陆运询价",
|
|
reply=lambda t, extra=None: replies.append(t) or "",
|
|
injected_facts={"起运港": "广州", "目的港": "深圳", "品名": "衣服"},
|
|
injected_mode="LAND",
|
|
)
|
|
self.assertEqual(phase, "need_land_options")
|
|
text = "\n".join(replies)
|
|
self.assertIn("线路类别", text)
|
|
self.assertIn("运输分类", text)
|
|
self.assertNotIn("请补充陆运类型：", text)
|
|
|
|
def test_land_session_survives_confirm(self) -> None:
|
|
msg1 = InboundMessage(sender_id="ld2", message_id="m-l2a", content="陆运")
|
|
facts = {
|
|
"起运港": "广州",
|
|
"目的港": "深圳",
|
|
"品名": "衣服",
|
|
"体积": "100CBM",
|
|
"毛重": "50kg",
|
|
}
|
|
phase = handle_text_inquiry(
|
|
msg1,
|
|
injected_facts=facts,
|
|
injected_mode="LAND",
|
|
store=self.store,
|
|
)
|
|
# 补选项走同一 sender
|
|
msg2 = InboundMessage(
|
|
sender_id="ld2",
|
|
message_id="m-l2b",
|
|
content="国内运输拼车 国内长途/零担 急件",
|
|
)
|
|
phase = handle_text_inquiry(
|
|
msg2,
|
|
injected_facts=facts,
|
|
injected_mode="LAND",
|
|
store=self.store,
|
|
)
|
|
self.assertEqual(phase, "wait_confirm")
|
|
msg3 = InboundMessage(sender_id="ld2", message_id="m-l2c", content="确定")
|
|
phase = handle_text_inquiry(msg3, store=self.store)
|
|
self.assertEqual(phase, "wait_collab")
|
|
|
|
def test_confirm_misrouted_as_other_still_creates(self) -> None:
|
|
store = get_message_store()
|
|
facts = {
|
|
"起运港": "广州",
|
|
"目的港": "深圳",
|
|
"品名": "衣服",
|
|
"体积": "100CBM",
|
|
"毛重": "50kg",
|
|
}
|
|
handle_text_inquiry(
|
|
InboundMessage(sender_id="ld3", message_id="m-l3a", content="陆运"),
|
|
injected_facts=facts,
|
|
injected_mode="LAND",
|
|
store=store,
|
|
)
|
|
handle_text_inquiry(
|
|
InboundMessage(
|
|
sender_id="ld3",
|
|
message_id="m-l3b",
|
|
content="国内运输拼车 国内长途/零担 急件",
|
|
),
|
|
injected_facts=facts,
|
|
injected_mode="LAND",
|
|
store=store,
|
|
)
|
|
self.assertEqual(get_land_text_flow().session_of("ld3").phase, "wait_confirm")
|
|
name = dispatch_inbound(
|
|
InboundMessage(sender_id="ld3", message_id="m-l3c", content="确定"),
|
|
RouteDecision(intent="ordinary_text_other"),
|
|
)
|
|
self.assertEqual(name, "ordinary_text_inquiry")
|
|
sess = get_land_text_flow().session_of("ld3")
|
|
self.assertEqual(sess.phase, "wait_collab")
|
|
self.assertTrue((sess.work_order_no or "").startswith("WO"))
|
|
bodies = "\n".join(i.content for i in store._outbox.values())
|
|
self.assertIn(sess.work_order_no, bodies)
|
|
last = list(store._outbox.values())[-1].content
|
|
self.assertNotIn("你好，我是询价机器人", last)
|
|
|
|
def test_confirm_after_memory_lost_still_creates(self) -> None:
|
|
store = get_message_store()
|
|
facts = {
|
|
"起运港": "广州",
|
|
"目的港": "深圳",
|
|
"品名": "衣服",
|
|
"体积": "100CBM",
|
|
"毛重": "50kg",
|
|
}
|
|
handle_text_inquiry(
|
|
InboundMessage(sender_id="ld4", message_id="m-l4a", content="陆运"),
|
|
injected_facts=facts,
|
|
injected_mode="LAND",
|
|
store=store,
|
|
)
|
|
handle_text_inquiry(
|
|
InboundMessage(
|
|
sender_id="ld4",
|
|
message_id="m-l4b",
|
|
content="国内运输拼车 国内长途/零担 急件",
|
|
),
|
|
injected_facts=facts,
|
|
injected_mode="LAND",
|
|
store=store,
|
|
)
|
|
self.assertEqual(get_land_text_flow().session_of("ld4").phase, "wait_confirm")
|
|
reset_land_text_flow_for_test()
|
|
self.assertIsNone(get_land_text_flow().session_of("ld4"))
|
|
name = dispatch_inbound(
|
|
InboundMessage(sender_id="ld4", message_id="m-l4c", content="确定"),
|
|
RouteDecision(intent="ordinary_text_other"),
|
|
)
|
|
self.assertEqual(name, "ordinary_text_inquiry")
|
|
sess = get_land_text_flow().session_of("ld4")
|
|
self.assertEqual(sess.phase, "wait_collab")
|
|
self.assertTrue((sess.work_order_no or "").startswith("WO"))
|
|
last = list(store._outbox.values())[-1].content
|
|
self.assertNotIn("你好，我是询价机器人", last)
|
|
|
|
def test_tms_tech_fail_pauses_until_admin_retry(self) -> None:
|
|
from agent.policy.inquiry_copy import SYS_EXC_TMS_FALLBACK
|
|
from tests.test_land_text_inquiry import CountingLedger
|
|
|
|
class AlwaysFail(CountingLedger):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.query_calls = 0
|
|
|
|
def query_tms(self, **kwargs):
|
|
self.query_calls += 1
|
|
return {
|
|
"ok": False,
|
|
"classification": "TMS_TECH_FAILURE",
|
|
"has_price": False,
|
|
"quote": {},
|
|
}
|
|
|
|
land = reset_land_text_flow_for_test()
|
|
ledger = AlwaysFail()
|
|
land.ledger = ledger
|
|
get_sea_text_flow().ledger = ledger
|
|
from agent.policy.air_text_flow import get_air_text_flow
|
|
|
|
get_air_text_flow().ledger = ledger
|
|
facts = {
|
|
"起运港": "广州",
|
|
"目的港": "深圳",
|
|
"品名": "衣服",
|
|
"体积": "100CBM",
|
|
"毛重": "50kg",
|
|
}
|
|
handle_text_inquiry(
|
|
InboundMessage(sender_id="ld5", message_id="m-l5a", content="陆运"),
|
|
injected_facts=facts,
|
|
injected_mode="LAND",
|
|
store=self.store,
|
|
)
|
|
handle_text_inquiry(
|
|
InboundMessage(
|
|
sender_id="ld5",
|
|
message_id="m-l5b",
|
|
content="国内运输拼车 国内长途/零担 急件",
|
|
),
|
|
injected_facts=facts,
|
|
injected_mode="LAND",
|
|
store=self.store,
|
|
)
|
|
handle_text_inquiry(
|
|
InboundMessage(sender_id="ld5", message_id="m-l5c", content="确定"),
|
|
store=self.store,
|
|
)
|
|
self.assertGreaterEqual(ledger.query_calls, 3)
|
|
wo = ledger.last_created_no
|
|
ticket = ledger.get_ticket(work_order_no=wo)
|
|
self.assertEqual(ticket.system_exception, "TMS异常")
|
|
bodies = "\n".join(i.content for i in self.store._outbox.values())
|
|
self.assertIn(SYS_EXC_TMS_FALLBACK, bodies)
|
|
self.assertIsNone(get_land_text_flow().session_of("ld5"))
|
|
name = dispatch_inbound(
|
|
InboundMessage(sender_id="ld5", message_id="m-l5d", content="确定"),
|
|
RouteDecision(intent="ordinary_text_other"),
|
|
)
|
|
self.assertEqual(name, "ordinary_text_other")
|
|
sess = get_land_text_flow().session_of("ld5")
|
|
if sess is not None:
|
|
self.assertNotEqual(sess.phase, "wait_collab")
|
|
handle_text_inquiry(
|
|
InboundMessage(sender_id="ld5", message_id="m-l5e", content=f"看看 {wo}"),
|
|
store=self.store,
|
|
)
|
|
last = list(self.store._outbox.values())[-1].content
|
|
self.assertIn(SYS_EXC_TMS_FALLBACK, last)
|
|
|
|
def test_sea_holds_confirm_before_quote(self) -> None:
|
|
"""海运私聊字段齐了先出核对卡，回复确定后才查价。"""
|
|
from agent.policy import inquiry_copy as copy
|
|
|
|
msg = InboundMessage(sender_id="s-sea", message_id="m-sea-keep", content="海运询价")
|
|
phase = handle_text_inquiry(
|
|
msg,
|
|
injected_facts=COMPLETE_SEA,
|
|
injected_mode="SEA",
|
|
store=self.store,
|
|
)
|
|
self.assertEqual(phase, "wait_confirm")
|
|
held = "\n".join(item.content for item in self.store._outbox.values())
|
|
self.assertIn(copy.SEA_CONFIRM_TAIL, held)
|
|
self.assertNotIn("工单号", held)
|
|
phase = handle_text_inquiry(
|
|
InboundMessage(sender_id="s-sea", message_id="m-sea-ok", content="确定"),
|
|
store=self.store,
|
|
)
|
|
self.assertEqual(phase, "wait_collab")
|
|
|
|
def _land_to_quoted(self, sender_id: str) -> None:
|
|
store = get_message_store()
|
|
facts = {
|
|
"起运港": "广州",
|
|
"目的港": "深圳",
|
|
"品名": "衣服",
|
|
"体积": "100CBM",
|
|
"毛重": "50kg",
|
|
}
|
|
handle_text_inquiry(
|
|
InboundMessage(sender_id=sender_id, message_id=f"{sender_id}-a", content="陆运"),
|
|
injected_facts=facts,
|
|
injected_mode="LAND",
|
|
store=store,
|
|
)
|
|
handle_text_inquiry(
|
|
InboundMessage(
|
|
sender_id=sender_id,
|
|
message_id=f"{sender_id}-b",
|
|
content="国内运输拼车 国内长途/零担 急件",
|
|
),
|
|
injected_facts=facts,
|
|
injected_mode="LAND",
|
|
store=store,
|
|
)
|
|
handle_text_inquiry(
|
|
InboundMessage(sender_id=sender_id, message_id=f"{sender_id}-c", content="确定"),
|
|
store=store,
|
|
)
|
|
self.assertEqual(get_land_text_flow().session_of(sender_id).phase, "wait_collab")
|
|
|
|
def test_quoted_rise_percent_not_first_chat(self) -> None:
|
|
"""已出 TMS 卡后，A 判成调价就必须追问费用名，不能回开场白。"""
|
|
store = get_message_store()
|
|
self._land_to_quoted("ld-rise")
|
|
name = dispatch_inbound(
|
|
InboundMessage(sender_id="ld-rise", message_id="m-rise", content="报价上升10%"),
|
|
RouteDecision(intent="quote_adjust"),
|
|
)
|
|
self.assertEqual(name, "quote_adjust")
|
|
last = list(store._outbox.values())[-1].content
|
|
self.assertNotIn("你好，我是询价机器人", last)
|
|
self.assertIn("请说明要调哪一项费用", last)
|
|
|
|
def test_quoted_land_wins_over_stale_sea(self) -> None:
|
|
"""旧海运书签不能把陆运调价抢走。"""
|
|
store = get_message_store()
|
|
handle_text_inquiry(
|
|
InboundMessage(sender_id="ld-mix", message_id="m-mix-sea", content="海运"),
|
|
injected_facts=COMPLETE_SEA,
|
|
injected_mode="SEA",
|
|
store=store,
|
|
)
|
|
self._land_to_quoted("ld-mix")
|
|
name = dispatch_inbound(
|
|
InboundMessage(sender_id="ld-mix", message_id="m-mix-rise", content="报价上升10%"),
|
|
RouteDecision(intent="quote_adjust"),
|
|
)
|
|
self.assertEqual(name, "quote_adjust")
|
|
from agent.handlers.text_inquiry import _pick_text_flow
|
|
|
|
self.assertIs(
|
|
_pick_text_flow(sender_id="ld-mix", text="报价上升10%", prefer_quoted=True),
|
|
get_land_text_flow(),
|
|
)
|
|
last = list(store._outbox.values())[-1].content
|
|
self.assertNotIn("你好，我是询价机器人", last)
|
|
self.assertIn("请说明要调哪一项费用", last)
|
|
|
|
def test_sea_field_supplement_not_stolen_by_stale_land(self) -> None:
|
|
"""
|
|
旧陆运选项草稿不能抢走海运补问。
|
|
|
|
测服：先陆运缺选项，再发海运缺贸易条款/运输分类，下一句只补这两项，
|
|
不能回陆运「参数提取不全」或填写样例图。
|
|
"""
|
|
store = get_message_store()
|
|
handle_text_inquiry(
|
|
InboundMessage(sender_id="mix-sea", message_id="m-land-old", content="陆运询价"),
|
|
injected_facts={"起运港": "广州", "目的港": "深圳", "品名": "衣服"},
|
|
injected_mode="LAND",
|
|
store=store,
|
|
)
|
|
self.assertEqual(get_land_text_flow().session_of("mix-sea").phase, "need_land_options")
|
|
incomplete = {
|
|
"起运港": "上海",
|
|
"目的港": "洛杉矶",
|
|
"品名": "普货",
|
|
"货量": "20吨",
|
|
"整柜或拼柜": "整柜",
|
|
"箱型箱量": "1x40HQ",
|
|
}
|
|
phase = handle_text_inquiry(
|
|
InboundMessage(sender_id="mix-sea", message_id="m-sea-ask", content="海运询价"),
|
|
injected_facts=incomplete,
|
|
injected_mode="SEA",
|
|
store=store,
|
|
)
|
|
self.assertEqual(phase, "clarify")
|
|
self.assertEqual(get_sea_text_flow().session_of("mix-sea").phase, "clarify")
|
|
self.assertIsNone(get_land_text_flow().session_of("mix-sea"))
|
|
before = len(store._outbox)
|
|
phase = handle_text_inquiry(
|
|
InboundMessage(
|
|
sender_id="mix-sea",
|
|
message_id="m-sea-fill",
|
|
content="贸易条款：FOB 运输分类：其他",
|
|
),
|
|
injected_facts={"贸易条款": "FOB", "运输分类": "其他"},
|
|
store=store,
|
|
)
|
|
self.assertNotEqual(phase, "need_land_options")
|
|
self.assertEqual(get_sea_text_flow().session_of("mix-sea").business_line, "SEA")
|
|
added = "\n".join(i.content for i in list(store._outbox.values())[before:])
|
|
self.assertNotIn("参数提取不全", added)
|
|
self.assertNotIn("填写样例", added)
|
|
self.assertNotIn("运输类型", added)
|
|
|
|
def test_clarify_short_reply_misrouted_as_other_stays_on_inquiry(self) -> None:
|
|
"""补问阶段回「111」即使被标闲聊，也不能发开场白。"""
|
|
store = get_message_store()
|
|
facts = {
|
|
"起运港": "广州",
|
|
"目的港": "深圳",
|
|
"运输类型": "国内运输拼车",
|
|
"线路类别": "国内长途/零担",
|
|
"运输分类": "急件",
|
|
}
|
|
phase = handle_text_inquiry(
|
|
InboundMessage(sender_id="ld-111", message_id="m-111a", content="陆运广州到深圳"),
|
|
injected_facts=facts,
|
|
injected_mode="LAND",
|
|
store=store,
|
|
)
|
|
self.assertEqual(phase, "clarify")
|
|
name = dispatch_inbound(
|
|
InboundMessage(sender_id="ld-111", message_id="m-111b", content="111"),
|
|
RouteDecision(intent="ordinary_text_other"),
|
|
)
|
|
self.assertEqual(name, "ordinary_text_inquiry")
|
|
sess = get_land_text_flow().session_of("ld-111")
|
|
self.assertIsNotNone(sess)
|
|
self.assertNotEqual(sess.phase, "")
|
|
last = list(store._outbox.values())[-1].content
|
|
self.assertNotIn("你好，我是询价机器人", last)
|
|
self.assertNotIn("请直接描述客户需求", last)
|
|
|
|
def test_lost_reason_not_taken_as_new_inquiry(self) -> None:
|
|
"""未成交后下一句是原因。旧报价单被判成调价时，不能改问运输方式。"""
|
|
from agent.policy.air_text_flow import FlowSession
|
|
|
|
sea = get_sea_text_flow()
|
|
created = sea._ledger.create_ticket(
|
|
sender_id="lost-1",
|
|
business_line="SEA",
|
|
facts={"起运港": "广州港", "目的港": "巴生港", "品名": "机器设备"},
|
|
)
|
|
wo = str(created.get("work_order_no") or "")
|
|
sea._ledger.transition(work_order_no=wo, to_status="未成交")
|
|
sea._save(
|
|
FlowSession(
|
|
sender_id="lost-1",
|
|
thread_id="lost-1",
|
|
phase="wait_lost_reason",
|
|
business_line="SEA",
|
|
work_order_no=wo,
|
|
status="未成交",
|
|
wait_version=2,
|
|
)
|
|
)
|
|
get_land_text_flow()._save(
|
|
FlowSession(
|
|
sender_id="lost-1",
|
|
thread_id="lost-1",
|
|
phase="wait_deal",
|
|
business_line="LAND",
|
|
work_order_no="WO202609240045",
|
|
status="已报价",
|
|
quote={"total": "CNY 1"},
|
|
)
|
|
)
|
|
phase = handle_text_inquiry(
|
|
InboundMessage(sender_id="lost-1", message_id="m-lost", content="客户不需要了"),
|
|
prefer_quoted=True,
|
|
store=self.store,
|
|
)
|
|
self.assertEqual(phase, "done")
|
|
body = self._outbox_text()
|
|
self.assertIn("已记下未成交原因：客户不需要了", body)
|
|
self.assertNotIn("告知我询价的运输方式", body)
|
|
|
|
def test_several_turns_before_mode_follow_the_chosen_line(self) -> None:
|
|
"""定运输方式前连续几句询价字段都要留下，选定后再整份带上。不只认上一句。"""
|
|
from agent.policy.air_text_flow import FlowSession, get_air_text_flow
|
|
|
|
turns = (
|
|
(
|
|
"a",
|
|
"10月1日，深圳-吉隆坡，散货、普货报价",
|
|
{
|
|
"起运港": "深圳",
|
|
"目的港": "吉隆坡",
|
|
"品名": "普货",
|
|
"包装方式": "散货",
|
|
"报价日期": "2026-10-01",
|
|
},
|
|
),
|
|
(
|
|
"b",
|
|
"10件，毛重100公斤，体积2方",
|
|
{"件数": "10", "毛重": "100", "体积": "2"},
|
|
),
|
|
)
|
|
for label, reply, mode in (
|
|
("空运", "3", "AIR"),
|
|
("海运", "海运", "SEA"),
|
|
("陆运", "1", "LAND"),
|
|
):
|
|
reset_sea_text_flow_for_test()
|
|
reset_land_text_flow_for_test()
|
|
reset_air_text_flow_for_test()
|
|
sid = "premode-" + mode
|
|
# 上一单还在海运，新的几句里没有运输方式，不能把这些字段丢掉。
|
|
get_sea_text_flow()._save(
|
|
FlowSession(
|
|
sender_id=sid,
|
|
thread_id=sid,
|
|
phase="tms_miss",
|
|
business_line="SEA",
|
|
work_order_no="WO-OLD",
|
|
facts={"目的港": "旧港口"},
|
|
)
|
|
)
|
|
for mid, content, facts in turns:
|
|
phase = handle_text_inquiry(
|
|
InboundMessage(sender_id=sid, message_id=sid + mid, content=content),
|
|
injected_facts=facts,
|
|
injected_mode="",
|
|
store=self.store,
|
|
)
|
|
self.assertEqual(phase, "need_mode", label)
|
|
waiting = get_sea_text_flow().session_of(sid)
|
|
self.assertEqual(waiting.facts.get("目的港"), "吉隆坡", label)
|
|
self.assertIn("10", waiting.facts.get("件数") or "", label)
|
|
self.assertIn("深圳-吉隆坡", waiting.immutable_text)
|
|
self.assertIn("10件", waiting.immutable_text)
|
|
before = len(self.store._outbox)
|
|
phase = handle_text_inquiry(
|
|
InboundMessage(sender_id=sid, message_id=sid + "mode", content=reply),
|
|
store=self.store,
|
|
)
|
|
self.assertNotEqual(phase, "need_mode", label)
|
|
added = "\n".join(i.content for i in list(self.store._outbox.values())[before:])
|
|
owner = {
|
|
"AIR": get_air_text_flow(),
|
|
"SEA": get_sea_text_flow(),
|
|
"LAND": get_land_text_flow(),
|
|
}[mode]
|
|
sess = owner.session_of(sid)
|
|
self.assertIsNotNone(sess, label)
|
|
self.assertEqual((sess.business_line or "").upper(), mode, label)
|
|
self.assertEqual(sess.facts.get("起运港"), "深圳", label)
|
|
self.assertEqual(sess.facts.get("目的港"), "吉隆坡", label)
|
|
self.assertEqual(sess.facts.get("品名"), "普货", label)
|
|
self.assertEqual(sess.facts.get("报价日期"), "2026-10-01", label)
|
|
self.assertNotEqual(sess.facts.get("目的港"), "旧港口", label)
|
|
if mode == "AIR":
|
|
self.assertIn("10", sess.facts.get("件数") or "", label)
|
|
self.assertNotIn("整柜或拼柜", sess.facts, label)
|
|
self.assertNotIn("箱型箱量", sess.facts, label)
|
|
elif mode == "SEA":
|
|
self.assertIn("10", sess.facts.get("货量") or "", label)
|
|
self.assertNotIn("件数", sess.facts, label)
|
|
self.assertNotIn("毛重", sess.facts, label)
|
|
self.assertNotIn("包装方式", sess.facts, label)
|
|
self.assertIn("货物数量", added)
|
|
self.assertIn("整柜或拼柜", added)
|
|
self.assertIn("箱型箱量", added)
|
|
self.assertIn("贸易条款", added)
|
|
self.assertNotIn("件数（必填）", added)
|
|
else:
|
|
self.assertIn("100", sess.facts.get("毛重") or "", label)
|
|
self.assertIn("2", sess.facts.get("体积") or "", label)
|
|
self.assertNotIn("件数", sess.facts, label)
|
|
self.assertNotIn("箱型箱量", sess.facts, label)
|
|
self.assertNotIn("贸易条款", sess.facts, label)
|
|
self.assertIn("运输类型", added)
|
|
self.assertNotIn("件数（必填）", added)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|