"""采购任务演练应用服务测试；全部使用 Mock，不连接真实手机。""" import tempfile import unittest from pathlib import Path from src.admin_gateway import ( AdminTask, AndroidDeviceInfo, ClaimCapabilities, ClientInfo, ) from src.mock_admin_gateway import MockAdminGateway from src.pdd_purchase_adapter import ( PddLivePurchaseAdapter, PddPurchaseAdapter, PddPurchaseError, PurchasePageState, ) from src.purchase_task_service import PurchaseTaskService from src.task_models import NewClaimedTask, TaskStatus, TaskType from src.task_repository import TaskRepository OPTIONS = {"color": "黑色", "size": "L", "bundle": "标准版"} class RecordingDryRunAdapter(PddPurchaseAdapter): """只记录调用的演练适配器，不提供提交订单能力。""" def __init__( self, *, goods_id: str = "737116531267", price_cent: int = 4200, candidate_count: int = 1, forced_page: str = "", wrong_options: bool = False, in_stock: bool = True, submit_candidate_count: int = 1, ) -> None: self.goods_id = goods_id self.price_cent = price_cent self.candidate_count = candidate_count self.forced_page = forced_page self.wrong_options = wrong_options self.in_stock = in_stock self.submit_candidate_count = submit_candidate_count self.options = {} self.quantity = 0 self.page_kind = "goods" self.calls = [] def open_goods(self, goods_url: str) -> None: self.calls.append(("open_goods", goods_url)) def read_state(self) -> PurchasePageState: self.calls.append(("read_state",)) options = self.options if self.wrong_options and options: options = {**options, "size": "XL"} return PurchasePageState( page_kind=self.forced_page or self.page_kind, goods_id=self.goods_id, selected_options=dict(options), quantity=self.quantity, price_cent=self.price_cent, candidate_count=self.candidate_count, in_stock=self.in_stock, submit_candidate_count=self.submit_candidate_count, ) def select_options(self, options) -> None: self.calls.append(("select_options", dict(options))) self.options = dict(options) def set_quantity(self, quantity: int) -> None: self.calls.append(("set_quantity", quantity)) self.quantity = quantity def enter_confirmation(self) -> None: self.calls.append(("enter_confirmation",)) self.page_kind = "order_confirmation" def stop_before_submit(self) -> None: self.calls.append(("stop_before_submit",)) def close(self) -> None: self.calls.append(("close",)) class RecordingLiveAdapter(RecordingDryRunAdapter, PddLivePurchaseAdapter): """只记录一次提交调用的 live 测试 Adapter。""" def __init__( self, *, submit_error: bool = False, address_error: bool = False, **kwargs, ) -> None: super().__init__(**kwargs) self.submit_error = submit_error self.address_error = address_error self.submit_count = 0 def update_shipping_address(self, purchase_number: str) -> None: self.calls.append(("update_shipping_address", purchase_number)) if self.address_error: raise PddPurchaseError( "PURCHASE_ADDRESS_PAGE_TIMEOUT", "地址页面切换超时", step="purchase_update_shipping_address", ) def submit_order_once(self) -> None: self.submit_count += 1 self.calls.append(("submit_order_once",)) if self.submit_error: raise PddPurchaseError( "DEVICE_DISCONNECTED", "点击后设备断开", step="purchase_submit_once", ) class UnavailableGoodsAdapter(RecordingDryRunAdapter): def open_goods(self, goods_url: str) -> None: self.calls.append(("open_goods", goods_url)) raise PddPurchaseError( "PDD_GOODS_UNAVAILABLE", "商品链接已失效，PDD 无法打开商品详情页并返回了首页", step="purchase_open_goods", retryable=False, ) class PurchaseTaskServiceTest(unittest.TestCase): def setUp(self) -> None: self.temp_dir = tempfile.TemporaryDirectory() self.repository = TaskRepository( Path(self.temp_dir.name) / "client.db" ) self.gateway = MockAdminGateway() self.client = ClientInfo("CLIENT-001", "测试客户端") def tearDown(self) -> None: self.temp_dir.cleanup() def _prepare_task( self, *, task_id: str = "PUR-001", execution_mode: str = "dry_run", ) -> None: task = AdminTask( task_id=task_id, task_type=TaskType.PURCHASE, version=1, priority=10, execution_mode=execution_mode, payload={ "goods_url": ( "https://mobile.yangkeduo.com/goods.html?" "goods_id=737116531267" ), "goods_id": "737116531267", "options": dict(OPTIONS), "quantity": 2, "max_price_cent": 5000, }, ) self.gateway.enqueue_task(task, self.client.client_id) claimed = self.gateway.claim_next( self.client, ClaimCapabilities( device=AndroidDeviceInfo("USB-001"), supported_types=(TaskType.PURCHASE,), purchase_mode=( "live" if execution_mode == "live" else "dry_run" ), ), ) assert claimed is not None self.repository.add_claimed_task( NewClaimedTask( remote_task_id=claimed.task_id, task_type=claimed.task_type, goods_url=str(claimed.payload["goods_url"]), execution_mode=claimed.execution_mode, goods_id=str(claimed.payload["goods_id"]), quantity=int(claimed.payload["quantity"]), priority=claimed.priority, version=claimed.version, admin_payload={ "id": claimed.task_id, "type": claimed.task_type.value, "version": claimed.version, "priority": claimed.priority, "payload": dict(claimed.payload), }, ) ) def _service(self, adapter: RecordingDryRunAdapter): return PurchaseTaskService( self.gateway, self.repository, self.client, "USB-001", lambda _address, _cancelled: adapter, ) def test_dynamic_options_dry_run_stops_before_order_submission(self): self._prepare_task() adapter = RecordingDryRunAdapter() outcome = self._service(adapter).execute_one_local() self.assertEqual(outcome.kind, "succeeded") self.assertIn("没有提交订单", outcome.message) self.assertIn(("select_options", OPTIONS), adapter.calls) self.assertIn(("set_quantity", 2), adapter.calls) self.assertIn(("enter_confirmation",), adapter.calls) self.assertIn(("stop_before_submit",), adapter.calls) detail = self.repository.get_task("PUR-001") assert detail is not None self.assertEqual(detail.status, TaskStatus.SUCCEEDED) purchase = detail.pdd_data["purchase"] self.assertEqual(purchase["mode"], "dry_run") self.assertEqual(purchase["requested"]["options"], OPTIONS) self.assertEqual(purchase["confirmed"]["options"], OPTIONS) self.assertFalse(purchase["order_submitted"]) self.assertFalse(purchase["payment_attempted"]) def test_started_callback_runs_after_database_enters_running(self): self._prepare_task(task_id="PUR-STARTED") observed = [] service = PurchaseTaskService( self.gateway, self.repository, self.client, "USB-001", lambda _address, _cancelled: RecordingDryRunAdapter(), started=lambda task_id: observed.append( (task_id, self.repository.get_task(task_id).status) ), ) service.execute_selected("PUR-STARTED") self.assertEqual(observed, [("PUR-STARTED", TaskStatus.RUNNING)]) def test_price_above_limit_stops_before_confirmation(self): self._prepare_task() adapter = RecordingDryRunAdapter(price_cent=5001) outcome = self._service(adapter).execute_one_local() self.assertEqual(outcome.kind, "task_failed") self.assertNotIn(("enter_confirmation",), adapter.calls) detail = self.repository.get_task("PUR-001") assert detail is not None self.assertEqual(detail.status, TaskStatus.FAILED) self.assertEqual(detail.last_error_code, "PURCHASE_PRICE_EXCEEDED") def test_unavailable_goods_stops_before_specs_and_is_not_retryable(self): self._prepare_task() adapter = UnavailableGoodsAdapter() outcome = self._service(adapter).execute_one_local() self.assertEqual("task_failed", outcome.kind) self.assertNotIn(("select_options", OPTIONS), adapter.calls) detail = self.repository.get_task("PUR-001") self.assertEqual(TaskStatus.FAILED, detail.status) self.assertEqual("PDD_GOODS_UNAVAILABLE", detail.last_error_code) def test_result_submit_timeout_does_not_run_adapter_twice(self): self._prepare_task() adapter = RecordingDryRunAdapter() self.gateway.timeout_next_call() first = self._service(adapter).execute_one_local() first_call_count = len(adapter.calls) second = self._service(adapter).execute_one_local() self.assertEqual(first.kind, "result_pending") self.assertEqual(second.kind, "succeeded") self.assertEqual(len(adapter.calls), first_call_count) self.assertEqual(self.gateway.submission_count, 1) def test_invalid_pages_and_ambiguous_target_fail_safely(self): cases = ( ("captcha", 1, "PDD_PAGE_CAPTCHA", "global_failed"), ("login_required", 1, "PDD_PAGE_LOGIN_REQUIRED", "global_failed"), ("unknown", 1, "PDD_PAGE_UNKNOWN", "global_failed"), ("", 2, "PURCHASE_AMBIGUOUS_TARGET", "task_failed"), ) for index, (page, candidates, expected_code, expected_kind) in enumerate(cases): with self.subTest(page=page, candidates=candidates): task_id = f"PUR-BLOCK-{index}" self._prepare_task(task_id=task_id) adapter = RecordingDryRunAdapter( forced_page=page, candidate_count=candidates, ) outcome = self._service(adapter).execute_selected(task_id) self.assertEqual(outcome.kind, expected_kind) self.assertNotIn(("enter_confirmation",), adapter.calls) detail = self.repository.get_task(task_id) assert detail is not None self.assertEqual(detail.last_error_code, expected_code) def test_wrong_goods_or_options_never_reaches_confirmation(self): cases = ( ({"goods_id": "OTHER"}, "PURCHASE_GOODS_MISMATCH"), ({"wrong_options": True}, "PURCHASE_OPTIONS_MISMATCH"), ) for index, (kwargs, expected_code) in enumerate(cases): with self.subTest(expected_code=expected_code): task_id = f"PUR-MISMATCH-{index}" self._prepare_task(task_id=task_id) adapter = RecordingDryRunAdapter(**kwargs) self._service(adapter).execute_selected(task_id) self.assertNotIn(("enter_confirmation",), adapter.calls) detail = self.repository.get_task(task_id) assert detail is not None self.assertEqual(detail.last_error_code, expected_code) def test_live_marks_irreversible_before_single_submit_and_enters_reconcile(self): self._prepare_task(execution_mode="live") adapter = RecordingLiveAdapter() outcome = self._service(adapter).execute_one_local() self.assertEqual(outcome.kind, "reconcile_pending") self.assertEqual(adapter.submit_count, 1) self.assertIn(("update_shipping_address", "PUR-001"), adapter.calls) detail = self.repository.get_task("PUR-001") run = self.repository.latest_task_run("PUR-001") self.assertEqual(detail.status, TaskStatus.MANUAL_REVIEW) self.assertEqual(detail.current_step, "reconcile_purchase") self.assertIsNotNone(run.irreversible_action_at) self.assertIsNotNone(run.order_submitted_at) self.assertEqual(run.current_step, "reconcile_purchase") self.assertIsNone(self.repository.next_pending_outbox()) def test_live_ambiguous_submit_target_stops_before_irreversible(self): self._prepare_task(execution_mode="live") adapter = RecordingLiveAdapter(submit_candidate_count=2) outcome = self._service(adapter).execute_one_local() self.assertEqual(outcome.kind, "task_failed") self.assertEqual(adapter.submit_count, 0) run = self.repository.latest_task_run("PUR-001") self.assertIsNone(run.irreversible_action_at) def test_live_address_failure_stops_before_irreversible(self): self._prepare_task(execution_mode="live") adapter = RecordingLiveAdapter(address_error=True) outcome = self._service(adapter).execute_one_local() self.assertEqual(outcome.kind, "task_failed") self.assertEqual(adapter.submit_count, 0) run = self.repository.latest_task_run("PUR-001") self.assertIsNone(run.irreversible_action_at) detail = self.repository.get_task("PUR-001") self.assertEqual( detail.last_error_code, "PURCHASE_ADDRESS_PAGE_TIMEOUT" ) def test_live_submit_error_never_retries_and_still_enters_reconcile(self): self._prepare_task(execution_mode="live") adapter = RecordingLiveAdapter(submit_error=True) outcome = self._service(adapter).execute_one_local() self.assertEqual(outcome.kind, "reconcile_pending") self.assertEqual(adapter.submit_count, 1) run = self.repository.latest_task_run("PUR-001") self.assertIsNotNone(run.irreversible_action_at) self.assertIsNone(run.order_submitted_at) self.assertEqual(run.current_step, "reconcile_purchase") self.assertIsNone(self.repository.next_pending_outbox()) def test_live_task_rejects_dry_adapter_before_device_action(self): self._prepare_task(execution_mode="live") adapter = RecordingDryRunAdapter() outcome = self._service(adapter).execute_one_local() self.assertEqual(outcome.kind, "task_failed") self.assertEqual(adapter.calls, [("close",)]) run = self.repository.latest_task_run("PUR-001") self.assertIsNone(run.irreversible_action_at) def test_purchase_task_with_missing_safety_fields_is_reported(self): task_id = "PUR-INVALID" task = AdminTask( task_id=task_id, task_type=TaskType.PURCHASE, version=1, priority=0, payload={"goods_url": "https://example.test/goods"}, ) self.gateway.enqueue_task(task, self.client.client_id) self.gateway.claim_next( self.client, ClaimCapabilities(supported_types=(TaskType.PURCHASE,)), ) self.repository.add_claimed_task( NewClaimedTask( remote_task_id=task_id, task_type=TaskType.PURCHASE, goods_url="https://example.test/goods", admin_payload={ "id": task_id, "type": "purchase", "version": 1, "payload": dict(task.payload), }, ) ) adapter = RecordingDryRunAdapter() outcome = self._service(adapter).execute_selected(task_id) self.assertEqual(outcome.kind, "task_failed") self.assertNotIn(("open_goods", "https://example.test/goods"), adapter.calls) detail = self.repository.get_task(task_id) assert detail is not None self.assertEqual(detail.last_error_code, "PURCHASE_TASK_INVALID") if __name__ == "__main__": unittest.main()