2114 lines
79 KiB
Python
2114 lines
79 KiB
Python
"""PDD 商品采集与 XML 解析测试，不连接真机。"""
|
||
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
import tempfile
|
||
import unittest
|
||
import xml.etree.ElementTree as ET
|
||
|
||
from src.pdd_collect_service import (
|
||
PddCollectError,
|
||
PddCollectService,
|
||
_is_spec_panel_open,
|
||
parse_goods_page,
|
||
parse_quantity,
|
||
parse_spec_panel,
|
||
)
|
||
from src.pdd_device_service import PddDeviceService
|
||
|
||
|
||
FIXTURES = Path(__file__).parent / "fixtures"
|
||
|
||
|
||
def style_package_spec_xml() -> str:
|
||
"""模拟口罩商品的“款式 + 套餐”规格面板。"""
|
||
|
||
return """<hierarchy>
|
||
<node class="android.widget.FrameLayout" bounds="[0,0][1080,2340]">
|
||
<node class="android.widget.ScrollView" scrollable="true"
|
||
bounds="[0,600][1080,2200]">
|
||
<node text="券后 ¥1.68" bounds="[300,650][600,710]" />
|
||
<node text="已选：成人款 单个装" bounds="[260,720][1000,780]" />
|
||
<node text="款式" bounds="[36,800][180,860]" />
|
||
<node text="成人款" clickable="true" selected="true"
|
||
bounds="[36,900][300,1000]" />
|
||
<node text="儿童款" clickable="true" selected="false"
|
||
bounds="[330,900][594,1000]" />
|
||
<node text="套餐" bounds="[36,1100][180,1160]" />
|
||
<node text="单个装" clickable="true" selected="true"
|
||
bounds="[36,1200][300,1300]" />
|
||
<node text="两个装" clickable="true" selected="false"
|
||
bounds="[330,1200][594,1300]" />
|
||
<node text="确定" clickable="true" bounds="[0,2000][1080,2150]" />
|
||
</node>
|
||
</node>
|
||
</hierarchy>"""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class FakeTask:
|
||
goods_url: str
|
||
goods_id: str | None = None
|
||
|
||
|
||
class FakeCollectDevice:
|
||
def __init__(self, home_xml: str, spec_xml: str):
|
||
self.home_xml = home_xml
|
||
self.spec_xml = spec_xml
|
||
self.panel_open = False
|
||
self.clicks = []
|
||
self.swipes = []
|
||
self.swipe_panel_states = []
|
||
self.app_wait_calls = 0
|
||
self.app_current_calls = 0
|
||
self.opened_url = None
|
||
|
||
def app_current(self):
|
||
self.app_current_calls += 1
|
||
return {"package": "com.xunmeng.pinduoduo"}
|
||
|
||
def app_start(self, _package):
|
||
raise AssertionError("PDD 已经在前台，不应重复启动")
|
||
|
||
def app_wait(self, _package, timeout=10):
|
||
self.app_wait_calls += 1
|
||
return 123
|
||
|
||
def open_url(self, url):
|
||
self.opened_url = url
|
||
|
||
def dump_hierarchy(self):
|
||
if self.opened_url is None:
|
||
return '<hierarchy><node package="com.xunmeng.pinduoduo" /></hierarchy>'
|
||
return self.spec_xml if self.panel_open else self.home_xml
|
||
|
||
def click(self, x, y):
|
||
self.clicks.append((x, y))
|
||
if y > 1600:
|
||
self.panel_open = True
|
||
|
||
def swipe(self, *args, **kwargs):
|
||
self.swipes.append((args, kwargs))
|
||
self.swipe_panel_states.append(self.panel_open)
|
||
|
||
|
||
class GoodsMetadataScrollDevice(FakeCollectDevice):
|
||
"""模拟评价和店铺分别出现在商品页后续屏幕。"""
|
||
|
||
def __init__(self, pages, spec_xml):
|
||
super().__init__(pages[0], spec_xml)
|
||
self.pages = pages
|
||
self.page_index = 0
|
||
|
||
def dump_hierarchy(self):
|
||
if self.opened_url is None and not self.panel_open:
|
||
return super().dump_hierarchy()
|
||
if self.panel_open:
|
||
return self.spec_xml
|
||
return self.pages[self.page_index]
|
||
|
||
def swipe(self, *args, **kwargs):
|
||
super().swipe(*args, **kwargs)
|
||
if not self.panel_open:
|
||
self.page_index = min(self.page_index + 1, len(self.pages) - 1)
|
||
|
||
|
||
class LoadingDevice(FakeCollectDevice):
|
||
def dump_hierarchy(self):
|
||
return '<hierarchy><node text="正在加载" /></hierarchy>'
|
||
|
||
|
||
class DisconnectingDevice(FakeCollectDevice):
|
||
def __init__(self, home_xml, spec_xml):
|
||
super().__init__(home_xml, spec_xml)
|
||
self.hierarchy_calls = 0
|
||
|
||
def dump_hierarchy(self):
|
||
self.hierarchy_calls += 1
|
||
if self.hierarchy_calls > 1:
|
||
raise RuntimeError("device not found")
|
||
return super().dump_hierarchy()
|
||
|
||
|
||
class FocusMismatchDevice(FakeCollectDevice):
|
||
"""系统焦点报告为设置页，但无障碍树实际属于 PDD。"""
|
||
|
||
def app_current(self):
|
||
return {"package": "com.oplus.wirelesssettings"}
|
||
|
||
def app_start(self, _package):
|
||
self.app_started = True
|
||
|
||
|
||
class SlowFocusMismatchDevice(FocusMismatchDevice):
|
||
def __init__(self, home_xml, spec_xml, clock):
|
||
super().__init__(home_xml, spec_xml)
|
||
self.clock = clock
|
||
self.app_current_calls = 0
|
||
|
||
def app_current(self):
|
||
self.app_current_calls += 1
|
||
self.clock.sleep(11.0)
|
||
return {"package": "com.android.settings"}
|
||
|
||
|
||
class PanelDoesNotOpenDevice(FakeCollectDevice):
|
||
def click(self, x, y):
|
||
self.clicks.append((x, y))
|
||
|
||
|
||
class QuickConfirmationDevice(FakeCollectDevice):
|
||
"""点击当前款式后，从快速确认页进入正常规格面板。"""
|
||
|
||
def __init__(self, quick_confirmation_xml, spec_xml):
|
||
super().__init__(quick_confirmation_xml, spec_xml)
|
||
self.opened_url = "https://mobile.yangkeduo.com/goods.html?goods_id=123"
|
||
self.panel_open = False
|
||
|
||
def dump_hierarchy(self):
|
||
return self.spec_xml if self.panel_open else self.home_xml
|
||
|
||
def click(self, x, y):
|
||
self.clicks.append((x, y))
|
||
if 32 <= x <= 470 and 1240 <= y <= 1320:
|
||
self.panel_open = True
|
||
|
||
|
||
class StuckQuickConfirmationDevice(QuickConfirmationDevice):
|
||
"""快速确认页接受点击后仍不打开规格选择区域。"""
|
||
|
||
def click(self, x, y):
|
||
self.clicks.append((x, y))
|
||
|
||
|
||
TRANSIENT_EMPTY_SPEC_XML = """<hierarchy>
|
||
<node class="android.widget.FrameLayout" package="com.xunmeng.pinduoduo"
|
||
bounds="[0,0][1080,2340]">
|
||
<node class="android.widget.LinearLayout" package="com.xunmeng.pinduoduo"
|
||
bounds="[0,0][1080,2214]" />
|
||
</node>
|
||
</hierarchy>"""
|
||
|
||
|
||
class TransientSpecTreeDevice(FakeCollectDevice):
|
||
"""规格面板阶段交替返回临时空树和有效树。"""
|
||
|
||
def __init__(self, home_xml, spec_xml):
|
||
super().__init__(home_xml, spec_xml)
|
||
self.panel_reads = 0
|
||
|
||
def dump_hierarchy(self):
|
||
if self.panel_open:
|
||
self.panel_reads += 1
|
||
if self.panel_reads % 2 == 0:
|
||
return TRANSIENT_EMPTY_SPEC_XML
|
||
return super().dump_hierarchy()
|
||
|
||
|
||
class SpecTreeStaysBlankDevice(FakeCollectDevice):
|
||
"""面板首次有效，随后持续返回空树。"""
|
||
|
||
def __init__(self, home_xml, spec_xml):
|
||
super().__init__(home_xml, spec_xml)
|
||
self.panel_reads = 0
|
||
|
||
def dump_hierarchy(self):
|
||
if self.panel_open:
|
||
self.panel_reads += 1
|
||
if self.panel_reads > 1:
|
||
return TRANSIENT_EMPTY_SPEC_XML
|
||
return super().dump_hierarchy()
|
||
|
||
|
||
class SnakeColorDevice(FakeCollectDevice):
|
||
"""模拟两行横向颜色网格和滚动后才出现的尺码区域。"""
|
||
|
||
prices = {
|
||
"A色": 1000,
|
||
"B色": 1000,
|
||
"C色": 1200,
|
||
"D色": 1300,
|
||
"E色": 1400,
|
||
"F色": 1500,
|
||
}
|
||
|
||
def __init__(self, home_xml):
|
||
super().__init__(home_xml, "")
|
||
self.page = 1
|
||
self.selected_color = "A色"
|
||
self.sizes_visible = False
|
||
self.clicked_colors = []
|
||
|
||
def dump_hierarchy(self):
|
||
if self.opened_url is None and not self.panel_open:
|
||
return super().dump_hierarchy()
|
||
if not self.panel_open:
|
||
return self.home_xml
|
||
return self._spec_xml()
|
||
|
||
def click(self, x, y):
|
||
self.clicks.append((x, y))
|
||
if not self.panel_open:
|
||
self.panel_open = True
|
||
return
|
||
for name, bounds in self._current_color_bounds().items():
|
||
left, top, right, bottom = bounds
|
||
if left <= x <= right and top <= y <= bottom:
|
||
self.selected_color = name
|
||
self.clicked_colors.append(name)
|
||
return
|
||
|
||
def swipe(self, x1, y1, x2, y2, duration=0.35):
|
||
self.swipes.append(((x1, y1, x2, y2), {"duration": duration}))
|
||
self.swipe_panel_states.append(self.panel_open)
|
||
if abs(x2 - x1) > abs(y2 - y1):
|
||
self.page = 1 if x1 > x2 else 0
|
||
else:
|
||
self.sizes_visible = True
|
||
|
||
def _current_color_bounds(self):
|
||
if self.page == 0:
|
||
return {
|
||
"A色": (36, 740, 340, 850),
|
||
"B色": (370, 740, 674, 850),
|
||
"C色": (1044, 740, 1080, 850),
|
||
"D色": (36, 900, 340, 1010),
|
||
"E色": (370, 900, 674, 1010),
|
||
"F色": (1044, 900, 1080, 1010),
|
||
}
|
||
return {
|
||
"A色": (0, 740, 20, 850),
|
||
"B色": (36, 740, 340, 850),
|
||
"C色": (370, 740, 674, 850),
|
||
"D色": (0, 900, 20, 1010),
|
||
"E色": (36, 900, 340, 1010),
|
||
"F色": (370, 900, 674, 1010),
|
||
}
|
||
|
||
def _spec_xml(self):
|
||
color_nodes = []
|
||
for name, bounds in self._current_color_bounds().items():
|
||
left, top, right, bottom = bounds
|
||
selected = "true" if name == self.selected_color else "false"
|
||
color_nodes.append(
|
||
f'<node class="android.view.ViewGroup" content-desc="{name}" '
|
||
f'clickable="true" selected="{selected}" '
|
||
f'bounds="[{left},{top}][{right},{bottom}]" '
|
||
'visible-to-user="true" enabled="true" />'
|
||
)
|
||
size_nodes = ""
|
||
if self.sizes_visible:
|
||
size_nodes = """
|
||
<node class="android.widget.TextView" text="尺码"
|
||
bounds="[36,1120][130,1180]" visible-to-user="true"
|
||
enabled="true" />
|
||
<node class="android.view.ViewGroup" content-desc="M"
|
||
clickable="true" bounds="[36,1210][340,1300]"
|
||
visible-to-user="true" enabled="true" />
|
||
<node class="android.view.ViewGroup" content-desc="L"
|
||
clickable="true" bounds="[370,1210][674,1300]"
|
||
visible-to-user="true" enabled="true" />
|
||
"""
|
||
price = self.prices[self.selected_color] / 100
|
||
return f"""<hierarchy>
|
||
<node class="android.widget.FrameLayout" bounds="[0,0][1080,2000]"
|
||
visible-to-user="true" enabled="true">
|
||
<node class="android.widget.TextView" text="¥{price:.2f}"
|
||
bounds="[390,300][650,380]" visible-to-user="true"
|
||
enabled="true" />
|
||
<node class="androidx.recyclerview.widget.RecyclerView"
|
||
scrollable="true" bounds="[0,600][1080,1700]"
|
||
visible-to-user="true" enabled="true">
|
||
<node class="android.widget.TextView" text="颜色分类"
|
||
bounds="[36,650][200,710]" visible-to-user="true"
|
||
enabled="true" />
|
||
<node class="androidx.recyclerview.widget.RecyclerView"
|
||
scrollable="true" bounds="[36,720][1044,1060]"
|
||
visible-to-user="true" enabled="true">
|
||
{''.join(color_nodes)}
|
||
</node>
|
||
{size_nodes}
|
||
</node>
|
||
</node>
|
||
</hierarchy>"""
|
||
|
||
|
||
class IgnoredColorClickDevice(SnakeColorDevice):
|
||
"""页面暴露选中状态，但模拟一次没有生效的颜色点击。"""
|
||
|
||
def click(self, x, y):
|
||
if not self.panel_open:
|
||
super().click(x, y)
|
||
return
|
||
self.clicks.append((x, y))
|
||
|
||
|
||
class AllMissingColorPriceDevice(IgnoredColorClickDevice):
|
||
"""所有颜色点击都未生效，模拟全部颜色采价为空。"""
|
||
|
||
def _spec_xml(self):
|
||
return super()._spec_xml().replace('selected="true"', 'selected="false"')
|
||
|
||
|
||
class NoColorPriceDevice(SnakeColorDevice):
|
||
"""颜色可以选中，但价格节点一直没有出现。"""
|
||
|
||
def _spec_xml(self):
|
||
price = self.prices[self.selected_color] / 100
|
||
return super()._spec_xml().replace(
|
||
f'text="¥{price:.2f}"', 'text="价格加载中"'
|
||
)
|
||
|
||
|
||
class DelayedHorizontalSwipeDevice(SnakeColorDevice):
|
||
"""水平滑动后第一次读取仍返回旧视口，第二次才更新。"""
|
||
|
||
def __init__(self, home_xml):
|
||
super().__init__(home_xml)
|
||
self.pending_page = None
|
||
self.old_page_reads = 0
|
||
|
||
def dump_hierarchy(self):
|
||
if self.panel_open and self.pending_page is not None:
|
||
if self.old_page_reads > 0:
|
||
self.old_page_reads -= 1
|
||
else:
|
||
self.page = self.pending_page
|
||
self.pending_page = None
|
||
return super().dump_hierarchy()
|
||
|
||
def swipe(self, x1, y1, x2, y2, duration=0.35):
|
||
self.swipes.append(((x1, y1, x2, y2), {"duration": duration}))
|
||
self.swipe_panel_states.append(self.panel_open)
|
||
if abs(x2 - x1) > abs(y2 - y1):
|
||
self.pending_page = 1 if x1 > x2 else 0
|
||
self.old_page_reads = 1
|
||
else:
|
||
self.sizes_visible = True
|
||
|
||
|
||
class ImageColorDevice(SnakeColorDevice):
|
||
"""图片型颜色卡片：点图片会进入大图，点底部文字才选中颜色。"""
|
||
|
||
def __init__(self, home_xml):
|
||
super().__init__(home_xml)
|
||
self.big_image_open = False
|
||
|
||
def dump_hierarchy(self):
|
||
if self.big_image_open:
|
||
return """<hierarchy>
|
||
<node class="android.support.v4.view.ViewPager"
|
||
scrollable="true" bounds="[0,0][1080,2000]">
|
||
<node class="android.widget.ImageView" content-desc="A色" />
|
||
</node>
|
||
<node class="android.widget.TextView" text="1/31"
|
||
bounds="[500,1850][580,1900]" />
|
||
</hierarchy>"""
|
||
return super().dump_hierarchy()
|
||
|
||
def click(self, x, y):
|
||
if self.panel_open and 740 <= y < 820:
|
||
self.clicks.append((x, y))
|
||
self.big_image_open = True
|
||
return
|
||
super().click(x, y)
|
||
|
||
def press(self, key):
|
||
if key == "back":
|
||
self.big_image_open = False
|
||
|
||
def _spec_xml(self):
|
||
xml_data = super()._spec_xml()
|
||
old = (
|
||
'<node class="android.view.ViewGroup" content-desc="A色" '
|
||
f'clickable="true" selected="{str(self.selected_color == "A色").lower()}" '
|
||
'bounds="[36,740][340,850]" visible-to-user="true" enabled="true" />'
|
||
)
|
||
new = (
|
||
'<node class="android.view.ViewGroup" content-desc="A色" '
|
||
f'clickable="true" selected="{str(self.selected_color == "A色").lower()}" '
|
||
'bounds="[36,740][340,850]" visible-to-user="true" enabled="true">'
|
||
'<node class="android.widget.ImageView" content-desc="A色" '
|
||
'clickable="true" bounds="[36,740][340,820]" />'
|
||
'<node class="android.widget.ImageView" content-desc="打开大图" '
|
||
'clickable="true" bounds="[280,740][340,800]" />'
|
||
'<node class="android.widget.TextView" text="A色" clickable="true" '
|
||
'bounds="[36,820][340,850]" visible-to-user="true" />'
|
||
'</node>'
|
||
)
|
||
return xml_data.replace(old, new)
|
||
|
||
|
||
class DefaultSizeDevice(SnakeColorDevice):
|
||
"""尺码在下方且默认选中，取消后才能完整遍历颜色。"""
|
||
|
||
def __init__(self, home_xml, *, can_clear=True):
|
||
super().__init__(home_xml)
|
||
self.size_selected = True
|
||
self.can_clear = can_clear
|
||
self.size_clicks = 0
|
||
|
||
def click(self, x, y):
|
||
if self.panel_open and self.sizes_visible and 1210 <= y <= 1300:
|
||
self.clicks.append((x, y))
|
||
self.size_clicks += 1
|
||
if self.can_clear:
|
||
self.size_selected = False
|
||
return
|
||
super().click(x, y)
|
||
|
||
def swipe(self, x1, y1, x2, y2, duration=0.35):
|
||
self.swipes.append(((x1, y1, x2, y2), {"duration": duration}))
|
||
self.swipe_panel_states.append(self.panel_open)
|
||
if abs(x2 - x1) > abs(y2 - y1):
|
||
self.page = 1 if x1 > x2 else 0
|
||
else:
|
||
self.sizes_visible = y1 > y2
|
||
|
||
def _spec_xml(self):
|
||
xml_data = super()._spec_xml()
|
||
if self.sizes_visible:
|
||
selected = str(self.size_selected).lower()
|
||
xml_data = xml_data.replace(
|
||
'content-desc="M"\n clickable="true"',
|
||
f'content-desc="M"\n clickable="true" selected="{selected}"',
|
||
)
|
||
return xml_data
|
||
|
||
|
||
class DefaultPackageDevice(DefaultSizeDevice):
|
||
"""“套餐”作为第二规格时沿用安全的默认选择取消流程。"""
|
||
|
||
def _spec_xml(self):
|
||
return super()._spec_xml().replace('text="尺码"', 'text="套餐"')
|
||
|
||
|
||
class HorizontalPackageDevice:
|
||
"""模拟独立横向套餐列表，每屏只有三个且相邻页面有重叠。"""
|
||
|
||
packages = tuple(f"套餐{i}" for i in range(1, 8))
|
||
|
||
def __init__(self, expected_count=7):
|
||
self.page = 1
|
||
self.expected_count = expected_count
|
||
self.swipes = []
|
||
self.clicks = []
|
||
|
||
def dump_hierarchy(self):
|
||
start = min(self.page * 2, len(self.packages) - 3)
|
||
visible = self.packages[start : start + 3]
|
||
options = []
|
||
for index, text in enumerate(visible):
|
||
left = 36 + index * 320
|
||
options.append(
|
||
f'<node text="{text}" clickable="true" '
|
||
f'bounds="[{left},1250][{left + 280},1370]" />'
|
||
)
|
||
return f"""<hierarchy>
|
||
<node bounds="[0,0][1080,2340]">
|
||
<node class="android.widget.ScrollView" scrollable="true"
|
||
bounds="[0,650][1080,2200]">
|
||
<node text="券后 ¥5.69" bounds="[300,680][600,740]" />
|
||
<node text="款式" bounds="[36,800][180,860]" />
|
||
<node text="口罩款" clickable="true" selected="true"
|
||
bounds="[36,900][300,1000]" />
|
||
<node text="套餐（{self.expected_count}）"
|
||
bounds="[36,1120][240,1180]" />
|
||
<node class="androidx.recyclerview.widget.RecyclerView"
|
||
scrollable="true" bounds="[24,1200][1056,1420]">
|
||
{''.join(options)}
|
||
</node>
|
||
</node>
|
||
</node>
|
||
</hierarchy>"""
|
||
|
||
def swipe(self, x1, y1, x2, y2, duration=0.35):
|
||
self.swipes.append((x1, y1, x2, y2, duration))
|
||
if abs(x2 - x1) <= abs(y2 - y1):
|
||
return
|
||
if x1 > x2:
|
||
self.page = min(self.page + 1, 2)
|
||
else:
|
||
self.page = max(self.page - 1, 0)
|
||
|
||
|
||
class VerticalLongPackageDevice:
|
||
"""模拟纵向长套餐：滑到第二屏后“套餐”标题不再出现在控件树。"""
|
||
|
||
packages = tuple(f"套餐{i}" for i in range(1, 17))
|
||
|
||
def __init__(self):
|
||
self.page = 0
|
||
self.swipes = []
|
||
self.clicks = []
|
||
|
||
def dump_hierarchy(self):
|
||
start = min(self.page * 3, len(self.packages) - 4)
|
||
visible = self.packages[start : start + 4]
|
||
headings = ""
|
||
if self.page == 0:
|
||
headings = """
|
||
<node text="款式" bounds="[36,720][180,780]" />
|
||
<node text="口罩款" clickable="true" selected="true"
|
||
bounds="[36,800][300,900]" />
|
||
<node text="套餐（16）" bounds="[36,980][240,1040]" />
|
||
"""
|
||
options = []
|
||
for index, text in enumerate(visible):
|
||
top = 1080 + index * 210
|
||
right = 500 if self.page == 0 else 1015
|
||
options.append(
|
||
f'<node text="{text} ¥5.69" clickable="true" '
|
||
f'bounds="[36,{top}][{right},{top + 150}]" />'
|
||
)
|
||
return f"""<hierarchy>
|
||
<node bounds="[0,0][1080,2340]">
|
||
<node class="android.widget.ScrollView" scrollable="true"
|
||
bounds="[0,650][1080,2200]">
|
||
<node text="已选：套餐1" bounds="[36,660][500,710]" />
|
||
{headings}
|
||
{''.join(options)}
|
||
<node text="券后 ¥5.69" clickable="true"
|
||
bounds="[36,1880][300,1940]" />
|
||
<node text="增加数量" clickable="true"
|
||
bounds="[36,1950][220,2050]" />
|
||
<node text="提交订单" clickable="true"
|
||
bounds="[300,2080][1040,2190]" />
|
||
</node>
|
||
</node>
|
||
</hierarchy>"""
|
||
|
||
def swipe(self, x1, y1, x2, y2, duration=0.35):
|
||
self.swipes.append((x1, y1, x2, y2, duration))
|
||
if abs(y2 - y1) <= abs(x2 - x1):
|
||
return
|
||
if y1 > y2:
|
||
self.page = min(self.page + 1, 4)
|
||
else:
|
||
self.page = max(self.page - 1, 0)
|
||
|
||
|
||
class FakeClock:
|
||
"""测试用时钟：sleep 只推进虚拟时间，不真的等待。"""
|
||
|
||
def __init__(self):
|
||
self.now = 0.0
|
||
self.sleeps = []
|
||
|
||
def monotonic(self):
|
||
return self.now
|
||
|
||
def sleep(self, seconds):
|
||
self.sleeps.append(seconds)
|
||
self.now += seconds
|
||
|
||
|
||
def keep_only_one_sku(xml_data: str) -> str:
|
||
"""从脱敏固件中删除蓝色和 L，只保留一个组合。"""
|
||
|
||
root = ET.fromstring(xml_data)
|
||
for parent in root.iter():
|
||
for child in list(parent):
|
||
own_label = (
|
||
child.get("text") or child.get("content-desc") or ""
|
||
).strip()
|
||
labels = " ".join(
|
||
(node.get("text") or node.get("content-desc") or "").strip()
|
||
for node in child.iter("node")
|
||
)
|
||
if own_label == "蓝色" or labels.strip() == "L":
|
||
parent.remove(child)
|
||
return ET.tostring(root, encoding="unicode")
|
||
|
||
|
||
class PddCollectParserTest(unittest.TestCase):
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
cls.home_xml = (FIXTURES / "pdd_goods_page.xml").read_text(encoding="utf-8")
|
||
cls.spec_xml = (FIXTURES / "pdd_spec_panel.xml").read_text(encoding="utf-8")
|
||
cls.count_heading_spec_xml = (
|
||
FIXTURES / "pdd_spec_panel_count_heading.xml"
|
||
).read_text(encoding="utf-8")
|
||
cls.non_scrollable_spec_xml = (
|
||
FIXTURES / "pdd_spec_panel_non_scrollable.xml"
|
||
).read_text(encoding="utf-8")
|
||
cls.submit_hint_spec_xml = (
|
||
FIXTURES / "pdd_spec_panel_submit_hint.xml"
|
||
).read_text(encoding="utf-8")
|
||
cls.quick_confirmation_xml = (
|
||
FIXTURES / "pdd_quick_confirmation.xml"
|
||
).read_text(encoding="utf-8")
|
||
cls.sold_out_xml = (
|
||
FIXTURES / "pdd_sold_out_recommendations.xml"
|
||
).read_text(encoding="utf-8")
|
||
|
||
def test_quantity_keeps_raw_value_and_approximate_flag(self):
|
||
result = parse_quantity("已拼1.2万+件")
|
||
self.assertEqual(result.value, 12000)
|
||
self.assertEqual(result.raw, "已拼1.2万+件")
|
||
self.assertTrue(result.approximate)
|
||
|
||
def test_parse_goods_page(self):
|
||
result = parse_goods_page(self.home_xml)
|
||
self.assertEqual(result.title, "测试纯棉短袖商品")
|
||
self.assertEqual(result.shop_name, "测试服饰旗舰店")
|
||
self.assertEqual(result.sales.value, 12000)
|
||
self.assertTrue(result.sales.approximate)
|
||
self.assertEqual(result.reviews.value, 2356)
|
||
|
||
def test_parse_review_title_and_shop_name_next_to_enter_shop(self):
|
||
xml_data = """<hierarchy>
|
||
<node class="android.widget.FrameLayout" bounds="[0,0][1080,2000]">
|
||
<node class="android.widget.TextView" text="商品评价(191)"
|
||
bounds="[30,700][300,760]" />
|
||
<node class="android.widget.TextView" text="承诺假一赔十"
|
||
bounds="[30,1050][300,1110]" />
|
||
<node class="android.widget.TextView" text="小熊服饰店"
|
||
bounds="[300,1200][650,1260]" />
|
||
<node class="android.widget.TextView" text="进店"
|
||
bounds="[850,1200][1030,1260]" />
|
||
</node>
|
||
</hierarchy>"""
|
||
|
||
result = parse_goods_page(xml_data)
|
||
|
||
self.assertEqual(result.reviews.value, 191)
|
||
self.assertEqual(result.shop_name, "小熊服饰店")
|
||
|
||
def test_parse_spec_panel_uses_generic_dimensions_and_cents(self):
|
||
result = parse_spec_panel(self.spec_xml)
|
||
self.assertEqual(result.price_cent, 1000)
|
||
self.assertEqual([item.key for item in result.dimensions], ["color", "size"])
|
||
self.assertEqual(
|
||
[value.text for value in result.dimensions[0].values],
|
||
["红色", "蓝色"],
|
||
)
|
||
self.assertFalse(result.dimensions[0].values[1].available)
|
||
self.assertEqual(
|
||
[value.text for value in result.dimensions[1].values],
|
||
["M", "L"],
|
||
)
|
||
|
||
def test_parse_spec_panel_accepts_count_in_color_heading(self):
|
||
for heading in ("颜色 (6)", "颜色（6）"):
|
||
with self.subTest(heading=heading):
|
||
xml_data = self.count_heading_spec_xml.replace("颜色 (6)", heading)
|
||
result = parse_spec_panel(xml_data)
|
||
colors = next(
|
||
item for item in result.dimensions if item.key == "color"
|
||
)
|
||
|
||
self.assertEqual(colors.name, heading)
|
||
self.assertEqual(
|
||
[value.text for value in colors.values],
|
||
["黑色", "红色", "蓝色", "紫色", "浅粉", "本色"],
|
||
)
|
||
|
||
def test_style_and_package_are_separate_color_and_size_dimensions(self):
|
||
result = parse_spec_panel(style_package_spec_xml())
|
||
|
||
self.assertEqual(
|
||
[(item.key, item.name) for item in result.dimensions],
|
||
[("color", "款式"), ("size", "套餐")],
|
||
)
|
||
self.assertEqual(
|
||
[value.text for value in result.dimensions[0].values],
|
||
["成人款", "儿童款"],
|
||
)
|
||
self.assertEqual(
|
||
[value.text for value in result.dimensions[1].values],
|
||
["单个装", "两个装"],
|
||
)
|
||
|
||
def test_package_only_is_primary_dimension(self):
|
||
root = ET.fromstring(style_package_spec_xml())
|
||
for parent in root.iter():
|
||
for child in list(parent):
|
||
if child.get("text") in {"款式", "成人款", "儿童款"}:
|
||
parent.remove(child)
|
||
|
||
result = parse_spec_panel(ET.tostring(root, encoding="unicode"))
|
||
|
||
self.assertEqual(
|
||
[(item.key, item.name) for item in result.dimensions],
|
||
[("color", "套餐")],
|
||
)
|
||
|
||
def test_third_dimension_is_not_silently_merged(self):
|
||
xml_data = style_package_spec_xml().replace(
|
||
'<node text="确定" clickable="true" bounds="[0,2000][1080,2150]" />',
|
||
'<node text="尺码" bounds="[36,1400][180,1460]" />\n'
|
||
'<node text="均码" clickable="true" selected="false" '
|
||
'bounds="[36,1500][300,1600]" />\n'
|
||
'<node text="确定" clickable="true" bounds="[0,2000][1080,2150]" />',
|
||
)
|
||
parsed = parse_spec_panel(xml_data)
|
||
self.assertEqual(
|
||
[item.key for item in parsed.dimensions],
|
||
["color", "size", "size_2"],
|
||
)
|
||
device = FakeCollectDevice(self.home_xml, xml_data)
|
||
device.panel_open = True
|
||
device.opened_url = "https://mobile.yangkeduo.com/goods.html?goods_id=123"
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
max_spec_swipes=0,
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service._collect_size_dimension(device)
|
||
|
||
self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_UNSUPPORTED")
|
||
|
||
def test_collect_preserves_style_and_package_names(self):
|
||
xml_data = style_package_spec_xml().replace(
|
||
'<node text="儿童款" clickable="true" selected="false"\n'
|
||
' bounds="[330,900][594,1000]" />',
|
||
"",
|
||
).replace(
|
||
'<node text="两个装" clickable="true" selected="false"\n'
|
||
' bounds="[330,1200][594,1300]" />',
|
||
"",
|
||
)
|
||
device = FakeCollectDevice(self.home_xml, xml_data)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
max_spec_swipes=0,
|
||
)
|
||
|
||
result = service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
|
||
self.assertEqual(
|
||
[(item.key, item.name) for item in result.dimensions],
|
||
[("color", "款式"), ("size", "套餐")],
|
||
)
|
||
self.assertEqual(
|
||
result.skus[0].options,
|
||
{"color": "成人款", "size": "单个装"},
|
||
)
|
||
|
||
def test_parse_non_scrollable_spec_panel(self):
|
||
result = parse_spec_panel(self.non_scrollable_spec_xml)
|
||
|
||
self.assertTrue(_is_spec_panel_open(self.non_scrollable_spec_xml))
|
||
self.assertEqual([item.key for item in result.dimensions], ["color", "size"])
|
||
self.assertEqual(
|
||
[value.text for value in result.dimensions[0].values],
|
||
[
|
||
"测试黑色",
|
||
"测试黑色+网袜",
|
||
"测试黑色+丝袜",
|
||
"测试黑色+短袜",
|
||
"测试黑色+长袜",
|
||
],
|
||
)
|
||
self.assertEqual(
|
||
[value.text for value in result.dimensions[1].values],
|
||
["均码（适合80-120斤）"],
|
||
)
|
||
self.assertEqual(result.price_cent, 817)
|
||
self.assertEqual(result.raw_price, "¥8.17")
|
||
self.assertNotIn("确认款式", [item.name for item in result.dimensions])
|
||
|
||
def test_parse_non_scrollable_submit_hint_spec_panel(self):
|
||
result = parse_spec_panel(self.submit_hint_spec_xml)
|
||
|
||
self.assertTrue(_is_spec_panel_open(self.submit_hint_spec_xml))
|
||
self.assertEqual(
|
||
[(item.key, item.name) for item in result.dimensions],
|
||
[("color", "颜色分类"), ("size", "尺码")],
|
||
)
|
||
self.assertEqual(
|
||
[value.text for value in result.dimensions[0].values],
|
||
["测试黑色", "测试红色"],
|
||
)
|
||
self.assertEqual(
|
||
[value.text for value in result.dimensions[1].values],
|
||
["M", "L"],
|
||
)
|
||
self.assertNotIn(
|
||
"请选择： 颜色分类 尺码",
|
||
[item.name for item in result.dimensions],
|
||
)
|
||
|
||
def test_collects_static_sizes_from_non_scrollable_submit_hint_panel(self):
|
||
device = FakeCollectDevice(self.home_xml, self.submit_hint_spec_xml)
|
||
device.opened_url = (
|
||
"https://mobile.yangkeduo.com/goods.html?goods_id=123"
|
||
)
|
||
device.panel_open = True
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
dimension = service.collect_second_dimension_candidates(device)
|
||
|
||
self.assertIsNotNone(dimension)
|
||
self.assertEqual(dimension.name, "尺码")
|
||
self.assertEqual(
|
||
[value.text for value in dimension.values],
|
||
["M", "L"],
|
||
)
|
||
self.assertEqual(device.swipes, [])
|
||
|
||
def test_static_sizes_keep_explicit_count_validation(self):
|
||
spec_xml = self.submit_hint_spec_xml.replace(
|
||
'text="尺码" bounds=', 'text="尺码（3）" bounds=', 1
|
||
)
|
||
device = FakeCollectDevice(self.home_xml, spec_xml)
|
||
device.opened_url = (
|
||
"https://mobile.yangkeduo.com/goods.html?goods_id=123"
|
||
)
|
||
device.panel_open = True
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service._collect_size_dimension(device)
|
||
|
||
self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
|
||
self.assertEqual(raised.exception.diagnostics["expected_count"], 3)
|
||
self.assertEqual(raised.exception.diagnostics["collected_count"], 2)
|
||
|
||
def test_static_sizes_require_strong_non_scrollable_panel_evidence(self):
|
||
snapshot = parse_spec_panel(self.submit_hint_spec_xml)
|
||
dimension = next(
|
||
item for item in snapshot.dimensions if item.key == "size"
|
||
)
|
||
incomplete_evidence_xml = self.submit_hint_spec_xml.replace(
|
||
'class="android.widget.EditText"',
|
||
'class="android.widget.TextView"',
|
||
1,
|
||
)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: object()),
|
||
"USB-001",
|
||
"client-001",
|
||
)
|
||
|
||
result = service._non_scrollable_second_dimension(
|
||
incomplete_evidence_xml, dimension
|
||
)
|
||
|
||
self.assertIsNone(result)
|
||
|
||
def test_isolated_submit_hint_is_not_a_spec_panel(self):
|
||
xml_data = """<hierarchy>
|
||
<node package="com.xunmeng.pinduoduo"
|
||
class="android.widget.FrameLayout" bounds="[0,0][1080,2376]">
|
||
<node package="com.xunmeng.pinduoduo" content-desc="关闭"
|
||
clickable="true" bounds="[975,273][1050,348]" />
|
||
<node package="com.xunmeng.pinduoduo"
|
||
text="选择规格后，提交订单" clickable="true"
|
||
bounds="[0,2181][1080,2328]" />
|
||
</node>
|
||
</hierarchy>"""
|
||
|
||
self.assertFalse(_is_spec_panel_open(xml_data))
|
||
|
||
def test_non_scrollable_spec_panel_exposes_three_color_rows(self):
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: object()),
|
||
"USB-001",
|
||
"client-001",
|
||
)
|
||
|
||
rows = service._visible_color_rows(self.non_scrollable_spec_xml)
|
||
|
||
self.assertEqual(
|
||
[[item.text for item in row] for row in rows],
|
||
[
|
||
["测试黑色", "测试黑色+网袜"],
|
||
["测试黑色+丝袜", "测试黑色+短袜"],
|
||
["测试黑色+长袜"],
|
||
],
|
||
)
|
||
self.assertTrue(all(item.available for row in rows for item in row))
|
||
self.assertTrue(all(item.bounds for row in rows for item in row))
|
||
|
||
def test_clickable_text_outside_spec_panel_is_not_an_option(self):
|
||
xml_data = """<hierarchy>
|
||
<node class="android.widget.FrameLayout" bounds="[0,0][1080,2400]">
|
||
<node text="测试黑色" clickable="true"
|
||
bounds="[36,1295][276,1382]" />
|
||
</node>
|
||
</hierarchy>"""
|
||
root = ET.fromstring(xml_data)
|
||
|
||
self.assertIsNone(PddCollectService._find_option_node(root, "测试黑色"))
|
||
|
||
def test_wait_accepts_non_scrollable_spec_panel_without_sleep(self):
|
||
device = FakeCollectDevice(self.home_xml, self.non_scrollable_spec_xml)
|
||
device.opened_url = "https://mobile.yangkeduo.com/goods.html?goods_id=123"
|
||
device.panel_open = True
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
spec_panel_timeout=1.0,
|
||
)
|
||
|
||
snapshot = service._wait_spec_panel(device)
|
||
|
||
self.assertEqual([item.key for item in snapshot.dimensions], ["color", "size"])
|
||
self.assertEqual(clock.sleeps, [])
|
||
self.assertEqual(device.clicks, [])
|
||
|
||
def test_quick_confirmation_is_not_a_complete_spec_panel(self):
|
||
self.assertFalse(_is_spec_panel_open(self.quick_confirmation_xml))
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
parse_spec_panel(self.quick_confirmation_xml)
|
||
self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
|
||
|
||
def test_wait_opens_spec_selector_from_quick_confirmation_once(self):
|
||
device = QuickConfirmationDevice(
|
||
self.quick_confirmation_xml,
|
||
self.non_scrollable_spec_xml,
|
||
)
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
spec_panel_timeout=1.0,
|
||
)
|
||
|
||
snapshot = service._wait_spec_panel(device)
|
||
|
||
self.assertEqual([item.key for item in snapshot.dimensions], ["color", "size"])
|
||
self.assertEqual(device.clicks, [(251, 1280)])
|
||
self.assertTrue(all(y < 2000 for _, y in device.clicks))
|
||
|
||
def test_quick_confirmation_recovery_is_attempted_only_once(self):
|
||
device = StuckQuickConfirmationDevice(
|
||
self.quick_confirmation_xml,
|
||
self.non_scrollable_spec_xml,
|
||
)
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
spec_panel_timeout=1.0,
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service._wait_spec_panel(device)
|
||
|
||
self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
|
||
self.assertIn("快速确认页", raised.exception.message)
|
||
self.assertEqual(device.clicks, [(251, 1280)])
|
||
self.assertTrue(
|
||
raised.exception.diagnostics["quick_confirmation_recovery_attempted"]
|
||
)
|
||
|
||
def test_quick_confirmation_without_payment_evidence_is_not_clicked(self):
|
||
incomplete_xml = self.quick_confirmation_xml.replace(
|
||
"使用微信支付，可更换支付方式",
|
||
"普通页面说明",
|
||
)
|
||
device = StuckQuickConfirmationDevice(
|
||
incomplete_xml,
|
||
self.non_scrollable_spec_xml,
|
||
)
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
spec_panel_timeout=1.0,
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service._wait_spec_panel(device)
|
||
|
||
self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
|
||
self.assertEqual(device.clicks, [])
|
||
self.assertFalse(
|
||
raised.exception.diagnostics["quick_confirmation_recovery_attempted"]
|
||
)
|
||
|
||
def test_goods_page_is_not_non_scrollable_spec_panel(self):
|
||
self.assertFalse(_is_spec_panel_open(self.home_xml))
|
||
|
||
def test_sold_out_recommendations_are_not_spec_options(self):
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
parse_spec_panel(self.sold_out_xml)
|
||
|
||
self.assertEqual(raised.exception.code, "PDD_GOODS_UNAVAILABLE")
|
||
self.assertEqual(
|
||
raised.exception.diagnostics, {"page_reason": "out_of_stock"}
|
||
)
|
||
|
||
page_without_marker = self.sold_out_xml.replace(
|
||
"商品已售罄，推荐以下相似商品", "为你推荐以下相似商品"
|
||
)
|
||
with self.assertRaises(PddCollectError) as incomplete:
|
||
parse_spec_panel(page_without_marker)
|
||
self.assertEqual(incomplete.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
|
||
|
||
def test_blank_pdd_tree_is_not_a_spec_panel(self):
|
||
xml_data = """<hierarchy>
|
||
<node class="android.widget.FrameLayout"
|
||
package="com.xunmeng.pinduoduo"
|
||
bounds="[0,0][1080,2340]" />
|
||
</hierarchy>"""
|
||
|
||
self.assertFalse(_is_spec_panel_open(xml_data))
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
parse_spec_panel(xml_data)
|
||
self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
|
||
|
||
def test_collect_stops_on_sold_out_page_without_clicking_recommendations(self):
|
||
device = FakeCollectDevice(self.home_xml, self.sold_out_xml)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
|
||
self.assertEqual(raised.exception.code, "PDD_GOODS_UNAVAILABLE")
|
||
self.assertEqual(
|
||
raised.exception.diagnostics["page_reason"], "out_of_stock"
|
||
)
|
||
# 只点击商品页规格入口，不点击任何推荐商品。
|
||
self.assertEqual(len(device.clicks), 1)
|
||
|
||
def test_full_window_with_scattered_labels_is_not_spec_panel(self):
|
||
xml_data = """<hierarchy>
|
||
<node class="android.widget.FrameLayout" bounds="[0,0][1080,2400]">
|
||
<node text="确认款式" bounds="[20,100][220,160]" />
|
||
<node text="已选择：测试值" bounds="[20,300][500,360]" />
|
||
<node text="颜色分类" bounds="[20,800][220,860]" />
|
||
<node text="确定" clickable="true" bounds="[0,2280][1080,2400]" />
|
||
</node>
|
||
</hierarchy>"""
|
||
|
||
self.assertFalse(_is_spec_panel_open(xml_data))
|
||
|
||
def test_open_panel_without_visible_options_is_not_reported_as_timeout(self):
|
||
panel_xml = """<hierarchy>
|
||
<node class="android.widget.FrameLayout" bounds="[0,0][1080,2400]">
|
||
<node text="请选择：颜色 尺码" bounds="[36,900][600,980]" />
|
||
<node class="androidx.recyclerview.widget.RecyclerView"
|
||
scrollable="true" bounds="[0,1077][1080,2079]" />
|
||
<node text="选择颜色及尺码后，提交订单"
|
||
bounds="[300,2200][1000,2300]" />
|
||
</node>
|
||
</hierarchy>"""
|
||
device = FakeCollectDevice(self.home_xml, panel_xml)
|
||
device.opened_url = "https://mobile.yangkeduo.com/goods.html?goods_id=123"
|
||
device.panel_open = True
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
spec_panel_timeout=1.0,
|
||
)
|
||
|
||
snapshot = service._wait_spec_panel(device)
|
||
|
||
self.assertEqual(snapshot.dimensions, ())
|
||
self.assertEqual(clock.sleeps, [])
|
||
|
||
def test_open_panel_with_incomplete_specs_reports_spec_incomplete(self):
|
||
panel_xml = """<hierarchy>
|
||
<node class="android.widget.FrameLayout" bounds="[0,0][1080,2400]">
|
||
<node text="请选择：颜色 尺码" bounds="[36,900][600,980]" />
|
||
<node class="androidx.recyclerview.widget.RecyclerView"
|
||
scrollable="true" bounds="[0,1077][1080,2079]" />
|
||
<node text="选择颜色及尺码后，提交订单"
|
||
bounds="[300,2200][1000,2300]" />
|
||
</node>
|
||
</hierarchy>"""
|
||
device = FakeCollectDevice(self.home_xml, panel_xml)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
max_spec_swipes=0,
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
|
||
self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
|
||
|
||
def test_truncated_spec_name_prefers_full_content_description(self):
|
||
xml_data = self.spec_xml.replace(
|
||
'content-desc="红色" clickable="true"',
|
||
'text="红…" content-desc="红色" clickable="true"',
|
||
)
|
||
result = parse_spec_panel(xml_data)
|
||
self.assertEqual(result.dimensions[0].values[0].text, "红色")
|
||
|
||
without_description = xml_data.replace(' content-desc="红色"', "")
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
parse_spec_panel(without_description)
|
||
self.assertEqual(raised.exception.code, "PDD_DATA_SKU_NAME_TRUNCATED")
|
||
|
||
def test_login_and_invalid_xml_have_different_error_codes(self):
|
||
with self.assertRaises(PddCollectError) as login:
|
||
parse_goods_page('<hierarchy><node text="手机号登录" /></hierarchy>')
|
||
with self.assertRaises(PddCollectError) as invalid:
|
||
parse_goods_page("<hierarchy>")
|
||
self.assertEqual(login.exception.code, "PDD_PAGE_LOGIN_REQUIRED")
|
||
self.assertEqual(invalid.exception.code, "PDD_DATA_XML_INVALID")
|
||
|
||
def test_captcha_and_incomplete_spec_have_specific_codes(self):
|
||
with self.assertRaises(PddCollectError) as captcha:
|
||
parse_goods_page('<hierarchy><node text="请完成验证" /></hierarchy>')
|
||
with self.assertRaises(PddCollectError) as incomplete:
|
||
parse_spec_panel('<hierarchy><node text="普通页面" /></hierarchy>')
|
||
self.assertEqual(captcha.exception.code, "PDD_PAGE_CAPTCHA")
|
||
self.assertEqual(incomplete.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
|
||
|
||
def test_collect_returns_versioned_pdd_data_without_real_device(self):
|
||
# 收窄到一个可用组合，便于验证完整流程而不模拟真实页面切换。
|
||
one_sku_xml = keep_only_one_sku(self.spec_xml)
|
||
device = FakeCollectDevice(self.home_xml, one_sku_xml)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
now=lambda: datetime(2026, 8, 7, 8, 0, tzinfo=timezone.utc),
|
||
max_spec_swipes=1,
|
||
)
|
||
|
||
result = service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
data = result.to_pdd_data()
|
||
|
||
self.assertEqual(data["schema_version"], 1)
|
||
self.assertEqual(data["goods_id"], "123")
|
||
self.assertEqual(data["title"], "测试纯棉短袖商品")
|
||
self.assertEqual(data["shop_name"], "测试服饰旗舰店")
|
||
self.assertEqual(data["price_granularity"], "color")
|
||
self.assertEqual(data["metrics"]["sales"]["value"], 12000)
|
||
self.assertEqual(data["dimensions"][0]["key"], "color")
|
||
self.assertEqual(data["skus"][0]["options"], {"color": "红色", "size": "M"})
|
||
self.assertEqual(data["skus"][0]["price_cent"], 1000)
|
||
self.assertEqual(
|
||
data["skus"][0]["price_observed_at"],
|
||
{"color": "红色"},
|
||
)
|
||
self.assertEqual(data["source"]["device_address"], "USB-001")
|
||
self.assertIsNone(data["purchase"])
|
||
self.assertEqual(device.app_wait_calls, 0)
|
||
self.assertEqual(device.app_current_calls, 1)
|
||
|
||
def test_collect_scrolls_before_spec_panel_and_accumulates_metadata(self):
|
||
initial_page = self.home_xml.replace(
|
||
'<node class="android.widget.TextView" text="2356条评价" '
|
||
'bounds="[330,950][600,1010]" visible-to-user="true" '
|
||
'enabled="true" />',
|
||
"",
|
||
).replace(
|
||
'<node class="android.widget.TextView" text="店铺：测试服饰旗舰店" '
|
||
'bounds="[30,1050][600,1110]" visible-to-user="true" '
|
||
'enabled="true" />',
|
||
"",
|
||
)
|
||
buy_button = """
|
||
<node class="android.view.ViewGroup" content-desc="¥10.00立即购买"
|
||
clickable="true" bounds="[500,1700][1080,1950]"
|
||
visible-to-user="true" enabled="true">
|
||
<node class="android.widget.TextView" text="¥10.00"
|
||
bounds="[650,1730][850,1810]"
|
||
visible-to-user="true" enabled="true" />
|
||
<node class="android.widget.TextView" text="立即购买"
|
||
bounds="[650,1820][900,1900]"
|
||
visible-to-user="true" enabled="true" />
|
||
</node>
|
||
"""
|
||
review_page = f"""<hierarchy>
|
||
<node class="android.widget.FrameLayout" bounds="[0,0][1080,2000]">
|
||
<node class="android.widget.TextView" text="商品评价(191)"
|
||
bounds="[30,700][300,760]" />
|
||
{buy_button}
|
||
</node>
|
||
</hierarchy>"""
|
||
shop_page = f"""<hierarchy>
|
||
<node class="android.widget.FrameLayout" bounds="[0,0][1080,2000]">
|
||
<node class="android.widget.TextView" text="小熊服饰店"
|
||
bounds="[300,1200][650,1260]" />
|
||
<node class="android.widget.TextView" text="进店"
|
||
bounds="[850,1200][1030,1260]" />
|
||
{buy_button}
|
||
</node>
|
||
</hierarchy>"""
|
||
device = GoodsMetadataScrollDevice(
|
||
[initial_page, review_page, shop_page],
|
||
keep_only_one_sku(self.spec_xml),
|
||
)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
max_spec_swipes=0,
|
||
)
|
||
|
||
result = service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
|
||
self.assertEqual(result.title, "测试纯棉短袖商品")
|
||
self.assertEqual(result.sales.value, 12000)
|
||
self.assertEqual(result.reviews.value, 191)
|
||
self.assertEqual(result.shop_name, "小熊服饰店")
|
||
self.assertEqual(device.swipe_panel_states[:2], [False, False])
|
||
self.assertEqual(device.page_index, 2)
|
||
self.assertTrue(device.clicks)
|
||
|
||
def test_pdd_hierarchy_is_ready_even_when_focused_package_is_settings(self):
|
||
pdd_home_xml = self.home_xml.replace(
|
||
"<node ", '<node package="com.xunmeng.pinduoduo" '
|
||
)
|
||
device = FocusMismatchDevice(
|
||
pdd_home_xml, keep_only_one_sku(self.spec_xml)
|
||
)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
max_spec_swipes=0,
|
||
)
|
||
|
||
result = service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
|
||
self.assertEqual(result.goods_id, "123")
|
||
self.assertTrue(device.app_started)
|
||
self.assertEqual(device.app_wait_calls, 1)
|
||
|
||
def test_price_is_sampled_once_per_available_color(self):
|
||
device = FakeCollectDevice(self.home_xml, self.spec_xml)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
max_spec_swipes=0,
|
||
)
|
||
|
||
data = service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
).to_pdd_data()
|
||
|
||
self.assertEqual(len(data["skus"]), 4)
|
||
red_skus = [
|
||
item for item in data["skus"] if item["options"]["color"] == "红色"
|
||
]
|
||
self.assertEqual(
|
||
{item["price_observed_at"]["color"] for item in red_skus},
|
||
{"红色"},
|
||
)
|
||
self.assertTrue(
|
||
all(
|
||
set(item["price_observed_at"]) == {"color"}
|
||
for item in red_skus
|
||
)
|
||
)
|
||
# 1 次打开规格面板，另有红色点 1 次；尺码只读不点击。
|
||
self.assertEqual(len(device.clicks), 2)
|
||
|
||
def test_colors_are_sampled_in_snake_order_before_sizes_are_read(self):
|
||
device = SnakeColorDevice(self.home_xml)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
data = service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
).to_pdd_data()
|
||
|
||
self.assertEqual(
|
||
device.clicked_colors,
|
||
["A色", "B色", "C色", "F色", "E色", "D色"],
|
||
)
|
||
self.assertEqual(
|
||
[item["text"] for item in data["dimensions"][0]["values"]],
|
||
["A色", "B色", "C色", "D色", "E色", "F色"],
|
||
)
|
||
self.assertEqual(
|
||
[item["text"] for item in data["dimensions"][1]["values"]],
|
||
["M", "L"],
|
||
)
|
||
color_prices = {
|
||
item["options"]["color"]: item["price_cent"]
|
||
for item in data["skus"]
|
||
if item["options"]["size"] == "M"
|
||
}
|
||
self.assertEqual(color_prices["A色"], 1000)
|
||
self.assertEqual(color_prices["B色"], 1000)
|
||
self.assertEqual(color_prices["F色"], 1500)
|
||
self.assertTrue(all(
|
||
item["price_observed_at"] == {"color": item["options"]["color"]}
|
||
for item in data["skus"]
|
||
))
|
||
|
||
def test_image_color_card_clicks_caption_instead_of_opening_big_image(self):
|
||
device = ImageColorDevice(self.home_xml)
|
||
device.panel_open = True
|
||
device.page = 0
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
sample = service._click_and_sample_color(device, "A色")
|
||
|
||
self.assertEqual(1000, sample.price_cent)
|
||
self.assertFalse(device.big_image_open)
|
||
self.assertGreaterEqual(device.clicks[0][1], 820)
|
||
|
||
def test_big_image_recovery_presses_back_only_once(self):
|
||
device = ImageColorDevice(self.home_xml)
|
||
device.panel_open = True
|
||
device.big_image_open = True
|
||
press_calls = []
|
||
original_press = device.press
|
||
|
||
def record_press(key):
|
||
press_calls.append(key)
|
||
original_press(key)
|
||
|
||
device.press = record_press
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
service._recover_spec_panel_from_big_image(device)
|
||
|
||
self.assertEqual(["back"], press_calls)
|
||
self.assertFalse(device.big_image_open)
|
||
|
||
def test_default_size_is_cleared_once_before_colors_and_sizes_are_collected(self):
|
||
device = DefaultSizeDevice(self.home_xml)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
data = service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
).to_pdd_data()
|
||
|
||
self.assertEqual(1, device.size_clicks)
|
||
self.assertFalse(device.size_selected)
|
||
self.assertEqual(
|
||
["A色", "B色", "C色", "F色", "E色", "D色"],
|
||
device.clicked_colors,
|
||
)
|
||
self.assertEqual(
|
||
["M", "L"],
|
||
[item["text"] for item in data["dimensions"][1]["values"]],
|
||
)
|
||
|
||
def test_size_that_cannot_be_cleared_is_not_clicked_repeatedly(self):
|
||
device = DefaultSizeDevice(self.home_xml, can_clear=False)
|
||
device.panel_open = True
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
service._clear_default_size_selection(device)
|
||
|
||
self.assertEqual(1, device.size_clicks)
|
||
self.assertTrue(device.size_selected)
|
||
|
||
def test_default_package_is_cleared_and_keeps_original_heading(self):
|
||
device = DefaultPackageDevice(self.home_xml)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
data = service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
).to_pdd_data()
|
||
|
||
self.assertEqual(1, device.size_clicks)
|
||
self.assertFalse(device.size_selected)
|
||
self.assertEqual(data["dimensions"][1]["key"], "size")
|
||
self.assertEqual(data["dimensions"][1]["name"], "套餐")
|
||
|
||
def test_horizontal_packages_are_scanned_from_left_to_right_without_clicks(self):
|
||
device = HorizontalPackageDevice()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
result = service.collect_second_dimension_candidates(device)
|
||
|
||
self.assertIsNotNone(result)
|
||
self.assertEqual(result.name, "套餐（7）")
|
||
self.assertEqual(
|
||
[value.text for value in result.values],
|
||
list(device.packages),
|
||
)
|
||
self.assertEqual(device.clicks, [])
|
||
self.assertTrue(
|
||
all(abs(x2 - x1) > abs(y2 - y1) for x1, y1, x2, y2, _ in device.swipes)
|
||
)
|
||
|
||
def test_package_heading_count_rejects_partial_collection(self):
|
||
device = HorizontalPackageDevice(expected_count=8)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service._collect_size_dimension(device)
|
||
|
||
self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
|
||
self.assertEqual(raised.exception.diagnostics["expected_count"], 8)
|
||
self.assertEqual(raised.exception.diagnostics["collected_count"], 7)
|
||
|
||
def test_vertical_packages_continue_after_heading_scrolls_out(self):
|
||
device = VerticalLongPackageDevice()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
result = service.collect_second_dimension_candidates(device)
|
||
|
||
self.assertIsNotNone(result)
|
||
self.assertEqual(result.name, "套餐（16）")
|
||
self.assertEqual(
|
||
[value.text for value in result.values],
|
||
list(device.packages),
|
||
)
|
||
self.assertNotIn("增加数量", [value.text for value in result.values])
|
||
self.assertNotIn("提交订单", [value.text for value in result.values])
|
||
self.assertNotIn("券后", [value.text for value in result.values])
|
||
self.assertTrue(all("¥" not in value.text for value in result.values))
|
||
self.assertEqual(device.clicks, [])
|
||
self.assertTrue(
|
||
all(abs(y2 - y1) > abs(x2 - x1) for x1, y1, x2, y2, _ in device.swipes)
|
||
)
|
||
|
||
def test_visible_click_is_fallback_when_page_exposes_no_selection_state(self):
|
||
xml_data = keep_only_one_sku(self.spec_xml)
|
||
root = ET.fromstring(xml_data)
|
||
for parent in root.iter():
|
||
for child in list(parent):
|
||
if (child.get("text") or "").startswith("已选"):
|
||
parent.remove(child)
|
||
xml_without_selection = ET.tostring(root, encoding="unicode")
|
||
device = FakeCollectDevice(self.home_xml, xml_without_selection)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
max_spec_swipes=0,
|
||
)
|
||
|
||
data = service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
).to_pdd_data()
|
||
|
||
self.assertEqual(data["skus"][0]["price_cent"], 1000)
|
||
self.assertEqual(
|
||
data["skus"][0]["price_observed_at"], {"color": "红色"}
|
||
)
|
||
|
||
def test_visible_text_does_not_confirm_an_ignored_click(self):
|
||
device = IgnoredColorClickDevice(self.home_xml)
|
||
device.panel_open = True
|
||
device.page = 0
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
sample = service._click_and_sample_color(device, "B色")
|
||
|
||
self.assertIsNone(sample.price_cent)
|
||
|
||
def test_color_price_is_checked_immediately_then_stabilized(self):
|
||
device = SnakeColorDevice(self.home_xml)
|
||
device.panel_open = True
|
||
device.page = 0
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
)
|
||
|
||
sample = service._click_and_sample_color(device, "A色")
|
||
|
||
self.assertEqual(sample.price_cent, 1000)
|
||
self.assertEqual(clock.sleeps, [0.1])
|
||
self.assertEqual(device.clicked_colors, ["A色"])
|
||
|
||
def test_ignored_color_click_stops_at_selection_timeout(self):
|
||
device = IgnoredColorClickDevice(self.home_xml)
|
||
device.panel_open = True
|
||
device.page = 0
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
color_poll_interval=0.1,
|
||
color_selection_timeout=0.3,
|
||
)
|
||
|
||
sample = service._click_and_sample_color(device, "B色")
|
||
|
||
self.assertIsNone(sample.price_cent)
|
||
self.assertLessEqual(clock.now, 0.31)
|
||
self.assertEqual(len(device.clicks), 1)
|
||
|
||
def test_selected_color_without_price_stops_at_price_timeout(self):
|
||
device = NoColorPriceDevice(self.home_xml)
|
||
device.panel_open = True
|
||
device.page = 0
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
color_poll_interval=0.1,
|
||
color_price_timeout=0.3,
|
||
)
|
||
|
||
sample = service._click_and_sample_color(device, "A色")
|
||
|
||
self.assertIsNone(sample.price_cent)
|
||
self.assertLessEqual(clock.now, 0.31)
|
||
self.assertEqual(device.clicked_colors, ["A色"])
|
||
|
||
def test_horizontal_swipe_waits_once_more_when_view_is_unchanged(self):
|
||
device = DelayedHorizontalSwipeDevice(self.home_xml)
|
||
device.panel_open = True
|
||
device.page = 0
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
)
|
||
previous = service._color_view_signature(device.dump_hierarchy())
|
||
device.swipe(800, 800, 200, 800, duration=0.35)
|
||
|
||
latest_xml = service._wait_for_horizontal_change(device, previous)
|
||
|
||
self.assertNotEqual(service._color_view_signature(latest_xml), previous)
|
||
self.assertEqual(clock.sleeps, [0.1, 0.1])
|
||
|
||
def test_color_wait_parameters_must_be_positive(self):
|
||
with self.assertRaisesRegex(ValueError, "颜色轮询间隔必须大于 0 秒"):
|
||
PddCollectService(
|
||
PddDeviceService(lambda _serial: object()),
|
||
"USB-001",
|
||
"client-001",
|
||
color_poll_interval=0,
|
||
)
|
||
|
||
def test_all_missing_color_prices_still_collects_sizes(self):
|
||
device = AllMissingColorPriceDevice(self.home_xml)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
data = service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
).to_pdd_data()
|
||
|
||
self.assertEqual(
|
||
[item["text"] for item in data["dimensions"][1]["values"]],
|
||
["M", "L"],
|
||
)
|
||
self.assertTrue(device.sizes_visible)
|
||
self.assertTrue(data["skus"])
|
||
self.assertTrue(all(item["price_cent"] is None for item in data["skus"]))
|
||
self.assertTrue(all(item["raw_price"] is None for item in data["skus"]))
|
||
self.assertTrue(
|
||
all(item["price_observed_at"] == {} for item in data["skus"])
|
||
)
|
||
self.assertTrue(all(item["available"] for item in data["skus"]))
|
||
|
||
def test_missing_goods_id_is_structured_error(self):
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: object()),
|
||
"USB-001",
|
||
"client-001",
|
||
)
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(FakeTask("https://mobile.yangkeduo.com/goods.html"))
|
||
self.assertEqual(raised.exception.code, "PDD_DATA_GOODS_ID_MISSING")
|
||
|
||
def test_non_pdd_goods_url_is_rejected_before_opening_device(self):
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: object()),
|
||
"USB-001",
|
||
"client-001",
|
||
)
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(FakeTask("https://example.com/goods.html?goods_id=123"))
|
||
self.assertEqual(raised.exception.code, "PDD_DATA_GOODS_URL_INVALID")
|
||
|
||
def test_task_goods_id_must_match_url(self):
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: object()),
|
||
"USB-001",
|
||
"client-001",
|
||
)
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(
|
||
FakeTask(
|
||
"https://mobile.yangkeduo.com/goods.html?goods_id=123",
|
||
goods_id="456",
|
||
)
|
||
)
|
||
self.assertEqual(raised.exception.code, "PDD_DATA_GOODS_ID_MISMATCH")
|
||
|
||
def test_page_timeout_has_specific_error_code(self):
|
||
device = LoadingDevice(self.home_xml, self.spec_xml)
|
||
ticks = iter((0.0, 0.0, 0.0, 2.0))
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
monotonic=lambda: next(ticks),
|
||
page_timeout=1.0,
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
self.assertEqual(raised.exception.code, "PDD_PAGE_TIMEOUT")
|
||
|
||
def test_stable_home_reopens_once_then_stops_before_spec_click(self):
|
||
home = (FIXTURES / "pdd_home_page.xml").read_text(encoding="utf-8")
|
||
device = FakeCollectDevice(home, self.spec_xml)
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
page_timeout=10.0,
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
|
||
self.assertEqual("PDD_GOODS_UNAVAILABLE", raised.exception.code)
|
||
self.assertEqual(2, raised.exception.diagnostics["open_attempts"])
|
||
self.assertEqual([], device.clicks)
|
||
|
||
def test_slow_wrong_focus_does_not_consume_home_reopen_timeout(self):
|
||
home = (FIXTURES / "pdd_home_page.xml").read_text(encoding="utf-8")
|
||
clock = FakeClock()
|
||
device = SlowFocusMismatchDevice(home, self.spec_xml, clock)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
page_timeout=10.0,
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
|
||
self.assertEqual("PDD_GOODS_UNAVAILABLE", raised.exception.code)
|
||
self.assertEqual(1, device.app_current_calls)
|
||
self.assertEqual([], device.clicks)
|
||
|
||
def test_non_pdd_tree_with_purchase_words_is_not_ready(self):
|
||
device = FocusMismatchDevice(
|
||
'<hierarchy><node package="com.android.settings" text="免拼购买" '
|
||
'bounds="[0,0][100,100]" /></hierarchy>',
|
||
self.spec_xml,
|
||
)
|
||
ticks = iter((0.0, 0.0, 0.0, 2.0))
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
monotonic=lambda: next(ticks),
|
||
page_timeout=1.0,
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
self.assertEqual(raised.exception.code, "PDD_PAGE_TIMEOUT")
|
||
|
||
def test_page_timeout_saves_last_xml_diagnostic(self):
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
device = LoadingDevice(self.home_xml, self.spec_xml)
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
page_timeout=1.0,
|
||
artifact_directory=Path(directory),
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
|
||
artifacts = raised.exception.diagnostics["artifacts"]
|
||
self.assertEqual(len(artifacts), 1)
|
||
self.assertTrue(Path(artifacts[0]["path"]).is_file())
|
||
self.assertEqual(len(artifacts[0]["sha256"]), 64)
|
||
|
||
def test_runtime_device_disconnect_has_stable_error_code(self):
|
||
service = PddCollectService(
|
||
PddDeviceService(
|
||
lambda _serial: DisconnectingDevice(self.home_xml, self.spec_xml)
|
||
),
|
||
"USB-001",
|
||
"client-001",
|
||
)
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
self.assertEqual(raised.exception.code, "DEVICE_DISCONNECTED")
|
||
|
||
def test_missing_shop_name_is_allowed(self):
|
||
incomplete_home = self.home_xml.replace(
|
||
'<node class="android.widget.TextView" text="店铺：测试服饰旗舰店" '
|
||
'bounds="[30,1050][600,1110]" visible-to-user="true" '
|
||
'enabled="true" />',
|
||
"",
|
||
)
|
||
device = FakeCollectDevice(incomplete_home, self.spec_xml)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
result = service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
self.assertIsNone(result.shop_name)
|
||
|
||
def test_missing_reviews_stops_when_goods_page_no_longer_changes(self):
|
||
home_without_reviews = self.home_xml.replace(
|
||
'<node class="android.widget.TextView" text="2356条评价" '
|
||
'bounds="[330,950][600,1010]" visible-to-user="true" '
|
||
'enabled="true" />',
|
||
"",
|
||
)
|
||
device = FakeCollectDevice(
|
||
home_without_reviews, keep_only_one_sku(self.spec_xml)
|
||
)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
max_spec_swipes=0,
|
||
)
|
||
|
||
result = service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
|
||
self.assertIsNone(result.reviews.raw)
|
||
self.assertTrue(device.clicks)
|
||
self.assertEqual(device.swipe_panel_states[:2], [False, False])
|
||
|
||
def test_goods_metadata_default_swipe_limit_is_three(self):
|
||
pages = [
|
||
f"""<hierarchy>
|
||
<node class="androidx.viewpager.widget.ViewPager"
|
||
content-desc="测试商品标题第{index}屏内容"
|
||
bounds="[0,100][1080,500]" />
|
||
<node class="android.widget.TextView" text="已拼12000件"
|
||
bounds="[30,600][300,660]" />
|
||
<node class="androidx.recyclerview.widget.RecyclerView"
|
||
scrollable="true" bounds="[0,0][1080,1800]" />
|
||
</hierarchy>"""
|
||
for index in range(7)
|
||
]
|
||
device = GoodsMetadataScrollDevice(pages, self.spec_xml)
|
||
device.opened_url = "https://mobile.yangkeduo.com/goods.html?goods_id=123"
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
)
|
||
|
||
result = service._collect_goods_details(device)
|
||
|
||
self.assertIsNone(result.shop_name)
|
||
self.assertIsNone(result.reviews.raw)
|
||
self.assertEqual(3, len(device.swipes))
|
||
self.assertEqual(3, device.page_index)
|
||
|
||
def test_cancel_during_goods_page_scroll_stops_before_spec_panel(self):
|
||
incomplete_home = self.home_xml.replace(
|
||
'<node class="android.widget.TextView" text="2356条评价" '
|
||
'bounds="[330,950][600,1010]" visible-to-user="true" '
|
||
'enabled="true" />',
|
||
"",
|
||
).replace(
|
||
'<node class="android.widget.TextView" text="店铺：测试服饰旗舰店" '
|
||
'bounds="[30,1050][600,1110]" visible-to-user="true" '
|
||
'enabled="true" />',
|
||
"",
|
||
)
|
||
cancelled = {"value": False}
|
||
device = FakeCollectDevice(incomplete_home, self.spec_xml)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: cancelled.update(value=True),
|
||
cancelled=lambda: cancelled["value"],
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
|
||
self.assertEqual(raised.exception.code, "PDD_CANCELLED")
|
||
self.assertEqual(device.swipe_panel_states, [False])
|
||
self.assertEqual(device.clicks, [])
|
||
|
||
def test_spec_panel_timeout_has_stable_code_and_xml_evidence(self):
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
device = PanelDoesNotOpenDevice(self.home_xml, self.spec_xml)
|
||
tick = {"value": -0.5}
|
||
|
||
def clock():
|
||
tick["value"] += 0.5
|
||
return tick["value"]
|
||
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
monotonic=clock,
|
||
spec_panel_timeout=1.0,
|
||
artifact_directory=Path(directory),
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
|
||
self.assertEqual(
|
||
raised.exception.code, "PDD_PAGE_SPEC_PANEL_TIMEOUT"
|
||
)
|
||
artifacts = raised.exception.diagnostics["artifacts"]
|
||
self.assertTrue(Path(artifacts[0]["path"]).is_file())
|
||
|
||
def test_open_confirmation_with_unrecognized_structure_is_not_timeout(self):
|
||
malformed_panel = self.submit_hint_spec_xml.replace(
|
||
'<node package="com.xunmeng.pinduoduo" class="android.widget.EditText"\n'
|
||
' text="1" clickable="true" bounds="[480,829][561,904]"\n'
|
||
' visible-to-user="true" enabled="true" />',
|
||
"",
|
||
)
|
||
device = FakeCollectDevice(self.home_xml, malformed_panel)
|
||
device.opened_url = "https://mobile.yangkeduo.com/goods.html?goods_id=123"
|
||
device.panel_open = True
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
spec_panel_timeout=1.0,
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service._wait_spec_panel(device)
|
||
|
||
self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
|
||
self.assertIn("已经打开", raised.exception.message)
|
||
|
||
def test_transient_empty_spec_trees_are_skipped_during_full_collection(self):
|
||
device = TransientSpecTreeDevice(
|
||
self.home_xml, keep_only_one_sku(self.spec_xml)
|
||
)
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=lambda _seconds: None,
|
||
max_spec_swipes=0,
|
||
)
|
||
|
||
result = service.collect(
|
||
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
|
||
)
|
||
|
||
self.assertEqual("测试纯棉短袖商品", result.title)
|
||
self.assertTrue(result.skus)
|
||
self.assertGreater(device.panel_reads, 4)
|
||
self.assertEqual(TRANSIENT_EMPTY_SPEC_XML, service._last_invalid_spec_xml)
|
||
self.assertIn("颜色分类", service._last_valid_spec_xml)
|
||
|
||
def test_persistent_empty_spec_tree_has_lost_code_and_dual_evidence(self):
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
device = SpecTreeStaysBlankDevice(
|
||
self.home_xml, keep_only_one_sku(self.spec_xml)
|
||
)
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
spec_panel_timeout=1.0,
|
||
max_spec_swipes=0,
|
||
artifact_directory=Path(directory),
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service.collect(
|
||
FakeTask(
|
||
"https://mobile.yangkeduo.com/goods.html?goods_id=123"
|
||
)
|
||
)
|
||
|
||
self.assertEqual("PDD_PAGE_SPEC_PANEL_LOST", raised.exception.code)
|
||
paths = [
|
||
Path(item["path"]).name
|
||
for item in raised.exception.diagnostics["artifacts"]
|
||
]
|
||
self.assertTrue(any(name.startswith("last-valid-spec-") for name in paths))
|
||
self.assertTrue(any(name.startswith("last-invalid-spec-") for name in paths))
|
||
self.assertLessEqual(clock.now, 1.2)
|
||
|
||
def test_special_page_is_not_retried_as_transient_empty_tree(self):
|
||
device = FakeCollectDevice(self.home_xml, self.spec_xml)
|
||
device.panel_open = True
|
||
device.opened_url = "https://mobile.yangkeduo.com/goods.html?goods_id=123"
|
||
device.spec_xml = (
|
||
'<hierarchy><node text="请完成验证" '
|
||
'package="com.xunmeng.pinduoduo" /></hierarchy>'
|
||
)
|
||
clock = FakeClock()
|
||
service = PddCollectService(
|
||
PddDeviceService(lambda _serial: device),
|
||
"USB-001",
|
||
"client-001",
|
||
sleeper=clock.sleep,
|
||
monotonic=clock.monotonic,
|
||
)
|
||
|
||
with self.assertRaises(PddCollectError) as raised:
|
||
service._read_valid_spec_panel(device)
|
||
|
||
self.assertEqual("PDD_PAGE_CAPTCHA", raised.exception.code)
|
||
self.assertEqual([], clock.sleeps)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|