221 lines
8.0 KiB
Python
221 lines
8.0 KiB
Python
"""
|
|
空运询价字段合同：起运港非必填，5 个新非必填，贸易条款不进出站。
|
|
"""
|
|
|
|
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")
|
|
|
|
from agent.ledger.memory_ledger import MemoryLedger
|
|
from agent.policy import inquiry_copy as copy
|
|
from agent.policy.air_text_flow import AirTextInquiryFlow
|
|
from agent.schema.field_validate import (
|
|
air_general_cargo_collab_defaults,
|
|
validate_required_fields,
|
|
)
|
|
from agent.schema.tms_air_query import assemble_air_query, volume_cbm_number
|
|
|
|
|
|
class AirFieldContractTests(unittest.TestCase):
|
|
def test_volume_dimension_product(self) -> None:
|
|
self.assertEqual(volume_cbm_number("1.1*1.2*3"), "3.96")
|
|
self.assertEqual(volume_cbm_number("1.1x1.2x3"), "3.96")
|
|
self.assertEqual(volume_cbm_number("2CBM"), "2")
|
|
out = assemble_air_query(
|
|
{
|
|
"目的港": "吉隆坡",
|
|
"品名": "普货",
|
|
"件数": "3件",
|
|
"毛重": "5",
|
|
"体积": "1.1*1.2*3",
|
|
"包装方式": "托盘",
|
|
"报价日期": "2026-10-08",
|
|
}
|
|
)
|
|
self.assertTrue(out["ok"], out)
|
|
self.assertEqual(out["facts"]["体积"], "1.1*1.2*3")
|
|
|
|
def test_origin_optional_still_ok(self) -> None:
|
|
result = validate_required_fields(
|
|
facts={
|
|
"目的港": "克拉克",
|
|
"品名": "茶叶",
|
|
"件数": "10",
|
|
"毛重": "100",
|
|
"体积": "2",
|
|
"包装方式": "托盘",
|
|
},
|
|
business_line="AIR",
|
|
)
|
|
self.assertTrue(result["ok"])
|
|
self.assertNotIn("起运港", result["missing"])
|
|
|
|
def test_dest_still_required(self) -> None:
|
|
result = validate_required_fields(
|
|
facts={
|
|
"品名": "茶叶",
|
|
"件数": "10",
|
|
"毛重": "100",
|
|
"体积": "2",
|
|
"包装方式": "托盘",
|
|
},
|
|
business_line="AIR",
|
|
)
|
|
self.assertFalse(result["ok"])
|
|
self.assertIn("目的港", result["missing"])
|
|
|
|
def test_assemble_without_origin_ok_and_keeps_optional(self) -> None:
|
|
out = assemble_air_query(
|
|
{
|
|
"目的港": "克拉克",
|
|
"品名": "茶叶",
|
|
"件数": "10件",
|
|
"毛重": "100KGS",
|
|
"体积": "2CBM",
|
|
"包装方式": "托盘",
|
|
"报价日期": "2026-09-22",
|
|
"货源地": "东莞",
|
|
"货好时间": "随时",
|
|
"提货地址": "东莞仓库",
|
|
"收货地址": "马尼拉",
|
|
"贸易条款": "FOB",
|
|
}
|
|
)
|
|
self.assertTrue(out["ok"])
|
|
self.assertNotIn("起运港", out["missing"])
|
|
facts = out["facts"]
|
|
self.assertEqual(facts["货源地"], "东莞")
|
|
self.assertEqual(facts["货好时间"], "随时")
|
|
self.assertEqual(facts["提货地址"], "东莞仓库")
|
|
self.assertEqual(facts["收货地址"], "马尼拉")
|
|
self.assertEqual(facts["贸易条款"], "FOB")
|
|
self.assertNotIn("tradeTerms", facts)
|
|
self.assertNotIn("transportTerms", facts)
|
|
|
|
def test_confirm_card_lists_optional_and_has_no_work_order(self) -> None:
|
|
text = copy.air_confirm_card(
|
|
{
|
|
"目的港": "克拉克",
|
|
"品名": "茶叶",
|
|
"件数": "10",
|
|
"毛重": "100",
|
|
"体积": "2",
|
|
"包装方式": "托盘",
|
|
"报价日期": "2026-09-22",
|
|
}
|
|
)
|
|
self.assertNotIn("工单号", text)
|
|
self.assertIn("起运地（非必填）：", text)
|
|
self.assertNotIn("起运港（非必填）", text)
|
|
self.assertLess(text.index("起运地（非必填）："), text.index("目的地（必填）：克拉克"))
|
|
self.assertIn("目的地（必填）：克拉克", text)
|
|
self.assertIn("货源地（非必填）：", text)
|
|
self.assertIn("贸易条款（非必填）：", text)
|
|
self.assertIn(copy.AIR_CONFIRM_TAIL, text)
|
|
|
|
def test_clarify_puts_origin_above_destination(self) -> None:
|
|
text = copy.ask_clarify(
|
|
facts={},
|
|
missing_keys=["目的港", "品名"],
|
|
transport_mode="空运",
|
|
)
|
|
self.assertIn("1.起运地（非必填）：\n(如：南京）", text)
|
|
self.assertIn("2.目的地（必填）：\n(如：吉隆坡）", text)
|
|
self.assertLess(text.index("起运地（非必填）"), text.index("目的地（必填）"))
|
|
self.assertNotIn("起运港（非必填）", text)
|
|
|
|
def test_empty_origin_confirm_then_quote(self) -> None:
|
|
flow = AirTextInquiryFlow(MemoryLedger())
|
|
replies: list[str] = []
|
|
|
|
def reply(text: str, extra=None) -> None:
|
|
replies.append(text)
|
|
_ = extra
|
|
|
|
phase = flow.on_text(
|
|
sender_id="u_no_pol",
|
|
text="空运",
|
|
reply=reply,
|
|
injected_facts={
|
|
"目的港": "克拉克",
|
|
"品名": "茶叶",
|
|
"件数": "10",
|
|
"毛重": "100",
|
|
"体积": "2",
|
|
"包装方式": "托盘",
|
|
},
|
|
injected_mode="AIR",
|
|
)
|
|
self.assertEqual(phase, "wait_confirm")
|
|
self.assertFalse(flow.session_of("u_no_pol").work_order_no)
|
|
self.assertIn(copy.AIR_CONFIRM_TAIL, replies[-1])
|
|
phase = flow.on_text(sender_id="u_no_pol", text="确定", reply=reply)
|
|
self.assertEqual(phase, "wait_collab")
|
|
self.assertTrue(flow.session_of("u_no_pol").work_order_no)
|
|
|
|
def test_assemble_keeps_raw_package_not_code(self) -> None:
|
|
"""包装出站跟销售原文，禁止编成 1/2/3。"""
|
|
out = assemble_air_query(
|
|
{
|
|
"起运港": "珠海",
|
|
"目的港": "克拉克",
|
|
"品名": "茶叶",
|
|
"件数": "10件",
|
|
"毛重": "100KGS",
|
|
"体积": "2CBM",
|
|
"包装方式": "托盘",
|
|
"报价日期": "2026-09-22",
|
|
}
|
|
)
|
|
self.assertTrue(out["ok"])
|
|
facts = out["facts"]
|
|
self.assertEqual(facts["包装方式"], "托盘")
|
|
self.assertEqual(facts["packageType"], "托盘")
|
|
self.assertEqual(facts["originCode"], "ZUH")
|
|
self.assertEqual(facts["destinationCode"], "CRK")
|
|
self.assertNotIn("origin", facts)
|
|
self.assertNotIn("destination", facts)
|
|
|
|
def test_assemble_missing_dest_still_blocks(self) -> None:
|
|
out = assemble_air_query(
|
|
{
|
|
"品名": "茶叶",
|
|
"件数": "10",
|
|
"毛重": "100",
|
|
"体积": "2",
|
|
"包装方式": "托盘",
|
|
"报价日期": "2026-09-22",
|
|
}
|
|
)
|
|
self.assertFalse(out["ok"])
|
|
self.assertIn("目的港", out["missing"])
|
|
|
|
def test_general_cargo_defaults_battery_and_magnet(self) -> None:
|
|
"""品名正好是普货，含电含磁还空着才默认否。"""
|
|
self.assertEqual(
|
|
air_general_cargo_collab_defaults({"品名": "普货"}),
|
|
{"是否含电": "否", "是否含磁": "否"},
|
|
)
|
|
self.assertEqual(
|
|
air_general_cargo_collab_defaults({"货物品名": "普货"}),
|
|
{"是否含电": "否", "是否含磁": "否"},
|
|
)
|
|
self.assertEqual(
|
|
air_general_cargo_collab_defaults({"品名": "普货", "是否含电": "是"}),
|
|
{"是否含磁": "否"},
|
|
)
|
|
self.assertEqual(air_general_cargo_collab_defaults({"品名": "配件"}), {})
|
|
self.assertEqual(air_general_cargo_collab_defaults({"品名": "普货纸箱"}), {})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|