"""Cloud billing breakdown — /api/v1/open/cloud/billing/breakdown.

Window total split along one dimension. The pins:

* all seven dimensions group correctly and shares sum to 1
* **unallocated is a row, not a rounding error** — facts the allocation
  ladder could not attribute (``auto_owner_source='none'``) appear as
  their own bucket in department / owner groupings
* **spend that cannot be split is disclosed, not dropped** — a
  subscription served by the daily-aggregate fallback has no service /
  owner / resource dimension, so its amount lands in
  ``meta.not_broken_down`` instead of silently vanishing
* resource grouping reports the spend with no ``resource_id`` (GCP's
  standard export has none at all) and its Top-N truncation
* whitelist + measure + window caps behave as on the sibling endpoints
"""
from __future__ import annotations

import uuid
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.api_clients.models import ApiClient
from app.api_clients.scopes import SCOPE_CLOUD_BILLING_READ
from app.api_clients.tokens import ApiClientTokenService
from app.audit.models import AuditLog
from app.domains.cmdb.models import (
    CmdbCloudBillingDaily,
    CmdbCloudScope,
    CmdbCostFact,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration

pytestmark = pytest.mark.asyncio

BREAKDOWN = "/api/v1/open/cloud/billing/breakdown"


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


async def _mk_person(db: AsyncSession, *, name: str) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"{name.lower()}-{uuid.uuid4().hex[:6]}@pingcap.com",
        display_name=name,
        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, *, visibility: dict | None = None,
) -> str:
    owner = await _mk_person(db, name="Client Owner")
    client = ApiClient(
        id=uuid.uuid4(),
        name="BI",
        slug=f"bi-{uuid.uuid4().hex[:6]}",
        owner_person_id=owner.id,
        scopes=[SCOPE_CLOUD_BILLING_READ],
        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_scope(
    db: AsyncSession, *, provider: str = "aws", name: str = "RD Compute",
    environment: str = "prod",
) -> tuple[CmdbCloudScope, 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()
    scope = CmdbCloudScope(
        id=uuid.uuid4(), integration_id=str(integ.id),
        scope_type=f"{provider}_account",
        external_id=f"{provider}-{uuid.uuid4().hex[:10]}",
        display_name=name, status="active", environment=environment,
        first_seen_at=datetime.now(UTC), last_synced_at=datetime.now(UTC),
    )
    db.add(scope)
    await db.flush()
    return scope, integ


async def _mk_fact(
    db: AsyncSession, *, scope: CmdbCloudScope, integ: Integration,
    day: date, service: str = "Amazon EC2", amount: str = "10",
    department: str | None = None, owner: IdentityPerson | None = None,
    auto_owner_source: str = "scope_owner",
    resource_id: str | None = None, resource_name: str | None = None,
    tags: dict | None = None, region: str | None = None,
) -> None:
    db.add(CmdbCostFact(
        fact_key=uuid.uuid4().hex + uuid.uuid4().hex[:32],
        date=day,
        integration_id=str(integ.id),
        scope_id=str(scope.id),
        service=service,
        resource_id=resource_id,
        resource_name=resource_name,
        tags=tags,
        region=region,
        amount_usd=Decimal(amount) / 2,
        amount_native=Decimal(amount) / 2,
        amount_usd_list=Decimal(amount),
        currency="USD",
        cost_owner_department=department,
        cost_owner_person_id=str(owner.id) if owner else None,
        auto_owner_source=auto_owner_source,
    ))
    await db.flush()


def _yesterday() -> date:
    return datetime.now(UTC).date() - timedelta(days=1)


def _window(day: date) -> str:
    return f"start={day}&end={day}"


# ── Dimensions ────────────────────────────────────────────────────────


async def test_group_by_service(client, db_session: AsyncSession):
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   service="Amazon EC2", amount="60")
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   service="Amazon S3", amount="40")
    await db_session.commit()

    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=service", headers=_headers(token),
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert [(r["key"], r["amount_usd"], r["share"]) for r in body["data"]] == [
        ("Amazon EC2", 60.0, 0.6),
        ("Amazon S3", 40.0, 0.4),
    ]
    assert body["meta"]["total_usd"] == 100.0
    assert body["meta"]["group_by"] == "service"
    assert body["meta"]["measure"] == "list"


async def test_group_by_provider_and_environment(
    client, db_session: AsyncSession,
):
    aws, aws_integ = await _mk_scope(db_session, provider="aws",
                                     environment="prod")
    gcp, gcp_integ = await _mk_scope(db_session, provider="gcp",
                                     environment="dev")
    day = _yesterday()
    await _mk_fact(db_session, scope=aws, integ=aws_integ, day=day, amount="70")
    await _mk_fact(db_session, scope=gcp, integ=gcp_integ, day=day, amount="30")
    await db_session.commit()

    token = await _mk_client(db_session)
    by_provider = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=provider", headers=_headers(token),
    )
    assert {r["key"]: r["amount_usd"] for r in by_provider.json()["data"]} == {
        "aws": 70.0, "gcp": 30.0,
    }
    by_env = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=environment",
        headers=_headers(token),
    )
    assert {r["key"]: r["amount_usd"] for r in by_env.json()["data"]} == {
        "prod": 70.0, "dev": 30.0,
    }


async def test_group_by_subscription_carries_identity_fields(
    client, db_session: AsyncSession,
):
    scope, integ = await _mk_scope(db_session, name="Data Platform")
    day = _yesterday()
    await _mk_fact(db_session, scope=scope, integ=integ, day=day, amount="12")
    await db_session.commit()

    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=subscription",
        headers=_headers(token),
    )
    row = resp.json()["data"][0]
    assert row["key"] == str(scope.id)
    assert row["label"] == "Data Platform"
    assert row["external_id"] == scope.external_id
    assert row["provider"] == "aws"


async def test_group_by_department_keeps_unallocated_as_its_own_row(
    client, db_session: AsyncSession,
):
    """Unattributed spend is a bucket, never folded into someone else."""
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   amount="60", department="RD Infra")
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   amount="40", department=None, auto_owner_source="none")
    await db_session.commit()

    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=department",
        headers=_headers(token),
    )
    rows = {r["key"]: r["amount_usd"] for r in resp.json()["data"]}
    assert rows == {"RD Infra": 60.0, "unallocated": 40.0}


async def test_group_by_owner_resolves_labels(client, db_session: AsyncSession):
    scope, integ = await _mk_scope(db_session)
    person = await _mk_person(db_session, name="Zhang San")
    day = _yesterday()
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   amount="25", owner=person, auto_owner_source="rule")
    await db_session.commit()

    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=owner", headers=_headers(token),
    )
    row = resp.json()["data"][0]
    assert row["key"] == str(person.id)
    assert row["label"] == "Zhang San"
    assert row["email"] == person.work_email


async def test_group_by_resource_reports_missing_ids_and_truncation(
    client, db_session: AsyncSession,
):
    """GCP's standard export has no resource column at all — a Top-N that
    silently covered a fraction of the spend would mislead."""
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    for i in range(5):
        await _mk_fact(
            db_session, scope=scope, integ=integ, day=day,
            amount=str(10 + i), resource_id=f"i-{i}", resource_name=f"node-{i}",
        )
    await _mk_fact(db_session, scope=scope, integ=integ, day=day, amount="7")
    await db_session.commit()

    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=resource&top=2",
        headers=_headers(token),
    )
    body = resp.json()
    assert [r["key"] for r in body["data"]] == ["i-4", "i-3"]
    assert body["data"][0]["label"] == "node-4"
    assert body["meta"]["truncated_to_top"] == 2
    assert body["meta"]["without_resource_id_usd"] == 7.0
    # Total covers everything in the window, not just the shown rows.
    assert body["meta"]["total_usd"] == pytest.approx(10 + 11 + 12 + 13 + 14 + 7)


# ── Fallback spend that cannot be split ───────────────────────────────


async def test_aggregate_fallback_spend_is_disclosed_not_dropped(
    client, db_session: AsyncSession,
):
    with_facts, facts_integ = await _mk_scope(db_session, name="Has Facts")
    aggregate_only, agg_integ = await _mk_scope(db_session, name="Aggregate Only")
    day = _yesterday()
    await _mk_fact(db_session, scope=with_facts, integ=facts_integ, day=day,
                   amount="60", service="Amazon EC2")
    db_session.add(CmdbCloudBillingDaily(
        id=uuid.uuid4(),
        integration_id=str(agg_integ.id),
        scope_id=str(aggregate_only.id),
        linked_account_id=aggregate_only.external_id or "",
        date=day,
        total_cost_usd=Decimal("30"),
        total_cost_usd_list=Decimal("40"),
        cost_by_service={"Amazon EC2": 30.0},
        currency="USD",
    ))
    await db_session.commit()

    token = await _mk_client(db_session)
    # A service grouping cannot split the aggregate row (its per-service
    # JSON is the invoice basis and may not be emitted at all).
    by_service = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=service", headers=_headers(token),
    )
    body = by_service.json()
    assert {r["key"]: r["amount_usd"] for r in body["data"]} == {
        "Amazon EC2": 60.0,
    }
    disclosure = body["meta"]["not_broken_down"]
    assert disclosure["amount_usd"] == 40.0
    assert disclosure["subscriptions"][0]["name"] == "Aggregate Only"
    # …and the headline total still accounts for it.
    assert body["meta"]["total_usd"] == 100.0

    # A subscription grouping CAN absorb it — no disclosure needed.
    by_sub = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=subscription",
        headers=_headers(token),
    )
    sub_body = by_sub.json()
    assert {r["label"]: r["amount_usd"] for r in sub_body["data"]} == {
        "Has Facts": 60.0, "Aggregate Only": 40.0,
    }
    assert "not_broken_down" not in sub_body["meta"]


async def test_negative_fallback_spend_is_disclosed_not_dropped(
    client, db_session: AsyncSession,
):
    """Regression, PR #2750 review round 2.

    A scope-month can net out BELOW zero on credits / refunds /
    adjustments. Skipping non-positive fallback amounts overstated
    ``total_usd`` and put the breakdown at odds with ``/billing/daily``
    and with ``group_by=subscription``, which absorb the same value
    unconditionally.
    """
    with_facts, facts_integ = await _mk_scope(db_session, name="Has Facts")
    credit_scope, credit_integ = await _mk_scope(db_session, name="Credit Month")
    day = _yesterday()
    await _mk_fact(db_session, scope=with_facts, integ=facts_integ, day=day,
                   amount="100", service="Amazon EC2")
    db_session.add(CmdbCloudBillingDaily(
        id=uuid.uuid4(),
        integration_id=str(credit_integ.id),
        scope_id=str(credit_scope.id),
        linked_account_id=credit_scope.external_id or "",
        date=day,
        total_cost_usd=Decimal("-30"),
        total_cost_usd_list=Decimal("-30"),
        cost_by_service={}, currency="USD",
    ))
    await db_session.commit()

    token = await _mk_client(db_session)
    by_service = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=service", headers=_headers(token),
    )
    meta = by_service.json()["meta"]
    assert meta["not_broken_down"]["amount_usd"] == -30.0
    assert meta["total_usd"] == 70.0

    # …matching what the daily series and the subscription grouping say.
    daily = await client.get(
        f"/api/v1/open/cloud/billing/daily?{_window(day)}",
        headers=_headers(token),
    )
    assert sum(r["amount_usd"] for r in daily.json()["data"]) == pytest.approx(70.0)
    by_sub = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=subscription",
        headers=_headers(token),
    )
    assert by_sub.json()["meta"]["total_usd"] == 70.0


async def test_blank_service_is_labelled_not_an_empty_slice(
    client, db_session: AsyncSession,
):
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   service="", amount="8")
    await db_session.commit()
    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=service", headers=_headers(token),
    )
    row = resp.json()["data"][0]
    assert row["key"] == ""
    assert row["label"] == "unknown"


async def test_resource_group_scope_is_attributed_not_dropped(
    client, db_session: AsyncSession,
):
    """Regression, PR #2750 review.

    An Azure resource-group subscription carries its facts under
    ``(integration_id, resource_group)`` with a NULL ``scope_id``. The
    subscription / provider / environment groupings key off ``scope_id``,
    so before the fix every RG fact fell into the NULL bucket and was
    dropped when rows were folded — the breakdown under-reported and
    disagreed with ``/billing/daily`` for the same subscription.
    """
    integ = Integration(
        id=uuid.uuid4(), kind="cloud", provider="azure",
        name=f"azure-{uuid.uuid4().hex[:6]}", slug=f"azure-{uuid.uuid4().hex[:6]}",
        enabled=True, is_primary=False, capabilities=["servers"],
    )
    db_session.add(integ)
    await db_session.flush()
    rg_name = f"rg-{uuid.uuid4().hex[:8]}"
    scope = CmdbCloudScope(
        id=uuid.uuid4(), integration_id=str(integ.id),
        scope_type="azure_resource_group", external_id=rg_name,
        display_name="NexusCRM RG", status="active", environment="prod",
        first_seen_at=datetime.now(UTC), last_synced_at=datetime.now(UTC),
    )
    db_session.add(scope)
    await db_session.flush()
    day = _yesterday()
    db_session.add(CmdbCostFact(
        fact_key=uuid.uuid4().hex + uuid.uuid4().hex[:32],
        date=day, integration_id=str(integ.id), scope_id=None,
        resource_group=rg_name, service="Virtual Machines",
        amount_usd=Decimal("12"), amount_native=Decimal("12"),
        amount_usd_list=Decimal("25"), currency="USD",
    ))
    await db_session.commit()

    token = await _mk_client(db_session)
    for group_by in ("subscription", "provider", "environment"):
        resp = await client.get(
            f"{BREAKDOWN}?{_window(day)}&group_by={group_by}",
            headers=_headers(token),
        )
        body = resp.json()
        assert body["meta"]["total_usd"] == 25.0, (group_by, body["meta"])
        assert [r["amount_usd"] for r in body["data"]] == [25.0], group_by

    # …and it agrees with the sibling daily endpoint to the cent.
    daily = await client.get(
        f"/api/v1/open/cloud/billing/daily?{_window(day)}",
        headers=_headers(token),
    )
    assert [r["amount_usd"] for r in daily.json()["data"]] == [25.0]


async def test_breakdown_reports_data_through(client, db_session: AsyncSession):
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    await _mk_fact(db_session, scope=scope, integ=integ, day=day)
    await db_session.commit()
    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=service", headers=_headers(token),
    )
    assert resp.json()["meta"]["data_through"] == day.isoformat()


async def test_share_is_null_when_the_total_is_not_positive(
    client, db_session: AsyncSession,
):
    """An all-credits window has no meaningful denominator. A `share`
    computed against it would be negative for a positive charge (or blow
    up at zero), which a chart would happily render."""
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   service="Credits", amount="-50")
    await db_session.commit()
    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=service", headers=_headers(token),
    )
    row = resp.json()["data"][0]
    assert row["amount_usd"] == -50.0
    assert row["share"] is None


# ── Guardrails shared with the sibling endpoints ──────────────────────


async def test_breakdown_respects_visibility(client, db_session: AsyncSession):
    mine, mine_integ = await _mk_scope(db_session, provider="aws")
    hidden, hidden_integ = await _mk_scope(db_session, provider="gcp")
    day = _yesterday()
    await _mk_fact(db_session, scope=mine, integ=mine_integ, day=day,
                   amount="10", service="Amazon EC2")
    await _mk_fact(db_session, scope=hidden, integ=hidden_integ, day=day,
                   amount="999", service="BigQuery")
    await db_session.commit()

    token = await _mk_client(db_session, visibility={"providers": ["aws"]})
    resp = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=service", headers=_headers(token),
    )
    body = resp.json()
    assert [r["key"] for r in body["data"]] == ["Amazon EC2"]
    assert body["meta"]["total_usd"] == 10.0


async def test_unknown_group_by_rejected(client, db_session: AsyncSession):
    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?group_by=continent", headers=_headers(token),
    )
    assert resp.status_code in (400, 422)
    # The error names the vocabulary: the caller is a machine integration
    # whose author cannot read our source.
    assert "region" in resp.text and "tag:" in resp.text


async def test_breakdown_window_cap(client, db_session: AsyncSession):
    token = await _mk_client(db_session)
    today = datetime.now(UTC).date()
    resp = await client.get(
        f"{BREAKDOWN}?start={today - timedelta(days=200)}&end={today}",
        headers=_headers(token),
    )
    assert resp.status_code in (400, 422)
    assert "max is 92" in resp.text


async def test_top_is_bounded(client, db_session: AsyncSession):
    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?group_by=resource&top=500", headers=_headers(token),
    )
    assert resp.status_code == 422


async def test_breakdown_is_audited(client, db_session: AsyncSession):
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    await _mk_fact(db_session, scope=scope, integ=integ, day=day)
    await db_session.commit()

    token = await _mk_client(db_session)
    await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=service", headers=_headers(token),
    )
    rows = (await db_session.execute(
        select(AuditLog).where(AuditLog.action == "cloud.billing.read_breakdown"),
    )).scalars().all()
    assert len(rows) == 1
    assert rows[0].actor_type == "service"
    assert (rows[0].detail or {}).get("group_by") == "service"


# ── Tag dimension + multi-dimension crossing (2026-08-15) ─────────────


async def test_group_by_tag_splits_and_keeps_untagged(
    client, db_session: AsyncSession,
):
    """The headline capability, and its headline honesty rule.

    Measured on production the day this shipped: ~21% of charges carry no
    tag at all, so `untagged` is a first-class row — a tag chart that
    dropped it would hide a fifth of the money.
    """
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   amount="60", tags={"usedby": "tidb-cloud"})
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   amount="30", tags={"usedby": "aiops"})
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   amount="10", tags={})
    await db_session.commit()

    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=tag:usedby",
        headers=_headers(token),
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert {r["key"]: r["amount_usd"] for r in body["data"]} == {
        "tidb-cloud": 60.0, "aiops": 30.0, "untagged": 10.0,
    }
    assert body["meta"]["dimensions"] == ["tag:usedby"]
    coverage = body["meta"]["tag_coverage"]
    assert coverage["untagged_usd"] == 10.0
    assert coverage["tagged_usd"] == 90.0
    assert coverage["tagged_share"] == 0.9


async def test_tag_key_case_is_preserved_not_folded(
    client, db_session: AsyncSession,
):
    """`tag:UsedBy` and `tag:usedby` are different keys in the cloud's own
    data (GCP labels are lower-case by rule, AWS tags are not). Folding
    the request to lower-case would match nothing and read as "this team
    spent nothing" — the worst failure mode for a cost API."""
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   amount="40", tags={"ManagedBy": "platform"})
    await db_session.commit()

    token = await _mk_client(db_session)
    exact = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=tag:ManagedBy",
        headers=_headers(token),
    )
    assert {r["key"]: r["amount_usd"] for r in exact.json()["data"]} == {
        "platform": 40.0,
    }
    # The other spelling genuinely has no data — reported as untagged,
    # never silently merged.
    other = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=tag:managedby",
        headers=_headers(token),
    )
    assert {r["key"] for r in other.json()["data"]} == {"untagged"}


async def test_two_dimensions_cross(client, db_session: AsyncSession):
    """`tag:team × environment` — the combination the integrator asked
    for. Rows carry both keys, and the cross still sums to the window."""
    prod, prod_integ = await _mk_scope(db_session, environment="prod")
    dev, dev_integ = await _mk_scope(db_session, environment="dev")
    day = _yesterday()
    await _mk_fact(db_session, scope=prod, integ=prod_integ, day=day,
                   amount="50", tags={"team": "infra"})
    await _mk_fact(db_session, scope=dev, integ=dev_integ, day=day,
                   amount="20", tags={"team": "infra"})
    await _mk_fact(db_session, scope=prod, integ=prod_integ, day=day,
                   amount="30", tags={"team": "data"})
    await db_session.commit()

    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=tag:team,environment",
        headers=_headers(token),
    )
    body = resp.json()
    got = {
        (r["keys"]["tag:team"], r["keys"]["environment"]): r["amount_usd"]
        for r in body["data"]
    }
    assert got == {
        ("infra", "prod"): 50.0,
        ("infra", "dev"): 20.0,
        ("data", "prod"): 30.0,
    }
    assert body["meta"]["dimensions"] == ["tag:team", "environment"]
    assert body["meta"]["total_usd"] == 100.0
    # Single-dimension consumers keep the flat `key` too.
    assert all(" · " in r["key"] for r in body["data"])


async def test_new_plain_dimensions(client, db_session: AsyncSession):
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   amount="15", region="ap-southeast-1")
    await db_session.commit()
    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=region", headers=_headers(token),
    )
    assert {r["key"]: r["amount_usd"] for r in resp.json()["data"]} == {
        "ap-southeast-1": 15.0,
    }


async def test_providers_without_tags_are_named(
    client, db_session: AsyncSession,
):
    """Azure's driver writes an empty tag map, so a tag chart cannot see
    Azure at all. Saying so is what stops a consumer reading "Azure spent
    nothing" off the chart."""
    aws, aws_integ = await _mk_scope(db_session, provider="aws")
    azure, azure_integ = await _mk_scope(db_session, provider="azure")
    day = _yesterday()
    await _mk_fact(db_session, scope=aws, integ=aws_integ, day=day,
                   amount="80", tags={"team": "infra"})
    await _mk_fact(db_session, scope=azure, integ=azure_integ, day=day,
                   amount="20", tags={})
    await db_session.commit()

    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?{_window(day)}&group_by=tag:team",
        headers=_headers(token),
    )
    meta = resp.json()["meta"]
    assert meta["providers_without_tags"] == ["azure"]
    assert meta["tag_coverage"]["untagged_usd"] == 20.0


async def test_too_many_dimensions_rejected(client, db_session: AsyncSession):
    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?group_by=tag:team,environment,service",
        headers=_headers(token),
    )
    assert resp.status_code in (400, 422)
    assert "at most 2" in resp.text


async def test_malformed_tag_dimension_rejected(
    client, db_session: AsyncSession,
):
    token = await _mk_client(db_session)
    for bad in ("tag:", "tag:a\"b", "tag:" + "x" * 200):
        resp = await client.get(
            f"{BREAKDOWN}?group_by={bad}", headers=_headers(token),
        )
        assert resp.status_code in (400, 422), bad


async def test_duplicate_dimension_rejected(client, db_session: AsyncSession):
    token = await _mk_client(db_session)
    resp = await client.get(
        f"{BREAKDOWN}?group_by=service,service", headers=_headers(token),
    )
    assert resp.status_code in (400, 422)


# ── Tag-key discovery ─────────────────────────────────────────────────


async def test_tag_keys_ranks_by_spend_and_names_providers(
    client, db_session: AsyncSession,
):
    """You cannot slice by a tag you do not know exists. Ranked by money,
    so the keys that carry real spend sort above the ones a Kubernetes
    controller sprays onto every node."""
    aws, aws_integ = await _mk_scope(db_session, provider="aws")
    gcp, gcp_integ = await _mk_scope(db_session, provider="gcp")
    day = _yesterday()
    await _mk_fact(db_session, scope=aws, integ=aws_integ, day=day,
                   amount="100", tags={"usedby": "tidb", "noise": "x"})
    await _mk_fact(db_session, scope=gcp, integ=gcp_integ, day=day,
                   amount="40", tags={"usedby": "aiops"})
    await _mk_fact(db_session, scope=aws, integ=aws_integ, day=day,
                   amount="5", tags={"noise": "y"})
    await db_session.commit()

    token = await _mk_client(db_session)
    resp = await client.get(
        f"/api/v1/open/cloud/billing/tag-keys?{_window(day)}",
        headers=_headers(token),
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    rows = {r["key"]: r for r in body["data"]}
    assert [r["key"] for r in body["data"]] == ["usedby", "noise"]
    assert rows["usedby"]["sampled_spend_usd"] == 140.0
    assert rows["usedby"]["providers"] == ["aws", "gcp"]
    # The number ranks; it is not a total, and the response says so.
    assert "not the key's total spend" in body["meta"]["sampled_spend_note"]


async def test_tag_keys_respects_visibility(client, db_session: AsyncSession):
    mine, mine_integ = await _mk_scope(db_session, provider="aws")
    hidden, hidden_integ = await _mk_scope(db_session, provider="gcp")
    day = _yesterday()
    await _mk_fact(db_session, scope=mine, integ=mine_integ, day=day,
                   amount="10", tags={"mine": "yes"})
    await _mk_fact(db_session, scope=hidden, integ=hidden_integ, day=day,
                   amount="10", tags={"secret": "yes"})
    await db_session.commit()

    token = await _mk_client(db_session, visibility={"providers": ["aws"]})
    resp = await client.get(
        f"/api/v1/open/cloud/billing/tag-keys?{_window(day)}",
        headers=_headers(token),
    )
    assert [r["key"] for r in resp.json()["data"]] == ["mine"]
    assert "secret" not in resp.text