"""Unit tests for the Tencent Cloud SDK throttle + retry shim.

The shim exists to keep ``cmdb.grants_sync`` running cleanly against
Tencent CAM's documented 5 RPS-per-tenant ceiling. The 2026-05-09
production incident hit this: the per-user
``ListAttachedUserAllPolicies`` loop in the IAM grants discovery path
trivially exceeded 5 RPS once a tenant had >5 active CAM users, and
the resulting partial snapshot was a mass-revoke hazard (mark-and-
sweep would close every grant the cap-truncated snapshot omitted).

Tests cover:

  * Token bucket actually paces calls at the configured RPS.
  * Per-credential isolation (two integrations don't share quota).
  * RequestLimitExceeded triggers retry with backoff; eventual success
    returns the success result.
  * Repeated RequestLimitExceeded eventually re-raises after
    ``max_retries`` exhausts (must NOT silently degrade to None / [],
    which the caller's mark-and-sweep would interpret as authoritative
    empty).
  * Non-rate-limit exceptions propagate immediately without retry —
    auth / not-found / 5xx etc. keep their existing semantics.
  * Sync flavour (used inside ``asyncio.to_thread``-wrapped flows)
    enforces the same throttle from a worker thread.
"""
from __future__ import annotations

import asyncio
import time

import pytest

from app.integrations import _tencent_throttle as throttle


@pytest.fixture(autouse=True)
def _reset_buckets():
    throttle._reset_buckets_for_tests()
    throttle._sync_state.clear()
    yield
    throttle._reset_buckets_for_tests()
    throttle._sync_state.clear()


# ── Stand-in TencentCloudSDKException ────────────────────────────────


class _FakeTencentExc(Exception):
    """Match the shape the real SDK exposes: ``.code`` + ``str(self)``."""

    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code


# ── Token bucket pacing ─────────────────────────────────────────────


@pytest.mark.asyncio
async def test_token_bucket_paces_calls_at_configured_rps() -> None:
    """5 calls at 10 RPS should take ~0.4s (4 inter-call gaps × 0.1s)."""
    counter = {"n": 0}

    def _ok(_req):
        counter["n"] += 1
        return "ok"

    start = time.monotonic()
    for _ in range(5):
        await throttle.tencent_call(
            _ok, None, secret_id="K", service="test", rate=10.0,
        )
    elapsed = time.monotonic() - start
    assert counter["n"] == 5
    # 4 gaps × 0.1s = 0.4s minimum. Allow 0.05s slack for scheduler jitter
    # but require it's at least mostly enforced.
    assert elapsed >= 0.35, f"throttle did not pace calls: elapsed={elapsed:.3f}s"


@pytest.mark.asyncio
async def test_distinct_secret_ids_have_independent_buckets() -> None:
    """Two tenants must not share a quota."""
    def _ok(_req):
        return "ok"

    # Two interleaved series at 5 RPS each. If they shared a bucket,
    # 10 calls would take ~1.8s; with separate buckets, ~0.9s.
    start = time.monotonic()
    tasks = []
    for i in range(5):
        tasks.append(throttle.tencent_call(
            _ok, i, secret_id="A", service="cam", rate=5.0,
        ))
        tasks.append(throttle.tencent_call(
            _ok, i, secret_id="B", service="cam", rate=5.0,
        ))
    await asyncio.gather(*tasks)
    elapsed = time.monotonic() - start
    # Independent buckets: ~0.8s ceiling. Shared: would be ~1.8s. We
    # assert the lower bound by checking it's clearly under the shared
    # case so the test catches a regression.
    assert elapsed < 1.4, (
        f"buckets appear shared across credentials: elapsed={elapsed:.3f}s"
    )


@pytest.mark.asyncio
async def test_distinct_services_have_independent_buckets() -> None:
    """CAM bucket and billing bucket are metered separately on Tencent's side."""
    def _ok(_req):
        return "ok"

    start = time.monotonic()
    tasks = []
    for i in range(5):
        tasks.append(throttle.tencent_call(
            _ok, i, secret_id="K", service="cam", rate=5.0,
        ))
        tasks.append(throttle.tencent_call(
            _ok, i, secret_id="K", service="billing", rate=5.0,
        ))
    await asyncio.gather(*tasks)
    elapsed = time.monotonic() - start
    assert elapsed < 1.4, (
        f"services appear to share a bucket: elapsed={elapsed:.3f}s"
    )


# ── Retry-on-RequestLimitExceeded ───────────────────────────────────


@pytest.mark.asyncio
async def test_retries_on_request_limit_exceeded_then_succeeds(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Two cap-rejections, third call returns OK — caller sees the OK."""
    monkeypatch.setattr(throttle, "BACKOFF_BASE", 0.01)  # speed up test
    monkeypatch.setattr(throttle, "BACKOFF_CAP", 0.01)

    attempts = {"n": 0}

    def _flaky(_req):
        attempts["n"] += 1
        if attempts["n"] <= 2:
            raise _FakeTencentExc(
                code="RequestLimitExceeded",
                message="您当前每秒请求 6 次，超过了每秒频率上限 5",
            )
        return "ok"

    out = await throttle.tencent_call(
        _flaky, None, secret_id="K", service="cam", rate=100.0,
    )
    assert out == "ok"
    assert attempts["n"] == 3


@pytest.mark.asyncio
async def test_exhausts_retries_then_re_raises(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A persistent rate-limit storm must surface as the SDK exception.

    Critical: returning a successful empty result here would let
    ``grants_sync`` treat the cap-truncated snapshot as authoritative
    and mass-revoke every Helix-tracked grant.
    """
    monkeypatch.setattr(throttle, "BACKOFF_BASE", 0.001)
    monkeypatch.setattr(throttle, "BACKOFF_CAP", 0.001)

    def _always_throttled(_req):
        raise _FakeTencentExc(
            code="RequestLimitExceeded",
            message="您当前每秒请求 6 次，超过了每秒频率上限 5",
        )

    with pytest.raises(_FakeTencentExc):
        await throttle.tencent_call(
            _always_throttled, None,
            secret_id="K", service="cam", rate=100.0, max_retries=2,
        )


@pytest.mark.asyncio
async def test_non_rate_limit_error_propagates_without_retry() -> None:
    """Auth errors and other SDK exceptions are NOT retried.

    Retrying ``AuthFailure`` would burn the retry budget on a credential
    the admin must fix. The throttle must only kick in on cap codes.
    """
    attempts = {"n": 0}

    def _auth_fail(_req):
        attempts["n"] += 1
        raise _FakeTencentExc(
            code="AuthFailure", message="invalid SecretKey",
        )

    with pytest.raises(_FakeTencentExc):
        await throttle.tencent_call(
            _auth_fail, None, secret_id="K", service="cam", rate=100.0,
        )
    assert attempts["n"] == 1


@pytest.mark.asyncio
async def test_substring_fallback_anchors_on_sdk_canonical_format() -> None:
    """An unrelated exception that merely *quotes* the literal must NOT retry.

    Pre-tightening the substring match was a bare ``"RequestLimitExceeded"
    in str(exc)``, which would happily retry any wrapper exception that
    quoted the rate-limit name in its message — e.g. an upstream
    ``RuntimeError("seen RequestLimitExceeded earlier in trace")``. The
    canonical SDK shape is ``[TencentCloudSDKException]
    code:RequestLimitExceeded message:Y``, so the matcher anchors on
    ``code:RequestLimitExceeded``.
    """

    def _quoting_wrapper(_req):
        # Plain Python exception that mentions the literal but is NOT
        # a real rate-limit. Must NOT trigger a retry.
        raise RuntimeError("we saw RequestLimitExceeded earlier and recovered")

    with pytest.raises(RuntimeError):
        await throttle.tencent_call(
            _quoting_wrapper, None,
            secret_id="K", service="cam", rate=1000.0, max_retries=2,
        )


@pytest.mark.asyncio
async def test_recognises_request_limit_subcodes() -> None:
    """The Tencent SDK uses dotted subcodes; the matcher must accept them."""
    monkeypatched_attempts = {"n": 0}

    def _subcode_throttle(_req):
        monkeypatched_attempts["n"] += 1
        if monkeypatched_attempts["n"] == 1:
            raise _FakeTencentExc(
                code="RequestLimitExceeded.UinLimitExceeded",
                message="账户级别 QPS 超限",
            )
        return "ok"

    out = await throttle.tencent_call(
        _subcode_throttle, None,
        secret_id="K", service="cam", rate=1000.0, max_retries=2,
    )
    assert out == "ok"
    assert monkeypatched_attempts["n"] == 2


# ── Sync flavour (worker-thread path) ───────────────────────────────


def test_sync_flavour_paces_calls() -> None:
    """``tencent_call_sync`` enforces the same throttle from a sync caller."""
    counter = {"n": 0}

    def _ok(_req):
        counter["n"] += 1
        return "ok"

    start = time.monotonic()
    for _ in range(5):
        throttle.tencent_call_sync(
            _ok, None, secret_id="S", service="test_sync", rate=10.0,
        )
    elapsed = time.monotonic() - start
    assert counter["n"] == 5
    assert elapsed >= 0.35, f"sync throttle did not pace: elapsed={elapsed:.3f}s"


def test_sync_flavour_retries_then_succeeds(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(throttle, "BACKOFF_BASE", 0.001)
    monkeypatch.setattr(throttle, "BACKOFF_CAP", 0.001)

    attempts = {"n": 0}

    def _flaky(_req):
        attempts["n"] += 1
        if attempts["n"] <= 1:
            raise _FakeTencentExc(
                code="RequestLimitExceeded",
                message="超过每秒频率上限",
            )
        return "ok"

    out = throttle.tencent_call_sync(
        _flaky, None, secret_id="S", service="cam", rate=100.0,
    )
    assert out == "ok"
    assert attempts["n"] == 2


def test_sync_flavour_exhausts_retries_then_re_raises(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(throttle, "BACKOFF_BASE", 0.001)
    monkeypatch.setattr(throttle, "BACKOFF_CAP", 0.001)

    def _always(_req):
        raise _FakeTencentExc(
            code="RequestLimitExceeded",
            message="超过每秒频率上限",
        )

    with pytest.raises(_FakeTencentExc):
        throttle.tencent_call_sync(
            _always, None,
            secret_id="S", service="cam", rate=100.0, max_retries=2,
        )


# ── Defensive: rejects invalid rate ────────────────────────────────


def test_zero_rate_is_rejected() -> None:
    """A rate <= 0 is a config error, not a "no-throttle" toggle."""
    with pytest.raises(ValueError):
        throttle._TokenBucket(rate=0)
    with pytest.raises(ValueError):
        throttle._TokenBucket(rate=-1)


def test_sync_flavour_zero_rate_is_rejected() -> None:
    def _ok(_req):
        return "ok"

    with pytest.raises(ValueError):
        throttle.tencent_call_sync(
            _ok, None, secret_id="S", service="test", rate=0,
        )


# ── Round-30 (2026-05-24) InternalError / ServiceUnavailable retry ──
#
# Sibling-bug coverage with Azure ``_arm_request`` and UCloud ``_call``
# 5xx retry. Pre-fix Tencent SDK ``InternalError`` mid-pagination on
# DescribeInstances stopped pagination + flipped the cron red on the
# very first transient — same self-healing-on-next-tick noise pattern.


@pytest.mark.asyncio
async def test_retries_on_internal_error_then_succeeds(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``InternalError`` (and dotted subcodes) is now retried alongside
    ``RequestLimitExceeded``. Two server-faults then a 200 OK."""
    monkeypatch.setattr(throttle, "BACKOFF_BASE", 0.001)
    monkeypatch.setattr(throttle, "BACKOFF_CAP", 0.001)
    attempts = {"n": 0}

    def _flaky(_req):
        attempts["n"] += 1
        if attempts["n"] <= 2:
            raise _FakeTencentExc(
                code="InternalError",
                message="服务端临时故障",
            )
        return "ok"

    out = await throttle.tencent_call(
        _flaky, None, secret_id="K", service="cvm", rate=100.0,
    )
    assert out == "ok"
    assert attempts["n"] == 3


@pytest.mark.asyncio
async def test_retries_on_internal_error_dotted_subcode(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Service-specific ``InternalError.X`` variants also retry —
    matches Tencent's documented dotted-code taxonomy."""
    monkeypatch.setattr(throttle, "BACKOFF_BASE", 0.001)
    monkeypatch.setattr(throttle, "BACKOFF_CAP", 0.001)
    attempts = {"n": 0}

    def _flaky(_req):
        attempts["n"] += 1
        if attempts["n"] == 1:
            raise _FakeTencentExc(
                code="InternalError.GetInstanceListError",
                message="后端服务异常",
            )
        return "ok"

    out = await throttle.tencent_call(
        _flaky, None, secret_id="K", service="cvm", rate=100.0,
    )
    assert out == "ok"
    assert attempts["n"] == 2


@pytest.mark.asyncio
async def test_retries_on_service_unavailable(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``ServiceUnavailable`` from a backend overload is retryable."""
    monkeypatch.setattr(throttle, "BACKOFF_BASE", 0.001)
    monkeypatch.setattr(throttle, "BACKOFF_CAP", 0.001)
    attempts = {"n": 0}

    def _flaky(_req):
        attempts["n"] += 1
        if attempts["n"] == 1:
            raise _FakeTencentExc(
                code="ServiceUnavailable",
                message="服务暂不可用",
            )
        return "ok"

    out = await throttle.tencent_call(
        _flaky, None, secret_id="K", service="cvm", rate=100.0,
    )
    assert out == "ok"
    assert attempts["n"] == 2


@pytest.mark.asyncio
async def test_auth_failure_still_not_retried_after_widening() -> None:
    """Widening the retry classifier must NOT pull in auth errors —
    a broken SecretKey burning the retry budget would just delay the
    inevitable surface to the admin."""
    attempts = {"n": 0}

    def _auth(_req):
        attempts["n"] += 1
        raise _FakeTencentExc(
            code="AuthFailure.SignatureFailure",
            message="invalid signature",
        )

    with pytest.raises(_FakeTencentExc):
        await throttle.tencent_call(
            _auth, None, secret_id="K", service="cvm", rate=100.0,
        )
    assert attempts["n"] == 1


@pytest.mark.asyncio
async def test_invalid_parameter_still_not_retried_after_widening() -> None:
    """Client-side ``InvalidParameter.X`` is permanent — must not retry."""
    attempts = {"n": 0}

    def _bad(_req):
        attempts["n"] += 1
        raise _FakeTencentExc(
            code="InvalidParameter.InstanceIdMalformed",
            message="bad id",
        )

    with pytest.raises(_FakeTencentExc):
        await throttle.tencent_call(
            _bad, None, secret_id="K", service="cvm", rate=100.0,
        )
    assert attempts["n"] == 1


def test_sync_flavour_retries_on_internal_error(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Sync flavour parity: ``InternalError`` retries via the same
    classifier."""
    monkeypatch.setattr(throttle, "BACKOFF_BASE", 0.001)
    monkeypatch.setattr(throttle, "BACKOFF_CAP", 0.001)
    attempts = {"n": 0}

    def _flaky(_req):
        attempts["n"] += 1
        if attempts["n"] <= 1:
            raise _FakeTencentExc(
                code="InternalError",
                message="临时故障",
            )
        return "ok"

    out = throttle.tencent_call_sync(
        _flaky, None, secret_id="S", service="cvm", rate=100.0,
    )
    assert out == "ok"
    assert attempts["n"] == 2