fix: 兼容快速确认页进入规格面板 (#190)

client/test/test_pdd_collect_service.py

@@ -158,6 +158,30 @@ class PanelDoesNotOpenDevice(FakeCollectDevice):
         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]">
@@ -612,6 +636,9 @@ class PddCollectParserTest(unittest.TestCase):
         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")
@@ -944,6 +971,87 @@ class PddCollectParserTest(unittest.TestCase):
 
         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))