"""UCloud ``list_cost_facts`` regression suite.

Background: UCloud's cost_facts ingestion landed empty for every prod
tenant because the driver inherited the base class's
``NotImplementedError`` and the ``cmdb.cost_facts.<slug>`` cron logged
``status='not_supported'``. Downstream chargeback / dashboard /
``ai_spending`` rollups had a UCloud-shaped hole.

The fix wraps ``ListUBillDetail`` (the same API ``list_billing_daily``
uses — UCloud has no separate line-item endpoint) and aggregates to
the cost_facts natural key ``(day × resource_id × usage_type)``
before emitting ``CostFact`` rows. The CNY → USD conversion uses the
service-layer-resolved ``cny_per_usd`` rate (parity with Aliyun +
Tencent).

These tests pin every billing-correctness invariant the PR claimed:

1. **FX guard** — missing / non-positive rate raises rather than
   silently emitting ``amount_usd=0``.
2. **Aggregation** — multiple rows sharing
   ``(day, resource_id, ChargeType)`` collapse into ONE CostFact with
   summed CNY (matters because the upsert is ``ON DUPLICATE KEY
   UPDATE`` keyed on those columns and would overwrite, not sum).
3. **AmountReal priority** matches ``list_billing_daily`` — same CNY
   totals on both paths so the daily-rollup and cost_facts tables
   reconcile.
4. **Asia/Shanghai day bucketing** — late-night CST rows don't drift
   into the previous UTC day.
5. **NotOpen / UnknownAction fail-open** — tenant without billing
   add-on returns ``[]`` instead of raising (cron logs ``status='ok',
   facts=0``).
6. **Mid-pagination errors raise** — partial cost_facts persistence
   is a worse failure mode than an explicit cron failure.
7. **Page-cap raises** — stricter than ``list_billing_daily``: a
   truncated cost_facts ingest would silently understate spend.
8. **Zero-amount rows skipped** — noise; would pollute buckets with
   empty dimensions for no money.
9. **Scope tagging** — every emitted CostFact carries the caller-
   supplied scope so the ingestion service's scope_lookup hits.
10. **AWS-specific kwargs accept-and-ignore** — ``principal_tag_keys``
    and ``bq_billing`` don't reach UCloud's API.
"""
from __future__ import annotations

from datetime import date, datetime
from typing import Any
from unittest.mock import AsyncMock, patch
from zoneinfo import ZoneInfo

import pytest

from app.integrations.driver_base import CloudScopeRef, CostFact
from app.integrations.drivers.ucloud_cloud import UCloudDriver


def _make_driver() -> UCloudDriver:
    return UCloudDriver(
        integration_id="int-ucloud-test",
        config={"public_key": "PUB", "private_key": "PRIV"},
    )


def _scope() -> CloudScopeRef:
    return CloudScopeRef(
        scope_type="standalone_iam",
        external_id="org-z2uz3o",
        display_name="UCloud Production",
    )


def _ts(year: int, month: int, day: int, hour: int = 12) -> int:
    """Unix-epoch seconds for an Asia/Shanghai wall-clock day."""
    return int(
        datetime(year, month, day, hour, tzinfo=ZoneInfo("Asia/Shanghai"))
        .timestamp()
    )


_FX = 7.20  # representative CNY-per-USD rate


# ── 1. FX guard ───────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_missing_fx_rate_raises() -> None:
    """``cny_per_usd=None`` is a service-layer bug — must raise so
    cost_facts ingestion fails loudly rather than silently landing
    ``amount_usd=0`` rows that understate every UCloud chargeback.
    """
    driver = _make_driver()
    with patch.object(driver, "_call", new=AsyncMock(return_value={})):
        with pytest.raises(RuntimeError, match="cny_per_usd"):
            await driver.list_cost_facts(
                date(2026, 5, 1), date(2026, 5, 1),
                scope=_scope(), cny_per_usd=None,
            )


@pytest.mark.asyncio
async def test_zero_fx_rate_raises() -> None:
    """``cny_per_usd <= 0`` would yield ``inf`` / ZeroDivisionError on
    every row; treat as the same service-layer bug as ``None``."""
    driver = _make_driver()
    with patch.object(driver, "_call", new=AsyncMock(return_value={})):
        with pytest.raises(RuntimeError, match="cny_per_usd"):
            await driver.list_cost_facts(
                date(2026, 5, 1), date(2026, 5, 1),
                scope=_scope(), cny_per_usd=0.0,
            )


def test_driver_advertises_requires_cny_fx() -> None:
    """The cost-ingestion service inspects this flag to decide whether
    to resolve the platform FX rate. UCloud bills in CNY → must be
    True or the service passes ``cny_per_usd=None`` and the FX guard
    above fires on every cron tick.
    """
    assert UCloudDriver.requires_cny_fx is True


# ── 2 + 3 + 4. Aggregation, AmountReal priority, TZ bucketing ────


@pytest.mark.asyncio
async def test_aggregates_by_day_resource_usage_type() -> None:
    """Three rows for the same (day, resource, ChargeType) MUST
    collapse into ONE CostFact with summed CNY. The ingest writes
    ``INSERT ... ON DUPLICATE KEY UPDATE`` keyed on the same tuple;
    pre-aggregation prevents the overwrite-drops-money bug.

    Also pins: ``AmountReal`` is preferred (matches ``list_billing_
    daily``), ``Asia/Shanghai`` TZ keeps the late-evening CST row
    on the same calendar day as the morning rows.
    """
    driver = _make_driver()
    rows = [
        {
            "StartTime": _ts(2026, 5, 1, hour=2),  # CST 02:00
            "AmountReal": "30.0",
            "Amount": "100.0",  # gross — must be IGNORED
            "ResourceId": "uhost-aaa",
            "ResourceType": "uhost",
            "Region": "cn-bj2",
            "ChargeType": "Dynamic",
            "ItemDetails": [{"ProductName": "云主机 UHost"}],
            "ProjectName": "default",
        },
        {
            "StartTime": _ts(2026, 5, 1, hour=14),
            "AmountReal": "12.5",
            "ResourceId": "uhost-aaa",
            "ResourceType": "uhost",
            "Region": "cn-bj2",
            "ChargeType": "Dynamic",
            "ItemDetails": [{"ProductName": "云主机 UHost"}],
        },
        {
            "StartTime": _ts(2026, 5, 1, hour=23),  # CST late-night
            "AmountReal": "7.5",
            "ResourceId": "uhost-aaa",
            "ResourceType": "uhost",
            "Region": "cn-bj2",
            "ChargeType": "Dynamic",
            "ItemDetails": [{"ProductName": "云主机 UHost"}],
        },
    ]
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": rows, "TotalCount": 3}),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=_scope(), cny_per_usd=_FX,
        )

    assert len(facts) == 1, "three same-key rows must aggregate to one"
    f = facts[0]
    assert f.date == date(2026, 5, 1), "CST 23:00 must stay on May 1"
    assert f.amount_native == pytest.approx(50.0), "30 + 12.5 + 7.5"
    assert f.currency == "CNY"
    assert f.amount_usd == pytest.approx(50.0 / _FX)
    assert f.resource_id == "uhost-aaa"
    assert f.resource_type == "uhost"
    assert f.region == "cn-bj2"
    assert f.service == "云主机 UHost"
    assert f.usage_type == "Dynamic"
    assert f.scope is not None and f.scope.external_id == "org-z2uz3o"
    assert f.integration_id == "int-ucloud-test"


@pytest.mark.asyncio
async def test_different_charge_types_stay_separate() -> None:
    """Same resource + day but different ChargeType (prepaid vs
    pay-as-you-go add-on) MUST emit two CostFact rows — they
    fact_key on distinct ``usage_type`` values, so aggregating them
    would falsely hide the granularity that downstream chargeback
    relies on.
    """
    driver = _make_driver()
    rows = [
        {
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "1200.0",
            "ResourceId": "uhost-bbb",
            "ResourceType": "uhost",
            "ChargeType": "Year",
            "ItemDetails": [{"ProductName": "云主机 UHost"}],
        },
        {
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "5.0",
            "ResourceId": "uhost-bbb",
            "ResourceType": "uhost",
            "ChargeType": "Dynamic",
            "ItemDetails": [{"ProductName": "云主机 UHost - 流量"}],
        },
    ]
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": rows}),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=_scope(), cny_per_usd=_FX,
        )
    by_ut = {(f.usage_type, f.service): f.amount_native for f in facts}
    assert by_ut[("Year", "云主机 UHost")] == pytest.approx(1200.0)
    assert by_ut[("Dynamic", "云主机 UHost - 流量")] == pytest.approx(5.0)


@pytest.mark.asyncio
async def test_falls_back_to_amount_then_total_pay_price() -> None:
    """``Amount`` and ``TotalPayPrice`` are accepted fallbacks (mirrors
    ``list_billing_daily``) so a row missing ``AmountReal`` still
    contributes to the fact total."""
    driver = _make_driver()
    rows = [
        {
            "StartTime": _ts(2026, 5, 2),
            "Amount": "10.0",  # no AmountReal
            "ResourceId": "r-a",
            "ChargeType": "Dynamic",
        },
        {
            "StartTime": _ts(2026, 5, 2),
            "TotalPayPrice": "5.0",  # only legacy field
            "ResourceId": "r-b",
            "ChargeType": "Dynamic",
        },
    ]
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": rows}),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 2), date(2026, 5, 2),
            scope=_scope(), cny_per_usd=_FX,
        )
    totals = {f.resource_id: f.amount_native for f in facts}
    assert totals == {"r-a": pytest.approx(10.0), "r-b": pytest.approx(5.0)}


# ── 5. NotOpen / UnknownAction fail-open ──────────────────────────


@pytest.mark.asyncio
async def test_unknown_action_fails_open_to_empty() -> None:
    """Billing add-on disabled → return ``[]`` so the cron records
    ``status='ok', facts=0``. Mirrors ``list_billing_daily``."""
    driver = _make_driver()

    async def boom(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False, project_id=None) -> dict[str, Any]:
        raise RuntimeError(
            "UCloud ListUBillDetail RetCode=160: UnknownAction",
        )

    with patch.object(driver, "_call", new=AsyncMock(side_effect=boom)):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=_scope(), cny_per_usd=_FX,
        )
    assert facts == []


@pytest.mark.asyncio
async def test_not_open_fails_open_to_empty() -> None:
    """Same fail-open path as ``UnknownAction``."""
    driver = _make_driver()

    async def boom(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False, project_id=None) -> dict[str, Any]:
        raise RuntimeError(
            "UCloud ListUBillDetail RetCode=170: NotOpen billing service",
        )

    with patch.object(driver, "_call", new=AsyncMock(side_effect=boom)):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=_scope(), cny_per_usd=_FX,
        )
    assert facts == []


# ── 6. Mid-pagination errors raise ────────────────────────────────


@pytest.mark.asyncio
async def test_mid_pagination_auth_error_raises() -> None:
    """A real auth / 5xx / throttle mid-pagination MUST raise so the
    lifecycle service records the per-scope failure. Silent fail-open
    would advance the cron watermark on a credential-broken tenant
    and leave stale data accumulating.
    """
    driver = _make_driver()

    async def boom(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False, project_id=None) -> dict[str, Any]:
        # Auth error string deliberately excludes "UnknownAction" /
        # "NotOpen" so the fail-open match doesn't catch it.
        raise RuntimeError("UCloud ListUBillDetail RetCode=100: NoAuth")

    with patch.object(driver, "_call", new=AsyncMock(side_effect=boom)):
        with pytest.raises(RuntimeError, match="cost_facts"):
            await driver.list_cost_facts(
                date(2026, 5, 1), date(2026, 5, 1),
                scope=_scope(), cny_per_usd=_FX,
            )


# ── 7. Page-cap raises ────────────────────────────────────────────


@pytest.mark.asyncio
async def test_page_cap_exhaustion_raises() -> None:
    """If every page returns a full 100-row batch we'd hit the
    500-page cap with no termination signal. Truncated cost_facts
    would land a plausible-but-incomplete dataset in the DB; raise so
    the operator splits the window or bumps the cap.
    """
    driver = _make_driver()
    # Every call returns exactly 100 items so the loop never short-
    # circuits via ``len(items) < limit``.
    row = {
        "StartTime": _ts(2026, 5, 1),
        "AmountReal": "0.01",
        "ResourceId": "r-x",
        "ChargeType": "Dynamic",
    }
    full_page = {"Items": [dict(row) for _ in range(100)]}
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value=full_page),
    ):
        with pytest.raises(RuntimeError, match="page cap"):
            await driver.list_cost_facts(
                date(2026, 5, 1), date(2026, 5, 1),
                scope=_scope(), cny_per_usd=_FX,
            )


# ── 8a. Non-finite (NaN / Inf) rows skipped — data-poisoning guard ──


@pytest.mark.asyncio
async def test_non_finite_amount_rows_skipped() -> None:
    """A poisoned or corrupt UCloud response carrying NaN / Infinity
    in ``AmountReal`` MUST be dropped at row-parse time. Without the
    guard, NaN propagates into ``amount_usd`` and corrupts every
    downstream ``SUM()`` aggregate (NaN + anything = NaN), silently
    breaking finance dashboards across the whole tenant.
    """
    driver = _make_driver()
    rows = [
        {
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "NaN",
            "ResourceId": "r-nan",
            "ChargeType": "Dynamic",
        },
        {
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "Infinity",
            "ResourceId": "r-inf",
            "ChargeType": "Dynamic",
        },
        {
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "-Infinity",
            "ResourceId": "r-neg-inf",
            "ChargeType": "Dynamic",
        },
        {
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "12.5",
            "ResourceId": "r-real",
            "ChargeType": "Dynamic",
        },
    ]
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": rows}),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=_scope(), cny_per_usd=_FX,
        )
    assert {f.resource_id for f in facts} == {"r-real"}
    # Sanity: no NaN/Inf reached amount_usd downstream.
    for f in facts:
        assert f.amount_usd == f.amount_usd  # pin "not NaN"
        assert f.amount_usd not in (float("inf"), float("-inf"))


# ── 8b. Zero-amount rows skipped ─────────────────────────────────


@pytest.mark.asyncio
async def test_zero_amount_rows_skipped() -> None:
    """Zero-yuan rows (placeholders / refunds netting to 0) carry no
    money — emitting them would pollute the bucket with empty
    dimensions and pad the row count downstream for nothing."""
    driver = _make_driver()
    rows = [
        {
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "0",
            "ResourceId": "r-zero",
            "ChargeType": "Dynamic",
        },
        {
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "1.5",
            "ResourceId": "r-real",
            "ChargeType": "Dynamic",
        },
    ]
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": rows}),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=_scope(), cny_per_usd=_FX,
        )
    assert {f.resource_id for f in facts} == {"r-real"}


# ── 9. Out-of-window rows skipped ─────────────────────────────────


@pytest.mark.asyncio
async def test_rows_outside_window_dropped() -> None:
    """``ListUBillDetail`` returns the whole monthly cycle; rows whose
    StartTime falls outside [start, end] must not contribute."""
    driver = _make_driver()
    rows = [
        {
            "StartTime": _ts(2026, 5, 3),
            "AmountReal": "10.0",
            "ResourceId": "r-in",
            "ChargeType": "Dynamic",
        },
        {
            "StartTime": _ts(2026, 5, 10),  # outside window
            "AmountReal": "99.0",
            "ResourceId": "r-out",
            "ChargeType": "Dynamic",
        },
    ]
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": rows}),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 5),
            scope=_scope(), cny_per_usd=_FX,
        )
    assert {f.resource_id for f in facts} == {"r-in"}


# ── 10. AWS-specific kwargs accept-and-ignore ────────────────────


@pytest.mark.asyncio
async def test_aws_kwargs_accept_and_ignore() -> None:
    """``principal_tag_keys`` and ``bq_billing`` are AWS-driver-
    specific kwargs the ingestion service may pass uniformly. UCloud
    has no equivalent dimension; the driver MUST accept and ignore
    them without altering the outbound API call shape.
    """
    driver = _make_driver()
    captured: list[dict[str, Any]] = []

    async def capture(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False, project_id=None) -> dict[str, Any]:
        captured.append({"action": action, "params": dict(params or {})})
        return {"Items": []}

    with patch.object(driver, "_call", new=AsyncMock(side_effect=capture)):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=_scope(), cny_per_usd=_FX,
            principal_tag_keys=["Owner"],
            bq_billing={"dataset": "ignored"},
        )
    assert facts == []
    # The outbound ListUBillDetail call must NOT carry AWS dims.
    assert captured, "driver should have called ListUBillDetail"
    for entry in captured:
        assert entry["action"] == "ListUBillDetail"
        assert "PrincipalTagKey" not in entry["params"]
        assert "BigQuery" not in entry["params"]


# ── 11. Empty driver response → empty facts ──────────────────────


@pytest.mark.asyncio
async def test_empty_response_returns_empty() -> None:
    """An empty ``Items`` list (legit "no billing this cycle") MUST
    short-circuit the pagination loop and return ``[]`` without
    tripping the page-cap raise."""
    driver = _make_driver()
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": []}),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=_scope(), cny_per_usd=_FX,
        )
    assert facts == []


# ── 12. Daily-rollup ↔ cost_facts CNY reconciliation ─────────────


@pytest.mark.asyncio
async def test_cost_facts_reconcile_with_list_billing_daily() -> None:
    """The same ``ListUBillDetail`` window MUST produce identical CNY
    totals on both the ``list_billing_daily`` path (daily rollup
    table) and the ``list_cost_facts`` path (line-item fact table).

    Pre-fix the daily and fact tables would diverge if cost_facts
    quietly used ``Amount`` (gross) while daily used ``AmountReal``
    (post-discount). This test pins the same priority on both paths
    so finance can trust the two views agree.

    Also includes a zero-amount row — which cost_facts skips and
    daily preserves (`total += 0` is a no-op). Both paths still
    reconcile (0 contributes nothing to either sum), which proves
    the zero-skip divergence is harmless.
    """
    driver = _make_driver()
    rows = [
        # Coupon-discounted row: AmountReal (post-coupon) is what
        # the customer actually paid; Amount is the gross sticker.
        {
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "80.0",
            "Amount": "100.0",
            "ResourceId": "uhost-a",
            "ResourceType": "uhost",
            "ChargeType": "Month",
            "ItemDetails": [{"ProductName": "UHost"}],
        },
        # Pay-as-you-go on a different resource.
        {
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "3.5",
            "ResourceId": "ulb-b",
            "ResourceType": "ulb",
            "ChargeType": "Dynamic",
            "ItemDetails": [{"ProductName": "ULB"}],
        },
        # Zero-amount placeholder — cost_facts skips, daily includes
        # but contributes 0. Reconciliation must still hold.
        {
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "0",
            "ResourceId": "uhost-zero",
            "ResourceType": "uhost",
            "ChargeType": "Dynamic",
            "ItemDetails": [{"ProductName": "UHost"}],
        },
        # Another day in the window.
        {
            "StartTime": _ts(2026, 5, 2),
            "AmountReal": "12.0",
            "ResourceId": "uhost-a",
            "ResourceType": "uhost",
            "ChargeType": "Dynamic",
            "ItemDetails": [{"ProductName": "UHost"}],
        },
    ]
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": rows}),
    ):
        daily = await driver.list_billing_daily(
            date(2026, 5, 1), date(2026, 5, 2), scope=_scope(),
        )
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": rows}),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 2),
            scope=_scope(), cny_per_usd=_FX,
        )

    daily_cny_total = sum(d.total_cost_usd for d in daily)
    facts_cny_total = sum(f.amount_native for f in facts)
    assert daily_cny_total == pytest.approx(80.0 + 3.5 + 12.0)
    assert facts_cny_total == pytest.approx(daily_cny_total), (
        "cost_facts CNY total must reconcile to list_billing_daily — "
        "both paths read AmountReal as priority"
    )
    # USD figures derive from the same CNY base — sanity check.
    assert sum(f.amount_usd for f in facts) == pytest.approx(
        daily_cny_total / _FX,
    )


# ── 13. Multi-page pagination terminates correctly ───────────────


@pytest.mark.asyncio
async def test_multi_page_pagination_terminates_when_partial_last_page() -> None:
    """Pagination loop MUST stop on the first page where
    ``len(items) < limit``. Pre-fix this was implicit; explicit
    coverage catches a future refactor that flips the termination
    condition.

    Page 1 = 100 items (full), Page 2 = 30 items (partial); the loop
    should fetch both then stop, producing facts from BOTH pages.
    """
    driver = _make_driver()
    page1 = [
        {
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "1.0",
            "ResourceId": f"r-p1-{i}",
            "ChargeType": "Dynamic",
        }
        for i in range(100)
    ]
    page2 = [
        {
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "1.0",
            "ResourceId": f"r-p2-{i}",
            "ChargeType": "Dynamic",
        }
        for i in range(30)
    ]
    calls = {"n": 0}

    async def paginated(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False, project_id=None) -> dict[str, Any]:
        calls["n"] += 1
        # Offset comes through as a str (per ``_signed_call``); compare
        # via int.
        offset = int(str((params or {}).get("Offset") or "0"))
        if offset == 0:
            return {"Items": page1}
        if offset == 100:
            return {"Items": page2}
        # If the loop calls a third time we want the test to fail loud
        # so a future refactor that adds an off-by-one fetch is caught.
        raise AssertionError(
            f"Unexpected third page request at offset={offset}",
        )

    with patch.object(driver, "_call", new=AsyncMock(side_effect=paginated)):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=_scope(), cny_per_usd=_FX,
        )
    assert calls["n"] == 2, "loop should fetch exactly 2 pages"
    assert len(facts) == 130, "all 130 unique-resource rows must emit"
    assert sum(f.amount_native for f in facts) == pytest.approx(130.0)


# ── 14. Cross-month window covers multiple cycles ────────────────


@pytest.mark.asyncio
async def test_cross_month_window_fetches_both_cycles() -> None:
    """Window spanning ``2026-04-29..2026-05-02`` MUST call
    ``ListUBillDetail`` once with ``BillingCycle=2026-04`` and once
    with ``BillingCycle=2026-05`` — UCloud's API only filters by
    month, so the driver MUST walk every intersecting cycle. Out-of-
    window rows within each cycle still get dropped by the per-row
    date filter.
    """
    driver = _make_driver()
    cycles_called: list[str] = []

    async def by_cycle(action: str, params: dict[str, Any] | None = None, *, retryable: bool = False, project_id=None) -> dict[str, Any]:
        cycle = str((params or {}).get("BillingCycle"))
        cycles_called.append(cycle)
        if cycle == "2026-04":
            return {"Items": [
                {
                    "StartTime": _ts(2026, 4, 15),  # outside window
                    "AmountReal": "99.0",
                    "ResourceId": "r-april-out",
                    "ChargeType": "Dynamic",
                },
                {
                    "StartTime": _ts(2026, 4, 29),  # inside
                    "AmountReal": "10.0",
                    "ResourceId": "r-april-in",
                    "ChargeType": "Dynamic",
                },
            ]}
        if cycle == "2026-05":
            return {"Items": [
                {
                    "StartTime": _ts(2026, 5, 2),  # inside
                    "AmountReal": "5.0",
                    "ResourceId": "r-may-in",
                    "ChargeType": "Dynamic",
                },
                {
                    "StartTime": _ts(2026, 5, 20),  # outside
                    "AmountReal": "99.0",
                    "ResourceId": "r-may-out",
                    "ChargeType": "Dynamic",
                },
            ]}
        raise AssertionError(f"Unexpected cycle fetched: {cycle}")

    with patch.object(driver, "_call", new=AsyncMock(side_effect=by_cycle)):
        facts = await driver.list_cost_facts(
            date(2026, 4, 29), date(2026, 5, 2),
            scope=_scope(), cny_per_usd=_FX,
        )
    assert cycles_called == ["2026-04", "2026-05"], (
        "driver MUST iterate every intersecting BillingCycle"
    )
    in_window = {f.resource_id for f in facts}
    assert in_window == {"r-april-in", "r-may-in"}
    assert sum(f.amount_native for f in facts) == pytest.approx(15.0)


# ── 15. NULL natural-key components handled correctly ────────────


@pytest.mark.asyncio
async def test_null_resource_id_and_charge_type_emit_with_none() -> None:
    """A row missing both ``ResourceId`` and ``ChargeType`` MUST still
    emit a fact (with NULL components). The cost-facts ingestor's
    ``compute_fact_key`` collapses NULL via ``_FACT_KEY_NULL_MARKER``
    so the row is upsert-safe; dropping the row silently would hide
    legitimate tenant-overhead spend.
    """
    driver = _make_driver()
    rows = [{
        "StartTime": _ts(2026, 5, 1),
        "AmountReal": "2.0",
        # No ResourceId, no ChargeType
        "ItemDetails": [{"ProductName": "支持服务"}],
    }]
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": rows}),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=_scope(), cny_per_usd=_FX,
        )
    assert len(facts) == 1
    f = facts[0]
    assert f.resource_id is None
    assert f.usage_type is None
    assert f.amount_native == pytest.approx(2.0)
    assert f.service == "支持服务"


# ── 16. Duplicate-key reconciliation (the WHOLE point of pre-agg) ──


@pytest.mark.asyncio
async def test_duplicate_key_reconciles_with_daily() -> None:
    """The most important reconciliation case: when MULTIPLE rows
    share ``(day, resource_id, ChargeType)`` (the cost_facts natural
    key), the cost_facts CNY sum MUST still equal the daily-rollup
    CNY sum. This is exactly the case where ``ON DUPLICATE KEY
    UPDATE`` would overwrite-drop-money WITHOUT driver pre-aggregation
    — pinning it here is the entire reason the pre-aggregation logic
    exists.
    """
    driver = _make_driver()
    rows = [
        # Three rows same (day, resource, ChargeType) — would collide
        # on fact_key without pre-aggregation.
        {
            "StartTime": _ts(2026, 5, 1, hour=2),
            "AmountReal": "10.0",
            "ResourceId": "uhost-X",
            "ChargeType": "Dynamic",
            "ItemDetails": [{"ProductName": "UHost"}],
        },
        {
            "StartTime": _ts(2026, 5, 1, hour=8),
            "AmountReal": "5.0",
            "ResourceId": "uhost-X",
            "ChargeType": "Dynamic",
            "ItemDetails": [{"ProductName": "UHost"}],
        },
        {
            "StartTime": _ts(2026, 5, 1, hour=20),
            "AmountReal": "3.5",
            "ResourceId": "uhost-X",
            "ChargeType": "Dynamic",
            "ItemDetails": [{"ProductName": "UHost"}],
        },
    ]
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": rows}),
    ):
        daily = await driver.list_billing_daily(
            date(2026, 5, 1), date(2026, 5, 1), scope=_scope(),
        )
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": rows}),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=_scope(), cny_per_usd=_FX,
        )

    daily_cny = sum(d.total_cost_usd for d in daily)
    facts_cny = sum(f.amount_native for f in facts)
    expected = 10.0 + 5.0 + 3.5

    assert daily_cny == pytest.approx(expected)
    assert facts_cny == pytest.approx(expected)
    assert facts_cny == pytest.approx(daily_cny), (
        "duplicate-key aggregation MUST reconcile cost_facts to daily"
    )
    # Three input rows MUST collapse to exactly one fact (otherwise
    # the ingestor's ON DUPLICATE KEY UPDATE would overwrite later
    # rows on top of earlier rows, losing money silently).
    assert len(facts) == 1


# ── 17. CostFact dataclass contract sanity ───────────────────────


@pytest.mark.asyncio
async def test_returns_costfact_instances() -> None:
    """Pin the return type so a future refactor swapping the
    dataclass for a dict (and breaking the ingestor) trips a test
    instead of leaking to prod."""
    driver = _make_driver()
    rows = [{
        "StartTime": _ts(2026, 5, 1),
        "AmountReal": "1.0",
        "ResourceId": "r-1",
        "ChargeType": "Dynamic",
    }]
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": rows}),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=_scope(), cny_per_usd=_FX,
        )
    assert facts and all(isinstance(f, CostFact) for f in facts)


# ── 12. Region + friendly-name recovery (2026-07-14) ─────────────


@pytest.mark.asyncio
async def test_region_from_azgroup_and_name_from_extendinfo() -> None:
    """UCloud bill rows leave ``Region`` null and ``ResourceName`` null,
    stashing the region in ``AzGroupCName`` (CN display name) and the
    console name in ``ResourceExtendInfo`` (``KeyId="name"``). The
    driver must recover both so per-region / per-resource "小账" views
    aren't blank. Explicit ``Region`` still wins when present.
    """
    driver = _make_driver()
    rows = [
        {   # no Region → falls back to AzGroupCName; name from ExtendInfo
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "12.0",
            "ResourceId": "eip-dhadbovb",
            "ResourceType": "eip",
            "AzGroupCName": "香港",
            "ResourceExtendInfo": [{"KeyId": "name", "Value": "authing-temp"}],
            "ChargeType": "Month",
            "ItemDetails": [{"ProductName": "BGP EIP"}],
            "ProjectName": "IT_Authing",
        },
        {   # explicit Region present → wins over AzGroupCName
            "StartTime": _ts(2026, 5, 1),
            "AmountReal": "5.0",
            "ResourceId": "uhost-zzz",
            "ResourceType": "uhost",
            "Region": "cn-bj2",
            "AzGroupCName": "华北（北京2）",
            "ChargeType": "Month",
            "ItemDetails": [{"ProductName": "云主机"}],
            "ProjectName": "IT_Infra",
        },
    ]
    with patch.object(
        driver, "_call",
        new=AsyncMock(return_value={"Items": rows, "TotalCount": 2}),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=_scope(), cny_per_usd=_FX,
        )
    by_id = {f.resource_id: f for f in facts}
    eip = by_id["eip-dhadbovb"]
    assert eip.region == "香港"                 # AzGroupCName fallback
    assert eip.resource_name == "authing-temp"  # ResourceExtendInfo name
    assert eip.resource_group == "IT_Authing"   # project
    uhost = by_id["uhost-zzz"]
    assert uhost.region == "cn-bj2"             # explicit Region wins
    assert uhost.resource_name == "uhost-zzz"   # no name → id fallback