feat(client): add Admin Gateway mock contract (#10)
This commit is contained in:
chengma
@@ -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
|
||||
Reference in New Issue
Block a user