from __future__ import annotations import json import unittest from cmbuyer_client.core.errors import ValidationError from cmbuyer_client.core.models import ClaimedTask, SecretToken from cmbuyer_client.core.validation import rfc3339_z_nanoseconds, strict_json_loads TASK_ID = "13c9f507-7473-4fa6-8d71-8786c34c6301" AUTH_ID = "73c9f507-7473-4fa6-8d71-8786c34c6301" ATTEMPT_ID = "53c9f507-7473-4fa6-8d71-8786c34c6301" TOKEN = "0123456789abcdef" * 4 def claim_wire() -> dict[str, object]: return { "task": { "id": TASK_ID, "version": 3, "title": "纯棉短袖", "product_url": "https://mobile.yangkeduo.com/goods.html?goods_id=937122477375", "goods_id": "937122477375", "sku_color": "黑色CHA(纯棉)", "sku_size": "M(建议100-115)", "quantity": 2, "max_total_price": "30.00", }, "authorization": {"id": AUTH_ID, "task_version": 2, "expires_at": "2026-08-04T10:00:00Z"}, "attempt": { "id": ATTEMPT_ID, "claim_token": TOKEN, "claim_generation": 1, "lease_expires_at": "2026-08-04T09:05:00Z", }, } class CoreModelsTests(unittest.TestCase): def test_claim_wire_round_trip_and_secret_repr(self) -> None: claimed = ClaimedTask.from_wire(claim_wire()) self.assertEqual(claimed.task.quantity, 2) self.assertNotIn(TOKEN, repr(claimed)) self.assertNotIn(TOKEN, repr(SecretToken(TOKEN))) def test_rejects_bool_float_wrong_url_and_version_drift(self) -> None: mutations = [] for mutate in ( lambda value: value["task"].__setitem__("quantity", True), lambda value: value["task"].__setitem__("max_total_price", "30.0"), lambda value: value["task"].__setitem__("max_total_price", "0.00"), lambda value: value["task"].__setitem__("product_url", "https://example.invalid/"), lambda value: value["task"].__setitem__("version", 2), ): value = claim_wire() mutate(value) mutations.append(value) for value in mutations: with self.subTest(value=value), self.assertRaises(ValidationError): ClaimedTask.from_wire(value) def test_strict_json_rejects_nested_duplicates_float_nan_bom_and_utf8(self) -> None: bad_values = ( b'{"task":{"id":1,"id":2}}', b'{"value":1.0}', b'{"value":NaN}', b'\xef\xbb\xbf{}', b'\xff', ('{"value":' + "9" * 5000 + '}').encode(), ) for raw in bad_values: with self.subTest(raw=raw), self.assertRaises(ValidationError): strict_json_loads(raw, maximum=1024) self.assertEqual(strict_json_loads(json.dumps({"value": 1}).encode(), maximum=1024), {"value": 1}) def test_rfc3339_nano_comparison_preserves_all_fraction_digits(self) -> None: equal = ( "2026-08-04T09:01:00.1Z", "2026-08-04T09:01:00.100000Z", "2026-08-04T09:01:00.100000000Z", ) self.assertEqual(len({rfc3339_z_nanoseconds(value) for value in equal}), 1) ordered = ( "2026-08-04T09:01:00Z", "2026-08-04T09:01:00.000001Z", "2026-08-04T09:01:00.0000011Z", "2026-08-04T09:01:00.000001101Z", "2026-08-04T09:01:01Z", ) self.assertEqual([rfc3339_z_nanoseconds(value) for value in ordered], sorted(rfc3339_z_nanoseconds(value) for value in ordered)) def test_money_accepts_positive_subunit_but_rejects_zero_and_noncanonical_forms(self) -> None: value = claim_wire() value["task"]["max_total_price"] = "0.01" self.assertEqual(ClaimedTask.from_wire(value).task.max_total_price, "0.01") for invalid in ("0.00", "00.01", "1.0", "1.000", "1", 1.0, "1.１２", "1.٠٠", "1２.00"): with self.subTest(invalid=invalid), self.assertRaises(ValidationError): changed = claim_wire() changed["task"]["max_total_price"] = invalid ClaimedTask.from_wire(changed) wide_quantity = claim_wire() wide_quantity["task"]["quantity"] = 2_147_483_648 self.assertEqual(ClaimedTask.from_wire(wide_quantity).task.quantity, 2_147_483_648) for invalid_goods in ("１２３", "1٢3"): changed = claim_wire() changed["task"]["goods_id"] = invalid_goods changed["task"]["product_url"] = "https://mobile.yangkeduo.com/goods.html?goods_id=" + invalid_goods with self.subTest(invalid_goods=invalid_goods), self.assertRaises(ValidationError): ClaimedTask.from_wire(changed) too_large = claim_wire() too_large["task"]["quantity"] = 9_223_372_036_854_775_808 with self.assertRaises(ValidationError): ClaimedTask.from_wire(too_large) def test_claim_fields_share_explicit_server_bounds(self) -> None: legal = claim_wire() legal_goods = "1" * 32 legal["task"].update( version=9_223_372_036_854_775_807, title="😀" * 120, goods_id=legal_goods, product_url="https://mobile.yangkeduo.com/goods.html?goods_id=" + legal_goods, sku_color="色" * 80, sku_size="码" * 80, max_total_price="1" * 29 + ".00", ) legal["authorization"]["task_version"] = 9_223_372_036_854_775_806 claimed = ClaimedTask.from_wire(legal) self.assertEqual(len(claimed.task.title), 120) # Python's default ensure_ascii=True expands astral characters to surrogate # escape pairs, so this is a conservative parser-budget proof as well. self.assertLess(len(json.dumps(legal, separators=(",", ":")).encode()), 32 * 1024) mutations = ( ("title", "😀" * 121), ("title", " title"), ("sku_color", "色" * 81), ("sku_color", "black "), ("sku_size", "码" * 81), ("sku_size", " M"), ("max_total_price", "1" * 30 + ".00"), ) for field, invalid in mutations: changed = claim_wire() changed["task"][field] = invalid with self.subTest(field=field, length=len(invalid)), self.assertRaises(ValidationError): ClaimedTask.from_wire(changed) overlong_goods = "1" * 33 changed = claim_wire() changed["task"].update( goods_id=overlong_goods, product_url="https://mobile.yangkeduo.com/goods.html?goods_id=" + overlong_goods, ) with self.assertRaises(ValidationError): ClaimedTask.from_wire(changed) def test_wire_strings_reject_lone_surrogates_but_accept_valid_pair(self) -> None: for escaped in (r'"\ud800"', r'"\udc00"'): value = claim_wire() value["task"]["title"] = json.loads(escaped) with self.subTest(escaped=escaped), self.assertRaises(ValidationError): ClaimedTask.from_wire(value) value = claim_wire() value["task"]["title"] = json.loads(r'"\ud83d\ude00"') self.assertEqual(ClaimedTask.from_wire(value).task.title, "😀") def test_title_rejects_ascii_and_unicode_whitespace_only(self) -> None: for title in ("", " \t\r\n", "\u3000", " \u3000\t", "\u00a0title", "title\u00a0"): value = claim_wire() value["task"]["title"] = title with self.subTest(title=repr(title)), self.assertRaises(ValidationError): ClaimedTask.from_wire(value) def test_persisted_text_has_runtime_independent_c0_and_nbsp_domain(self) -> None: for field in ("title", "sku_color", "sku_size"): for invalid in ("\u001cvalue", "value\u001f", "value\u001dinside", "\u00a0value", "value\u00a0"): value = claim_wire() value["task"][field] = invalid with self.subTest(field=field, invalid=repr(invalid)), self.assertRaises(ValidationError): ClaimedTask.from_wire(value) value = claim_wire() value["task"][field] = "left\u00a0right" self.assertEqual(getattr(ClaimedTask.from_wire(value).task, field), "left\u00a0right")