2351 lines
86 KiB
Python
2351 lines
86 KiB
Python
"""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 """<hierarchy>
|
|
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2376]">
|
|
<node package="com.xunmeng.pinduoduo" clickable="true"
|
|
enabled="true" visible-to-user="true"
|
|
bounds="[500,2100][1080,2328]">
|
|
<node package="com.xunmeng.pinduoduo" text="立即购买 ¥5.03"
|
|
enabled="true" visible-to-user="true"
|
|
bounds="[600,2150][1000,2280]"/>
|
|
</node>
|
|
</node>
|
|
</hierarchy>"""
|
|
|
|
|
|
def panel_xml(quantity: int = 1) -> str:
|
|
return f"""<hierarchy>
|
|
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2376]">
|
|
<node text="已选： 黑色 3XL【140-165斤】" bounds="[300,500][1000,560]"/>
|
|
<node content-desc="黑色" selected="true" bounds="[30,900][330,1000]"/>
|
|
<node text="3XL【140-165斤】" selected="true"
|
|
bounds="[40,1200][500,1300]"/>
|
|
<node text="折后¥5.03" bounds="[300,400][600,470]"/>
|
|
<node class="android.widget.EditText" text="{quantity}"
|
|
bounds="[480,700][560,780]"/>
|
|
<node content-desc="减少数量" clickable="true" bounds="[390,700][470,780]"/>
|
|
<node content-desc="增加数量" clickable="true" bounds="[570,700][650,780]"/>
|
|
<node clickable="true" bounds="[0,2181][1080,2328]">
|
|
<node text="提交订单 ¥5.03" enabled="true" visible-to-user="true"
|
|
bounds="[380,2220][700,2290]"/>
|
|
</node>
|
|
</node>
|
|
</hierarchy>"""
|
|
|
|
|
|
def contextual_confirm_panel_xml(quantity: int = 1) -> str:
|
|
"""模拟商品 897186799891 的非滚动“确定”规格面板。"""
|
|
|
|
return f"""<hierarchy>
|
|
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2340]">
|
|
<node package="com.xunmeng.pinduoduo" class="android.widget.RelativeLayout"
|
|
bounds="[0,720][1080,2214]" visible-to-user="true" enabled="true">
|
|
<node package="com.xunmeng.pinduoduo" text="确认款式"
|
|
bounds="[438,746][642,814]"/>
|
|
<node text="券后 ¥8.17" bounds="[324,852][573,919]"/>
|
|
<node text="已选择： 黑色 均码" bounds="[324,1009][1044,1062]"/>
|
|
<node class="android.widget.EditText" text="{quantity}"
|
|
bounds="[408,1080][489,1155]"/>
|
|
<node content-desc="减少数量" clickable="true" enabled="true"
|
|
visible-to-user="true" bounds="[324,1080][402,1155]"/>
|
|
<node content-desc="增加数量" clickable="true" enabled="true"
|
|
visible-to-user="true" bounds="[495,1080][573,1155]"/>
|
|
<node text="颜色分类" bounds="[36,1214][216,1267]"/>
|
|
<node text="黑色" clickable="true" selected="true"
|
|
bounds="[36,1295][276,1382]"/>
|
|
<node text="尺码" bounds="[36,1684][126,1737]"/>
|
|
<node text="均码" clickable="true" selected="true"
|
|
bounds="[36,1765][711,1852]"/>
|
|
<node text="确定" clickable="true" enabled="true"
|
|
visible-to-user="true" bounds="[0,2067][1080,2214]"/>
|
|
</node>
|
|
</node>
|
|
</hierarchy>"""
|
|
|
|
|
|
def vertically_scrolled_purchase_panel_xml(*, at_color_top: bool) -> str:
|
|
"""构造规格面板位于颜色顶部或停在中部的脱敏控件树。"""
|
|
|
|
if at_color_top:
|
|
scroll_content = """
|
|
<node text="颜色分类" bounds="[32,1120][190,1170]"/>
|
|
<node text="黑色" clickable="true" enabled="true"
|
|
visible-to-user="true" bounds="[32,1190][332,1320]"/>
|
|
<node text="白色" clickable="true" enabled="true"
|
|
visible-to-user="true" bounds="[359,1190][659,1320]"/>
|
|
<node text="尺码" bounds="[32,1500][114,1550]"/>
|
|
<node text="均码" clickable="true" enabled="true"
|
|
visible-to-user="true" bounds="[32,1580][391,1660]"/>
|
|
"""
|
|
else:
|
|
scroll_content = """
|
|
<node text="黑色 ¥6.93" clickable="true" enabled="true"
|
|
visible-to-user="true" bounds="[32,1095][332,1460]"/>
|
|
<node text="白色 ¥6.93" clickable="true" enabled="true"
|
|
visible-to-user="true" bounds="[359,1095][659,1460]"/>
|
|
<node text="尺码" bounds="[32,1605][114,1653]"/>
|
|
<node text="均码" clickable="true" selected="true"
|
|
enabled="true" visible-to-user="true"
|
|
bounds="[32,1677][391,1753]"/>
|
|
"""
|
|
return f"""<hierarchy>
|
|
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2376]">
|
|
<node text="确认款式" bounds="[430,650][650,710]"/>
|
|
<node text="请选择: 颜色分类" bounds="[360,770][990,830]"/>
|
|
<node class="android.widget.EditText" text="1"
|
|
bounds="[440,870][510,940]"/>
|
|
<node content-desc="减少数量" clickable="true"
|
|
bounds="[360,870][430,940]"/>
|
|
<node content-desc="增加数量" clickable="true"
|
|
bounds="[515,870][585,940]"/>
|
|
<node package="com.xunmeng.pinduoduo"
|
|
class="androidx.recyclerview.widget.RecyclerView"
|
|
scrollable="true" enabled="true" visible-to-user="true"
|
|
bounds="[0,1095][1020,1918]">
|
|
{scroll_content}
|
|
</node>
|
|
<node text="选择颜色分类后，提交订单"
|
|
bounds="[0,2181][1080,2328]"/>
|
|
</node>
|
|
</hierarchy>"""
|
|
|
|
|
|
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 '<hierarchy><node package="com.xunmeng.pinduoduo" /></hierarchy>'
|
|
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 '<hierarchy><node package="com.xunmeng.pinduoduo" /></hierarchy>'
|
|
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 '<hierarchy><node package="com.xunmeng.pinduoduo" /></hierarchy>'
|
|
if self.mode == "panel":
|
|
return self.panel_data
|
|
return home_xml()
|
|
|
|
|
|
def address_confirmation_xml(address: str) -> str:
|
|
return f"""<hierarchy>
|
|
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2376]">
|
|
<node clickable="true" enabled="true" visible-to-user="true"
|
|
bounds="[0,300][1080,520]">
|
|
<node text="测试用户，000****0000" bounds="[300,330][800,380]"/>
|
|
<node text="{address}" bounds="[120,390][1000,470]"/>
|
|
</node>
|
|
<node text="已选： 黑色 3XL【140-165斤】" bounds="[300,550][1000,610]"/>
|
|
<node content-desc="黑色" selected="true" bounds="[30,900][330,1000]"/>
|
|
<node text="3XL【140-165斤】" selected="true" bounds="[40,1200][500,1300]"/>
|
|
<node text="折后¥5.03" bounds="[300,530][600,590]"/>
|
|
<node class="android.widget.EditText" text="1" bounds="[480,700][560,780]"/>
|
|
<node content-desc="减少数量" clickable="true" bounds="[390,700][470,780]"/>
|
|
<node content-desc="增加数量" clickable="true" bounds="[570,700][650,780]"/>
|
|
<node clickable="true" enabled="true" visible-to-user="true"
|
|
bounds="[0,2181][1080,2328]">
|
|
<node text="提交订单 ¥5.03" enabled="true" visible-to-user="true"
|
|
bounds="[380,2220][700,2290]"/>
|
|
</node>
|
|
</node>
|
|
</hierarchy>"""
|
|
|
|
|
|
def address_panel_xml(address: str) -> str:
|
|
return f"""<hierarchy>
|
|
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2376]">
|
|
<node clickable="true" enabled="true" bounds="[0,120][240,258]">
|
|
<node text="返回" bounds="[42,154][108,223]"/>
|
|
</node>
|
|
<node text="收货地址" bounds="[438,120][642,258]"/>
|
|
<node text="{address}" bounds="[36,400][900,590]"/>
|
|
<node clickable="true" enabled="true" bounds="[900,600][1040,700]">
|
|
<node text="修改" bounds="[930,620][1020,680]"/>
|
|
</node>
|
|
</node>
|
|
</hierarchy>"""
|
|
|
|
|
|
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"""
|
|
<node package="com.xunmeng.pinduoduo"
|
|
class="android.view.ViewGroup" bounds="[0,{top}][1020,{bottom}]">
|
|
<node package="com.xunmeng.pinduoduo" text="{address_text}"
|
|
class="android.widget.TextView" bounds="[32,{top + 120}][825,{top + 240}]"/>
|
|
<node package="com.xunmeng.pinduoduo" clickable="true"
|
|
bounds="[32,{marker_top - 20}][249,{marker_top + 75}]">
|
|
<node package="com.xunmeng.pinduoduo" text="{marker}"
|
|
class="android.widget.TextView"
|
|
bounds="[89,{marker_top}][249,{marker_top + 48}]"/>
|
|
</node>
|
|
<node package="com.xunmeng.pinduoduo" clickable="true"
|
|
bounds="[908,{modify_top}][988,{modify_top + 54}]">
|
|
<node package="com.xunmeng.pinduoduo" text="修改"
|
|
class="android.widget.TextView"
|
|
bounds="[908,{modify_top}][988,{modify_top + 54}]"/>
|
|
</node>
|
|
</node>""")
|
|
return f"""<hierarchy>
|
|
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2376]">
|
|
<node package="com.xunmeng.pinduoduo" clickable="true"
|
|
bounds="[0,102][217,226]">
|
|
<node package="com.xunmeng.pinduoduo" text="返回"
|
|
class="android.widget.TextView" bounds="[38,132][98,195]"/>
|
|
</node>
|
|
<node package="com.xunmeng.pinduoduo" text="收货地址"
|
|
class="android.widget.TextView" bounds="[418,102][602,226]"/>
|
|
<node package="com.xunmeng.pinduoduo"
|
|
class="androidx.recyclerview.widget.RecyclerView"
|
|
bounds="[0,226][1020,2140]">
|
|
{''.join(cards)}
|
|
</node>
|
|
</node>
|
|
<node package="com.android.systemui" content-desc="返回"
|
|
class="android.widget.ImageView" clickable="true"
|
|
bounds="[640,2140][802,2250]"/>
|
|
</hierarchy>"""
|
|
|
|
|
|
def address_edit_xml(address: str) -> str:
|
|
return f"""<hierarchy>
|
|
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2376]">
|
|
<node text="修改收货地址" bounds="[380,350][700,450]"/>
|
|
<node text="收货人" bounds="[120,550][260,610]"/>
|
|
<node class="android.widget.EditText" text="测试用户"
|
|
enabled="true" visible-to-user="true" bounds="[327,510][906,654]"/>
|
|
<node text="手机号码" bounds="[120,700][260,760]"/>
|
|
<node class="android.widget.EditText" text="00000000000"
|
|
enabled="true" visible-to-user="true" bounds="[327,654][906,798]"/>
|
|
<node text="详细地址" bounds="[120,990][300,1043]"/>
|
|
<node class="android.widget.EditText" text="{address}"
|
|
enabled="true" visible-to-user="true" bounds="[327,990][813,1116]"/>
|
|
<node clickable="true" enabled="true" bounds="[120,1194][960,1329]">
|
|
<node text="保存" bounds="[489,1232][591,1291]"/>
|
|
</node>
|
|
</node>
|
|
</hierarchy>"""
|
|
|
|
|
|
def coloros_password_overlay_xml() -> str:
|
|
return """<hierarchy>
|
|
<node package="com.xunmeng.pinduoduo" class="android.widget.FrameLayout"
|
|
bounds="[0,0][1080,2376]"/>
|
|
<node package="com.coloros.codebook" class="android.widget.FrameLayout"
|
|
clickable="true" focusable="true" focused="true"
|
|
visible-to-user="true" bounds="[0,0][1080,2328]">
|
|
<node package="com.coloros.codebook" text="系统提示"
|
|
class="android.widget.TextView" visible-to-user="true"
|
|
bounds="[96,1965][516,2022]"/>
|
|
</node>
|
|
</hierarchy>"""
|
|
|
|
|
|
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 '<hierarchy><node package="com.xunmeng.pinduoduo" /></hierarchy>'
|
|
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 """<hierarchy>
|
|
<node package="com.xunmeng.pinduoduo"
|
|
bounds="[0,0][1080,2376]">
|
|
<node package="com.xunmeng.pinduoduo" clickable="true"
|
|
bounds="[0,102][217,226]">
|
|
<node package="com.xunmeng.pinduoduo" text="返回"
|
|
bounds="[38,132][98,195]"/>
|
|
</node>
|
|
<node package="com.xunmeng.pinduoduo" text="收货地址"
|
|
bounds="[418,102][602,226]"/>
|
|
</node>
|
|
</hierarchy>"""
|
|
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 """<hierarchy>
|
|
<node package="com.example.unknown" text="未知页面"
|
|
focused="true" visible-to-user="true"
|
|
bounds="[0,0][1080,2376]"/>
|
|
</hierarchy>"""
|
|
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 """<hierarchy>
|
|
<node package="com.xunmeng.pinduoduo" text="其他页面"
|
|
bounds="[0,0][1080,2376]"/>
|
|
</hierarchy>"""
|
|
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(
|
|
"</node>\n </node>\n </hierarchy>",
|
|
"""</node>
|
|
<node clickable="true" enabled="true" bounds="[700,600][850,700]">
|
|
<node text="修改" bounds="[730,620][820,680]"/>
|
|
</node>
|
|
</node>
|
|
</hierarchy>""",
|
|
)
|
|
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_accepts_selected_simplified_color_for_traditional_target(self):
|
|
device = FakeDevice()
|
|
selected = []
|
|
traditional_color = "縹緲仙姬綰【黑色】"
|
|
simplified_color = "缥缈仙姬绾【黑色】"
|
|
resolved_size = "均码【80-120斤】"
|
|
snapshot = PurchaseSpecCandidateSnapshot.build(
|
|
goods_id="753136429979",
|
|
selected_color=traditional_color,
|
|
target_size="均碼【40.0-60.0公斤】",
|
|
dimension_name="尺码",
|
|
observations=(PurchaseSizeObservation(resolved_size, True),),
|
|
observed_at="2026-08-18T01:01:36Z",
|
|
)
|
|
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(resolved_size, True),)
|
|
),
|
|
)
|
|
adapter.open_goods(GOODS_URL)
|
|
device.mode = "special"
|
|
device.special_xml = panel_xml().replace(
|
|
"黑色", simplified_color
|
|
).replace("3XL【140-165斤】", resolved_size)
|
|
|
|
adapter.apply_resolved_size(snapshot, snapshot.candidates[0])
|
|
|
|
self.assertEqual(selected, [resolved_size])
|
|
adapter._requested_options = {
|
|
"color": traditional_color,
|
|
"size": resolved_size,
|
|
}
|
|
self.assertEqual(
|
|
adapter.read_state().selected_options,
|
|
{"color": traditional_color, "size": resolved_size},
|
|
)
|
|
adapter.close()
|
|
|
|
def test_apply_resolved_size_distinguishes_page_and_color_context_changes(self):
|
|
resolved_size = "均码【80-120斤】"
|
|
snapshot = PurchaseSpecCandidateSnapshot.build(
|
|
goods_id="753136429979",
|
|
selected_color="縹緲仙姬綰【黑色】",
|
|
target_size="均碼【40.0-60.0公斤】",
|
|
dimension_name="尺码",
|
|
observations=(PurchaseSizeObservation(resolved_size, True),),
|
|
observed_at="2026-08-18T01:01:36Z",
|
|
)
|
|
|
|
cases = (
|
|
("page_kind", home_xml(), "页面已离开规格确认页"),
|
|
(
|
|
"selected_color",
|
|
panel_xml().replace("selected=\"true\"", "selected=\"false\"")
|
|
.replace("已选： 黑色", "已选： 白色")
|
|
.replace("黑色", "缥缈仙姬绾【白色】")
|
|
.replace("3XL【140-165斤】", resolved_size),
|
|
"已选颜色发生变化",
|
|
),
|
|
)
|
|
for context_check, current_xml, expected_message in cases:
|
|
with self.subTest(context_check=context_check):
|
|
device = FakeDevice()
|
|
adapter = U2PddPurchaseAdapter(
|
|
"USB-001",
|
|
device_service=PddDeviceService(
|
|
connector=lambda _serial, current=device: current
|
|
),
|
|
sleeper=lambda _seconds: None,
|
|
select_size_fn=lambda *_args, **_kwargs: True,
|
|
size_candidate_collector=lambda _device: SpecDimension(
|
|
"size", "尺码", (DimensionValue(resolved_size, True),)
|
|
),
|
|
)
|
|
adapter.open_goods(GOODS_URL)
|
|
device.mode = "special"
|
|
device.special_xml = current_xml
|
|
|
|
with self.assertRaises(PddPurchaseError) as raised:
|
|
adapter.apply_resolved_size(snapshot, snapshot.candidates[0])
|
|
|
|
self.assertEqual(
|
|
raised.exception.code, "PURCHASE_SPEC_CONTEXT_CHANGED"
|
|
)
|
|
self.assertIn(expected_message, raised.exception.message)
|
|
self.assertEqual(
|
|
raised.exception.diagnostics["context_check"], context_check
|
|
)
|
|
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("""<hierarchy>
|
|
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2340]">
|
|
<node package="com.xunmeng.pinduoduo" text="提示"
|
|
bounds="[100,800][980,1400]">
|
|
<node package="com.xunmeng.pinduoduo" text="确定"
|
|
clickable="true" enabled="true" visible-to-user="true"
|
|
bounds="[300,1250][780,1380]"/>
|
|
</node>
|
|
</node>
|
|
</hierarchy>""")
|
|
|
|
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 = (
|
|
'<hierarchy><node package="com.xunmeng.pinduoduo" '
|
|
'text="占位"/></hierarchy>'
|
|
)
|
|
|
|
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(
|
|
'<hierarchy><node package="com.xunmeng.pinduoduo" '
|
|
'text="请完成验证" bounds="[0,0][1080,2376]"/></hierarchy>'
|
|
)
|
|
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(
|
|
"</hierarchy>",
|
|
'<node text="确认购买" enabled="true" visible-to-user="true" '
|
|
'bounds="[50,2200][300,2300]"/></hierarchy>',
|
|
)
|
|
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()
|