151 lines
5.7 KiB
Python
151 lines
5.7 KiB
Python
"""
|
|
报价模板 sheet_map：JSON 剥离、缺项检测、补跑单测（不打外网）。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
|
|
from agent.llm.mode_sheet_map import (
|
|
assess_mapping_gaps,
|
|
build_user_message,
|
|
invoke_sheet_map,
|
|
parse_mapping_json,
|
|
strip_json_fence,
|
|
)
|
|
|
|
|
|
class TestSheetMapParse(unittest.TestCase):
|
|
def test_strip_fence(self) -> None:
|
|
raw = '```json\n{"schemaVersion":"quote-template-mapping-v1","fields":{}}\n```'
|
|
self.assertIn("schemaVersion", strip_json_fence(raw))
|
|
|
|
def test_parse_ok(self) -> None:
|
|
raw = '{"schemaVersion":"quote-template-mapping-v1","fields":{"quote_no":"S!A1"},"feeRegions":[]}'
|
|
out = parse_mapping_json(raw)
|
|
self.assertTrue(out["ok"])
|
|
self.assertEqual(out["mapping"]["fields"]["quote_no"], "S!A1")
|
|
|
|
def test_parse_invalid(self) -> None:
|
|
out = parse_mapping_json("not-json")
|
|
self.assertFalse(out["ok"])
|
|
|
|
def test_user_message_contains_inventory(self) -> None:
|
|
msg = build_user_message(
|
|
{
|
|
"transportMode": "sea",
|
|
"fileName": "海运报价单模板.xlsx",
|
|
"sheet": "Sheet1",
|
|
"cellInventory": ["Sheet1!A1 | 报价日期 | - | FFFF00"],
|
|
}
|
|
)
|
|
self.assertIn("海运", msg)
|
|
self.assertIn("Sheet1!A1", msg)
|
|
self.assertIn("自检", msg)
|
|
|
|
def test_user_message_gaps_block(self) -> None:
|
|
msg = build_user_message(
|
|
{
|
|
"transportMode": "sea",
|
|
"fileName": "x.xlsx",
|
|
"sheet": "S",
|
|
"cellInventory": ["S!A1 | x | - | -"],
|
|
},
|
|
gaps=["费用区缺失：始发港费用"],
|
|
)
|
|
self.assertIn("补跑缺口", msg)
|
|
self.assertIn("始发港费用", msg)
|
|
|
|
def test_assess_gaps_sea_fee_and_fields(self) -> None:
|
|
payload = {
|
|
"transportMode": "sea",
|
|
"cellInventory": [
|
|
"海运询价单!A18 | 始发港费用/Origin Charges | - | -",
|
|
"海运询价单!A30 | 海运费用/Ocean Freight | - | -",
|
|
"海运询价单!A35 | 目的港费用/Destination Charges | - | -",
|
|
"海运询价单!E6 | 提货地址/Place of Receipt | - | -",
|
|
"海运询价单!A11 | 货值/Cargo Value | - | -",
|
|
],
|
|
}
|
|
mapping = {
|
|
"fields": {"quote_no": "海运询价单!B3"},
|
|
"feeRegions": [],
|
|
}
|
|
gaps = assess_mapping_gaps(payload, mapping)
|
|
self.assertTrue(any("始发港费用" in g for g in gaps))
|
|
self.assertTrue(any("海运费用" in g for g in gaps))
|
|
self.assertTrue(any("目的港费用" in g for g in gaps))
|
|
self.assertTrue(any("pickup_address" in g for g in gaps))
|
|
self.assertTrue(any("cargo_value" in g for g in gaps))
|
|
|
|
def test_assess_gaps_complete_empty(self) -> None:
|
|
payload = {
|
|
"transportMode": "sea",
|
|
"cellInventory": [
|
|
"S!A18 | 始发港费用/Origin Charges | - | -",
|
|
"S!E6 | 提货地址 | - | -",
|
|
],
|
|
}
|
|
mapping = {
|
|
"fields": {"pickup_address": "S!H6"},
|
|
"feeRegions": [
|
|
{
|
|
"id": "origin_charges",
|
|
"title": "始发港费用",
|
|
"tmsSource": "polCostItems",
|
|
"stopAtText": ["x"],
|
|
"anchor": "S!A19",
|
|
}
|
|
],
|
|
}
|
|
self.assertEqual(assess_mapping_gaps(payload, mapping), [])
|
|
|
|
def test_invoke_retries_when_gaps(self) -> None:
|
|
calls: list[str] = []
|
|
|
|
def fake_chat(*, mode, messages, temperature=0.0): # noqa: ARG001
|
|
calls.append(messages[-1]["content"])
|
|
if len(calls) == 1:
|
|
content = (
|
|
'{"schemaVersion":"quote-template-mapping-v1",'
|
|
'"fields":{"quote_no":"S!B3"},"feeRegions":[]}'
|
|
)
|
|
else:
|
|
content = (
|
|
'{"schemaVersion":"quote-template-mapping-v1",'
|
|
'"fields":{"quote_no":"S!B3","pickup_address":"S!H6","cargo_value":"S!B11"},'
|
|
'"feeRegions":[{"id":"origin_charges","title":"始发港费用",'
|
|
'"tmsSource":"polCostItems","anchor":"S!A19","stopAtText":["海运费用"],'
|
|
'"grow":"insert_before_stop","styleCopyRows":1}]}'
|
|
)
|
|
return SimpleNamespace(ok=True, content=content, provider="stub", error="")
|
|
|
|
payload = {
|
|
"versionId": "v1",
|
|
"transportMode": "sea",
|
|
"fileName": "sea.xlsx",
|
|
"sheet": "S",
|
|
"cellInventory": [
|
|
"S!A18 | 始发港费用/Origin Charges | - | -",
|
|
"S!E6 | 提货地址/Place of Receipt | - | -",
|
|
"S!A11 | 货值/Cargo Value | - | -",
|
|
],
|
|
}
|
|
out = invoke_sheet_map(payload, chat_fn=fake_chat, allow_network=False)
|
|
self.assertTrue(out["ok"])
|
|
self.assertEqual(out.get("rounds"), 2)
|
|
self.assertEqual(len(calls), 2)
|
|
self.assertIn("补跑缺口", calls[1])
|
|
self.assertIn("pickup_address", out["mapping"]["fields"])
|
|
self.assertEqual(len(out["mapping"]["feeRegions"]), 1)
|
|
self.assertIn("mappingGaps", out["mapping"])
|
|
self.assertIsInstance(out["mapping"]["mappingGaps"], list)
|
|
# 第二轮已补齐清单内缺口 → mappingGaps 为空且与 gaps_after 一致
|
|
self.assertEqual(out.get("gaps_after"), out["mapping"]["mappingGaps"])
|
|
self.assertEqual(out["mapping"]["mappingGaps"], [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|