"""统一任务分派器测试;不连接真实 Admin 或手机。""" import tempfile import unittest from pathlib import Path from src.admin_gateway import AdminTask, ClientInfo from src.mock_admin_gateway import MockAdminGateway from src.pdd_purchase_adapter import 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 def purchase_task(task_id="PUR-001"): return AdminTask( task_id=task_id, task_type=TaskType.PURCHASE, version=1, priority=10, 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): 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, ) 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_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_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_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()