"""任务数据模型和 TaskRepository 测试。""" import tempfile import unittest from pathlib import Path from src.db import open_database from src.task_models import ( NewClaimedTask, OutboxEventType, OutboxStatus, RunStatus, TaskFilters, TaskStatus, TaskType, ) from src.task_repository import DuplicateTaskError, TaskRepository class TaskRepositoryTests(unittest.TestCase): def setUp(self) -> None: self._temporary_directory = tempfile.TemporaryDirectory() self.db_path = Path(self._temporary_directory.name) / "client.db" self.repository = TaskRepository(self.db_path) def tearDown(self) -> None: self._temporary_directory.cleanup() @staticmethod def _task( remote_task_id: str, task_type: TaskType = TaskType.COLLECT, title: str = "测试商品", goods_id: str = "10001", ) -> NewClaimedTask: return NewClaimedTask( remote_task_id=remote_task_id, task_type=task_type, goods_id=goods_id, goods_url=f"https://example.test/goods/{goods_id}", title=title, target_color="黑色" if task_type == TaskType.PURCHASE else None, target_size="L" if task_type == TaskType.PURCHASE else None, price_cent=3990, quantity=2 if task_type == TaskType.PURCHASE else None, admin_payload={"schema_version": 1, "task_id": remote_task_id}, ) def test_enum_values_match_database_constraints(self) -> None: self.assertEqual({item.value for item in TaskType}, {"collect", "purchase"}) self.assertEqual( {item.value for item in TaskStatus}, { "claimed", "running", "result_pending", "retry_wait", "manual_review", "succeeded", "failed", "cancelled", }, ) self.assertEqual( {item.value for item in RunStatus}, {"running", "succeeded", "failed", "cancelled", "manual_review"}, ) self.assertEqual( {item.value for item in OutboxEventType}, {"collect_result", "purchase_result", "task_failure"}, ) self.assertEqual( {item.value for item in OutboxStatus}, {"pending", "sending", "sent", "failed"}, ) def test_add_claimed_task_and_read_detail(self) -> None: task_id = self.repository.add_claimed_task( self._task("TASK-001"), "2026-08-06T08:00:00Z" ) connection = open_database(self.db_path) try: with connection: connection.execute( "UPDATE pdd_tasks SET pdd_data = ? WHERE id = ?", ('{"schema_version": 1, "goods": {"title": "测试商品"}}', task_id), ) finally: connection.close() detail = self.repository.get_task("TASK-001") self.assertIsNotNone(detail) self.assertEqual(detail.id, task_id) self.assertEqual(detail.status, TaskStatus.CLAIMED) self.assertEqual(detail.task_type, TaskType.COLLECT) self.assertEqual(detail.admin_payload["task_id"], "TASK-001") self.assertEqual(detail.pdd_data["schema_version"], 1) def test_duplicate_remote_task_id_does_not_overwrite(self) -> None: self.repository.add_claimed_task(self._task("TASK-001")) with self.assertRaises(DuplicateTaskError): self.repository.add_claimed_task( self._task("TASK-001", title="不应覆盖的新标题") ) detail = self.repository.get_task("TASK-001") self.assertEqual(detail.title, "测试商品") def test_list_tasks_uses_stable_paging_order(self) -> None: same_time = "2026-08-06T08:00:00Z" self.repository.add_claimed_task(self._task("TASK-001"), same_time) self.repository.add_claimed_task(self._task("TASK-002"), same_time) self.repository.add_claimed_task( self._task("TASK-003"), "2026-08-06T09:00:00Z" ) first_page = self.repository.list_tasks(limit=2) second_page = self.repository.list_tasks(limit=2, offset=2) self.assertEqual( [task.remote_task_id for task in first_page], ["TASK-003", "TASK-002"] ) self.assertEqual( [task.remote_task_id for task in second_page], ["TASK-001"] ) self.assertFalse(hasattr(first_page[0], "pdd_data")) self.assertFalse(hasattr(first_page[0], "admin_payload")) def test_filters_use_and_relationship_and_count_matches(self) -> None: self.repository.add_claimed_task( self._task("COLLECT-BLACK", title="黑色短袖", goods_id="20001") ) self.repository.add_claimed_task( self._task( "PURCHASE-BLACK", task_type=TaskType.PURCHASE, title="黑色长裙", goods_id="20002", ) ) self.repository.add_claimed_task( self._task("COLLECT-WHITE", title="白色短袖", goods_id="20003") ) connection = open_database(self.db_path) try: with connection: connection.execute( "UPDATE pdd_tasks SET status = 'running'" " WHERE remote_task_id = 'COLLECT-BLACK'" ) finally: connection.close() filters = TaskFilters( task_type=TaskType.COLLECT, status=TaskStatus.RUNNING, keyword="黑色", ) tasks = self.repository.list_tasks(filters) self.assertEqual([task.remote_task_id for task in tasks], ["COLLECT-BLACK"]) self.assertEqual(self.repository.count_tasks(filters), 1) def test_keyword_treats_percent_as_normal_text(self) -> None: self.repository.add_claimed_task(self._task("TASK-100%", title="百分号")) self.repository.add_claimed_task(self._task("TASK-OTHER", title="普通商品")) tasks = self.repository.list_tasks(TaskFilters(keyword="100%")) self.assertEqual([task.remote_task_id for task in tasks], ["TASK-100%"]) def test_invalid_model_and_page_parameters_are_rejected(self) -> None: with self.assertRaisesRegex(ValueError, "price_cent"): NewClaimedTask( remote_task_id="TASK-001", task_type=TaskType.COLLECT, goods_url="https://example.test/goods", price_cent=-1, ) with self.assertRaisesRegex(ValueError, "limit"): self.repository.list_tasks(limit=0) with self.assertRaisesRegex(ValueError, "offset"): self.repository.list_tasks(offset=-1) def test_result_and_outbox_are_saved_together_then_marked_succeeded(self): self.repository.add_claimed_task(self._task("TASK-RUN")) started = self.repository.start_collect_run("TASK-RUN", "USB-001") event = self.repository.save_collect_result( "TASK-RUN", started.attempt_id, { "goods_id": "10001", "title": "测试商品", "price_granularity": "color", "dimensions": [], "skus": [{"price_cent": 990}], }, ) self.assertEqual( self.repository.get_task("TASK-RUN").status, TaskStatus.RESULT_PENDING, ) self.assertEqual(event.status, OutboxStatus.PENDING) self.repository.mark_outbox_sent(event.id) self.assertEqual( self.repository.get_task("TASK-RUN").status, TaskStatus.SUCCEEDED, ) def test_recovery_restores_sending_and_interrupted_running(self): self.repository.add_claimed_task(self._task("TASK-RECOVER")) started = self.repository.start_collect_run("TASK-RECOVER", "USB-001") self.repository.save_collect_failure( "TASK-RECOVER", started.attempt_id, TaskStatus.RETRY_WAIT, "DEVICE_OFFLINE", "设备离线", True, ) event = self.repository.next_pending_outbox() self.repository.mark_outbox_sending(event.id) self.repository.recover_interrupted_work() self.assertEqual( self.repository.get_outbox_event(event.id).status, OutboxStatus.PENDING, ) if __name__ == "__main__": unittest.main()