598 lines
21 KiB
Python
598 lines
21 KiB
Python
"""不访问网络的 AdminGateway 测试实现。"""
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import unicodedata
|
|
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,
|
|
RegistrationReceipt,
|
|
SpecResolutionMatch,
|
|
SpecResolutionReceipt,
|
|
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._spec_resolutions: Dict[
|
|
str, Tuple[str, SpecResolutionReceipt]
|
|
] = {}
|
|
self._next_spec_resolution: Optional[SpecResolutionReceipt] = None
|
|
self._next_error: Optional[AdminGatewayError] = None
|
|
self._reject_next_submission = False
|
|
self._registrations: Dict[str, Tuple[ClientInfo, ClaimCapabilities]] = {}
|
|
self._lock = Lock()
|
|
|
|
def register_client(
|
|
self, client: ClientInfo, capabilities: ClaimCapabilities
|
|
) -> RegistrationReceipt:
|
|
"""幂等登记 Client，并保留最后一次上报内容供测试检查。"""
|
|
|
|
with self._lock:
|
|
self._raise_forced_error()
|
|
normalized = ClientInfo(client.client_id.strip(), client.name.strip())
|
|
self._registrations[normalized.client_id] = (
|
|
normalized,
|
|
deepcopy(capabilities),
|
|
)
|
|
return RegistrationReceipt(
|
|
registered=True,
|
|
client_id=normalized.client_id,
|
|
registered_at=utc_now_iso(),
|
|
)
|
|
|
|
@property
|
|
def registration_count(self) -> int:
|
|
"""返回不同 Client ID 的登记数量。"""
|
|
|
|
return len(self._registrations)
|
|
|
|
def registered_client(
|
|
self, client_id: str
|
|
) -> Optional[Tuple[ClientInfo, ClaimCapabilities]]:
|
|
"""测试辅助：返回某个 Client 最后一次登记的资料。"""
|
|
|
|
value = self._registrations.get(client_id)
|
|
return deepcopy(value) if value is not None else None
|
|
|
|
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
|
|
if (
|
|
item.task.execution_mode == "live"
|
|
and capabilities.purchase_mode != "live"
|
|
):
|
|
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 set_next_spec_resolution(
|
|
self, receipt: SpecResolutionReceipt
|
|
) -> None:
|
|
"""测试辅助：设置下一次新规格解析请求的业务响应。"""
|
|
|
|
self._next_spec_resolution = deepcopy(receipt)
|
|
|
|
@property
|
|
def spec_resolution_count(self) -> int:
|
|
return len(self._spec_resolutions)
|
|
|
|
def resolve_purchase_spec(
|
|
self,
|
|
task_id: str,
|
|
idempotency_key: str,
|
|
observation: Mapping[str, Any],
|
|
) -> SpecResolutionReceipt:
|
|
with self._lock:
|
|
self._raise_forced_error()
|
|
self._validate_spec_observation(
|
|
task_id, idempotency_key, observation
|
|
)
|
|
task = self._validate_spec_resolution_target(
|
|
task_id, observation
|
|
)
|
|
fingerprint = self._fingerprint(
|
|
"spec_resolution", task_id, observation
|
|
)
|
|
previous = self._spec_resolutions.get(idempotency_key)
|
|
if previous is not None:
|
|
old_fingerprint, receipt = previous
|
|
if old_fingerprint != fingerprint:
|
|
raise AdminGatewayError(
|
|
"IDEMPOTENCY_CONFLICT",
|
|
"相同规格解析幂等键携带了不同内容",
|
|
False,
|
|
)
|
|
return deepcopy(receipt)
|
|
|
|
receipt = self._next_spec_resolution or SpecResolutionReceipt(
|
|
schema_version=1,
|
|
resolution_id=str(uuid4()),
|
|
outcome="uncertain",
|
|
source=None,
|
|
candidate_snapshot_hash=str(
|
|
observation["candidate_snapshot_hash"]
|
|
),
|
|
match=None,
|
|
confidence_bps=None,
|
|
reason="Mock 未配置唯一匹配",
|
|
resolved_at=utc_now_iso(),
|
|
)
|
|
self._next_spec_resolution = None
|
|
self._spec_resolutions[idempotency_key] = (
|
|
fingerprint,
|
|
deepcopy(receipt),
|
|
)
|
|
return deepcopy(receipt)
|
|
|
|
@staticmethod
|
|
def _validate_spec_observation(
|
|
task_id: str,
|
|
idempotency_key: str,
|
|
observation: Mapping[str, Any],
|
|
) -> None:
|
|
try:
|
|
encoded = json.dumps(
|
|
observation,
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
except (TypeError, ValueError) as exc:
|
|
raise AdminGatewayError(
|
|
"INVALID_BODY", "规格解析请求不是有效 JSON", False
|
|
) from exc
|
|
if len(encoded) > 64 * 1024:
|
|
raise AdminGatewayError(
|
|
"INVALID_BODY", "规格解析请求超过 64 KiB", False
|
|
)
|
|
if observation.get("schema_version") != 1:
|
|
raise AdminGatewayError(
|
|
"INVALID_SPEC_RESOLUTION_SCHEMA", "规格解析版本无效", False
|
|
)
|
|
task_version = observation.get("task_version")
|
|
original_options = observation.get("original_options")
|
|
candidates = observation.get("candidates")
|
|
snapshot_hash = observation.get("candidate_snapshot_hash")
|
|
selected_color = observation.get("selected_color")
|
|
target_size = observation.get("target_size")
|
|
observed_at = observation.get("observed_at")
|
|
if (
|
|
not isinstance(task_version, int)
|
|
or isinstance(task_version, bool)
|
|
or task_version <= 0
|
|
or not MockAdminGateway._valid_spec_text(
|
|
observation.get("attempt_id")
|
|
)
|
|
or not MockAdminGateway._valid_spec_text(
|
|
observation.get("pdd_goods_id")
|
|
)
|
|
or not MockAdminGateway._valid_spec_text(selected_color)
|
|
or not MockAdminGateway._valid_spec_text(target_size)
|
|
or not isinstance(original_options, Mapping)
|
|
or not 1 <= len(original_options) <= 16
|
|
or any(
|
|
not MockAdminGateway._valid_spec_text(key)
|
|
or not MockAdminGateway._valid_spec_text(value)
|
|
for key, value in original_options.items()
|
|
)
|
|
or not isinstance(candidates, list)
|
|
or not 1 <= len(candidates) <= 100
|
|
or not isinstance(snapshot_hash, str)
|
|
or re.fullmatch(r"[0-9a-f]{64}", snapshot_hash) is None
|
|
or not MockAdminGateway._valid_observed_at(observed_at)
|
|
):
|
|
raise AdminGatewayError(
|
|
"INVALID_SPEC_RESOLUTION_REQUEST", "规格候选结构无效", False
|
|
)
|
|
if any(
|
|
not isinstance(candidate, Mapping)
|
|
or candidate.get("candidate_id") != f"c{index}"
|
|
or not MockAdminGateway._valid_spec_text(
|
|
candidate.get("raw_text")
|
|
)
|
|
or dict(candidate.get("options") or {})
|
|
!= {
|
|
"color": selected_color,
|
|
"size": candidate.get("raw_text"),
|
|
}
|
|
for index, candidate in enumerate(candidates, start=1)
|
|
):
|
|
raise AdminGatewayError(
|
|
"INVALID_SPEC_RESOLUTION_REQUEST", "规格候选字段无效", False
|
|
)
|
|
candidate_material = "".join(
|
|
(
|
|
MockAdminGateway._frame("spec-resolution-v1"),
|
|
MockAdminGateway._frame(str(observation["pdd_goods_id"])),
|
|
MockAdminGateway._frame(str(selected_color)),
|
|
MockAdminGateway._frame(str(len(candidates))),
|
|
*(
|
|
"".join(
|
|
MockAdminGateway._frame(str(value))
|
|
for value in (
|
|
candidate.get("candidate_id"),
|
|
candidate.get("raw_text"),
|
|
(candidate.get("options") or {}).get("color"),
|
|
(candidate.get("options") or {}).get("size"),
|
|
)
|
|
)
|
|
for candidate in candidates
|
|
if isinstance(candidate, Mapping)
|
|
),
|
|
)
|
|
)
|
|
expected_snapshot_hash = hashlib.sha256(
|
|
candidate_material.encode("utf-8")
|
|
).hexdigest()
|
|
if snapshot_hash != expected_snapshot_hash:
|
|
raise AdminGatewayError(
|
|
"SPEC_RESOLUTION_HASH_MISMATCH", "候选快照哈希不一致", False
|
|
)
|
|
identity = "".join(
|
|
MockAdminGateway._frame(str(value))
|
|
for value in (
|
|
task_id,
|
|
observation["attempt_id"],
|
|
snapshot_hash,
|
|
"spec-resolution-v1",
|
|
)
|
|
)
|
|
expected_key = "spec-resolution-v1:" + hashlib.sha256(
|
|
identity.encode("utf-8")
|
|
).hexdigest()
|
|
if idempotency_key != expected_key:
|
|
raise AdminGatewayError(
|
|
"SPEC_RESOLUTION_HASH_MISMATCH", "规格解析幂等键不一致", False
|
|
)
|
|
|
|
def _validate_spec_resolution_target(
|
|
self,
|
|
task_id: str,
|
|
observation: Mapping[str, Any],
|
|
) -> AdminTask:
|
|
item = self._find_task(task_id)
|
|
if item is None:
|
|
raise AdminGatewayError(
|
|
"TASK_NOT_FOUND", "规格解析任务不存在", False
|
|
)
|
|
task = item.task
|
|
if task.task_type.value != "purchase":
|
|
raise AdminGatewayError(
|
|
"TASK_NOT_PURCHASE", "当前任务不是采购任务", False
|
|
)
|
|
if observation.get("task_version") != task.version:
|
|
raise AdminGatewayError(
|
|
"TASK_VERSION_CONFLICT", "规格解析任务版本不一致", False
|
|
)
|
|
payload = task.payload
|
|
if observation.get("pdd_goods_id") != payload.get("goods_id"):
|
|
raise AdminGatewayError(
|
|
"PDD_GOODS_MISMATCH", "规格解析商品编号不一致", False
|
|
)
|
|
original_options = dict(observation.get("original_options") or {})
|
|
if (
|
|
original_options != dict(payload.get("options") or {})
|
|
or not MockAdminGateway._target_option_was_claimed(
|
|
original_options,
|
|
"color",
|
|
str(observation.get("selected_color") or ""),
|
|
)
|
|
or not MockAdminGateway._target_option_was_claimed(
|
|
original_options,
|
|
"size",
|
|
str(observation.get("target_size") or ""),
|
|
)
|
|
):
|
|
raise AdminGatewayError(
|
|
"INVALID_SPEC_RESOLUTION_REQUEST", "原始规格不一致", False
|
|
)
|
|
if task_id not in self._claimed_task_ids:
|
|
raise AdminGatewayError(
|
|
"TASK_NOT_CLAIMED_BY_CLIENT",
|
|
"该 Client 从未领取过此任务",
|
|
False,
|
|
)
|
|
return task
|
|
|
|
@staticmethod
|
|
def _target_option_was_claimed(
|
|
options: Mapping[str, Any], preferred_key: str, target: str
|
|
) -> bool:
|
|
if preferred_key in options:
|
|
return options[preferred_key] == target
|
|
return target in options.values()
|
|
|
|
@staticmethod
|
|
def _valid_spec_text(value: object) -> bool:
|
|
return (
|
|
isinstance(value, str)
|
|
and 1 <= len(value) <= 191
|
|
and not any(
|
|
unicodedata.category(character) == "Cc"
|
|
for character in value
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def _valid_observed_at(value: object) -> bool:
|
|
if not isinstance(value, str) or re.fullmatch(
|
|
r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}"
|
|
r"(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})",
|
|
value,
|
|
) is None:
|
|
return False
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return False
|
|
return parsed.tzinfo is not None
|
|
|
|
@staticmethod
|
|
def _frame(value: str) -> str:
|
|
return f"{len(value.encode('utf-8'))}:{value}"
|
|
|
|
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
|