290 lines
10 KiB
Python
290 lines
10 KiB
Python
"""采购崩溃恢复和安全门禁测试；不连接真实手机。"""
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from src.admin_gateway import AdminTask, ClaimCapabilities, ClientInfo
|
|
from src.db import open_database
|
|
from src.mock_admin_gateway import MockAdminGateway
|
|
from src.pdd_purchase_adapter import (
|
|
PddPurchaseAdapter,
|
|
PddPurchaseError,
|
|
PurchasePageState,
|
|
)
|
|
from src.pdd_purchase_reconcile_adapter import (
|
|
PddPurchaseReconcileAdapter,
|
|
PurchaseReconcileObservation,
|
|
)
|
|
from src.purchase_task_service import PurchaseTaskService
|
|
from src.task_dispatcher import TaskDispatcher, admin_task_to_new_claimed_task
|
|
from src.task_models import RunStatus, TaskStatus, TaskType
|
|
from src.task_repository import TaskRepository
|
|
|
|
|
|
OPTIONS = {"color": "黑色", "size": "L"}
|
|
|
|
|
|
def purchase_admin_task(task_id: str = "PUR-RECOVER") -> AdminTask:
|
|
return AdminTask(
|
|
task_id,
|
|
TaskType.PURCHASE,
|
|
1,
|
|
0,
|
|
{
|
|
"goods_url": "https://example.test/goods/737116531267",
|
|
"goods_id": "737116531267",
|
|
"options": dict(OPTIONS),
|
|
"quantity": 2,
|
|
"max_price_cent": 5000,
|
|
},
|
|
)
|
|
|
|
|
|
class FaultAdapter(PddPurchaseAdapter):
|
|
"""在指定动作抛错，用来模拟设备断开或进程崩溃。"""
|
|
|
|
def __init__(self, fail_action: str = "") -> None:
|
|
self.fail_action = fail_action
|
|
self.options = {}
|
|
self.quantity = 0
|
|
self.page_kind = "goods"
|
|
|
|
def _fail(self, action: str) -> None:
|
|
if self.fail_action == action:
|
|
raise PddPurchaseError(
|
|
"DEVICE_DISCONNECTED",
|
|
"Android 设备连接中断",
|
|
step=action,
|
|
retryable=True,
|
|
diagnostics={"action": action},
|
|
)
|
|
|
|
def open_goods(self, _goods_url: str) -> None:
|
|
self._fail("open_goods")
|
|
|
|
def read_state(self) -> PurchasePageState:
|
|
self._fail("read_state")
|
|
return PurchasePageState(
|
|
self.page_kind,
|
|
"737116531267",
|
|
dict(self.options),
|
|
self.quantity,
|
|
4200,
|
|
1,
|
|
)
|
|
|
|
def select_options(self, options) -> None:
|
|
self._fail("select_options")
|
|
self.options = dict(options)
|
|
|
|
def set_quantity(self, quantity: int) -> None:
|
|
self._fail("set_quantity")
|
|
self.quantity = quantity
|
|
|
|
def enter_confirmation(self) -> None:
|
|
self._fail("enter_confirmation")
|
|
self.page_kind = "order_confirmation"
|
|
|
|
def stop_before_submit(self) -> None:
|
|
self._fail("stop_before_submit")
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
|
|
class ReadOnlyReconcileAdapter(PddPurchaseReconcileAdapter):
|
|
def __init__(self, calls) -> None:
|
|
self.calls = calls
|
|
|
|
def read_order_match(self, query):
|
|
self.calls.append(("reconcile", query.goods_id))
|
|
return PurchaseReconcileObservation(
|
|
"matched", "ORDER-001", "2026-08-10T08:00:00Z"
|
|
)
|
|
|
|
def close(self) -> None:
|
|
self.calls.append(("reconcile_close",))
|
|
|
|
|
|
class PurchaseRecoveryTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.db_path = Path(self.temporary.name) / "client.db"
|
|
self.repository = TaskRepository(self.db_path)
|
|
self.gateway = MockAdminGateway()
|
|
self.client = ClientInfo("CLIENT-001")
|
|
|
|
def tearDown(self) -> None:
|
|
self.temporary.cleanup()
|
|
|
|
def _add(self, task_id: str = "PUR-RECOVER") -> None:
|
|
task = purchase_admin_task(task_id)
|
|
self.gateway.enqueue_task(task, self.client.client_id)
|
|
claimed = self.gateway.claim_next(
|
|
self.client,
|
|
ClaimCapabilities(supported_types=(TaskType.PURCHASE,)),
|
|
)
|
|
assert claimed is not None
|
|
self.repository.add_claimed_task(
|
|
admin_task_to_new_claimed_task(claimed)
|
|
)
|
|
|
|
def _service(self, adapter, *, cancelled=lambda: False):
|
|
return PurchaseTaskService(
|
|
self.gateway,
|
|
self.repository,
|
|
self.client,
|
|
"USB-001",
|
|
lambda _address, _cancelled: adapter,
|
|
cancelled=cancelled,
|
|
)
|
|
|
|
def _interrupt_after_irreversible(self, task_id: str) -> None:
|
|
self._add(task_id)
|
|
started = self.repository.start_purchase_run(task_id, "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-10T08:00:00Z", started.attempt_id),
|
|
)
|
|
finally:
|
|
connection.close()
|
|
self.repository.recover_interrupted_work()
|
|
|
|
def test_critical_action_failure_keeps_last_persisted_step(self):
|
|
cases = {
|
|
"open_goods": "purchase_open_goods",
|
|
"select_options": "purchase_select_options",
|
|
"set_quantity": "purchase_set_quantity",
|
|
"enter_confirmation": "purchase_enter_confirmation",
|
|
"stop_before_submit": "purchase_dry_run_stopped",
|
|
}
|
|
for index, (action, expected_step) in enumerate(cases.items()):
|
|
with self.subTest(action=action):
|
|
task_id = f"PUR-FAULT-{index}"
|
|
self._add(task_id)
|
|
|
|
self._service(FaultAdapter(action)).execute_selected(task_id)
|
|
|
|
run = self.repository.latest_task_run(task_id)
|
|
assert run is not None
|
|
self.assertEqual(run.current_step, expected_step)
|
|
self.assertEqual(run.error_code, "DEVICE_DISCONNECTED")
|
|
self.assertEqual(run.diagnostics_json["action"], action)
|
|
detail = self.repository.get_task(task_id)
|
|
assert detail is not None
|
|
self.assertEqual(detail.status, TaskStatus.MANUAL_REVIEW)
|
|
|
|
def test_stop_request_is_saved_before_any_device_action(self):
|
|
self._add("PUR-STOP")
|
|
checks = iter((False, True))
|
|
adapter = FaultAdapter()
|
|
|
|
outcome = self._service(
|
|
adapter, cancelled=lambda: next(checks, True)
|
|
).execute_selected("PUR-STOP")
|
|
|
|
self.assertEqual(outcome.kind, "failed")
|
|
detail = self.repository.get_task("PUR-STOP")
|
|
run = self.repository.latest_task_run("PUR-STOP")
|
|
assert detail is not None and run is not None
|
|
self.assertEqual(detail.status, TaskStatus.CANCELLED)
|
|
self.assertEqual(run.current_step, "purchase_open_goods")
|
|
self.assertEqual(run.run_status, RunStatus.CANCELLED)
|
|
|
|
def test_restart_before_irreversible_closes_old_run_then_retries(self):
|
|
self._add()
|
|
first = self.repository.start_purchase_run("PUR-RECOVER", "USB-001")
|
|
self.repository.update_purchase_step(
|
|
"PUR-RECOVER", first.attempt_id, "purchase_select_options"
|
|
)
|
|
|
|
self.repository.recover_interrupted_work()
|
|
|
|
old_run = self.repository.latest_task_run("PUR-RECOVER")
|
|
detail = self.repository.get_task("PUR-RECOVER")
|
|
assert old_run is not None and detail is not None
|
|
self.assertEqual(old_run.run_status, RunStatus.FAILED)
|
|
self.assertEqual(old_run.current_step, "purchase_select_options")
|
|
self.assertEqual(detail.status, TaskStatus.CLAIMED)
|
|
second = self.repository.start_purchase_run(
|
|
"PUR-RECOVER", "USB-001"
|
|
)
|
|
self.assertEqual(second.attempt_no, 2)
|
|
with self.assertRaisesRegex(ValueError, "不能开始演练"):
|
|
self.repository.start_purchase_run("PUR-RECOVER", "USB-001")
|
|
|
|
def test_irreversible_restart_only_reconciles_and_never_purchases(self):
|
|
self._interrupt_after_irreversible("PUR-RECOVER")
|
|
calls = []
|
|
|
|
def purchase_factory(_address, _cancelled):
|
|
calls.append(("purchase",))
|
|
return FaultAdapter()
|
|
|
|
dispatcher = TaskDispatcher(
|
|
self.gateway,
|
|
self.repository,
|
|
self.client,
|
|
"USB-001",
|
|
purchase_adapter_factory=purchase_factory,
|
|
purchase_reconcile_factory=(
|
|
lambda _address, _cancelled: ReadOnlyReconcileAdapter(calls)
|
|
),
|
|
)
|
|
|
|
first = dispatcher.execute_one()
|
|
second = dispatcher.execute_one()
|
|
|
|
self.assertEqual(first.kind, "manual_review")
|
|
self.assertEqual(second.kind, "no_task")
|
|
self.assertNotIn(("purchase",), calls)
|
|
self.assertEqual(calls.count(("reconcile", "737116531267")), 1)
|
|
detail = self.repository.get_task("PUR-RECOVER")
|
|
run = self.repository.latest_task_run("PUR-RECOVER")
|
|
assert detail is not None and run is not None
|
|
self.assertEqual(detail.current_step, "reconcile_completed")
|
|
self.assertEqual(run.diagnostics_json["mode"], "reconcile_only")
|
|
|
|
def test_reconcile_device_failure_is_recorded_as_unknown(self):
|
|
task_id = "PUR-RECONCILE-OFFLINE"
|
|
self._interrupt_after_irreversible(task_id)
|
|
purchase_calls = []
|
|
|
|
def unavailable_reconcile(_address, _cancelled):
|
|
raise ConnectionError("核对设备已断开")
|
|
|
|
outcome = TaskDispatcher(
|
|
self.gateway,
|
|
self.repository,
|
|
self.client,
|
|
"USB-001",
|
|
purchase_adapter_factory=(
|
|
lambda _address, _cancelled: purchase_calls.append("purchase")
|
|
),
|
|
purchase_reconcile_factory=unavailable_reconcile,
|
|
).execute_one()
|
|
|
|
self.assertEqual(outcome.kind, "manual_review")
|
|
self.assertEqual(purchase_calls, [])
|
|
detail = self.repository.get_task(task_id)
|
|
run = self.repository.latest_task_run(task_id)
|
|
assert detail is not None and run is not None
|
|
self.assertEqual(detail.current_step, "reconcile_manual_review")
|
|
self.assertIn("核对设备已断开", run.diagnostics_json["error"])
|
|
|
|
def test_live_mode_and_order_submission_methods_are_unavailable(self):
|
|
with self.assertRaisesRegex(ValueError, "dry_run"):
|
|
ClaimCapabilities(purchase_mode="live")
|
|
for name in ("submit_order", "pay", "payment"):
|
|
self.assertFalse(hasattr(PddPurchaseAdapter, name))
|
|
self.assertFalse(hasattr(PddPurchaseReconcileAdapter, name))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|