cmbuyer/client/tests/remote/test_task_source.py

from __future__ import annotations

import json
import unittest

from cmbuyer_client.core.errors import AmbiguousRemoteError, CredentialRemoteError, ManualRemoteError, ProtocolRemoteError
from cmbuyer_client.core.models import ClaimRequest, DeviceCredentials, RenewRequest, SecretToken
from cmbuyer_client.remote.http_transport import HttpResponse
from cmbuyer_client.remote.task_source import HttpTaskSource
from tests.core.test_models import ATTEMPT_ID, TASK_ID, TOKEN, claim_wire
from tests.remote.test_http_transport import _exception_graph


DEVICE_ID = "e3c9f507-7473-4fa6-8d71-8786c34c6301"
SESSION_ID = "23c9f507-7473-4fa6-8d71-8786c34c6301"
REQUEST_ID = "33c9f507-7473-4fa6-8d71-8786c34c6301"
RENEW_ID = "43c9f507-7473-4fa6-8d71-8786c34c6301"


class FakeTransport:
    def __init__(self, response: HttpResponse) -> None:
        self.response = response
        self.calls: list[tuple[object, ...]] = []

    def request(self, *args: object, **kwargs: object) -> HttpResponse:
        self.calls.append(args + (kwargs,))
        return self.response


def response(status: int, value: object | None = None) -> HttpResponse:
    body = b"" if value is None else json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()
    headers = () if value is None else (("Content-Type", "application/json; charset=utf-8"),)
    return HttpResponse(status, headers, body)


class TaskSourceTests(unittest.TestCase):
    def setUp(self) -> None:
        self.credentials = DeviceCredentials(DEVICE_ID, SecretToken(TOKEN))

    def test_claim_success_and_empty_each_send_once_with_exact_headers(self) -> None:
        transport = FakeTransport(response(200, claim_wire()))
        claimed = HttpTaskSource(transport).claim_next(self.credentials, ClaimRequest(SESSION_ID, REQUEST_ID))
        self.assertEqual(claimed.task.id, TASK_ID)
        self.assertEqual(len(transport.calls), 1)
        args = transport.calls[0]
        self.assertEqual(args[1], "/api/v1/tasks/claim-next")
        headers = dict(args[2])
        self.assertEqual(headers["Authorization"], "Bearer " + TOKEN)
        self.assertEqual(headers["X-CMBuyer-Device-ID"], DEVICE_ID)

        empty = FakeTransport(response(204))
        self.assertIsNone(HttpTaskSource(empty).claim_next(self.credentials, ClaimRequest(SESSION_ID, REQUEST_ID)))
        self.assertEqual(len(empty.calls), 1)

    def test_invalid_2xx_is_unknown_and_redirect_is_not_followed(self) -> None:
        malformed = FakeTransport(HttpResponse(200, (("Content-Type", "application/json"),), b'{"task":'))
        with self.assertRaises(AmbiguousRemoteError):
            HttpTaskSource(malformed).claim_next(self.credentials, ClaimRequest(SESSION_ID, REQUEST_ID))
        self.assertEqual(len(malformed.calls), 1)

        secret_body = b'{"claim_token":"' + TOKEN.encode()
        leaking = FakeTransport(HttpResponse(200, (("Content-Type", "application/json"),), secret_body))
        with self.assertRaises(AmbiguousRemoteError) as captured:
            HttpTaskSource(leaking).claim_next(self.credentials, ClaimRequest(SESSION_ID, REQUEST_ID))
        self.assertNotIn(TOKEN, _exception_graph(captured.exception))

        for status in (201, 202, 206):
            with self.subTest(status=status), self.assertRaises(AmbiguousRemoteError):
                HttpTaskSource(FakeTransport(response(status, claim_wire()))).claim_next(
                    self.credentials, ClaimRequest(SESSION_ID, REQUEST_ID)
                )

        redirect = FakeTransport(HttpResponse(302, (("Location", "http://example.invalid"),), b""))
        with self.assertRaises(ProtocolRemoteError):
            HttpTaskSource(redirect).claim_next(self.credentials, ClaimRequest(SESSION_ID, REQUEST_ID))
        self.assertEqual(len(redirect.calls), 1)

    def test_claim_rejects_service_field_bound_drift_as_ambiguous(self) -> None:
        mutations = (
            ("title", "😀" * 121),
            ("sku_color", "色" * 81),
            ("sku_size", "码" * 81),
            ("max_total_price", "1" * 30 + ".00"),
        )
        for field, invalid in mutations:
            value = claim_wire()
            value["task"][field] = invalid
            transport = FakeTransport(response(200, value))
            with self.subTest(field=field), self.assertRaises(AmbiguousRemoteError):
                HttpTaskSource(transport).claim_next(self.credentials, ClaimRequest(SESSION_ID, REQUEST_ID))
            self.assertEqual(len(transport.calls), 1)

        goods_id = "1" * 33
        value = claim_wire()
        value["task"].update(
            goods_id=goods_id,
            product_url="https://mobile.yangkeduo.com/goods.html?goods_id=" + goods_id,
        )
        with self.assertRaises(AmbiguousRemoteError):
            HttpTaskSource(FakeTransport(response(200, value))).claim_next(
                self.credentials, ClaimRequest(SESSION_ID, REQUEST_ID)
            )

    def test_fixed_conflict_and_renew_cas(self) -> None:
        conflict = FakeTransport(response(409, {"error": "claim_requires_manual"}))
        with self.assertRaises(ManualRemoteError):
            HttpTaskSource(conflict).claim_next(self.credentials, ClaimRequest(SESSION_ID, REQUEST_ID))

        request = RenewRequest(TASK_ID, RENEW_ID, SESSION_ID, ATTEMPT_ID, 1, SecretToken(TOKEN), "2026-08-04T09:05:00Z", "2026-08-04T10:00:00Z")
        renewed = response(200, {"task_id": TASK_ID, "attempt_id": ATTEMPT_ID, "claim_generation": 1, "lease_expires_at": "2026-08-04T09:06:00Z"})
        result = HttpTaskSource(FakeTransport(renewed)).renew(self.credentials, request)
        self.assertEqual(result.claim_generation, 1)

        capped = RenewRequest(TASK_ID, RENEW_ID, SESSION_ID, ATTEMPT_ID, 1, SecretToken(TOKEN), "2026-08-04T10:00:00.000000000Z", "2026-08-04T10:00:00Z")
        capped_result = response(200, {"task_id": TASK_ID, "attempt_id": ATTEMPT_ID, "claim_generation": 1, "lease_expires_at": "2026-08-04T10:00:00Z"})
        self.assertEqual(HttpTaskSource(FakeTransport(capped_result)).renew(self.credentials, capped).lease_expires_at, "2026-08-04T10:00:00Z")

        beyond_cap = response(200, {"task_id": TASK_ID, "attempt_id": ATTEMPT_ID, "claim_generation": 1, "lease_expires_at": "2026-08-04T10:00:00.000000001Z"})
        with self.assertRaises(AmbiguousRemoteError):
            HttpTaskSource(FakeTransport(beyond_cap)).renew(self.credentials, request)

        stale = response(200, {"task_id": TASK_ID, "attempt_id": ATTEMPT_ID, "claim_generation": 1, "lease_expires_at": "2026-08-04T09:04:00Z"})
        with self.assertRaises(AmbiguousRemoteError):
            HttpTaskSource(FakeTransport(stale)).renew(self.credentials, request)
        for status in (201, 204):
            with self.subTest(status=status), self.assertRaises(AmbiguousRemoteError):
                HttpTaskSource(FakeTransport(response(status, None if status == 204 else {
                    "task_id": TASK_ID,
                    "attempt_id": ATTEMPT_ID,
                    "claim_generation": 1,
                    "lease_expires_at": "2026-08-04T09:06:00Z",
                }))).renew(self.credentials, request)

    def test_claim_and_renew_error_status_matrix(self) -> None:
        claim_request = ClaimRequest(SESSION_ID, REQUEST_ID)
        claim_cases = (
            (HttpResponse(401, (), b""), CredentialRemoteError),
            (response(400, {"error": "invalid_request"}), ProtocolRemoteError),
            (HttpResponse(403, (), b""), ProtocolRemoteError),
            (response(413, {"error": "request_too_large"}), ProtocolRemoteError),
            (response(415, {"error": "unsupported_media_type"}), ProtocolRemoteError),
            (HttpResponse(500, (), b""), AmbiguousRemoteError),
            (HttpResponse(503, (), b""), AmbiguousRemoteError),
            (HttpResponse(418, (), b""), ProtocolRemoteError),
        )
        for wire_response, expected in claim_cases:
            with self.subTest(status=wire_response.status), self.assertRaises(expected):
                HttpTaskSource(FakeTransport(wire_response)).claim_next(self.credentials, claim_request)

        renew_request = RenewRequest(
            TASK_ID,
            RENEW_ID,
            SESSION_ID,
            ATTEMPT_ID,
            1,
            SecretToken(TOKEN),
            "2026-08-04T09:05:00Z",
            "2026-08-04T10:00:00Z",
        )
        for code in ("idempotency_conflict", "claim_not_current"):
            with self.subTest(code=code), self.assertRaises(ManualRemoteError):
                HttpTaskSource(FakeTransport(response(409, {"error": code}))).renew(self.credentials, renew_request)