"""统一任务分派器测试；不连接真实 Admin 或手机。""" import tempfile import unittest from pathlib import Path from src.android_device_service import AndroidDeviceSearchError from src.admin_gateway import AdminTask, ClientInfo, SubmissionReceipt from src.mock_admin_gateway import MockAdminGateway from src.pdd_purchase_adapter import ( PddLivePurchaseAdapter, PddPurchaseAdapter, PurchasePageState, ) from src.task_dispatcher import TaskDispatcher, admin_task_to_new_claimed_task from src.task_models import TaskStatus, TaskType from src.task_repository import TaskRepository class FakeCollectResult: def to_pdd_data(self): return { "schema_version": 1, "goods_id": "COL-GOODS", "title": "采集测试商品", "price_granularity": "color", "dimensions": [], "skus": [], } class RecordingCollector: def __init__(self, calls): self.calls = calls def collect(self, task): self.calls.append(("collect", task.remote_task_id)) return FakeCollectResult() class ReadyPurchaseAdapter(PddPurchaseAdapter): def __init__(self, calls): self.calls = calls self.options = {} self.quantity = 0 self.page_kind = "goods" def open_goods(self, goods_url): self.calls.append(("purchase", "open", goods_url)) def read_state(self): return PurchasePageState( self.page_kind, "PUR-GOODS", dict(self.options), self.quantity, 990, 1, ) def select_options(self, options): self.options = dict(options) def set_quantity(self, quantity): self.quantity = quantity def enter_confirmation(self): self.page_kind = "order_confirmation" def stop_before_submit(self): self.calls.append(("purchase", "stopped")) def close(self): pass class ReadyLivePurchaseAdapter(ReadyPurchaseAdapter, PddLivePurchaseAdapter): def read_state(self): state = super().read_state() return PurchasePageState( page_kind=state.page_kind, goods_id=state.goods_id, selected_options=state.selected_options, quantity=state.quantity, price_cent=state.price_cent, candidate_count=state.candidate_count, in_stock=True, submit_candidate_count=1, ) def submit_order_once(self): self.calls.append(("purchase", "submitted_once")) def purchase_task(task_id="PUR-001", execution_mode="dry_run"): return AdminTask( task_id=task_id, task_type=TaskType.PURCHASE, version=1, priority=10, execution_mode=execution_mode, payload={ "goods_url": "https://example.test/PUR-GOODS", "goods_id": "PUR-GOODS", "options": { "color": "黑色", "size": "L", "bundle": "标准版", }, "quantity": 2, "max_price_cent": 1200, }, ) class TaskDispatcherTest(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.repository = TaskRepository( Path(self.temporary.name) / "client.db" ) self.gateway = MockAdminGateway() self.client = ClientInfo("CLIENT-001", "测试客户端") self.calls = [] def tearDown(self): self.temporary.cleanup() def _dispatcher( self, *, purchase_ready, live_ready=False, purchase_mode="dry_run", device_checker=lambda _serial: None, task_saved=lambda _task_id: None, ): purchase_factory = None if purchase_ready: purchase_factory = ( lambda _address, _cancelled: ReadyPurchaseAdapter(self.calls) ) return TaskDispatcher( self.gateway, self.repository, self.client, "USB-001", collect_service_factory=( lambda *_args: RecordingCollector(self.calls) ), purchase_adapter_factory=purchase_factory, live_purchase_adapter_factory=( ( lambda _address, _cancelled: ReadyLivePurchaseAdapter( self.calls ) ) if live_ready else None ), purchase_mode=purchase_mode, device_connection_checker=device_checker, task_saved=task_saved, ) def test_capability_only_includes_purchase_when_adapter_is_ready(self): collect_only = self._dispatcher(purchase_ready=False) ready = self._dispatcher(purchase_ready=True) self.assertEqual( collect_only.claim_capabilities().supported_types, (TaskType.COLLECT,), ) self.assertEqual( ready.claim_capabilities().supported_types, (TaskType.COLLECT, TaskType.PURCHASE), ) self.assertEqual( ready.claim_capabilities().purchase_mode, "dry_run" ) def test_live_capability_requires_authorization_and_live_factory(self): no_factory = self._dispatcher( purchase_ready=True, purchase_mode="live" ) authorized = self._dispatcher( purchase_ready=True, live_ready=True, purchase_mode="live", ) self.assertEqual( no_factory.claim_capabilities().purchase_mode, "dry_run" ) self.assertEqual( authorized.claim_capabilities().purchase_mode, "live" ) def test_local_live_task_never_runs_after_authorization_is_disabled(self): task = purchase_task(execution_mode="live") self.repository.add_claimed_task(admin_task_to_new_claimed_task(task)) with self.assertRaisesRegex(RuntimeError, "授权已关闭"): self._dispatcher( purchase_ready=True, live_ready=True, purchase_mode="dry_run", ).execute_one() self.assertNotIn(("purchase", "submitted_once"), self.calls) self.assertEqual( self.repository.get_task(task.task_id).status, TaskStatus.CLAIMED, ) def test_live_task_is_claimed_only_by_authorized_live_dispatcher(self): task = purchase_task(execution_mode="live") self.gateway.enqueue_task(task, self.client.client_id) dry_outcome = self._dispatcher(purchase_ready=True).execute_one() live_outcome = self._dispatcher( purchase_ready=True, live_ready=True, purchase_mode="live", ).execute_one() self.assertEqual(dry_outcome.kind, "no_task") self.assertEqual(live_outcome.kind, "manual_review") self.assertEqual( self.repository.get_task(task.task_id).execution_mode, "live", ) self.assertEqual( self.calls.count(("purchase", "submitted_once")), 1 ) def test_claims_saves_then_dispatches_purchase_dry_run(self): task = purchase_task() self.gateway.enqueue_task(task, self.client.client_id) outcome = self._dispatcher(purchase_ready=True).execute_one() self.assertEqual(outcome.kind, "succeeded") detail = self.repository.get_task(task.task_id) assert detail is not None self.assertEqual(detail.task_type, TaskType.PURCHASE) self.assertEqual(detail.status, TaskStatus.SUCCEEDED) self.assertEqual( detail.admin_payload["payload"]["options"]["bundle"], "标准版", ) self.assertIn(("purchase", "stopped"), self.calls) def test_new_task_notifies_after_save_and_before_execution(self): task = AdminTask( "COL-SAVED", TaskType.COLLECT, 1, 0, { "goods_url": "https://example.test/COL-SAVED", "goods_id": "COL-GOODS", }, ) self.gateway.enqueue_task(task, self.client.client_id) notifications = [] def record_saved(task_id): detail = self.repository.get_task(task_id) notifications.append( (task_id, detail.status if detail else None, list(self.calls)) ) outcome = self._dispatcher( purchase_ready=False, task_saved=record_saved ).execute_one() self.assertEqual(outcome.kind, "succeeded") self.assertEqual( notifications, [("COL-SAVED", TaskStatus.CLAIMED, [])], ) def test_purchase_is_not_claimed_when_runtime_is_not_ready(self): self.gateway.enqueue_task(purchase_task(), self.client.client_id) outcome = self._dispatcher(purchase_ready=False).execute_one() self.assertEqual(outcome.kind, "no_task") self.assertEqual(self.repository.count_tasks(), 0) def test_local_purchase_stops_dispatch_when_runtime_is_not_ready(self): task = purchase_task() self.repository.add_claimed_task( admin_task_to_new_claimed_task(task) ) with self.assertRaisesRegex(RuntimeError, "停止领取新任务"): self._dispatcher(purchase_ready=False).execute_one() detail = self.repository.get_task(task.task_id) assert detail is not None self.assertEqual(detail.status, TaskStatus.CLAIMED) def test_disconnected_device_does_not_claim_new_task(self): self.gateway.enqueue_task(purchase_task(), self.client.client_id) def disconnected(_serial): raise AndroidDeviceSearchError("USB Android 设备 USB-001 未连接") with self.assertRaisesRegex(AndroidDeviceSearchError, "未连接"): self._dispatcher( purchase_ready=True, device_checker=disconnected ).execute_one() self.assertEqual(self.repository.count_tasks(), 0) def test_pending_outbox_submits_without_connected_device(self): task = admin_task_to_new_claimed_task( AdminTask( "COL-OUTBOX", TaskType.COLLECT, 1, 0, {"goods_url": "https://example.test", "goods_id": "G"}, ) ) self.repository.add_claimed_task(task) started = self.repository.start_collect_run("COL-OUTBOX", "USB-001") self.repository.save_collect_result( "COL-OUTBOX", started.attempt_id, FakeCollectResult().to_pdd_data() ) checks = [] class AcceptGateway: def submit_result(self, *_args): return SubmissionReceipt( True, "RESULT-001", "2026-08-10T08:00:00Z" ) outcome = TaskDispatcher( AcceptGateway(), self.repository, self.client, "USB-001", purchase_adapter_factory=( lambda _address, _cancelled: ReadyPurchaseAdapter(self.calls) ), device_connection_checker=lambda serial: checks.append(serial), ).execute_one() self.assertEqual(outcome.kind, "succeeded") self.assertEqual(checks, []) def test_purchase_mapper_rejects_missing_or_invalid_safety_fields(self): invalid_payloads = ( {"goods_url": "https://example.test"}, { "goods_url": "https://example.test", "goods_id": "G", "options": {}, "quantity": 1, "max_price_cent": 100, }, { "goods_url": "https://example.test", "goods_id": "G", "options": {"size": "L"}, "quantity": 0, "max_price_cent": 100, }, { "goods_url": "https://example.test", "goods_id": "G", "options": {"size": "L"}, "quantity": 1, "max_price_cent": 0, }, ) for index, payload in enumerate(invalid_payloads): with self.subTest(index=index): task = AdminTask( f"PUR-BAD-{index}", TaskType.PURCHASE, 1, 0, payload, ) with self.assertRaises(ValueError): admin_task_to_new_claimed_task(task) if __name__ == "__main__": unittest.main()