From c412e8afe23a48df2c30229e437c0e6d7e568c86 Mon Sep 17 00:00:00 2001 From: chengma Date: Thu, 6 Aug 2026 16:38:28 +0800 Subject: [PATCH] feat(client): add Admin Gateway mock contract (#10) --- client/src/admin_gateway.py | 142 ++++++++++ client/src/mock_admin_gateway.py | 285 +++++++++++++++++++++ client/test/test_admin_gateway_contract.py | 220 ++++++++++++++++ 3 files changed, 647 insertions(+) create mode 100644 client/src/admin_gateway.py create mode 100644 client/src/mock_admin_gateway.py create mode 100644 client/test/test_admin_gateway_contract.py diff --git a/client/src/admin_gateway.py b/client/src/admin_gateway.py new file mode 100644 index 0000000..29086ac --- /dev/null +++ b/client/src/admin_gateway.py @@ -0,0 +1,142 @@ +"""Client 访问 Admin 的稳定边界和简单数据对象。 + +AdminGateway 只有领取任务、提交成功结果、提交失败结果三个业务方法。 +业务层不应直接依赖 HTTP 请求或 Mock 的内部实现。 +""" + +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from typing import Any, Mapping, Optional, Tuple + +from .task_models import TaskType + + +@dataclass(frozen=True) +class ClientInfo: + """发起领取请求的 Client 身份，不保存访问令牌。""" + + client_id: str + + def __post_init__(self) -> None: + if not self.client_id.strip(): + raise ValueError("client_id 不能为空") + + +@dataclass(frozen=True) +class AndroidDeviceInfo: + """领取任务时上报的 Android 设备信息。""" + + address: str + platform: str = "android" + pdd_package: str = "com.xunmeng.pinduoduo" + + def __post_init__(self) -> None: + if not self.address.strip(): + raise ValueError("设备地址不能为空") + if self.platform != "android": + raise ValueError("当前只支持 android 平台") + if not self.pdd_package.strip(): + raise ValueError("PDD 包名不能为空") + + +@dataclass(frozen=True) +class ClaimCapabilities: + """Client 领取任务时声明的设备与执行能力。""" + + device: AndroidDeviceInfo + supported_types: Tuple[TaskType, ...] = ( + TaskType.COLLECT, + TaskType.PURCHASE, + ) + purchase_mode: str = "dry_run" + schema_versions: Tuple[int, ...] = (1,) + + def __post_init__(self) -> None: + if not self.supported_types: + raise ValueError("supported_types 不能为空") + if any(not isinstance(value, TaskType) for value in self.supported_types): + raise ValueError("supported_types 必须使用 TaskType") + if self.purchase_mode not in {"dry_run", "live"}: + raise ValueError("purchase_mode 只能是 dry_run 或 live") + if not self.schema_versions or any( + version <= 0 for version in self.schema_versions + ): + raise ValueError("schema_versions 必须是正整数") + + +@dataclass(frozen=True) +class AdminTask: + """Admin 派发给 Client 的一个任务。""" + + task_id: str + task_type: TaskType + version: int + priority: int + payload: Mapping[str, Any] = field(default_factory=dict) + created_at: str = "" + updated_at: str = "" + + def __post_init__(self) -> None: + if not self.task_id.strip(): + raise ValueError("task_id 不能为空") + if not isinstance(self.task_type, TaskType): + raise ValueError("task_type 必须使用 TaskType") + if self.version <= 0: + raise ValueError("version 必须大于 0") + if not isinstance(self.payload, Mapping): + raise ValueError("payload 必须是对象") + + +@dataclass(frozen=True) +class SubmissionReceipt: + """Admin 已接收并保存一次提交的确认。""" + + accepted: bool + result_id: str + accepted_at: str + + +class AdminGatewayError(RuntimeError): + """带稳定错误代码和可重试标志的 Admin 边界错误。""" + + def __init__( + self, + code: str, + message: str, + retryable: bool, + request_id: str = "", + details: Optional[Mapping[str, Any]] = None, + ): + super().__init__(message) + self.code = code + self.retryable = retryable + self.request_id = request_id + self.details = dict(details or {}) + + +class AdminGateway(ABC): + """Admin 边界；不得增加状态查询、心跳或租约方法。""" + + @abstractmethod + def claim_next( + self, client: ClientInfo, capabilities: ClaimCapabilities + ) -> Optional[AdminTask]: + """领取至多一个分配给当前 Client 的任务。""" + + @abstractmethod + def submit_result( + self, + task_id: str, + idempotency_key: str, + result: Mapping[str, Any], + ) -> SubmissionReceipt: + """幂等提交采集或采购成功结果。""" + + @abstractmethod + def submit_failure( + self, + task_id: str, + idempotency_key: str, + failure: Mapping[str, Any], + ) -> SubmissionReceipt: + """幂等提交失败、取消或人工处理结果。""" diff --git a/client/src/mock_admin_gateway.py b/client/src/mock_admin_gateway.py new file mode 100644 index 0000000..ee4f47d --- /dev/null +++ b/client/src/mock_admin_gateway.py @@ -0,0 +1,285 @@ +"""不访问网络的 AdminGateway 测试实现。""" + +import hashlib +import json +from copy import deepcopy +from dataclasses import dataclass +from datetime import datetime, timezone +from threading import Lock +from typing import Any, Dict, List, Mapping, Optional, Tuple +from uuid import uuid4 + +from .admin_gateway import ( + AdminGateway, + AdminGatewayError, + AdminTask, + ClaimCapabilities, + ClientInfo, + SubmissionReceipt, +) + + +FAILURE_STATUSES = {"retry_wait", "manual_review", "failed", "cancelled"} + + +def utc_now_iso() -> str: + """返回精确到秒的 UTC 时间。""" + + return datetime.now(timezone.utc).isoformat(timespec="seconds").replace( + "+00:00", "Z" + ) + + +@dataclass +class _QueuedTask: + task: AdminTask + assigned_client_id: str + claimed: bool = False + cancelled: bool = False + + +class MockAdminGateway(AdminGateway): + """支持任务派发、故障模拟和幂等提交的内存 Mock。""" + + def __init__(self): + self._tasks: List[_QueuedTask] = [] + self._claimed_task_ids = set() + self._submissions: Dict[ + str, Tuple[str, SubmissionReceipt] + ] = {} + self._next_error: Optional[AdminGatewayError] = None + self._reject_next_submission = False + self._lock = Lock() + + def enqueue_task(self, task: AdminTask, assigned_client_id: str) -> None: + """测试辅助：加入一条分配给指定 Client 的任务。""" + + if not assigned_client_id.strip(): + raise ValueError("assigned_client_id 不能为空") + with self._lock: + if any(item.task.task_id == task.task_id for item in self._tasks): + raise ValueError(f"任务 {task.task_id} 已存在") + self._tasks.append( + _QueuedTask(deepcopy(task), assigned_client_id.strip()) + ) + + def cancel_task(self, task_id: str) -> None: + """测试辅助：模拟 Admin 在任务派发后取消任务。""" + + with self._lock: + item = self._find_task(task_id) + if item is None: + raise ValueError(f"任务 {task_id} 不存在") + item.cancelled = True + + def timeout_next_call(self) -> None: + """测试辅助：让下一次 Gateway 调用模拟网络超时。""" + + self._next_error = AdminGatewayError( + "ADMIN_TIMEOUT", "Admin 请求超时", True + ) + + def fail_next_call_temporarily(self) -> None: + """测试辅助：让下一次调用模拟 Admin 暂时故障。""" + + self._next_error = AdminGatewayError( + "ADMIN_UNAVAILABLE", "Admin 暂时不可用", True + ) + + def reject_next_submission(self) -> None: + """测试辅助：让下一次提交模拟结果校验失败。""" + + self._reject_next_submission = True + + @property + def submission_count(self) -> int: + """返回已接受的不同幂等提交数量。""" + + return len(self._submissions) + + def claim_next( + self, client: ClientInfo, capabilities: ClaimCapabilities + ) -> Optional[AdminTask]: + with self._lock: + self._raise_forced_error() + for item in self._tasks: + if item.claimed: + continue + if item.assigned_client_id != client.client_id: + continue + if item.task.task_type not in capabilities.supported_types: + continue + + item.claimed = True + self._claimed_task_ids.add(item.task.task_id) + return deepcopy(item.task) + return None + + def submit_result( + self, + task_id: str, + idempotency_key: str, + result: Mapping[str, Any], + ) -> SubmissionReceipt: + return self._submit("result", task_id, idempotency_key, result) + + def submit_failure( + self, + task_id: str, + idempotency_key: str, + failure: Mapping[str, Any], + ) -> SubmissionReceipt: + return self._submit("failure", task_id, idempotency_key, failure) + + def _submit( + self, + submission_type: str, + task_id: str, + idempotency_key: str, + payload: Mapping[str, Any], + ) -> SubmissionReceipt: + with self._lock: + self._raise_forced_error() + if self._reject_next_submission: + self._reject_next_submission = False + raise AdminGatewayError( + "ADMIN_RESULT_INVALID", "Admin 拒绝了无效结果", False + ) + + task = self._validate_submission_target(task_id, idempotency_key) + if submission_type == "result": + self._validate_result(task, payload) + else: + self._validate_failure(task, payload) + + fingerprint = self._fingerprint( + submission_type, task_id, payload + ) + previous = self._submissions.get(idempotency_key) + if previous is not None: + old_fingerprint, receipt = previous + if old_fingerprint != fingerprint: + raise AdminGatewayError( + "IDEMPOTENCY_CONFLICT", + "相同幂等键提交了不同内容", + False, + ) + return receipt + + receipt = SubmissionReceipt( + accepted=True, + result_id=str(uuid4()), + accepted_at=utc_now_iso(), + ) + self._submissions[idempotency_key] = (fingerprint, receipt) + return receipt + + def _validate_submission_target( + self, task_id: str, idempotency_key: str + ) -> AdminTask: + if not task_id.strip(): + raise AdminGatewayError( + "ADMIN_TASK_INVALID", "task_id 不能为空", False + ) + if not idempotency_key.strip(): + raise AdminGatewayError( + "ADMIN_IDEMPOTENCY_KEY_INVALID", "幂等键不能为空", False + ) + item = self._find_task(task_id) + if item is None or task_id not in self._claimed_task_ids: + raise AdminGatewayError( + "ADMIN_TASK_NOT_ASSIGNED", + "任务未派发给当前 Client", + False, + ) + # item.cancelled 故意不参与判断：已派发任务即使取消也必须接收。 + return item.task + + @staticmethod + def _validate_result(task: AdminTask, result: Mapping[str, Any]) -> None: + MockAdminGateway._require_mapping(result, "result") + MockAdminGateway._validate_common_submission(task, result) + if result.get("result_type") != task.task_type.value: + raise AdminGatewayError( + "ADMIN_RESULT_INVALID", "result_type 与任务类型不一致", False + ) + if not str(result.get("completed_at", "")).strip(): + raise AdminGatewayError( + "ADMIN_RESULT_INVALID", "completed_at 不能为空", False + ) + if not isinstance(result.get("pdd_data"), Mapping): + raise AdminGatewayError( + "ADMIN_RESULT_INVALID", "pdd_data 必须是对象", False + ) + + @staticmethod + def _validate_failure(task: AdminTask, failure: Mapping[str, Any]) -> None: + MockAdminGateway._require_mapping(failure, "failure") + MockAdminGateway._validate_common_submission(task, failure) + if failure.get("status") not in FAILURE_STATUSES: + raise AdminGatewayError( + "ADMIN_RESULT_INVALID", "失败状态无效", False + ) + error = failure.get("error") + if not isinstance(error, Mapping) or not str(error.get("code", "")).strip(): + raise AdminGatewayError( + "ADMIN_RESULT_INVALID", "error.code 不能为空", False + ) + if not str(failure.get("reported_at", "")).strip(): + raise AdminGatewayError( + "ADMIN_RESULT_INVALID", "reported_at 不能为空", False + ) + + @staticmethod + def _validate_common_submission( + task: AdminTask, payload: Mapping[str, Any] + ) -> None: + version = payload.get("task_version") + if isinstance(version, bool) or version != task.version: + raise AdminGatewayError( + "ADMIN_RESULT_INVALID", "task_version 与任务不一致", False + ) + if not str(payload.get("attempt_id", "")).strip(): + raise AdminGatewayError( + "ADMIN_RESULT_INVALID", "attempt_id 不能为空", False + ) + + @staticmethod + def _require_mapping(value: Mapping[str, Any], name: str) -> None: + if not isinstance(value, Mapping): + raise AdminGatewayError( + "ADMIN_RESULT_INVALID", f"{name} 必须是对象", False + ) + + @staticmethod + def _fingerprint( + submission_type: str, task_id: str, payload: Mapping[str, Any] + ) -> str: + try: + canonical = json.dumps( + { + "submission_type": submission_type, + "task_id": task_id, + "payload": payload, + }, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + except (TypeError, ValueError) as exc: + raise AdminGatewayError( + "ADMIN_RESULT_INVALID", "提交内容不是有效 JSON", False + ) from exc + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + def _raise_forced_error(self) -> None: + error = self._next_error + self._next_error = None + if error is not None: + raise error + + def _find_task(self, task_id: str) -> Optional[_QueuedTask]: + for item in self._tasks: + if item.task.task_id == task_id: + return item + return None diff --git a/client/test/test_admin_gateway_contract.py b/client/test/test_admin_gateway_contract.py new file mode 100644 index 0000000..8c91341 --- /dev/null +++ b/client/test/test_admin_gateway_contract.py @@ -0,0 +1,220 @@ +"""AdminGateway 边界和 Mock 契约测试。""" + +import unittest + +from src.admin_gateway import ( + AdminGateway, + AdminGatewayError, + AdminTask, + AndroidDeviceInfo, + ClaimCapabilities, + ClientInfo, +) +from src.mock_admin_gateway import MockAdminGateway +from src.task_models import TaskType + + +class MockAdminGatewayContractTest(unittest.TestCase): + def setUp(self): + self.gateway = MockAdminGateway() + self.client = ClientInfo("client-001") + self.all_capabilities = ClaimCapabilities( + device=AndroidDeviceInfo("192.168.0.173:5555") + ) + + @staticmethod + def _task(task_id: str, task_type: TaskType) -> AdminTask: + return AdminTask( + task_id=task_id, + task_type=task_type, + version=3, + priority=10, + payload={"goods_id": "737116531267"}, + created_at="2026-08-06T07:00:00Z", + updated_at="2026-08-06T07:05:00Z", + ) + + @staticmethod + def _result(result_type: str = "collect"): + return { + "task_version": 3, + "attempt_id": "attempt-001", + "result_type": result_type, + "completed_at": "2026-08-06T08:03:00Z", + "pdd_data": {"title": "测试商品"}, + } + + @staticmethod + def _failure(): + return { + "task_version": 3, + "attempt_id": "attempt-001", + "status": "manual_review", + "error": { + "code": "AMBIGUOUS_ORDER_MATCH", + "message": "发现多个候选订单", + "retryable": False, + "step": "reconcile_order", + }, + "diagnostics": {"artifact_ids": ["artifact-001"]}, + "reported_at": "2026-08-06T08:03:00Z", + } + + def test_gateway_has_only_three_business_methods(self): + self.assertEqual( + AdminGateway.__abstractmethods__, + {"claim_next", "submit_result", "submit_failure"}, + ) + for forbidden in ("get_status", "heartbeat", "renew_lease"): + self.assertFalse(hasattr(AdminGateway, forbidden)) + + def test_claim_returns_none_when_no_task_exists(self): + self.assertIsNone( + self.gateway.claim_next(self.client, self.all_capabilities) + ) + + def test_claim_respects_client_and_supported_types(self): + self.gateway.enqueue_task( + self._task("COLLECT-001", TaskType.COLLECT), "client-001" + ) + self.gateway.enqueue_task( + self._task("PURCHASE-001", TaskType.PURCHASE), "client-001" + ) + self.gateway.enqueue_task( + self._task("COLLECT-OTHER", TaskType.COLLECT), "client-002" + ) + collect_only = ClaimCapabilities( + device=AndroidDeviceInfo("emulator-5554"), + supported_types=(TaskType.COLLECT,), + ) + + first = self.gateway.claim_next(self.client, collect_only) + + self.assertEqual(first.task_id, "COLLECT-001") + self.assertIsNone(self.gateway.claim_next(self.client, collect_only)) + + second = self.gateway.claim_next(self.client, self.all_capabilities) + self.assertEqual(second.task_id, "PURCHASE-001") + self.assertIsNone( + self.gateway.claim_next(self.client, self.all_capabilities) + ) + + def test_claimed_task_is_not_returned_twice(self): + self.gateway.enqueue_task( + self._task("TASK-001", TaskType.COLLECT), "client-001" + ) + + self.assertIsNotNone( + self.gateway.claim_next(self.client, self.all_capabilities) + ) + self.assertIsNone( + self.gateway.claim_next(self.client, self.all_capabilities) + ) + + def test_can_simulate_timeout_and_temporary_failure_once(self): + self.gateway.timeout_next_call() + with self.assertRaises(AdminGatewayError) as timeout_context: + self.gateway.claim_next(self.client, self.all_capabilities) + self.assertEqual(timeout_context.exception.code, "ADMIN_TIMEOUT") + self.assertTrue(timeout_context.exception.retryable) + self.assertIsNone( + self.gateway.claim_next(self.client, self.all_capabilities) + ) + + self.gateway.fail_next_call_temporarily() + with self.assertRaises(AdminGatewayError) as unavailable_context: + self.gateway.claim_next(self.client, self.all_capabilities) + self.assertEqual( + unavailable_context.exception.code, "ADMIN_UNAVAILABLE" + ) + self.assertTrue(unavailable_context.exception.retryable) + + def test_same_idempotency_key_and_content_reuses_receipt(self): + self.gateway.enqueue_task( + self._task("TASK-001", TaskType.COLLECT), "client-001" + ) + self.gateway.claim_next(self.client, self.all_capabilities) + result = self._result() + + first = self.gateway.submit_result("TASK-001", "stable-key", result) + second = self.gateway.submit_result("TASK-001", "stable-key", result) + + self.assertTrue(first.accepted) + self.assertEqual(first, second) + self.assertEqual(self.gateway.submission_count, 1) + + def test_same_idempotency_key_with_different_content_conflicts(self): + self.gateway.enqueue_task( + self._task("TASK-001", TaskType.COLLECT), "client-001" + ) + self.gateway.claim_next(self.client, self.all_capabilities) + self.gateway.submit_result("TASK-001", "stable-key", self._result()) + changed = self._result() + changed["pdd_data"] = {"title": "另一个商品"} + + with self.assertRaises(AdminGatewayError) as context: + self.gateway.submit_result("TASK-001", "stable-key", changed) + + self.assertEqual(context.exception.code, "IDEMPOTENCY_CONFLICT") + self.assertFalse(context.exception.retryable) + self.assertEqual(self.gateway.submission_count, 1) + + def test_cancelled_task_still_accepts_result_and_failure(self): + self.gateway.enqueue_task( + self._task("TASK-001", TaskType.COLLECT), "client-001" + ) + self.gateway.claim_next(self.client, self.all_capabilities) + self.gateway.cancel_task("TASK-001") + + result_receipt = self.gateway.submit_result( + "TASK-001", "result-key", self._result() + ) + failure_receipt = self.gateway.submit_failure( + "TASK-001", "failure-key", self._failure() + ) + + self.assertTrue(result_receipt.accepted) + self.assertTrue(failure_receipt.accepted) + self.assertEqual(self.gateway.submission_count, 2) + + def test_unclaimed_task_submission_is_rejected(self): + self.gateway.enqueue_task( + self._task("TASK-001", TaskType.COLLECT), "client-001" + ) + + with self.assertRaises(AdminGatewayError) as context: + self.gateway.submit_result("TASK-001", "result-key", self._result()) + + self.assertEqual(context.exception.code, "ADMIN_TASK_NOT_ASSIGNED") + self.assertFalse(context.exception.retryable) + + def test_can_simulate_result_validation_failure(self): + self.gateway.enqueue_task( + self._task("TASK-001", TaskType.COLLECT), "client-001" + ) + self.gateway.claim_next(self.client, self.all_capabilities) + self.gateway.reject_next_submission() + + with self.assertRaises(AdminGatewayError) as context: + self.gateway.submit_result("TASK-001", "result-key", self._result()) + + self.assertEqual(context.exception.code, "ADMIN_RESULT_INVALID") + self.assertFalse(context.exception.retryable) + self.assertEqual(self.gateway.submission_count, 0) + + def test_payload_validation_rejects_wrong_result_type(self): + self.gateway.enqueue_task( + self._task("TASK-001", TaskType.COLLECT), "client-001" + ) + self.gateway.claim_next(self.client, self.all_capabilities) + + with self.assertRaises(AdminGatewayError) as context: + self.gateway.submit_result( + "TASK-001", "result-key", self._result("purchase") + ) + + self.assertEqual(context.exception.code, "ADMIN_RESULT_INVALID") + + +if __name__ == "__main__": + unittest.main()