445 lines
16 KiB
Python
445 lines
16 KiB
Python
"""采购任务演练应用服务测试；全部使用 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_order_total_above_limit_stops_on_confirmation(self):
|
|
self._prepare_task()
|
|
adapter = RecordingDryRunAdapter(price_cent=5001)
|
|
|
|
outcome = self._service(adapter).execute_one_local()
|
|
|
|
self.assertEqual(outcome.kind, "task_failed")
|
|
self.assertIn(("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")
|
|
self.assertIn("订单总价", detail.last_error_message)
|
|
|
|
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()
|