256 lines
9.1 KiB
Python
256 lines
9.1 KiB
Python
"""任务数据模型和 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,
|
|
)
|
|
|
|
def test_recovery_retries_old_ambiguous_admin_response(self):
|
|
self.repository.add_claimed_task(self._task("TASK-AMBIGUOUS"))
|
|
started = self.repository.start_collect_run("TASK-AMBIGUOUS", "USB-001")
|
|
event = self.repository.save_collect_failure(
|
|
"TASK-AMBIGUOUS",
|
|
started.attempt_id,
|
|
TaskStatus.RETRY_WAIT,
|
|
"PDD_PAGE_TIMEOUT",
|
|
"页面超时",
|
|
True,
|
|
)
|
|
self.repository.mark_outbox_failed(event.id, "Admin 提交响应字段不完整")
|
|
|
|
self.repository.recover_interrupted_work()
|
|
|
|
self.assertEqual(
|
|
self.repository.get_outbox_event(event.id).status,
|
|
OutboxStatus.PENDING,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|