"""Regression: UCloud driver-layer ``_signed_call`` transient retry.

Companion to the CMDB provider ``_call`` fix (prod 2026-05-27, jpn-tky
UHost ``RetCode=8498`` "Call third api service timeout"). The
lifecycle/billing driver (``UCloudDriver``) signs its own calls through
``_signed_call`` and, pre-fix, had NO retry budget at all — a single
gateway timeout failed the cost_facts cron / IAM read outright.

The fix adds a reads-only retry budget gated by ``retryable``:

  * ``retryable=True``  (idempotent reads: GetRegion / List* /
    ListUBillDetail) — 3 attempts, exp backoff, on transport errors,
    HTTP 5xx, and body-level transient RetCodes.
  * ``retryable=False`` (default; NON-idempotent mutations: CreateUser /
    DeleteUser / Attach/DetachPolicies / DeleteAccessKey) — single shot,
    EXACT historical behavior. A timed-out mutation must NOT silently
    re-fire (it could have succeeded cloud-side; a retry would surface a
    spurious "already exists" / "not found" and corrupt the verdict).
"""
from __future__ import annotations

from typing import Any

import httpx
import pytest

from app.integrations.drivers import ucloud_cloud as _mod
from app.integrations.drivers.ucloud_cloud import _signed_call


class _FakeResp:
    def __init__(
        self, status_code: int, payload: dict | None = None, text: str = "",
    ) -> None:
        self.status_code = status_code
        self._payload = payload or {}
        self.text = text
        self.aclosed = False

    def json(self) -> dict[str, Any]:
        return self._payload

    def raise_for_status(self) -> None:
        if self.status_code >= 400:
            raise RuntimeError(f"unexpected status {self.status_code}")

    async def aclose(self) -> None:
        self.aclosed = True


class _FakeClient:
    """Returns each queued response/exception in turn; repeats the last."""

    def __init__(self, queue: list[Any]) -> None:
        self._queue = queue
        self.calls = 0

    async def post(self, _url: str, data: dict | None = None) -> _FakeResp:
        item = self._queue[min(self.calls, len(self._queue) - 1)]
        self.calls += 1
        if isinstance(item, Exception):
            raise item
        return item


@pytest.fixture(autouse=True)
def _no_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
    async def _noop(_d: float) -> None:
        return None

    monkeypatch.setattr(_mod.asyncio, "sleep", _noop)


_TIMEOUT_BODY = {
    "RetCode": 8498,
    "Message": "Call third api service timeout, please retry",
}


@pytest.mark.asyncio
async def test_read_retries_transient_retcode_then_succeeds() -> None:
    client = _FakeClient([
        _FakeResp(200, _TIMEOUT_BODY),
        _FakeResp(200, {"RetCode": 0, "DataSet": []}),
    ])
    data = await _signed_call(
        client, "PUB", "PRIV", "ListUBillDetail", {"BillingCycle": "2026-05"},
        retryable=True,
    )
    assert data == {"RetCode": 0, "DataSet": []}
    assert client.calls == 2


@pytest.mark.asyncio
async def test_read_persistent_transient_exhausts_then_raises() -> None:
    client = _FakeClient([_FakeResp(200, _TIMEOUT_BODY)])
    with pytest.raises(RuntimeError, match="8498"):
        await _signed_call(
            client, "PUB", "PRIV", "ListUsers", {"Offset": 0, "Limit": 100},
            retryable=True,
        )
    assert client.calls == 3


@pytest.mark.asyncio
async def test_mutation_does_not_retry_transient_retcode() -> None:
    """CRITICAL: a mutation (retryable=False) must fire exactly once on a
    transient RetCode — no silent double-execution of a write."""
    client = _FakeClient([_FakeResp(200, _TIMEOUT_BODY)])
    with pytest.raises(RuntimeError, match="8498"):
        await _signed_call(
            client, "PUB", "PRIV", "DeleteUser", {"UserName": "u1"},
            # retryable defaults to False
        )
    assert client.calls == 1


@pytest.mark.asyncio
async def test_read_retries_transport_error_then_succeeds() -> None:
    client = _FakeClient([
        httpx.ConnectError("connection reset"),
        _FakeResp(200, {"RetCode": 0, "DataSet": []}),
    ])
    data = await _signed_call(
        client, "PUB", "PRIV", "GetRegion", None, retryable=True,
    )
    assert data == {"RetCode": 0, "DataSet": []}
    assert client.calls == 2


@pytest.mark.asyncio
async def test_mutation_transport_error_propagates_raw_once() -> None:
    """A mutation transport error keeps the historical raw-propagation
    contract (httpx error, not wrapped) and does NOT retry."""
    client = _FakeClient([httpx.ConnectError("connection reset")])
    with pytest.raises(httpx.ConnectError):
        await _signed_call(
            client, "PUB", "PRIV", "CreateUser", {"UserName": "u1"},
        )
    assert client.calls == 1


@pytest.mark.asyncio
async def test_read_retries_5xx_then_succeeds() -> None:
    five_oh_three = _FakeResp(503, text="busy")
    client = _FakeClient([
        five_oh_three,
        _FakeResp(200, {"RetCode": 0, "DataSet": []}),
    ])
    data = await _signed_call(
        client, "PUB", "PRIV", "ListUBillDetail", {"BillingCycle": "2026-05"},
        retryable=True,
    )
    assert data == {"RetCode": 0, "DataSet": []}
    assert client.calls == 2
    assert five_oh_three.aclosed is True  # connection released before sleep


@pytest.mark.asyncio
async def test_read_success_first_try_no_retry() -> None:
    client = _FakeClient([_FakeResp(200, {"RetCode": 0, "DataSet": [1]})])
    data = await _signed_call(
        client, "PUB", "PRIV", "GetRegion", None, retryable=True,
    )
    assert data == {"RetCode": 0, "DataSet": [1]}
    assert client.calls == 1


@pytest.mark.asyncio
async def test_auth_status_not_retried_even_for_reads() -> None:
    """HTTP 401/403 is a credential problem — retrying is pointless and
    must surface immediately even on a read."""
    client = _FakeClient([_FakeResp(403, text="forbidden")])
    with pytest.raises(RuntimeError, match="auth error"):
        await _signed_call(
            client, "PUB", "PRIV", "ListUsers", {"Offset": 0}, retryable=True,
        )
    assert client.calls == 1


@pytest.mark.asyncio
async def test_auth_retcode_with_transient_hint_not_retried_on_read() -> None:
    """Body-level auth RetCode (100 'Invalid PublicKey') must NOT be
    retried even on a read path, even if its Message happens to carry a
    transient-looking phrase. ``_signed_call`` has no body-level auth
    classification (only HTTP 401/403), so the guard lives in
    ``_is_retryable_retcode``'s never-retry bucket. Regression for the
    2026-05-27 independent-review LOW finding."""
    client = _FakeClient([_FakeResp(200, {
        "RetCode": 100,
        "Message": "Invalid PublicKey, please retry",
    })])
    with pytest.raises(RuntimeError, match="100"):
        await _signed_call(
            client, "PUB", "PRIV", "ListUsers", {"Offset": 0}, retryable=True,
        )
    assert client.calls == 1  # auth code is never retried


def test_predicate_excludes_auth_and_region_skip_buckets() -> None:
    """Unit-pin :func:`_is_retryable_retcode`'s never-retry set so the
    predicate stays self-contained regardless of call site."""
    from app.domains.cmdb.providers.ucloud import _is_retryable_retcode

    # Auth + region-skip codes: never retried, even with a timeout hint.
    for code in (100, 150, 230, 231, 299):
        assert _is_retryable_retcode(code, "please retry, timeout") is False
    # Known transient code + message-hint codes: retried.
    assert _is_retryable_retcode(8498, "Call third api service timeout") is True
    assert _is_retryable_retcode(9999, "backend time out, try again") is True
    # Deterministic client error (no hint): not retried.
    assert _is_retryable_retcode(171, "Action not supported") is False