"""uiautomator2 采购演练 Adapter 测试；不连接真实手机。""" from datetime import datetime, timezone from pathlib import Path import tempfile import unittest import xml.etree.ElementTree as ET from src.pdd_device_service import PddDeviceService from src.pdd_collect_service import DimensionValue, PddCollectError, SpecDimension from src.pdd_purchase_adapter import ( PddPurchaseError, PddPurchaseSpecResolutionRequired, PurchaseSizeObservation, PurchaseSpecCandidateSnapshot, ) from src.performance_timing import TaskPerformanceTrace from src.pdd_u2_purchase_adapter import ( U2PddLivePurchaseAdapter, U2PddPurchaseAdapter, _final_submit_targets, _page_kind, _tagged_shipping_address, ) GOODS_URL = "https://mobile.yangkeduo.com/goods.html?goods_id=753136429979" FIXTURES = Path(__file__).parent / "fixtures" class FakeClock: def __init__(self) -> None: self.now = 0.0 self.sleeps = [] def monotonic(self) -> float: return self.now def sleep(self, seconds: float) -> None: self.sleeps.append(seconds) self.now += seconds def home_xml() -> str: return """ """ def panel_xml(quantity: int = 1) -> str: return f""" """ def contextual_confirm_panel_xml(quantity: int = 1) -> str: """模拟商品 897186799891 的非滚动“确定”规格面板。""" return f""" """ def vertically_scrolled_purchase_panel_xml(*, at_color_top: bool) -> str: """构造规格面板位于颜色顶部或停在中部的脱敏控件树。""" if at_color_top: scroll_content = """ """ else: scroll_content = """ """ return f""" """ class FakeEditor: def __init__(self, device) -> None: self.device = device self.count = 1 def set_text(self, value: str) -> None: self.device.editor_values.append(value) self.device.quantity = int(value) if self.device.show_keyboard_on_edit: self.device.keyboard_shown = True class FakeDevice: def __init__(self, initial_xml: str = "") -> None: self.mode = "special" if initial_xml else "home" self.special_xml = initial_xml self.quantity = 1 self.clicks = [] self.opened_urls = [] self.app_wait_calls = 0 self.app_current_calls = 0 self.has_opened = False self.editor_values = [] self.show_keyboard_on_edit = True self.keyboard_shown = False self.keyboard_status_known = True self.keyboard_can_dismiss = True self.close_panel_on_back = False self.presses = [] def app_current(self): self.app_current_calls += 1 return {"package": "com.xunmeng.pinduoduo"} def app_start(self, _package): pass def app_wait(self, _package, timeout=10): self.app_wait_calls += 1 return True def open_url(self, url): self.opened_urls.append(url) self.has_opened = True def dump_hierarchy(self): if not self.has_opened: return '' if self.mode == "special": return self.special_xml return home_xml() if self.mode == "home" else panel_xml(self.quantity) def click(self, x, y): self.clicks.append((x, y)) if self.mode == "panel" and 570 <= x <= 650 and 700 <= y <= 780: self.quantity += 1 return if self.mode == "panel" and 390 <= x <= 470 and 700 <= y <= 780: self.quantity -= 1 return self.mode = "panel" def shell(self, command): self.asserted_shell_command = command if not self.keyboard_status_known: return "input method state unavailable" value = "true" if self.keyboard_shown else "false" return f"mInputShown={value} mIsInputViewShown={value}" def press(self, key): self.presses.append(key) if key == "back" and self.keyboard_can_dismiss: self.keyboard_shown = False if self.close_panel_on_back: self.mode = "home" def __call__(self, **selector): if selector == {"className": "android.widget.EditText"}: return FakeEditor(self) raise AssertionError(f"未预期的 selector: {selector}") class VerticallyScrolledColorPanelDevice(FakeDevice): """模拟 PDD 记住规格面板纵向位置。""" def __init__(self, *, starts_at_top: bool, restores_on_swipe: bool = True): super().__init__() self.at_color_top = starts_at_top self.restores_on_swipe = restores_on_swipe self.swipes = [] def dump_hierarchy(self): if not self.has_opened: return super().dump_hierarchy() if self.mode == "home": return home_xml() if self.mode == "panel": return vertically_scrolled_purchase_panel_xml( at_color_top=self.at_color_top ) return super().dump_hierarchy() def swipe(self, x1, y1, x2, y2, duration=0.0): self.swipes.append((x1, y1, x2, y2, duration)) if self.restores_on_swipe: self.at_color_top = True class ColdStartDevice(FakeDevice): def __init__(self) -> None: super().__init__() self.current_calls = 0 self.app_started = False def app_current(self): self.current_calls += 1 package = ( "com.android.launcher" if self.current_calls == 1 else "com.xunmeng.pinduoduo" ) return {"package": package} def app_start(self, _package): self.app_started = True class StaleGoodsDevice(FakeDevice): """深链打开前后都停留在同一个旧商品页。""" def dump_hierarchy(self): return home_xml() class SlowSettingsFocusDevice(FakeDevice): """模拟真机：前台查询慢，而且错误报告为设置页。""" def __init__(self, clock: FakeClock) -> None: super().__init__() self.clock = clock def app_current(self): self.app_current_calls += 1 self.clock.sleep(11.0) return {"package": "com.android.settings"} class PanelDoesNotOpenDevice(FakeDevice): def click(self, x, y): self.clicks.append((x, y)) class ContextualConfirmPanelDevice(FakeDevice): def dump_hierarchy(self): if not self.has_opened: return '' if self.mode == "special": return self.special_xml if self.mode == "panel": return contextual_confirm_panel_xml(self.quantity) return home_xml() def click(self, x, y): self.clicks.append((x, y)) if self.mode == "panel" and 495 <= x <= 573 and 1080 <= y <= 1155: self.quantity += 1 return self.mode = "panel" class SubmitHintPanelDevice(FakeDevice): """模拟带提交提示的非滚动图片颜色面板。""" def __init__(self, panel_data: str) -> None: super().__init__() self.panel_data = panel_data def dump_hierarchy(self): if not self.has_opened: return '' if self.mode == "panel": return self.panel_data return home_xml() def address_confirmation_xml(address: str) -> str: return f""" """ def address_panel_xml(address: str) -> str: return f""" """ def multiple_address_panel_xml( address: str, *, default_indexes=(0,) ) -> str: """按真机多地址结构构造不含个人数据的最小控件树。""" cards = [] for index, (top, bottom) in enumerate(((226, 654), (654, 1083))): modify_top = 544 if index == 0 else 973 marker = "已设默认" if index in default_indexes else "使用" marker_top = 547 if index == 0 else 817 address_text = address if index in default_indexes else "备用测试地址" cards.append(f""" """) return f""" """ def address_edit_xml(address: str) -> str: return f""" """ def coloros_password_overlay_xml() -> str: return """ """ class FakeAddressEditor: def __init__(self, device, matched: bool) -> None: self.device = device self.count = 1 if matched else 0 def set_text(self, value: str) -> None: self.device.address = value class AddressFlowDevice(FakeDevice): def __init__(self) -> None: super().__init__() self.address = "测试省测试市测试区测试路1号-1dd212" def dump_hierarchy(self): if not self.has_opened: return '' if self.mode == "home": return home_xml() if self.mode == "panel": return address_confirmation_xml(self.address) if self.mode in {"address_panel", "address_saved"}: return address_panel_xml(self.address) if self.mode == "address_edit": return address_edit_xml(self.address) raise AssertionError(f"未知测试页面：{self.mode}") def click(self, x, y): self.clicks.append((x, y)) if self.mode == "home": self.mode = "panel" elif self.mode == "panel" and y < 600: self.mode = "address_panel" elif self.mode == "address_panel": self.mode = "address_edit" elif self.mode == "address_edit": self.mode = "address_saved" elif self.mode == "address_saved": self.mode = "panel" def __call__(self, **selector): if ( selector.get("className") == "android.widget.EditText" and "text" in selector ): return FakeAddressEditor(self, selector["text"] == self.address) return super().__call__(**selector) class MultipleAddressFlowDevice(AddressFlowDevice): """模拟多地址面板，并记录是否点击默认卡片和 PDD 标题栏返回。""" def __init__(self, default_indexes=(0,)) -> None: super().__init__() self.default_indexes = default_indexes self.address_modify_clicks = [] self.address_back_clicks = [] def dump_hierarchy(self): if self.mode in {"address_panel", "address_saved"}: return multiple_address_panel_xml( self.address, default_indexes=self.default_indexes ) return super().dump_hierarchy() def click(self, x, y): if self.mode == "address_panel": self.clicks.append((x, y)) self.address_modify_clicks.append((x, y)) expected_y = 571 if self.default_indexes == (0,) else 1000 if (x, y) == (948, expected_y): self.mode = "address_edit" return if self.mode == "address_saved": self.clicks.append((x, y)) self.address_back_clicks.append((x, y)) if x <= 240 and y <= 320: self.mode = "panel" return super().click(x, y) class DelayedMultipleAddressFlowDevice(MultipleAddressFlowDevice): """模拟地址标题先出现，地址列表稍后才进入控件树。""" def __init__(self, loading_reads=2) -> None: super().__init__(default_indexes=(0,)) self.loading_reads = loading_reads self.address_panel_reads = 0 def dump_hierarchy(self): if self.mode == "address_panel": self.address_panel_reads += 1 if self.address_panel_reads <= self.loading_reads: return """ """ return super().dump_hierarchy() class AlternatingDefaultAddressDevice(MultipleAddressFlowDevice): """模拟默认地址目标坐标持续变化。""" def __init__(self) -> None: super().__init__(default_indexes=(0,)) self.address_panel_reads = 0 def dump_hierarchy(self): if self.mode == "address_panel": self.address_panel_reads += 1 indexes = (self.address_panel_reads % 2,) return multiple_address_panel_xml( self.address, default_indexes=indexes ) return super().dump_hierarchy() class AmbiguousDefaultAddressDevice(MultipleAddressFlowDevice): """模拟默认地址卡片内部出现两个修改目标。""" def dump_hierarchy(self): xml_data = super().dump_hierarchy() if self.mode != "address_panel": return xml_data root = ET.fromstring(xml_data) recycler = next( node for node in root.iter("node") if node.get("class") == "androidx.recyclerview.widget.RecyclerView" ) default_card = list(recycler)[0] extra = ET.SubElement( default_card, "node", { "package": "com.xunmeng.pinduoduo", "clickable": "true", "bounds": "[800,544][880,598]", }, ) ET.SubElement( extra, "node", { "package": "com.xunmeng.pinduoduo", "text": "修改", "bounds": "[800,544][880,598]", }, ) return ET.tostring(root, encoding="unicode") class AddressSaveOverlayDevice(AddressFlowDevice): """模拟保存后出现 ColorOS 密码本浮层。""" def __init__(self, clear_after_reads=None) -> None: super().__init__() self.clear_after_reads = clear_after_reads self.overlay_reads = 0 self.save_clicks = 0 def dump_hierarchy(self): if self.mode == "address_overlay": self.overlay_reads += 1 if ( self.clear_after_reads is not None and self.overlay_reads > self.clear_after_reads ): self.mode = "address_saved" return super().dump_hierarchy() return coloros_password_overlay_xml() return super().dump_hierarchy() def click(self, x, y): if self.mode == "address_edit": self.clicks.append((x, y)) self.save_clicks += 1 self.mode = "address_overlay" return super().click(x, y) def press(self, key): self.presses.append(key) if self.mode == "address_overlay" and key == "back": self.mode = "address_saved" class PersistentAddressSaveOverlayDevice(AddressSaveOverlayDevice): """模拟系统返回键也没有关闭密码本浮层。""" def press(self, key): self.presses.append(key) class UnknownAfterAddressSaveDevice(AddressFlowDevice): """模拟保存后进入非白名单未知浮层。""" def __init__(self) -> None: super().__init__() self.save_clicks = 0 def dump_hierarchy(self): if self.mode == "address_save_unknown": return """ """ return super().dump_hierarchy() def click(self, x, y): if self.mode == "address_edit": self.clicks.append((x, y)) self.save_clicks += 1 self.mode = "address_save_unknown" return super().click(x, y) class HiddenSavedAddressDevice(AddressFlowDevice): """模拟地址面板不在控件树中暴露完整地址。""" def dump_hierarchy(self): if self.mode == "address_saved": return address_panel_xml("") return super().dump_hierarchy() class DelayedAddressEditor(FakeAddressEditor): """先记录输入，等待控件树刷新后才显示新地址。""" def set_text(self, value: str) -> None: self.device.pending_address = value class DelayedStableSaveTargetDevice(AddressFlowDevice): """输入完成后保存目标先变化，随后才稳定。""" def __init__(self) -> None: super().__init__() self.save_form_reads = 0 self.save_clicks = 0 self.pending_address = None def dump_hierarchy(self): if self.mode != "address_edit" or self.pending_address is None: return super().dump_hierarchy() self.save_form_reads += 1 if self.save_form_reads >= 2: self.address = self.pending_address xml_data = super().dump_hierarchy() if not self.address.endswith("_cg2"): return xml_data shift = min(self.save_form_reads, 3) * 10 return xml_data.replace( 'bounds="[120,1194][960,1329]"', f'bounds="[{120 + shift},1194][{960 + shift},1329]"', ) def click(self, x, y): if self.mode == "address_edit": self.save_clicks += 1 super().click(x, y) def __call__(self, **selector): if ( selector.get("className") == "android.widget.EditText" and "text" in selector ): return DelayedAddressEditor(self, selector["text"] == self.address) return super().__call__(**selector) class NeverStableSaveTargetDevice(DelayedStableSaveTargetDevice): """保存目标在两个位置之间持续变化。""" def dump_hierarchy(self): if self.mode != "address_edit" or self.pending_address is None: return AddressFlowDevice.dump_hierarchy(self) self.save_form_reads += 1 if self.save_form_reads >= 2: self.address = self.pending_address xml_data = AddressFlowDevice.dump_hierarchy(self) if not self.address.endswith("_cg2"): return xml_data shift = 10 if self.save_form_reads % 2 else 30 return xml_data.replace( 'bounds="[120,1194][960,1329]"', f'bounds="[{120 + shift},1194][{960 + shift},1329]"', ) class LostSaveClickAddressDevice(AddressFlowDevice): """保存按钮点击一次后页面仍停留在编辑页。""" def __init__(self) -> None: super().__init__() self.save_clicks = 0 def click(self, x, y): if self.mode == "address_edit": self.clicks.append((x, y)) self.save_clicks += 1 return super().click(x, y) class LostFirstModifyClickAddressDevice(AddressFlowDevice): """模拟第一次“修改”点击被页面吞掉。""" def __init__(self) -> None: super().__init__() self.modify_clicks = 0 def click(self, x, y): if self.mode != "address_panel": return super().click(x, y) self.clicks.append((x, y)) self.modify_clicks += 1 if self.modify_clicks >= 2: self.mode = "address_edit" class NeverOpenAddressEditDevice(AddressFlowDevice): """模拟两次“修改”点击都未打开编辑页。""" def __init__(self) -> None: super().__init__() self.modify_clicks = 0 def click(self, x, y): if self.mode != "address_panel": return super().click(x, y) self.clicks.append((x, y)) self.modify_clicks += 1 class UnknownAfterModifyAddressDevice(AddressFlowDevice): """模拟点击“修改”后进入无法识别的中间页面。""" def __init__(self) -> None: super().__init__() self.modify_clicks = 0 def dump_hierarchy(self): if self.mode == "address_unknown": return """ """ return super().dump_hierarchy() def click(self, x, y): if self.mode != "address_panel": return super().click(x, y) self.clicks.append((x, y)) self.modify_clicks += 1 self.mode = "address_unknown" class AmbiguousAfterModifyAddressDevice(AddressFlowDevice): """模拟点击后地址面板同时出现两个“修改”按钮。""" def __init__(self) -> None: super().__init__() self.modify_clicks = 0 def dump_hierarchy(self): if self.mode == "address_ambiguous": xml_data = address_panel_xml(self.address) return xml_data.replace( "\n \n ", """ """, ) return super().dump_hierarchy() def click(self, x, y): if self.mode != "address_panel": return super().click(x, y) self.clicks.append((x, y)) self.modify_clicks += 1 self.mode = "address_ambiguous" class U2PddPurchaseAdapterTest(unittest.TestCase): def _adapter(self, device, calls): def select_color_fn(_device, _xml, target, **_kwargs): calls.append(("color", target)) return target == "黑色" def select_size_fn(_device, _xml, target, **_kwargs): calls.append(("size", target)) return target == "3XL【140-165斤】" return U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), sleeper=lambda _seconds: None, select_color_fn=select_color_fn, select_size_fn=select_size_fn, ) def _live_adapter(self, device, calls, **adapter_options): def select_color_fn(_device, _xml, target, **_kwargs): calls.append(("color", target)) return target == "黑色" def select_size_fn(_device, _xml, target, **_kwargs): calls.append(("size", target)) return target == "3XL【140-165斤】" options = { "sleeper": lambda _seconds: None, "select_color_fn": select_color_fn, "select_size_fn": select_size_fn, } options.update(adapter_options) return U2PddLivePurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), **options, ) def test_dry_run_reaches_confirmation_without_clicking_submit(self): device = FakeDevice() calls = [] adapter = self._adapter(device, calls) adapter.open_goods(GOODS_URL) goods = adapter.read_state() adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) adapter.set_quantity(2) confirmation = adapter.read_state() adapter.enter_confirmation() adapter.stop_before_submit() adapter.close() self.assertEqual(goods.page_kind, "goods") self.assertEqual(goods.goods_id, "753136429979") self.assertEqual( confirmation.selected_options, {"color": "黑色", "size": "3XL【140-165斤】"}, ) self.assertEqual(confirmation.quantity, 2) self.assertEqual(confirmation.price_cent, 503) self.assertEqual(confirmation.page_kind, "order_confirmation") self.assertEqual(calls, [("color", "黑色"), ("size", "3XL【140-165斤】")]) # 点击一次商品页采购入口和一次加号；最终“提交订单”从不点击。 self.assertEqual(len(device.clicks), 2) self.assertEqual(device.editor_values, []) self.assertEqual(device.app_wait_calls, 0) def test_missing_size_returns_complete_structured_candidate_snapshot(self): device = FakeDevice() collected = [] def collect_candidates(current_device): collected.append(current_device) return SpecDimension( "size", "尺码", ( DimensionValue("M", True), DimensionValue("L（售罄）", False), DimensionValue("XL", True), ), ) adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), sleeper=lambda _seconds: None, select_color_fn=lambda *_args, **_kwargs: True, select_size_fn=lambda *_args, **_kwargs: False, size_candidate_collector=collect_candidates, now=lambda: datetime(2026, 8, 17, 8, 0, tzinfo=timezone.utc), ) adapter.open_goods(GOODS_URL) with self.assertRaises(PddPurchaseSpecResolutionRequired) as raised: adapter.select_options({"color": "黑色", "size": "XXL"}) snapshot = raised.exception.snapshot self.assertEqual(collected, [device]) self.assertEqual(snapshot.goods_id, "753136429979") self.assertEqual(snapshot.dimension_name, "尺码") self.assertEqual(snapshot.observed_at, "2026-08-17T08:00:00Z") self.assertEqual( [(item.raw_text, item.available) for item in snapshot.observations], [("M", True), ("L（售罄）", False), ("XL", True)], ) self.assertEqual( [item.to_dict() for item in snapshot.candidates], [ { "candidate_id": "c1", "raw_text": "M", "options": {"color": "黑色", "size": "M"}, }, { "candidate_id": "c2", "raw_text": "XL", "options": {"color": "黑色", "size": "XL"}, }, ], ) self.assertEqual( snapshot.candidate_snapshot_hash, "fcc35e4429625c830e6b2cbde9c618b63536339c18d9d452c34ec8347d1f1630", ) self.assertEqual( raised.exception.diagnostics["candidate_snapshot_hash"], snapshot.candidate_snapshot_hash, ) adapter.close() def test_apply_resolved_size_rechecks_same_snapshot_before_exact_selection(self): device = FakeDevice() selected = [] snapshot = PurchaseSpecCandidateSnapshot.build( goods_id="753136429979", selected_color="黑色", target_size="XXL", dimension_name="尺码", observations=( PurchaseSizeObservation("3XL【140-165斤】", True), ), observed_at="2026-08-17T08:00:00Z", ) adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), sleeper=lambda _seconds: None, select_size_fn=lambda _device, _xml, target, **_kwargs: ( selected.append(target) or True ), size_candidate_collector=lambda _device: SpecDimension( "size", "尺码", (DimensionValue("3XL【140-165斤】", True),), ), ) adapter.open_goods(GOODS_URL) device.click(790, 2214) adapter.apply_resolved_size(snapshot, snapshot.candidates[0]) self.assertEqual(selected, ["3XL【140-165斤】"]) self.assertEqual( adapter.read_state().selected_options, {"size": "3XL【140-165斤】"}, ) adapter.close() def test_apply_resolved_size_stops_when_candidate_snapshot_changed(self): device = FakeDevice() selections = [] snapshot = PurchaseSpecCandidateSnapshot.build( goods_id="753136429979", selected_color="黑色", target_size="XXL", dimension_name="尺码", observations=(PurchaseSizeObservation("M", True),), observed_at="2026-08-17T08:00:00Z", ) adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), sleeper=lambda _seconds: None, select_size_fn=lambda *_args, **_kwargs: selections.append(True), size_candidate_collector=lambda _device: SpecDimension( "size", "尺码", (DimensionValue("L", True),) ), ) adapter.open_goods(GOODS_URL) device.click(790, 2214) with self.assertRaises(PddPurchaseError) as raised: adapter.apply_resolved_size(snapshot, snapshot.candidates[0]) self.assertEqual( raised.exception.code, "PURCHASE_SPEC_CANDIDATES_CHANGED" ) self.assertEqual(selections, []) adapter.close() def test_candidate_snapshot_is_not_used_when_target_exists_but_click_fails(self): device = FakeDevice() adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), sleeper=lambda _seconds: None, select_color_fn=lambda *_args, **_kwargs: True, select_size_fn=lambda *_args, **_kwargs: False, size_candidate_collector=lambda _device: SpecDimension( "size", "尺码", (DimensionValue("XXL", True),) ), ) adapter.open_goods(GOODS_URL) with self.assertRaises(PddPurchaseError) as raised: adapter.select_options({"color": "黑色", "size": "XXL"}) self.assertNotIsInstance( raised.exception, PddPurchaseSpecResolutionRequired ) self.assertEqual(raised.exception.code, "PURCHASE_OPTIONS_MISMATCH") self.assertEqual( raised.exception.diagnostics["selection_failure"], "selection_unconfirmed", ) adapter.close() def test_incomplete_candidate_scan_remains_normal_purchase_failure(self): device = FakeDevice() def incomplete_scan(_device): raise PddCollectError( "PDD_DATA_SPEC_INCOMPLETE", "达到滑动上限" ) adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), sleeper=lambda _seconds: None, select_color_fn=lambda *_args, **_kwargs: True, select_size_fn=lambda *_args, **_kwargs: False, size_candidate_collector=incomplete_scan, ) adapter.open_goods(GOODS_URL) with self.assertRaises(PddPurchaseError) as raised: adapter.select_options({"color": "黑色", "size": "XXL"}) self.assertNotIsInstance( raised.exception, PddPurchaseSpecResolutionRequired ) self.assertEqual( raised.exception.code, "PURCHASE_SIZE_CANDIDATE_SCAN_INCOMPLETE", ) self.assertEqual( raised.exception.diagnostics["candidate_scan_error"], "PDD_DATA_SPEC_INCOMPLETE", ) adapter.close() def test_invalid_candidate_list_never_requests_remote_resolution(self): for values, expected_reason in ( ( (DimensionValue("M", True), DimensionValue("M", False)), "candidate_duplicated", ), ( tuple( DimensionValue(f"尺码{index}", True) for index in range(101) ), "candidate_count_exceeded", ), ): with self.subTest(expected_reason=expected_reason): device = FakeDevice() adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService( connector=lambda _serial: device ), sleeper=lambda _seconds: None, select_color_fn=lambda *_args, **_kwargs: True, select_size_fn=lambda *_args, **_kwargs: False, size_candidate_collector=lambda _device, items=values: ( SpecDimension("size", "尺码", items) ), ) adapter.open_goods(GOODS_URL) with self.assertRaises(PddPurchaseError) as raised: adapter.select_options( {"color": "黑色", "size": "不存在"} ) self.assertNotIsInstance( raised.exception, PddPurchaseSpecResolutionRequired ) self.assertEqual( raised.exception.code, "PURCHASE_SIZE_CANDIDATE_SCAN_INCOMPLETE", ) self.assertEqual( raised.exception.diagnostics["candidate_validation"], expected_reason, ) adapter.close() def test_size_without_selected_color_never_requests_remote_resolution(self): device = FakeDevice() scans = [] adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), sleeper=lambda _seconds: None, select_size_fn=lambda *_args, **_kwargs: False, size_candidate_collector=lambda _device: scans.append(True), ) adapter.open_goods(GOODS_URL) with self.assertRaises(PddPurchaseError) as raised: adapter.select_options({"size": "XXL"}) self.assertEqual(raised.exception.code, "PURCHASE_OPTIONS_MISMATCH") self.assertEqual(scans, []) adapter.close() def test_contextual_confirm_panel_is_reliable_without_clicking_confirm(self): device = ContextualConfirmPanelDevice() calls = [] def select_color_fn(_device, _xml, target, **_kwargs): calls.append(("color", target)) return target == "黑色" def select_size_fn(_device, _xml, target, **_kwargs): calls.append(("size", target)) return target == "均码" adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), sleeper=lambda _seconds: None, select_color_fn=select_color_fn, select_size_fn=select_size_fn, ) adapter.open_goods(GOODS_URL) adapter.select_options({"color": "黑色", "size": "均码"}) state = adapter.read_state() adapter.stop_before_submit() self.assertEqual(state.page_kind, "order_confirmation") self.assertEqual(state.submit_candidate_count, 1) self.assertEqual(calls, [("color", "黑色"), ("size", "均码")]) # dry-run 只点击商品页采购入口，不点击底部“确定”。 self.assertEqual(device.clicks, [(790, 2214)]) adapter.close() def test_purchase_panel_recognizes_style_and_package_headings(self): xml_data = contextual_confirm_panel_xml().replace( 'text="颜色分类"', 'text="款式"' ).replace('text="尺码"', 'text="套餐"') root = ET.fromstring(xml_data) self.assertEqual(_page_kind(root, ""), "order_confirmation") self.assertEqual(len(_final_submit_targets(root)), 1) def test_plain_confirm_dialog_is_not_a_submit_target(self): root = ET.fromstring(""" """) self.assertEqual(_page_kind(root, ""), "unknown") self.assertEqual(_final_submit_targets(root), []) def test_live_adapter_clicks_contextual_confirm_only_once(self): device = ContextualConfirmPanelDevice() adapter = U2PddLivePurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), sleeper=lambda _seconds: None, select_color_fn=lambda *_args, **_kwargs: True, select_size_fn=lambda *_args, **_kwargs: True, ) adapter.open_goods(GOODS_URL) adapter.select_options({"color": "黑色", "size": "均码"}) adapter.submit_order_once() with self.assertRaises(PddPurchaseError) as raised: adapter.submit_order_once() self.assertEqual(raised.exception.code, "PURCHASE_SUBMIT_ALREADY_ATTEMPTED") # 采购入口一次，底部“确定”一次。 self.assertEqual(len(device.clicks), 2) self.assertEqual(device.clicks[-1], (540, 2140)) adapter.close() def test_purchase_restores_color_region_before_selecting_color(self): clock = FakeClock() device = VerticallyScrolledColorPanelDevice(starts_at_top=False) selected_xml = [] def select_color_fn(_device, xml_data, _target, **_kwargs): selected_xml.append(xml_data) return True adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, select_color_fn=select_color_fn, ) adapter.open_goods(GOODS_URL) adapter.select_options({"color": "黑色"}) self.assertEqual(len(device.swipes), 1) self.assertLess(device.swipes[0][1], device.swipes[0][3]) self.assertEqual(len(selected_xml), 1) self.assertIn('text="颜色分类"', selected_xml[0]) adapter.close() def test_purchase_does_not_scroll_when_color_region_is_already_visible(self): device = VerticallyScrolledColorPanelDevice(starts_at_top=True) adapter = self._live_adapter( device, [], select_color_fn=lambda *_args, **_kwargs: True, ) adapter.open_goods(GOODS_URL) adapter.select_options({"color": "黑色"}) self.assertEqual(device.swipes, []) adapter.close() def test_purchase_stops_when_color_region_cannot_be_restored(self): clock = FakeClock() device = VerticallyScrolledColorPanelDevice( starts_at_top=False, restores_on_swipe=False, ) select_calls = [] adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, select_color_fn=lambda *_args, **_kwargs: select_calls.append(True), ) adapter.open_goods(GOODS_URL) with self.assertRaises(PddPurchaseError) as raised: adapter.select_options({"color": "黑色"}) self.assertEqual( raised.exception.code, "PURCHASE_PANEL_COLOR_REGION_NOT_FOUND", ) self.assertEqual(len(device.swipes), 2) self.assertEqual(select_calls, []) self.assertEqual(len(device.clicks), 1) adapter.close() def test_address_is_split_at_first_separator_before_tagging(self): self.assertEqual( _tagged_shipping_address( "测试省测试市测试区测试路1号-1dd212", "cg2" ), "测试省测试市测试区测试路1号_cg2", ) self.assertEqual( _tagged_shipping_address( "测试省测试市测试区测试路1号_cg2", "cg3" ), "测试省测试市测试区测试路1号_cg3", ) self.assertEqual( _tagged_shipping_address( "测试省-开发区测试路1号", "cg4" ), "测试省_cg4", ) self.assertEqual( _tagged_shipping_address( "测试省测试市测试路1号_旧采购编号", "cg4" ), "测试省测试市测试路1号_cg4", ) def test_live_adapter_updates_address_and_returns_to_confirmation(self): clock = FakeClock() device = AddressFlowDevice() adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) adapter.set_quantity(1) adapter.update_shipping_address("cg2") state = adapter.read_state() self.assertEqual( device.address, "测试省测试市测试区测试路1号_cg2" ) self.assertEqual(device.mode, "panel") self.assertEqual(state.page_kind, "order_confirmation") self.assertEqual(state.quantity, 1) self.assertEqual(state.price_cent, 503) self.assertEqual(clock.sleeps.count(1.0), 5) adapter.close() def test_multiple_addresses_modify_only_default_card(self): device = MultipleAddressFlowDevice(default_indexes=(0,)) adapter = self._live_adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) adapter.update_shipping_address("cg8") self.assertEqual(device.address_modify_clicks, [(948, 571)]) self.assertEqual(device.address_back_clicks, [(108, 164)]) self.assertEqual(device.mode, "panel") adapter.close() def test_address_list_can_load_after_panel_title(self): device = DelayedMultipleAddressFlowDevice(loading_reads=2) adapter = self._live_adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) adapter.update_shipping_address("cg11") self.assertGreaterEqual(device.address_panel_reads, 4) self.assertEqual(device.address_modify_clicks, [(948, 571)]) self.assertEqual(device.mode, "panel") adapter.close() def test_address_list_missing_until_timeout_has_structural_diagnostics(self): clock = FakeClock() device = DelayedMultipleAddressFlowDevice(loading_reads=100) with tempfile.TemporaryDirectory() as directory: adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, panel_timeout=1.0, artifact_directory=Path(directory), ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) with self.assertRaises(PddPurchaseError) as raised: adapter.update_shipping_address("cg11") diagnostics = raised.exception.diagnostics self.assertEqual( raised.exception.code, "PURCHASE_ADDRESS_EDIT_AMBIGUOUS" ) self.assertEqual(diagnostics["address_card_count"], 0) self.assertEqual(diagnostics["default_card_count"], 0) self.assertEqual(diagnostics["candidate_count"], 0) self.assertEqual(diagnostics["stable_reads"], 0) self.assertEqual(device.address_modify_clicks, []) artifact = diagnostics["artifacts"][0] self.assertTrue(Path(artifact["path"]).is_file()) adapter.close() def test_default_address_can_be_second_card(self): device = MultipleAddressFlowDevice(default_indexes=(1,)) adapter = self._live_adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) adapter.update_shipping_address("cg8") self.assertEqual(device.address_modify_clicks, [(948, 1000)]) self.assertEqual(device.address_back_clicks, [(108, 164)]) self.assertEqual(device.mode, "panel") adapter.close() def test_multiple_addresses_without_unique_default_stop_safely(self): for default_indexes in ((), (0, 1)): with self.subTest(default_indexes=default_indexes): device = MultipleAddressFlowDevice( default_indexes=default_indexes ) clock = FakeClock() adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, panel_timeout=1.0, ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) with self.assertRaises(PddPurchaseError) as raised: adapter.update_shipping_address("cg8") self.assertEqual( raised.exception.code, "PURCHASE_ADDRESS_EDIT_AMBIGUOUS", ) self.assertEqual(device.address_modify_clicks, []) self.assertEqual(device.address_back_clicks, []) self.assertEqual( raised.exception.diagnostics["address_card_count"], 2 ) self.assertEqual( raised.exception.diagnostics["default_card_count"], len(default_indexes), ) adapter.close() def test_default_address_with_multiple_modify_targets_stops_safely(self): device = AmbiguousDefaultAddressDevice() clock = FakeClock() adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, panel_timeout=1.0, ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) with self.assertRaises(PddPurchaseError) as raised: adapter.update_shipping_address("cg8") self.assertEqual( raised.exception.code, "PURCHASE_ADDRESS_EDIT_AMBIGUOUS" ) self.assertEqual(raised.exception.diagnostics["candidate_count"], 2) self.assertEqual(device.address_modify_clicks, []) adapter.close() def test_changing_default_target_never_clicks_until_timeout(self): clock = FakeClock() device = AlternatingDefaultAddressDevice() adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, panel_timeout=1.0, ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) with self.assertRaises(PddPurchaseError) as raised: adapter.update_shipping_address("cg11") self.assertEqual(raised.exception.code, "PURCHASE_ADDRESS_PAGE_TIMEOUT") self.assertEqual(raised.exception.diagnostics["candidate_count"], 1) self.assertLessEqual(raised.exception.diagnostics["stable_reads"], 1) self.assertEqual(device.address_modify_clicks, []) adapter.close() def test_saved_address_panel_can_hide_full_address(self): device = HiddenSavedAddressDevice() adapter = self._live_adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) adapter.update_shipping_address("cg2") self.assertEqual(device.mode, "panel") self.assertEqual(device.presses, []) adapter.close() def test_address_save_waits_for_stable_target_before_single_click(self): clock = FakeClock() device = DelayedStableSaveTargetDevice() adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) adapter.update_shipping_address("cg2") self.assertGreaterEqual(device.save_form_reads, 5) self.assertEqual(device.save_clicks, 1) self.assertEqual(device.mode, "panel") self.assertGreaterEqual(clock.now, 0.7) adapter.close() def test_unstable_address_save_target_is_never_clicked(self): clock = FakeClock() device = NeverStableSaveTargetDevice() expected_address = "测试省测试市测试区测试路1号_cg2" adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, panel_timeout=1.0, ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) with self.assertRaises(PddPurchaseError) as raised: adapter.update_shipping_address("cg2") self.assertEqual( raised.exception.code, "PURCHASE_ADDRESS_SAVE_NOT_READY", ) self.assertEqual(device.save_clicks, 0) self.assertNotIn(expected_address, repr(raised.exception.diagnostics)) adapter.close() def test_lost_save_click_reports_not_effective_without_retry(self): clock = FakeClock() device = LostSaveClickAddressDevice() expected_address = "测试省测试市测试区测试路1号_cg2" adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, panel_timeout=1.0, ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) with self.assertRaises(PddPurchaseError) as raised: adapter.update_shipping_address("cg2") self.assertEqual( raised.exception.code, "PURCHASE_ADDRESS_SAVE_NOT_EFFECTIVE", ) self.assertEqual(raised.exception.diagnostics["observed_page"], "edit") self.assertEqual(device.save_clicks, 1) self.assertNotIn(expected_address, repr(raised.exception.diagnostics)) adapter.close() def test_coloros_overlay_can_disappear_without_system_back(self): device = AddressSaveOverlayDevice(clear_after_reads=2) adapter = self._live_adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) adapter.update_shipping_address("cg2") self.assertEqual(device.save_clicks, 1) self.assertEqual(device.presses, []) self.assertEqual(device.mode, "panel") adapter.close() def test_persistent_coloros_overlay_is_dismissed_with_one_system_back(self): clock = FakeClock() device = AddressSaveOverlayDevice() adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) adapter.update_shipping_address("cg2") self.assertEqual(device.save_clicks, 1) self.assertEqual(device.presses, ["back"]) self.assertEqual(device.mode, "panel") self.assertEqual(clock.sleeps.count(1.0), 6) adapter.close() def test_coloros_overlay_is_never_dismissed_more_than_once(self): clock = FakeClock() device = PersistentAddressSaveOverlayDevice() adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, panel_timeout=1.0, ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) with self.assertRaises(PddPurchaseError) as raised: adapter.update_shipping_address("cg2") self.assertEqual(raised.exception.code, "PURCHASE_ADDRESS_PAGE_TIMEOUT") self.assertTrue( raised.exception.diagnostics["coloros_password_overlay_seen"] ) self.assertEqual(raised.exception.diagnostics["system_back_attempts"], 1) self.assertEqual(device.save_clicks, 1) self.assertEqual(device.presses, ["back"]) adapter.close() def test_unknown_page_after_save_does_not_press_back_or_click_again(self): clock = FakeClock() device = UnknownAfterAddressSaveDevice() adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, panel_timeout=1.0, ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) with self.assertRaises(PddPurchaseError) as raised: adapter.update_shipping_address("cg2") self.assertEqual(raised.exception.code, "PURCHASE_ADDRESS_PAGE_TIMEOUT") self.assertEqual(raised.exception.diagnostics["observed_page"], "unknown") self.assertFalse( raised.exception.diagnostics["coloros_password_overlay_seen"] ) self.assertEqual(raised.exception.diagnostics["system_back_attempts"], 0) self.assertEqual(device.save_clicks, 1) self.assertEqual(device.presses, []) adapter.close() def test_address_edit_retries_once_when_first_modify_click_is_lost(self): device = LostFirstModifyClickAddressDevice() adapter = self._live_adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) adapter.update_shipping_address("cg2") self.assertEqual(device.modify_clicks, 2) self.assertEqual(device.address, "测试省测试市测试区测试路1号_cg2") self.assertEqual(device.mode, "panel") adapter.close() def test_address_edit_stops_after_two_lost_modify_clicks(self): clock = FakeClock() device = NeverOpenAddressEditDevice() original_address = device.address with tempfile.TemporaryDirectory() as directory: adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, panel_timeout=1.0, artifact_directory=Path(directory), ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) with self.assertRaises(PddPurchaseError) as raised: adapter.update_shipping_address("cg2") self.assertEqual( raised.exception.code, "PURCHASE_ADDRESS_PAGE_TIMEOUT" ) diagnostics = raised.exception.diagnostics self.assertEqual(diagnostics["expected_page"], "edit") self.assertEqual(diagnostics["observed_page"], "panel") self.assertEqual(diagnostics["click_attempts"], 2) self.assertEqual(diagnostics["candidate_count"], 1) self.assertEqual(device.modify_clicks, 2) self.assertEqual(device.address, original_address) artifact = diagnostics["artifacts"][0] content = Path(artifact["path"]).read_text(encoding="utf-8") self.assertNotIn(original_address, content) adapter.close() def test_address_edit_does_not_retry_on_unknown_page(self): clock = FakeClock() device = UnknownAfterModifyAddressDevice() adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, panel_timeout=1.0, ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) with self.assertRaises(PddPurchaseError) as raised: adapter.update_shipping_address("cg2") self.assertEqual(raised.exception.code, "PURCHASE_ADDRESS_PAGE_TIMEOUT") self.assertEqual(raised.exception.diagnostics["observed_page"], "unknown") self.assertEqual(raised.exception.diagnostics["click_attempts"], 1) self.assertEqual(device.modify_clicks, 1) adapter.close() def test_address_edit_does_not_retry_when_modify_target_is_ambiguous(self): clock = FakeClock() device = AmbiguousAfterModifyAddressDevice() adapter = self._live_adapter( device, [], sleeper=clock.sleep, monotonic=clock.monotonic, panel_timeout=1.0, ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) with self.assertRaises(PddPurchaseError) as raised: adapter.update_shipping_address("cg2") self.assertEqual( raised.exception.code, "PURCHASE_ADDRESS_EDIT_AMBIGUOUS" ) self.assertEqual(raised.exception.diagnostics["candidate_count"], 2) self.assertEqual(raised.exception.diagnostics["click_attempts"], 1) self.assertEqual(device.modify_clicks, 1) adapter.close() def test_reliable_pdd_tree_skips_repeated_app_current(self): device = FakeDevice() adapter = self._adapter(device, []) adapter.open_goods(GOODS_URL) calls_after_open = device.app_current_calls state = adapter.read_state() self.assertEqual(state.page_kind, "goods") self.assertEqual(device.app_current_calls, calls_after_open) adapter.close() def test_ambiguous_tree_falls_back_to_app_current(self): device = FakeDevice() adapter = self._adapter(device, []) adapter.open_goods(GOODS_URL) calls_after_open = device.app_current_calls device.mode = "special" device.special_xml = ( '' ) adapter.read_state() self.assertEqual(device.app_current_calls, calls_after_open + 1) adapter.close() def test_purchase_state_and_option_stages_are_recorded(self): device = FakeDevice() adapter = self._adapter(device, []) records = [] trace = TaskPerformanceTrace(sink=records.append) trace.bind_task("PUR-PERFORMANCE") with trace.activate(): adapter.open_goods(GOODS_URL) adapter.read_state() adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) operations = [record["operation"] for record in records] self.assertIn("purchase_read_state", operations) self.assertIn("purchase_select_color", operations) self.assertIn("purchase_select_size", operations) adapter.close() def test_failed_option_stage_is_recorded_as_failed(self): cases = ( ({"color": "不存在的颜色"}, "purchase_select_color"), ( {"color": "黑色", "size": "不存在的尺码"}, "purchase_select_size", ), ) for options, expected_operation in cases: with self.subTest(operation=expected_operation): device = FakeDevice() adapter = self._adapter(device, []) records = [] trace = TaskPerformanceTrace(sink=records.append) trace.bind_task(f"PUR-{expected_operation}") adapter.open_goods(GOODS_URL) with trace.activate(), self.assertRaises(PddPurchaseError): adapter.select_options(options) matching = [ record for record in records if record["operation"] == expected_operation ] self.assertEqual(len(matching), 1) self.assertEqual(matching[0]["result"], "failed") adapter.close() def test_same_quantity_does_not_focus_editor(self): device = FakeDevice() adapter = self._adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options({"color": "黑色", "size": "3XL【140-165斤】"}) adapter.set_quantity(1) self.assertEqual(device.editor_values, []) self.assertEqual(device.presses, []) self.assertEqual(len(device.clicks), 1) adapter.close() def test_small_quantity_difference_uses_buttons(self): device = FakeDevice() adapter = self._adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options({"color": "黑色", "size": "3XL【140-165斤】"}) adapter.set_quantity(4) self.assertEqual(device.quantity, 4) self.assertEqual(device.editor_values, []) self.assertEqual(device.presses, []) # 采购入口一次，加号三次。 self.assertEqual(len(device.clicks), 4) adapter.close() def test_large_quantity_difference_closes_visible_keyboard_once(self): device = FakeDevice() adapter = self._adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options({"color": "黑色", "size": "3XL【140-165斤】"}) adapter.set_quantity(7) self.assertEqual(device.editor_values, ["7"]) self.assertEqual(device.presses, ["back"]) self.assertFalse(device.keyboard_shown) adapter.close() def test_editor_does_not_press_back_when_keyboard_is_already_hidden(self): device = FakeDevice() device.show_keyboard_on_edit = False adapter = self._adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options({"color": "黑色", "size": "3XL【140-165斤】"}) adapter.set_quantity(7) self.assertEqual(device.editor_values, ["7"]) self.assertEqual(device.presses, []) adapter.close() def test_editor_stops_when_keyboard_cannot_be_closed(self): device = FakeDevice() device.keyboard_can_dismiss = False adapter = self._adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options({"color": "黑色", "size": "3XL【140-165斤】"}) with self.assertRaises(PddPurchaseError) as raised: adapter.set_quantity(7) self.assertEqual(raised.exception.code, "PURCHASE_KEYBOARD_DISMISS_FAILED") self.assertEqual(device.presses, ["back"]) adapter.close() def test_editor_does_not_press_back_when_keyboard_state_is_unknown(self): device = FakeDevice() device.keyboard_status_known = False adapter = self._adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options({"color": "黑色", "size": "3XL【140-165斤】"}) with self.assertRaises(PddPurchaseError) as raised: adapter.set_quantity(7) self.assertEqual(raised.exception.code, "PURCHASE_KEYBOARD_DISMISS_FAILED") self.assertEqual(device.presses, []) adapter.close() def test_editor_stops_when_back_closes_specification_panel(self): device = FakeDevice() device.close_panel_on_back = True adapter = self._adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options({"color": "黑色", "size": "3XL【140-165斤】"}) with self.assertRaises(PddPurchaseError) as raised: adapter.set_quantity(7) self.assertEqual(raised.exception.code, "PURCHASE_PANEL_LOST_AFTER_KEYBOARD") self.assertEqual(device.presses, ["back"]) adapter.close() def test_unsupported_dynamic_option_stops_before_click(self): device = FakeDevice() adapter = self._adapter(device, []) adapter.open_goods(GOODS_URL) with self.assertRaises(PddPurchaseError) as raised: adapter.select_options({"bundle": "标准版"}) self.assertEqual(raised.exception.code, "PURCHASE_OPTIONS_UNSUPPORTED") self.assertEqual(device.clicks, []) adapter.close() def test_cold_start_waits_for_pdd_once(self): device = ColdStartDevice() adapter = self._adapter(device, []) adapter.open_goods(GOODS_URL) self.assertTrue(device.app_started) self.assertEqual(device.app_wait_calls, 1) adapter.close() def test_captcha_stops_without_waiting_or_clicking(self): device = FakeDevice( '' ) adapter = self._adapter(device, []) with self.assertRaises(PddPurchaseError) as raised: adapter.open_goods(GOODS_URL) self.assertEqual(raised.exception.code, "PDD_PAGE_CAPTCHA") self.assertEqual(device.clicks, []) def test_stable_home_reopens_once_then_stops_before_click(self): device = FakeDevice( (FIXTURES / "pdd_home_page.xml").read_text(encoding="utf-8") ) clock = FakeClock() adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), sleeper=clock.sleep, monotonic=clock.monotonic, page_timeout=10.0, ) with self.assertRaises(PddPurchaseError) as raised: adapter.open_goods(GOODS_URL) self.assertEqual("PDD_GOODS_UNAVAILABLE", raised.exception.code) self.assertFalse(raised.exception.retryable) self.assertEqual(2, len(device.opened_urls)) self.assertEqual([], device.clicks) def test_slow_wrong_focus_still_completes_home_reopen_state_machine(self): clock = FakeClock() device = SlowSettingsFocusDevice(clock) device.mode = "special" device.special_xml = ( FIXTURES / "pdd_home_page.xml" ).read_text(encoding="utf-8") adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), sleeper=clock.sleep, monotonic=clock.monotonic, page_timeout=10.0, ) with self.assertRaises(PddPurchaseError) as raised: adapter.open_goods(GOODS_URL) self.assertEqual("PDD_GOODS_UNAVAILABLE", raised.exception.code) self.assertEqual(1, device.app_current_calls) self.assertEqual(2, len(device.opened_urls)) adapter.close() def test_stale_goods_page_is_not_accepted_as_target(self): device = StaleGoodsDevice() clock = FakeClock() adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), sleeper=clock.sleep, monotonic=clock.monotonic, page_timeout=1.0, ) with self.assertRaises(PddPurchaseError) as raised: adapter.open_goods(GOODS_URL) self.assertEqual("PDD_GOODS_IDENTITY_UNCONFIRMED", raised.exception.code) self.assertEqual([], device.clicks) adapter.close() def test_slow_wrong_focus_does_not_consume_panel_timeout(self): clock = FakeClock() device = SlowSettingsFocusDevice(clock) adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService(connector=lambda _serial: device), sleeper=clock.sleep, monotonic=clock.monotonic, panel_timeout=1.0, select_color_fn=lambda *_args, **_kwargs: True, select_size_fn=lambda *_args, **_kwargs: True, ) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) self.assertEqual(1, device.app_current_calls) self.assertEqual(11.0, clock.now) self.assertEqual(1, len(device.clicks)) adapter.close() def test_panel_timeout_saves_sanitized_last_hierarchy(self): clock = FakeClock() device = PanelDoesNotOpenDevice() with tempfile.TemporaryDirectory() as directory: adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService( connector=lambda _serial: device ), sleeper=clock.sleep, monotonic=clock.monotonic, panel_timeout=0.5, artifact_directory=Path(directory), ) adapter.open_goods(GOODS_URL) with self.assertRaises(PddPurchaseError) as raised: adapter.select_options({"color": "黑色"}) diagnostics = raised.exception.diagnostics self.assertEqual("goods", diagnostics["page_kind"]) self.assertGreaterEqual(diagnostics["hierarchy_reads"], 2) self.assertGreaterEqual(diagnostics["panel_hierarchy_reads"], 1) artifact = diagnostics["artifacts"][0] content = Path(artifact["path"]).read_text(encoding="utf-8") self.assertNotIn("5.03", content) self.assertNotIn("立即购买", content) self.assertIn("[已脱敏]", content) adapter.close() def test_color_mismatch_reports_reason_and_saves_sanitized_tree(self): panel_data = ( FIXTURES / "pdd_spec_panel_submit_hint.xml" ).read_text(encoding="utf-8") device = SubmitHintPanelDevice(panel_data) with tempfile.TemporaryDirectory() as directory: adapter = U2PddPurchaseAdapter( "USB-001", device_service=PddDeviceService( connector=lambda _serial: device ), sleeper=lambda _seconds: None, select_color_fn=lambda *_args, **_kwargs: False, artifact_directory=Path(directory), ) adapter.open_goods(GOODS_URL) with self.assertRaises(PddPurchaseError) as raised: adapter.select_options({"color": "测试黑色"}) self.assertEqual( raised.exception.code, "PURCHASE_OPTIONS_MISMATCH" ) self.assertIn("点击颜色后", raised.exception.message) self.assertEqual( raised.exception.diagnostics["selection_failure"], "selection_unconfirmed", ) artifact = raised.exception.diagnostics["artifacts"][0] content = Path(artifact["path"]).read_text(encoding="utf-8") self.assertNotIn("测试黑色", content) self.assertIn("[已脱敏]", content) adapter.close() def test_invalid_non_pdd_url_is_rejected_before_connect(self): device = FakeDevice() adapter = self._adapter(device, []) with self.assertRaises(PddPurchaseError) as raised: adapter.open_goods("https://example.test/goods?goods_id=1") self.assertEqual(raised.exception.code, "PURCHASE_GOODS_URL_INVALID") self.assertEqual(device.opened_urls, []) def test_live_adapter_clicks_unique_submit_target_only_once(self): device = FakeDevice() adapter = self._live_adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) adapter.set_quantity(2) state = adapter.read_state() calls_before_submit = device.app_current_calls adapter.submit_order_once() with self.assertRaises(PddPurchaseError) as raised: adapter.submit_order_once() self.assertEqual(state.submit_candidate_count, 1) self.assertEqual(device.app_current_calls, calls_before_submit + 1) self.assertEqual(raised.exception.code, "PURCHASE_SUBMIT_ALREADY_ATTEMPTED") # 依次点击采购入口、数量加号和最终提交。 self.assertEqual(len(device.clicks), 3) adapter.close() def test_live_adapter_rejects_multiple_submit_targets_without_click(self): device = FakeDevice() adapter = self._live_adapter(device, []) adapter.open_goods(GOODS_URL) adapter.select_options( {"color": "黑色", "size": "3XL【140-165斤】"} ) device.special_xml = panel_xml().replace( "", '', ) device.mode = "special" with self.assertRaises(PddPurchaseError) as raised: adapter.submit_order_once() self.assertEqual(raised.exception.code, "PURCHASE_SUBMIT_TARGET_AMBIGUOUS") self.assertEqual(len(device.clicks), 1) adapter.close() if __name__ == "__main__": unittest.main()