585 lines
20 KiB
Python
585 lines
20 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 (
|
|
PddPurchaseReconcileError,
|
|
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",
|
|
order_submitted_at="2026-08-10T08:03: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 FakeClock:
|
|
def __init__(self):
|
|
self.now = 0.0
|
|
|
|
def monotonic(self):
|
|
return self.now
|
|
|
|
def sleep(self, seconds):
|
|
self.now += seconds
|
|
|
|
|
|
def xml_package_page(package: str, text: str = "") -> str:
|
|
return (
|
|
'<hierarchy>'
|
|
f'<node package="{package}" text="{text}" />'
|
|
f'<node package="{package}" text="" />'
|
|
f'<node package="{package}" text="" />'
|
|
"</hierarchy>"
|
|
)
|
|
|
|
|
|
def xml_android_wechat_chooser() -> str:
|
|
return (
|
|
'<hierarchy>'
|
|
'<node package="android" text="选择要使用的应用" />'
|
|
'<node package="android" text="微信" />'
|
|
'<node package="android" text="微信分身" />'
|
|
'</hierarchy>'
|
|
)
|
|
|
|
|
|
class TransitionDevice(DetailDevice):
|
|
def __init__(
|
|
self,
|
|
pages,
|
|
current_package="com.tencent.mm",
|
|
current_activity="",
|
|
):
|
|
super().__init__(pages)
|
|
self.current_package = current_package
|
|
self.current_activity = current_activity
|
|
self.dump_calls = 0
|
|
self.back_calls = 0
|
|
|
|
def dump_hierarchy(self, compressed=False):
|
|
index = min(self.dump_calls, len(self.pages) - 1)
|
|
self.dump_calls += 1
|
|
return self.pages[index]
|
|
|
|
def app_current(self):
|
|
self.app_current_calls += 1
|
|
return {
|
|
"package": self.current_package,
|
|
"activity": self.current_activity,
|
|
}
|
|
|
|
def press(self, key):
|
|
if key == "back":
|
|
self.back_calls += 1
|
|
|
|
def swipe(self, *args):
|
|
self.swipes += 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_minimal_order_and_time_candidate_uses_five_minute_window(self):
|
|
within_window = PurchaseOrderCandidate(
|
|
order_no="ORDER-MINIMAL",
|
|
ordered_at="2026-08-10T08:08:00Z",
|
|
payment_status="unpaid",
|
|
)
|
|
outside_window = PurchaseOrderCandidate(
|
|
order_no="ORDER-LATE",
|
|
ordered_at="2026-08-10T08:08:01Z",
|
|
payment_status="unpaid",
|
|
)
|
|
|
|
self.assertTrue(
|
|
PurchaseReconcileService._candidate_matches(
|
|
within_window, query()
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
PurchaseReconcileService._candidate_matches(
|
|
outside_window, query()
|
|
)
|
|
)
|
|
|
|
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))
|
|
|
|
def test_waits_for_pdd_detail_after_external_transition(self):
|
|
complete_detail = xml_detail(
|
|
"待付款 订单编号：ORDER-20260810 "
|
|
"商品编号：737116531267 黑色 L 共2件 合计 ¥84.00 "
|
|
"下单时间：2026-08-10 16:03:00"
|
|
)
|
|
device = TransitionDevice(
|
|
[
|
|
xml_package_page("com.tencent.mm", "微信登录"),
|
|
xml_package_page("com.xunmeng.pinduoduo", "页面加载中"),
|
|
complete_detail,
|
|
]
|
|
)
|
|
clock = FakeClock()
|
|
adapter = U2PddPurchaseReconcileAdapter(
|
|
"serial",
|
|
device_service=FakeDeviceService(device),
|
|
settle_seconds=1,
|
|
transition_timeout_seconds=30,
|
|
max_pages=2,
|
|
monotonic=clock.monotonic,
|
|
sleep=clock.sleep,
|
|
)
|
|
|
|
scan = adapter.read_order_candidates(query())
|
|
adapter.close()
|
|
|
|
self.assertEqual(len(scan.candidates), 1)
|
|
self.assertEqual(device.app_start_calls, 1)
|
|
self.assertEqual(device.app_current_calls, 0)
|
|
self.assertLess(clock.now, 30)
|
|
|
|
def test_android_wechat_chooser_is_closed_once_before_reading_detail(self):
|
|
device = TransitionDevice(
|
|
[
|
|
xml_android_wechat_chooser(),
|
|
xml_package_page("com.xunmeng.pinduoduo", "页面加载中"),
|
|
xml_detail("待付款"),
|
|
],
|
|
current_package="android",
|
|
current_activity="com.android.internal.app.ChooserActivity",
|
|
)
|
|
clock = FakeClock()
|
|
adapter = U2PddPurchaseReconcileAdapter(
|
|
"serial",
|
|
device_service=FakeDeviceService(device),
|
|
settle_seconds=1,
|
|
transition_timeout_seconds=30,
|
|
monotonic=clock.monotonic,
|
|
sleep=clock.sleep,
|
|
)
|
|
|
|
adapter._open_unpaid_orders(device)
|
|
|
|
self.assertEqual(device.back_calls, 1)
|
|
self.assertEqual(device.app_start_calls, 0)
|
|
self.assertEqual(device.app_current_calls, 1)
|
|
|
|
def test_mixed_package_chooser_with_duplicate_accessibility_values_closes(self):
|
|
chooser_over_pdd = (
|
|
'<hierarchy>'
|
|
'<node package="com.xunmeng.pinduoduo" text="页面加载中" />'
|
|
'<node package="com.xunmeng.pinduoduo" text="" />'
|
|
'<node package="com.xunmeng.pinduoduo" text="" />'
|
|
'<node package="android" text="选择要使用的应用" '
|
|
'content-desc="选择要使用的应用" />'
|
|
'<node package="com.tencent.mm" text="微信" '
|
|
'content-desc="微信" />'
|
|
'<node package="com.tencent.mm" text="微信 2" />'
|
|
'</hierarchy>'
|
|
)
|
|
device = TransitionDevice(
|
|
[chooser_over_pdd, xml_detail("待付款")],
|
|
current_package="android",
|
|
current_activity="com.android.internal.app.ChooserActivity",
|
|
)
|
|
clock = FakeClock()
|
|
adapter = U2PddPurchaseReconcileAdapter(
|
|
"serial",
|
|
device_service=FakeDeviceService(device),
|
|
settle_seconds=1,
|
|
transition_timeout_seconds=30,
|
|
monotonic=clock.monotonic,
|
|
sleep=clock.sleep,
|
|
)
|
|
|
|
adapter._open_unpaid_orders(device)
|
|
|
|
self.assertEqual(device.back_calls, 1)
|
|
self.assertEqual(device.app_start_calls, 0)
|
|
self.assertEqual(device.app_current_calls, 1)
|
|
|
|
def test_android_resolver_package_and_title_variant_are_supported(self):
|
|
resolver = (
|
|
'<hierarchy>'
|
|
'<node package="com.android.intentresolver" '
|
|
'text="使用以下应用打开" />'
|
|
'<node package="com.tencent.mm" text="微信" />'
|
|
'<node package="com.tencent.mm" text="微信分身" />'
|
|
'</hierarchy>'
|
|
)
|
|
device = TransitionDevice(
|
|
[resolver, xml_detail("待付款")],
|
|
current_package="com.android.intentresolver",
|
|
current_activity="com.android.intentresolver.ResolverActivity",
|
|
)
|
|
clock = FakeClock()
|
|
adapter = U2PddPurchaseReconcileAdapter(
|
|
"serial",
|
|
device_service=FakeDeviceService(device),
|
|
settle_seconds=1,
|
|
transition_timeout_seconds=30,
|
|
monotonic=clock.monotonic,
|
|
sleep=clock.sleep,
|
|
)
|
|
|
|
adapter._open_unpaid_orders(device)
|
|
|
|
self.assertEqual(device.back_calls, 1)
|
|
self.assertEqual(device.app_start_calls, 0)
|
|
|
|
def test_android_wechat_chooser_stops_when_one_back_does_not_close_it(self):
|
|
device = TransitionDevice(
|
|
[xml_android_wechat_chooser(), xml_android_wechat_chooser()],
|
|
current_package="android",
|
|
current_activity="com.android.internal.app.ResolverActivity",
|
|
)
|
|
clock = FakeClock()
|
|
adapter = U2PddPurchaseReconcileAdapter(
|
|
"serial",
|
|
device_service=FakeDeviceService(device),
|
|
settle_seconds=1,
|
|
transition_timeout_seconds=30,
|
|
monotonic=clock.monotonic,
|
|
sleep=clock.sleep,
|
|
)
|
|
|
|
with self.assertRaises(PddPurchaseReconcileError) as captured:
|
|
adapter._open_unpaid_orders(device)
|
|
|
|
self.assertEqual(
|
|
captured.exception.code, "RECONCILE_ANDROID_CHOOSER_STUCK"
|
|
)
|
|
self.assertEqual(device.back_calls, 1)
|
|
self.assertEqual(device.app_start_calls, 0)
|
|
|
|
def test_android_wechat_chooser_with_unknown_activity_stops_safely(self):
|
|
device = TransitionDevice(
|
|
[xml_android_wechat_chooser()],
|
|
current_package="android",
|
|
current_activity="com.example.UnknownSystemActivity",
|
|
)
|
|
clock = FakeClock()
|
|
adapter = U2PddPurchaseReconcileAdapter(
|
|
"serial",
|
|
device_service=FakeDeviceService(device),
|
|
settle_seconds=1,
|
|
transition_timeout_seconds=30,
|
|
monotonic=clock.monotonic,
|
|
sleep=clock.sleep,
|
|
)
|
|
|
|
with self.assertRaises(PddPurchaseReconcileError) as captured:
|
|
adapter._open_unpaid_orders(device)
|
|
|
|
self.assertEqual(
|
|
captured.exception.code,
|
|
"RECONCILE_ANDROID_CHOOSER_UNRECOGNIZED",
|
|
)
|
|
self.assertEqual(device.back_calls, 0)
|
|
self.assertEqual(device.app_start_calls, 0)
|
|
|
|
def test_unknown_android_page_is_not_treated_as_wechat_chooser(self):
|
|
unknown_system_page = xml_package_page("android", "系统设置")
|
|
device = TransitionDevice(
|
|
[unknown_system_page, xml_detail("待付款")],
|
|
current_package="android",
|
|
current_activity="com.android.settings.Settings",
|
|
)
|
|
clock = FakeClock()
|
|
adapter = U2PddPurchaseReconcileAdapter(
|
|
"serial",
|
|
device_service=FakeDeviceService(device),
|
|
settle_seconds=1,
|
|
transition_timeout_seconds=30,
|
|
monotonic=clock.monotonic,
|
|
sleep=clock.sleep,
|
|
)
|
|
|
|
adapter._open_unpaid_orders(device)
|
|
|
|
self.assertEqual(device.back_calls, 0)
|
|
self.assertEqual(device.app_start_calls, 1)
|
|
|
|
def test_payment_back_is_pressed_at_most_once_while_page_changes(self):
|
|
payment_page = xml_package_page(
|
|
"com.xunmeng.pinduoduo", "立即支付"
|
|
)
|
|
device = TransitionDevice(
|
|
[payment_page, payment_page, xml_detail("待付款")],
|
|
current_package="com.xunmeng.pinduoduo",
|
|
)
|
|
clock = FakeClock()
|
|
adapter = U2PddPurchaseReconcileAdapter(
|
|
"serial",
|
|
device_service=FakeDeviceService(device),
|
|
settle_seconds=1,
|
|
transition_timeout_seconds=30,
|
|
monotonic=clock.monotonic,
|
|
sleep=clock.sleep,
|
|
)
|
|
|
|
adapter._open_unpaid_orders(device)
|
|
|
|
self.assertEqual(device.back_calls, 1)
|
|
|
|
def test_transition_timeout_reports_last_waiting_reason(self):
|
|
device = TransitionDevice(
|
|
[xml_package_page("com.xunmeng.pinduoduo", "页面加载中")],
|
|
current_package="com.xunmeng.pinduoduo",
|
|
)
|
|
clock = FakeClock()
|
|
adapter = U2PddPurchaseReconcileAdapter(
|
|
"serial",
|
|
device_service=FakeDeviceService(device),
|
|
settle_seconds=1,
|
|
transition_timeout_seconds=2,
|
|
monotonic=clock.monotonic,
|
|
sleep=clock.sleep,
|
|
)
|
|
|
|
with self.assertRaises(PddPurchaseReconcileError) as captured:
|
|
adapter._open_unpaid_orders(device)
|
|
|
|
self.assertEqual(captured.exception.code, "RECONCILE_ORDER_PAGE_TIMEOUT")
|
|
self.assertIn("页面仍在过渡", str(captured.exception))
|
|
|
|
def test_transition_wait_can_be_cancelled(self):
|
|
device = TransitionDevice(
|
|
[xml_package_page("com.xunmeng.pinduoduo", "页面加载中")],
|
|
current_package="com.xunmeng.pinduoduo",
|
|
)
|
|
clock = FakeClock()
|
|
checks = iter((False, False, True))
|
|
adapter = U2PddPurchaseReconcileAdapter(
|
|
"serial",
|
|
device_service=FakeDeviceService(device),
|
|
cancelled=lambda: next(checks, True),
|
|
settle_seconds=1,
|
|
transition_timeout_seconds=30,
|
|
monotonic=clock.monotonic,
|
|
sleep=clock.sleep,
|
|
)
|
|
|
|
with self.assertRaises(PddPurchaseReconcileError) as captured:
|
|
adapter.read_order_candidates(query())
|
|
|
|
self.assertEqual(captured.exception.code, "RECONCILE_CANCELLED")
|
|
|
|
def test_security_pages_still_stop_transition_wait(self):
|
|
cases = (
|
|
("请完成验证", "RECONCILE_CAPTCHA"),
|
|
("手机号登录", "RECONCILE_LOGIN_REQUIRED"),
|
|
("操作频繁", "RECONCILE_RISK_CONTROL"),
|
|
)
|
|
for marker, error_code in cases:
|
|
with self.subTest(marker=marker):
|
|
device = TransitionDevice(
|
|
[xml_package_page("com.xunmeng.pinduoduo", marker)],
|
|
current_package="com.xunmeng.pinduoduo",
|
|
)
|
|
clock = FakeClock()
|
|
adapter = U2PddPurchaseReconcileAdapter(
|
|
"serial",
|
|
device_service=FakeDeviceService(device),
|
|
settle_seconds=1,
|
|
transition_timeout_seconds=30,
|
|
monotonic=clock.monotonic,
|
|
sleep=clock.sleep,
|
|
)
|
|
|
|
with self.assertRaises(PddPurchaseReconcileError) as captured:
|
|
adapter._open_unpaid_orders(device)
|
|
|
|
self.assertEqual(captured.exception.code, error_code)
|
|
self.assertEqual(device.app_start_calls, 0)
|
|
self.assertEqual(device.back_calls, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|