747 lines
29 KiB
Python
747 lines
29 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 (
|
|
CollectRerunError,
|
|
DuplicateTaskError,
|
|
TaskRemovalError,
|
|
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()
|
|
|
|
def _set_task_status(self, task_id: str, status: TaskStatus) -> None:
|
|
connection = open_database(self.db_path)
|
|
try:
|
|
with connection:
|
|
connection.execute(
|
|
"UPDATE pdd_tasks SET status = ? WHERE remote_task_id = ?",
|
|
(status.value, task_id),
|
|
)
|
|
finally:
|
|
connection.close()
|
|
|
|
@staticmethod
|
|
def _task(
|
|
remote_task_id: str,
|
|
task_type: TaskType = TaskType.COLLECT,
|
|
title: str = "测试商品",
|
|
goods_id: str = "10001",
|
|
execution_mode: str = "dry_run",
|
|
) -> NewClaimedTask:
|
|
return NewClaimedTask(
|
|
remote_task_id=remote_task_id,
|
|
task_type=task_type,
|
|
execution_mode=execution_mode,
|
|
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_live_mode_and_irreversible_marker_are_persisted_once(self) -> None:
|
|
self.repository.add_claimed_task(
|
|
self._task(
|
|
"PURCHASE-LIVE",
|
|
TaskType.PURCHASE,
|
|
execution_mode="live",
|
|
)
|
|
)
|
|
detail = self.repository.get_task("PURCHASE-LIVE")
|
|
self.assertEqual(detail.execution_mode, "live")
|
|
|
|
started = self.repository.start_purchase_run(
|
|
"PURCHASE-LIVE", "192.168.0.173:5555"
|
|
)
|
|
marked_at = self.repository.mark_purchase_irreversible(
|
|
"PURCHASE-LIVE",
|
|
started.attempt_id,
|
|
{
|
|
"options": {"color": "黑色", "size": "L"},
|
|
"quantity": 2,
|
|
"unit_price_cent": 3990,
|
|
"total_price_cent": 7980,
|
|
},
|
|
)
|
|
run = self.repository.latest_task_run("PURCHASE-LIVE")
|
|
self.assertEqual(run.irreversible_action_at, marked_at)
|
|
self.assertEqual(
|
|
run.diagnostics_json["final_confirmation"]["total_price_cent"],
|
|
7980,
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "已经存在"):
|
|
self.repository.mark_purchase_irreversible(
|
|
"PURCHASE-LIVE", started.attempt_id
|
|
)
|
|
|
|
def test_dry_run_cannot_enter_irreversible_stage(self) -> None:
|
|
self.repository.add_claimed_task(
|
|
self._task("PURCHASE-DRY", TaskType.PURCHASE)
|
|
)
|
|
started = self.repository.start_purchase_run(
|
|
"PURCHASE-DRY", "USB-001"
|
|
)
|
|
|
|
with self.assertRaisesRegex(ValueError, "演练任务"):
|
|
self.repository.mark_purchase_irreversible(
|
|
"PURCHASE-DRY", started.attempt_id
|
|
)
|
|
|
|
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_list_and_keyword_search_extract_purchase_order_no_safely(self) -> None:
|
|
self.repository.add_claimed_task(
|
|
self._task("PURCHASE-ORDER", TaskType.PURCHASE)
|
|
)
|
|
self.repository.add_claimed_task(self._task("COLLECT-BROKEN"))
|
|
connection = open_database(self.db_path)
|
|
try:
|
|
with connection:
|
|
connection.execute(
|
|
"UPDATE pdd_tasks SET pdd_data = ? WHERE remote_task_id = ?",
|
|
(
|
|
'{"purchase":{"order_no":"ORDER-20260811"}}',
|
|
"PURCHASE-ORDER",
|
|
),
|
|
)
|
|
connection.execute(
|
|
"UPDATE pdd_tasks SET pdd_data = ? WHERE remote_task_id = ?",
|
|
("不是有效JSON", "COLLECT-BROKEN"),
|
|
)
|
|
finally:
|
|
connection.close()
|
|
|
|
tasks = self.repository.list_tasks()
|
|
by_id = {task.remote_task_id: task for task in tasks}
|
|
searched = self.repository.list_tasks(
|
|
TaskFilters(keyword="20260811")
|
|
)
|
|
|
|
self.assertEqual(by_id["PURCHASE-ORDER"].order_no, "ORDER-20260811")
|
|
self.assertIsNone(by_id["COLLECT-BROKEN"].order_no)
|
|
self.assertEqual(
|
|
[task.remote_task_id for task in searched], ["PURCHASE-ORDER"]
|
|
)
|
|
self.assertEqual(
|
|
self.repository.count_tasks(TaskFilters(keyword="ORDER-20260811")),
|
|
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": "测试商品",
|
|
"shop_name": "测试店铺",
|
|
"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)
|
|
summary = self.repository.list_tasks()[0]
|
|
self.assertEqual(summary.shop_name, "测试店铺")
|
|
self.assertEqual(summary.latest_run_status, RunStatus.SUCCEEDED)
|
|
self.repository.mark_outbox_sent(event.id)
|
|
self.assertEqual(
|
|
self.repository.get_task("TASK-RUN").status,
|
|
TaskStatus.SUCCEEDED,
|
|
)
|
|
|
|
def test_list_uses_latest_run_duration_and_clamps_negative_value(self):
|
|
self.repository.add_claimed_task(self._task("TASK-DURATION"))
|
|
started = self.repository.start_collect_run("TASK-DURATION", "USB-001")
|
|
connection = open_database(self.db_path)
|
|
try:
|
|
with connection:
|
|
connection.execute(
|
|
"UPDATE task_runs SET run_status = 'failed',"
|
|
" started_at = ?, finished_at = ? WHERE attempt_id = ?",
|
|
(
|
|
"2026-08-11T08:00:10Z",
|
|
"2026-08-11T08:00:05Z",
|
|
started.attempt_id,
|
|
),
|
|
)
|
|
finally:
|
|
connection.close()
|
|
|
|
summary = self.repository.list_tasks()[0]
|
|
|
|
self.assertEqual(summary.latest_run_status, RunStatus.FAILED)
|
|
self.assertEqual(summary.duration_seconds, 0)
|
|
|
|
def test_latest_result_outbox_excludes_newer_failure_event(self):
|
|
self.repository.add_claimed_task(self._task("TASK-REPORT"))
|
|
first = self.repository.start_collect_run("TASK-REPORT", "USB-001")
|
|
result = self.repository.save_collect_result(
|
|
"TASK-REPORT",
|
|
first.attempt_id,
|
|
{"goods_id": "10001", "title": "测试商品", "skus": []},
|
|
)
|
|
self.repository.mark_outbox_sent(result.id)
|
|
self.repository.prepare_collect_rerun("TASK-REPORT")
|
|
second = self.repository.start_collect_run("TASK-REPORT", "USB-001")
|
|
failure = self.repository.save_collect_failure(
|
|
"TASK-REPORT",
|
|
second.attempt_id,
|
|
TaskStatus.FAILED,
|
|
"TEST_FAILURE",
|
|
"测试失败",
|
|
False,
|
|
)
|
|
|
|
latest = self.repository.latest_result_outbox("TASK-REPORT")
|
|
|
|
self.assertEqual(latest.id, result.id)
|
|
self.assertEqual(latest.event_type, OutboxEventType.COLLECT_RESULT)
|
|
self.assertNotEqual(latest.id, failure.id)
|
|
self.assertEqual(
|
|
self.repository.outbox_for_resubmit("TASK-REPORT").id,
|
|
failure.id,
|
|
)
|
|
self.repository.mark_outbox_sent(failure.id)
|
|
self.assertEqual(
|
|
self.repository.outbox_for_resubmit("TASK-REPORT").id,
|
|
result.id,
|
|
)
|
|
self.assertIsNone(self.repository.latest_result_outbox("MISSING"))
|
|
self.assertIsNone(self.repository.outbox_for_resubmit("MISSING"))
|
|
|
|
def test_replaying_old_result_does_not_hide_latest_failure(self):
|
|
self.repository.add_claimed_task(self._task("TASK-STATE"))
|
|
first = self.repository.start_collect_run("TASK-STATE", "USB-001")
|
|
result = self.repository.save_collect_result(
|
|
"TASK-STATE",
|
|
first.attempt_id,
|
|
{"goods_id": "10001", "title": "旧结果", "skus": []},
|
|
)
|
|
self.repository.mark_outbox_sent(result.id)
|
|
self.repository.prepare_collect_rerun("TASK-STATE")
|
|
second = self.repository.start_collect_run("TASK-STATE", "USB-001")
|
|
failure = self.repository.save_collect_failure(
|
|
"TASK-STATE",
|
|
second.attempt_id,
|
|
TaskStatus.FAILED,
|
|
"SKU_PANEL_NOT_FOUND",
|
|
"规格面板加载超时",
|
|
False,
|
|
)
|
|
|
|
self.repository.mark_outbox_sending(result.id)
|
|
self.repository.mark_outbox_sent(result.id)
|
|
|
|
detail = self.repository.get_task("TASK-STATE")
|
|
self.assertEqual(detail.status, TaskStatus.FAILED)
|
|
self.assertEqual(detail.last_error_code, "SKU_PANEL_NOT_FOUND")
|
|
|
|
# 模拟旧版本已经把主状态错误改成“已完成”。发送最新失败信息后
|
|
# 应当依据最新执行记录恢复，而不是继续保留假完成状态。
|
|
connection = open_database(self.db_path)
|
|
try:
|
|
with connection:
|
|
connection.execute(
|
|
"UPDATE pdd_tasks SET status = 'succeeded',"
|
|
" current_step = 'completed' WHERE remote_task_id = ?",
|
|
("TASK-STATE",),
|
|
)
|
|
finally:
|
|
connection.close()
|
|
self.repository.mark_outbox_sending(failure.id)
|
|
self.repository.mark_outbox_sent(failure.id)
|
|
|
|
repaired = self.repository.get_task("TASK-STATE")
|
|
self.assertEqual(repaired.status, TaskStatus.FAILED)
|
|
self.assertEqual(repaired.current_step, "failed")
|
|
self.assertEqual(repaired.last_error_message, "规格面板加载超时")
|
|
|
|
def test_replaying_old_failure_does_not_override_new_success(self):
|
|
self.repository.add_claimed_task(self._task("TASK-NEW-SUCCESS"))
|
|
first = self.repository.start_collect_run(
|
|
"TASK-NEW-SUCCESS", "USB-001"
|
|
)
|
|
old_failure = self.repository.save_collect_failure(
|
|
"TASK-NEW-SUCCESS",
|
|
first.attempt_id,
|
|
TaskStatus.FAILED,
|
|
"OLD_FAILURE",
|
|
"旧执行失败",
|
|
False,
|
|
)
|
|
self.repository.mark_outbox_sent(old_failure.id)
|
|
self.repository.prepare_collect_rerun("TASK-NEW-SUCCESS")
|
|
second = self.repository.start_collect_run(
|
|
"TASK-NEW-SUCCESS", "USB-001"
|
|
)
|
|
new_result = self.repository.save_collect_result(
|
|
"TASK-NEW-SUCCESS",
|
|
second.attempt_id,
|
|
{"goods_id": "10001", "title": "新结果", "skus": []},
|
|
)
|
|
self.repository.mark_outbox_sent(new_result.id)
|
|
|
|
self.repository.mark_outbox_sending(old_failure.id)
|
|
self.repository.mark_outbox_sent(old_failure.id)
|
|
|
|
detail = self.repository.get_task("TASK-NEW-SUCCESS")
|
|
self.assertEqual(detail.status, TaskStatus.SUCCEEDED)
|
|
self.assertEqual(detail.current_step, "completed")
|
|
|
|
def test_prepare_rerun_preserves_old_result_and_creates_new_attempt(self):
|
|
self.repository.add_claimed_task(self._task("TASK-RERUN"))
|
|
first = self.repository.start_collect_run("TASK-RERUN", "USB-001")
|
|
first_data = {"goods_id": "10001", "title": "旧标题", "skus": []}
|
|
first_event = self.repository.save_collect_result(
|
|
"TASK-RERUN", first.attempt_id, first_data
|
|
)
|
|
self.repository.mark_outbox_sent(first_event.id)
|
|
|
|
prepared = self.repository.prepare_collect_rerun("TASK-RERUN")
|
|
second = self.repository.start_collect_run("TASK-RERUN", "USB-001")
|
|
second_data = {"goods_id": "10001", "title": "新标题", "skus": []}
|
|
self.repository.save_collect_result(
|
|
"TASK-RERUN", second.attempt_id, second_data
|
|
)
|
|
|
|
connection = open_database(self.db_path)
|
|
try:
|
|
rows = connection.execute(
|
|
"SELECT attempt_id, result_data FROM task_runs"
|
|
" WHERE task_id = ? ORDER BY attempt_no",
|
|
(prepared.id,),
|
|
).fetchall()
|
|
finally:
|
|
connection.close()
|
|
self.assertEqual(second.attempt_no, 2)
|
|
self.assertIn("旧标题", rows[0]["result_data"])
|
|
self.assertIn("新标题", rows[1]["result_data"])
|
|
self.assertEqual(self.repository.get_task("TASK-RERUN").title, "新标题")
|
|
|
|
def test_rerun_rejects_purchase_active_and_unsent_tasks(self):
|
|
self.repository.add_claimed_task(
|
|
self._task("PURCHASE-RERUN", TaskType.PURCHASE)
|
|
)
|
|
with self.assertRaisesRegex(CollectRerunError, "采购任务"):
|
|
self.repository.validate_collect_rerun("PURCHASE-RERUN")
|
|
|
|
self.repository.add_claimed_task(self._task("ACTIVE-RERUN"))
|
|
self.assertEqual(
|
|
self.repository.validate_collect_rerun("ACTIVE-RERUN").status,
|
|
TaskStatus.CLAIMED,
|
|
)
|
|
|
|
self.repository.add_claimed_task(self._task("UNSENT-RERUN"))
|
|
started = self.repository.start_collect_run("UNSENT-RERUN", "USB-001")
|
|
self.repository.save_collect_result(
|
|
"UNSENT-RERUN", started.attempt_id, {"title": "结果", "skus": []}
|
|
)
|
|
connection = open_database(self.db_path)
|
|
try:
|
|
with connection:
|
|
connection.execute(
|
|
"UPDATE pdd_tasks SET status = 'failed'"
|
|
" WHERE remote_task_id = 'UNSENT-RERUN'"
|
|
)
|
|
finally:
|
|
connection.close()
|
|
with self.assertRaisesRegex(CollectRerunError, "未发送"):
|
|
self.repository.validate_collect_rerun("UNSENT-RERUN")
|
|
|
|
def test_batch_rerun_plan_filters_other_task_type(self):
|
|
self.repository.add_claimed_task(self._task("COLLECT-PLAN"))
|
|
self.repository.add_claimed_task(
|
|
self._task("PURCHASE-PLAN", TaskType.PURCHASE)
|
|
)
|
|
|
|
plan = self.repository.plan_rerun_batch(
|
|
("COLLECT-PLAN", "PURCHASE-PLAN"), TaskType.COLLECT
|
|
)
|
|
|
|
self.assertEqual(plan.eligible_task_ids, ("COLLECT-PLAN",))
|
|
self.assertEqual(plan.filtered_task_ids, ("PURCHASE-PLAN",))
|
|
self.assertEqual(plan.blocked, ())
|
|
|
|
def test_purchase_rerun_permanently_rejects_irreversible_history(self):
|
|
self.repository.add_claimed_task(
|
|
self._task("PURCHASE-SAFE", TaskType.PURCHASE)
|
|
)
|
|
started = self.repository.start_purchase_run("PURCHASE-SAFE", "USB-001")
|
|
connection = open_database(self.db_path)
|
|
try:
|
|
with connection:
|
|
connection.execute(
|
|
"UPDATE task_runs SET irreversible_action_at = ? WHERE attempt_id = ?",
|
|
("2026-08-11T00:00:00Z", started.attempt_id),
|
|
)
|
|
connection.execute(
|
|
"UPDATE pdd_tasks SET status = 'manual_review' WHERE remote_task_id = ?",
|
|
("PURCHASE-SAFE",),
|
|
)
|
|
finally:
|
|
connection.close()
|
|
|
|
plan = self.repository.plan_rerun_batch(
|
|
("PURCHASE-SAFE",), TaskType.PURCHASE
|
|
)
|
|
|
|
self.assertEqual(plan.eligible_task_ids, ())
|
|
self.assertIn("只准核对订单", plan.blocked[0][1])
|
|
|
|
def test_retry_wait_can_rerun_only_after_failure_outbox_is_sent(self):
|
|
self.repository.add_claimed_task(self._task("RETRY-RERUN"))
|
|
first = self.repository.start_collect_run("RETRY-RERUN", "USB-001")
|
|
event = self.repository.save_collect_failure(
|
|
"RETRY-RERUN",
|
|
first.attempt_id,
|
|
TaskStatus.RETRY_WAIT,
|
|
"DEVICE_OFFLINE",
|
|
"设备离线",
|
|
True,
|
|
)
|
|
|
|
with self.assertRaisesRegex(CollectRerunError, "未上报的失败信息"):
|
|
self.repository.validate_collect_rerun("RETRY-RERUN")
|
|
|
|
self.repository.mark_outbox_sent(event.id)
|
|
prepared = self.repository.prepare_collect_rerun("RETRY-RERUN")
|
|
second = self.repository.start_collect_run("RETRY-RERUN", "USB-001")
|
|
|
|
self.assertEqual(prepared.status, TaskStatus.CLAIMED)
|
|
self.assertEqual(second.attempt_no, 2)
|
|
|
|
def test_retry_wait_explains_permanently_failed_outbox(self):
|
|
self.repository.add_claimed_task(self._task("RETRY-REJECTED"))
|
|
started = self.repository.start_collect_run(
|
|
"RETRY-REJECTED", "USB-001"
|
|
)
|
|
event = self.repository.save_collect_failure(
|
|
"RETRY-REJECTED",
|
|
started.attempt_id,
|
|
TaskStatus.RETRY_WAIT,
|
|
"DEVICE_OFFLINE",
|
|
"设备离线",
|
|
True,
|
|
)
|
|
self.repository.mark_outbox_failed(event.id, "任务不存在")
|
|
|
|
with self.assertRaisesRegex(
|
|
CollectRerunError, "上次失败信息提交失败：任务不存在"
|
|
):
|
|
self.repository.validate_collect_rerun("RETRY-REJECTED")
|
|
|
|
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,
|
|
)
|
|
|
|
def test_terminal_task_is_soft_removed_but_audit_record_remains(self):
|
|
self.repository.add_claimed_task(self._task("REMOVE-OK"))
|
|
self._set_task_status("REMOVE-OK", TaskStatus.SUCCEEDED)
|
|
|
|
removed_count = self.repository.remove_tasks_from_list(("REMOVE-OK",))
|
|
|
|
self.assertEqual(removed_count, 1)
|
|
self.assertEqual(self.repository.list_tasks(), [])
|
|
self.assertEqual(self.repository.count_tasks(), 0)
|
|
self.assertIsNotNone(self.repository.get_task("REMOVE-OK"))
|
|
connection = open_database(self.db_path)
|
|
try:
|
|
row = connection.execute(
|
|
"SELECT removed_at FROM pdd_tasks"
|
|
" WHERE remote_task_id = 'REMOVE-OK'"
|
|
).fetchone()
|
|
finally:
|
|
connection.close()
|
|
self.assertIsNotNone(row[0])
|
|
with self.assertRaises(DuplicateTaskError):
|
|
self.repository.add_claimed_task(self._task("REMOVE-OK"))
|
|
|
|
def test_batch_remove_is_all_or_nothing_for_non_terminal_task(self):
|
|
self.repository.add_claimed_task(self._task("REMOVE-ENDED"))
|
|
self.repository.add_claimed_task(self._task("REMOVE-RUNNING"))
|
|
self._set_task_status("REMOVE-ENDED", TaskStatus.FAILED)
|
|
self._set_task_status("REMOVE-RUNNING", TaskStatus.RUNNING)
|
|
|
|
with self.assertRaisesRegex(TaskRemovalError, "尚未结束"):
|
|
self.repository.remove_tasks_from_list(
|
|
("REMOVE-ENDED", "REMOVE-RUNNING")
|
|
)
|
|
|
|
self.assertEqual(self.repository.count_tasks(), 2)
|
|
|
|
def test_task_with_unsent_outbox_cannot_be_removed(self):
|
|
self.repository.add_claimed_task(self._task("REMOVE-UNSENT"))
|
|
started = self.repository.start_collect_run("REMOVE-UNSENT", "USB-001")
|
|
self.repository.save_collect_failure(
|
|
"REMOVE-UNSENT",
|
|
started.attempt_id,
|
|
TaskStatus.FAILED,
|
|
"TEST_FAILURE",
|
|
"测试失败",
|
|
False,
|
|
)
|
|
|
|
with self.assertRaisesRegex(TaskRemovalError, "未发送完成"):
|
|
self.repository.remove_tasks_from_list(("REMOVE-UNSENT",))
|
|
|
|
self.assertEqual(self.repository.count_tasks(), 1)
|
|
|
|
def test_irreversible_task_cannot_be_removed_even_when_terminal(self):
|
|
self.repository.add_claimed_task(
|
|
self._task(
|
|
"REMOVE-IRREVERSIBLE",
|
|
TaskType.PURCHASE,
|
|
execution_mode="live",
|
|
)
|
|
)
|
|
started = self.repository.start_purchase_run(
|
|
"REMOVE-IRREVERSIBLE", "USB-001"
|
|
)
|
|
self.repository.mark_purchase_irreversible(
|
|
"REMOVE-IRREVERSIBLE", started.attempt_id
|
|
)
|
|
self._set_task_status("REMOVE-IRREVERSIBLE", TaskStatus.SUCCEEDED)
|
|
|
|
with self.assertRaisesRegex(TaskRemovalError, "不可逆阶段"):
|
|
self.repository.remove_tasks_from_list(("REMOVE-IRREVERSIBLE",))
|
|
|
|
self.assertEqual(self.repository.count_tasks(), 1)
|
|
|
|
def test_sent_outbox_does_not_block_soft_remove(self):
|
|
self.repository.add_claimed_task(self._task("REMOVE-SENT"))
|
|
started = self.repository.start_collect_run("REMOVE-SENT", "USB-001")
|
|
event = self.repository.save_collect_result(
|
|
"REMOVE-SENT",
|
|
started.attempt_id,
|
|
{"goods_id": "10001", "title": "测试商品", "skus": []},
|
|
)
|
|
self.repository.mark_outbox_sent(event.id)
|
|
|
|
self.assertEqual(
|
|
self.repository.remove_tasks_from_list(("REMOVE-SENT",)), 1
|
|
)
|
|
self.assertEqual(self.repository.count_tasks(), 0)
|
|
connection = open_database(self.db_path)
|
|
try:
|
|
run_count = connection.execute(
|
|
"SELECT COUNT(*) FROM task_runs r JOIN pdd_tasks t"
|
|
" ON t.id = r.task_id WHERE t.remote_task_id = 'REMOVE-SENT'"
|
|
).fetchone()[0]
|
|
outbox_row = connection.execute(
|
|
"SELECT o.status FROM outbox_events o JOIN pdd_tasks t"
|
|
" ON t.id = o.task_id WHERE t.remote_task_id = 'REMOVE-SENT'"
|
|
).fetchone()
|
|
finally:
|
|
connection.close()
|
|
self.assertEqual(run_count, 1)
|
|
self.assertEqual(outbox_row[0], OutboxStatus.SENT.value)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|