cmbuyer/client/tests/core/test_models.py

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 = claim_wire()
        wide_goods = "1" * 33
        wide["task"].update(
            goods_id=wide_goods,
            product_url="https://mobile.yangkeduo.com/goods.html?goods_id=" + wide_goods,
            max_total_price="1" * 31 + ".00",
            quantity=2_147_483_648,
        )
        self.assertEqual(ClaimedTask.from_wire(wide).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_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"):
            value = claim_wire()
            value["task"]["title"] = title
            with self.subTest(title=repr(title)), self.assertRaises(ValidationError):
                ClaimedTask.from_wire(value)