feat(client): connect PDD task list to SQLite (#9)

2026-08-06 16:31:34 +08:00
parent be6ee3d880
commit 583e441e17
4 changed files with 334 additions and 90 deletions
@@ -0,0 +1,142 @@
+"""PDD 任务列表接入 SQLite 的离屏测试。"""
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
+
+from PyQt5.QtWidgets import QApplication
+
+from src.pdd_ui import PDDTaskPage
+from src.pdd_ui_event import PDDTaskPageEvent, summary_to_row
+from src.task_models import NewClaimedTask, TaskStatus, TaskSummary, TaskType
+from src.task_repository import TaskRepository
+from src.ui_main import MainWindow
+
+
+class BrokenRepository:
+    """模拟无法读取数据库的 Repository。"""
+
+    def list_tasks(self, filters=None, limit=50, offset=0):
+        raise RuntimeError("database is unavailable")
+
+
+class PDDTaskPageEventTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.app = QApplication.instance() or QApplication([])
+
+    def setUp(self):
+        self.temp_directory = tempfile.TemporaryDirectory()
+        self.db_path = Path(self.temp_directory.name) / "client.db"
+        self.repository = TaskRepository(self.db_path)
+
+    def tearDown(self):
+        self.temp_directory.cleanup()
+
+    def _add_task(
+        self,
+        number: int,
+        task_type: TaskType = TaskType.COLLECT,
+        title: str = "测试商品",
+    ) -> None:
+        self.repository.add_claimed_task(
+            NewClaimedTask(
+                remote_task_id=f"PDD-{number:03d}",
+                task_type=task_type,
+                goods_url=f"pdd://goods/{number}",
+                goods_id=str(number),
+                title=title,
+                target_color="黑色",
+                target_size="M",
+                price_cent=3990,
+                quantity=1 if task_type is TaskType.PURCHASE else None,
+            ),
+            received_at="2026-08-06T08:00:00Z",
+        )
+
+    def test_initial_load_and_fetch_next_page(self):
+        for number in range(51):
+            self._add_task(number)
+
+        page = PDDTaskPage()
+        events = PDDTaskPageEvent(page, self.repository)
+        events.load_initial_tasks()
+
+        self.assertEqual(page.taskModel.data_row_count(), 50)
+        self.assertEqual(page.taskModel.row_at(0).remote_task_id, "PDD-050")
+        self.assertTrue(page.taskModel.canFetchMore())
+
+        page.taskModel.fetchMore()
+
+        self.assertEqual(page.taskModel.data_row_count(), 51)
+        self.assertEqual(page.taskModel.row_at(50).remote_task_id, "PDD-000")
+        self.assertFalse(page.taskModel.canFetchMore())
+        page.deleteLater()
+
+    def test_search_uses_type_status_and_keyword_together(self):
+        self._add_task(1, TaskType.COLLECT, "目标短袖")
+        self._add_task(2, TaskType.PURCHASE, "目标短袖")
+        self._add_task(3, TaskType.COLLECT, "其他商品")
+
+        page = PDDTaskPage()
+        events = PDDTaskPageEvent(page, self.repository)
+        page.searchRequested.emit(
+            {"task_type": "采集", "status": "待执行", "keyword": "目标"}
+        )
+
+        self.assertEqual(page.taskModel.data_row_count(), 1)
+        row = page.taskModel.row_at(0)
+        self.assertEqual(row.remote_task_id, "PDD-001")
+        self.assertEqual(row.task_type, "采集")
+        self.assertEqual(row.status, "待执行")
+        self.assertEqual(row.price_cents, 3990)
+        page.deleteLater()
+
+    def test_query_failure_shows_readable_error(self):
+        page = PDDTaskPage()
+        events = PDDTaskPageEvent(page, BrokenRepository())
+
+        events.load_initial_tasks()
+
+        self.assertFalse(page.emptyStateCard.isHidden())
+        self.assertEqual(page.emptyTitleLabel.text(), "任务加载失败")
+        self.assertIn("检查数据库", page.emptyMessageLabel.text())
+        self.assertFalse(page.taskModel.canFetchMore())
+        page.deleteLater()
+
+    def test_summary_to_row_does_not_copy_detail_json(self):
+        summary = TaskSummary(
+            id=1,
+            remote_task_id="PDD-001",
+            task_type=TaskType.PURCHASE,
+            goods_id=None,
+            title=None,
+            target_color=None,
+            target_size=None,
+            price_cent=None,
+            quantity=2,
+            status=TaskStatus.MANUAL_REVIEW,
+            updated_at="2026-08-06T08:00:00Z",
+        )
+
+        row = summary_to_row(summary)
+
+        self.assertEqual(row.task_type, "采购")
+        self.assertEqual(row.status, "需要人工处理")
+        self.assertEqual(row.goods_id, "")
+        self.assertFalse(hasattr(row, "pdd_data"))
+
+    def test_main_window_keeps_event_object_alive(self):
+        window = MainWindow(task_repository=self.repository)
+
+        self.assertIsInstance(window.pddTaskPageEvent, PDDTaskPageEvent)
+        self.assertEqual(window.pddTaskPage.taskModel.data_row_count(), 0)
+        window.close()
+        window.deleteLater()
+
+
+if __name__ == "__main__":
+    unittest.main()