diff --git a/client/src/mock_admin_gateway.py b/client/src/mock_admin_gateway.py index 970eab4..8db42a6 100644 --- a/client/src/mock_admin_gateway.py +++ b/client/src/mock_admin_gateway.py @@ -2,6 +2,8 @@ import hashlib import json +import re +import unicodedata from copy import deepcopy from dataclasses import dataclass from datetime import datetime, timezone @@ -195,8 +197,12 @@ class MockAdminGateway(AdminGateway): ) -> SpecResolutionReceipt: with self._lock: self._raise_forced_error() - task = self._validate_submission_target(task_id, idempotency_key) - self._validate_spec_observation(task, idempotency_key, observation) + 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 ) @@ -233,42 +239,59 @@ class MockAdminGateway(AdminGateway): @staticmethod def _validate_spec_observation( - task: AdminTask, + task_id: str, idempotency_key: str, observation: Mapping[str, Any], ) -> None: - if task.task_type.value != "purchase": + try: + encoded = json.dumps( + observation, + ensure_ascii=False, + separators=(",", ":"), + ).encode("utf-8") + except (TypeError, ValueError) as exc: raise AdminGatewayError( - "TASK_NOT_PURCHASE", "当前任务不是采购任务", False + "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 ) - 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 - ) - if dict(observation.get("original_options") or {}) != dict( - payload.get("options") or {} - ): - raise AdminGatewayError( - "INVALID_SPEC_RESOLUTION_REQUEST", "原始规格不一致", 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(candidates, list) + 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 len(snapshot_hash) != 64 - or not str(observation.get("attempt_id") or "").strip() + 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 @@ -276,8 +299,9 @@ class MockAdminGateway(AdminGateway): if any( not isinstance(candidate, Mapping) or candidate.get("candidate_id") != f"c{index}" - or not isinstance(candidate.get("raw_text"), str) - or not candidate["raw_text"] + or not MockAdminGateway._valid_spec_text( + candidate.get("raw_text") + ) or dict(candidate.get("options") or {}) != { "color": selected_color, @@ -319,7 +343,7 @@ class MockAdminGateway(AdminGateway): identity = "".join( MockAdminGateway._frame(str(value)) for value in ( - task.task_id, + task_id, observation["attempt_id"], snapshot_hash, "spec-resolution-v1", @@ -333,6 +357,88 @@ class MockAdminGateway(AdminGateway): "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}" diff --git a/client/test/test_admin_gateway_contract.py b/client/test/test_admin_gateway_contract.py index 23d6f8c..a6df6eb 100644 --- a/client/test/test_admin_gateway_contract.py +++ b/client/test/test_admin_gateway_contract.py @@ -2,6 +2,7 @@ import hashlib import unittest +from copy import deepcopy from src.admin_gateway import ( AdminGateway, @@ -17,6 +18,63 @@ from src.mock_admin_gateway import MockAdminGateway from src.task_models import TaskType +def _frame(value: str) -> str: + return f"{len(value.encode('utf-8'))}:{value}" + + +def _finalize_spec_observation(task_id: str, observation: dict) -> str: + candidates = observation["candidates"] + snapshot_values = [ + "spec-resolution-v1", + observation["pdd_goods_id"], + observation["selected_color"], + str(len(candidates)), + ] + for candidate in candidates: + snapshot_values.extend( + ( + candidate["candidate_id"], + candidate["raw_text"], + candidate["options"]["color"], + candidate["options"]["size"], + ) + ) + observation["candidate_snapshot_hash"] = hashlib.sha256( + "".join(_frame(value) for value in snapshot_values).encode("utf-8") + ).hexdigest() + identity_values = ( + task_id, + observation["attempt_id"], + observation["candidate_snapshot_hash"], + "spec-resolution-v1", + ) + return "spec-resolution-v1:" + hashlib.sha256( + "".join(_frame(value) for value in identity_values).encode("utf-8") + ).hexdigest() + + +def _spec_observation(task_id: str = "PUR-SPEC") -> tuple[dict, str]: + observation = { + "schema_version": 1, + "task_version": 3, + "attempt_id": "attempt-001", + "pdd_goods_id": "737116531267", + "original_options": {"color": "黑色", "size": "60公斤"}, + "selected_color": "黑色", + "target_size": "60公斤", + "candidates": [ + { + "candidate_id": "c1", + "raw_text": "120斤", + "options": {"color": "黑色", "size": "120斤"}, + } + ], + "candidate_snapshot_hash": "", + "observed_at": "2026-08-17T08:00:00Z", + } + return observation, _finalize_spec_observation(task_id, observation) + + class MockAdminGatewayContractTest(unittest.TestCase): def setUp(self): self.gateway = MockAdminGateway() @@ -252,6 +310,115 @@ class MockAdminGatewayContractTest(unittest.TestCase): self.assertEqual(second, configured) self.assertEqual(self.gateway.spec_resolution_count, 1) + def test_spec_resolution_validation_matches_admin_contract(self): + task_id = "PUR-SPEC-VALIDATION" + task = AdminTask( + task_id=task_id, + task_type=TaskType.PURCHASE, + version=3, + priority=1, + payload={ + "goods_id": "737116531267", + "options": {"color": "黑色", "size": "60公斤"}, + }, + ) + self.gateway.enqueue_task(task, self.client.client_id) + self.gateway.claim_next(self.client, self.all_capabilities) + original, _key = _spec_observation(task_id) + + cases = [] + + wrong_color = deepcopy(original) + wrong_color["selected_color"] = "白色" + wrong_color["candidates"][0]["options"]["color"] = "白色" + cases.append(( + "已选颜色不是领取规格", + wrong_color, + _finalize_spec_observation(task_id, wrong_color), + "INVALID_SPEC_RESOLUTION_REQUEST", + )) + + wrong_size = deepcopy(original) + wrong_size["target_size"] = "70公斤" + cases.append(( + "目标尺码不是领取规格", + wrong_size, + _finalize_spec_observation(task_id, wrong_size), + "INVALID_SPEC_RESOLUTION_REQUEST", + )) + + missing_timezone = deepcopy(original) + missing_timezone["observed_at"] = "2026-08-17T08:00:00" + cases.append(( + "观测时间没有时区", + missing_timezone, + _finalize_spec_observation(task_id, missing_timezone), + "INVALID_SPEC_RESOLUTION_REQUEST", + )) + + oversized_attempt = deepcopy(original) + oversized_attempt["attempt_id"] = "a" * 192 + cases.append(( + "执行尝试编号超长", + oversized_attempt, + _finalize_spec_observation(task_id, oversized_attempt), + "INVALID_SPEC_RESOLUTION_REQUEST", + )) + + control_character = deepcopy(original) + control_character["target_size"] = "60\n公斤" + cases.append(( + "字段包含控制字符", + control_character, + _finalize_spec_observation(task_id, control_character), + "INVALID_SPEC_RESOLUTION_REQUEST", + )) + + oversized_body = deepcopy(original) + oversized_body["future_padding"] = "x" * (64 * 1024) + cases.append(( + "请求体超过上限", + oversized_body, + _finalize_spec_observation(task_id, oversized_body), + "INVALID_BODY", + )) + + for name, request, key, expected_code in cases: + with self.subTest(name=name): + with self.assertRaises(AdminGatewayError) as raised: + self.gateway.resolve_purchase_spec(task_id, key, request) + self.assertEqual(raised.exception.code, expected_code) + self.assertFalse(raised.exception.retryable) + + def test_spec_resolution_uses_admin_task_and_claim_error_codes(self): + observation, key = _spec_observation("PUR-MISSING") + with self.assertRaises(AdminGatewayError) as missing: + self.gateway.resolve_purchase_spec( + "PUR-MISSING", key, observation + ) + self.assertEqual(missing.exception.code, "TASK_NOT_FOUND") + + task_id = "PUR-UNCLAIMED" + self.gateway.enqueue_task( + AdminTask( + task_id=task_id, + task_type=TaskType.PURCHASE, + version=3, + priority=1, + payload={ + "goods_id": "737116531267", + "options": {"color": "黑色", "size": "60公斤"}, + }, + ), + self.client.client_id, + ) + observation, key = _spec_observation(task_id) + with self.assertRaises(AdminGatewayError) as unclaimed: + self.gateway.resolve_purchase_spec(task_id, key, observation) + self.assertEqual( + unclaimed.exception.code, "TASK_NOT_CLAIMED_BY_CLIENT" + ) + def test_same_idempotency_key_and_content_reuses_receipt(self): self.gateway.enqueue_task( self._task("TASK-001", TaskType.COLLECT), "client-001"