"""Cloud billing External API — /api/v1/open/cloud/billing/{daily,monthly}.

The pins, ordered by what each one prevents:

* **口径红线** — amounts are the pre-discount (list) basis and there is
  no ``measure`` parameter at all; passing one must not switch the basis
  (invoice ÷ list = the company's contract discount rate).
* **whitelist** — a billing read is narrowed by ``cloud_visibility``
  exactly like the inventory read; an unlisted subscription contributes
  nothing and is not named in the response.
* **facts-first, aggregate fallback** — a scope-month with no line items
  falls back to ``cmdb_cloud_billing_daily`` and says ``source``; without
  it the API reports $0 for a scope that is actually spending.
* **FX honesty** — a row with no resolvable rate is dropped and
  disclosed, never emitted as ``0``.
* **freshness honesty** — recent days are ``is_provisional`` and the
  running month is ``month_is_partial``.
* **month == sum(days)** — the monthly series is folded from the same
  cells the daily series returns.
* **bounds** — oversized windows 422 instead of scanning; the service
  breakdown carries its own tighter cap.
* **audit** — every read leaves one service-attributed row.
"""
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,
    SCOPE_CLOUD_SUBSCRIPTIONS_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.integrations.models import Integration

pytestmark = pytest.mark.asyncio

DAILY = "/api/v1/open/cloud/billing/daily"
MONTHLY = "/api/v1/open/cloud/billing/monthly"


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


async def _mk_client(
    db: AsyncSession, *, scopes: list[str], visibility: dict | None = None,
) -> str:
    from app.identity.person_models import IdentityPerson

    owner = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"co-{uuid.uuid4().hex[:8]}@pingcap.com",
        display_name="Owner",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:10]}",
    )
    db.add(owner)
    await db.flush()
    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_scope(
    db: AsyncSession, *, provider: str = "aws", name: str = "RD Compute",
    environment: str = "prod", scope_type: str | None = None,
    external_id: str | None = None,
) -> 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=scope_type or f"{provider}_account",
        external_id=external_id or 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


def _fact_key() -> str:
    return uuid.uuid4().hex + uuid.uuid4().hex[:0]


async def _mk_fact(
    db: AsyncSession, *, scope: CmdbCloudScope, integ: Integration, day: date,
    service: str = "Amazon EC2", amount_list: str | None = "10",
    amount_invoice: str = "4", resource_group: str | None = None,
    scope_linked: bool = True,
) -> None:
    db.add(CmdbCostFact(
        fact_key=uuid.uuid4().hex[:32] + uuid.uuid4().hex[:32],
        date=day,
        integration_id=str(integ.id),
        scope_id=str(scope.id) if scope_linked else None,
        resource_group=resource_group,
        service=service,
        amount_usd=Decimal(amount_invoice),
        amount_native=Decimal(amount_invoice),
        amount_usd_list=Decimal(amount_list) if amount_list is not None else None,
        currency="USD",
    ))
    await db.flush()


async def _mk_daily_aggregate(
    db: AsyncSession, *, scope: CmdbCloudScope, integ: Integration, day: date,
    total: str = "7", total_list: str | None = "9", currency: str = "USD",
) -> None:
    db.add(CmdbCloudBillingDaily(
        id=uuid.uuid4(),
        integration_id=str(integ.id),
        scope_id=str(scope.id),
        linked_account_id=scope.external_id or "",
        date=day,
        total_cost_usd=Decimal(total),
        total_cost_usd_list=(
            Decimal(total_list) if total_list is not None else None
        ),
        cost_by_service={"Amazon EC2": float(total)},
        currency=currency,
    ))
    await db.flush()


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


# ── Scope gating ──────────────────────────────────────────────────────


async def test_requires_billing_scope(client, db_session: AsyncSession):
    token = await _mk_client(
        db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ],
    )
    resp = await client.get(DAILY, headers=_headers(token))
    assert resp.status_code == 403
    assert "cloud.billing:read" in resp.text


# ── Measure: the confidentiality red line ─────────────────────────────


async def test_amounts_are_list_basis_and_measure_is_not_switchable(
    client, db_session: AsyncSession,
):
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    # list 10 vs invoice 4 — a response carrying 4 would be the net
    # figure, and net alongside list divides out to the discount rate.
    await _mk_fact(
        db_session, scope=scope, integ=integ, day=day,
        amount_list="10", amount_invoice="4",
    )
    await db_session.commit()

    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    for query in ("", "&measure=invoice", "&measure=net"):
        resp = await client.get(
            f"{DAILY}?start={day}&end={day}{query}", headers=_headers(token),
        )
        assert resp.status_code == 200, resp.text
        body = resp.json()
        assert body["meta"]["measure"] == "list"
        assert [r["amount_usd"] for r in body["data"]] == [10.0]


async def test_list_basis_falls_back_to_invoice_only_where_list_is_null(
    client, db_session: AsyncSession,
):
    """Transitional COALESCE semantics: a cloud whose driver has no list
    caliber yet must still report its spend, not a hole."""
    scope, integ = await _mk_scope(db_session, provider="tencent")
    day = _yesterday()
    await _mk_fact(
        db_session, scope=scope, integ=integ, day=day,
        amount_list=None, amount_invoice="6",
    )
    await db_session.commit()
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    resp = await client.get(
        f"{DAILY}?start={day}&end={day}", headers=_headers(token),
    )
    assert [r["amount_usd"] for r in resp.json()["data"]] == [6.0]


# ── Whitelist ─────────────────────────────────────────────────────────


async def test_billing_is_narrowed_by_cloud_visibility(
    client, db_session: AsyncSession,
):
    mine, mine_integ = await _mk_scope(db_session, provider="aws", name="Mine")
    hidden, hidden_integ = await _mk_scope(
        db_session, provider="gcp", name="Hidden",
    )
    day = _yesterday()
    await _mk_fact(db_session, scope=mine, integ=mine_integ, day=day,
                   amount_list="10")
    await _mk_fact(db_session, scope=hidden, integ=hidden_integ, day=day,
                   amount_list="999")
    await db_session.commit()

    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_BILLING_READ],
        visibility={"providers": ["aws"]},
    )
    resp = await client.get(
        f"{DAILY}?start={day}&end={day}", headers=_headers(token),
    )
    body = resp.json()
    assert [r["external_id"] for r in body["data"]] == [mine.external_id]
    assert "Hidden" not in resp.text
    assert 999.0 not in [r["amount_usd"] for r in body["data"]]


async def test_explicit_subscription_filter_cannot_widen_past_whitelist(
    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)
    await _mk_fact(db_session, scope=hidden, integ=hidden_integ, day=day,
                   amount_list="999")
    await db_session.commit()

    token = await _mk_client(
        db_session,
        scopes=[SCOPE_CLOUD_BILLING_READ],
        visibility={"providers": ["aws"]},
    )
    resp = await client.get(
        f"{DAILY}?start={day}&end={day}&subscription={hidden.id}",
        headers=_headers(token),
    )
    assert resp.status_code == 200
    assert resp.json()["data"] == []


# ── Source routing + honesty ──────────────────────────────────────────


async def test_falls_back_to_daily_aggregate_when_no_line_items(
    client, db_session: AsyncSession,
):
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    await _mk_daily_aggregate(
        db_session, scope=scope, integ=integ, day=day,
        total="7", total_list="9",
    )
    await db_session.commit()

    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    resp = await client.get(
        f"{DAILY}?start={day}&end={day}", headers=_headers(token),
    )
    body = resp.json()
    assert len(body["data"]) == 1
    row = body["data"][0]
    assert row["source"] == "billing_daily"
    # list column, not the net one.
    assert row["amount_usd"] == 9.0
    assert body["meta"]["sources"] == ["billing_daily"]


async def test_facts_win_over_aggregate_for_the_same_scope_month(
    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_list="10")
    await _mk_daily_aggregate(db_session, scope=scope, integ=integ, day=day,
                             total_list="9999")
    await db_session.commit()

    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    resp = await client.get(
        f"{DAILY}?start={day}&end={day}", headers=_headers(token),
    )
    rows = resp.json()["data"]
    assert [r["amount_usd"] for r in rows] == [10.0]
    assert {r["source"] for r in rows} == {"cost_facts"}


async def test_unconvertible_currency_row_is_dropped_not_zeroed(
    client, db_session: AsyncSession,
):
    """A CNY row with no resolvable rate must vanish + be disclosed. A
    ``0`` would read as "this day cost nothing"."""
    scope, integ = await _mk_scope(db_session, provider="aliyun")
    day = _yesterday()
    await _mk_daily_aggregate(
        db_session, scope=scope, integ=integ, day=day,
        total="700", total_list="700", currency="XYZ",
    )
    await db_session.commit()

    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    resp = await client.get(
        f"{DAILY}?start={day}&end={day}", headers=_headers(token),
    )
    body = resp.json()
    assert body["data"] == []
    assert "XYZ" in (body["meta"].get("unconverted_by_currency") or {})


async def test_recent_days_are_flagged_provisional(
    client, db_session: AsyncSession,
):
    scope, integ = await _mk_scope(db_session)
    today = datetime.now(UTC).date()
    old_day = today - timedelta(days=20)
    fresh_day = today - timedelta(days=1)
    await _mk_fact(db_session, scope=scope, integ=integ, day=old_day)
    await _mk_fact(db_session, scope=scope, integ=integ, day=fresh_day)
    await db_session.commit()

    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    resp = await client.get(
        f"{DAILY}?start={old_day}&end={today}", headers=_headers(token),
    )
    body = resp.json()
    flags = {r["date"]: r["is_provisional"] for r in body["data"]}
    assert flags[old_day.isoformat()] is False
    assert flags[fresh_day.isoformat()] is True
    assert body["meta"]["data_through"] == fresh_day.isoformat()
    assert body["meta"]["excludes_unallocatable"] is True


async def test_unallocated_facts_do_not_leak_into_a_subscription(
    client, db_session: AsyncSession,
):
    """Tenant-level charges (``scope_id IS NULL``) belong to no
    subscription and must not be attributed to one."""
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   amount_list="10")
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   amount_list="500", scope_linked=False, service="Tax")
    await db_session.commit()

    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    resp = await client.get(
        f"{DAILY}?start={day}&end={day}", headers=_headers(token),
    )
    assert [r["amount_usd"] for r in resp.json()["data"]] == [10.0]


# ── Service breakdown ─────────────────────────────────────────────────


async def test_service_breakdown_splits_and_sums_back(
    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_list="6")
    await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                   service="Amazon S3", amount_list="4")
    await db_session.commit()

    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    resp = await client.get(
        f"{DAILY}?start={day}&end={day}&group_by=subscription,service",
        headers=_headers(token),
    )
    rows = resp.json()["data"]
    assert {r["service"]: r["amount_usd"] for r in rows} == {
        "Amazon EC2": 6.0, "Amazon S3": 4.0,
    }

    plain = await client.get(
        f"{DAILY}?start={day}&end={day}", headers=_headers(token),
    )
    assert [r["amount_usd"] for r in plain.json()["data"]] == [10.0]
    assert "service" not in plain.json()["data"][0]


async def test_service_breakdown_has_a_tighter_window_cap(
    client, db_session: AsyncSession,
):
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    today = datetime.now(UTC).date()
    start = today - timedelta(days=200)
    ok = await client.get(
        f"{DAILY}?start={start}&end={today}", headers=_headers(token),
    )
    assert ok.status_code == 200
    too_wide = await client.get(
        f"{DAILY}?start={start}&end={today}&group_by=subscription,service",
        headers=_headers(token),
    )
    assert too_wide.status_code in (400, 422)
    assert "max is 92" in too_wide.text


# ── Bounds ────────────────────────────────────────────────────────────


async def test_daily_window_cap(client, db_session: AsyncSession):
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    today = datetime.now(UTC).date()
    resp = await client.get(
        f"{DAILY}?start={today - timedelta(days=800)}&end={today}",
        headers=_headers(token),
    )
    assert resp.status_code in (400, 422)
    assert "max is 366" in resp.text


async def test_inverted_window_rejected(client, db_session: AsyncSession):
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    today = datetime.now(UTC).date()
    resp = await client.get(
        f"{DAILY}?start={today}&end={today - timedelta(days=3)}",
        headers=_headers(token),
    )
    assert resp.status_code in (400, 422)


async def test_bad_environment_rejected(client, db_session: AsyncSession):
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    resp = await client.get(
        f"{DAILY}?environment=production", headers=_headers(token),
    )
    assert resp.status_code in (400, 422)


# ── Monthly ───────────────────────────────────────────────────────────


async def test_monthly_equals_sum_of_days_and_flags_partial(
    client, db_session: AsyncSession,
):
    scope, integ = await _mk_scope(db_session)
    today = datetime.now(UTC).date()
    first_of_month = date(today.year, today.month, 1)
    days = [first_of_month, first_of_month + timedelta(days=1)]
    if days[1] > today:
        days = [first_of_month]
    for day in days:
        await _mk_fact(db_session, scope=scope, integ=integ, day=day,
                       amount_list="10")
    await db_session.commit()

    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    month = f"{today.year:04d}-{today.month:02d}"
    monthly = await client.get(
        f"{MONTHLY}?start_month={month}&end_month={month}",
        headers=_headers(token),
    )
    assert monthly.status_code == 200, monthly.text
    rows = monthly.json()["data"]
    assert len(rows) == 1
    assert rows[0]["amount_usd"] == pytest.approx(10.0 * len(days))
    # The running month is never a closed number.
    assert rows[0]["month_is_partial"] is True
    assert rows[0]["month"] == month

    daily = await client.get(
        f"{DAILY}?start={first_of_month}&end={today}", headers=_headers(token),
    )
    assert sum(r["amount_usd"] for r in daily.json()["data"]) == pytest.approx(
        rows[0]["amount_usd"],
    )


async def test_monthly_month_format_validated(client, db_session: AsyncSession):
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    resp = await client.get(
        f"{MONTHLY}?start_month=2026/07", headers=_headers(token),
    )
    assert resp.status_code in (400, 422)
    assert "YYYY-MM" in resp.text


# ── Audit ─────────────────────────────────────────────────────────────


async def test_read_is_audited_as_service_principal(
    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, scopes=[SCOPE_CLOUD_BILLING_READ])
    await client.get(f"{DAILY}?start={day}&end={day}", headers=_headers(token))

    rows = (await db_session.execute(
        select(AuditLog).where(AuditLog.action == "cloud.billing.read_daily"),
    )).scalars().all()
    assert len(rows) == 1
    row = rows[0]
    assert row.actor_type == "service"
    assert row.actor_user_id is None
    assert (row.detail or {}).get("measure") == "list"
    assert (row.detail or {}).get("rows") == 1


# ── Catalog wiring ────────────────────────────────────────────────────


def test_billing_scope_is_in_the_catalog():
    from app.api_clients.scopes import KNOWN_SCOPES, catalog_payload

    assert SCOPE_CLOUD_BILLING_READ in KNOWN_SCOPES
    assert any(
        s["key"] == SCOPE_CLOUD_BILLING_READ and s["desc_id"] == "cloud_billing_read"
        for s in catalog_payload()
    )


# ── Review round (post-merge self-review, 2026-08-15) ─────────────────


async def test_monthly_folds_per_month_not_over_the_whole_window(
    client, db_session: AsyncSession, monkeypatch,
):
    """A monthly pull must not trip the daily materialization guard.

    Folding months out of one fully materialized daily series meant
    ``months × days × subscriptions`` cells live at once: 36 months (this
    endpoint's own documented limit) across a few hundred subscriptions
    is ~200k cells, which 422'd a request the API says it supports.
    Proving that at real volume would mean seeding 200k rows, so the cap
    is shrunk instead — with per-month folding, a window whose TOTAL
    cells exceed the cap still succeeds as long as no single month does,
    while the daily endpoint over the same window still refuses.
    """
    import app.domains.cmdb.open_billing_service as billing_service

    monkeypatch.setattr(billing_service, "MAX_RESULT_CELLS", 3)

    scope, integ = await _mk_scope(db_session)
    # Both months sit fully in the past, and each fact is placed on a day
    # that exists in every month. Anchoring the pair to the RUNNING month
    # instead put its second fact on *tomorrow* whenever the suite ran on
    # the 1st: ``monthly_billing`` clamps the window to
    # ``end=min(month_end, today)``, so that fact fell outside the query
    # and the month folded to 20.0 rather than 40.0 (green on the 3rd,
    # red on the 1st). Nothing here needs the current month — these
    # assertions are about per-month folding against the cell cap, not
    # about partiality; ``test_monthly_equals_sum_of_days_and_flags_partial``
    # owns the running month.
    today = datetime.now(UTC).date()
    later_first = (today.replace(day=1) - timedelta(days=1)).replace(day=1)
    earlier_first = (later_first - timedelta(days=1)).replace(day=1)

    # 2 cells in each of two adjacent months: 4 in total (over the cap),
    # 2 per month (under it).
    for offset in range(2):
        await _mk_fact(db_session, scope=scope, integ=integ,
                       day=earlier_first + timedelta(days=offset),
                       amount_list="10")
        await _mk_fact(db_session, scope=scope, integ=integ,
                       day=later_first + timedelta(days=offset),
                       amount_list="20")
    await db_session.commit()

    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    start_month = f"{earlier_first.year:04d}-{earlier_first.month:02d}"
    end_month = f"{later_first.year:04d}-{later_first.month:02d}"
    resp = await client.get(
        f"{MONTHLY}?start_month={start_month}&end_month={end_month}",
        headers=_headers(token),
    )
    assert resp.status_code == 200, resp.text
    amounts = {r["month"]: r["amount_usd"] for r in resp.json()["data"]}
    assert amounts[start_month] == 20.0
    assert amounts[end_month] == 40.0

    # …while the daily surface still refuses the same oversized window.
    too_many = await client.get(
        f"{DAILY}?start={earlier_first}&end={later_first + timedelta(days=1)}",
        headers=_headers(token),
    )
    assert too_many.status_code in (400, 422)
    assert "Result too large" in too_many.text


async def test_audit_failure_does_not_fail_the_read(
    client, db_session: AsyncSession, monkeypatch,
):
    """The audit write is best-effort by design — but swallowing its
    exception left the session poisoned, so ``get_db``'s closing commit
    raised and turned a served 200 into a 500."""
    scope, integ = await _mk_scope(db_session)
    day = _yesterday()
    await _mk_fact(db_session, scope=scope, integ=integ, day=day)
    await db_session.commit()

    from sqlalchemy import text as sa_text

    from app.audit import service as audit_service

    async def _boom(self, **kwargs):
        # Fail the way a real audit write fails: a statement that errors
        # inside the request's own session, leaving it unusable until
        # someone rolls it back. Raising before touching the DB would
        # not reproduce the bug (``get_db``'s closing commit is what
        # turned the served 200 into a 500).
        await self.db.execute(sa_text("SELECT * FROM table_that_does_not_exist"))

    monkeypatch.setattr(audit_service.AuditService, "log", _boom)

    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_BILLING_READ])
    resp = await client.get(
        f"{DAILY}?start={day}&end={day}", headers=_headers(token),
    )
    assert resp.status_code == 200, resp.text
    assert [r["amount_usd"] for r in resp.json()["data"]] == [10.0]