"""Regression: UCloud RetCode=230 classification (2026-04-24).

UCloud's 230 error code is a generic "parameter error" bucket that
covers both genuine auth failures ("Invalid PublicKey") AND per-region
availability rejections ("Params [Region] not available"). Previously
we treated every 230 as a non-retryable auth error, so a single
unavailable region (e.g. ``afr-nigeria``, ``idn-jakarta``) would
collapse the entire UHost sweep — observed in prod 2026-04-24 where
``cmdb.sync.ucloud-primary.servers`` failed with ``Params [Region]
not available``.

The fix routes the region-not-available case through
``ProviderTransientError`` so ``list_servers``'s per-region loop skips
that region and keeps sweeping the rest. Genuine 230/231/299 auth
failures still surface as ``ProviderAuthError``.
"""
from __future__ import annotations

from unittest.mock import patch

import pytest

from app.domains.cmdb.providers.base import ProviderCredentials
from app.domains.cmdb.providers.exceptions import (
    ProviderAuthError,
    ProviderTransientError,
)
from app.domains.cmdb.providers.ucloud import (
    UCloudProvider,
    _is_region_skip_signal,
)


def _creds(regions: str = "cn-bj2") -> ProviderCredentials:
    return ProviderCredentials(
        provider_id="test-ucloud",
        provider_type="ucloud",
        account_alias="test",
        fields={
            "public_key": "PUB",
            "private_key": "PRIV",
            "regions": regions,
        },
    )


@pytest.mark.asyncio
async def test_call_region_unavailable_raises_transient_not_auth() -> None:
    """RetCode=230 with "Region ... not available" must be transient."""
    provider = UCloudProvider(_creds())

    async def fake_post(*_args, **_kwargs):
        class _Resp:
            status_code = 200
            text = ""
            def json(self):  # noqa: ANN001
                return {
                    "RetCode": 230,
                    "Message": "Params [Region] not available",
                }
            def raise_for_status(self):
                return None
        return _Resp()

    with patch.object(provider._client, "post", side_effect=fake_post):
        with pytest.raises(ProviderTransientError, match="not available"):
            await provider._call("DescribeUHostInstance", {"Region": "afr-nigeria"})


@pytest.mark.asyncio
async def test_call_invalid_key_still_raises_auth() -> None:
    """RetCode=230 with "Invalid PublicKey" must still surface as auth."""
    provider = UCloudProvider(_creds())

    async def fake_post(*_args, **_kwargs):
        class _Resp:
            status_code = 200
            text = ""
            def json(self):  # noqa: ANN001
                return {"RetCode": 230, "Message": "Invalid PublicKey"}
            def raise_for_status(self):
                return None
        return _Resp()

    with patch.object(provider._client, "post", side_effect=fake_post):
        with pytest.raises(ProviderAuthError):
            await provider._call("DescribeUHostInstance", {"Region": "cn-bj2"})


@pytest.mark.asyncio
async def test_list_servers_skips_unavailable_region_and_sweeps_rest() -> None:
    """Unavailable region must be skipped, remaining regions still fetched."""
    provider = UCloudProvider(_creds(regions="cn-bj2,afr-nigeria"))

    async def fake_call(action, params, **_kw):  # noqa: ANN001
        region = params.get("Region")
        if region == "afr-nigeria":
            raise ProviderTransientError(
                "UCloud DescribeUHostInstance RetCode=230: Params [Region] not available",
            )
        if region == "cn-bj2":
            return {"UHostSet": [], "TotalCount": 0}
        raise AssertionError(f"unexpected region {region!r}")

    with patch.object(provider, "_call", side_effect=fake_call):
        rows = await provider.list_servers()

    # cn-bj2 returned empty, afr-nigeria was skipped — no exception propagates.
    assert rows == []


# ── Regression: ULB RetCode=150 "Service unavailable" (prod 2026-04-24 15:15) ──
#
# ULB is a separate UCloud product from UHost; even tenants with UHost
# enabled in a region may lack ULB there. UCloud returns RetCode=150
# "Service unavailable" for such regions, which used to propagate out
# of ``list_load_balancers`` and fail the whole LB sync. Now
# ``_is_region_skip_signal`` recognizes that signature and the
# per-region loop skips and continues.


def test_is_region_skip_signal_matches_ulb_150() -> None:
    assert _is_region_skip_signal(
        "UCloud DescribeULB RetCode=150: Service unavailable",
    ) is True


def test_is_region_skip_signal_matches_uhost_230() -> None:
    assert _is_region_skip_signal(
        "UCloud DescribeUHostInstance RetCode=230: Params [Region] not available",
    ) is True


def test_is_region_skip_signal_does_not_match_real_failures() -> None:
    # A genuine service outage is RetCode=50000-ish with a different message;
    # we must NOT silently skip those.
    assert _is_region_skip_signal("UCloud DescribeULB RetCode=50000: internal error") is False
    # Plain HTTP error surface (no RetCode= prefix) — don't misinterpret.
    assert _is_region_skip_signal("UCloud HTTP error: ConnectionReset") is False
    assert _is_region_skip_signal("") is False


@pytest.mark.asyncio
async def test_list_load_balancers_skips_service_unavailable_region() -> None:
    """ULB RetCode=150 on one region must skip + continue, not fail all."""
    provider = UCloudProvider(_creds(regions="cn-bj2,afr-nigeria"))

    async def fake_call(action, params, **_kw):  # noqa: ANN001
        region = params.get("Region")
        if region == "afr-nigeria":
            raise ProviderTransientError(
                "UCloud DescribeULB RetCode=150: Service unavailable",
            )
        if region == "cn-bj2":
            return {"DataSet": [], "TotalCount": 0}
        raise AssertionError(f"unexpected region {region!r}")

    with patch.object(provider, "_call", side_effect=fake_call):
        rows = await provider.list_load_balancers()

    assert rows == []


@pytest.mark.asyncio
async def test_list_load_balancers_still_propagates_real_transient() -> None:
    """Non-skip transient (e.g. 5xx) must still fail the sweep so the
    scheduled-job handler sees a real error and retries on next tick."""
    provider = UCloudProvider(_creds(regions="cn-bj2"))

    async def fake_call(*_args, **_kwargs):
        raise ProviderTransientError("UCloud HTTP 503: server busy")

    with patch.object(provider, "_call", side_effect=fake_call):
        with pytest.raises(ProviderTransientError, match="503"):
            await provider.list_load_balancers()


# ── Transport-layer retry on UCloud _call (2026-05-24) ──────────────
#
# Sibling-bug coverage with Azure object_storage / Aliyun OSS. UCloud
# ``_call`` pre-fix wrapped ``httpx.HTTPError`` into ProviderTransientError
# with NO retry — a single mid-flight ConnectError dropped the whole
# capability for the cron tick. Aliyun's RPC ``_call`` already had a
# 3-attempt retry budget; parity fix here.


@pytest.mark.asyncio
async def test_call_retries_on_transport_error_then_succeeds(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A single ConnectError on the first POST retries and succeeds on
    the second attempt — same scope as the Azure helper but adapted to
    UCloud's local 3-attempt loop in ``_call``."""
    import httpx

    from app.domains.cmdb.providers import ucloud as _ucloud_mod

    provider = UCloudProvider(_creds())

    class _OK:
        status_code = 200
        text = ""

        def json(self):  # noqa: ANN001
            return {"RetCode": 0, "DataSet": []}

        def raise_for_status(self):
            return None

    call_count = {"n": 0}

    async def fake_post(*_args, **_kwargs):
        call_count["n"] += 1
        if call_count["n"] == 1:
            raise httpx.ConnectError("simulated flake")
        return _OK()

    async def _noop_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr(_ucloud_mod.asyncio, "sleep", _noop_sleep)

    with patch.object(provider._client, "post", side_effect=fake_post):
        data = await provider._call("DescribeUHostInstance", {"Region": "cn-bj2"})

    assert data == {"RetCode": 0, "DataSet": []}
    assert call_count["n"] == 2


@pytest.mark.asyncio
async def test_call_transport_retry_budget_3_then_raises_transient(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """After 3 transport-error attempts, ``_call`` surfaces a
    ProviderTransientError with the LAST exception's redacted text.
    Caller logic still treats this as a normal transient and the cron
    tick re-runs."""
    import httpx

    from app.domains.cmdb.providers import ucloud as _ucloud_mod

    provider = UCloudProvider(_creds())

    call_count = {"n": 0}

    async def fake_post(*_args, **_kwargs):
        call_count["n"] += 1
        raise httpx.ReadTimeout("permanent")

    async def _noop_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr(_ucloud_mod.asyncio, "sleep", _noop_sleep)

    with patch.object(provider._client, "post", side_effect=fake_post):
        with pytest.raises(ProviderTransientError, match="UCloud HTTP error"):
            await provider._call("DescribeUHostInstance", {"Region": "cn-bj2"})

    # All 3 attempts burned.
    assert call_count["n"] == 3


# ── Round-30 (2026-05-24) UCloud 5xx + 429 retry ────────────────────


class _UCloudStubResp:
    """Stub httpx.Response carrying status + JSON body for retry tests."""

    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):
        return self._payload

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

    async def aclose(self):
        # Round-30 security-review fix: ``_call`` now calls ``aclose()``
        # on intermediate 5xx / 429 responses to release the connection
        # back to the httpx pool before the retry sleep. The stub
        # records the call so tests can assert pool-hygiene if needed.
        self.aclosed = True


@pytest.mark.asyncio
async def test_call_retries_on_5xx_then_succeeds(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """UCloud ``_call`` now retries 5xx within the same 3-attempt
    budget as transport errors — pre-fix a single 5xx on a region's
    DescribeUHostInstance dropped the entire region's instances.
    Parity with Aliyun ``_call``'s existing 5xx retry."""
    from app.domains.cmdb.providers import ucloud as _ucloud_mod

    provider = UCloudProvider(_creds())
    call_count = {"n": 0}

    async def fake_post(*_args, **_kwargs):
        call_count["n"] += 1
        if call_count["n"] < 3:
            return _UCloudStubResp(502, text="bad gateway")
        return _UCloudStubResp(200, {"RetCode": 0, "DataSet": []})

    async def _noop_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr(_ucloud_mod.asyncio, "sleep", _noop_sleep)
    with patch.object(provider._client, "post", side_effect=fake_post):
        data = await provider._call(
            "DescribeUHostInstance", {"Region": "cn-bj2"},
        )
    assert data == {"RetCode": 0, "DataSet": []}
    assert call_count["n"] == 3


@pytest.mark.asyncio
async def test_call_5xx_budget_exhausted_raises_transient(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """All 3 attempts return 503 → ProviderTransientError surfaced with
    "(after 3 attempts)" in the message so admins can tell budget-
    exhaustion from a first-touch failure."""
    from app.domains.cmdb.providers import ucloud as _ucloud_mod

    provider = UCloudProvider(_creds())
    call_count = {"n": 0}

    async def fake_post(*_args, **_kwargs):
        call_count["n"] += 1
        return _UCloudStubResp(503, text="busy")

    async def _noop_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr(_ucloud_mod.asyncio, "sleep", _noop_sleep)
    with patch.object(provider._client, "post", side_effect=fake_post):
        with pytest.raises(ProviderTransientError) as exc_info:
            await provider._call(
                "DescribeUHostInstance", {"Region": "cn-bj2"},
            )
    assert call_count["n"] == 3
    msg = str(exc_info.value)
    assert "503" in msg
    assert "after 3 attempts" in msg


@pytest.mark.asyncio
async def test_call_429_within_budget_retries_then_succeeds(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """429 within budget is now retried (was raised immediately)."""
    from app.domains.cmdb.providers import ucloud as _ucloud_mod

    provider = UCloudProvider(_creds())
    call_count = {"n": 0}

    async def fake_post(*_args, **_kwargs):
        call_count["n"] += 1
        if call_count["n"] == 1:
            return _UCloudStubResp(429, text="throttle")
        return _UCloudStubResp(200, {"RetCode": 0, "DataSet": []})

    async def _noop_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr(_ucloud_mod.asyncio, "sleep", _noop_sleep)
    with patch.object(provider._client, "post", side_effect=fake_post):
        data = await provider._call(
            "DescribeUHostInstance", {"Region": "cn-bj2"},
        )
    assert data == {"RetCode": 0, "DataSet": []}
    assert call_count["n"] == 2


# ── Regression: body-level transient RetCode retry (prod 2026-05-27) ──
#
# UCloud returns HTTP 200 + a non-zero RetCode for transient server
# faults too. RetCode=8498 "Call third api service timeout, please
# retry" on the jpn-tky UHost sweep failed the whole ``servers``
# capability and paged — because the transport/5xx retry budget never
# saw a body-level RetCode (it was classified + raised after the loop
# broke on the 2xx status). The fix folds RetCode classification into
# the retry loop so a transient RetCode is retried in-place.


@pytest.mark.asyncio
async def test_call_transient_retcode_8498_retries_then_succeeds(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """RetCode=8498 (gateway timeout) is retried in-place and recovers."""
    from app.domains.cmdb.providers import ucloud as _ucloud_mod

    provider = UCloudProvider(_creds())
    call_count = {"n": 0}

    async def fake_post(*_args, **_kwargs):
        call_count["n"] += 1
        if call_count["n"] == 1:
            return _UCloudStubResp(200, {
                "RetCode": 8498,
                "Message": (
                    "Call third api service timeout, please retry or "
                    "contact technical support"
                ),
            })
        return _UCloudStubResp(200, {"RetCode": 0, "UHostSet": [], "TotalCount": 0})

    async def _noop_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr(_ucloud_mod.asyncio, "sleep", _noop_sleep)
    with patch.object(provider._client, "post", side_effect=fake_post):
        data = await provider._call(
            "DescribeUHostInstance", {"Region": "jpn-tky"},
        )
    assert data["RetCode"] == 0
    assert call_count["n"] == 2  # one transient, one recovery


@pytest.mark.asyncio
async def test_call_transient_retcode_8498_exhausts_then_raises(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A persistent 8498 exhausts the budget then raises transient (so
    the per-region caller still records its partial_failure leg)."""
    from app.domains.cmdb.providers import ucloud as _ucloud_mod

    provider = UCloudProvider(_creds())
    call_count = {"n": 0}

    async def fake_post(*_args, **_kwargs):
        call_count["n"] += 1
        return _UCloudStubResp(200, {
            "RetCode": 8498,
            "Message": "Call third api service timeout, please retry",
        })

    async def _noop_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr(_ucloud_mod.asyncio, "sleep", _noop_sleep)
    with patch.object(provider._client, "post", side_effect=fake_post):
        with pytest.raises(ProviderTransientError, match="8498"):
            await provider._call(
                "DescribeUHostInstance", {"Region": "jpn-tky"},
            )
    assert call_count["n"] == 3  # full 3-attempt budget burned


@pytest.mark.asyncio
async def test_call_transient_message_hint_retries(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A non-8498 code whose Message carries a timeout hint is also
    retried — the transient-timeout class isn't limited to one code."""
    from app.domains.cmdb.providers import ucloud as _ucloud_mod

    provider = UCloudProvider(_creds())
    call_count = {"n": 0}

    async def fake_post(*_args, **_kwargs):
        call_count["n"] += 1
        if call_count["n"] == 1:
            return _UCloudStubResp(200, {
                "RetCode": 9999,
                "Message": "backend request time out, please try again",
            })
        return _UCloudStubResp(200, {"RetCode": 0, "UHostSet": []})

    async def _noop_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr(_ucloud_mod.asyncio, "sleep", _noop_sleep)
    with patch.object(provider._client, "post", side_effect=fake_post):
        data = await provider._call("DescribeUHostInstance", {"Region": "cn-bj2"})
    assert data["RetCode"] == 0
    assert call_count["n"] == 2


@pytest.mark.asyncio
async def test_call_permanent_retcode_not_retried(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A deterministic client error (no timeout hint) must NOT burn the
    retry budget — it raises on the first attempt."""
    from app.domains.cmdb.providers import ucloud as _ucloud_mod

    provider = UCloudProvider(_creds())
    call_count = {"n": 0}

    async def fake_post(*_args, **_kwargs):
        call_count["n"] += 1
        return _UCloudStubResp(200, {
            "RetCode": 171,
            "Message": "Action [DescribeUHostInstanceXX] not supported",
        })

    async def _noop_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr(_ucloud_mod.asyncio, "sleep", _noop_sleep)
    with patch.object(provider._client, "post", side_effect=fake_post):
        with pytest.raises(ProviderTransientError, match="171"):
            await provider._call("DescribeUHostInstance", {"Region": "cn-bj2"})
    assert call_count["n"] == 1  # no retry for a non-transient code


@pytest.mark.asyncio
async def test_call_region_skip_code_150_not_retried(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """RetCode=150 'Service unavailable' (region-not-provisioned) is a
    fast per-region skip — it must NOT burn three retries per region."""
    from app.domains.cmdb.providers import ucloud as _ucloud_mod

    provider = UCloudProvider(_creds())
    call_count = {"n": 0}

    async def fake_post(*_args, **_kwargs):
        call_count["n"] += 1
        return _UCloudStubResp(200, {
            "RetCode": 150,
            "Message": "Service unavailable",
        })

    async def _noop_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr(_ucloud_mod.asyncio, "sleep", _noop_sleep)
    with patch.object(provider._client, "post", side_effect=fake_post):
        with pytest.raises(ProviderTransientError):
            await provider._call("DescribeULB", {"Region": "afr-nigeria"})
    assert call_count["n"] == 1  # region-skip is immediate, no retry


@pytest.mark.asyncio
async def test_list_servers_recovers_from_transient_and_sweeps_region(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """End-to-end: a single 8498 timeout on jpn-tky now recovers on
    retry, so the region's servers are swept instead of the whole
    capability failing + paging (the prod 2026-05-27 incident)."""
    from app.domains.cmdb.providers import ucloud as _ucloud_mod

    provider = UCloudProvider(_creds(regions="jpn-tky"))
    call_count = {"n": 0}

    async def fake_post(*_args, **_kwargs):
        call_count["n"] += 1
        if call_count["n"] == 1:
            return _UCloudStubResp(200, {
                "RetCode": 8498,
                "Message": "Call third api service timeout, please retry",
            })
        return _UCloudStubResp(200, {
            "RetCode": 0,
            "UHostSet": [{"UHostId": "uhost-jp1", "Name": "jp1", "State": "Running"}],
            "TotalCount": 1,
        })

    async def _noop_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr(_ucloud_mod.asyncio, "sleep", _noop_sleep)
    with patch.object(provider._client, "post", side_effect=fake_post):
        rows = await provider.list_servers()

    assert [s.instance_id for s in rows] == ["uhost-jp1"]
    assert not provider.partial_failures  # no leg recorded — fully recovered
    assert call_count["n"] == 2


# ---------------------------------------------------------------------------
# Multi-project config (2026-07-14 — DLP-test drift root cause)
#
# The provider has read ``project_ids`` (CSV) since PR#558, but the field
# was never in ``UCloudDriver.config_schema`` — so the admin UI had no way
# to set it and prod ran with only the default project synced, leaving
# resources in every other project (incl. DLP-test in org-eneelp) invisible
# to Helix. These tests pin BOTH the schema exposure and the parse
# precedence so the field cannot silently disappear again.
# ---------------------------------------------------------------------------
def _creds_with(**extra: str) -> ProviderCredentials:
    fields = {"public_key": "PUB", "private_key": "PRIV", "regions": "cn-bj2"}
    fields.update(extra)
    return ProviderCredentials(
        provider_id="test-ucloud",
        provider_type="ucloud",
        account_alias="test",
        fields=fields,
    )


def test_config_schema_exposes_project_ids() -> None:
    from app.integrations.drivers.ucloud_cloud import UCloudDriver

    keys = {f.key for f in UCloudDriver.config_schema}
    # Both the multi-project field and the legacy single field are present.
    assert "project_ids" in keys
    assert "project_id" in keys


def test_project_ids_plural_iterates_every_project() -> None:
    provider = UCloudProvider(_creds_with(project_ids="org-aaa,org-bbb, org-ccc"))
    # Whitespace trimmed, order preserved, all three iterated.
    assert provider._project_ids == ["org-aaa", "org-bbb", "org-ccc"]


def test_project_ids_plural_supersedes_singular() -> None:
    provider = UCloudProvider(
        _creds_with(project_ids="org-plural", project_id="org-single")
    )
    assert provider._project_ids == ["org-plural"]


def test_project_id_singular_used_when_plural_absent() -> None:
    provider = UCloudProvider(_creds_with(project_id="org-single"))
    assert provider._project_ids == ["org-single"]


def test_no_project_config_falls_back_to_default_sentinel() -> None:
    # Neither field set → single ``None`` sentinel → _call omits ProjectId,
    # deferring to the AK's default project (pre-multi-project behavior).
    provider = UCloudProvider(_creds_with())
    assert provider._project_ids == [None]