104 lines
4.0 KiB
Python
104 lines
4.0 KiB
Python
"""HttpLedger 只测 Jeecg 包装解析，不打 8180 / TMS。"""
|
|
|
|
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("LEDGER_BACKEND", "memory")
|
|
|
|
from agent.ledger import LedgerResponse
|
|
from agent.ledger.http_ledger import HttpLedger, _unwrap
|
|
from agent.schema.fingerprint import ascii_idempotency_key, inquiry_fingerprint
|
|
|
|
|
|
class _FakeLedgerClient:
|
|
"""只记路径，不打 8180。"""
|
|
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, str, dict]] = []
|
|
self.timeout_seconds = 10
|
|
|
|
def request_json(self, method, path, *, json_body=None, idempotency_key=""):
|
|
self.calls.append((method, path, dict(json_body or {})))
|
|
_ = idempotency_key
|
|
return LedgerResponse(ok=True, status_code=200, data={"success": True, "result": {"ok": True}})
|
|
|
|
def close(self) -> None:
|
|
return None
|
|
|
|
|
|
class HttpLedgerParseTests(unittest.TestCase):
|
|
def test_unwrap_jeecg_ok(self) -> None:
|
|
inner = _unwrap({"success": True, "result": {"ok": True, "has_price": True, "classification": "HAS_PRICE"}})
|
|
self.assertTrue(inner["ok"])
|
|
self.assertEqual(inner["classification"], "HAS_PRICE")
|
|
|
|
def test_unwrap_jeecg_fail(self) -> None:
|
|
inner = _unwrap({"success": False, "message": "unauthorized", "result": {}})
|
|
self.assertFalse(inner["ok"])
|
|
|
|
def test_chinese_fingerprint_header_is_ascii(self) -> None:
|
|
raw = inquiry_fingerprint(
|
|
{"起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "毛重": "100", "体积": "1"},
|
|
business_line="AIR",
|
|
)
|
|
key = ascii_idempotency_key(raw)
|
|
key.encode("ascii")
|
|
self.assertTrue(key.startswith("sha256:"))
|
|
self.assertEqual(ascii_idempotency_key("tms:WO1"), "tms:WO1")
|
|
|
|
def test_service_token_not_sent_as_jwt(self) -> None:
|
|
from agent.ledger import LedgerClient
|
|
|
|
client = LedgerClient(base_url="http://127.0.0.1:8180/jeecgboot", service_token="not-a-jwt")
|
|
headers = client._http().headers
|
|
self.assertEqual(headers.get("X-Agent-Token"), "not-a-jwt")
|
|
self.assertFalse(headers.get("X-Access-Token"))
|
|
client.close()
|
|
|
|
def test_update_facts_posts_agent_path(self) -> None:
|
|
fake = _FakeLedgerClient()
|
|
out = HttpLedger(client=fake).update_facts(
|
|
work_order_no="WO202609220001", facts={"起运港": "深圳"}
|
|
)
|
|
self.assertTrue(out.get("ok"))
|
|
method, path, body = fake.calls[0]
|
|
self.assertEqual(method, "POST")
|
|
self.assertEqual(path, "/inquiry/agent/ticket/updateFacts")
|
|
self.assertEqual(body["workOrderNo"], "WO202609220001")
|
|
self.assertEqual(body["facts"]["起运港"], "深圳")
|
|
|
|
def test_upsert_quote_posts_replace_tms(self) -> None:
|
|
fake = _FakeLedgerClient()
|
|
out = HttpLedger(client=fake).upsert_quote(
|
|
work_order_no="WO202609220015",
|
|
quote={"total": "17100", "currency": "CNY", "route_line": "上海-谅山-胡志明"},
|
|
to_status="已报价",
|
|
)
|
|
self.assertTrue(out.get("ok"))
|
|
self.assertEqual(len(fake.calls), 1)
|
|
method, path, body = fake.calls[0]
|
|
self.assertEqual(method, "POST")
|
|
self.assertEqual(path, "/inquiry/agent/ticket/replaceTmsQuote")
|
|
self.assertEqual(body["workOrderNo"], "WO202609220015")
|
|
self.assertEqual(body["quote"]["total"], "17100")
|
|
|
|
def test_void_quote_files_posts_agent_path(self) -> None:
|
|
fake = _FakeLedgerClient()
|
|
out = HttpLedger(client=fake).void_quote_files(work_order_no="WO202609220001")
|
|
self.assertTrue(out.get("ok"))
|
|
method, path, body = fake.calls[0]
|
|
self.assertEqual(method, "POST")
|
|
self.assertEqual(path, "/inquiry/agent/ticket/voidQuoteFiles")
|
|
self.assertEqual(body["workOrderNo"], "WO202609220001")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|