""" 海运文字询价主路径单测（不连企微 / 真实 TMS / 8180）。 覆盖：海运补问、询价确认卡、有价/无价卡、跳过协同、拉群、群里静默补协同字段。 空运话术不得出现在海运路径。 """ 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.wecom.models import InboundMessage from agent.handlers.group_collab import handle_group_collab 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.sea_text_flow import SeaTextInquiryFlow, reset_sea_text_flow_for_test from agent.schema.field_validate import validate_required_fields from agent.schema.tms_sea_query import assemble_sea_query COMPLETE_SEA = { "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "货量": "20吨", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", } class FakeGroupClient: """单测假企微建群：不打外网。""" def __init__(self) -> None: self.created: list[dict] = [] self.sent: list[dict] = [] self.fail = False self.next_chat_id = "wr_sea_1" def create_group(self, *, name: str, userids: list[str]) -> dict: if self.fail: return {"ok": False, "error": "wecom_create_fail", "chat_id": ""} self.created.append({"name": name, "userids": list(userids)}) return {"ok": True, "chat_id": self.next_chat_id, "error": ""} def add_group_members(self, *, chat_id: str, userids: list[str]) -> dict: self.created.append({"name": "", "userids": list(userids), "chat_id": chat_id, "add": True}) return {"ok": True} def send_group(self, *, chat_id: str, content: str) -> dict: self.sent.append({"chat_id": chat_id, "content": content}) return {"ok": True} class SeaTextInquiryTests(unittest.TestCase): def setUp(self) -> None: self.ledger = MemoryLedger() self.ledger.set_quote_templates( [ { "templateId": "mem-default-sea", "name": "海运默认", "bizType": "海运", "keywords": [], "isDefault": True, } ] ) self.groups = FakeGroupClient() self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups) 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_sea_pieces_fill_cargo_qty(self) -> None: facts = { "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "件数": "15件", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", } check = validate_required_fields(facts=facts, business_line="SEA") self.assertTrue(check["ok"]) self.assertEqual(check["facts"]["货量"], "15件") self.assertNotIn("货量", check["missing"]) def test_sea_missing_container_only_asks_container(self) -> None: facts = { "起运港": "上海", "目的港": "洛杉矶", "品名": "茶叶", "货量": "20吨", "整柜或拼柜": "整柜", } check = validate_required_fields(facts=facts, business_line="SEA") self.assertFalse(check["ok"]) self.assertEqual(check["missing"], ["箱型箱量"]) self.assertIn("箱型箱量", check["missing_display"]) self.assertNotIn("件数", check["missing"]) self.assertNotIn("毛重", check["missing"]) text = copy.ask_clarify( facts=check["facts"], missing_keys=check["missing"], transport_mode="海运", ) self.assertIn("运输方式：海运", text) self.assertIn("起运港：上海", text) self.assertIn("货物数量：20吨", text) self.assertIn("箱型箱量：(如", text) self.assertNotIn("件数", text) self.assertNotIn("重量", text) self.assertNotIn("报价日期：(如", text) self.assertIn("报价日期：", text) def test_complete_sea_inquiry_card_uses_seven_fields(self) -> None: phase = self.flow.on_text( sender_id="s1", text="海运完整", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.assertEqual(phase, "wait_collab") joined = "\n".join(self.replies) self.assertIn("运输方式：海运", joined) self.assertIn("起运港：上海", joined) self.assertIn("目的港：洛杉矶", joined) self.assertIn("货物数量：20吨", joined) self.assertIn("整柜或拼柜：整柜", joined) self.assertIn("箱型箱量：1x40HQ", joined) self.assertIn("正在查询TMS报价", joined) self.assertNotIn("件数：", joined.split("正在查询TMS报价")[0]) self.assertNotIn("本轮先开通空运", joined) sess = self.flow.session_of("s1") self.assertEqual(sess.business_line, "SEA") self.assertEqual(sess.status, "已报价") def test_handler_routes_sea_not_air_deferred(self) -> None: reset_sea_text_flow_for_test() from agent.channel.queue import MemoryMessageStore store = MemoryMessageStore() msg = InboundMessage( sender_id="s2", message_id="m-sea-1", content="海运询价", ) phase = handle_text_inquiry( msg, injected_facts=COMPLETE_SEA, injected_mode="SEA", flow=self.flow, store=store, ) self.assertEqual(phase, "wait_collab") self.assertNotEqual(phase, "sea_deferred") class SeaTmsCardTests(unittest.TestCase): def setUp(self) -> None: self.ledger = MemoryLedger() self.groups = FakeGroupClient() self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups) 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_assemble_sea_query_does_not_need_air_measures(self) -> None: ready = assemble_sea_query(COMPLETE_SEA) self.assertTrue(ready["ok"]) self.assertEqual(ready["missing"], []) self.assertNotIn("件数", ready.get("missing") or []) def test_long_fee_lines_go_to_text_not_clipped_card(self) -> None: """海运多项费用：完整明细必须在文本里，卡片引用条不得半截。""" long_quote = { "total": "2287.65", "currency": "CNY", "eta": "1天", "valid_until": "2026-09-30", "source_label": "TMS 标准报价", "fee_lines": [ {"name": "海运费", "amount": "100.00", "currency": "USD"}, {"name": "燃油附加费", "amount": "15.00", "currency": "USD"}, {"name": "操作费", "amount": "680.00", "currency": "CNY"}, {"name": "文件费", "amount": "450.00", "currency": "CNY"}, {"name": "封条费（SEAI）", "amount": "50.00", "currency": "CNY"}, {"name": "设备费", "amount": "50.00", "currency": "CNY"}, {"name": "报关费", "amount": "250.00", "currency": "CNY"}, ], } text = copy.sea_quote_detail_text( work_order_no="WO202609150029", quote=long_quote ) self.assertIn("海运费：USD 100.00", text) self.assertIn("报关费：CNY 250.00", text) self.assertIn("总销售价合计：CNY 2287.65", text) card_text = copy.quote_area_safe_text(long_quote) self.assertIn("总销售价合计：CNY 2287.65", card_text) self.assertNotIn("海运费：USD 100.00\n燃油附加费", card_text) self.assertLessEqual(len(card_text), 80) payload = copy.sea_tms_hit_wecom_payload( work_order_no="WO202609150029", quote=long_quote ) self.assertEqual( payload["template_card"]["quote_area"]["quote_text"], card_text ) def test_sea_hit_card_has_price_and_two_buttons(self) -> None: phase = self.flow.on_text( sender_id="s1", text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.assertEqual(phase, "wait_collab") joined = "\n".join(self.replies) self.assertIn("价格明细", joined) self.assertIn("是否拉产品人员进群协同", joined) self.assertIn(copy.BTN_PULL_COLLAB, joined) self.assertIn(copy.BTN_SKIP_COLLAB, joined) self.assertNotIn("请手动拉相关人员", joined) self.assertNotIn("若需要咨询内部航线人员", joined) cards = [x for x in self.extras if x.get("msgtype") == "template_card"] self.assertTrue(cards) buttons = [b["text"] for b in cards[-1]["template_card"]["button_list"]] self.assertEqual(buttons, [copy.BTN_PULL_COLLAB, copy.BTN_SKIP_COLLAB]) def test_sea_miss_one_card_with_pull_button(self) -> None: facts = dict(COMPLETE_SEA) facts["目的港"] = "无价港" phase = self.flow.on_text( sender_id="s1", text="海运无价", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.assertEqual(phase, "tms_miss") sess = self.flow.session_of("s1") self.assertEqual(sess.status, "询价中") joined = "\n".join(self.replies) self.assertIn("暂无匹配报价", joined) self.assertIn("询价中", joined) self.assertIn("未查到合适报价", joined) self.assertIn("是否需要拉内部产品人员进群", joined) self.assertIn(copy.BTN_PULL_COLLAB, joined) self.assertNotIn("请手动拉相关人员和智能体到原有群", joined) cards = [x for x in self.extras if x.get("msgtype") == "template_card"] self.assertTrue(cards) buttons = [b["text"] for b in cards[-1]["template_card"].get("button_list") or []] self.assertEqual(buttons, [copy.BTN_PULL_COLLAB]) def test_sea_tech_fail_is_not_miss(self) -> None: facts = dict(COMPLETE_SEA) facts["tms_force"] = "tech" phase = self.flow.on_text( sender_id="s1", text="海运技术失败", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.assertEqual(phase, "tms_tech") self.assertIn("TMS 查询失败", "\n".join(self.replies)) self.assertNotIn(copy.BTN_PULL_COLLAB, "\n".join(self.replies)) class SeaSkipAndPullTests(unittest.TestCase): def setUp(self) -> None: os.environ["WECOM_ARCHIVE_SEAT_USER_IDS"] = "" from agent.config import get_settings get_settings.cache_clear() self.ledger = MemoryLedger() self.ledger.set_quote_templates( [ { "templateId": "mem-default-sea", "name": "海运默认", "bizType": "海运", "keywords": [], "isDefault": True, } ] ) self.groups = FakeGroupClient() self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups) self.replies: list[str] = [] self.extras: list[dict] = [] def _reply(self, text: str, extra=None) -> str: self.replies.append(text) self.extras.append(dict(extra) if extra else {}) return "" def _quote_first(self, sender: str = "s1") -> None: self.flow.on_text( sender_id=sender, text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) def test_skip_collab_reuses_air_scheme(self) -> None: self._quote_first() phase = self.flow.on_button( sender_id="s1", action=copy.BTN_SKIP_COLLAB, reply=self._reply, card_meta={"response_code": "rc-sea-1"}, ) self.assertEqual(phase, "wait_adopt") self.assertIn("已跳过内部协同", "\n".join(self.replies)) scheme = "\n".join(x for x in self.replies if "报价方案" in x) self.assertIn("运输方式：海运", scheme) self.assertIn("货物数量：", scheme) self.assertIn("整柜或拼柜：", scheme) self.assertIn("箱型箱量：", scheme) self.assertNotIn("件数：", scheme) self.assertNotIn("重量：", scheme) self.assertNotIn("体积：", scheme) self.assertNotIn("包装类型：", scheme) cards = [ x.get("template_card") or {} for x in self.extras if (x.get("template_card") or {}).get("main_title", {}).get("title", "").endswith("报价方案") ] self.assertTrue(cards) rows = {r["keyname"]: r["value"] for r in cards[0].get("horizontal_content_list") or []} self.assertIn("海运", rows.get("航线", "")) self.assertIn("箱型", rows) self.assertNotIn("包装", rows) again = self.flow.on_button( sender_id="s1", action=copy.BTN_SKIP_COLLAB, reply=self._reply ) self.assertEqual(again, "card_discard") def test_pull_group_success_greys_buttons(self) -> None: self.ledger.set_sea_staff( [ { "name": "张产品", "wecomId": "prod1", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", } ] ) self._quote_first() wo = self.flow.session_of("s1").work_order_no phase = self.flow.on_button( sender_id="s1", action=copy.BTN_PULL_COLLAB, reply=self._reply, card_meta={"response_code": "rc-pull-1"}, ) self.assertEqual(phase, "collab_group") joined = "\n".join(self.replies) self.assertIn("协同群已创建", joined) self.assertIn(wo, joined) self.assertIn("销售", joined) self.assertIn("产品张产品", joined) self.assertEqual(self.groups.created[0]["name"], wo) self.assertIn("prod1", self.groups.created[0]["userids"]) self.assertIn("s1", self.groups.created[0]["userids"]) self.assertNotIn("AI001", self.groups.created[0]["userids"]) self.assertTrue(self.groups.sent) group_text = self.groups.sent[0]["content"] self.assertIn(wo, group_text) self.assertIn("起运港", group_text) self.assertIn("贸易条款", group_text) self.assertIn("TMS", group_text) sess = self.flow.session_of("s1") self.assertNotIn(copy.BTN_PULL_COLLAB, sess.allowed) self.assertNotIn(copy.BTN_SKIP_COLLAB, sess.allowed) again = self.flow.on_button( sender_id="s1", action=copy.BTN_SKIP_COLLAB, reply=self._reply ) self.assertEqual(again, "card_discard") def test_pull_group_includes_archive_bot(self) -> None: self.ledger.upsert_staff( { "name": "询价机器人", "wecomId": "AI001", "roleCode": "sales", "status": "active", } ) self.ledger.set_sea_staff( [ { "name": "张产品", "wecomId": "prod1", "roleCode": "sea", "routes": "海运全线", "status": "active", } ] ) self._quote_first() self.flow.on_button( sender_id="s1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) userids = self.groups.created[0]["userids"] self.assertIn("AI001", userids) self.assertIn("s1", userids) self.assertIn("prod1", userids) joined = "\n".join(self.replies) self.assertIn("产品张产品", joined) self.assertNotIn("产品询价机器人", joined) self.assertNotIn("销售询价机器人", joined) def test_pull_no_staff_keeps_button(self) -> None: self.ledger.set_sea_staff([]) self._quote_first() phase = self.flow.on_button( sender_id="s1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) self.assertEqual(phase, "wait_collab") self.assertIn("没有可拉的产品人员", "\n".join(self.replies)) self.assertEqual(self.groups.created, []) self.assertIn(copy.BTN_PULL_COLLAB, self.flow.session_of("s1").allowed) def test_pull_already_has_group_no_second(self) -> None: self.ledger.set_sea_staff( [ { "name": "李产品", "wecomId": "prod2", "roleCode": "sea", "routes": "上海", "status": "active", } ] ) self._quote_first() self.flow.on_button( sender_id="s1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) n = len(self.groups.created) self.flow.session_of("s1").allowed = (copy.BTN_PULL_COLLAB,) phase = self.flow.on_button( sender_id="s1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) self.assertEqual(phase, "collab_group") self.assertEqual(len(self.groups.created), n) self.assertIn("群已在", "\n".join(self.replies)) def test_miss_group_message_writes_no_quote(self) -> None: self.ledger.set_sea_staff( [ { "name": "王产品", "wecomId": "prod3", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", } ] ) facts = dict(COMPLETE_SEA) facts["目的港"] = "无价港" self.flow.on_text( sender_id="s1", text="海运无价", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.flow.on_button( sender_id="s1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) self.assertTrue(self.groups.sent) self.assertIn("暂无匹配报价", self.groups.sent[0]["content"]) class SeaGroupCollabFieldTests(unittest.TestCase): def setUp(self) -> None: self.ledger = MemoryLedger() self.groups = FakeGroupClient() self.flow = SeaTextInquiryFlow(self.ledger, group_client=self.groups) self.replies: list[str] = [] def _reply(self, text: str, extra=None) -> None: self.replies.append(text) def test_group_text_writes_incoterm_silently(self) -> None: self.ledger.set_sea_staff( [ { "name": "赵产品", "wecomId": "prod4", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", } ] ) self.flow.on_text( sender_id="sales1", text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.flow.on_button( sender_id="sales1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) wo = self.flow.session_of("sales1").work_order_no ticket = self.ledger.get_ticket(work_order_no=wo) chat_id = ticket.collab_chat_id n = len(self.replies) phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="g1", content="贸易条款：FOB", chat_id=chat_id, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "collab_field_saved") self.assertEqual(len(self.replies), n) self.assertEqual(ticket.collab_facts.get("贸易条款"), "FOB") def test_group_price_chat_is_ignored(self) -> None: self.ledger.set_sea_staff( [ { "name": "赵产品", "wecomId": "prod4", "roleCode": "sea", "routes": "海运·全部线路", "status": "active", } ] ) self.flow.on_text( sender_id="sales1", text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.flow.on_button( sender_id="sales1", action=copy.BTN_PULL_COLLAB, reply=self._reply, ) wo = self.flow.session_of("sales1").work_order_no ticket = self.ledger.get_ticket(work_order_no=wo) phase = handle_group_collab( InboundMessage( sender_id="prod4", message_id="g2", content="报价 1200 美元可以做", chat_id=ticket.collab_chat_id, chat_type="group", ), flow=self.flow, ledger=self.ledger, ) self.assertEqual(phase, "group_ignore") self.assertFalse(ticket.quote_version > 1 and "1200" in str(ticket.quote)) class SeaContinueTests(unittest.TestCase): def setUp(self) -> None: from agent.redis_coord.runtime import shutdown_redis_runtime shutdown_redis_runtime() self.ledger = MemoryLedger() self.ledger.set_quote_templates( [ { "templateId": "mem-default-sea", "name": "海运默认", "bizType": "海运", "keywords": [], "isDefault": True, } ] ) self.flow = SeaTextInquiryFlow(self.ledger) self.replies: list[str] = [] self.extras: list[dict] = [] def _reply(self, text: str, extra=None) -> str: self.replies.append(text) self.extras.append(dict(extra) if extra else {}) return "" def test_continue_after_skip_replays_sea_scheme_not_air(self) -> None: """已跳过协同后再续办：必须海运方案卡，不能套空运件数包装。""" from unittest.mock import patch from agent.channel.queue import MemoryMessageStore from agent.handlers.continue_thread import handle_continue_thread from agent.policy.air_text_flow import AirTextInquiryFlow self.flow.on_text( sender_id="c1", text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.flow.on_button( sender_id="c1", action=copy.BTN_SKIP_COLLAB, reply=self._reply, card_meta={"response_code": "rc-cont-1"}, ) wo = self.flow.session_of("c1").work_order_no store = MemoryMessageStore() msg = InboundMessage( sender_id="c1", message_id="m-cont-sea", content=f"继续工单{wo}", ) air_trap = AirTextInquiryFlow(MemoryLedger()) with patch( "agent.handlers.continue_thread.get_sea_text_flow", return_value=self.flow ), patch( "agent.handlers.continue_thread.get_air_text_flow", return_value=air_trap ): phase = handle_continue_thread(msg, thread_id=wo, store=store) self.assertEqual(phase, "wait_adopt") cards = [ (item.payload or {}).get("template_card") or {} for item in store._outbox.values() ] titles = [ (c.get("main_title") or {}).get("title") or "" for c in cards if c ] self.assertTrue(any(t == f"{wo}报价方案" for t in titles)) rows = {} for card in cards: if (card.get("main_title") or {}).get("title") == f"{wo}报价方案": rows = { r["keyname"]: r["value"] for r in card.get("horizontal_content_list") or [] } self.assertIn("海运", rows.get("航线", "")) self.assertIn("箱型", rows) self.assertNotIn("包装", rows) def test_continue_from_ledger_uses_sea_collab_card(self) -> None: """进程无书签：主账海运已报价应回收海运协同卡，不是空运卡。""" from unittest.mock import patch from agent.channel.queue import MemoryMessageStore from agent.handlers.continue_thread import handle_continue_thread from agent.policy.air_text_flow import AirTextInquiryFlow created = self.ledger.create_ticket( sender_id="c2", business_line="SEA", facts=COMPLETE_SEA ) wo = created["work_order_no"] self.ledger.upsert_quote( work_order_no=wo, quote={ "total": "USD 1,200", "eta": "18天", "valid_until": "2026-09-20", "source_label": "TMS 标准报价", }, to_status="已报价", ) empty = SeaTextInquiryFlow(self.ledger) store = MemoryMessageStore() msg = InboundMessage( sender_id="c2", message_id="m-cont-ledger", content=f"继续工单{wo}", ) air_trap = AirTextInquiryFlow(MemoryLedger()) with patch( "agent.handlers.continue_thread.get_sea_text_flow", return_value=empty ), patch( "agent.handlers.continue_thread.get_air_text_flow", return_value=air_trap ): phase = handle_continue_thread(msg, thread_id=wo, store=store) self.assertEqual(phase, "wait_collab") sess = empty.session_by_ticket("c2", wo) self.assertIsNotNone(sess) self.assertEqual(sess.business_line, "SEA") joined = "\n".join(item.content for item in store._outbox.values()) self.assertIn(copy.BTN_PULL_COLLAB, joined) self.assertIn("是否拉产品人员进群协同", joined) self.assertNotIn("包装方式", joined) def test_continue_after_quote_file_returns_deal(self) -> None: """报价单已发出后再续办：必须成交跟进，不能退回生成报价单。""" from unittest.mock import patch from agent.channel.queue import MemoryMessageStore from agent.handlers.continue_thread import handle_continue_thread from agent.policy.air_text_flow import AirTextInquiryFlow self.flow.on_text( sender_id="c3", text="海运有价", reply=self._reply, injected_facts=COMPLETE_SEA, injected_mode="SEA", ) self.flow.on_button( sender_id="c3", action=copy.BTN_SKIP_COLLAB, reply=self._reply, card_meta={"response_code": "rc-deal-1"}, ) wo = self.flow.session_of("c3").work_order_no self.flow.on_button( sender_id="c3", action=copy.BTN_ADOPT_EXCEL, reply=self._reply, card_meta={"event_key": f"adopt_excel:{wo}", "response_code": "rc-xl"}, ) self.assertEqual(self.flow.session_of("c3").phase, "wait_deal") empty = SeaTextInquiryFlow(self.ledger) store = MemoryMessageStore() msg = InboundMessage( sender_id="c3", message_id="m-cont-deal", content=f"继续工单{wo}", ) air_trap = AirTextInquiryFlow(MemoryLedger()) with patch( "agent.handlers.continue_thread.get_sea_text_flow", return_value=empty ), patch( "agent.handlers.continue_thread.get_air_text_flow", return_value=air_trap ): phase = handle_continue_thread(msg, thread_id=wo, store=store) self.assertEqual(phase, "wait_deal") joined = "\n".join(item.content for item in store._outbox.values()) self.assertTrue(any(x in joined for x in (copy.BTN_DEAL, "成交"))) self.assertNotIn(copy.BTN_SKIP_COLLAB, joined) class SeaGroupCopyTests(unittest.TestCase): def test_scheme_card_uses_sea_fields(self) -> None: facts = { "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "货量": "15件", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", "报价日期": "2026-09-15", } quote = {"total": "USD 1200", "eta": "18天", "valid_until": "2026-09-20"} text = copy.scheme_card( work_order_no="WO202609150021", status="已报价", facts=facts, quote=quote, transport_mode="海运", ) self.assertIn("运输方式：海运", text) self.assertIn("货物数量：15件", text) self.assertIn("箱型箱量：1x40HQ", text) self.assertNotIn("件数：", text) self.assertNotIn("包装类型：", text) rows = { x["keyname"]: x["value"] for x in copy.scheme_horizontal_rows( work_order_no="WO202609150021", facts=facts, quote=quote, transport_mode="海运", ) } self.assertIn("海运 上海→洛杉矶", rows["航线"]) self.assertIn("15件", rows["货物"]) self.assertIn("1x40HQ", rows["箱型"]) self.assertNotIn("包装", rows) def test_invited_names_carry_role(self) -> None: text = copy.sea_group_created( work_order_no="WO202609150019", sales_name="吴季林", product_names=["罗小华"], ) self.assertEqual( text, "协同群已创建，群名与工单号一致：WO202609150019。" "已邀请：销售吴季林、产品罗小华。请在群聊中继续确认报价。", ) if __name__ == "__main__": unittest.main()