"""Per-client cloud visibility whitelist (migration 470).

A scope says WHAT KIND of cloud data a client may read; the whitelist
says WHICH accounts. Pins, in order of how badly each would hurt:

* fail-closed — a restricted client whose whitelist resolves to nothing
  sees NOTHING (never "no constraint" → whole inventory)
* list / get / patch all enforce it; out-of-whitelist is 404, never 403
  (403 would confirm the account exists to a client not entitled to
  know it does)
* an ambiguous external id resolves fine when the whitelist leaves
  exactly one candidate visible
* unset whitelist = unrestricted, i.e. every pre-migration client keeps
  working unchanged
* the shape validator rejects an unknown key rather than ignoring it —
  ``{"provider": [...]}`` (typo) must not normalize to "unrestricted"
* governance: unknown provider / unknown subscription → 422 at save
  time, and both create and update audit the grant shape
* a stored whitelist that is present but NOT a JSON object is malformed,
  and malformed fails CLOSED (reads nothing) — it must never fall back
  to the ``NULL`` meaning of "unrestricted"; the console must show it as
  broken, and clearing it must actually clear it
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import select, text
from sqlalchemy import update as sa_update
from sqlalchemy.ext.asyncio import AsyncSession

from app.api_clients.auth import ApiClientContext
from app.api_clients.models import ApiClient
from app.api_clients.scopes import (
    SCOPE_CLOUD_SUBSCRIPTIONS_READ,
    SCOPE_CLOUD_SUBSCRIPTIONS_WRITE,
)
from app.api_clients.tokens import ApiClientTokenService
from app.audit.models import AuditLog
from app.domains.cmdb.models import CmdbCloudScope
from app.domains.cmdb.open_visibility import (
    resolve_client_cloud_visibility,
    resolve_cloud_visibility,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration

pytestmark = pytest.mark.asyncio

BASE = "/api/v1/open/cloud/subscriptions"


def _headers(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


async def _mk_person(db: AsyncSession, *, email: str) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name="Owner",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:10]}",
    )
    db.add(p)
    await db.flush()
    return p


async def _mk_client(
    db: AsyncSession, *, scopes: list[str], visibility: dict | None = None,
) -> str:
    owner = await _mk_person(db, email=f"co-{uuid.uuid4().hex[:8]}@pingcap.com")
    client = ApiClient(
        id=uuid.uuid4(),
        name="BI Platform",
        slug=f"bi-{uuid.uuid4().hex[:6]}",
        owner_person_id=owner.id,
        scopes=scopes,
        cloud_visibility=visibility,
    )
    db.add(client)
    await db.flush()
    await db.refresh(client)
    _row, plaintext = await ApiClientTokenService(db).mint(
        client=client, name="t", expires_in_days=None, created_by=None,
    )
    await db.commit()
    return plaintext


async def _mk_integration(db: AsyncSession, provider: str) -> Integration:
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider=provider,
        name=f"{provider}-{uuid.uuid4().hex[:6]}",
        slug=f"{provider}-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db.add(integ)
    await db.flush()
    return integ


async def _mk_scope(
    db: AsyncSession,
    *,
    integration: Integration,
    external_id: str | None = None,
    display_name: str = "RD Compute",
) -> CmdbCloudScope:
    from datetime import UTC, datetime

    scope = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integration.id),
        scope_type="aws_account",
        external_id=external_id or f"acct-{uuid.uuid4().hex[:10]}",
        display_name=display_name,
        status="active",
        environment="dev",
        first_seen_at=datetime.now(UTC),
        last_synced_at=datetime.now(UTC),
    )
    db.add(scope)
    await db.flush()
    return scope


# ── Resolver: fail-closed ─────────────────────────────────────────────


async def test_whitelist_matching_nothing_resolves_to_false_not_unrestricted(
    db_session: AsyncSession,
):
    """The invariant everything else rests on."""
    resolved = await resolve_cloud_visibility(
        db_session, {"providers": ["provider-that-does-not-exist"]},
    )
    assert resolved.unrestricted is False
    conds = resolved.scope_conditions()
    assert len(conds) == 1
    rows = (await db_session.execute(
        select(CmdbCloudScope.id).where(*conds),
    )).scalars().all()
    assert rows == []


async def test_malformed_whitelist_resolves_to_nothing_not_unrestricted(
    db_session: AsyncSession,
):
    """A stored whitelist that is present but not a JSON object (a
    hand-edited row, a restore from an older shape) cannot be evaluated,
    and "cannot evaluate" must mean NOTHING.

    The tempting reading is the opposite one: ``cloud_visibility_is_``
    ``unrestricted`` answers True for any non-dict, so an unreadable
    value used to sail straight through the resolver as "no constraint"
    and hand a deliberately narrowed client the entire cloud inventory —
    and, with ``cloud.billing:read``, the whole spend picture.
    """
    integ = await _mk_integration(db_session, "aws")
    await _mk_scope(db_session, integration=integ)
    await db_session.commit()

    for broken in ("not-an-object", ["aws"], 7):
        resolved = await resolve_cloud_visibility(db_session, broken)
        assert resolved.unrestricted is False, broken
        assert resolved.malformed is True, broken
        rows = (await db_session.execute(
            select(CmdbCloudScope.id).where(*resolved.scope_conditions()),
        )).scalars().all()
        assert rows == [], broken


async def test_context_flag_fails_closed_even_though_the_value_coerced_to_none(
    db_session: AsyncSession,
):
    """``ApiClientContext.cloud_visibility`` coerces an unreadable value
    to ``None`` — indistinguishable from a legitimately unrestricted
    grant. The truth rides on the companion flag, so the context-shaped
    resolver must consult it; passing the coerced value alone would fail
    OPEN with no way for the caller to notice."""
    integ = await _mk_integration(db_session, "aws")
    scope = await _mk_scope(db_session, integration=integ)
    await db_session.commit()

    ctx = ApiClientContext(
        client_id=uuid.uuid4(), slug="s", name="n", scopes=frozenset(),
        cloud_visibility=None, cloud_visibility_malformed=True,
    )
    resolved = await resolve_client_cloud_visibility(db_session, ctx)
    assert resolved.unrestricted is False
    assert resolved.allows(scope) is False

    healthy = ApiClientContext(
        client_id=uuid.uuid4(), slug="s", name="n", scopes=frozenset(),
        cloud_visibility=None, cloud_visibility_malformed=False,
    )
    assert (await resolve_client_cloud_visibility(
        db_session, healthy,
    )).unrestricted is True


# ── Enforcement on the data surface ───────────────────────────────────


async def test_list_is_narrowed_to_whitelisted_provider(
    client, db_session: AsyncSession,
):
    aws = await _mk_integration(db_session, "aws")
    gcp = await _mk_integration(db_session, "gcp")
    mine = await _mk_scope(db_session, integration=aws, display_name="Mine")
    await _mk_scope(db_session, integration=gcp, display_name="NotMine")
    await db_session.commit()

    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ],
        visibility={"providers": ["aws"]},
    )
    resp = await client.get(BASE, headers=_headers(token))
    assert resp.status_code == 200
    body = resp.json()
    ids = [row["id"] for row in body["data"]]
    assert ids == [str(mine.id)]
    assert body["meta"]["total"] == 1


async def test_list_union_of_providers_and_named_subscriptions(
    client, db_session: AsyncSession,
):
    aws = await _mk_integration(db_session, "aws")
    gcp = await _mk_integration(db_session, "gcp")
    aws_scope = await _mk_scope(db_session, integration=aws)
    named_gcp = await _mk_scope(db_session, integration=gcp)
    await _mk_scope(db_session, integration=gcp)  # not named → invisible
    await db_session.commit()

    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ],
        visibility={
            "providers": ["aws"],
            "subscription_ids": [str(named_gcp.id)],
        },
    )
    resp = await client.get(BASE, headers=_headers(token))
    assert resp.status_code == 200
    ids = {row["id"] for row in resp.json()["data"]}
    assert ids == {str(aws_scope.id), str(named_gcp.id)}


async def test_unset_visibility_is_unrestricted(client, db_session: AsyncSession):
    aws = await _mk_integration(db_session, "aws")
    gcp = await _mk_integration(db_session, "gcp")
    a = await _mk_scope(db_session, integration=aws)
    g = await _mk_scope(db_session, integration=gcp)
    await db_session.commit()

    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ])
    resp = await client.get(BASE, headers=_headers(token))
    ids = {row["id"] for row in resp.json()["data"]}
    assert {str(a.id), str(g.id)} <= ids


async def test_get_out_of_whitelist_is_404_not_403(client, db_session: AsyncSession):
    aws = await _mk_integration(db_session, "aws")
    gcp = await _mk_integration(db_session, "gcp")
    await _mk_scope(db_session, integration=aws)
    hidden = await _mk_scope(db_session, integration=gcp)
    await db_session.commit()

    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ],
        visibility={"providers": ["aws"]},
    )
    resp = await client.get(f"{BASE}/{hidden.id}", headers=_headers(token))
    # 404, not 403: a 403 confirms the subscription exists.
    assert resp.status_code == 404


async def test_get_by_external_id_out_of_whitelist_is_404(
    client, db_session: AsyncSession,
):
    gcp = await _mk_integration(db_session, "gcp")
    hidden = await _mk_scope(db_session, integration=gcp, external_id="999888777666")
    await db_session.commit()

    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ],
        visibility={"providers": ["aws"]},
    )
    resp = await client.get(f"{BASE}/{hidden.external_id}", headers=_headers(token))
    assert resp.status_code == 404
    assert hidden.display_name not in resp.text


async def test_ambiguous_external_id_resolves_when_whitelist_leaves_one(
    client, db_session: AsyncSession,
):
    """Two tenants share an account number; the client is whitelisted to
    one of them. Filtering to visible rows BEFORE the ambiguity check is
    what turns a 400 the client cannot act on into its own row."""
    aws = await _mk_integration(db_session, "aws")
    gcp = await _mk_integration(db_session, "gcp")
    mine = await _mk_scope(db_session, integration=aws, external_id="dup-1234")
    await _mk_scope(db_session, integration=gcp, external_id="dup-1234")
    await db_session.commit()

    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ],
        visibility={"providers": ["aws"]},
    )
    resp = await client.get(f"{BASE}/dup-1234", headers=_headers(token))
    assert resp.status_code == 200
    assert resp.json()["data"]["id"] == str(mine.id)


async def test_patch_out_of_whitelist_is_404(client, db_session: AsyncSession):
    gcp = await _mk_integration(db_session, "gcp")
    hidden = await _mk_scope(db_session, integration=gcp)
    await db_session.commit()

    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ, SCOPE_CLOUD_SUBSCRIPTIONS_WRITE],
        visibility={"providers": ["aws"]},
    )
    resp = await client.patch(
        f"{BASE}/{hidden.id}", headers=_headers(token), json={"notes": "hi"},
    )
    assert resp.status_code == 404
    await db_session.refresh(hidden)
    assert hidden.notes != "hi"


async def test_me_reports_visibility(client, db_session: AsyncSession):
    await _mk_integration(db_session, "aws")
    await db_session.commit()
    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ],
        visibility={"providers": ["aws"]},
    )
    resp = await client.get("/api/v1/open/me", headers=_headers(token))
    assert resp.status_code == 200
    vis = resp.json()["data"]["cloud_visibility"]
    assert vis["restricted"] is True
    assert vis["providers"] == ["aws"]


# ── Governance surface ────────────────────────────────────────────────


async def test_admin_create_with_unknown_provider_rejected(
    client, admin_headers, db_session: AsyncSession,
):
    owner = await _mk_person(db_session, email=f"o-{uuid.uuid4().hex[:6]}@pingcap.com")
    await db_session.commit()
    resp = await client.post(
        "/api/v1/api-clients",
        headers=admin_headers,
        json={
            "name": "BI", "slug": f"bi-{uuid.uuid4().hex[:6]}",
            "owner_person_id": str(owner.id),
            "scopes": [SCOPE_CLOUD_SUBSCRIPTIONS_READ],
            "cloud_visibility": {"providers": ["nosuchcloud"]},
        },
    )
    assert resp.status_code in (400, 422)
    assert "nosuchcloud" in resp.text


async def test_admin_create_with_unknown_subscription_rejected(
    client, admin_headers, db_session: AsyncSession,
):
    owner = await _mk_person(db_session, email=f"o-{uuid.uuid4().hex[:6]}@pingcap.com")
    await db_session.commit()
    ghost = str(uuid.uuid4())
    resp = await client.post(
        "/api/v1/api-clients",
        headers=admin_headers,
        json={
            "name": "BI", "slug": f"bi-{uuid.uuid4().hex[:6]}",
            "owner_person_id": str(owner.id),
            "scopes": [SCOPE_CLOUD_SUBSCRIPTIONS_READ],
            "cloud_visibility": {"subscription_ids": [ghost]},
        },
    )
    assert resp.status_code in (400, 422)
    assert ghost in resp.text


async def test_admin_can_grant_and_widen_with_audit(
    client, admin_headers, db_session: AsyncSession,
):
    aws = await _mk_integration(db_session, "aws")
    scope = await _mk_scope(db_session, integration=aws)
    owner = await _mk_person(db_session, email=f"o-{uuid.uuid4().hex[:6]}@pingcap.com")
    await db_session.commit()

    created = await client.post(
        "/api/v1/api-clients",
        headers=admin_headers,
        json={
            "name": "BI", "slug": f"bi-{uuid.uuid4().hex[:6]}",
            "owner_person_id": str(owner.id),
            "scopes": [SCOPE_CLOUD_SUBSCRIPTIONS_READ],
            "cloud_visibility": {"subscription_ids": [str(scope.id)]},
        },
    )
    assert created.status_code == 200, created.text
    client_id = created.json()["data"]["id"]
    assert created.json()["data"]["cloud_visibility"] == {
        "subscription_ids": [str(scope.id)],
    }

    # Detail carries the label map so the console shows names, not UUIDs.
    detail = await client.get(f"/api/v1/api-clients/{client_id}", headers=admin_headers)
    labels = detail.json()["data"]["cloud_visibility_labels"]
    assert labels[str(scope.id)]["external_id"] == scope.external_id

    # Explicit null widens back to the whole inventory — and is audited.
    widened = await client.patch(
        f"/api/v1/api-clients/{client_id}",
        headers=admin_headers,
        json={"cloud_visibility": None},
    )
    assert widened.status_code == 200
    assert widened.json()["data"]["cloud_visibility"] is None

    rows = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.resource_type == "api_client",
            AuditLog.resource_id == client_id,
        ),
    )).scalars().all()
    details = [r.detail or {} for r in rows]
    assert any(d.get("cloud_visibility") == "restricted" for d in details)
    assert any(
        (d.get("changes") or {}).get("cloud_visibility", {}).get("to", {})
        .get("cloud_visibility") == "unrestricted"
        for d in details
    )


async def test_admin_patch_omitting_visibility_leaves_it_unchanged(
    client, admin_headers, db_session: AsyncSession,
):
    """Widening must be deliberate — never a side effect of a PATCH that
    simply didn't mention the field."""
    aws = await _mk_integration(db_session, "aws")
    owner = await _mk_person(db_session, email=f"o-{uuid.uuid4().hex[:6]}@pingcap.com")
    await db_session.commit()
    created = await client.post(
        "/api/v1/api-clients",
        headers=admin_headers,
        json={
            "name": "BI", "slug": f"bi-{uuid.uuid4().hex[:6]}",
            "owner_person_id": str(owner.id),
            "scopes": [SCOPE_CLOUD_SUBSCRIPTIONS_READ],
            "cloud_visibility": {"providers": [aws.provider]},
        },
    )
    client_id = created.json()["data"]["id"]

    patched = await client.patch(
        f"/api/v1/api-clients/{client_id}",
        headers=admin_headers,
        json={"name": "BI renamed"},
    )
    assert patched.status_code == 200
    assert patched.json()["data"]["cloud_visibility"] == {"providers": [aws.provider]}


# ── Malformed whitelist: fail closed, be visible, be repairable ───────


async def test_data_plane_fails_closed_on_a_malformed_whitelist(
    client, db_session: AsyncSession,
):
    """End-to-end proof over the route, not just the resolver: a client
    whose stored whitelist is unreadable lists ZERO subscriptions, even
    though the same request with no whitelist at all would list them."""
    aws = await _mk_integration(db_session, "aws")
    await _mk_scope(db_session, integration=aws, display_name="Visible")
    await db_session.commit()

    token = await _mk_client(
        db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ], visibility=None,
    )
    # Baseline: NULL really is unrestricted, so the assertion below
    # proves fail-closed rather than an empty fixture.
    baseline = await client.get(BASE, headers=_headers(token))
    assert baseline.json()["meta"]["total"] == 1

    # Bypass the schema validator the way a hand-edited row or a restore
    # from an older stored shape would.
    await db_session.execute(
        text("UPDATE api_clients SET cloud_visibility = :v WHERE slug LIKE 'bi-%'"),
        {"v": '"not-an-object"'},
    )
    await db_session.commit()
    # The route runs on this very session (conftest overrides get_db with
    # db_session, expire_on_commit=False), so the ApiClient loaded by the
    # baseline request is still in the identity map holding the old
    # value — a Core UPDATE does not touch it. Expire so the route
    # genuinely re-reads the malformed row.
    db_session.expire_all()

    resp = await client.get(BASE, headers=_headers(token))
    assert resp.status_code == 200, resp.text
    assert resp.json()["data"] == []
    assert resp.json()["meta"]["total"] == 0


async def test_me_does_not_call_an_unreadable_whitelist_unrestricted(
    client, db_session: AsyncSession,
):
    """`/open/me` is the ONLY place an integrator can learn their own
    reach is narrowed. Reporting `restricted: false` for a whitelist the
    data plane cannot parse would send them debugging their own query
    while every cloud endpoint returns nothing."""
    await _mk_integration(db_session, "aws")
    await db_session.commit()
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ])
    await db_session.execute(
        text("UPDATE api_clients SET cloud_visibility = :v WHERE slug LIKE 'bi-%'"),
        {"v": '"not-an-object"'},
    )
    await db_session.commit()
    db_session.expire_all()  # see note above — Core UPDATE vs identity map

    resp = await client.get("/api/v1/open/me", headers=_headers(token))
    assert resp.status_code == 200, resp.text
    vis = resp.json()["data"]["cloud_visibility"]
    assert vis["restricted"] is True
    assert vis["misconfigured"] is True
    assert "re-save" in vis["detail"]


async def test_malformed_cloud_visibility_is_visible_and_repairable(
    client, admin_headers, db_session: AsyncSession,
):
    """The admin half of the same defect. Two things must hold: the
    console must not render an unreadable whitelist as *unrestricted*,
    and the repair — clearing it — must actually clear it.

    Both failed at first: the response coerced the value to ``None``
    (reading as unrestricted), and the PATCH compared the incoming body
    against that same coerced ``None``, so an explicit-null PATCH
    compared ``None != None``, was skipped, and left the client reading
    nothing forever with no API-side way out.
    """
    owner = await _mk_person(db_session, email=f"o-{uuid.uuid4().hex[:6]}@pingcap.com")
    await db_session.commit()
    created = (await client.post(
        "/api/v1/api-clients",
        headers=admin_headers,
        json={
            "name": "BI", "slug": f"bi-{uuid.uuid4().hex[:6]}",
            "owner_person_id": str(owner.id),
            "scopes": [SCOPE_CLOUD_SUBSCRIPTIONS_READ],
        },
    )).json()["data"]

    await db_session.execute(
        sa_update(ApiClient)
        .where(ApiClient.id == uuid.UUID(created["id"]))
        .values(cloud_visibility='"not-an-object"'),
    )
    await db_session.commit()
    db_session.expire_all()  # see note above — Core UPDATE vs identity map

    detail = (await client.get(
        f"/api/v1/api-clients/{created['id']}", headers=admin_headers,
    )).json()["data"]
    assert detail["cloud_visibility_misconfigured"] is True
    # The console renders `cloud_visibility: null` as "all cloud
    # accounts", so the flag is the only thing standing between the admin
    # and the opposite of the truth.
    assert detail["cloud_visibility"] is None

    repaired = await client.patch(
        f"/api/v1/api-clients/{created['id']}",
        headers=admin_headers,
        json={"cloud_visibility": None},
    )
    assert repaired.status_code == 200, repaired.text
    assert repaired.json()["data"]["cloud_visibility_misconfigured"] is False

    row = await db_session.get(ApiClient, uuid.UUID(created["id"]))
    await db_session.refresh(row)
    assert row.cloud_visibility is None

    # The repair takes the client from "reads nothing" to "reads the
    # whole inventory" — the largest widening on this surface. It must be
    # legible in the audit trail, not recorded as unrestricted →
    # unrestricted.
    rows = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.resource_type == "api_client",
            AuditLog.resource_id == created["id"],
            AuditLog.action == "api_client.update",
        ),
    )).scalars().all()
    change = next(
        r.detail["changes"]["cloud_visibility"] for r in rows
        if "cloud_visibility" in (r.detail.get("changes") or {})
    )
    assert change["from"]["cloud_visibility"] == "misconfigured"
    assert change["to"]["cloud_visibility"] == "unrestricted"


async def test_malformed_whitelist_can_also_be_repaired_by_replacement(
    client, admin_headers, db_session: AsyncSession,
):
    """Clearing is one repair; replacing with a real whitelist is the
    other, and is the one an admin actually wants when the client was
    narrowed on purpose. The raw comparison must let it through too."""
    aws = await _mk_integration(db_session, "aws")
    # Bound to a plain string now: ``expire_all()`` below would turn a
    # later ``aws.provider`` into a lazy refresh, and a lazy load inside
    # the synchronous request-body dict raises MissingGreenlet.
    provider = aws.provider
    owner = await _mk_person(db_session, email=f"o-{uuid.uuid4().hex[:6]}@pingcap.com")
    await db_session.commit()
    created = (await client.post(
        "/api/v1/api-clients",
        headers=admin_headers,
        json={
            "name": "BI", "slug": f"bi-{uuid.uuid4().hex[:6]}",
            "owner_person_id": str(owner.id),
            "scopes": [SCOPE_CLOUD_SUBSCRIPTIONS_READ],
        },
    )).json()["data"]
    await db_session.execute(
        sa_update(ApiClient)
        .where(ApiClient.id == uuid.UUID(created["id"]))
        .values(cloud_visibility='"not-an-object"'),
    )
    await db_session.commit()
    db_session.expire_all()  # see note above — Core UPDATE vs identity map

    resp = await client.patch(
        f"/api/v1/api-clients/{created['id']}",
        headers=admin_headers,
        json={"cloud_visibility": {"providers": [provider]}},
    )
    assert resp.status_code == 200, resp.text
    assert resp.json()["data"]["cloud_visibility"] == {"providers": [provider]}
    assert resp.json()["data"]["cloud_visibility_misconfigured"] is False