"""Unit tests for the GCP CMDB provider.

Same mocking approach as the GCP driver tests: patch ``_svc`` on the
provider instance to return stubbed Google API clients.
"""
from __future__ import annotations

import json
from typing import Any
from unittest.mock import MagicMock, patch

import pytest

from app.domains.cmdb.providers.base import (
    ProviderCapability,
    ProviderCredentials,
)
from app.domains.cmdb.providers.gcp import GcpProvider

_SA_JSON = json.dumps({
    "type": "service_account",
    "project_id": "test-project",
    "private_key_id": "x",
    "private_key": "-----BEGIN PRIVATE KEY-----\nfake\n-----END PRIVATE KEY-----\n",
    "client_email": "sa@test-project.iam.gserviceaccount.com",
    "client_id": "0",
    "auth_uri": "https://accounts.google.com/o/oauth2/auth",
    "token_uri": "https://oauth2.googleapis.com/token",
})


@pytest.fixture(autouse=True)
def _mock_sa_credentials():
    """Skip real RSA-key parse during tests — we're not exercising google-auth.

    Applies to every test in this module. The ``test_constructor_rejects_*``
    cases override this back to the real behaviour.
    """
    with patch(
        "app.domains.cmdb.providers.gcp.service_account.Credentials.from_service_account_info",
        return_value=MagicMock(),
    ) as mock:
        yield mock


def _provider() -> GcpProvider:
    creds = ProviderCredentials(
        provider_id="test-gcp",
        provider_type="gcp",
        account_alias="test",
        fields={
            "service_account_json": _SA_JSON,
            "organization_id": "478164292948",
        },
    )
    return GcpProvider(creds)


def _with_svc(provider: GcpProvider, svcs: dict[tuple[str, str], MagicMock]):
    """Patch the provider instance's _svc to return the right mock."""
    crm = svcs.get(("cloudresourcemanager", "v3"))
    if crm is not None:
        projects = crm.projects.return_value

        def _list_projects(
            *, parent: str, pageToken: str | None = None,
            showDeleted: bool = False,
        ) -> MagicMock:
            del showDeleted
            legacy_request = projects.search(
                query="state:ACTIVE", pageToken=pageToken,
            )
            request = MagicMock()

            def _execute(**_kwargs: Any) -> dict[str, Any]:
                response = legacy_request.execute()
                for project in response.get("projects") or []:
                    project.setdefault("parent", parent)
                return response

            request.execute.side_effect = _execute
            return request

        projects.list.side_effect = _list_projects
        crm.folders.return_value.list.return_value.execute.return_value = {
            "folders": [],
        }

    def _factory(service: str, version: str):
        key = (service, version)
        if key not in svcs:
            raise AssertionError(f"Unexpected _svc({service!r}, {version!r})")
        return svcs[key]
    return patch.object(provider, "_svc", side_effect=_factory)


def test_supported_capabilities_match_impl() -> None:
    caps = GcpProvider.supported_capabilities()
    assert ProviderCapability.SERVERS in caps
    assert ProviderCapability.DNS_RECORDS in caps
    assert ProviderCapability.OBJECT_STORAGE in caps
    # The ones we deliberately skip.
    assert ProviderCapability.DOMAINS not in caps
    assert ProviderCapability.LOAD_BALANCERS not in caps
    assert ProviderCapability.CDN not in caps


def test_constructor_rejects_missing_json() -> None:
    from app.domains.cmdb.providers.exceptions import ProviderAuthError
    creds = ProviderCredentials(
        provider_id="x", provider_type="gcp", account_alias="t", fields={},
    )
    with pytest.raises(ProviderAuthError):
        GcpProvider(creds)


def test_constructor_rejects_malformed_json() -> None:
    from app.domains.cmdb.providers.exceptions import ProviderAuthError
    creds = ProviderCredentials(
        provider_id="x", provider_type="gcp", account_alias="t",
        fields={"service_account_json": "not json"},
    )
    with pytest.raises(ProviderAuthError):
        GcpProvider(creds)


# ── list_servers ────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_list_servers_parses_aggregated_response() -> None:
    p = _provider()
    crm = MagicMock()
    # Single project visible.
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-a"}],
    }
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.return_value = {
        "items": {
            "zones/us-central1-a": {
                "instances": [
                    {
                        "id": "1234567890",
                        "name": "vm-web-1",
                        "status": "RUNNING",
                        "machineType": "https://.../zones/us-central1-a/machineTypes/n2-standard-4",
                        "networkInterfaces": [
                            {
                                "networkIP": "10.0.0.5",
                                "accessConfigs": [{"natIP": "35.1.2.3"}],
                            },
                        ],
                        "labels": {"env": "prod", "team": "web"},
                        "disks": [
                            {"boot": True, "licenses": [
                                "https://.../debian-cloud/licenses/debian-11",
                            ], "source": "projects/x/disks/disk-1"},
                        ],
                    },
                ],
            },
            "zones/us-east1-b": {
                # No instances — aggregated responses include empty zones.
                "warning": {"code": "NO_RESULTS_ON_PAGE"},
            },
        },
    }
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert len(servers) == 1
    s = servers[0]
    assert s.name == "vm-web-1"
    assert s.status == "running"
    assert s.zone == "us-central1-a"
    assert s.region == "us-central1"
    assert s.vcpu == 4
    assert s.public_ips == ["35.1.2.3"]
    assert s.private_ips == ["10.0.0.5"]
    assert s.os_type == "linux"
    assert s.tags["env"] == "prod"
    assert s.raw["_project_id"] == "proj-a"


@pytest.mark.asyncio
async def test_list_servers_records_zone_warning_as_partial_failure() -> None:
    """A zone returning a non-benign ``warning`` block under
    ``returnPartialSuccess=True`` must produce a ``partial_failures``
    leg so the failed zone is visible to the sync-history admin.

    Regression for production incident 2026-05-17: GCP servers cron
    blocked by sweep guard for 8+ hours (provider returned ~870 vs
    baseline ~938) because the driver only iterated ``zone_block[
    "instances"]`` and ignored ``zone_block["warning"]`` — silent
    instance loss with no leg recorded to identify the failing zone.
    """
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-mixed"}],
    }
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.return_value = {
        "items": {
            "zones/us-central1-a": {
                "instances": [{
                    "id": "1", "name": "vm-1", "status": "RUNNING",
                    "machineType": ".../machineTypes/e2-medium",
                    "networkInterfaces": [],
                    "disks": [{"boot": True, "licenses": []}],
                }],
            },
            # Benign — empty zone signal, must NOT record a leg.
            "zones/us-east1-b": {"warning": {"code": "NO_RESULTS_ON_PAGE"}},
            # Non-benign — zone unreachable, instances silently dropped.
            "zones/asia-east1-c": {
                "warning": {
                    "code": "UNREACHABLE",
                    "message": "Failed to reach zone asia-east1-c.",
                },
            },
            # Non-benign — quota/permission/transient mix.
            "zones/europe-west1-b": {
                "warning": {
                    "code": "ZONE_NOT_AVAILABLE",
                    "message": "Zone is undergoing maintenance.",
                },
            },
        },
    }
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert len(servers) == 1
    legs = [
        leg for leg in p.partial_failures
        if leg.get("capability") == "servers"
    ]
    reasons = sorted(leg["reason"] for leg in legs)
    assert reasons == [
        "aggregatedList.zone_warning_UNREACHABLE",
        "aggregatedList.zone_warning_ZONE_NOT_AVAILABLE",
    ]
    zone_legs = sorted(leg["leg"] for leg in legs)
    # Leg shape ``project:<id>:zone:<name>`` mirrors the sibling DNS leg
    # at gcp.py::_list_dns_records_for_project so failure-notifier dedup
    # keys are consistent across capabilities; ``zones/`` URL prefix
    # stripped — the zone name alone is the meaningful identifier.
    assert zone_legs == [
        "project:proj-mixed:zone:asia-east1-c",
        "project:proj-mixed:zone:europe-west1-b",
    ]


@pytest.mark.asyncio
async def test_list_servers_zone_warnings_plural_form_also_recorded() -> None:
    """Defensive handling of the rare ``warnings`` (list) shape — some
    GCP API responses use this when a single zone has multiple
    sub-errors. Without this branch the driver re-introduces the same
    silent-drop blind spot the singular-form fix was meant to close.
    """
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-x"}],
    }
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.return_value = {
        "items": {
            "zones/us-central1-a": {
                "warnings": [
                    {"code": "NO_RESULTS_ON_PAGE"},     # benign — ignored
                    {"code": "DEPRECATED_RESOURCE_USED"},  # surfaced
                ],
            },
        },
    }
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert servers == []
    legs = [
        leg for leg in p.partial_failures
        if leg.get("capability") == "servers"
    ]
    assert len(legs) == 1
    assert legs[0]["reason"] == "aggregatedList.zone_warning_DEPRECATED_RESOURCE_USED"


@pytest.mark.asyncio
async def test_list_servers_zone_with_instances_and_warning_records_both() -> None:
    """A zone block can legitimately carry BOTH ``instances`` AND a
    non-benign ``warning`` simultaneously (partial success at the zone
    level). The driver must collect every instance AND surface the
    warning leg — silently picking one over the other would either
    drop visible servers or hide a real per-zone problem.
    """
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-both"}],
    }
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.return_value = {
        "items": {
            "zones/us-central1-a": {
                "instances": [{
                    "id": "1", "name": "vm-survivor", "status": "RUNNING",
                    "machineType": ".../machineTypes/e2-medium",
                    "networkInterfaces": [],
                    "disks": [{"boot": True, "licenses": []}],
                }],
                "warning": {
                    "code": "DEPRECATED_RESOURCE_USED",
                    "message": "Instance uses a deprecated machine type.",
                },
            },
        },
    }
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert len(servers) == 1
    assert servers[0].name == "vm-survivor"
    server_legs = [
        leg for leg in p.partial_failures
        if leg.get("capability") == "servers"
    ]
    assert len(server_legs) == 1
    assert server_legs[0]["reason"] == (
        "aggregatedList.zone_warning_DEPRECATED_RESOURCE_USED"
    )


@pytest.mark.asyncio
async def test_list_servers_zone_warning_deduped_across_pagination() -> None:
    """``aggregatedList`` paginates; a problem zone persisting across
    multiple pages must produce ONE leg, not N. Pre-dedup-fix every
    page emitted a duplicate leg into ``partial_failures``, polluting
    the ``cmdb_sync_jobs.capability_results`` JSON column and the
    sync-history admin UI on >500-instance projects.
    """
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-paginated"}],
    }
    compute = MagicMock()
    page_responses = [
        # Page 1 — has nextPageToken; bad zone appears.
        {
            "items": {
                "zones/asia-east1-c": {
                    "warning": {"code": "UNREACHABLE", "message": "..."},
                },
                "zones/us-central1-a": {
                    "instances": [{
                        "id": "1", "name": "vm-a", "status": "RUNNING",
                        "machineType": ".../machineTypes/e2-medium",
                        "networkInterfaces": [],
                        "disks": [{"boot": True, "licenses": []}],
                    }],
                },
            },
            "nextPageToken": "page2",
        },
        # Page 2 — same bad zone still warning (persistent); different
        # zone has instances.
        {
            "items": {
                "zones/asia-east1-c": {
                    "warning": {"code": "UNREACHABLE", "message": "..."},
                },
                "zones/us-east1-b": {
                    "instances": [{
                        "id": "2", "name": "vm-b", "status": "RUNNING",
                        "machineType": ".../machineTypes/e2-medium",
                        "networkInterfaces": [],
                        "disks": [{"boot": True, "licenses": []}],
                    }],
                },
            },
        },
    ]
    page_idx = {"i": 0}

    def _aggregated_list(project: str, **kwargs):
        mock = MagicMock()
        mock.execute.return_value = page_responses[page_idx["i"]]
        page_idx["i"] += 1
        return mock

    compute.instances.return_value.aggregatedList.side_effect = _aggregated_list
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert len(servers) == 2
    server_legs = [
        leg for leg in p.partial_failures
        if leg.get("capability") == "servers"
    ]
    assert len(server_legs) == 1, (
        f"Expected 1 deduped leg for persistent bad zone across pages, "
        f"got {len(server_legs)}: {server_legs}"
    )
    assert server_legs[0]["leg"] == "project:proj-paginated:zone:asia-east1-c"
    assert server_legs[0]["reason"] == "aggregatedList.zone_warning_UNREACHABLE"


@pytest.mark.asyncio
async def test_list_servers_sanitizes_zone_warning_code() -> None:
    """Cloud-controlled warning ``code`` strings are interpolated into
    the ``reason`` field, which is logged via
    ``logger.warning("...%s...", reason)`` downstream. A malicious /
    malformed code with embedded newlines would forge log lines (log
    injection). The driver strips to ``[A-Z0-9_]`` defensively.
    """
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-evil"}],
    }
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.return_value = {
        "items": {
            "zones/us-central1-a": {
                # Newline-injected code — would forge a separate log line.
                "warning": {
                    "code": "UNREACHABLE\nALERT: false_admin_action_logged",
                    "message": "x",
                },
            },
        },
    }
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            await p.list_servers()
    finally:
        await p.close()
    server_legs = [
        leg for leg in p.partial_failures
        if leg.get("capability") == "servers"
    ]
    assert len(server_legs) == 1
    reason = server_legs[0]["reason"]
    assert "\n" not in reason
    assert "false_admin_action_logged" not in reason
    # The sanitized version keeps the canonical-looking prefix so admins
    # can still triage; ``[^A-Z0-9_]`` strips both newline and lowercase
    # ``alert: false_admin_action_logged`` text.
    assert reason.startswith("aggregatedList.zone_warning_UNREACHABLE")


@pytest.mark.asyncio
async def test_list_servers_benign_zone_warning_does_not_record_leg() -> None:
    """``NO_RESULTS_ON_PAGE`` (zone has no instances) must stay silent —
    every empty zone in GCP carries this warning and surfacing each
    one would flood ``partial_failures`` with ~30 legs per project per
    tick across PingCAP's project tree.
    """
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-empty"}],
    }
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.return_value = {
        "items": {
            "zones/us-central1-a": {"warning": {"code": "NO_RESULTS_ON_PAGE"}},
            "zones/us-east1-b": {"warning": {"code": "NO_RESULTS_FOR_LIST"}},
        },
    }
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert servers == []
    server_legs = [
        leg for leg in p.partial_failures
        if leg.get("capability") == "servers"
    ]
    assert server_legs == []


@pytest.mark.asyncio
async def test_list_servers_403_on_one_project_is_skipped() -> None:
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [
            {"projectId": "proj-ok"},
            {"projectId": "proj-denied"},
        ],
    }
    compute = MagicMock()
    ok_resp = {"items": {
        "zones/us-central1-a": {
            "instances": [{
                "id": "1", "name": "vm", "status": "RUNNING",
                "machineType": ".../machineTypes/e2-medium",
                "networkInterfaces": [],
                "disks": [{"boot": True, "licenses": []}],
            }],
        },
    }}
    denied_err = HttpError(
        resp=MagicMock(status=403),
        content=b'{"error":{"message":"Compute API disabled"}}',
    )
    # First call (proj-ok) returns ok_resp, second (proj-denied) raises 403.
    # aggregatedList is called with project=..., so we use side_effect
    # on the method call itself.
    calls = []
    def _aggregated_list(project: str, **kwargs):
        calls.append(project)
        mock = MagicMock()
        if project == "proj-ok":
            mock.execute.return_value = ok_resp
        else:
            mock.execute.side_effect = denied_err
        return mock
    compute.instances.return_value.aggregatedList.side_effect = _aggregated_list
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    # Exactly one server — from proj-ok; proj-denied swallowed.
    assert len(servers) == 1
    assert sorted(calls) == ["proj-denied", "proj-ok"]


# ── list_dns_records ────────────────────────────────────────────


@pytest.mark.asyncio
async def test_list_dns_records_fans_rrdata_into_rows() -> None:
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-a"}],
    }
    dns = MagicMock()
    dns.managedZones.return_value.list.return_value.execute.return_value = {
        "managedZones": [
            {"name": "example-com", "dnsName": "example.com.",
             "visibility": "public"},
        ],
    }
    dns.resourceRecordSets.return_value.list.return_value.execute.return_value = {
        "rrsets": [
            {
                "name": "www.example.com.",
                "type": "A",
                "ttl": 300,
                "rrdatas": ["1.2.3.4", "5.6.7.8"],
            },
            {
                "name": "example.com.",
                "type": "MX",
                "ttl": 3600,
                "rrdatas": ["10 mail.example.com."],
            },
        ],
    }
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("dns", "v1"): dns,
        }):
            records = await p.list_dns_records()
    finally:
        await p.close()
    # Two rrdata values on the A record → two rows.
    assert len(records) == 3
    a_rows = [r for r in records if r.record_type == "A"]
    assert {r.value for r in a_rows} == {"1.2.3.4", "5.6.7.8"}
    for r in records:
        assert r.zone_name == "example.com."
        assert r.status == "enable"


# ── list_object_storage ─────────────────────────────────────────


@pytest.mark.asyncio
async def test_list_object_storage_normalizes_access_type() -> None:
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-a"}],
    }
    storage = MagicMock()
    storage.buckets.return_value.list.return_value.execute.return_value = {
        "items": [
            {
                "name": "public-website",
                "location": "US",
                "storageClass": "STANDARD",
                # No ubla → effectively public-accessible by legacy ACL.
                "iamConfiguration": {},
            },
            {
                "name": "ubla-locked",
                "location": "ASIA-NORTHEAST1",
                "storageClass": "STANDARD",
                "iamConfiguration": {
                    "uniformBucketLevelAccess": {"enabled": True},
                    "publicAccessPrevention": "enforced",
                },
                "owner": {"entity": "project-owners-123"},
            },
        ],
    }
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("storage", "v1"): storage,
        }):
            buckets = await p.list_object_storage()
    finally:
        await p.close()
    assert len(buckets) == 2
    assert buckets[0]["bucket_name"] == "public-website"
    assert buckets[0]["access_type"] == "public"
    assert buckets[1]["bucket_name"] == "ubla-locked"
    assert buckets[1]["access_type"] == "private"
    assert buckets[1]["owner_raw"] == "project-owners-123"


# ── API-not-enabled (SERVICE_DISABLED / accessNotConfigured) ─────
#
# Production incident 2026-05-17: a 500-employee org's GCP servers
# cron stayed red on every tick because 474 projects in the org
# legitimately have no Compute Engine API enabled (BigQuery / Storage /
# AppEngine / Firebase workloads). The driver pre-fix bucketed those
# 403s into ``partial_failures`` alongside real permission denials, so
# every cron run produced "474 × http_403" legs and paged IT admin.
#
# Fix: parse the GCP structured error body's
# ``error.details[*].reason`` or ``error.errors[*].reason`` and route
# SERVICE_DISABLED / accessNotConfigured 403s to ``skipped_legs`` so
# the capability stays green while remaining visible to admins.


def test_classify_gcp_api_disabled_modern_shape() -> None:
    """Current Google error shape carries ``details[*].reason='SERVICE_DISABLED''``
    with ``metadata.service`` holding the canonical service name."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_api_disabled

    body = {
        "error": {
            "code": 403,
            "message": (
                "Compute Engine API has not been used in project "
                "abiding-window-424223-t9 before or it is disabled."
            ),
            "status": "PERMISSION_DENIED",
            "details": [
                {
                    "@type": "type.googleapis.com/google.rpc.ErrorInfo",
                    "reason": "SERVICE_DISABLED",
                    "domain": "googleapis.com",
                    "metadata": {
                        "service": "compute.googleapis.com",
                        "consumer": "projects/abiding-window-424223-t9",
                    },
                },
            ],
        },
    }
    exc = HttpError(
        resp=MagicMock(status=403),
        content=json.dumps(body).encode("utf-8"),
    )
    assert _classify_gcp_api_disabled(exc) == "compute.googleapis.com"


def test_classify_gcp_api_disabled_legacy_shape() -> None:
    """Legacy v1-style error block carries ``errors[*].reason='accessNotConfigured''``
    without a service name attached. Return the literal ``"unknown"``."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_api_disabled

    body = {
        "error": {
            "code": 403,
            "message": "Access Not Configured. DNS API not enabled.",
            "errors": [
                {
                    "message": "Access Not Configured. DNS API not enabled.",
                    "domain": "usageLimits",
                    "reason": "accessNotConfigured",
                },
            ],
        },
    }
    exc = HttpError(
        resp=MagicMock(status=403),
        content=json.dumps(body).encode("utf-8"),
    )
    assert _classify_gcp_api_disabled(exc) == "unknown"


def test_classify_gcp_api_disabled_real_permission_denied_returns_none() -> None:
    """A genuine permission denial (no SERVICE_DISABLED / accessNotConfigured
    marker) must return None so the leg stays in ``partial_failures``.
    Returning a non-None here would silently mask a real IAM gap."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_api_disabled

    body = {
        "error": {
            "code": 403,
            "message": "The caller does not have permission",
            "status": "PERMISSION_DENIED",
            "details": [
                {
                    "@type": "type.googleapis.com/google.rpc.ErrorInfo",
                    "reason": "IAM_PERMISSION_DENIED",
                    "domain": "iam.googleapis.com",
                    "metadata": {
                        "permission": "compute.instances.list",
                        "resource": "projects/foo",
                    },
                },
            ],
        },
    }
    exc = HttpError(
        resp=MagicMock(status=403),
        content=json.dumps(body).encode("utf-8"),
    )
    assert _classify_gcp_api_disabled(exc) is None


def test_classify_gcp_api_disabled_malformed_body_returns_none() -> None:
    """Pathological inputs (no body / non-JSON / bizarre shape) must
    NOT be classified as api-disabled — that would silently mask real
    permission denials. Default to None and let the caller treat as
    a real 403."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_api_disabled

    # No content.
    exc1 = HttpError(resp=MagicMock(status=403), content=b"")
    assert _classify_gcp_api_disabled(exc1) is None

    # Non-JSON HTML error page (load-balancer interstitial).
    exc2 = HttpError(
        resp=MagicMock(status=403),
        content=b"<html><body>403 Forbidden</body></html>",
    )
    assert _classify_gcp_api_disabled(exc2) is None

    # JSON but not the expected shape.
    exc3 = HttpError(
        resp=MagicMock(status=403),
        content=b'{"foo": "bar"}',
    )
    assert _classify_gcp_api_disabled(exc3) is None

    # JSON array instead of object.
    exc4 = HttpError(
        resp=MagicMock(status=403),
        content=b'[1, 2, 3]',
    )
    assert _classify_gcp_api_disabled(exc4) is None


def test_classify_gcp_api_disabled_oversized_body_short_circuits() -> None:
    """A 200 KB body is almost certainly an HTML page, not a structured
    GCP error. Reject without attempting JSON parse to bound CPU work."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_api_disabled

    huge = b"x" * 200_000
    exc = HttpError(resp=MagicMock(status=403), content=huge)
    assert _classify_gcp_api_disabled(exc) is None


def test_classify_gcp_api_disabled_service_name_sanitized() -> None:
    """The ``service`` metadata is interpolated into a leg ``reason``
    that lands in admin-visible ``capability_results``. Sanitize aggressively
    against control chars / unbounded values (defense-in-depth — GCP's
    real service names match ``[a-z]+\\.googleapis\\.com``)."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_api_disabled

    body = {
        "error": {
            "details": [
                {
                    "reason": "SERVICE_DISABLED",
                    "metadata": {
                        # Embedded newline + null byte + 5 KB of garbage.
                        "service": "compute.googleapis.com\n\x00" + ("A" * 5000),
                    },
                },
            ],
        },
    }
    exc = HttpError(
        resp=MagicMock(status=403),
        content=json.dumps(body).encode("utf-8"),
    )
    result = _classify_gcp_api_disabled(exc)
    assert result is not None
    # Sanitizer keeps only ``[a-zA-Z0-9._-]`` and caps at 128 chars.
    assert "\n" not in result
    assert "\x00" not in result
    assert result == "compute.googleapis.com" + (
        "A" * (128 - len("compute.googleapis.com"))
    )


# ── End-to-end: API-disabled 403 routes to skipped_legs ─────────


@pytest.mark.asyncio
async def test_list_servers_api_disabled_403_goes_to_skipped_legs() -> None:
    """The bug: 474 projects without Compute API enabled returned 403
    and polluted ``partial_failures`` (capability flipped to
    ``partial_failure``, failure-notifier paged on every cron tick).
    The fix: a SERVICE_DISABLED 403 must land in ``skipped_legs`` so
    the capability stays green."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "bigquery-only-proj"}],
    }
    api_disabled_body = json.dumps({
        "error": {
            "code": 403,
            "status": "PERMISSION_DENIED",
            "details": [
                {
                    "reason": "SERVICE_DISABLED",
                    "metadata": {"service": "compute.googleapis.com"},
                },
            ],
        },
    }).encode("utf-8")
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=403),
        content=api_disabled_body,
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert servers == []
    # Did NOT pollute partial_failures (would have flipped sync to red).
    assert p.partial_failures == []
    # DID record an informational skipped leg.
    assert len(p.skipped_legs) == 1
    leg = p.skipped_legs[0]
    assert leg["leg"] == "project:bigquery-only-proj"
    assert leg["capability"] == "servers"
    assert leg["reason"] == "api_disabled:compute.googleapis.com"


@pytest.mark.asyncio
async def test_list_servers_real_permission_denied_stays_in_partial_failures() -> None:
    """Symmetric to the api-disabled test: a real permission denial
    MUST stay in ``partial_failures`` so the failure-notifier pages —
    the SA legitimately needs the IAM binding and silently dropping
    the row would mask the issue."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "real-iam-gap-proj"}],
    }
    # No SERVICE_DISABLED / accessNotConfigured — just a bare 403.
    body = json.dumps({
        "error": {
            "code": 403,
            "status": "PERMISSION_DENIED",
            "message": "The caller does not have permission",
        },
    }).encode("utf-8")
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=403),
        content=body,
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert servers == []
    assert p.skipped_legs == []
    assert len(p.partial_failures) == 1
    leg = p.partial_failures[0]
    assert leg["leg"] == "project:real-iam-gap-proj"
    assert leg["reason"] == "http_403"


@pytest.mark.asyncio
async def test_list_dns_records_api_disabled_403_goes_to_skipped_legs() -> None:
    """Same disambiguation for the DNS managedZones.list branch."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "no-dns-proj"}],
    }
    api_disabled_body = json.dumps({
        "error": {
            "details": [{
                "reason": "SERVICE_DISABLED",
                "metadata": {"service": "dns.googleapis.com"},
            }],
        },
    }).encode("utf-8")
    dns = MagicMock()
    dns.managedZones.return_value.list.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=403),
        content=api_disabled_body,
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("dns", "v1"): dns,
        }):
            records = await p.list_dns_records()
    finally:
        await p.close()
    assert records == []
    assert p.partial_failures == []
    assert len(p.skipped_legs) == 1
    assert p.skipped_legs[0]["reason"] == "api_disabled:dns.googleapis.com"


@pytest.mark.asyncio
async def test_list_object_storage_api_disabled_403_goes_to_skipped_legs() -> None:
    """Same disambiguation for the GCS buckets.list branch."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "no-storage-proj"}],
    }
    api_disabled_body = json.dumps({
        "error": {
            "details": [{
                "reason": "SERVICE_DISABLED",
                "metadata": {"service": "storage.googleapis.com"},
            }],
        },
    }).encode("utf-8")
    storage = MagicMock()
    storage.buckets.return_value.list.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=403),
        content=api_disabled_body,
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("storage", "v1"): storage,
        }):
            buckets = await p.list_object_storage()
    finally:
        await p.close()
    assert buckets == []
    assert p.partial_failures == []
    assert len(p.skipped_legs) == 1
    assert p.skipped_legs[0]["reason"] == "api_disabled:storage.googleapis.com"


@pytest.mark.asyncio
async def test_404_still_lands_in_partial_failures_not_skipped() -> None:
    """404 is a different shape (project deleted mid-sync, not API
    disabled). Stays in partial_failures so the admin can confirm
    project lifecycle changes."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "deleted-proj"}],
    }
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=404),
        content=b'{"error":{"code":404,"message":"Project not found"}}',
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert servers == []
    assert p.skipped_legs == []
    assert len(p.partial_failures) == 1
    assert p.partial_failures[0]["reason"] == "http_404"


def test_classify_gcp_api_disabled_hybrid_body_prefers_iam_denial() -> None:
    """Round-1 review hardening: if GCP returns a body with BOTH
    ``SERVICE_DISABLED`` and ``IAM_PERMISSION_DENIED`` markers in
    ``details`` (rare race condition during a project rollout where the
    API was just enabled but the SA still lacks the binding), prefer
    the IAM denial — never silently mask a real permission gap."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_api_disabled

    body = {
        "error": {
            "code": 403,
            "details": [
                {
                    "reason": "SERVICE_DISABLED",
                    "metadata": {"service": "compute.googleapis.com"},
                },
                {
                    "reason": "IAM_PERMISSION_DENIED",
                    "metadata": {"permission": "compute.instances.list"},
                },
            ],
        },
    }
    exc = HttpError(
        resp=MagicMock(status=403),
        content=json.dumps(body).encode("utf-8"),
    )
    # Even though SERVICE_DISABLED appears first, IAM denial must win.
    assert _classify_gcp_api_disabled(exc) is None

    # Reverse order — same outcome (IAM denial still wins).
    body["error"]["details"].reverse()
    exc2 = HttpError(
        resp=MagicMock(status=403),
        content=json.dumps(body).encode("utf-8"),
    )
    assert _classify_gcp_api_disabled(exc2) is None


def test_classify_gcp_api_disabled_memoryview_content() -> None:
    """Defense-in-depth: ``googleapiclient.HttpError`` itself enforces
    bytes at construction (rejects memoryview / bytearray with
    TypeError), but the classifier accepts any exception-shaped object
    that exposes ``.content`` — useful for hand-rolled SDK wrappers
    that bypass HttpError. Bypass HttpError's bytes guard with a
    SimpleNamespace stand-in to assert classifier robustness."""
    from types import SimpleNamespace

    from app.domains.cmdb.providers.gcp import _classify_gcp_api_disabled

    body = json.dumps({
        "error": {
            "details": [{
                "reason": "SERVICE_DISABLED",
                "metadata": {"service": "compute.googleapis.com"},
            }],
        },
    }).encode("utf-8")
    fake_exc = SimpleNamespace(content=memoryview(body), resp=MagicMock(status=403))
    assert _classify_gcp_api_disabled(fake_exc) == "compute.googleapis.com"

    fake_exc_ba = SimpleNamespace(content=bytearray(body), resp=MagicMock(status=403))
    assert _classify_gcp_api_disabled(fake_exc_ba) == "compute.googleapis.com"


def test_classify_gcp_api_disabled_tiny_empty_error_body() -> None:
    """Pathological tiny body ``{"error":{}}`` lands in default-None
    branch (no details, no errors). Stays in partial_failures —
    correct because we have no positive signal that the API is
    legitimately disabled."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_api_disabled

    exc = HttpError(
        resp=MagicMock(status=403),
        content=b'{"error":{}}',
    )
    assert _classify_gcp_api_disabled(exc) is None


@pytest.mark.asyncio
async def test_list_dns_records_rrsets_level_api_disabled_403_goes_to_skipped_legs() -> None:
    """Round-1 review: end-to-end coverage for the rrsets-level (4th)
    403 disambiguation branch. managedZones.list succeeds but the
    per-zone rrsets.list raises SERVICE_DISABLED — should land in
    skipped_legs with the project+zone leg shape, not partial_failures."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-a"}],
    }
    dns = MagicMock()
    dns.managedZones.return_value.list.return_value.execute.return_value = {
        "managedZones": [
            {
                "name": "example-com",
                "dnsName": "example.com.",
                "visibility": "public",
            },
        ],
    }
    api_disabled_body = json.dumps({
        "error": {
            "details": [{
                "reason": "SERVICE_DISABLED",
                "metadata": {"service": "dns.googleapis.com"},
            }],
        },
    }).encode("utf-8")
    dns.resourceRecordSets.return_value.list.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=403),
        content=api_disabled_body,
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("dns", "v1"): dns,
        }):
            records = await p.list_dns_records()
    finally:
        await p.close()
    assert records == []
    assert p.partial_failures == []
    assert len(p.skipped_legs) == 1
    leg = p.skipped_legs[0]
    assert leg["leg"] == "project:proj-a:zone:example-com"
    assert leg["capability"] == "dns_records"
    assert leg["reason"] == "api_disabled:dns.googleapis.com"


@pytest.mark.parametrize(
    "capability,svc_key,svc_setup",
    [
        # (capability label, _svc() key, lambda(MagicMock) → wire up the 500)
        (
            "servers", ("compute", "v1"),
            lambda m: setattr(
                m.instances.return_value.aggregatedList.return_value,
                "execute",
                MagicMock(side_effect=__import__(
                    "googleapiclient.errors", fromlist=["HttpError"],
                ).HttpError(
                    resp=MagicMock(status=500),
                    content=b'{"error":{"code":500,"message":"transient"}}',
                )),
            ),
        ),
        (
            "dns_records", ("dns", "v1"),
            lambda m: setattr(
                m.managedZones.return_value.list.return_value,
                "execute",
                MagicMock(side_effect=__import__(
                    "googleapiclient.errors", fromlist=["HttpError"],
                ).HttpError(
                    resp=MagicMock(status=500),
                    content=b'{"error":{"code":500,"message":"transient"}}',
                )),
            ),
        ),
        (
            "object_storage", ("storage", "v1"),
            lambda m: setattr(
                m.buckets.return_value.list.return_value,
                "execute",
                MagicMock(side_effect=__import__(
                    "googleapiclient.errors", fromlist=["HttpError"],
                ).HttpError(
                    resp=MagicMock(status=500),
                    content=b'{"error":{"code":500,"message":"transient"}}',
                )),
            ),
        ),
    ],
)
@pytest.mark.asyncio
async def test_fan_out_per_project_catch_all_leg_has_capability_key(
    capability: str,
    svc_key: tuple[str, str],
    svc_setup: Any,
) -> None:
    """Round-1 + Round-2 review: the bare-except fallback in
    ``_fan_out_per_project`` previously emitted partial_failures legs
    WITHOUT a ``capability`` field, which downstream admin UI rendered
    as ``(unknown capability)``. Every other leg-emitter sets
    ``capability`` explicitly — the catch-all must too, for ALL three
    fan-out callers (servers / dns_records / object_storage).

    Round-2 caught that ``list_object_storage`` had been missed in the
    Round-1 plumbing fix; this parametrized test covers each so a
    future kwarg-drop is caught by CI."""
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-broken"}],
    }
    svc_mock = MagicMock()
    svc_setup(svc_mock)
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            svc_key: svc_mock,
        }):
            if capability == "servers":
                _ = await p.list_servers()
            elif capability == "dns_records":
                _ = await p.list_dns_records()
            else:
                _ = await p.list_object_storage()
    finally:
        await p.close()
    # The 500 falls past the 403/404 branches in the per-project helpers
    # (which re-raise non-403/404) and gets caught by _one()'s catch-all.
    assert len(p.partial_failures) == 1
    leg = p.partial_failures[0]
    assert leg["leg"] == "project:proj-broken"
    assert leg["capability"] == capability
    assert "HttpError" in leg["reason"]


def test_consume_skipped_legs_clears_list() -> None:
    """The drain helper must clear the list so a second capability's
    drain doesn't see the first capability's skips."""
    p = _provider()
    p.skipped_legs.append({"leg": "project:foo", "reason": "api_disabled:x"})
    assert len(p.consume_skipped_legs()) == 1
    assert p.skipped_legs == []
    assert p.consume_skipped_legs() == []


# ── Round-26 (2026-05-18) classifier broadening ─────────────────────
#
# Production incident: ``project:tidb-376913`` flagged ``http_403 × 1``
# on the GCP servers cron for >24h after the admin claimed the IAM
# binding was fixed. The pre-fix leg gave zero context — the admin
# could not tell whether it was a still-propagating IAM binding, a
# VPC-SC perimeter rule, an Org Policy constraint, billing disabled
# on the project, or a real permission gap. Round-26 broadens the
# classifier to recognize all five and enriches the bare-403 leg
# with the GCP ``error.message`` snippet.


def test_classify_gcp_403_category_returns_api_disabled_modern_shape() -> None:
    """Backwards compat: SERVICE_DISABLED still classified as api_disabled."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "details": [{
                "@type": "type.googleapis.com/google.rpc.ErrorInfo",
                "reason": "SERVICE_DISABLED",
                "metadata": {"service": "compute.googleapis.com"},
            }],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) == (
        "api_disabled", "compute.googleapis.com",
    )


def test_classify_gcp_403_category_vpc_sc_via_errorinfo() -> None:
    """VPC-SC perimeter denial via ErrorInfo reason=SECURITY_POLICY_VIOLATED
    + metadata.violationReason=VPC_SERVICE_CONTROLS."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "message": "Request is prohibited by organization's policy.",
            "details": [{
                "@type": "type.googleapis.com/google.rpc.ErrorInfo",
                "reason": "SECURITY_POLICY_VIOLATED",
                "metadata": {
                    "service": "compute.googleapis.com",
                    "violationReason": "VPC_SERVICE_CONTROLS",
                },
            }],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) == (
        "vpc_sc", "compute.googleapis.com",
    )


def test_classify_gcp_403_category_vpc_sc_via_precondition_failure() -> None:
    """VPC-SC also surfaces as PreconditionFailure with violations[*].type
    = VPC_SERVICE_CONTROLS (Cloud DNS / IAM Policy paths)."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "details": [{
                "@type": "type.googleapis.com/google.rpc.PreconditionFailure",
                "violations": [{
                    "type": "VPC_SERVICE_CONTROLS",
                    "subject": "projects/123/services/dns.googleapis.com",
                    "description": "Request is prohibited by org perimeter",
                }],
            }],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    # No service name in the PreconditionFailure shape — fall back to "unknown".
    assert _classify_gcp_403_category(exc) == ("vpc_sc", "unknown")


def test_classify_gcp_403_category_org_policy_via_errorinfo() -> None:
    """Org Policy constraint via ErrorInfo reason=ORG_POLICY_CONSTRAINT_VIOLATED
    with metadata.constraint."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "details": [{
                "@type": "type.googleapis.com/google.rpc.ErrorInfo",
                "reason": "ORG_POLICY_CONSTRAINT_VIOLATED",
                "metadata": {
                    "constraint": "constraints/iam.disableServiceAccountKeyCreation",
                },
            }],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) == (
        "org_policy", "constraints/iam.disableServiceAccountKeyCreation",
    )


def test_classify_gcp_403_category_org_policy_via_precondition_violation() -> None:
    """Org Policy can also appear as PreconditionFailure with
    violation type starting with ``constraints/``."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "details": [{
                "@type": "type.googleapis.com/google.rpc.PreconditionFailure",
                "violations": [{
                    "type": "constraints/compute.disableSerialPortAccess",
                    "subject": "projects/123",
                }],
            }],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) == (
        "org_policy", "constraints/compute.disableSerialPortAccess",
    )


def test_classify_gcp_403_category_billing_disabled() -> None:
    """Billing not enabled: most GCP APIs 403 with reason=BILLING_DISABLED."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "message": "Billing must be enabled for activation",
            "details": [{
                "@type": "type.googleapis.com/google.rpc.ErrorInfo",
                "reason": "BILLING_DISABLED",
                "metadata": {"service": "compute.googleapis.com"},
            }],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) == (
        "billing_disabled", "compute.googleapis.com",
    )


def test_classify_gcp_403_category_consumer_invalid() -> None:
    """CONSUMER_INVALID — project in deletion / suspended."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "details": [{
                "reason": "CONSUMER_INVALID",
                "metadata": {"service": "compute.googleapis.com"},
            }],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) == (
        "consumer_invalid", "compute.googleapis.com",
    )


def test_classify_gcp_403_category_iam_denied_dominates_all_categories() -> None:
    """IAM_PERMISSION_DENIED appearing alongside ANY other structural
    marker MUST return None — silently masking the real permission gap
    would defeat the failure-notifier."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    # IAM denial + SERVICE_DISABLED
    body1 = json.dumps({
        "error": {
            "code": 403,
            "details": [
                {"reason": "SERVICE_DISABLED",
                 "metadata": {"service": "compute.googleapis.com"}},
                {"reason": "IAM_PERMISSION_DENIED",
                 "metadata": {"permission": "compute.instances.list"}},
            ],
        },
    }).encode("utf-8")
    assert _classify_gcp_403_category(
        HttpError(resp=MagicMock(status=403), content=body1),
    ) is None

    # IAM denial + VPC_SERVICE_CONTROLS
    body2 = json.dumps({
        "error": {
            "code": 403,
            "details": [
                {"reason": "SECURITY_POLICY_VIOLATED",
                 "metadata": {"violationReason": "VPC_SERVICE_CONTROLS",
                              "service": "compute.googleapis.com"}},
                {"reason": "IAM_PERMISSION_DENIED"},
            ],
        },
    }).encode("utf-8")
    assert _classify_gcp_403_category(
        HttpError(resp=MagicMock(status=403), content=body2),
    ) is None

    # IAM denial + BILLING_DISABLED
    body3 = json.dumps({
        "error": {
            "code": 403,
            "details": [
                {"reason": "BILLING_DISABLED",
                 "metadata": {"service": "compute.googleapis.com"}},
                {"reason": "IAM_PERMISSION_DENIED"},
            ],
        },
    }).encode("utf-8")
    assert _classify_gcp_403_category(
        HttpError(resp=MagicMock(status=403), content=body3),
    ) is None


def test_classify_gcp_403_category_real_permission_denied_returns_none() -> None:
    """No structural marker, bare PERMISSION_DENIED → real IAM gap."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "status": "PERMISSION_DENIED",
            "message": (
                "Permission 'compute.instances.list' denied on resource "
                "'projects/tidb-376913' (or it may not exist)."
            ),
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) is None


def test_classify_gcp_403_category_sanitizes_service_name() -> None:
    """Malformed metadata.service must be sanitized (defense-in-depth)."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "details": [{
                "reason": "SERVICE_DISABLED",
                "metadata": {
                    "service": "compute.googleapis.com\n<script>x</script>",
                },
            }],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    result = _classify_gcp_403_category(exc)
    assert result is not None
    category, detail = result
    assert category == "api_disabled"
    # The legitimate prefix survives; the injection is stripped.
    assert detail is not None
    assert "<" not in detail
    assert "\n" not in detail
    assert detail == "compute.googleapis.comscriptx/script"


def test_classify_gcp_403_category_security_policy_non_vpc_sc_stays_real() -> None:
    """Round-26 review: a SECURITY_POLICY_VIOLATED reason with a
    violationReason OTHER than VPC_SERVICE_CONTROLS must NOT be
    classified as vpc_sc — it could be a real customer-defined policy
    the admin needs to see."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "details": [{
                "reason": "SECURITY_POLICY_VIOLATED",
                "metadata": {
                    "violationReason": "SOME_OTHER_POLICY",
                    "service": "compute.googleapis.com",
                },
            }],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    # Falls through — no other markers, not classified as a skip.
    assert _classify_gcp_403_category(exc) is None


# ── End-to-end: new categories route to skipped_legs ────────────────


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "category,details,expected_reason",
    [
        (
            "vpc_sc",
            [{"reason": "SECURITY_POLICY_VIOLATED",
              "metadata": {"violationReason": "VPC_SERVICE_CONTROLS",
                           "service": "compute.googleapis.com"}}],
            "vpc_sc:compute.googleapis.com",
        ),
        (
            "org_policy",
            [{"reason": "ORG_POLICY_CONSTRAINT_VIOLATED",
              "metadata": {"constraint": "constraints/iam.disableServiceAccountKeyCreation"}}],
            "org_policy:constraints/iam.disableServiceAccountKeyCreation",
        ),
        (
            "billing_disabled",
            [{"reason": "BILLING_DISABLED",
              "metadata": {"service": "compute.googleapis.com"}}],
            "billing_disabled:compute.googleapis.com",
        ),
        (
            "consumer_invalid",
            [{"reason": "CONSUMER_INVALID",
              "metadata": {"service": "compute.googleapis.com"}}],
            "consumer_invalid:compute.googleapis.com",
        ),
    ],
)
async def test_list_servers_routes_new_skip_categories_to_skipped_legs(
    category: str, details: list, expected_reason: str,
) -> None:
    """Each Round-26 category must land in skipped_legs (not partial_failures)
    so the capability stays green and the failure-notifier stays quiet."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-blocked"}],
    }
    body = json.dumps({"error": {"code": 403, "details": details}}).encode("utf-8")
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=403),
        content=body,
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert servers == []
    assert p.partial_failures == [], (
        f"{category} must NOT pollute partial_failures: got {p.partial_failures}"
    )
    assert len(p.skipped_legs) == 1
    leg = p.skipped_legs[0]
    assert leg["leg"] == "project:proj-blocked"
    assert leg["capability"] == "servers"
    assert leg["reason"] == expected_reason


# ── Round-26 review P1-4: same parametrize for DNS + object_storage ─


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "category,details,expected_reason",
    [
        (
            "vpc_sc",
            [{"reason": "SECURITY_POLICY_VIOLATED",
              "metadata": {"violationReason": "VPC_SERVICE_CONTROLS",
                           "service": "dns.googleapis.com"}}],
            "vpc_sc:dns.googleapis.com",
        ),
        (
            "org_policy",
            [{"reason": "ORG_POLICY_CONSTRAINT_VIOLATED",
              "metadata": {"constraint": "constraints/dns.someConstraint"}}],
            "org_policy:constraints/dns.someConstraint",
        ),
        (
            "billing_disabled",
            [{"reason": "BILLING_DISABLED",
              "metadata": {"service": "dns.googleapis.com"}}],
            "billing_disabled:dns.googleapis.com",
        ),
        (
            "consumer_invalid",
            [{"reason": "CONSUMER_INVALID",
              "metadata": {"service": "dns.googleapis.com"}}],
            "consumer_invalid:dns.googleapis.com",
        ),
    ],
)
async def test_list_dns_records_routes_new_skip_categories(
    category: str, details: list, expected_reason: str,
) -> None:
    """DNS managedZones.list branch must also recognize the new
    Round-26 skip categories — symmetric to servers."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-blocked"}],
    }
    body = json.dumps({"error": {"code": 403, "details": details}}).encode("utf-8")
    dns = MagicMock()
    dns.managedZones.return_value.list.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=403),
        content=body,
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("dns", "v1"): dns,
        }):
            records = await p.list_dns_records()
    finally:
        await p.close()
    assert records == []
    assert p.partial_failures == []
    assert len(p.skipped_legs) == 1
    leg = p.skipped_legs[0]
    assert leg["leg"] == "project:proj-blocked"
    assert leg["capability"] == "dns_records"
    assert leg["reason"] == expected_reason


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "category,details,expected_reason",
    [
        (
            "vpc_sc",
            [{"reason": "SECURITY_POLICY_VIOLATED",
              "metadata": {"violationReason": "VPC_SERVICE_CONTROLS",
                           "service": "storage.googleapis.com"}}],
            "vpc_sc:storage.googleapis.com",
        ),
        (
            "org_policy",
            [{"reason": "ORG_POLICY_CONSTRAINT_VIOLATED",
              "metadata": {"constraint": "constraints/storage.publicAccessPrevention"}}],
            "org_policy:constraints/storage.publicAccessPrevention",
        ),
        (
            "billing_disabled",
            [{"reason": "BILLING_DISABLED",
              "metadata": {"service": "storage.googleapis.com"}}],
            "billing_disabled:storage.googleapis.com",
        ),
        (
            "consumer_invalid",
            [{"reason": "CONSUMER_INVALID",
              "metadata": {"service": "storage.googleapis.com"}}],
            "consumer_invalid:storage.googleapis.com",
        ),
    ],
)
async def test_list_object_storage_routes_new_skip_categories(
    category: str, details: list, expected_reason: str,
) -> None:
    """GCS buckets.list branch must also recognize the new
    Round-26 skip categories — symmetric to servers."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-blocked"}],
    }
    body = json.dumps({"error": {"code": 403, "details": details}}).encode("utf-8")
    storage = MagicMock()
    storage.buckets.return_value.list.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=403),
        content=body,
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("storage", "v1"): storage,
        }):
            buckets = await p.list_object_storage()
    finally:
        await p.close()
    assert buckets == []
    assert p.partial_failures == []
    assert len(p.skipped_legs) == 1
    leg = p.skipped_legs[0]
    assert leg["leg"] == "project:proj-blocked"
    assert leg["capability"] == "object_storage"
    assert leg["reason"] == expected_reason


# ── Round-26 review P0-1: SECURITY_POLICY_VIOLATED without violationReason ─


def test_classify_gcp_403_category_security_policy_no_violation_reason_stays_real() -> None:
    """P0-1 fix: ``reason=SECURITY_POLICY_VIOLATED`` with NO
    ``metadata.violationReason`` MUST stay None (real 403) — could be
    a Cloud Armor / BeyondCorp / customer-defined policy the admin
    needs to investigate. Pre-P0 the classifier accepted any
    SECURITY_POLICY_VIOLATED as VPC-SC unless explicitly contradicted;
    that silently masked real security policy denials."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "details": [{
                "reason": "SECURITY_POLICY_VIOLATED",
                "metadata": {"service": "compute.googleapis.com"},
                # NO violationReason key at all.
            }],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    # P0 fix: stays None (real failure), NOT vpc_sc.
    assert _classify_gcp_403_category(exc) is None


def test_classify_gcp_403_category_security_policy_empty_violation_reason_stays_real() -> None:
    """Edge case: ``violationReason`` present but empty string. Same P0
    decision — empty isn't a positive match for VPC_SERVICE_CONTROLS."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "details": [{
                "reason": "SECURITY_POLICY_VIOLATED",
                "metadata": {
                    "violationReason": "",
                    "service": "compute.googleapis.com",
                },
            }],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) is None


# ── Round-31 (2026-05-18): legacy REST + message-pattern billing 403 ─
#
# Prod incident ``project:tidb-376913`` paged http_403 for >24h because
# GCP Compute V1 returns the legacy ``error.errors[*].reason`` body
# shape (NOT the modern ``error.details[*].reason`` gRPC ErrorInfo) for
# billing-disabled, and the classifier only recognized the modern shape.
# These tests lock in the three new code paths:
#   a. legacy ``errors[].reason == "billingNotEnabled"``
#   b. legacy ``errors[].reason == "billingDisabled"`` (synonym)
#   c. last-resort top-level ``error.message`` pattern match


def test_classify_gcp_403_category_legacy_billing_not_enabled() -> None:
    """Reproduces the prod ``project:tidb-376913`` body verbatim."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "message": (
                "This API method requires billing to be enabled. "
                "Please enable billing on project #tidb-376913 by "
                "visiting https://console.developers.google.com/billing"
                "/enable?project=tidb-376913 then retry."
            ),
            "errors": [{
                "domain": "global",
                "reason": "billingNotEnabled",
                "message": "This API method requires billing to be enabled.",
            }],
            "status": "PERMISSION_DENIED",
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) == ("billing_disabled", "unknown")


def test_classify_gcp_403_category_legacy_billing_disabled_synonym() -> None:
    """Cloud Functions / older Datastore endpoints spell the legacy
    reason ``billingDisabled``; same category as ``billingNotEnabled``."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "errors": [{"reason": "billingDisabled"}],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) == ("billing_disabled", "unknown")


def test_classify_gcp_403_category_legacy_consumer_invalid() -> None:
    """Legacy ``consumerInvalid`` reason — project deleted / suspended."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "errors": [{"reason": "consumerInvalid"}],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) == ("consumer_invalid", "unknown")


def test_classify_gcp_403_category_legacy_account_disabled_synonym() -> None:
    """``accountDisabled`` legacy synonym for consumer_invalid."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "errors": [{"reason": "accountDisabled"}],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) == ("consumer_invalid", "unknown")


def test_classify_gcp_403_category_legacy_precedence_api_over_billing() -> None:
    """When both legacy reasons appear in ``errors[]`` (defensive case
    GCP shouldn't emit but might), api_disabled wins — matches the
    modern path's precedence."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "errors": [
                {"reason": "billingNotEnabled"},
                {"reason": "accessNotConfigured"},
            ],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) == ("api_disabled", "unknown")


def test_classify_gcp_403_category_legacy_unknown_reason_returns_none() -> None:
    """Legacy ``errors[]`` carrying an unrecognized reason must NOT be
    silently classified — falls through to the message-pattern fallback
    and from there to None (= real 403)."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "errors": [{"reason": "forbidden",
                         "message": "Permission denied."}],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) is None


def test_classify_gcp_403_category_message_only_billing_required() -> None:
    """No ``details[]``, no ``errors[]`` — only the human-readable
    ``error.message`` carries the billing signal. The strict pattern
    fallback must still classify."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "message": (
                "This API method requires billing to be enabled. "
                "Please enable billing on project #tidb-376913."
            ),
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) == ("billing_disabled", "unknown")


def test_classify_gcp_403_category_message_pattern_iam_denial_dominates() -> None:
    """CRITICAL false-positive guard: an error whose message mentions
    both "enable billing" (as remediation hint) AND a real IAM denial
    phrasing must stay None — silently downgrading a real permission
    gap defeats the failure-notifier."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "message": (
                "Permission 'compute.instances.list' denied on resource "
                "'projects/tidb-376913'. If you recently enabled billing "
                "on project tidb-376913, wait a few minutes."
            ),
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) is None


def test_classify_gcp_403_category_message_pattern_caller_no_permission_dominates() -> None:
    """Alternate IAM-denial phrasing GCP uses (``caller does not have
    permission``) — also dominates a billing-message coincidence."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "message": (
                "The caller does not have permission. To proceed, please "
                "enable billing on project tidb-foo."
            ),
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) is None


def test_classify_gcp_403_category_message_only_unrelated_403_returns_none() -> None:
    """Top-level message that doesn't match the billing pattern stays
    None — message-pattern fallback must not over-trigger."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "message": "Some other 403 condition we don't classify.",
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) is None


def test_classify_gcp_403_category_iam_denied_in_details_dominates_legacy_billing() -> None:
    """Defensive: if modern ``details[]`` carries IAM_PERMISSION_DENIED
    AND legacy ``errors[]`` carries billingNotEnabled (a hypothetical
    hybrid body), IAM dominance must win — None, not billing_disabled.

    The modern-path scan runs first and sets ``iam_denied=True``, which
    short-circuits before the legacy block is reached."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "details": [{"reason": "IAM_PERMISSION_DENIED"}],
            "errors": [{"reason": "billingNotEnabled"}],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) is None


def test_classify_gcp_403_category_modern_details_dominates_legacy_errors() -> None:
    """Hybrid body where the modern ``details[]`` carries a structural
    marker AND the legacy ``errors[]`` carries a DIFFERENT one — modern
    path wins because the legacy block only runs when no modern marker
    set an accumulator. Locks in the cross-shape precedence so future
    refactors don't accidentally make legacy win."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "details": [{
                "reason": "SERVICE_DISABLED",
                "metadata": {"service": "compute.googleapis.com"},
            }],
            # Legacy block would say billing_disabled — must be ignored.
            "errors": [{"reason": "billingNotEnabled"}],
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    # Modern api_disabled (with rich detail) wins.
    assert _classify_gcp_403_category(exc) == (
        "api_disabled", "compute.googleapis.com",
    )


def test_classify_gcp_403_category_iam_remediation_hint_stays_real() -> None:
    """Round-31 review MEDIUM regression: an IAM-denial message that
    contains the past-tense ``enabled billing`` remediation hint must
    NOT match the billing pattern. Pre-fix the pattern ``enable billing
    on project`` also matched ``enabled billing on project`` and would
    have silently downgraded a real IAM gap to billing_disabled."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "message": (
                "If you recently enabled billing on project foo, "
                "wait a few minutes for IAM propagation."
            ),
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    # No imperative billing phrasing → stays None → real partial_failure.
    assert _classify_gcp_403_category(exc) is None


def test_classify_gcp_403_category_billing_account_closed_message() -> None:
    """Billing-account-closed phrasing (separate from project-billing-
    disabled) — the third arm of the billing pattern."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_403_category

    body = json.dumps({
        "error": {
            "code": 403,
            "message": (
                "The billing account 0X1234-ABCDEF-567890 is closed. "
                "Re-open the account or attach a different one."
            ),
        },
    }).encode("utf-8")
    exc = HttpError(resp=MagicMock(status=403), content=body)
    assert _classify_gcp_403_category(exc) == ("billing_disabled", "unknown")


@pytest.mark.asyncio
async def test_list_servers_legacy_billing_routes_to_skipped_legs() -> None:
    """End-to-end: prod-shaped ``project:tidb-376913`` legacy billing
    body now lands in ``skipped_legs`` (not ``partial_failures``), so
    the failure-notifier stops paging."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "tidb-376913"}],
    }
    body = json.dumps({
        "error": {
            "code": 403,
            "message": (
                "This API method requires billing to be enabled. "
                "Please enable billing on project #tidb-376913."
            ),
            "errors": [{"domain": "global", "reason": "billingNotEnabled"}],
        },
    }).encode("utf-8")
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=403),
        content=body,
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert servers == []
    # Round-31 fix: leg now in skipped_legs, NOT partial_failures.
    assert p.partial_failures == []
    assert len(p.skipped_legs) == 1
    leg = p.skipped_legs[0]
    assert leg["leg"] == "project:tidb-376913"
    assert leg["capability"] == "servers"
    assert leg["reason"] == "billing_disabled:unknown"


# ── End-to-end: bare-403 leg carries enriched error_message ─────────


@pytest.mark.asyncio
async def test_list_servers_real_permission_denied_attaches_error_message() -> None:
    """Round-26: a real ``http_403`` partial_failures leg now carries
    the redacted ``error.message`` so admins can diagnose immediately
    from the sync-history row without grepping logs.

    Reproduces the prod ``project:tidb-376913`` incident message.
    """
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "tidb-376913"}],
    }
    body = json.dumps({
        "error": {
            "code": 403,
            "status": "PERMISSION_DENIED",
            "message": (
                "Permission 'compute.instances.list' denied on resource "
                "'projects/tidb-376913' (or it may not exist)."
            ),
        },
    }).encode("utf-8")
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=403),
        content=body,
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert servers == []
    assert p.skipped_legs == []
    assert len(p.partial_failures) == 1
    leg = p.partial_failures[0]
    assert leg["leg"] == "project:tidb-376913"
    assert leg["reason"] == "http_403"
    # Round-26 enrichment: error_message present and informative.
    assert "error_message" in leg
    assert "compute.instances.list" in leg["error_message"]
    assert "tidb-376913" in leg["error_message"]


@pytest.mark.asyncio
async def test_list_servers_http_404_carries_error_message() -> None:
    """Same enrichment applies on 404 — admin needs to see the message
    to know if the project really was deleted vs an SDK glitch."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "deleted-proj"}],
    }
    body = json.dumps({
        "error": {
            "code": 404,
            "message": "Project 'projects/deleted-proj' not found.",
        },
    }).encode("utf-8")
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=404),
        content=body,
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert servers == []
    assert len(p.partial_failures) == 1
    leg = p.partial_failures[0]
    assert leg["reason"] == "http_404"
    assert "error_message" in leg
    assert "not found" in leg["error_message"]


@pytest.mark.asyncio
async def test_list_servers_403_no_body_falls_back_to_bare_reason() -> None:
    """Defense-in-depth: a 403 with NO body (or malformed) must still
    produce a usable leg with bare ``http_403`` reason and no
    ``error_message`` key — preserves dedup signature space."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "no-body-proj"}],
    }
    compute = MagicMock()
    compute.instances.return_value.aggregatedList.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=403),
        content=b"",
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            servers = await p.list_servers()
    finally:
        await p.close()
    assert servers == []
    assert len(p.partial_failures) == 1
    leg = p.partial_failures[0]
    assert leg["reason"] == "http_403"
    assert "error_message" not in leg


@pytest.mark.asyncio
async def test_list_dns_records_managed_zones_403_carries_error_message() -> None:  # noqa: N802 — referencing GCP "managedZones" SDK endpoint name
    """Symmetric to servers — DNS managedZones.list 403 also enriches."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "iam-gap-proj"}],
    }
    body = json.dumps({
        "error": {
            "code": 403,
            "message": "Permission denied for managedZones.list.",
        },
    }).encode("utf-8")
    dns = MagicMock()
    dns.managedZones.return_value.list.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=403),
        content=body,
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("dns", "v1"): dns,
        }):
            records = await p.list_dns_records()
    finally:
        await p.close()
    assert records == []
    assert len(p.partial_failures) == 1
    leg = p.partial_failures[0]
    assert leg["reason"] == "managedZones.list_http_403"
    assert "error_message" in leg
    assert "Permission denied" in leg["error_message"]


@pytest.mark.asyncio
async def test_list_object_storage_403_carries_error_message() -> None:
    """Symmetric to servers — GCS buckets.list 403 also enriches."""
    from googleapiclient.errors import HttpError
    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "no-storage-perm"}],
    }
    body = json.dumps({
        "error": {
            "code": 403,
            "message": "user@example.com does not have storage.buckets.list",
        },
    }).encode("utf-8")
    storage = MagicMock()
    storage.buckets.return_value.list.return_value.execute.side_effect = HttpError(
        resp=MagicMock(status=403),
        content=body,
    )
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("storage", "v1"): storage,
        }):
            buckets = await p.list_object_storage()
    finally:
        await p.close()
    assert buckets == []
    assert len(p.partial_failures) == 1
    leg = p.partial_failures[0]
    assert leg["reason"] == "buckets.list_http_403"
    assert "error_message" in leg
    assert "storage.buckets.list" in leg["error_message"]


def test_gcp_403_leg_helper_omits_message_when_absent() -> None:
    """The helper should not pollute the leg dict with empty ``error_message``
    when the body has no message — keeps the JSON column tight on the
    common no-body 4xx case."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _gcp_403_leg

    exc = HttpError(resp=MagicMock(status=403), content=b"")
    leg = _gcp_403_leg("proj-x", "servers", exc, bare_reason="http_403")
    assert leg == {
        "leg": "project:proj-x",
        "capability": "servers",
        "reason": "http_403",
    }
    assert "error_message" not in leg


def test_gcp_403_leg_helper_zone_variant() -> None:
    """rrsets branch passes ``zone_name`` — leg id should include it."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _gcp_403_leg

    exc = HttpError(resp=MagicMock(status=403), content=b"")
    leg = _gcp_403_leg(
        "proj-x", "dns_records", exc,
        bare_reason="rrsets.list_http_403",
        zone_name="zone-foo",
    )
    assert leg["leg"] == "project:proj-x:zone:zone-foo"


def test_gcp_403_legs_with_same_reason_group_together_in_summary() -> None:
    """Round-26 regression guard: even though each leg now carries a
    unique-per-project ``error_message``, the ``reason`` stays bare so
    ``_summarize_leg_failures`` groups them into one ``http_403 × N``
    row — preserving the failure-notifier dedup signature."""
    from app.integrations.drivers._cloud_shared import _summarize_leg_failures
    legs = [
        {
            "leg": f"project:proj-{i}",
            "capability": "servers",
            "reason": "http_403",
            "error_message": (
                f"Permission denied on resource projects/proj-{i}"
            ),
        }
        for i in range(3)
    ]
    summary = _summarize_leg_failures(legs)
    # All three group into one row keyed on the bare reason.
    assert "http_403 × 3" in summary
    # The mode (any of the 3 different messages will do — they all
    # appear once, lexicographic-min wins the tie).
    assert "(msg: " in summary
    assert "Permission denied" in summary


def test_summarize_leg_failures_picks_most_common_error_message() -> None:
    """When messages differ in count, the most-common one is rendered."""
    from app.integrations.drivers._cloud_shared import _summarize_leg_failures
    legs = [
        {"leg": "project:a", "reason": "http_403",
         "error_message": "common message"},
        {"leg": "project:b", "reason": "http_403",
         "error_message": "common message"},
        {"leg": "project:c", "reason": "http_403",
         "error_message": "rare message"},
    ]
    summary = _summarize_leg_failures(legs)
    assert "msg: common message" in summary
    assert "rare message" not in summary


def test_redact_leg_entry_redacts_error_message() -> None:
    """The new ``error_message`` field MUST pass through the redactor —
    leg entries land in admin-readable surfaces."""
    from app.domains.cmdb.providers.exceptions import redact_leg_entry
    entry = {
        "leg": "project:foo",
        "reason": "http_403",
        "error_message": (
            "AccessKeyId=AKIA1234567890ABCDEF caused PERMISSION_DENIED"
        ),
    }
    out = redact_leg_entry(entry)
    assert out["error_message"]
    assert "AKIA1234567890ABCDEF" not in out["error_message"]
    assert "<redacted>" in out["error_message"]


def test_classify_gcp_api_disabled_shim_still_works() -> None:
    """Backwards-compat: the legacy ``_classify_gcp_api_disabled`` shim
    still returns the service name for api_disabled and None for
    everything else (including the new categories)."""
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.gcp import _classify_gcp_api_disabled

    # api_disabled → service name (legacy contract)
    body1 = json.dumps({
        "error": {
            "details": [{
                "reason": "SERVICE_DISABLED",
                "metadata": {"service": "compute.googleapis.com"},
            }],
        },
    }).encode("utf-8")
    assert _classify_gcp_api_disabled(
        HttpError(resp=MagicMock(status=403), content=body1),
    ) == "compute.googleapis.com"

    # vpc_sc → None via shim (the new classifier handles it, but the
    # legacy shim is "api_disabled only")
    body2 = json.dumps({
        "error": {
            "details": [{
                "reason": "SECURITY_POLICY_VIOLATED",
                "metadata": {"violationReason": "VPC_SERVICE_CONTROLS",
                             "service": "compute.googleapis.com"}},
            ],
        },
    }).encode("utf-8")
    assert _classify_gcp_api_disabled(
        HttpError(resp=MagicMock(status=403), content=body2),
    ) is None


# ── Round-30 (2026-05-24) execute() num_retries plumbing ────────────
#
# googleapiclient ``HttpRequest.execute(num_retries=N)`` retries on
# transport errors + 5xx responses + 429 with built-in exponential
# backoff + jitter. Pre-fix every ``.execute()`` call ran with the
# default ``num_retries=0`` so a single mid-pagination 5xx (compute
# aggregatedList page 3 of 12) dropped the rest of the project's
# instances and flipped the cron red. New behaviour: every execute
# call site passes ``num_retries=_GCP_EXECUTE_NUM_RETRIES`` (3).


def test_gcp_execute_num_retries_constant_is_three() -> None:
    """Lock the documented retry budget to 3. A drift here changes the
    "transient self-heal vs admin paged" trade-off for every GCP cron
    tick, so the constant gets a dedicated assertion."""
    from app.domains.cmdb.providers.gcp import _GCP_EXECUTE_NUM_RETRIES

    assert _GCP_EXECUTE_NUM_RETRIES == 3


@pytest.mark.asyncio
async def test_list_servers_passes_num_retries_to_execute() -> None:
    """``compute.instances().aggregatedList(...).execute(num_retries=3)``
    — verifies the kwarg actually reaches the execute call site so a
    future refactor doesn't silently drop it."""
    from app.domains.cmdb.providers.gcp import _GCP_EXECUTE_NUM_RETRIES

    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-a"}],
    }
    compute = MagicMock()
    execute_mock = compute.instances.return_value.aggregatedList.return_value.execute
    execute_mock.return_value = {"items": {}}
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("compute", "v1"): compute,
        }):
            await p.list_servers()
    finally:
        await p.close()
    # Every execute() call site that hits a paginated GCP API now
    # passes num_retries=_GCP_EXECUTE_NUM_RETRIES.
    execute_mock.assert_called_with(num_retries=_GCP_EXECUTE_NUM_RETRIES)


@pytest.mark.asyncio
async def test_list_dns_records_passes_num_retries_to_execute() -> None:
    from app.domains.cmdb.providers.gcp import _GCP_EXECUTE_NUM_RETRIES

    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-a"}],
    }
    dns = MagicMock()
    zones_execute = dns.managedZones.return_value.list.return_value.execute
    zones_execute.return_value = {"managedZones": []}
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("dns", "v1"): dns,
        }):
            await p.list_dns_records()
    finally:
        await p.close()
    zones_execute.assert_called_with(num_retries=_GCP_EXECUTE_NUM_RETRIES)


@pytest.mark.asyncio
async def test_list_object_storage_passes_num_retries_to_execute() -> None:
    from app.domains.cmdb.providers.gcp import _GCP_EXECUTE_NUM_RETRIES

    p = _provider()
    crm = MagicMock()
    crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [{"projectId": "proj-a"}],
    }
    storage = MagicMock()
    buckets_execute = storage.buckets.return_value.list.return_value.execute
    buckets_execute.return_value = {"items": []}
    try:
        with _with_svc(p, {
            ("cloudresourcemanager", "v3"): crm,
            ("storage", "v1"): storage,
        }):
            await p.list_object_storage()
    finally:
        await p.close()
    buckets_execute.assert_called_with(num_retries=_GCP_EXECUTE_NUM_RETRIES)


def test_organization_project_listing_recurses_only_configured_hierarchy() -> None:
    from app.domains.cmdb.providers.gcp import list_gcp_organization_projects

    crm = MagicMock()
    project_pages = {
        "organizations/478164292948": [
            {"projectId": "root-project", "parent": "organizations/478164292948"},
        ],
        "folders/111": [
            {"projectId": "nested-project", "parent": "folders/111"},
        ],
    }
    folder_pages = {
        "organizations/478164292948": [
            {"name": "folders/111", "parent": "organizations/478164292948"},
        ],
        "folders/111": [],
    }

    def _request(payload: dict[str, Any]) -> MagicMock:
        request = MagicMock()
        request.execute.return_value = payload
        return request

    crm.projects.return_value.list.side_effect = lambda **kwargs: _request({
        "projects": project_pages[kwargs["parent"]],
    })
    crm.folders.return_value.list.side_effect = lambda **kwargs: _request({
        "folders": folder_pages[kwargs["parent"]],
    })

    projects = list_gcp_organization_projects(crm, "478164292948")

    assert [project["projectId"] for project in projects] == [
        "root-project",
        "nested-project",
    ]
    assert {
        call.kwargs["parent"]
        for call in crm.projects.return_value.list.call_args_list
    } == {"organizations/478164292948", "folders/111"}
    assert not crm.projects.return_value.search.called


def test_organization_project_listing_rejects_parent_mismatch() -> None:
    from app.domains.cmdb.providers.gcp import list_gcp_organization_projects

    crm = MagicMock()
    crm.projects.return_value.list.return_value.execute.return_value = {
        "projects": [
            {
                "projectId": "foreign-project",
                "parent": "organizations/999999999999",
            },
        ],
    }

    with pytest.raises(RuntimeError, match="outside requested parent"):
        list_gcp_organization_projects(crm, "478164292948")


def test_organization_project_listing_rejects_repeated_page_token() -> None:
    from app.domains.cmdb.providers.gcp import list_gcp_organization_projects

    crm = MagicMock()
    crm.projects.return_value.list.return_value.execute.side_effect = [
        {"projects": [], "nextPageToken": "repeat"},
        {"projects": [], "nextPageToken": "repeat"},
    ]

    with pytest.raises(RuntimeError, match="repeated nextPageToken"):
        list_gcp_organization_projects(crm, "478164292948")