2885 lines
107 KiB
Python
2885 lines
107 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,
|
||
_address_entry_targets,
|
||
_final_submit_targets,
|
||
_needs_color_region_restore,
|
||
_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>"""
|
||
|
||
|
||
def numbered_vertical_color_panel_xml(
|
||
*,
|
||
at_color_top: bool,
|
||
target_selected: bool = False,
|
||
middle_option_count: int = 8,
|
||
extra_scrollable: bool = False,
|
||
) -> str:
|
||
"""模拟摘要不含维度名、当前停在 E09～E16 的纵向颜色列表。"""
|
||
|
||
if at_color_top:
|
||
selected = ' selected="true"' if target_selected else ""
|
||
content = f"""
|
||
<node text="颜色分类" bounds="[36,1012][216,1065]"/>
|
||
<node text="宠粉诱惑-E01【一件】 ¥3.13" clickable="true"{selected}
|
||
enabled="true" visible-to-user="true" bounds="[36,1090][604,1175]"/>
|
||
<node text="宠粉诱惑-E02【一件】 ¥3.37" clickable="true"
|
||
enabled="true" visible-to-user="true" bounds="[36,1205][604,1290]"/>
|
||
<node text="宠粉诱惑-E03【一件】 ¥3.95" clickable="true"
|
||
enabled="true" visible-to-user="true" bounds="[36,1320][604,1405]"/>
|
||
<node text="尺码" bounds="[36,1515][126,1568]"/>
|
||
<node text="均码【70-140斤】" clickable="true"
|
||
enabled="true" visible-to-user="true" bounds="[36,1590][520,1675]"/>
|
||
"""
|
||
else:
|
||
cards = []
|
||
for index in range(middle_option_count):
|
||
number = index + 9
|
||
top = 997 + index * 115
|
||
bottom = top + 85
|
||
cards.append(
|
||
f'<node text="宠粉诱惑-E{number:02d}【一件】 ¥3.13" '
|
||
f'clickable="true" enabled="true" visible-to-user="true" '
|
||
f'bounds="[36,{top}][604,{bottom}]"/>'
|
||
)
|
||
content = "\n".join(cards) + """
|
||
<node text="尺码" bounds="[36,1927][126,1980]"/>
|
||
<node text="均码【70-140斤】" clickable="true"
|
||
enabled="true" visible-to-user="true" bounds="[36,2006][520,2079]"/>
|
||
"""
|
||
extra = """
|
||
<node package="com.xunmeng.pinduoduo"
|
||
class="androidx.recyclerview.widget.RecyclerView"
|
||
scrollable="true" enabled="true" visible-to-user="true"
|
||
bounds="[0,300][1080,900]"/>
|
||
""" if extra_scrollable else ""
|
||
summary = (
|
||
"已选： 宠粉诱惑-E01【一件】 均码【70-140斤】"
|
||
if target_selected
|
||
else "已选： 宠粉诱惑-E10【一件】 均码【70-140斤】"
|
||
)
|
||
return f"""<hierarchy>
|
||
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2376]">
|
||
<node text="{summary}" bounds="[396,731][1053,845]"/>
|
||
<node class="android.widget.EditText" text="1" bounds="[462,881][540,959]"/>
|
||
<node content-desc="减少数量" clickable="true" bounds="[396,881][462,959]"/>
|
||
<node content-desc="增加数量" clickable="true" bounds="[540,881][606,959]"/>
|
||
<node package="com.xunmeng.pinduoduo"
|
||
class="androidx.recyclerview.widget.RecyclerView"
|
||
scrollable="true" enabled="true" visible-to-user="true"
|
||
bounds="[0,997][1080,2079]">
|
||
{content}
|
||
</node>
|
||
{extra}
|
||
<node text="提交订单 ¥3.13" clickable="true"
|
||
enabled="true" visible-to-user="true" bounds="[0,2181][1080,2328]"/>
|
||
</node>
|
||
</hierarchy>"""
|
||
|
||
|
||
def size_only_panel_xml() -> str:
|
||
"""单规格尺码面板不能被结构兜底误认为颜色列表中部。"""
|
||
|
||
return """<hierarchy>
|
||
<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2376]">
|
||
<node text="已选： 均码" bounds="[396,731][1053,845]"/>
|
||
<node content-desc="增加数量" clickable="true" bounds="[540,881][606,959]"/>
|
||
<node package="com.xunmeng.pinduoduo"
|
||
class="androidx.recyclerview.widget.RecyclerView"
|
||
scrollable="true" enabled="true" visible-to-user="true"
|
||
bounds="[0,997][1080,2079]">
|
||
<node text="尺码" bounds="[36,1012][126,1065]"/>
|
||
<node text="均码" clickable="true" enabled="true"
|
||
visible-to-user="true" bounds="[36,1090][520,1175]"/>
|
||
<node text="大码" clickable="true" enabled="true"
|
||
visible-to-user="true" bounds="[36,1205][520,1290]"/>
|
||
</node>
|
||
<node text="提交订单 ¥3.13" clickable="true"
|
||
enabled="true" visible-to-user="true" 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 SoldOutRefreshPurchaseDevice(FakeDevice):
|
||
"""模拟采购深链先显示售罄，下拉一次后恢复或继续售罄。"""
|
||
|
||
def __init__(self, sold_out_xml: str, *, persistent: bool = False) -> None:
|
||
super().__init__(sold_out_xml)
|
||
self.sold_out_xml = sold_out_xml
|
||
self.persistent = persistent
|
||
self.refreshed = False
|
||
self.swipes = []
|
||
|
||
def dump_hierarchy(self):
|
||
if not self.has_opened:
|
||
return '<hierarchy><node package="com.xunmeng.pinduoduo" /></hierarchy>'
|
||
if not self.refreshed or self.persistent:
|
||
return self.sold_out_xml
|
||
return home_xml()
|
||
|
||
def swipe(self, x1, y1, x2, y2, duration=0.0):
|
||
self.swipes.append((x1, y1, x2, y2, duration))
|
||
if y2 > y1 and x1 == x2:
|
||
self.refreshed = True
|
||
|
||
|
||
class VerticallyScrolledColorPanelDevice(FakeDevice):
|
||
"""模拟 PDD 记住规格面板纵向位置。"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
starts_at_top: bool,
|
||
restores_on_swipe: bool = True,
|
||
selected_color: str = "",
|
||
show_selected_summary: bool = False,
|
||
):
|
||
super().__init__()
|
||
self.at_color_top = starts_at_top
|
||
self.restores_on_swipe = restores_on_swipe
|
||
self.selected_color = selected_color
|
||
self.selected_size = ""
|
||
self.show_selected_summary = show_selected_summary
|
||
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":
|
||
xml_data = vertically_scrolled_purchase_panel_xml(
|
||
at_color_top=self.at_color_top
|
||
)
|
||
if self.at_color_top and self.selected_color == "黑色":
|
||
xml_data = xml_data.replace(
|
||
'<node text="黑色" clickable="true"',
|
||
'<node text="黑色" selected="true" clickable="true"',
|
||
)
|
||
if self.at_color_top and self.selected_size == "均码":
|
||
xml_data = xml_data.replace(
|
||
'<node text="均码" clickable="true"',
|
||
'<node text="均码" selected="true" clickable="true"',
|
||
)
|
||
if (
|
||
not self.at_color_top
|
||
and self.show_selected_summary
|
||
and self.selected_color
|
||
):
|
||
summary = self.selected_color
|
||
if self.selected_size:
|
||
summary = f"{summary} {self.selected_size}"
|
||
xml_data = xml_data.replace(
|
||
'text="请选择: 颜色分类"',
|
||
f'text="已选： {summary}"',
|
||
)
|
||
return xml_data
|
||
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 NumberedVerticalColorPanelDevice(FakeDevice):
|
||
"""模拟目标 E01 在纵向列表顶部、当前视口停在 E09～E16。"""
|
||
|
||
def __init__(self) -> None:
|
||
super().__init__()
|
||
self.at_color_top = False
|
||
self.target_selected = False
|
||
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 numbered_vertical_color_panel_xml(
|
||
at_color_top=self.at_color_top,
|
||
target_selected=self.target_selected,
|
||
)
|
||
return super().dump_hierarchy()
|
||
|
||
def click(self, x, y):
|
||
self.clicks.append((x, y))
|
||
if self.mode == "panel" and self.at_color_top and 1090 <= y <= 1175:
|
||
self.target_selected = True
|
||
return
|
||
self.mode = "panel"
|
||
|
||
def swipe(self, x1, y1, x2, y2, duration=0.0):
|
||
self.swipes.append((x1, y1, x2, y2, duration))
|
||
if y2 > y1:
|
||
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 duplicate_address_entry_xml(*, separated: bool = False) -> str:
|
||
"""模拟同一卡片重复节点，或两个位置明显不同的真实候选。"""
|
||
|
||
second_top, second_bottom = ((620, 727) if separated else (366, 472))
|
||
return f"""<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="[0,366][1080,473]">
|
||
<node package="com.xunmeng.pinduoduo"
|
||
text="测试用户，000****0000" bounds="[412,366][993,384]"/>
|
||
</node>
|
||
<node package="com.xunmeng.pinduoduo" clickable="true"
|
||
enabled="true" visible-to-user="true"
|
||
bounds="[0,{second_top}][1080,{second_bottom}]">
|
||
<node package="com.xunmeng.pinduoduo"
|
||
text="测试用户，111****1111"
|
||
bounds="[93,{second_top + 30}][960,{second_top + 76}]"/>
|
||
</node>
|
||
</node>
|
||
</hierarchy>"""
|
||
|
||
|
||
def clipped_address_confirmation_xml(address: str) -> str:
|
||
"""首次地址卡片被顶部裁剪时的脱敏控件树。"""
|
||
|
||
return address_confirmation_xml(address).replace(
|
||
'<node text="测试用户，000****0000" bounds="[300,330][800,380]"/>',
|
||
"",
|
||
).replace(
|
||
'<node package="com.xunmeng.pinduoduo" bounds="[0,0][1080,2376]">',
|
||
'<node package="com.xunmeng.pinduoduo" scrollable="true" '
|
||
'visible-to-user="true" bounds="[0,0][1080,2079]">',
|
||
)
|
||
|
||
|
||
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 ClippedAddressFlowDevice(AddressFlowDevice):
|
||
"""模拟规格面板下滑后才显示联系人和手机号。"""
|
||
|
||
def __init__(self) -> None:
|
||
super().__init__()
|
||
self.swipes = []
|
||
self.address_visible = False
|
||
|
||
def dump_hierarchy(self):
|
||
if self.mode == "panel" and not self.address_visible:
|
||
return clipped_address_confirmation_xml(self.address)
|
||
return super().dump_hierarchy()
|
||
|
||
def swipe(self, x1, y1, x2, y2, duration=0.0):
|
||
self.swipes.append((x1, y1, x2, y2, duration))
|
||
if y2 > y1:
|
||
self.address_visible = True
|
||
|
||
|
||
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 test_address_entry_merges_one_pixel_duplicate_nodes(self):
|
||
root = ET.fromstring(duplicate_address_entry_xml())
|
||
|
||
self.assertEqual(_address_entry_targets(root), [(0, 366, 1080, 472)])
|
||
|
||
def test_address_entry_keeps_separate_cards_ambiguous(self):
|
||
root = ET.fromstring(duplicate_address_entry_xml(separated=True))
|
||
|
||
self.assertEqual(len(_address_entry_targets(root)), 2)
|
||
|
||
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_uses_complete_summary_without_color_restore(self):
|
||
clock = FakeClock()
|
||
device = VerticallyScrolledColorPanelDevice(
|
||
starts_at_top=False,
|
||
selected_color="黑色",
|
||
show_selected_summary=True,
|
||
)
|
||
selected_sizes = []
|
||
snapshot = PurchaseSpecCandidateSnapshot.build(
|
||
goods_id="753136429979",
|
||
selected_color="黑色",
|
||
target_size="50公斤",
|
||
dimension_name="尺码",
|
||
observations=(PurchaseSizeObservation("均码", True),),
|
||
observed_at="2026-08-18T04:00:00Z",
|
||
)
|
||
|
||
def select_size(_device, _xml, target, **_kwargs):
|
||
selected_sizes.append(target)
|
||
device.selected_size = target
|
||
return True
|
||
|
||
adapter = U2PddPurchaseAdapter(
|
||
"USB-001",
|
||
device_service=PddDeviceService(connector=lambda _serial: device),
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
select_size_fn=select_size,
|
||
size_candidate_collector=lambda _device: SpecDimension(
|
||
"size", "尺码", (DimensionValue("均码", True),)
|
||
),
|
||
)
|
||
adapter.open_goods(GOODS_URL)
|
||
device.mode = "panel"
|
||
|
||
adapter.apply_resolved_size(snapshot, snapshot.candidates[0])
|
||
|
||
self.assertEqual(device.swipes, [])
|
||
self.assertEqual(selected_sizes, ["均码"])
|
||
adapter.close()
|
||
|
||
def test_apply_resolved_size_restores_color_when_summary_is_incomplete(self):
|
||
clock = FakeClock()
|
||
device = VerticallyScrolledColorPanelDevice(
|
||
starts_at_top=False,
|
||
selected_color="黑色",
|
||
)
|
||
selected_sizes = []
|
||
snapshot = PurchaseSpecCandidateSnapshot.build(
|
||
goods_id="753136429979",
|
||
selected_color="黑色",
|
||
target_size="50公斤",
|
||
dimension_name="尺码",
|
||
observations=(PurchaseSizeObservation("均码", True),),
|
||
observed_at="2026-08-18T04:00:00Z",
|
||
)
|
||
|
||
def select_size(_device, _xml, target, **_kwargs):
|
||
selected_sizes.append(target)
|
||
device.selected_size = target
|
||
return True
|
||
|
||
adapter = U2PddPurchaseAdapter(
|
||
"USB-001",
|
||
device_service=PddDeviceService(connector=lambda _serial: device),
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
select_size_fn=select_size,
|
||
size_candidate_collector=lambda _device: SpecDimension(
|
||
"size", "尺码", (DimensionValue("均码", True),)
|
||
),
|
||
)
|
||
adapter.open_goods(GOODS_URL)
|
||
device.mode = "panel"
|
||
|
||
adapter.apply_resolved_size(snapshot, snapshot.candidates[0])
|
||
|
||
self.assertEqual(len(device.swipes), 1)
|
||
self.assertLess(device.swipes[0][1], device.swipes[0][3])
|
||
self.assertEqual(selected_sizes, ["均码"])
|
||
adapter.close()
|
||
|
||
def test_apply_resolved_size_stops_when_color_fallback_cannot_be_restored(self):
|
||
clock = FakeClock()
|
||
device = VerticallyScrolledColorPanelDevice(
|
||
starts_at_top=False,
|
||
restores_on_swipe=False,
|
||
selected_color="黑色",
|
||
)
|
||
selected_sizes = []
|
||
candidate_scans = []
|
||
snapshot = PurchaseSpecCandidateSnapshot.build(
|
||
goods_id="753136429979",
|
||
selected_color="黑色",
|
||
target_size="50公斤",
|
||
dimension_name="尺码",
|
||
observations=(PurchaseSizeObservation("均码", True),),
|
||
observed_at="2026-08-18T04:00:00Z",
|
||
)
|
||
|
||
def select_size(_device, _xml, target, **_kwargs):
|
||
selected_sizes.append(target)
|
||
device.selected_size = target
|
||
return True
|
||
|
||
def collect_candidates(_device):
|
||
candidate_scans.append(True)
|
||
return SpecDimension(
|
||
"size", "尺码", (DimensionValue("均码", True),)
|
||
)
|
||
|
||
adapter = U2PddPurchaseAdapter(
|
||
"USB-001",
|
||
device_service=PddDeviceService(connector=lambda _serial: device),
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
select_size_fn=select_size,
|
||
size_candidate_collector=collect_candidates,
|
||
)
|
||
adapter.open_goods(GOODS_URL)
|
||
device.mode = "panel"
|
||
|
||
with self.assertRaises(PddPurchaseError) as raised:
|
||
adapter.apply_resolved_size(snapshot, snapshot.candidates[0])
|
||
|
||
self.assertEqual(
|
||
raised.exception.code,
|
||
"PURCHASE_PANEL_COLOR_REGION_NOT_FOUND",
|
||
)
|
||
self.assertEqual(len(device.swipes), 2)
|
||
self.assertEqual(candidate_scans, [True])
|
||
self.assertEqual(selected_sizes, ["均码"])
|
||
adapter.close()
|
||
|
||
def test_apply_resolved_size_rechecks_panel_after_selecting_returned_size(self):
|
||
device = FakeDevice()
|
||
resolved_size = "均码【80-120斤】"
|
||
snapshot = PurchaseSpecCandidateSnapshot.build(
|
||
goods_id="753136429979",
|
||
selected_color="黑色",
|
||
target_size="均码【40-60公斤】",
|
||
dimension_name="尺码",
|
||
observations=(PurchaseSizeObservation(resolved_size, True),),
|
||
observed_at="2026-08-18T04:00:00Z",
|
||
)
|
||
|
||
def leave_panel_after_size(*_args, **_kwargs):
|
||
device.special_xml = home_xml()
|
||
return True
|
||
|
||
adapter = U2PddPurchaseAdapter(
|
||
"USB-001",
|
||
device_service=PddDeviceService(connector=lambda _serial: device),
|
||
sleeper=lambda _seconds: None,
|
||
select_size_fn=leave_panel_after_size,
|
||
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(
|
||
"3XL【140-165斤】", resolved_size
|
||
)
|
||
|
||
with self.assertRaises(PddPurchaseError) as raised:
|
||
adapter.apply_resolved_size(snapshot, snapshot.candidates[0])
|
||
|
||
self.assertEqual(raised.exception.code, "PURCHASE_SPEC_CONTEXT_CHANGED")
|
||
self.assertEqual(
|
||
raised.exception.diagnostics["context_check"],
|
||
"page_kind_after_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_restores_plain_summary_and_matches_traditional_target(self):
|
||
clock = FakeClock()
|
||
device = NumberedVerticalColorPanelDevice()
|
||
adapter = U2PddPurchaseAdapter(
|
||
"USB-001",
|
||
device_service=PddDeviceService(connector=lambda _serial: device),
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
)
|
||
adapter.open_goods(GOODS_URL)
|
||
|
||
adapter.select_options({"color": "寵粉誘惑-E01【一件】"})
|
||
|
||
self.assertTrue(device.at_color_top)
|
||
self.assertTrue(device.target_selected)
|
||
self.assertEqual(len(device.swipes), 1)
|
||
# 一次打开规格面板、一次点击唯一的简体 E01 卡片。
|
||
self.assertEqual(len(device.clicks), 2)
|
||
adapter.close()
|
||
|
||
def test_plain_summary_restore_requires_strong_first_dimension_structure(self):
|
||
cases = (
|
||
(
|
||
"valid_vertical_first_dimension",
|
||
numbered_vertical_color_panel_xml(at_color_top=False),
|
||
True,
|
||
),
|
||
(
|
||
"only_one_candidate_before_size",
|
||
numbered_vertical_color_panel_xml(
|
||
at_color_top=False,
|
||
middle_option_count=1,
|
||
),
|
||
False,
|
||
),
|
||
(
|
||
"multiple_scrollables",
|
||
numbered_vertical_color_panel_xml(
|
||
at_color_top=False,
|
||
extra_scrollable=True,
|
||
),
|
||
False,
|
||
),
|
||
("size_only", size_only_panel_xml(), False),
|
||
)
|
||
for name, xml_data, expected in cases:
|
||
with self.subTest(name=name):
|
||
self.assertEqual(
|
||
_needs_color_region_restore(ET.fromstring(xml_data)),
|
||
expected,
|
||
)
|
||
|
||
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_address_entry_scrolls_down_once_when_contact_is_clipped(self):
|
||
device = ClippedAddressFlowDevice()
|
||
adapter = self._live_adapter(device, [])
|
||
adapter.open_goods(GOODS_URL)
|
||
adapter.select_options(
|
||
{"color": "黑色", "size": "3XL【140-165斤】"}
|
||
)
|
||
|
||
adapter.update_shipping_address("cg12")
|
||
|
||
self.assertEqual(len(device.swipes), 1)
|
||
self.assertLess(device.swipes[0][1], device.swipes[0][3])
|
||
self.assertEqual(
|
||
device.address, "测试省测试市测试区测试路1号_cg12"
|
||
)
|
||
self.assertEqual(device.mode, "panel")
|
||
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_sold_out_page_pulls_down_once_then_recovers_goods(self):
|
||
sold_out_xml = (
|
||
FIXTURES / "pdd_sold_out_recommendations.xml"
|
||
).read_text(encoding="utf-8")
|
||
device = SoldOutRefreshPurchaseDevice(sold_out_xml)
|
||
adapter = self._adapter(device, [])
|
||
|
||
adapter.open_goods(GOODS_URL)
|
||
|
||
self.assertEqual(len(device.swipes), 1)
|
||
x1, y1, x2, y2, duration = device.swipes[0]
|
||
self.assertEqual(x1, x2)
|
||
self.assertGreater(y2, y1)
|
||
self.assertEqual(duration, 0.5)
|
||
self.assertEqual(device.clicks, [])
|
||
adapter.close()
|
||
|
||
def test_sold_out_after_one_refresh_is_unavailable(self):
|
||
sold_out_xml = (
|
||
FIXTURES / "pdd_sold_out_recommendations.xml"
|
||
).read_text(encoding="utf-8")
|
||
device = SoldOutRefreshPurchaseDevice(
|
||
sold_out_xml,
|
||
persistent=True,
|
||
)
|
||
adapter = self._adapter(device, [])
|
||
|
||
with self.assertRaises(PddPurchaseError) as raised:
|
||
adapter.open_goods(GOODS_URL)
|
||
|
||
self.assertEqual(raised.exception.code, "PDD_GOODS_UNAVAILABLE")
|
||
self.assertFalse(raised.exception.retryable)
|
||
self.assertEqual(
|
||
raised.exception.diagnostics,
|
||
{"page_reason": "out_of_stock", "refresh_attempts": 1},
|
||
)
|
||
self.assertEqual(len(device.swipes), 1)
|
||
self.assertEqual(device.clicks, [])
|
||
adapter.close()
|
||
|
||
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()
|