"""GCP billing-account gate (PR 2026-05-19).

Background: PR #389 added a name/projectId pattern filter that caught
~3 families of personal sandbox projects (``gen-lang-client-*`` /
``"My First Project"`` / ``"Untitled project*"``), but post-deploy
analysis on 2026-05-19 found ~111 org-root personal projects with NO
regular naming pattern that the filter could not reach. The durable
fix is to gate org-root projects on **billing-account membership**:
projects linked to PingCAP's central billing account are corp; the
rest are personal sandboxes (or free-trial / disabled-billing
projects) and must not flow into ``cmdb_cloud_scopes``.

Surfaces covered here:

1. ``normalize_gcp_billing_account_name`` — config-string normalization
2. ``fetch_gcp_billed_project_ids`` — paginated API helper + the
   ``billing_enabled=False`` filter + raise-on-API-error contract
3. ``is_gcp_org_root_unbilled`` — pure predicate, defensive cases
4. ``GcpProvider._list_active_projects_sync`` — folder pass-through +
   org-root gating + ``ProviderAuthError`` on cloudbilling 403
5. ``GcpCloudDriver._list_projects_sync`` — same filter at the
   scope-listing surface (writes ``cmdb_cloud_scopes``)
6. Disabled gate (no billing account configured) — current behavior
   preserved for orgs that haven't centralized billing yet
"""
from __future__ import annotations

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

import pytest

# ───────────────────────── 1. Normalize helper ────────────────────


def test_normalize_gcp_billing_account_name_full_form_pass_through() -> None:
    from app.domains.cmdb.providers.gcp import (
        normalize_gcp_billing_account_name,
    )

    assert (
        normalize_gcp_billing_account_name(
            "billingAccounts/01D088-8F9CF2-8AF1C6",
        )
        == "billingAccounts/01D088-8F9CF2-8AF1C6"
    )


def test_normalize_gcp_billing_account_name_short_form_prefixed() -> None:
    from app.domains.cmdb.providers.gcp import (
        normalize_gcp_billing_account_name,
    )

    assert (
        normalize_gcp_billing_account_name("01D088-8F9CF2-8AF1C6")
        == "billingAccounts/01D088-8F9CF2-8AF1C6"
    )


def test_normalize_gcp_billing_account_name_empty_returns_none() -> None:
    from app.domains.cmdb.providers.gcp import (
        normalize_gcp_billing_account_name,
    )

    assert normalize_gcp_billing_account_name(None) is None
    assert normalize_gcp_billing_account_name("") is None
    assert normalize_gcp_billing_account_name("   ") is None


# ───────────────── 2. fetch_gcp_billed_project_ids ────────────────


def _fake_cloudbilling(pages: list[dict[str, Any]]) -> Any:
    """Build a MagicMock that returns ``pages`` from successive
    ``list().execute()`` calls.

    Each page is a dict shaped like
    ``{"projectBillingInfo": [...], "nextPageToken": str | None}``.
    """
    svc = MagicMock()
    execute = svc.billingAccounts.return_value.projects.return_value.list.return_value.execute
    execute.side_effect = pages
    # ``GcpCloudDriver.list_scopes`` also enumerates billing accounts as
    # grantable scopes via ``billingAccounts().list()`` (added 2026-06,
    # PR #762). Return a single empty page so that pager terminates and
    # these gate tests see project scopes only (zero billing-account
    # scopes is the intended best-effort path here).
    svc.billingAccounts.return_value.list.return_value.execute.return_value = {
        "billingAccounts": [],
    }
    return svc


def test_fetch_gcp_billed_project_ids_filters_disabled_billing() -> None:
    """``billingEnabled=False`` rows (project once linked, now
    disassociated / suspended) MUST be skipped — a disabled link does
    not vouch for the project being corp-paid."""
    from app.domains.cmdb.providers.gcp import fetch_gcp_billed_project_ids

    svc = _fake_cloudbilling([
        {
            "projectBillingInfo": [
                {"projectId": "tidbcloud-prod-000", "billingEnabled": True},
                {"projectId": "old-disabled", "billingEnabled": False},
                {"projectId": "doc-wiki", "billingEnabled": True},
            ],
            "nextPageToken": None,
        },
    ])
    out = fetch_gcp_billed_project_ids(
        svc, "billingAccounts/01D088-8F9CF2-8AF1C6",
    )
    assert out == {"tidbcloud-prod-000", "doc-wiki"}


def test_fetch_gcp_billed_project_ids_paginates_through_nextpagetoken() -> None:
    """Two-page response — the helper MUST walk ``nextPageToken``
    until exhaustion. Real billing accounts often have 50+ projects
    and the API returns 100 per page by default."""
    from app.domains.cmdb.providers.gcp import fetch_gcp_billed_project_ids

    svc = _fake_cloudbilling([
        {
            "projectBillingInfo": [
                {"projectId": "a", "billingEnabled": True},
                {"projectId": "b", "billingEnabled": True},
            ],
            "nextPageToken": "page-2",
        },
        {
            "projectBillingInfo": [
                {"projectId": "c", "billingEnabled": True},
            ],
            "nextPageToken": None,
        },
    ])
    out = fetch_gcp_billed_project_ids(
        svc, "billingAccounts/01D088-8F9CF2-8AF1C6",
    )
    assert out == {"a", "b", "c"}


def test_fetch_gcp_billed_project_ids_repeated_token_fails() -> None:
    from app.domains.cmdb.providers.gcp import fetch_gcp_billed_project_ids

    svc = _fake_cloudbilling([
        {"projectBillingInfo": [], "nextPageToken": "repeat"},
        {"projectBillingInfo": [], "nextPageToken": "repeat"},
    ])
    with pytest.raises(RuntimeError, match="repeated nextPageToken"):
        fetch_gcp_billed_project_ids(svc, "billingAccounts/AAAA-BBBB-CCCC")


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

    svc = _fake_cloudbilling([
        {"projectBillingInfo": [], "nextPageToken": "page-2"},
        {"projectBillingInfo": [], "nextPageToken": "page-3"},
    ])
    with patch.object(gcp, "_MAX_BILLED_PROJECT_PAGES", 2):
        with pytest.raises(RuntimeError, match="page cap"):
            gcp.fetch_gcp_billed_project_ids(
                svc, "billingAccounts/AAAA-BBBB-CCCC",
            )


def test_fetch_gcp_billed_project_ids_empty_response_ok() -> None:
    """No projects on the billing account — return empty set; do NOT
    raise. The gate then rejects every org-root project (as it
    should — no billed projects means everything at org-root is
    suspicious)."""
    from app.domains.cmdb.providers.gcp import fetch_gcp_billed_project_ids

    svc = _fake_cloudbilling([
        {"projectBillingInfo": [], "nextPageToken": None},
    ])
    assert fetch_gcp_billed_project_ids(
        svc, "billingAccounts/01D088-8F9CF2-8AF1C6",
    ) == set()


def test_fetch_gcp_billed_project_ids_missing_project_id_fails_closed() -> None:
    """A billing-enabled row without its identity makes the allowlist partial."""
    from app.domains.cmdb.providers.gcp import fetch_gcp_billed_project_ids

    svc = _fake_cloudbilling([
        {
            "projectBillingInfo": [
                {"projectId": "real", "billingEnabled": True},
                {"projectId": None, "billingEnabled": True},
                {"billingEnabled": True},
            ],
            "nextPageToken": None,
        },
    ])
    with pytest.raises(RuntimeError, match="without projectId"):
        fetch_gcp_billed_project_ids(
            svc, "billingAccounts/01D088-8F9CF2-8AF1C6",
        )


# ─────────────────── 3. is_gcp_org_root_unbilled ──────────────────


def test_is_gcp_org_root_unbilled_disabled_when_billed_is_none() -> None:
    """``billed_project_ids=None`` MUST disable the gate entirely
    (legacy behavior for integrations without billing centralization)."""
    from app.domains.cmdb.providers.gcp import is_gcp_org_root_unbilled

    assert is_gcp_org_root_unbilled(
        {"projectId": "x", "parent": "organizations/12345"}, None,
    ) is False


def test_is_gcp_org_root_unbilled_folders_always_pass() -> None:
    """Folder-hosted projects are inside the corp folder structure
    and pass without billing check, even if the billed set is empty."""
    from app.domains.cmdb.providers.gcp import is_gcp_org_root_unbilled

    proj = {"projectId": "tidbcloud-prod-000", "parent": "folders/988345611051"}
    assert is_gcp_org_root_unbilled(proj, set()) is False
    assert is_gcp_org_root_unbilled(proj, {"tidbcloud-prod-000"}) is False


def test_is_gcp_org_root_unbilled_billed_org_root_passes() -> None:
    """Legit org-root project that IS in the billing-account set —
    passes (returns False = not unbilled)."""
    from app.domains.cmdb.providers.gcp import is_gcp_org_root_unbilled

    proj = {"projectId": "doc-wiki", "parent": "organizations/478164292948"}
    assert is_gcp_org_root_unbilled(proj, {"doc-wiki", "other"}) is False


def test_is_gcp_org_root_unbilled_unbilled_org_root_filtered() -> None:
    """Personal sandbox at org-root NOT on the billing account —
    rejected (returns True = unbilled, filter it)."""
    from app.domains.cmdb.providers.gcp import is_gcp_org_root_unbilled

    proj = {"projectId": "bosns-project", "parent": "organizations/478164292948"}
    assert is_gcp_org_root_unbilled(proj, {"doc-wiki"}) is True


def test_is_gcp_org_root_unbilled_defensive_edge_cases() -> None:
    from app.domains.cmdb.providers.gcp import is_gcp_org_root_unbilled

    assert is_gcp_org_root_unbilled(None, set()) is False  # type: ignore[arg-type]
    assert is_gcp_org_root_unbilled({}, set()) is False
    assert is_gcp_org_root_unbilled({"projectId": "x"}, set()) is False
    assert is_gcp_org_root_unbilled(
        {"projectId": "x", "parent": None}, set(),
    ) is False
    assert is_gcp_org_root_unbilled(
        {"projectId": None, "parent": "organizations/1"}, set(),
    ) is False


# ──────── 4. Provider — _list_active_projects_sync billing gate ───


def _build_provider_with_billing(
    fields: dict[str, str],
) -> Any:
    """Construct a ``GcpProvider`` with the SA-credential parse mocked
    so we don't need a real google-auth ``service_account_info``.

    ``fields`` is the credentials.fields dict — must at least carry
    ``service_account_json``; pass ``create_scope_billing_account`` to
    activate the gate.
    """
    from app.domains.cmdb.providers.base import ProviderCredentials
    from app.domains.cmdb.providers.gcp import GcpProvider

    with patch(
        "app.domains.cmdb.providers.gcp.service_account."
        "Credentials.from_service_account_info",
        return_value=MagicMock(),
    ):
        return GcpProvider(ProviderCredentials(
            provider_id="x", provider_type="gcp", account_alias="x",
            fields={"organization_id": "478164292948", **fields},
        ))


def _adapt_parent_bounded_crm(
    crm: MagicMock,
    *,
    organization_id: str = "478164292948",
) -> None:
    """Expose strict projects/folders.list over compact legacy fixtures."""
    root = f"organizations/{organization_id}"
    payload = crm.projects.return_value.search.return_value.execute.return_value
    projects = payload.get("projects", [])
    folder_names = sorted({
        str(project.get("parent"))
        for project in projects
        if isinstance(project, dict)
        and str(project.get("parent") or "").startswith("folders/")
    })

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

    def _list_projects(**kwargs: Any) -> MagicMock:
        parent = kwargs["parent"]
        matching = [
            project
            for project in projects
            if not isinstance(project, dict)
            or str(project.get("parent") or root) == parent
        ]
        return _request({"projects": matching})

    def _list_folders(**kwargs: Any) -> MagicMock:
        folders = (
            [{"name": name, "parent": root} for name in folder_names]
            if kwargs["parent"] == root
            else []
        )
        return _request({"folders": folders})

    crm.projects.return_value.list.side_effect = _list_projects
    crm.folders.return_value.list.side_effect = _list_folders


def test_provider_billing_gate_filters_unbilled_org_root() -> None:
    """End-to-end: provider lists projects through
    ``cloudresourcemanager.projects.search``, calls cloudbilling for
    the billed set, applies all three filters (sys / personal /
    billing-gate), and returns ONLY the legit corp projects."""
    p = _build_provider_with_billing({
        "service_account_json": "{}",
        "create_scope_billing_account": "billingAccounts/01D088-8F9CF2-8AF1C6",
    })

    fake_crm = MagicMock()
    fake_crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [
            # folder-hosted — pass without billing check
            {"projectId": "tidbcloud-prod-000",
             "displayName": "tidbcloud-prod-000",
             "parent": "folders/988345611051"},
            # org-root + on billing → pass
            {"projectId": "doc-wiki",
             "displayName": "Doc-Wiki",
             "parent": "organizations/478164292948"},
            # org-root + NOT on billing → filtered (the new 111-class)
            {"projectId": "bosns-project-for-testing",
             "displayName": "Bosn's Project for Testing",
             "parent": "organizations/478164292948"},
            # personal-pattern (PR #389 defense-in-depth) still fires
            {"projectId": "gen-lang-client-9999",
             "displayName": "x",
             "parent": "organizations/478164292948"},
            # sys-* still filtered
            {"projectId": "sys-1234",
             "parent": "organizations/478164292948"},
        ],
        "nextPageToken": None,
    }
    _adapt_parent_bounded_crm(fake_crm)

    fake_billing = _fake_cloudbilling([
        {
            "projectBillingInfo": [
                {"projectId": "tidbcloud-prod-000", "billingEnabled": True},
                {"projectId": "doc-wiki", "billingEnabled": True},
            ],
            "nextPageToken": None,
        },
    ])

    def _svc_side_effect(service: str, version: str) -> Any:
        if service == "cloudresourcemanager":
            return fake_crm
        if service == "cloudbilling":
            return fake_billing
        raise AssertionError(f"unexpected _svc call: {service}/{version}")

    with patch.object(p, "_svc", side_effect=_svc_side_effect):
        out = p._list_active_projects_sync()

    assert set(out) == {"tidbcloud-prod-000", "doc-wiki"}


def test_provider_billing_gate_disabled_when_account_not_configured() -> None:
    """No ``create_scope_billing_account`` → cloudbilling is never
    called, gate is off, only the legacy filters apply. Preserves
    behavior for orgs that haven't centralized billing yet."""
    p = _build_provider_with_billing({"service_account_json": "{}"})

    fake_crm = MagicMock()
    fake_crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [
            {"projectId": "any-org-root-project",
             "displayName": "Any",
             "parent": "organizations/478164292948"},
        ],
        "nextPageToken": None,
    }
    _adapt_parent_bounded_crm(fake_crm)

    def _svc_side_effect(service: str, version: str) -> Any:
        if service == "cloudresourcemanager":
            return fake_crm
        raise AssertionError(
            f"cloudbilling MUST NOT be called when gate is disabled; "
            f"got {service}/{version}",
        )

    with patch.object(p, "_svc", side_effect=_svc_side_effect):
        out = p._list_active_projects_sync()

    assert out == ["any-org-root-project"]


@pytest.mark.parametrize("project", [{"projectId": None}, {"projectId": ""}, "bad"])
def test_provider_project_discovery_malformed_identity_fails_closed(
    project: object,
) -> None:
    from app.domains.cmdb.providers.exceptions import ProviderTransientError

    p = _build_provider_with_billing({"service_account_json": "{}"})
    fake_crm = MagicMock()
    fake_crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [project],
    }
    _adapt_parent_bounded_crm(fake_crm)

    with patch.object(p, "_svc", return_value=fake_crm):
        with pytest.raises(ProviderTransientError, match="snapshot is incomplete"):
            p._list_active_projects_sync()


def test_provider_billing_gate_raises_on_cloudbilling_403() -> None:
    """SA missing ``roles/billing.viewer`` on the billing account →
    must raise ``ProviderAuthError`` (NOT silently fall back). This is
    the load-bearing failure-mode contract: if we silently allowed,
    the next sync after a perm hiccup would re-flood ``cmdb_cloud_scopes``
    with 1000+ personal projects.
    """
    from googleapiclient.errors import HttpError

    from app.domains.cmdb.providers.exceptions import ProviderAuthError

    p = _build_provider_with_billing({
        "service_account_json": "{}",
        "create_scope_billing_account": "billingAccounts/01D088-8F9CF2-8AF1C6",
    })

    fake_billing = MagicMock()
    fake_resp = MagicMock()
    fake_resp.status = 403
    fake_resp.reason = "Forbidden"
    fake_billing.billingAccounts.return_value.projects.return_value.list.return_value.execute.side_effect = HttpError(  # type: ignore[arg-type]
        resp=fake_resp,
        content=b'{"error":{"code":403,"message":"Permission denied"}}',
    )

    def _svc_side_effect(service: str, version: str) -> Any:
        if service == "cloudbilling":
            return fake_billing
        raise AssertionError(f"unexpected _svc call: {service}/{version}")

    with patch.object(p, "_svc", side_effect=_svc_side_effect):
        with pytest.raises(ProviderAuthError, match="cloudbilling"):
            p._list_active_projects_sync()


# ────────── 5. Driver — _list_projects_sync billing gate ─────────


@pytest.mark.asyncio
async def test_driver_list_scopes_billing_gate_filters_unbilled_org_root() -> None:
    """Driver-side surface (writes ``cmdb_cloud_scopes`` rows). With
    the gate active, the 1028-project flood from 2026-05-16 is impossible
    even if upstream display-name / projectId signals all miss."""
    from app.integrations.drivers import gcp_cloud

    driver = gcp_cloud.GcpCloudDriver(
        integration_id="x",
        config={
            "service_account_json": "{}",
            "organization_id": "478164292948",
            "create_scope_billing_account": "billingAccounts/01D088-8F9CF2-8AF1C6",
        },
    )

    fake_crm = MagicMock()
    fake_crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [
            # folder-hosted — pass
            {"projectId": "tidbcloud-prod-000",
             "displayName": "tidbcloud-prod-000",
             "parent": "folders/988345611051"},
            # org-root + billed — pass
            {"projectId": "doc-wiki",
             "displayName": "Doc-Wiki",
             "parent": "organizations/478164292948"},
            # org-root + unbilled (random-name personal) — filtered
            {"projectId": "bosns-project-for-testing",
             "displayName": "Bosn's Project for Testing",
             "parent": "organizations/478164292948"},
            # org-root + unbilled (My Project NNNN — leaked past PR #389) — filtered
            {"projectId": "trim-mote-496114-v5",
             "displayName": "My Project 27153",
             "parent": "organizations/478164292948"},
            # sys-* — filtered (existing)
            {"projectId": "sys-1234",
             "parent": "organizations/478164292948"},
        ],
        "nextPageToken": None,
    }
    _adapt_parent_bounded_crm(fake_crm)

    fake_billing = _fake_cloudbilling([
        {
            "projectBillingInfo": [
                {"projectId": "tidbcloud-prod-000", "billingEnabled": True},
                {"projectId": "doc-wiki", "billingEnabled": True},
            ],
            "nextPageToken": None,
        },
    ])

    def _svc_side_effect(service: str, version: str, **_: Any) -> Any:
        if service == "cloudresourcemanager":
            return fake_crm
        if service == "cloudbilling":
            return fake_billing
        raise AssertionError(f"unexpected _svc call: {service}/{version}")

    with patch.object(driver, "_svc", side_effect=_svc_side_effect):
        scopes = await driver.list_scopes()

    ids = {s.external_id for s in scopes}
    assert ids == {"tidbcloud-prod-000", "doc-wiki"}


@pytest.mark.asyncio
async def test_driver_list_scopes_gate_disabled_without_billing_account() -> None:
    """No ``create_scope_billing_account`` → driver behaves exactly as
    before this PR (only legacy filters). Cloudbilling MUST NOT be
    invoked — verifies no spurious extra API call for orgs without
    centralized billing."""
    from app.integrations.drivers import gcp_cloud

    driver = gcp_cloud.GcpCloudDriver(
        integration_id="x",
        config={
            "service_account_json": "{}",
            "organization_id": "478164292948",
        },
    )

    fake_crm = MagicMock()
    fake_crm.projects.return_value.search.return_value.execute.return_value = {
        "projects": [
            {"projectId": "any-project",
             "displayName": "Anything",
             "parent": "organizations/478164292948"},
        ],
        "nextPageToken": None,
    }
    _adapt_parent_bounded_crm(fake_crm)

    def _svc_side_effect(service: str, version: str, **_: Any) -> Any:
        if service == "cloudresourcemanager":
            return fake_crm
        raise AssertionError(
            f"cloudbilling MUST NOT be called when gate is disabled; "
            f"got {service}/{version}",
        )

    with patch.object(driver, "_svc", side_effect=_svc_side_effect):
        scopes = await driver.list_scopes()

    assert {s.external_id for s in scopes} == {"any-project"}


@pytest.mark.asyncio
async def test_driver_list_scopes_raises_on_cloudbilling_failure() -> None:
    """Driver MUST propagate cloudbilling API errors (no silent
    fallback). Matches the provider's contract and the project-wide
    ``handler_status_must_raise`` invariant — scope sync should land
    as ``status='failed'`` so failure_notifier alerts the admin."""
    from googleapiclient.errors import HttpError

    from app.integrations.drivers import gcp_cloud

    driver = gcp_cloud.GcpCloudDriver(
        integration_id="x",
        config={
            "service_account_json": "{}",
            "organization_id": "1",
            "create_scope_billing_account": "billingAccounts/01D088-8F9CF2-8AF1C6",
        },
    )

    fake_billing = MagicMock()
    fake_resp = MagicMock()
    fake_resp.status = 403
    fake_resp.reason = "Forbidden"
    fake_billing.billingAccounts.return_value.projects.return_value.list.return_value.execute.side_effect = HttpError(  # type: ignore[arg-type]
        resp=fake_resp,
        content=b'{"error":{"code":403,"message":"Permission denied"}}',
    )

    def _svc_side_effect(service: str, version: str, **_: Any) -> Any:
        if service == "cloudbilling":
            return fake_billing
        raise AssertionError(f"unexpected _svc call: {service}/{version}")

    with patch.object(driver, "_svc", side_effect=_svc_side_effect):
        with pytest.raises(HttpError):
            await driver.list_scopes()