"""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 """ """ @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 '' 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 SoldOutRefreshCollectDevice(FakeCollectDevice): """模拟深链先显示售罄，向下拉一次后恢复或继续售罄。""" def __init__(self, sold_out_xml, recovered_xml, *, persistent=False): super().__init__(recovered_xml, "") self.sold_out_xml = sold_out_xml self.recovered_xml = recovered_xml self.persistent = persistent self.refreshed = False def dump_hierarchy(self): if self.opened_url is None: return '' if not self.refreshed or self.persistent: return self.sold_out_xml return self.recovered_xml def swipe(self, x1, y1, x2, y2, duration=0.0): self.swipes.append(((x1, y1, x2, y2), {"duration": duration})) if y2 > y1 and x1 == x2: self.refreshed = True 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 '' 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 = """ """ 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'' ) size_nodes = "" if self.sizes_visible: size_nodes = """ """ price = self.prices[self.selected_color] / 100 return f""" """ 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 """ """ 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 = ( '' ) new = ( '' '' '' '' ) 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'' ) return f""" """ 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 = """ """ options = [] for index, text in enumerate(visible): top = 1080 + index * 210 right = 500 if self.page == 0 else 1015 options.append( f'' ) return f""" """ 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 = """ """ 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( '', '\n' '\n' '', ) 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( '', "", ).replace( '', "", ) 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 = """ """ 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 = """ """ 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 = """ """ 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 = """ """ self.assertFalse(_is_spec_panel_open(xml_data)) def test_open_panel_without_visible_options_is_not_reported_as_timeout(self): panel_xml = """ """ 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 = """ """ 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('') with self.assertRaises(PddCollectError) as invalid: parse_goods_page("") 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('') with self.assertRaises(PddCollectError) as incomplete: parse_spec_panel('') 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( '', "", ).replace( '', "", ) buy_button = """ """ review_page = f""" """ shop_page = f""" """ 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( " 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_sold_out_page_pulls_down_once_then_recovers_goods(self): device = SoldOutRefreshCollectDevice( self.sold_out_xml, self.home_xml, ) service = PddCollectService( PddDeviceService(lambda _serial: device), "USB-001", "client-001", sleeper=lambda _seconds: None, ) service._open_goods( device, "https://mobile.yangkeduo.com/goods.html?goods_id=123", {"package": "com.xunmeng.pinduoduo"}, ) self.assertEqual(len(device.swipes), 1) (x1, y1, x2, y2), options = device.swipes[0] self.assertEqual(x1, x2) self.assertGreater(y2, y1) self.assertEqual(options["duration"], 0.5) self.assertEqual(device.clicks, []) def test_sold_out_after_one_refresh_is_unavailable(self): device = SoldOutRefreshCollectDevice( self.sold_out_xml, self.home_xml, persistent=True, ) service = PddCollectService( PddDeviceService(lambda _serial: device), "USB-001", "client-001", sleeper=lambda _seconds: None, ) with self.assertRaises(PddCollectError) as raised: service._open_goods( device, "https://mobile.yangkeduo.com/goods.html?goods_id=123", {"package": "com.xunmeng.pinduoduo"}, ) self.assertEqual(raised.exception.code, "PDD_GOODS_UNAVAILABLE") self.assertEqual( raised.exception.diagnostics, {"page_reason": "out_of_stock", "refresh_attempts": 1}, ) self.assertEqual(len(device.swipes), 1) self.assertEqual(device.clicks, []) 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( '', 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( '', "", ) 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( '', "", ) 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""" """ 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( '', "", ).replace( '', "", ) 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( '', "", ) 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 = ( '' ) 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) def test_valid_initial_spec_frame_is_used_before_reading_device_again(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" service = PddCollectService( PddDeviceService(lambda _serial: device), "USB-001", "client-001", sleeper=lambda _seconds: None, ) xml_data, snapshot = service._read_valid_spec_panel( device, initial_xml=self.spec_xml ) self.assertEqual(self.spec_xml, xml_data) self.assertTrue(snapshot.dimensions) self.assertEqual(0, service._goods_screens_checked) if __name__ == "__main__": unittest.main()