208 lines
6.8 KiB
Python
208 lines
6.8 KiB
Python
"""只读订单核对 XML 解析测试；不连接真实手机。"""
|
|
|
|
import unittest
|
|
from contextlib import AbstractContextManager
|
|
|
|
from src.pdd_purchase_reconcile_adapter import (
|
|
PurchaseOrderCandidate,
|
|
PurchaseReconcileQuery,
|
|
)
|
|
from src.pdd_u2_purchase_reconcile_adapter import (
|
|
U2PddPurchaseReconcileAdapter,
|
|
_is_single_order_detail,
|
|
_merge_order_evidence,
|
|
_parse_xml,
|
|
parse_order_candidates,
|
|
)
|
|
from src.purchase_reconcile_service import PurchaseReconcileService
|
|
|
|
|
|
def query() -> PurchaseReconcileQuery:
|
|
return PurchaseReconcileQuery(
|
|
goods_id="737116531267",
|
|
options={"color": "黑色", "size": "L"},
|
|
quantity=2,
|
|
unit_price_cent=4200,
|
|
total_price_cent=8400,
|
|
irreversible_action_at="2026-08-10T08:00:00Z",
|
|
reconcile_started_at="2026-08-10T08:05:00Z",
|
|
)
|
|
|
|
|
|
def xml_card(text: str) -> str:
|
|
return (
|
|
'<?xml version="1.0" encoding="UTF-8"?>'
|
|
'<hierarchy rotation="0">'
|
|
f'<node text="" content-desc="{text}" bounds="[0,100][1080,900]" '
|
|
'visible-to-user="true" enabled="true" />'
|
|
"</hierarchy>"
|
|
)
|
|
|
|
|
|
def xml_detail(text: str) -> str:
|
|
package = "com.xunmeng.pinduoduo"
|
|
return (
|
|
'<?xml version="1.0" encoding="UTF-8"?>'
|
|
'<hierarchy rotation="0">'
|
|
f'<node package="{package}" text="{text}" />'
|
|
f'<node package="{package}" text="去支付" />'
|
|
f'<node package="{package}" text="取消订单" />'
|
|
"</hierarchy>"
|
|
)
|
|
|
|
|
|
class FakeSession(AbstractContextManager):
|
|
def __init__(self, device):
|
|
self.device = device
|
|
|
|
def __enter__(self):
|
|
return self.device
|
|
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
|
return None
|
|
|
|
|
|
class FakeDeviceService:
|
|
def __init__(self, device):
|
|
self.device = device
|
|
|
|
def connect(self, device_address):
|
|
return FakeSession(self.device)
|
|
|
|
|
|
class DetailDevice:
|
|
def __init__(self, pages):
|
|
self.pages = pages
|
|
self.page_index = 0
|
|
self.app_current_calls = 0
|
|
self.app_start_calls = 0
|
|
self.swipes = 0
|
|
|
|
def dump_hierarchy(self, compressed=False):
|
|
return self.pages[self.page_index]
|
|
|
|
def app_current(self):
|
|
self.app_current_calls += 1
|
|
return {"package": "com.android.settings"}
|
|
|
|
def app_start(self, package):
|
|
self.app_start_calls += 1
|
|
|
|
def window_size(self):
|
|
return 1080, 2400
|
|
|
|
def swipe(self, *args):
|
|
self.swipes += 1
|
|
self.page_index = min(self.page_index + 1, len(self.pages) - 1)
|
|
|
|
|
|
class PddU2PurchaseReconcileAdapterTest(unittest.TestCase):
|
|
def test_parser_extracts_only_required_unpaid_order_fields(self):
|
|
candidates = parse_order_candidates(
|
|
xml_card(
|
|
"待付款 订单编号：ORDER-20260810 商品编号：737116531267 "
|
|
"黑色 L 共2件 合计 ¥84.00 下单时间：2026-08-10 16:03:00"
|
|
),
|
|
query(),
|
|
)
|
|
|
|
self.assertEqual(len(candidates), 1)
|
|
candidate = candidates[0]
|
|
self.assertEqual(candidate.order_no, "ORDER-20260810")
|
|
self.assertEqual(candidate.goods_id, "737116531267")
|
|
self.assertEqual(candidate.options, {"color": "黑色", "size": "L"})
|
|
self.assertEqual(candidate.quantity, 2)
|
|
self.assertEqual(candidate.total_price_cent, 8400)
|
|
self.assertEqual(candidate.ordered_at, "2026-08-10T08:03:00Z")
|
|
self.assertEqual(candidate.payment_status, "unpaid")
|
|
self.assertFalse(hasattr(candidate, "recipient"))
|
|
self.assertFalse(hasattr(candidate, "address"))
|
|
|
|
def test_paid_or_cancelled_text_never_becomes_unpaid(self):
|
|
candidates = parse_order_candidates(
|
|
xml_card(
|
|
"待付款 已付款 订单编号：ORDER-PAID 商品编号：737116531267 "
|
|
"黑色 L 共2件 合计 ¥84.00 下单时间：2026-08-10 16:03:00"
|
|
),
|
|
query(),
|
|
)
|
|
|
|
self.assertEqual(candidates[0].payment_status, "other")
|
|
|
|
def test_missing_goods_id_and_time_are_kept_incomplete_for_manual_review(self):
|
|
candidates = parse_order_candidates(
|
|
xml_card("待付款 订单编号：ORDER-INCOMPLETE 黑色 L 共2件 合计 ¥84.00"),
|
|
query(),
|
|
)
|
|
|
|
self.assertEqual(candidates[0].goods_id, "")
|
|
self.assertEqual(candidates[0].ordered_at, "")
|
|
|
|
def test_adapter_merges_one_order_detail_across_viewports(self):
|
|
device = DetailDevice(
|
|
[
|
|
xml_detail(
|
|
"待付款 商品编号：737116531267 黑色 L "
|
|
"共2件 合计 ¥84.00"
|
|
),
|
|
xml_detail(
|
|
"订单编号：ORDER-20260810 "
|
|
"下单时间：2026-08-10 16:03:00"
|
|
),
|
|
]
|
|
)
|
|
adapter = U2PddPurchaseReconcileAdapter(
|
|
"serial",
|
|
device_service=FakeDeviceService(device),
|
|
settle_seconds=0,
|
|
max_pages=3,
|
|
)
|
|
|
|
scan = adapter.read_order_candidates(query())
|
|
adapter.close()
|
|
|
|
self.assertEqual(len(scan.candidates), 1)
|
|
candidate = scan.candidates[0]
|
|
self.assertEqual(candidate.order_no, "ORDER-20260810")
|
|
self.assertEqual(candidate.goods_id, "737116531267")
|
|
self.assertEqual(candidate.options, {"color": "黑色", "size": "L"})
|
|
self.assertEqual(candidate.quantity, 2)
|
|
self.assertEqual(candidate.total_price_cent, 8400)
|
|
self.assertEqual(candidate.ordered_at, "2026-08-10T08:03:00Z")
|
|
self.assertEqual(candidate.payment_status, "unpaid")
|
|
self.assertTrue(PurchaseReconcileService._candidate_matches(candidate, query()))
|
|
self.assertEqual(device.app_current_calls, 0)
|
|
self.assertEqual(device.app_start_calls, 0)
|
|
|
|
def test_conflicting_order_numbers_are_not_merged(self):
|
|
first = PurchaseOrderCandidate(
|
|
order_no="ORDER-A",
|
|
goods_id="737116531267",
|
|
payment_status="unpaid",
|
|
)
|
|
second = PurchaseOrderCandidate(
|
|
order_no="ORDER-B",
|
|
quantity=2,
|
|
total_price_cent=8400,
|
|
payment_status="unpaid",
|
|
)
|
|
|
|
self.assertIsNone(_merge_order_evidence(first, second))
|
|
|
|
def test_multiple_order_actions_disable_detail_evidence_merge(self):
|
|
root = _parse_xml(
|
|
'<hierarchy><node package="com.xunmeng.pinduoduo" '
|
|
'text="订单编号：ORDER-A" />'
|
|
'<node package="com.xunmeng.pinduoduo" text="去支付" />'
|
|
'<node package="com.xunmeng.pinduoduo" '
|
|
'text="订单编号：ORDER-B" />'
|
|
'<node package="com.xunmeng.pinduoduo" text="去支付" />'
|
|
"</hierarchy>"
|
|
)
|
|
|
|
self.assertFalse(_is_single_order_detail(root))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|