perf: 优化颜色采集等待时间 (#44)

client/test/test_pdd_collect_service.py

@@ -239,6 +239,58 @@ class AllMissingColorPriceDevice(IgnoredColorClickDevice):
         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 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，只保留一个组合。"""
 
@@ -569,6 +621,87 @@ class PddCollectParserTest(unittest.TestCase):
 
         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_all_missing_color_prices_still_collects_sizes(self):
         device = AllMissingColorPriceDevice(self.home_xml)
         service = PddCollectService(