"""采购崩溃恢复和安全门禁测试；不连接真实手机。""" import tempfile import unittest from dataclasses import replace from pathlib import Path from src.admin_gateway import AdminTask, ClaimCapabilities, ClientInfo from src.mock_admin_gateway import MockAdminGateway from src.pdd_purchase_adapter import ( PddLivePurchaseAdapter, PddPurchaseAdapter, PddPurchaseError, PurchasePageState, ) from src.pdd_purchase_reconcile_adapter import ( PddPurchaseReconcileAdapter, PurchaseOrderCandidate, PurchaseReconcileScan, ) 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", execution_mode: str = "dry_run" ) -> 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, }, execution_mode=execution_mode, ) 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, candidates=None, error=None) -> None: self.calls = calls self.candidates = candidates self.error = error def read_order_candidates(self, query): self.calls.append(("reconcile", query.goods_id)) if self.error is not None: raise self.error candidates = self.candidates if candidates is None: candidates = ( PurchaseOrderCandidate( order_no="ORDER-001", goods_id=query.goods_id, options=dict(query.options), quantity=query.quantity, total_price_cent=query.total_price_cent, ordered_at=query.irreversible_action_at, ordered_at_raw="2026-08-10 16:00:00", payment_status="unpaid", ), ) else: candidates = tuple( replace(candidate, ordered_at=query.irreversible_action_at) for candidate in candidates ) return PurchaseReconcileScan( candidates=candidates, diagnostics={"pages_scanned": 1}, ) 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", execution_mode: str = "dry_run" ) -> None: task = purchase_admin_task(task_id, execution_mode) self.gateway.enqueue_task(task, self.client.client_id) claimed = self.gateway.claim_next( self.client, ClaimCapabilities( supported_types=(TaskType.PURCHASE,), purchase_mode=( "live" if execution_mode == "live" else "dry_run" ), ), ) 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, "live") started = self.repository.start_purchase_run(task_id, "USB-001") self.repository.mark_purchase_irreversible( task_id, started.attempt_id, { "options": dict(OPTIONS), "quantity": 2, "unit_price_cent": 4200, "total_price_cent": 8400, }, ) 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) ), device_connection_checker=lambda _serial: None, ) first = dispatcher.execute_one() pending = self.repository.next_pending_outbox() matched_detail = self.repository.get_task("PUR-RECOVER") matched_run = self.repository.latest_task_run("PUR-RECOVER") duplicate = self.repository.save_matched_purchase_reconciliation( "PUR-RECOVER", matched_run.attempt_id, matched_detail.pdd_data, {"mode": "reconcile_only"}, ) self.assertEqual(duplicate.id, pending.id) second = dispatcher.execute_one() third = dispatcher.execute_one() self.assertEqual(first.kind, "result_pending") self.assertEqual(second.kind, "succeeded") self.assertEqual(third.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, "completed") self.assertEqual( run.diagnostics_json["reconciliation"]["mode"], "reconcile_only", ) self.assertEqual( detail.pdd_data["purchase"]["payment_status"], "unpaid" ) self.assertEqual(detail.pdd_data["purchase"]["order_no"], "ORDER-001") 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, device_connection_checker=lambda _serial: None, ).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["reconciliation"]["error_message"], ) def test_no_multiple_mismatched_and_paid_candidates_need_manual_review(self): cases = { "EMPTY": ((), "ORDER_NOT_FOUND"), "MULTIPLE": ( ( PurchaseOrderCandidate( "ORDER-A", "737116531267", OPTIONS, 2, 8400, "2026-08-10T08:00:00Z", "2026-08-10 16:00:00", "unpaid", ), PurchaseOrderCandidate( "ORDER-B", "737116531267", OPTIONS, 2, 8400, "2026-08-10T08:00:00Z", "2026-08-10 16:00:00", "unpaid", ), ), "AMBIGUOUS_ORDER_MATCH", ), "MISMATCH": ( ( PurchaseOrderCandidate( "ORDER-C", "OTHER-GOODS", OPTIONS, 2, 8400, "2026-08-10T08:00:00Z", "2026-08-10 16:00:00", "unpaid", ), ), "ORDER_MATCH_UNCERTAIN", ), "PAID": ( ( PurchaseOrderCandidate( "ORDER-D", "737116531267", OPTIONS, 2, 8400, "2026-08-10T08:00:00Z", "2026-08-10 16:00:00", "paid", ), ), "ORDER_MATCH_UNCERTAIN", ), } for suffix, (candidates, error_code) in cases.items(): with self.subTest(suffix=suffix): task_id = f"PUR-{suffix}" self._interrupt_after_irreversible(task_id) calls = [] outcome = TaskDispatcher( self.gateway, self.repository, self.client, "USB-001", purchase_adapter_factory=lambda *_args: calls.append( "purchase" ), purchase_reconcile_factory=( lambda _address, _cancelled, values=candidates: ReadOnlyReconcileAdapter(calls, values) ), device_connection_checker=lambda _serial: None, ).execute_one() self.assertEqual(outcome.kind, "manual_review") self.assertNotIn("purchase", calls) detail = self.repository.get_task(task_id) self.assertEqual(detail.status, TaskStatus.MANUAL_REVIEW) self.assertEqual(detail.last_error_code, error_code) self.assertIsNone(self.repository.next_pending_outbox()) def test_restart_after_match_only_submits_outbox_without_reading_phone(self): task_id = "PUR-OUTBOX-RESTART" self._interrupt_after_irreversible(task_id) first_calls = [] first = TaskDispatcher( self.gateway, self.repository, self.client, "USB-001", purchase_reconcile_factory=( lambda _address, _cancelled: ReadOnlyReconcileAdapter( first_calls ) ), device_connection_checker=lambda _serial: None, ).execute_one() self.assertEqual(first.kind, "result_pending") restarted_repository = TaskRepository(self.db_path) restarted_calls = [] second = TaskDispatcher( self.gateway, restarted_repository, self.client, "", purchase_adapter_factory=lambda *_args: restarted_calls.append( "purchase" ), purchase_reconcile_factory=lambda *_args: restarted_calls.append( "reconcile" ), ).execute_one() self.assertEqual(second.kind, "succeeded") self.assertEqual(restarted_calls, []) self.assertEqual( restarted_repository.get_task(task_id).status, TaskStatus.SUCCEEDED, ) def test_live_submit_is_isolated_and_payment_methods_are_unavailable(self): self.assertEqual(ClaimCapabilities(purchase_mode="live").purchase_mode, "live") self.assertFalse(hasattr(PddPurchaseAdapter, "submit_order_once")) self.assertTrue(hasattr(PddLivePurchaseAdapter, "submit_order_once")) for name in ("submit_order", "pay", "payment", "cancel_order"): self.assertFalse(hasattr(PddPurchaseAdapter, name)) self.assertFalse(hasattr(PddLivePurchaseAdapter, name)) self.assertFalse(hasattr(PddPurchaseReconcileAdapter, name)) if __name__ == "__main__": unittest.main()