"""Unit tests for the AWS driver's BigQuery CUR billing path.

The driver supports two billing sources: Cost Explorer (CE) for AWS
SG / single-account IAM-mode integrations, and BigQuery CUR for the
PingCAP main AWS Org whose IAM user lacks ``ce:GetCostAndUsage*`` and
whose CUR is exported to ``gcp-digital-bi.aws_nonprod_billing``. These
tests cover routing, schema mapping, partition-pruning predicate
construction, and the error surface for misconfigured BQ sources.
"""
from __future__ import annotations

from datetime import date, timedelta
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest

from app.integrations.driver_base import CloudScopeRef
from app.integrations.drivers.aws_cloud import AwsCloudDriver, _BqTablePick

_BQ_CFG_BASE = {
    "access_key_id": "AKIA-test",
    "secret_access_key": "test-sk",
    "billing_source": "bigquery",
    "billing_bq_dataset": "fake-bi.aws_nonprod_billing",
    "billing_bq_credentials_integration_id": (
        "00000000-0000-0000-0000-000000000001"
    ),
    "billing_bq_table_pattern": "{account_id}_aws_billing_data",
}

_CE_CFG = {
    "access_key_id": "AKIA-test",
    "secret_access_key": "test-sk",
    "billing_source": "cost_explorer",
}

_FAKE_SA = {
    "type": "service_account",
    "project_id": "gcp-digital-infra",
    "client_email": "pch@gcp-digital-infra.iam.gserviceaccount.com",
}

_BQ_BILLING_OK = {
    "credentials_json": _FAKE_SA,
    "dataset": "fake-bi.aws_nonprod_billing",
    "table_pattern": "{account_id}_aws_billing_data",
}


def _bq_driver(**override: object) -> AwsCloudDriver:
    cfg = {**_BQ_CFG_BASE, **override}
    return AwsCloudDriver(integration_id="aws-test", config=cfg)


# ── _billing_source property ────────────────────────────────────


def test_billing_source_default_is_cost_explorer() -> None:
    d = AwsCloudDriver(
        integration_id="i", config={"access_key_id": "k", "secret_access_key": "s"},
    )
    assert d._billing_source == "cost_explorer"


def test_billing_source_normalises_whitespace_and_case() -> None:
    d = AwsCloudDriver(
        integration_id="i",
        config={"access_key_id": "k", "secret_access_key": "s",
                "billing_source": "  BigQuery "},
    )
    assert d._billing_source == "bigquery"


# ── _bq_month_predicate ─────────────────────────────────────────


def test_bq_month_predicate_single_month_one_pair() -> None:
    sql, params = AwsCloudDriver._bq_month_predicate(
        date(2026, 5, 1), date(2026, 5, 15),
    )
    assert sql == "((year = @bq_yr_0 AND month = @bq_mo_0))"
    assert [(p.name, p.value) for p in params] == [
        ("bq_yr_0", 2026), ("bq_mo_0", 5),
    ]


def test_bq_month_predicate_spans_year_boundary() -> None:
    sql, params = AwsCloudDriver._bq_month_predicate(
        date(2025, 12, 28), date(2026, 1, 3),
    )
    # Exactly two (year, month) pairs: (2025,12) and (2026,1).
    assert sql.count(" OR ") == 1
    assert [(p.name, p.value) for p in params] == [
        ("bq_yr_0", 2025), ("bq_mo_0", 12),
        ("bq_yr_1", 2026), ("bq_mo_1", 1),
    ]


def test_bq_month_predicate_thirteen_month_window() -> None:
    # A worst-case 13-month backfill spans 14 month-pairs.
    sql, params = AwsCloudDriver._bq_month_predicate(
        date(2025, 4, 1), date(2026, 5, 31),
    )
    assert sql.count(" OR ") == 13  # 14 clauses → 13 OR separators
    assert len(params) == 28  # 14 × 2 (year + month)


# ── _bq_cost_facts_windows (per-query bytes-billed chunking) ─────


def test_bq_cost_facts_windows_single_month_is_one_chunk() -> None:
    """A full single calendar month (the ``cmdb.cost_facts_monthclose``
    window) is ONE chunk → one BQ query, identical to the pre-chunking
    behaviour. The cap raise is what lets this ~70 GiB query complete."""
    w = AwsCloudDriver._bq_cost_facts_windows(
        "per_month", date(2026, 6, 1), date(2026, 6, 30),
    )
    assert w == [(date(2026, 6, 1), date(2026, 6, 30))]


def test_bq_cost_facts_windows_sub_month_is_one_chunk() -> None:
    """A window inside one calendar month stays whole (the daily cron's
    3-day rolling window, an admin's mid-month backfill)."""
    w = AwsCloudDriver._bq_cost_facts_windows(
        "per_month", date(2026, 6, 10), date(2026, 6, 20),
    )
    assert w == [(date(2026, 6, 10), date(2026, 6, 20))]


def test_bq_cost_facts_windows_splits_on_calendar_month_boundaries() -> None:
    """A multi-month window (admin-override backfill) is sliced on
    CALENDAR-MONTH boundaries — not arbitrary day counts — so each
    chunk reproduces a standalone monthly close (correct at the
    month-boundary spanning-row edge) and each query scans ~one
    month."""
    w = AwsCloudDriver._bq_cost_facts_windows(
        "per_month", date(2026, 4, 15), date(2026, 6, 10),
    )
    assert w == [
        (date(2026, 4, 15), date(2026, 4, 30)),
        (date(2026, 5, 1), date(2026, 5, 31)),
        (date(2026, 6, 1), date(2026, 6, 10)),
    ]
    # Disjoint + contiguous + full coverage by DATE(start_date).
    for prev, nxt in zip(w, w[1:]):
        assert (nxt[0] - prev[1]).days == 1
    assert w[0][0] == date(2026, 4, 15)
    assert w[-1][1] == date(2026, 6, 10)


def test_bq_cost_facts_windows_handles_year_boundary() -> None:
    w = AwsCloudDriver._bq_cost_facts_windows(
        "per_month", date(2025, 12, 20), date(2026, 1, 10),
    )
    assert w == [
        (date(2025, 12, 20), date(2025, 12, 31)),
        (date(2026, 1, 1), date(2026, 1, 10)),
    ]


def test_bq_cost_facts_windows_per_account_never_chunks() -> None:
    """CUR 1.0 per_account tables are MONTH-partitioned and one table
    per account — chunking a multi-month window gives no byte savings,
    so it stays whole."""
    w = AwsCloudDriver._bq_cost_facts_windows(
        "per_account", date(2026, 4, 15), date(2026, 6, 10),
    )
    assert w == [(date(2026, 4, 15), date(2026, 6, 10))]


# ── _bq_tag_column ──────────────────────────────────────────────


def test_bq_tag_column_lowercases_and_keeps_alnum() -> None:
    assert AwsCloudDriver._bq_tag_column("Owner") == (
        "resource_tags_user_owner"
    )
    assert AwsCloudDriver._bq_tag_column("ProjectX") == (
        "resource_tags_user_projectx"
    )


def test_bq_tag_column_normalises_non_alnum_to_underscore() -> None:
    assert AwsCloudDriver._bq_tag_column("kubernetes.io/service-name") == (
        "resource_tags_user_kubernetes_io_service_name"
    )


def test_bq_tag_column_rejects_empty() -> None:
    with pytest.raises(ValueError, match="non-empty"):
        AwsCloudDriver._bq_tag_column("")
    with pytest.raises(ValueError, match="non-empty"):
        AwsCloudDriver._bq_tag_column("   ")


# ── _bq_table_for_account ───────────────────────────────────────


def test_bq_table_substitutes_account_placeholder() -> None:
    d = _bq_driver()
    assert d._bq_table_for_account(
        "123456789012",
        dataset="fake-bi.aws_nonprod_billing",
        pattern="{account_id}_aws_billing_data",
    ) == "fake-bi.aws_nonprod_billing.123456789012_aws_billing_data"


def test_bq_table_pattern_must_contain_placeholder() -> None:
    d = _bq_driver()
    with pytest.raises(RuntimeError, match="must contain"):
        d._bq_table_for_account(
            "123456789012",
            dataset="fake-bi.aws_nonprod_billing",
            pattern="literal_pattern",
        )


# ── _bq_resolve_options ─────────────────────────────────────────


def test_bq_resolve_options_rejects_none() -> None:
    with pytest.raises(RuntimeError, match="not threaded through"):
        AwsCloudDriver._bq_resolve_options(None)


def test_bq_resolve_options_rejects_missing_credentials() -> None:
    with pytest.raises(RuntimeError, match="credentials_json missing"):
        AwsCloudDriver._bq_resolve_options({
            "dataset": "x.y", "table_pattern": "{account_id}_t",
        })


def test_bq_resolve_options_rejects_missing_dataset() -> None:
    with pytest.raises(RuntimeError, match="dataset missing"):
        AwsCloudDriver._bq_resolve_options({
            "credentials_json": _FAKE_SA,
        })


def test_bq_resolve_options_rejects_unqualified_dataset() -> None:
    with pytest.raises(RuntimeError, match="project.dataset"):
        AwsCloudDriver._bq_resolve_options({
            "credentials_json": _FAKE_SA,
            "dataset": "single_part_only",
        })


def test_bq_resolve_options_defaults_table_pattern() -> None:
    creds, ds, pat, layout, tag_schema, fact_grain = AwsCloudDriver._bq_resolve_options({
        "credentials_json": _FAKE_SA,
        "dataset": "p.d",
    })
    assert creds is _FAKE_SA
    assert ds == "p.d"
    assert pat == "{account_id}_aws_billing_data"
    # Defaults preserve CUR 1.0 layout for backwards compat — existing
    # ``aws-primary`` integrations (created before the per_month +
    # nested switches existed) keep their original behaviour through
    # the deploy of this refactor; the prod flip happens via explicit
    # config_public mutation, not via default migration.
    assert layout == "per_account"
    assert tag_schema == "flat"


# ── list_cost_facts: routing ─────────────────────────────────────


@pytest.mark.asyncio
async def test_list_cost_facts_routes_to_bq_when_source_is_bigquery() -> None:
    driver = _bq_driver()
    scope = CloudScopeRef(scope_type="aws_account", external_id="123456789012")
    with patch.object(
        driver, "_list_cost_facts_bq_sync", return_value=[],
    ) as bq_sync, patch.object(
        driver, "_list_cost_facts_sync",
    ) as ce_sync, patch.object(
        driver, "_list_cost_facts_principal_tag_sync",
    ) as ptag_sync:
        result = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    assert result == []
    bq_sync.assert_called_once()
    ce_sync.assert_not_called()
    ptag_sync.assert_not_called()


@pytest.mark.asyncio
async def test_list_cost_facts_bq_mode_rejects_principal_tag_combo() -> None:
    driver = _bq_driver()
    with pytest.raises(RuntimeError, match="incompatible with billing_source=bigquery"):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=CloudScopeRef(scope_type="aws_account", external_id="123456789012"),
            principal_tag_keys=["Owner"],
            bq_billing=_BQ_BILLING_OK,
        )


@pytest.mark.asyncio
async def test_list_cost_facts_ce_path_unchanged_when_source_is_default() -> None:
    """SG-style integration with no billing_source set continues to
    invoke the CE codepath; the new bq_billing kwarg is silently
    ignored. Guard against the routing inverting in a future refactor.
    """
    driver = AwsCloudDriver(integration_id="sg-test", config=_CE_CFG)
    scope = CloudScopeRef(scope_type="aws_account", external_id="123456789012")
    with patch.object(
        driver, "_list_cost_facts_bq_sync",
    ) as bq_sync, patch.object(
        driver, "_list_cost_facts_sync", return_value=[],
    ) as ce_sync:
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,  # passed but ignored
        )
    bq_sync.assert_not_called()
    ce_sync.assert_called_once()


# ── _list_cost_facts_bq_sync: row construction ──────────────────


# Default CUR schema used by stub tables — mirrors PingCAP prod's
# main-account CUR (built with "Include resource IDs" + all five
# tracked tag columns). Tests for degraded schemas pass an explicit
# narrower column list.
_DEFAULT_CUR_COLUMNS: tuple[str, ...] = (
    "line_item_usage_start_date",
    "line_item_usage_account_id",
    "line_item_product_code",
    "product_product_name",
    "line_item_resource_id",
    "line_item_currency_code",
    "line_item_usage_type",
    "product_region",
    "pricing_unit",
    "line_item_unblended_cost",
    "line_item_usage_amount",
    "line_item_line_item_type",
    "year",
    "month",
    "resource_tags_user_owner",
    "resource_tags_user_application",
    "resource_tags_user_environment",
    "resource_tags_user_project",
    "resource_tags_user_department",
)


def _stub_table(columns: tuple[str, ...] | list[str] = _DEFAULT_CUR_COLUMNS) -> MagicMock:
    """Build a stub BQ Table object whose ``.schema`` carries the
    given column-name set. The driver reads ``.schema[].name`` to
    drop missing CUR columns from SELECT/GROUP BY at runtime —
    see ``_bq_table_columns``."""
    return MagicMock(schema=[SimpleNamespace(name=c) for c in columns])


def _stub_bq_client_with_rows(
    table_rows: list[SimpleNamespace],
    *, table_exists: bool = True,
    columns: tuple[str, ...] | list[str] = _DEFAULT_CUR_COLUMNS,
) -> MagicMock:
    client = MagicMock()
    if not table_exists:
        from google.api_core.exceptions import NotFound
        client.get_table.side_effect = NotFound("missing")
    else:
        client.get_table.return_value = _stub_table(columns)
    job = MagicMock()
    job.result.return_value = iter(table_rows)
    client.query.return_value = job
    return client


@pytest.mark.asyncio
async def test_list_cost_facts_bq_emits_one_fact_per_row() -> None:
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    row = SimpleNamespace(
        day=date(2026, 5, 14),
        account_id="123456789012",
        product_code="AmazonEC2",
        service_name="Amazon Elastic Compute Cloud",
        resource_id="i-0abcd1234",
        currency="USD",
        region="us-west-2",
        usage_type="BoxUsage:m5.large",
        usage_unit="Hrs",
        tag_owner="alice@example.com",
        tag_application="checkout",
        tag_environment="prod",
        tag_project=None,
        tag_department=None,
        native_cost=12.34,
        usage_amount=24.0,
    )
    with patch.object(
        driver, "_bq_client",
        return_value=_stub_bq_client_with_rows([row]),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    assert len(facts) == 1
    f = facts[0]
    assert f.amount_usd == pytest.approx(12.34)
    assert f.amount_native == pytest.approx(12.34)
    assert f.currency == "USD"
    assert f.service == "Amazon Elastic Compute Cloud"
    assert f.resource_id == "i-0abcd1234"
    assert f.region == "us-west-2"
    assert f.tags == {"owner": "alice@example.com",
                      "application": "checkout",
                      "environment": "prod"}
    assert f.resource_type == "ec2/instance"


@pytest.mark.asyncio
async def test_list_cost_facts_bq_emits_gross_when_present() -> None:
    """折扣前(原价)= CUR pricing_public_on_demand_cost. When the
    aggregated row carries ``gross_native``, the fact surfaces
    amount_native_gross / amount_usd_gross (USD via same FX path)."""
    driver = _bq_driver()
    scope = CloudScopeRef(scope_type="aws_account", external_id="123456789012")
    row = SimpleNamespace(
        day=date(2026, 5, 14), account_id="123456789012",
        product_code="AmazonBedrock",
        service_name="Amazon Bedrock", resource_id="model/claude",
        currency="USD", region="us-east-1", usage_type="Tokens",
        usage_unit="tokens",
        tag_owner=None, tag_application=None, tag_environment=None,
        tag_project=None, tag_department=None,
        native_cost=40.0, gross_native=100.0, usage_amount=1000.0,
    )
    cols = (*_DEFAULT_CUR_COLUMNS, "pricing_public_on_demand_cost")
    client = _stub_bq_client_with_rows([row], columns=cols)
    with patch.object(driver, "_bq_client", return_value=client):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    assert len(facts) == 1
    f = facts[0]
    assert f.amount_native == pytest.approx(40.0)
    assert f.amount_native_gross == pytest.approx(100.0)
    assert f.amount_usd_gross == pytest.approx(100.0)
    # SQL must reference the real column when the table has it.
    sql = client.query.call_args.args[0]
    assert "pricing_public_on_demand_cost" in sql


@pytest.mark.asyncio
async def test_list_cost_facts_bq_gross_present_on_cur2_nested_path() -> None:
    """Regression: CUR 2.0 (per_month) picks carry ``columns=None`` (the
    schema is uniform, per-table probing is skipped). The gross-column
    presence check must therefore key off ``tag_schema`` (like resource_id
    / tags do), NOT off ``pick.columns`` — otherwise the prod CUR 2.0 path
    silently projects ``CAST(NULL)`` and gross never populates. Patch the
    table resolver to yield a single ``columns=None`` pick and assert the
    SQL references the real list-price column."""
    driver = _bq_driver_v2()
    scope = CloudScopeRef(scope_type="aws_account", external_id="123456789012")
    pick = _BqTablePick(
        label="2026-05",
        fq_table="fake-bi.aws_billing_no_payer.aws_billing_20260501",
        account_filter=None,
        columns=None,  # CUR 2.0 sentinel
    )
    row = SimpleNamespace(
        day=date(2026, 5, 14), account_id="123456789012",
        product_code="AmazonBedrock", service_name="Amazon Bedrock",
        resource_id="model/claude", currency="USD", region="us-east-1",
        usage_type="Tokens", usage_unit="tokens", tag_kv=[],
        native_cost=40.0, gross_native=100.0, usage_amount=1000.0,
    )
    client = _stub_bq_client_with_rows([row])
    with patch.object(
        driver, "_bq_resolve_tables", return_value=[pick],
    ), patch.object(driver, "_bq_client", return_value=client):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_CUR_V2,
        )
    sql = client.query.call_args.args[0]
    assert "SUM(pricing_public_on_demand_cost) AS gross_native" in sql
    assert "CAST(NULL AS FLOAT64)) AS gross_native" not in sql
    assert len(facts) == 1
    assert facts[0].amount_native_gross == pytest.approx(100.0)
    assert facts[0].amount_usd_gross == pytest.approx(100.0)


@pytest.mark.asyncio
async def test_list_cost_facts_bq_gross_null_when_column_absent() -> None:
    """Older per-account CURs lack pricing_public_on_demand_cost. The
    SELECT must project NULL (never reference the missing column) so the
    whole ingest doesn't 400, and the fact's gross stays None (unknown)."""
    driver = _bq_driver()
    scope = CloudScopeRef(scope_type="aws_account", external_id="123456789012")
    row = SimpleNamespace(
        day=date(2026, 5, 14), account_id="123456789012",
        product_code="AmazonEC2", service_name="EC2",
        resource_id="i-x", currency="USD", region="us-west-2",
        usage_type="BoxUsage", usage_unit="Hrs",
        tag_owner=None, tag_application=None, tag_environment=None,
        tag_project=None, tag_department=None,
        native_cost=5.0, usage_amount=1.0,
        # No gross_native attr (column absent → driver projects NULL).
    )
    client = _stub_bq_client_with_rows([row])  # _DEFAULT_CUR_COLUMNS, no pricing col
    with patch.object(driver, "_bq_client", return_value=client):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    assert len(facts) == 1
    assert facts[0].amount_native_gross is None
    assert facts[0].amount_usd_gross is None
    sql = client.query.call_args.args[0]
    assert "SUM(CAST(NULL AS FLOAT64)) AS gross_native" in sql
    assert "pricing_public_on_demand_cost" not in sql


@pytest.mark.asyncio
async def test_list_cost_facts_bq_non_usd_zeros_amount_usd() -> None:
    """AWS-China CUR partitions (Sinnet / NWCD) bill in CNY. Match the
    CE codepath's behaviour: keep native, zero out USD so the dashboard
    USD sum isn't silently inflated."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    row = SimpleNamespace(
        day=date(2026, 5, 14),
        account_id="123456789012",
        product_code="AmazonEC2",
        service_name="Amazon Elastic Compute Cloud",
        resource_id="i-cn1",
        currency="CNY",
        region="cn-north-1",
        usage_type="BoxUsage",
        usage_unit="Hrs",
        tag_owner=None, tag_application=None, tag_environment=None,
        tag_project=None, tag_department=None,
        native_cost=100.0,
        usage_amount=1.0,
    )
    with patch.object(
        driver, "_bq_client",
        return_value=_stub_bq_client_with_rows([row]),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    assert facts[0].amount_usd == 0.0
    assert facts[0].amount_native == pytest.approx(100.0)
    assert facts[0].currency == "CNY"


@pytest.mark.asyncio
async def test_list_cost_facts_bq_synthesises_resource_id_when_null() -> None:
    """Some CUR rows (SavingsPlan negation pre-filter, marketplace fees)
    have resource_id=NULL. The driver synthesises a stable key so the
    upsert's natural-key hash stays unique."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    row = SimpleNamespace(
        day=date(2026, 5, 14),
        account_id="123456789012",
        product_code="AWSMarketplaceFee",
        service_name="AWS Marketplace",
        resource_id=None,
        currency="USD",
        region=None,
        usage_type=None,
        usage_unit=None,
        tag_owner=None, tag_application=None, tag_environment=None,
        tag_project=None, tag_department=None,
        native_cost=5.0,
        usage_amount=None,
    )
    with patch.object(
        driver, "_bq_client",
        return_value=_stub_bq_client_with_rows([row]),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    assert facts[0].resource_id == (
        "aws-account:123456789012:AWSMarketplaceFee"
    )
    assert facts[0].resource_type == "aws-aggregated"


@pytest.mark.asyncio
async def test_list_cost_facts_bq_buckets_edp_discount_separately() -> None:
    """CUR 2.0 surfaces enterprise discount and private-rate discount as
    negative line items with NULL product_code. Without the
    ``discount_bucket`` CASE the two would collide on the synthesised
    ``aws-account:<acc>:None`` resource_id (same fact_key → one row
    wins the upsert, the other is lost).

    Two synthetic rows below: one EDP, one Private Rate. Driver should
    emit TWO facts with distinct resource_ids and distinct
    ``service`` labels."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="385595570414",
    )
    edp_row = SimpleNamespace(
        day=date(2026, 5, 1),
        account_id="385595570414",
        product_code=None,
        service_name=None,
        discount_bucket="EDP Discount",
        resource_id=None,
        currency="USD",
        region=None,
        usage_type=None,
        usage_unit=None,
        tag_owner=None, tag_application=None, tag_environment=None,
        tag_project=None, tag_department=None,
        native_cost=-566.39,
        usage_amount=None,
    )
    prd_row = SimpleNamespace(
        day=date(2026, 5, 1),
        account_id="385595570414",
        product_code=None,
        service_name=None,
        discount_bucket="Private Rate Discount",
        resource_id=None,
        currency="USD",
        region=None,
        usage_type=None,
        usage_unit=None,
        tag_owner=None, tag_application=None, tag_environment=None,
        tag_project=None, tag_department=None,
        native_cost=-45.31,
        usage_amount=None,
    )
    with patch.object(
        driver, "_bq_client",
        return_value=_stub_bq_client_with_rows([edp_row, prd_row]),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    assert len(facts) == 2
    by_service = {f.service: f for f in facts}
    assert "EDP Discount" in by_service
    assert "Private Rate Discount" in by_service
    assert by_service["EDP Discount"].amount_native == pytest.approx(-566.39)
    assert by_service["Private Rate Discount"].amount_native == pytest.approx(-45.31)
    assert (
        by_service["EDP Discount"].resource_id
        == "aws-account:385595570414:EDP Discount"
    )
    assert (
        by_service["Private Rate Discount"].resource_id
        == "aws-account:385595570414:Private Rate Discount"
    )


@pytest.mark.asyncio
async def test_list_cost_facts_bq_skips_missing_table() -> None:
    """When a linked account doesn't have a CUR export table yet
    (newly added accounts, the Org master itself), the driver returns
    [] rather than crashing the cron — matches the CE codepath's
    "no data" semantics."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    with patch.object(
        driver, "_bq_client",
        return_value=_stub_bq_client_with_rows([], table_exists=False),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    assert facts == []


@pytest.mark.asyncio
async def test_list_cost_facts_bq_requires_scope() -> None:
    driver = _bq_driver()
    with pytest.raises(NotImplementedError, match="explicit scope"):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=None, bq_billing=_BQ_BILLING_OK,
        )


@pytest.mark.asyncio
async def test_list_cost_facts_bq_rejects_non_12_digit_scope() -> None:
    driver = _bq_driver()
    with pytest.raises(RuntimeError, match="12-digit"):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=CloudScopeRef(scope_type="aws_account", external_id="abc"),
            bq_billing=_BQ_BILLING_OK,
        )


@pytest.mark.asyncio
async def test_list_cost_facts_bq_drops_zero_cost_rows() -> None:
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    rows = [
        SimpleNamespace(
            day=date(2026, 5, 14),
            account_id="123456789012",
            product_code="AmazonEC2",
            service_name="EC2",
            resource_id="i-1",
            currency="USD",
            region="us-west-2", usage_type="BoxUsage", usage_unit="Hrs",
            tag_owner=None, tag_application=None, tag_environment=None,
            tag_project=None, tag_department=None,
            native_cost=0.0, usage_amount=0.0,
        ),
        SimpleNamespace(
            day=date(2026, 5, 14),
            account_id="123456789012",
            product_code="AmazonEC2",
            service_name="EC2",
            resource_id="i-2",
            currency="USD",
            region="us-west-2", usage_type="BoxUsage", usage_unit="Hrs",
            tag_owner=None, tag_application=None, tag_environment=None,
            tag_project=None, tag_department=None,
            native_cost=1.5, usage_amount=1.0,
        ),
    ]
    with patch.object(
        driver, "_bq_client",
        return_value=_stub_bq_client_with_rows(rows),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    assert len(facts) == 1
    assert facts[0].resource_id == "i-2"


# ── Schema heterogeneity across CUR Parquet tables ──────────────
#
# Regression coverage for the 2026-05-15 prod incident: PR #311
# generalised an INFORMATION_SCHEMA probe taken on the master-account
# CUR table ("``line_item_resource_id`` exists") to all ~60
# linked-account tables and the cron failed on the three linked
# accounts whose CUR was actually built without "Include resource
# IDs". The driver now introspects each table's schema and degrades
# gracefully rather than emitting SQL that BQ rejects.


@pytest.mark.asyncio
async def test_list_cost_facts_bq_resource_id_column_absent_uses_synthetic_id() -> None:
    """When ``line_item_resource_id`` is absent from the table schema
    (CUR built without 'Include resource IDs'), the SQL must NOT
    reference that column anywhere, and every emitted fact must use
    the synthetic ``aws-account:<acc>:<product>`` id with
    ``resource_type='aws-aggregated'`` — same fallback used for
    NULL-resource_id rows on tables that DO have the column."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="470438934515",
    )
    columns = tuple(
        c for c in _DEFAULT_CUR_COLUMNS if c != "line_item_resource_id"
    )
    captured_sqls: list[str] = []
    # CUR 1.0 flat + missing resource_id column: the refactored
    # driver projects ``CAST(NULL AS STRING) AS resource_id`` so every
    # UNION leg carries a uniform shape. The row stub therefore MUST
    # carry ``resource_id=None`` (the SELECT projects it, BQ would
    # produce NULL on every row).
    row = SimpleNamespace(
        day=date(2026, 5, 14),
        account_id="470438934515",
        product_code="AmazonEC2",
        service_name="Amazon Elastic Compute Cloud",
        resource_id=None,
        currency="USD",
        region="us-west-2",
        usage_type="BoxUsage:m5.large",
        usage_unit="Hrs",
        # CUR 1.0 flat schema — tags arrive as ``tag_<alias>`` row
        # attributes consumed by ``_cur1_extract_tags_from_row``.
        tag_owner="alice@example.com",
        tag_application=None, tag_environment=None,
        tag_project=None, tag_department=None,
        native_cost=42.0,
        usage_amount=24.0,
    )

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([row])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(columns)
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )

    sql = captured_sqls[0]
    # The absent column itself must never appear by name — that would
    # cause BigQuery's ``invalidQuery: Unrecognized name`` failure
    # PR #311 originally fixed. ``CAST(NULL AS STRING) AS resource_id``
    # is the safe substitution.
    assert "line_item_resource_id" not in sql, (
        f"SQL must omit the absent column. Got: {sql}"
    )
    assert "CAST(NULL AS STRING) AS resource_id" in sql

    assert len(facts) == 1
    f = facts[0]
    assert f.resource_id == "aws-account:470438934515:AmazonEC2"
    assert f.resource_type == "aws-aggregated"
    assert f.amount_native == pytest.approx(42.0)
    # Tag columns still present in this table — owner tag survives.
    assert f.tags == {"owner": "alice@example.com"}


@pytest.mark.asyncio
async def test_list_cost_facts_bq_tag_columns_absent_emit_empty_tags() -> None:
    """When some/all ``resource_tags_user_*`` columns are absent (a
    common case for accounts whose resources were never tagged with
    those keys), the SQL must drop the missing columns from SELECT
    rather than crash, and ``CostFact.tags`` for those rows is
    simply an empty dict."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="225664694250",
    )
    # Only ``resource_tags_user_owner`` present; the four others missing.
    columns = tuple(
        c for c in _DEFAULT_CUR_COLUMNS
        if c not in {
            "resource_tags_user_application",
            "resource_tags_user_environment",
            "resource_tags_user_project",
            "resource_tags_user_department",
        }
    )
    captured_sqls: list[str] = []
    # Refactored CUR 1.0 driver projects ``CAST(NULL AS STRING) AS
    # tag_<alias>`` for absent flat tag columns so every UNION leg
    # carries a uniform shape. The row stub therefore MUST carry
    # ``tag_application=None`` (etc.) — that's what BQ would land
    # for those columns on the Row object.
    row = SimpleNamespace(
        day=date(2026, 5, 14),
        account_id="225664694250",
        product_code="AmazonS3",
        service_name="Amazon Simple Storage Service",
        resource_id="arn:aws:s3:::my-bucket",
        currency="USD",
        region="us-east-1",
        usage_type="TimedStorage-ByteHrs",
        usage_unit="GB-Mo",
        tag_owner="bob@example.com",
        tag_application=None, tag_environment=None,
        tag_project=None, tag_department=None,
        native_cost=7.5,
        usage_amount=1024.0,
    )

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([row])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(columns)
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )

    sql = captured_sqls[0]
    assert "resource_tags_user_owner" in sql
    for absent in (
        "resource_tags_user_application",
        "resource_tags_user_environment",
        "resource_tags_user_project",
        "resource_tags_user_department",
    ):
        assert absent not in sql, f"SQL must omit absent column {absent}"

    assert len(facts) == 1
    assert facts[0].tags == {"owner": "bob@example.com"}


@pytest.mark.asyncio
async def test_list_cost_facts_bq_all_tag_columns_absent_projects_null_tags() -> None:
    """When NONE of the tracked tag columns are present in the CUR 1.0
    schema, the refactored driver projects every tag alias as
    ``CAST(NULL AS STRING) AS tag_<alias>`` so every per-pick SELECT
    has the same column shape (required for UNION ALL compatibility
    across per_month month-tables, and a noop simplification for
    per_account single-table). Bare column references must still NOT
    appear — that's what would crash with
    ``invalidQuery: Unrecognized name``.
    """
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="335771843383",
    )
    columns = tuple(
        c for c in _DEFAULT_CUR_COLUMNS
        if not c.startswith("resource_tags_user_")
    )
    captured_sqls: list[str] = []
    row = SimpleNamespace(
        day=date(2026, 5, 14),
        account_id="335771843383",
        product_code="AmazonEC2",
        service_name="Amazon Elastic Compute Cloud",
        resource_id="i-no-tags",
        currency="USD",
        region="us-west-2",
        usage_type="BoxUsage",
        usage_unit="Hrs",
        # Every tag alias projected as ``CAST(NULL AS STRING) AS
        # tag_<alias>`` — BQ Row carries all of them as None.
        tag_owner=None, tag_application=None, tag_environment=None,
        tag_project=None, tag_department=None,
        native_cost=1.0,
        usage_amount=1.0,
    )

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([row])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(columns)
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )

    sql = captured_sqls[0]
    # No bare ``resource_tags_user_*`` column reference (BQ would
    # raise ``invalidQuery: Unrecognized name`` on those absent
    # columns).
    assert "resource_tags_user_" not in sql
    # Every tag alias projected as CAST(NULL AS STRING) so UNION-leg
    # column shape stays uniform across picks.
    for alias in ("owner", "application", "environment", "project", "department"):
        assert f"CAST(NULL AS STRING) AS tag_{alias}" in sql
    # No doubled comma artifact — the comma-aggregation joint is well-formed.
    assert ",," not in sql
    assert len(facts) == 1
    assert facts[0].tags == {}


@pytest.mark.asyncio
async def test_list_cost_facts_bq_synthetic_id_rows_produce_distinct_fact_keys() -> None:
    """End-to-end natural-key guarantee for the synthetic-id branch.

    When ``line_item_resource_id`` is absent, multiple BQ rows for the
    same (day × account × product) but different ``usage_type`` must
    produce DISTINCT ``compute_fact_key`` hashes — otherwise the upsert
    on ``cmdb_cost_facts.fact_key`` (UNIQUE) would collapse them and
    silently undercount cost. Round-3 logic-trace finding; locks in
    the contract end-to-end so a future driver refactor can't drop
    ``usage_type`` from the synthetic-id branch's SELECT/GROUP BY
    without immediately failing this test.
    """
    from app.domains.cmdb.cost_service import compute_fact_key

    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="470438934515",
    )
    columns = tuple(
        c for c in _DEFAULT_CUR_COLUMNS if c != "line_item_resource_id"
    )
    rows = [
        SimpleNamespace(
            day=date(2026, 5, 14),
            account_id="470438934515",
            product_code="AmazonEC2",
            service_name="EC2",
            currency="USD",
            region="us-west-2",
            usage_type="BoxUsage:m5.large",
            usage_unit="Hrs",
            resource_id=None,
            tag_owner=None, tag_application=None,
            tag_environment=None, tag_project=None, tag_department=None,
            native_cost=10.0, usage_amount=24.0,
        ),
        SimpleNamespace(
            day=date(2026, 5, 14),
            account_id="470438934515",
            product_code="AmazonEC2",
            service_name="EC2",
            currency="USD",
            region="us-west-2",
            usage_type="EBSOptimized:m5",
            usage_unit="Hrs",
            resource_id=None,
            tag_owner=None, tag_application=None,
            tag_environment=None, tag_project=None, tag_department=None,
            native_cost=0.5, usage_amount=24.0,
        ),
        SimpleNamespace(
            day=date(2026, 5, 14),
            account_id="470438934515",
            product_code="AmazonS3",  # different product, same day
            service_name="S3",
            currency="USD",
            region="us-east-1",
            usage_type="TimedStorage-ByteHrs",
            usage_unit="GB-Mo",
            resource_id=None,
            tag_owner=None, tag_application=None,
            tag_environment=None, tag_project=None, tag_department=None,
            native_cost=3.0, usage_amount=100.0,
        ),
    ]
    with patch.object(
        driver, "_bq_client",
        return_value=_stub_bq_client_with_rows(rows, columns=columns),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )

    # 3 BQ rows → 3 CostFacts (synthetic id collapses NONE of them
    # because the BQ GROUP BY already includes usage_type AND
    # product_code, and the driver synthesises a per-product id).
    assert len(facts) == 3
    keys = {
        compute_fact_key(
            integration_id=f.integration_id,
            fact_date=f.date,
            scope_external_id=str(f.scope.external_id),
            resource_id=f.resource_id,
            usage_type=f.usage_type,
        )
        for f in facts
    }
    assert len(keys) == 3, (
        f"Synthetic-id rows collapsed onto duplicate fact_keys — "
        f"upsert would silently undercount cost. fact_keys={keys}"
    )
    # Sanity-check the resource ids themselves: two EC2 rows share
    # the same synthetic id (different only on usage_type, which IS
    # in the fact_key but not in the id); S3 row has a different
    # synthetic id (different product_code).
    ids = {f.resource_id for f in facts}
    assert ids == {
        "aws-account:470438934515:AmazonEC2",
        "aws-account:470438934515:AmazonS3",
    }


@pytest.mark.asyncio
async def test_list_cost_facts_bq_resource_id_and_all_tags_absent_combined() -> None:
    """Worst-case combined: the actual prod-incident shape for the
    three failing accounts on 2026-05-15 (CUR built without "Include
    resource IDs" AND no resources ever tagged with the tracked keys).
    Both axes degrade simultaneously — the SQL must still parse, the
    fact must use the synthetic id, and tags must be empty."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="335771843383",
    )
    columns = tuple(
        c for c in _DEFAULT_CUR_COLUMNS
        if c != "line_item_resource_id"
        and not c.startswith("resource_tags_user_")
    )
    captured_sqls: list[str] = []
    row = SimpleNamespace(
        day=date(2026, 5, 14),
        account_id="335771843383",
        product_code="AmazonEC2",
        service_name="Amazon Elastic Compute Cloud",
        resource_id=None,
        currency="USD",
        region="us-west-2",
        usage_type="BoxUsage:m5.large",
        usage_unit="Hrs",
        # Both axes degraded — both projected as CAST(NULL AS STRING).
        tag_owner=None, tag_application=None, tag_environment=None,
        tag_project=None, tag_department=None,
        native_cost=100.0,
        usage_amount=24.0,
    )

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([row])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(columns)
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )

    sql = captured_sqls[0]
    assert "line_item_resource_id" not in sql
    assert "resource_tags_user_" not in sql
    assert "CAST(NULL AS STRING) AS resource_id" in sql
    for alias in ("owner", "application", "environment", "project", "department"):
        assert f"CAST(NULL AS STRING) AS tag_{alias}" in sql
    # GROUP BY now always includes ``resource_id`` — when its value is
    # always NULL (CAST NULL substitution), BigQuery treats NULL as a
    # single bucket so the effective grain is identical to omitting
    # the column from GROUP BY. The synthetic-id fallback in Python
    # then keys the natural-key hash on (account × product × usage_type
    # × currency) so the upsert's UNIQUE(fact_key) stays unique.
    assert len(facts) == 1
    assert facts[0].resource_id == "aws-account:335771843383:AmazonEC2"
    assert facts[0].resource_type == "aws-aggregated"
    assert facts[0].tags == {}


# ── _bq_resolve_options charset hardening (R1.3 security review) ──


def test_bq_resolve_options_rejects_dataset_with_backtick() -> None:
    """An admin (or compromised config) entering a dataset string
    containing a backtick could break out of the backtick-quoted
    identifier in the INFORMATION_SCHEMA probe / CUR data SELECTs.
    Reject at the validator boundary."""
    with pytest.raises(RuntimeError, match="characters outside"):
        AwsCloudDriver._bq_resolve_options({
            "credentials_json": _FAKE_SA,
            "dataset": "evil`.x",
            "table_pattern": "{account_id}_t",
        })


def test_bq_resolve_options_rejects_dataset_with_space() -> None:
    with pytest.raises(RuntimeError, match="characters outside"):
        AwsCloudDriver._bq_resolve_options({
            "credentials_json": _FAKE_SA,
            "dataset": "evil .x",
        })


def test_bq_resolve_options_rejects_dataset_with_quote() -> None:
    with pytest.raises(RuntimeError, match="characters outside"):
        AwsCloudDriver._bq_resolve_options({
            "credentials_json": _FAKE_SA,
            "dataset": "evil'.x",
        })


def test_bq_resolve_options_accepts_hyphenated_project() -> None:
    """PingCAP prod uses ``gcp-digital-bi.aws_nonprod_billing`` —
    hyphens in the project ID are legitimate BigQuery identifiers."""
    _, dataset, _, _, _, _ = AwsCloudDriver._bq_resolve_options({
        "credentials_json": _FAKE_SA,
        "dataset": "gcp-digital-bi.aws_nonprod_billing",
    })
    assert dataset == "gcp-digital-bi.aws_nonprod_billing"


def test_bq_resolve_options_rejects_pattern_with_backtick() -> None:
    with pytest.raises(RuntimeError, match="characters outside"):
        AwsCloudDriver._bq_resolve_options({
            "credentials_json": _FAKE_SA,
            "dataset": "p.d",
            "table_pattern": "{account_id}`_evil",
        })


def test_bq_resolve_options_rejects_pattern_with_semicolon() -> None:
    with pytest.raises(RuntimeError, match="characters outside"):
        AwsCloudDriver._bq_resolve_options({
            "credentials_json": _FAKE_SA,
            "dataset": "p.d",
            "table_pattern": "{account_id};DROP",
        })


# ── _bq_resolve_table_columns invariant checks (R1.2 correctness) ──


def test_bq_resolve_table_columns_rejects_non_fully_qualified_name() -> None:
    """The signature reads general-purpose; a caller passing a bare
    short name (``some_table`` instead of ``project.dataset.some_table``)
    would silently fail the INFORMATION_SCHEMA probe (no match → empty
    columns → table dropped downstream). Raise loudly at the boundary."""
    driver = _bq_driver()
    client = MagicMock()
    with pytest.raises(RuntimeError, match="not fully qualified"):
        driver._bq_resolve_table_columns(
            client,
            dataset="p.d",
            table_names=["bare_name"],
            columns_of_interest={"resource_tags_user_owner"},
        )


def test_bq_resolve_table_columns_rejects_short_name_collision() -> None:
    """Two FQ tables sharing the same short name in the same call
    cannot be disambiguated by ``INFORMATION_SCHEMA.COLUMNS.table_name``.
    Defensive: raise so a future caller mixing datasets doesn't
    silently land one table in the 'columns absent' bucket."""
    driver = _bq_driver()
    client = MagicMock()
    with pytest.raises(RuntimeError, match="short-name collision"):
        driver._bq_resolve_table_columns(
            client,
            dataset="p.d",
            table_names=[
                "p1.d1.shared_name",
                "p2.d2.shared_name",
            ],
            columns_of_interest={"resource_tags_user_owner"},
        )


def test_bq_resolve_table_columns_dedupe_is_safe() -> None:
    """Passing the same FQ name twice is benign — same entry, not a
    collision — so the short-name guard tolerates it."""
    driver = _bq_driver()
    client = MagicMock()
    job = MagicMock()
    job.result.return_value = iter([])
    client.query.return_value = job
    result = driver._bq_resolve_table_columns(
        client,
        dataset="p.d",
        table_names=["p.d.t", "p.d.t"],
        columns_of_interest={"resource_tags_user_owner"},
    )
    assert "p.d.t" in result


def test_bq_resolve_table_columns_empty_inputs_short_circuit() -> None:
    driver = _bq_driver()
    client = MagicMock()
    # No query issued either path.
    assert driver._bq_resolve_table_columns(
        client, dataset="p.d", table_names=[],
        columns_of_interest={"x"},
    ) == {}
    assert driver._bq_resolve_table_columns(
        client, dataset="p.d", table_names=["p.d.t"],
        columns_of_interest=set(),
    ) == {"p.d.t": set()}
    client.query.assert_not_called()


# ── _list_billing_daily_bq_sync ─────────────────────────────────


@pytest.mark.asyncio
async def test_list_billing_daily_bq_aggregates_service_breakdown() -> None:
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    totals = [SimpleNamespace(
        day=date(2026, 5, 14),
        account_id="123456789012",
        currency="USD",
        native_cost=200.0,
    )]
    # Service breakdown rows now also carry ``currency`` so the output
    # loop can attach the right breakdown to each currency-keyed
    # output row. A future mixed-currency day produces atomic
    # per-(day, account, currency, service) rows.
    services = [
        SimpleNamespace(
            day=date(2026, 5, 14),
            account_id="123456789012",
            currency="USD",
            service="Amazon EC2",
            native_cost=150.0,
        ),
        SimpleNamespace(
            day=date(2026, 5, 14),
            account_id="123456789012",
            currency="USD",
            service="Amazon S3",
            native_cost=50.0,
        ),
    ]

    client = MagicMock()
    client.get_table.return_value = _stub_table()
    # Three .query() calls: totals, services, owner_breakdown
    totals_job, svc_job, owner_job = MagicMock(), MagicMock(), MagicMock()
    totals_job.result.return_value = iter(totals)
    svc_job.result.return_value = iter(services)
    owner_job.result.return_value = iter([])  # nothing tagged
    client.query.side_effect = [totals_job, svc_job, owner_job]

    with patch.object(driver, "_bq_client", return_value=client):
        rows = await driver.list_billing_daily(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, include_owner_breakdown=True,
            bq_billing=_BQ_BILLING_OK,
        )
    assert len(rows) == 1
    r = rows[0]
    assert r.total_cost_usd == pytest.approx(200.0)
    assert r.cost_by_service == {
        "Amazon EC2": 150.0, "Amazon S3": 50.0,
    }
    assert r.owner_breakdown == {}  # include_owner_breakdown=True, none tagged


@pytest.mark.asyncio
async def test_list_billing_daily_bq_skips_query_when_no_owner_breakdown() -> None:
    """include_owner_breakdown=False must not issue the owner query —
    saves one BQ job per scope on the daily cron's hot path."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    client = MagicMock()
    client.get_table.return_value = _stub_table()
    totals_job, svc_job = MagicMock(), MagicMock()
    totals_job.result.return_value = iter([])
    svc_job.result.return_value = iter([])
    client.query.side_effect = [totals_job, svc_job]

    with patch.object(driver, "_bq_client", return_value=client):
        rows = await driver.list_billing_daily(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, include_owner_breakdown=False,
            bq_billing=_BQ_BILLING_OK,
        )
    assert rows == []
    assert client.query.call_count == 2  # totals + services only


@pytest.mark.asyncio
async def test_list_billing_daily_bq_owner_breakdown_skipped_when_column_absent() -> None:
    """When the scoped account's CUR has no ``resource_tags_user_owner``
    column, the owner_breakdown query must NOT be issued (issuing it
    would land on the same ``Unrecognized name`` failure as the
    cost_facts_sync 2026-05-15 incident). The output row carries an
    empty owner_breakdown dict — matches the "no rows tagged" semantics
    so the dashboard stays consistent."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="225664694250",
    )
    columns_no_owner = tuple(
        c for c in _DEFAULT_CUR_COLUMNS
        if c != "resource_tags_user_owner"
    )
    totals = [SimpleNamespace(
        day=date(2026, 5, 14),
        account_id="225664694250",
        currency="USD",
        native_cost=300.0,
    )]
    client = MagicMock()
    client.get_table.return_value = _stub_table(columns_no_owner)
    totals_job, svc_job = MagicMock(), MagicMock()
    totals_job.result.return_value = iter(totals)
    svc_job.result.return_value = iter([])
    client.query.side_effect = [totals_job, svc_job]

    with patch.object(driver, "_bq_client", return_value=client):
        rows = await driver.list_billing_daily(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, include_owner_breakdown=True,
            bq_billing=_BQ_BILLING_OK,
        )
    assert client.query.call_count == 2  # totals + services only, NO owner
    assert len(rows) == 1
    assert rows[0].owner_breakdown == {}


# ── _list_billing_by_dimension_bq_sync ──────────────────────────


@pytest.mark.asyncio
async def test_list_billing_by_dimension_bq_unknown_dimension_raises() -> None:
    driver = _bq_driver()
    with pytest.raises(ValueError, match="does not support dimension"):
        await driver.list_billing_by_dimension(
            date(2026, 5, 14), date(2026, 5, 14),
            group_by=["totally_made_up_dim"],
            bq_billing=_BQ_BILLING_OK,
        )


@pytest.mark.asyncio
async def test_list_billing_by_dimension_bq_caps_at_four_dimensions() -> None:
    driver = _bq_driver()
    with pytest.raises(ValueError, match="cap of 4"):
        await driver.list_billing_by_dimension(
            date(2026, 5, 14), date(2026, 5, 14),
            group_by=["service", "region", "account", "usage_type", "operation"],
            bq_billing=_BQ_BILLING_OK,
        )


@pytest.mark.asyncio
async def test_list_billing_by_dimension_bq_rejects_non_12_digit_scope() -> None:
    """Parity with the cost_facts / daily helpers: a malformed
    ``scope.external_id`` must fail loudly at the boundary, not silently
    compose a bogus table name and surface a confusing NotFound
    downstream. Round-3 review consistency fix."""
    driver = _bq_driver()
    with pytest.raises(RuntimeError, match="12-digit"):
        await driver.list_billing_by_dimension(
            date(2026, 5, 14), date(2026, 5, 14),
            group_by=["service"],
            scope=CloudScopeRef(
                scope_type="aws_account", external_id="not-an-account",
            ),
            bq_billing=_BQ_BILLING_OK,
        )


@pytest.mark.asyncio
async def test_list_billing_by_dimension_bq_skips_missing_table_for_scope() -> None:
    """Per-scope path now mirrors cost_facts/daily: ``client.get_table``
    existence-probe + log + return ``[]`` rather than silently issuing
    a SELECT against a non-existent table. R2 review parity fix."""
    from google.api_core.exceptions import NotFound

    driver = _bq_driver()
    client = MagicMock()
    client.get_table.side_effect = NotFound("missing")
    # ``query`` should never be reached.
    client.query.side_effect = AssertionError(
        "no data query should be issued when get_table returns NotFound"
    )

    with patch.object(driver, "_bq_client", return_value=client):
        rows = await driver.list_billing_by_dimension(
            date(2026, 5, 14), date(2026, 5, 14),
            group_by=["service"],
            scope=CloudScopeRef(
                scope_type="aws_account", external_id="123456789012",
            ),
            bq_billing=_BQ_BILLING_OK,
        )
    assert rows == []
    client.close.assert_called_once()


@pytest.mark.asyncio
async def test_list_billing_by_dimension_bq_tag_key_resolves_to_column() -> None:
    """tag:Owner must translate to resource_tags_user_owner in the
    SELECT/GROUP BY when the column exists in the per-account CUR.

    The driver issues two queries:
      1. INFORMATION_SCHEMA.COLUMNS lookup to learn which tag columns
         are present on each table (heterogeneous CUR schemas);
      2. The actual per-day-by-dimension aggregate SELECT.

    Stub #1 to report the column as present, then assert #2 (the data
    SELECT) projects ``resource_tags_user_owner`` rather than the
    ``CAST(NULL AS STRING)`` fallback used when the column is absent.
    """
    driver = _bq_driver()
    captured_sqls: list[str] = []

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        if "INFORMATION_SCHEMA.COLUMNS" in sql:
            # Report owner column as present on the scoped account's
            # table so the data SELECT can reference it.
            job.result.return_value = iter([
                SimpleNamespace(
                    table_name="123456789012_aws_billing_data",
                    column_name="resource_tags_user_owner",
                ),
            ])
        else:
            job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table()
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_billing_by_dimension(
            date(2026, 5, 14), date(2026, 5, 14),
            group_by=["service", "tag:Owner"],
            scope=CloudScopeRef(
                scope_type="aws_account", external_id="123456789012",
            ),
            bq_billing=_BQ_BILLING_OK,
        )
    # Two queries: the schema probe + the data SELECT.
    assert len(captured_sqls) == 2
    data_sql = next(
        s for s in captured_sqls if "INFORMATION_SCHEMA" not in s
    )
    assert "resource_tags_user_owner AS d1" in data_sql
    assert "product_product_name" in data_sql


@pytest.mark.asyncio
async def test_list_billing_by_dimension_bq_tag_dim_absent_drops_table() -> None:
    """When a requested tag DIMENSION column is absent from the
    scoped account's CUR (the account never tagged any resource with
    that key), the table is dropped from the UNION rather than
    contributing rows with a ``CAST(NULL AS STRING)`` placeholder.
    Dropping is the honest call: a NULL placeholder would silently
    merge "owner tag not observable" rows with "owner tag legitimately
    NULL" rows on the same dashboard cell. Round-1 review finding,
    same incident family as cost_facts_sync 2026-05-15."""
    driver = _bq_driver()
    captured_sqls: list[str] = []

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        # Schema probe returns no rows → column is absent.
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table()
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        rows = await driver.list_billing_by_dimension(
            date(2026, 5, 14), date(2026, 5, 14),
            group_by=["service", "tag:RareKey"],
            scope=CloudScopeRef(
                scope_type="aws_account", external_id="123456789012",
            ),
            bq_billing=_BQ_BILLING_OK,
        )
    assert rows == []
    # No data SELECT issued — schema-probe-only round trip.
    data_sqls = [s for s in captured_sqls if "INFORMATION_SCHEMA" not in s]
    assert data_sqls == []


@pytest.mark.asyncio
async def test_list_billing_by_dimension_bq_tag_dim_present_includes_column() -> None:
    """Inverse of the drop-on-absence test: when the requested tag-dim
    column IS present, the data SELECT references it directly (no NULL
    placeholder)."""
    driver = _bq_driver()
    captured_sqls: list[str] = []

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        if "INFORMATION_SCHEMA.COLUMNS" in sql:
            job.result.return_value = iter([
                SimpleNamespace(
                    table_name="123456789012_aws_billing_data",
                    column_name="resource_tags_user_owner",
                ),
            ])
        else:
            job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table()
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_billing_by_dimension(
            date(2026, 5, 14), date(2026, 5, 14),
            group_by=["service", "tag:Owner"],
            scope=CloudScopeRef(
                scope_type="aws_account", external_id="123456789012",
            ),
            bq_billing=_BQ_BILLING_OK,
        )
    data_sql = next(
        s for s in captured_sqls if "INFORMATION_SCHEMA" not in s
    )
    assert "resource_tags_user_owner AS d1" in data_sql
    assert "CAST(NULL AS STRING)" not in data_sql


@pytest.mark.asyncio
async def test_list_billing_by_dimension_bq_tag_filter_drops_tableless_account() -> None:
    """When a tag *filter* references a column that no table in the
    scope has, the driver returns [] rather than emitting a UNION ALL
    where every leg is unsatisfiable. Tenant-wide flow: discover N
    tables, schema-probe says none carry ``resource_tags_user_owner``,
    nothing to scan."""
    driver = _bq_driver()
    captured_sqls: list[str] = []

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        if "INFORMATION_SCHEMA.TABLES" in sql:
            # Discovery: one table in the dataset.
            job.result.return_value = iter([
                SimpleNamespace(table_name="111111111111_aws_billing_data"),
            ])
        else:
            # Schema probe: column not present on the table.
            job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table()
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        rows = await driver.list_billing_by_dimension(
            date(2026, 5, 14), date(2026, 5, 14),
            group_by=["service"],
            scope=None,
            filters={"tag:Owner": ["alice@example.com"]},
            bq_billing=_BQ_BILLING_OK,
        )
    assert rows == []
    # No data SELECT issued — discovery + schema-probe only.
    data_sqls = [
        s for s in captured_sqls
        if "INFORMATION_SCHEMA" not in s
    ]
    assert data_sqls == []


@pytest.mark.asyncio
async def test_list_billing_by_dimension_bq_filter_in_unnest() -> None:
    """A ``filters={'service': ['EC2']}`` must lower into an
    ``IN UNNEST(@flt_0)`` clause with the param bound to the
    allowed-value list.
    """
    driver = _bq_driver()
    captured_params: list[list] = []

    def _capture(sql, *, job_config=None, **kw):
        captured_params.append(job_config.query_parameters)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table()
    client.query.side_effect = _capture

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_billing_by_dimension(
            date(2026, 5, 14), date(2026, 5, 14),
            group_by=["service"],
            scope=CloudScopeRef(
                scope_type="aws_account", external_id="123456789012",
            ),
            filters={"service": ["Amazon EC2", "AmazonRDS"]},
            bq_billing=_BQ_BILLING_OK,
        )
    # First (and only) job's params include the ARRAY filter.
    flt_params = [p for p in captured_params[0] if p.name == "flt_0"]
    assert len(flt_params) == 1
    assert flt_params[0].values == ["Amazon EC2", "AmazonRDS"]


# ── list_billing_by_dimension routing: CE vs BQ ─────────────────


@pytest.mark.asyncio
async def test_list_billing_by_dimension_routes_to_ce_when_default() -> None:
    """Default integration (no billing_source set) must stay on the
    CE code path even when bq_billing is passed. SG depends on this
    invariant."""
    driver = AwsCloudDriver(integration_id="sg-test", config=_CE_CFG)
    with patch.object(
        driver, "_list_billing_by_dimension_bq_sync",
    ) as bq_sync, patch.object(
        driver, "_list_billing_by_dimension_sync", return_value=[],
    ) as ce_sync:
        await driver.list_billing_by_dimension(
            date(2026, 5, 14), date(2026, 5, 14),
            group_by=["service"],
            bq_billing=_BQ_BILLING_OK,
        )
    bq_sync.assert_not_called()
    ce_sync.assert_called_once()


# ── Resource-cleanup invariants (FD-leak guard) ──────────────────


@pytest.mark.asyncio
async def test_list_cost_facts_bq_closes_client_on_success() -> None:
    """The BigQuery Client holds pooled HTTP connections; the cron
    creates a fresh client per call so leaking accumulates ~180 unclosed
    clients per cost_facts_sync tick (3 BQ ops × 60 scopes). Verify the
    driver closes deterministically on the success path."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    client = _stub_bq_client_with_rows([])
    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    client.close.assert_called_once()


@pytest.mark.asyncio
async def test_list_cost_facts_bq_closes_client_on_error() -> None:
    """Same invariant on the error path — a BQ-side exception must
    still trigger client.close() via the try/finally."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    client = MagicMock()
    client.get_table.return_value = _stub_table()
    client.query.side_effect = RuntimeError("simulated BQ failure")
    with patch.object(driver, "_bq_client", return_value=client):
        with pytest.raises(RuntimeError, match="simulated BQ failure"):
            await driver.list_cost_facts(
                date(2026, 5, 14), date(2026, 5, 14),
                scope=scope, bq_billing=_BQ_BILLING_OK,
            )
    client.close.assert_called_once()


@pytest.mark.asyncio
async def test_list_billing_daily_bq_closes_client_on_table_not_found() -> None:
    """Early-return paths (e.g. table doesn't exist) must still close.
    Without try/finally this leaks one client per skipped account on
    every cron run."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    client = _stub_bq_client_with_rows([], table_exists=False)
    with patch.object(driver, "_bq_client", return_value=client):
        rows = await driver.list_billing_daily(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    assert rows == []
    client.close.assert_called_once()


@pytest.mark.asyncio
async def test_list_billing_by_dimension_bq_closes_client_on_success() -> None:
    driver = _bq_driver()
    client = MagicMock()
    client.get_table.return_value = _stub_table()
    job = MagicMock()
    job.result.return_value = iter([])
    client.query.return_value = job
    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_billing_by_dimension(
            date(2026, 5, 14), date(2026, 5, 14),
            group_by=["service"],
            scope=CloudScopeRef(
                scope_type="aws_account", external_id="123456789012",
            ),
            bq_billing=_BQ_BILLING_OK,
        )
    client.close.assert_called_once()


# ── Bytes-billed + job timeout caps ──────────────────────────────


@pytest.mark.asyncio
async def test_bq_query_job_config_applies_safety_caps() -> None:
    """Every BQ query the driver issues must run with
    ``maximum_bytes_billed`` + ``job_timeout_ms`` set. A malformed
    predicate (e.g. losing the year/month partition filter through
    a refactor) could otherwise scan the whole CUR dataset (~250 GB
    in prod) on a single query."""
    driver = _bq_driver()
    captured_configs: list = []

    def _capture_query(sql, *, job_config=None, **kw):
        captured_configs.append(job_config)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table()
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=CloudScopeRef(
                scope_type="aws_account", external_id="123456789012",
            ),
            bq_billing=_BQ_BILLING_OK,
        )

    assert len(captured_configs) >= 1
    for cfg in captured_configs:
        # ``maximum_bytes_billed`` is stored as int. ``job_timeout_ms``
        # is stored as str on the BQ API surface; coerce both ways
        # before comparing so a future BQ-library change in either
        # direction doesn't break this assertion.
        assert int(cfg.maximum_bytes_billed) == AwsCloudDriver._BQ_MAX_BYTES_BILLED
        assert int(cfg.job_timeout_ms) == AwsCloudDriver._BQ_JOB_TIMEOUT_MS


# ── usage_type granularity preservation (compute_fact_key parity) ──


@pytest.mark.asyncio
async def test_list_cost_facts_bq_groups_by_usage_type() -> None:
    """``compute_fact_key`` includes ``usage_type`` — two CUR rows for
    the same resource on the same day but different usage_types are
    distinct facts. The SQL MUST group by usage_type so the BQ
    aggregation produces one row per fact_key rather than ANY_VALUE-
    folding them into one and silently undercounting."""
    driver = _bq_driver()
    captured_sqls: list[str] = []

    def _capture_query(sql, *, job_config=None, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table()
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=CloudScopeRef(
                scope_type="aws_account", external_id="123456789012",
            ),
            bq_billing=_BQ_BILLING_OK,
        )

    sql = captured_sqls[0]
    # Both surfaces matter: SELECT projects it raw (NOT ANY_VALUE),
    # GROUP BY includes it so the aggregator splits per usage_type.
    assert "line_item_usage_type AS usage_type" in sql
    assert "usage_type" in sql.split("GROUP BY")[1]


@pytest.mark.asyncio
async def test_list_cost_facts_bq_emits_distinct_facts_per_usage_type() -> None:
    """End-to-end check: two BQ rows with same resource_id but different
    usage_type must yield two distinct CostFact rows (with usage_type
    set), not one collapsed row."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    rows = [
        SimpleNamespace(
            day=date(2026, 5, 14),
            account_id="123456789012",
            product_code="AmazonEC2",
            service_name="Amazon Elastic Compute Cloud",
            resource_id="i-deadbeef",
            currency="USD",
            region="us-west-2",
            usage_type="BoxUsage:m5.large",
            usage_unit="Hrs",
            tag_owner=None, tag_application=None, tag_environment=None,
            tag_project=None, tag_department=None,
            native_cost=12.34, usage_amount=24.0,
        ),
        SimpleNamespace(
            day=date(2026, 5, 14),
            account_id="123456789012",
            product_code="AmazonEC2",
            service_name="Amazon Elastic Compute Cloud",
            resource_id="i-deadbeef",
            currency="USD",
            region="us-west-2",
            usage_type="EBSOptimized:m5",
            usage_unit="Hrs",
            tag_owner=None, tag_application=None, tag_environment=None,
            tag_project=None, tag_department=None,
            native_cost=0.50, usage_amount=24.0,
        ),
    ]
    with patch.object(
        driver, "_bq_client",
        return_value=_stub_bq_client_with_rows(rows),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    assert len(facts) == 2
    usage_types = {f.usage_type for f in facts}
    assert usage_types == {"BoxUsage:m5.large", "EBSOptimized:m5"}


# ── Other drivers accept bq_billing kwarg silently ───────────────


@pytest.mark.asyncio
async def test_gcp_driver_accepts_bq_billing_kwarg() -> None:
    """Cross-driver: the service layer passes ``bq_billing`` uniformly
    to every cloud driver. Non-AWS drivers must accept and ignore it,
    not TypeError. Smoke check against GCP — the other 5 drivers
    follow the same pattern."""
    import inspect

    from app.integrations.drivers.gcp_cloud import GcpCloudDriver
    for method_name in (
        "list_billing_daily", "list_cost_facts",
    ):
        sig = inspect.signature(getattr(GcpCloudDriver, method_name))
        assert "bq_billing" in sig.parameters, (
            f"GcpCloudDriver.{method_name} missing bq_billing kwarg — "
            f"will TypeError when cost_service passes it"
        )


@pytest.mark.asyncio
async def test_list_billing_daily_bq_rejects_multi_currency_day() -> None:
    """`cmdb_cloud_billing_daily` UNIQUE is (integration, scope, date)
    — there's no currency column. Emitting two rows for the same day
    with different currencies would silently collide on UPSERT and
    drop one. Catch the future-AWS-China case at the driver layer
    with a clear error so the operator knows to extend the UNIQUE
    constraint first."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    totals = [
        SimpleNamespace(
            day=date(2026, 5, 14),
            account_id="123456789012",
            currency="USD",
            native_cost=200.0,
        ),
        SimpleNamespace(
            day=date(2026, 5, 14),
            account_id="123456789012",
            currency="CNY",
            native_cost=1400.0,
        ),
    ]
    client = MagicMock()
    client.get_table.return_value = _stub_table()
    totals_job, svc_job = MagicMock(), MagicMock()
    totals_job.result.return_value = iter(totals)
    svc_job.result.return_value = iter([])
    client.query.side_effect = [totals_job, svc_job]

    with patch.object(driver, "_bq_client", return_value=client):
        with pytest.raises(RuntimeError, match="multi-currency"):
            await driver.list_billing_daily(
                date(2026, 5, 14), date(2026, 5, 14),
                scope=scope, bq_billing=_BQ_BILLING_OK,
            )
    # Even on this error path, the client is still closed (try/finally).
    client.close.assert_called_once()


def test_bq_resolve_options_whitespace_pattern_falls_back_to_default() -> None:
    """A whitespace-only ``table_pattern`` (admin typo) must fall back
    to the default rather than landing as an empty string downstream
    where the {account_id} placeholder check would fail with a
    confusing error."""
    _, _, pattern, _, _, _ = AwsCloudDriver._bq_resolve_options({
        "credentials_json": _FAKE_SA,
        "dataset": "p.d",
        "table_pattern": "   ",
    })
    assert pattern == "{account_id}_aws_billing_data"


@pytest.mark.asyncio
async def test_all_cloud_drivers_accept_bq_billing_kwarg() -> None:
    """Defence-in-depth across every cloud driver. Adding a new driver
    that forgets the kwarg silently turns into a runtime TypeError on
    the first billing cron tick — catch it at test time."""
    import inspect

    from app.integrations.drivers.alicloud_intl_cloud import (
        AlicloudIntlCloudDriver,
    )
    from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver
    from app.integrations.drivers.azure_cloud import AzureCloudDriver
    from app.integrations.drivers.gcp_cloud import GcpCloudDriver
    from app.integrations.drivers.tencent_cloud import TencentCloudDriver
    from app.integrations.drivers.ucloud_cloud import UCloudDriver

    # (DriverClass, method_name) pairs — only methods the driver
    # actually overrides matter; if a driver inherits the base
    # implementation it picks up ``bq_billing`` automatically.
    driver_methods = [
        (AzureCloudDriver, "list_billing_daily"),
        (AzureCloudDriver, "list_cost_facts"),
        (AzureCloudDriver, "list_billing_by_dimension"),
        (GcpCloudDriver, "list_billing_daily"),
        (GcpCloudDriver, "list_cost_facts"),
        (AliyunCloudDriver, "list_billing_daily"),
        (AliyunCloudDriver, "list_cost_facts"),
        (AlicloudIntlCloudDriver, "list_billing_daily"),
        (TencentCloudDriver, "list_billing_daily"),
        (TencentCloudDriver, "list_cost_facts"),
        (UCloudDriver, "list_billing_daily"),
    ]
    missing = []
    for cls, method_name in driver_methods:
        sig = inspect.signature(getattr(cls, method_name))
        if "bq_billing" not in sig.parameters:
            missing.append(f"{cls.__name__}.{method_name}")
    assert not missing, (
        f"Drivers missing bq_billing kwarg "
        f"(will TypeError when service layer passes it): {missing}"
    )


# ── Savings Plan discount accounting (PR #339) ──────────────────
#
# CUR ships SP-covered usage as THREE rows: a positive ``Usage`` at
# the on-demand list price, a negative ``SavingsPlanNegation`` for
# the discount portion, and a ``SavingsPlanCoveredUsage`` tracking
# row (unblended_cost=0). Filtering ``SavingsPlanNegation`` /
# ``ComputeSavingsPlans`` out of CUR SUMs drops the discount and
# inflates reported cost by the SP rate (~30%). Prod incident:
# AWS account 552185537300 April 2026 showed $8,360.87 in Helix vs
# $5,505.88 on the invoice — 34% over.
#
# These tests pin the WHERE clause shape so a future
# "let's clean up the SQL" edit can't silently reintroduce the
# inflation.


def _capture_query_sqls() -> tuple[MagicMock, list[str]]:
    """Build a BQ client whose ``query()`` calls are captured to
    a list (returning empty result jobs). The caller asserts on the
    captured SQL strings."""
    sqls: list[str] = []

    def _capture(sql, **_kw):
        sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table()
    client.query.side_effect = _capture
    return client, sqls


@pytest.mark.asyncio
async def test_list_billing_daily_bq_does_not_filter_sp_negation() -> None:
    """Regression: the totals SQL must NOT exclude SP negation rows or
    the ComputeSavingsPlans product code. Those rows ARE the SP
    discount and the SP recurring fee; excluding them inflates
    reported cost ~30% (PR #339 / account 552185537300 April 2026)."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    client, sqls = _capture_query_sqls()
    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_billing_daily(
            date(2026, 5, 1), date(2026, 5, 15),
            scope=scope, include_owner_breakdown=False,
            bq_billing=_BQ_BILLING_OK,
        )
    assert sqls, "no BQ query issued"
    # Totals SQL is the first query. Same shape applies to the
    # service-breakdown SQL (second query) — both must keep SP rows.
    # Pattern check (NOT bare-string check) — the service-breakdown
    # SQL legitimately references the SP line-item-type strings inside
    # a ``CASE`` that buckets them under ``'Savings Plans'`` for the
    # donut. What we forbid is the WHERE-clause filter form
    # ``!= 'SavingsPlanNegation'`` / ``!= 'ComputeSavingsPlans'`` that
    # used to drop the rows entirely.
    for label, sql in zip(("totals", "services"), sqls):
        assert "!= 'SavingsPlanNegation'" not in sql, (
            f"{label} SQL still has the WHERE-exclude for "
            f"SavingsPlanNegation — would drop the SP discount and "
            f"inflate cost.\nSQL:\n{sql}"
        )
        assert "!= 'ComputeSavingsPlans'" not in sql, (
            f"{label} SQL still has the WHERE-exclude for "
            f"product_code='ComputeSavingsPlans' — would drop the SP "
            f"recurring fee on payer accounts.\nSQL:\n{sql}"
        )


@pytest.mark.asyncio
async def test_list_billing_daily_bq_service_breakdown_buckets_sp() -> None:
    """The service breakdown SQL must label SP-related rows under a
    single ``Savings Plans`` pseudo-service so the per-scope donut
    has one clearly labelled discount slice instead of a CUR-internal
    product name (``Savings Plans for AWS Compute usage``) sharing
    its legend entry with positive SP fee rows."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    client, sqls = _capture_query_sqls()
    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_billing_daily(
            date(2026, 5, 1), date(2026, 5, 15),
            scope=scope, include_owner_breakdown=False,
            bq_billing=_BQ_BILLING_OK,
        )
    services_sql = sqls[1]
    assert "'Savings Plans'" in services_sql
    assert "SavingsPlanNegation" in services_sql  # in the CASE bucket list
    assert "SavingsPlanCoveredUsage" in services_sql
    assert "SavingsPlanRecurringFee" in services_sql
    # CUR 2.0 enterprise discount lines must also bucket under named
    # pseudo-services so the donut legend reads "EDP Discount: -$X"
    # instead of an anonymous "Other" slice (PR #4xx, 2026-05-20).
    assert "'EDP Discount'" in services_sql
    assert "EdpDiscount" in services_sql
    assert "'Private Rate Discount'" in services_sql
    assert "PrivateRateDiscount" in services_sql


@pytest.mark.asyncio
async def test_list_billing_daily_bq_sums_sp_negation_into_total() -> None:
    """End-to-end: when the BQ result contains both a positive Usage
    row and a negative SavingsPlanNegation row for the same day, the
    daily total must net them. This is the prod-incident pattern from
    account 552185537300."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="552185537300",
    )
    # The totals SQL pre-aggregates by (day, account, currency) so
    # the driver sees ONE row per day with the netted native_cost.
    # Simulate the BQ aggregator returning the net.
    totals_rows = [SimpleNamespace(
        day=date(2026, 4, 30),
        account_id="552185537300",
        currency="USD",
        native_cost=5505.88,  # net of SP discount
    )]
    service_rows = [
        SimpleNamespace(
            day=date(2026, 4, 30),
            account_id="552185537300",
            currency="USD",
            service="Amazon Elastic Compute Cloud",
            native_cost=8360.87,  # gross EC2
        ),
        SimpleNamespace(
            day=date(2026, 4, 30),
            account_id="552185537300",
            currency="USD",
            service="Savings Plans",
            native_cost=-2854.99,  # SP discount as a negative line
        ),
    ]
    client = MagicMock()
    client.get_table.return_value = _stub_table()
    totals_job, svc_job = MagicMock(), MagicMock()
    totals_job.result.return_value = iter(totals_rows)
    svc_job.result.return_value = iter(service_rows)
    client.query.side_effect = [totals_job, svc_job]

    with patch.object(driver, "_bq_client", return_value=client):
        rows = await driver.list_billing_daily(
            date(2026, 4, 30), date(2026, 4, 30),
            scope=scope, include_owner_breakdown=False,
            bq_billing=_BQ_BILLING_OK,
        )
    assert len(rows) == 1
    assert rows[0].total_cost_usd == pytest.approx(5505.88)
    # Service breakdown carries the gross EC2 line and the negative
    # SP line — sum equals total.
    bd = rows[0].cost_by_service
    assert bd.get("Amazon Elastic Compute Cloud") == pytest.approx(8360.87)
    assert bd.get("Savings Plans") == pytest.approx(-2854.99)
    assert sum(bd.values()) == pytest.approx(5505.88)


@pytest.mark.asyncio
async def test_list_cost_facts_bq_does_not_filter_sp_negation() -> None:
    """Same regression guard for the cost_facts ingest path. The
    cost_facts table backs the per-scope billing-detail drawer, scope
    ranking, and chargeback rule ladder — an SP filter here cascades
    into every analytical surface."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    client, sqls = _capture_query_sqls()
    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 15),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    assert sqls
    sql = sqls[0]
    assert "!= 'SavingsPlanNegation'" not in sql, (
        f"cost_facts SQL still has the WHERE-exclude for "
        f"SavingsPlanNegation.\nSQL:\n{sql}"
    )
    assert "!= 'ComputeSavingsPlans'" not in sql, (
        f"cost_facts SQL still has the WHERE-exclude for "
        f"ComputeSavingsPlans.\nSQL:\n{sql}"
    )


@pytest.mark.asyncio
async def test_list_billing_by_dimension_bq_does_not_filter_sp_negation() -> None:
    """Same regression guard for the by-dimension reporting path
    (powers cross-account / per-service drilldowns)."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    client, sqls = _capture_query_sqls()
    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_billing_by_dimension(
            date(2026, 5, 1), date(2026, 5, 15),
            scope=scope, group_by=["service"],
            bq_billing=_BQ_BILLING_OK,
        )
    assert sqls
    sql = sqls[0]
    assert "!= 'SavingsPlanNegation'" not in sql, (
        f"by-dimension SQL still has the WHERE-exclude for "
        f"SavingsPlanNegation.\nSQL:\n{sql}"
    )
    assert "!= 'ComputeSavingsPlans'" not in sql, (
        f"by-dimension SQL still has the WHERE-exclude for "
        f"ComputeSavingsPlans.\nSQL:\n{sql}"
    )


@pytest.mark.asyncio
async def test_list_billing_daily_bq_payer_account_carries_recurring_fee() -> None:
    """Payer-account scope's CUR carries SavingsPlanRecurringFee rows
    (positive — the SP commitment fee for the whole org) alongside
    any own-usage rows. Linked-account scopes carry only negative
    SavingsPlanNegation rows for their share of the discount. Both
    classes get bucketed under 'Savings Plans' by the service-breakdown
    CASE; the totals SQL nets both correctly without any SP-aware
    branching needed in the driver. This test pins the payer half of
    that contract — the prod 552185537300 case (PR #339 round-1)
    covered the linked-account half."""
    driver = _bq_driver()
    payer_scope = CloudScopeRef(
        scope_type="aws_account", external_id="317766989874",  # PingCAP Org master
    )
    # Payer CUR aggregator returns one row carrying own-usage net of
    # own SP discount PLUS the SP commitment fee. The SP fee dwarfs the
    # tiny own-usage on a master / org-management account that runs
    # almost no compute itself.
    totals_rows = [SimpleNamespace(
        day=date(2026, 4, 30),
        account_id="317766989874",
        currency="USD",
        native_cost=5200.00,  # ~$5000 SP fee + $200 own-usage net
    )]
    service_rows = [
        SimpleNamespace(
            day=date(2026, 4, 30),
            account_id="317766989874",
            currency="USD",
            service="Amazon Elastic Compute Cloud",
            native_cost=200.00,  # minimal own usage on master
        ),
        SimpleNamespace(
            day=date(2026, 4, 30),
            account_id="317766989874",
            currency="USD",
            service="Savings Plans",
            native_cost=5000.00,  # SP commitment fee — POSITIVE on payer
        ),
    ]
    client = MagicMock()
    client.get_table.return_value = _stub_table()
    totals_job, svc_job = MagicMock(), MagicMock()
    totals_job.result.return_value = iter(totals_rows)
    svc_job.result.return_value = iter(service_rows)
    client.query.side_effect = [totals_job, svc_job]

    with patch.object(driver, "_bq_client", return_value=client):
        rows = await driver.list_billing_daily(
            date(2026, 4, 30), date(2026, 4, 30),
            scope=payer_scope, include_owner_breakdown=False,
            bq_billing=_BQ_BILLING_OK,
        )
    assert len(rows) == 1
    assert rows[0].total_cost_usd == pytest.approx(5200.00)
    bd = rows[0].cost_by_service
    # Both branches of the SP CASE are exercised: a payer account's
    # SP fee lands positive under "Savings Plans" — same legend entry
    # as a linked account's negative discount line, distinguishable
    # only by sign in the frontend.
    assert bd.get("Savings Plans") == pytest.approx(5000.00)
    assert bd.get("Amazon Elastic Compute Cloud") == pytest.approx(200.00)


@pytest.mark.asyncio
async def test_list_cost_facts_bq_buckets_sp_rows_to_savings_plans_service() -> None:
    """cost_facts service field is the source of truth for the
    per-scope detail drawer's donut AND for the chargeback allocation
    ladder. SP-flavor product codes (ComputeSavingsPlans /
    EC2InstanceSavingsPlans / SageMakerSavingsPlans) must all bucket
    under the same 'Savings Plans' label so cost_facts and the
    billing_daily.service_breakdown (which already applies the same
    CASE in _service_select) agree on the legend key. Without the
    rebucketing, the two surfaces show different strings for the same
    row class, and chargeback skip-allocation (next test) wouldn't
    catch all flavours."""
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    sp_rows = [
        SimpleNamespace(
            day=date(2026, 5, 1),
            account_id="123456789012",
            product_code="ComputeSavingsPlans",
            service_name="Savings Plans for AWS Compute usage",
            resource_id=None,
            currency="USD",
            region=None,
            usage_type="ComputeSP:1yr-NoUpfront",
            usage_unit=None,
            native_cost=-1234.56,
            usage_amount=1.0,
            # tag_* must be present per the SELECT contract — empty values OK
            tag_owner=None, tag_application=None, tag_environment=None,
            tag_project=None, tag_department=None,
        ),
        SimpleNamespace(
            day=date(2026, 5, 1),
            account_id="123456789012",
            product_code="EC2InstanceSavingsPlans",
            service_name="EC2 Instance Savings Plans",
            resource_id=None,
            currency="USD",
            region=None,
            usage_type="EC2SP-Usage",
            usage_unit=None,
            native_cost=-42.10,
            usage_amount=1.0,
            tag_owner=None, tag_application=None, tag_environment=None,
            tag_project=None, tag_department=None,
        ),
        # Regular EC2 usage row — should NOT bucket as 'Savings Plans'
        SimpleNamespace(
            day=date(2026, 5, 1),
            account_id="123456789012",
            product_code="AmazonEC2",
            service_name="Amazon Elastic Compute Cloud",
            resource_id="i-0abc",
            currency="USD",
            region="us-west-2",
            usage_type="BoxUsage:m5.large",
            usage_unit="Hrs",
            native_cost=1.92,
            usage_amount=24.0,
            tag_owner=None, tag_application=None, tag_environment=None,
            tag_project=None, tag_department=None,
        ),
    ]
    client = MagicMock()
    client.get_table.return_value = _stub_table()
    job = MagicMock()
    job.result.return_value = iter(sp_rows)
    client.query.return_value = job

    with patch.object(driver, "_bq_client", return_value=client):
        facts = await driver.list_cost_facts(
            date(2026, 5, 1), date(2026, 5, 1),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )
    # 3 input rows, 3 output facts (none zero-dropped).
    assert len(facts) == 3
    by_pc = {f.raw["product_code"]: f for f in facts}
    assert by_pc["ComputeSavingsPlans"].service == "Savings Plans"
    assert by_pc["EC2InstanceSavingsPlans"].service == "Savings Plans"
    # Regular EC2 keeps its product_product_name verbatim.
    assert (
        by_pc["AmazonEC2"].service
        == "Amazon Elastic Compute Cloud"
    )


def test_cost_facts_ingest_skips_allocation_for_savings_plans_rows() -> None:
    """The cost_facts ingest path's allocation step (and the
    recompute_allocation rebatcher) must short-circuit when
    ``fact.service == 'Savings Plans'``. SP rows are org-level
    commitment netting, not user-attributable spend — allocating
    them to a scope owner makes per-person chargeback views carry
    spurious negative (linked-account negation) or positive (payer
    recurring fee) lines that are not the owner's own activity. The
    sentinel pinned here is the ``auto_owner_source='savings_plan'``
    value that both code paths stamp; both pass-throughs of the
    branch live in ``server/app/domains/cmdb/cost_service.py``.

    This is a STATIC code-shape test (no DB needed) — it inspects
    the source for the SP-aware token set. The check is intentionally
    PATTERN-based (substring presence), not EXACT-LITERAL-based, so a
    future refactor that extracts the SP label into a constant or
    factors the branch into a helper does NOT trip this test for a
    non-regression. The required signals are:
      1. The literal ``"Savings Plans"`` string appears (the service
         label is the matched value in both branches).
      2. The literal ``"savings_plan"`` sentinel appears (stamped
         into ``auto_owner_source``).
      3. The token ``is_sp_row`` appears (the local boolean that
         gates both allocation skip + scope-owner-fallback skip).
    Behavioural tests for the allocation outcomes (e.g., a real
    cost_facts row with service='Savings Plans' lands with
    cost_owner_person_id=None after ``recompute_allocation``) live
    alongside the broader cost-allocation suite when an in-process
    DB is available; the static guard here is the canary that fires
    BEFORE the DB-backed tests can even run."""
    import inspect

    from app.domains.cmdb import cost_service

    source = inspect.getsource(cost_service)
    assert "Savings Plans" in source, (
        "SP service label missing — both ingest_cost_facts and "
        "recompute_allocation MUST gate on fact/row .service == "
        "'Savings Plans' before allocating to a scope owner. Without "
        "the gate, ~30%-magnitude negative SP discount rows attribute "
        "to scope owners and pollute per-person chargeback views."
    )
    assert "savings_plan" in source, (
        "auto_owner_source='savings_plan' sentinel missing — admin "
        "auditing tools that filter by this column value treat SP "
        "rows as unallocated rather than intentionally-skipped."
    )
    assert "is_sp_row" in source, (
        "is_sp_row local missing — both ingest and recompute branches "
        "use this boolean to (a) zero out the allocation tuple and "
        "(b) skip the scope-owner fallback. Refactoring either branch "
        "to inline the check is fine but the SP-skip path must remain."
    )


# ── CUR 2.0 / Data Exports support (per_month + nested) ─────────
#
# Verifies the driver's dual-layout / dual-tag-schema implementation
# against the shape of PingCAP's ``gcp-digital-bi.aws_billing_no_payer``
# dataset: one native-partitioned table per month
# (``aws_billing_YYYYMMDD``), all linked accounts merged inside, tags
# nested in ``resource_tags.key_value`` ARRAY<STRUCT<key,value>>.
# The same helper code path serves the legacy CUR 1.0
# ``aws_nonprod_billing`` (per_account + flat) without behaviour
# change — guaranteed by the existing 72 tests above.

_BQ_BILLING_CUR_V2 = {
    "credentials_json": _FAKE_SA,
    "dataset": "fake-bi.aws_billing_no_payer",
    "table_pattern": "aws_billing_{yyyymmdd}",
    "table_layout": "per_month",
    "tag_schema": "nested",
}

_BQ_CFG_CUR_V2 = {
    **_BQ_CFG_BASE,
    "billing_bq_dataset": "fake-bi.aws_billing_no_payer",
    "billing_bq_table_pattern": "aws_billing_{yyyymmdd}",
    "billing_bq_table_layout": "per_month",
    "billing_bq_tag_schema": "nested",
}


def _bq_driver_v2(**override: object) -> AwsCloudDriver:
    cfg = {**_BQ_CFG_CUR_V2, **override}
    return AwsCloudDriver(integration_id="aws-test-v2", config=cfg)


# ── Config validation: layout/tag_schema/pattern coherence ─────


def test_bq_resolve_options_rejects_invalid_layout() -> None:
    with pytest.raises(RuntimeError, match="table_layout"):
        AwsCloudDriver._bq_resolve_options({
            "credentials_json": _FAKE_SA,
            "dataset": "p.d",
            "table_layout": "per_decade",
        })


def test_bq_resolve_options_rejects_invalid_tag_schema() -> None:
    with pytest.raises(RuntimeError, match="tag_schema"):
        AwsCloudDriver._bq_resolve_options({
            "credentials_json": _FAKE_SA,
            "dataset": "p.d",
            "tag_schema": "graph",
        })


def test_bq_resolve_options_rejects_per_month_pattern_with_account_id() -> None:
    """per_month layout expects a ``{yyyymmdd}`` placeholder — passing
    the per_account placeholder is an admin config typo and we fail
    loudly at the boundary rather than producing an unparseable BQ
    table name on the first scope."""
    with pytest.raises(RuntimeError, match="{yyyymmdd}"):
        AwsCloudDriver._bq_resolve_options({
            "credentials_json": _FAKE_SA,
            "dataset": "p.d",
            "table_layout": "per_month",
            "table_pattern": "{account_id}_aws_billing_data",
        })


def test_bq_resolve_options_rejects_per_account_pattern_with_yyyymmdd() -> None:
    """Symmetric to the previous test — per_account expects
    ``{account_id}``."""
    with pytest.raises(RuntimeError, match="{account_id}"):
        AwsCloudDriver._bq_resolve_options({
            "credentials_json": _FAKE_SA,
            "dataset": "p.d",
            "table_layout": "per_account",
            "table_pattern": "aws_billing_{yyyymmdd}",
        })


def test_bq_resolve_options_per_month_defaults_pattern() -> None:
    """When ``table_layout=per_month`` is set without an explicit
    pattern, the helper falls back to the prod default
    ``aws_billing_{yyyymmdd}`` — keeps the admin UI bare-bones."""
    _, _, pat, layout, tag_schema, fact_grain = AwsCloudDriver._bq_resolve_options({
        "credentials_json": _FAKE_SA,
        "dataset": "p.d",
        "table_layout": "per_month",
        "tag_schema": "nested",
    })
    assert pat == "aws_billing_{yyyymmdd}"
    assert layout == "per_month"
    assert tag_schema == "nested"
    # Default fact_grain — per_product keeps cost_facts row volume
    # bounded; per_resource is the explicit opt-in.
    assert fact_grain == "per_product"


# ── _bq_table_for_month / _bq_months_in_window ─────────────────


def test_bq_table_for_month_substitutes_yyyymmdd() -> None:
    d = _bq_driver_v2()
    assert d._bq_table_for_month(
        date(2026, 5, 1),
        dataset="p.aws_billing_no_payer",
        pattern="aws_billing_{yyyymmdd}",
    ) == "p.aws_billing_no_payer.aws_billing_20260501"


def test_bq_table_for_month_rejects_non_first_day() -> None:
    """The CUR 2.0 month-table token is the month's first day; passing
    a mid-month date is a logic bug (the caller forgot to normalise to
    the month start) and we fail loudly rather than silently produce
    an unparseable table name."""
    d = _bq_driver_v2()
    with pytest.raises(RuntimeError, match="first day"):
        d._bq_table_for_month(
            date(2026, 5, 14),
            dataset="p.d",
            pattern="aws_billing_{yyyymmdd}",
        )


def test_bq_months_in_window_single_month() -> None:
    assert AwsCloudDriver._bq_months_in_window(
        date(2026, 5, 1), date(2026, 5, 31),
    ) == [date(2026, 5, 1)]


def test_bq_months_in_window_spans_year_boundary() -> None:
    """A window crossing Dec→Jan must enumerate both months' first-day
    markers. The _bq_table_for_window caller turns each into a month-
    table name; missing the year-rollover would silently drop January
    spend from the result."""
    months = AwsCloudDriver._bq_months_in_window(
        date(2025, 12, 28), date(2026, 1, 3),
    )
    assert months == [date(2025, 12, 1), date(2026, 1, 1)]


def test_bq_months_in_window_three_month_backfill() -> None:
    months = AwsCloudDriver._bq_months_in_window(
        date(2026, 3, 1), date(2026, 5, 31),
    )
    assert months == [date(2026, 3, 1), date(2026, 4, 1), date(2026, 5, 1)]


# ── _bq_partition_predicate (layout-aware) ─────────────────────


def test_bq_partition_predicate_per_account_uses_hive() -> None:
    """per_account (CUR 1.0 EXTERNAL) prunes on Hive ``year``+``month``
    INT columns — the legacy behaviour preserved for back-compat."""
    sql, params = AwsCloudDriver._bq_partition_predicate(
        "per_account", date(2026, 5, 1), date(2026, 5, 15),
    )
    assert "year =" in sql and "month =" in sql
    assert [(p.name, p.value) for p in params] == [
        ("bq_yr_0", 2026), ("bq_mo_0", 5),
    ]


def test_bq_partition_predicate_per_month_uses_native_day() -> None:
    """per_month (CUR 2.0 native) prunes on the table's DAY partition
    column ``line_item_usage_end_date`` via a timestamp range — Hive
    year/month INT columns DON'T EXIST on CUR 2.0 tables so the
    per_account predicate would fail with 'Unrecognized name year'."""
    sql, params = AwsCloudDriver._bq_partition_predicate(
        "per_month", date(2026, 5, 1), date(2026, 5, 15),
    )
    assert "line_item_usage_end_date" in sql
    assert "year" not in sql and "month" not in sql.replace("line_item", "")
    # Two timestamp params: start (inclusive) and end (exclusive, padded).
    by_name = {p.name: p for p in params}
    assert "bq_pp_start" in by_name
    assert "bq_pp_end" in by_name
    assert by_name["bq_pp_start"].type_ == "TIMESTAMP"


# ── _bq_tag_expression / _bq_product_field_expression ──────────


def test_bq_tag_expression_flat_returns_column_name() -> None:
    """CUR 1.0 tags are one flat column per key."""
    assert AwsCloudDriver._bq_tag_expression(
        "flat", "Owner",
    ) == "resource_tags_user_owner"


def test_bq_tag_expression_nested_returns_unnest_subquery() -> None:
    """CUR 2.0 tags live in ``resource_tags.key_value`` ARRAY<STRUCT> —
    array element key carries the ``user_`` prefix
    (``user_owner``, not ``owner``)."""
    expr = AwsCloudDriver._bq_tag_expression("nested", "Owner")
    assert "UNNEST(resource_tags.key_value)" in expr
    assert "v.key = 'user_owner'" in expr
    assert "LIMIT 1" in expr


def test_bq_tag_expression_nested_sanitises_non_alnum() -> None:
    """Non-alphanumeric characters in the tag key sanitise to ``_``
    in both schemas — required because the array key in CUR 2.0 is
    a string literal that must be a valid BQ identifier subset."""
    expr = AwsCloudDriver._bq_tag_expression("nested", "kubernetes.io/service-name")
    # Same sanitisation as ``_bq_tag_column``: dots/dashes → underscore.
    assert "v.key = 'user_kubernetes_io_service_name'" in expr


def test_bq_product_field_expression_flat_returns_prefixed_column() -> None:
    assert AwsCloudDriver._bq_product_field_expression(
        "flat", "product_name",
    ) == "product_product_name"


def test_bq_product_field_expression_nested_returns_unnest_subquery() -> None:
    expr = AwsCloudDriver._bq_product_field_expression("nested", "product_name")
    assert "UNNEST(product.key_value)" in expr
    assert "v.key = 'product_name'" in expr


def test_bq_dimension_expression_service_nested_uses_product_key_value() -> None:
    """The ``service`` dimension maps to ``product_product_name`` in
    CUR 1.0 (flat top-level column) and to ``UNNEST(product.key_value)
    WHERE key='product_name'`` in CUR 2.0 (nested record)."""
    expr_flat = AwsCloudDriver._bq_dimension_expression("flat", "service")
    expr_nested = AwsCloudDriver._bq_dimension_expression("nested", "service")
    assert expr_flat == "product_product_name"
    assert "UNNEST(product.key_value)" in expr_nested
    assert "v.key = 'product_name'" in expr_nested


def test_bq_dimension_expression_account_unchanged_across_schemas() -> None:
    """``linked_account`` / ``usage_type`` / ``operation`` etc. map to
    top-level ``line_item_*`` columns in both CUR 1.0 and CUR 2.0 —
    only the ``product_*`` family migrates to the nested record."""
    for alias in ("linked_account", "usage_type", "operation", "purchase_type"):
        assert (
            AwsCloudDriver._bq_dimension_expression("flat", alias)
            == AwsCloudDriver._bq_dimension_expression("nested", alias)
        )


# ── _bq_resolve_tables (per_month layout) ──────────────────────


def test_bq_resolve_tables_per_month_scoped_sets_account_filter() -> None:
    """per_month scope=acc resolves to N month-tables plus an
    account_filter that the caller injects as
    ``WHERE line_item_usage_account_id = @scope_acct``.

    Enumeration window is widened by +2 days on the end to catch
    rows that straddle month boundaries (start in month-N, end in
    month-(N+1) → row lives in N+1's table). For end=May 31, the
    padded end (June 2) brings June's table into the enumeration.
    """
    driver = _bq_driver_v2()
    client = MagicMock()
    client.get_table.return_value = _stub_table(())  # CUR 2.0 schema-uniform
    picks = driver._bq_resolve_tables(
        client,
        layout="per_month",
        dataset="p.d",
        pattern="aws_billing_{yyyymmdd}",
        scope=CloudScopeRef(scope_type="aws_account", external_id="385595570414"),
        start=date(2026, 4, 1), end=date(2026, 5, 31),
    )
    assert [p.fq_table for p in picks] == [
        "p.d.aws_billing_20260401",
        "p.d.aws_billing_20260501",
        "p.d.aws_billing_20260601",  # +2-day boundary pad
    ]
    assert all(p.account_filter == "385595570414" for p in picks)
    # CUR 2.0 schema-uniform → no per-table column probe needed.
    assert all(p.columns is None for p in picks)


def test_bq_resolve_tables_per_month_tenant_wide_no_account_filter() -> None:
    """per_month tenant-wide (scope=None) UNIONs every month-table in
    the (window + 2-day end pad) with NO account filter — every linked
    account's spend contributes."""
    driver = _bq_driver_v2()
    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    picks = driver._bq_resolve_tables(
        client,
        layout="per_month",
        dataset="p.d",
        pattern="aws_billing_{yyyymmdd}",
        scope=None,
        start=date(2026, 5, 1), end=date(2026, 5, 31),
    )
    # May + June (boundary pad) — see scoped-counterpart for rationale.
    assert [p.fq_table for p in picks] == [
        "p.d.aws_billing_20260501",
        "p.d.aws_billing_20260601",
    ]
    assert all(p.account_filter is None for p in picks)


def test_bq_resolve_tables_per_month_skips_notfound_months() -> None:
    """Window extends back to a month before the CUR-export began —
    those month-tables don't exist. Skip them rather than failing the
    whole run; same NotFound-graceful-skip contract as the per_account
    missing-table path.
    """
    from google.api_core.exceptions import NotFound

    driver = _bq_driver_v2()
    client = MagicMock()
    # Only the May 2026 table exists; April + June raise NotFound
    # (April predates CUR-export, June is a future-month that the
    # boundary pad pulls in but the CUR-export hasn't created yet).
    def _get_table(fq: str):
        if fq.endswith("aws_billing_20260501"):
            return _stub_table(())
        raise NotFound("missing")

    client.get_table.side_effect = _get_table
    picks = driver._bq_resolve_tables(
        client,
        layout="per_month",
        dataset="p.d",
        pattern="aws_billing_{yyyymmdd}",
        scope=CloudScopeRef(scope_type="aws_account", external_id="385595570414"),
        start=date(2026, 4, 1), end=date(2026, 5, 31),
    )
    assert [p.fq_table for p in picks] == ["p.d.aws_billing_20260501"]


def test_bq_resolve_tables_per_month_boundary_pad_within_same_month() -> None:
    """Window entirely within mid-month (start/end ≥ 2 days from any
    month boundary): the +2-day end pad MUST NOT add an extra month.
    Verifies the pad is bounded and doesn't double-scan for typical
    daily-cron queries.

    Pin: start=May 14, end=May 14 → padded end = May 16, still in May.
    Enumerate May only (NOT June).
    """
    driver = _bq_driver_v2()
    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    picks = driver._bq_resolve_tables(
        client,
        layout="per_month",
        dataset="p.d",
        pattern="aws_billing_{yyyymmdd}",
        scope=CloudScopeRef(scope_type="aws_account", external_id="385595570414"),
        start=date(2026, 5, 14), end=date(2026, 5, 14),
    )
    assert [p.fq_table for p in picks] == [
        "p.d.aws_billing_20260501",
    ]


def test_bq_resolve_tables_per_month_boundary_pad_catches_next_month_for_month_end_query() -> None:
    """The core bug class the boundary pad fixes: a row with
    ``start_date=April 30 23:30Z, end_date=May 1 00:30Z`` lives in
    MAY's month-table (partitioned on end_date) but has
    ``DAY=April 30``. Without the pad, an April-window query would
    enumerate April's table only → row invisible → April 30 cost
    silently undercounts.

    Pin: end=April 30 → padded end = May 2 → enumerate April + May.
    The day-grain filter then drops pure-May rows from May's table
    so we don't double-count; only boundary-straddling rows survive.
    """
    driver = _bq_driver_v2()
    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    picks = driver._bq_resolve_tables(
        client,
        layout="per_month",
        dataset="p.d",
        pattern="aws_billing_{yyyymmdd}",
        scope=CloudScopeRef(scope_type="aws_account", external_id="385595570414"),
        start=date(2026, 4, 1), end=date(2026, 4, 30),
    )
    assert [p.fq_table for p in picks] == [
        "p.d.aws_billing_20260401",
        "p.d.aws_billing_20260501",  # boundary pad
    ]


def test_bq_resolve_tables_per_account_unchanged() -> None:
    """Smoke check that the per_account path still resolves to one
    pick per scope with ``account_filter=None`` (the table name itself
    filters by account)."""
    driver = _bq_driver()
    client = MagicMock()
    client.get_table.return_value = _stub_table(_DEFAULT_CUR_COLUMNS)
    picks = driver._bq_resolve_tables(
        client,
        layout="per_account",
        dataset="p.d",
        pattern="{account_id}_aws_billing_data",
        scope=CloudScopeRef(scope_type="aws_account", external_id="123456789012"),
        start=date(2026, 5, 1), end=date(2026, 5, 31),
    )
    assert len(picks) == 1
    assert picks[0].fq_table == "p.d.123456789012_aws_billing_data"
    assert picks[0].account_filter is None
    assert picks[0].columns is not None  # schema columns probed


# ── Dataset-existence probe (per_month dataset-typo guard) ─────


@pytest.mark.parametrize("layout,pattern,scope_ext", [
    ("per_month", "aws_billing_{yyyymmdd}", "385595570414"),
    ("per_account", "{account_id}_aws_billing_data", "385595570414"),
    ("per_account", "{account_id}_aws_billing_data", None),  # tenant-wide
])
def test_bq_resolve_tables_raises_when_dataset_missing(
    layout: str, pattern: str, scope_ext: str | None,
) -> None:
    """Admin typo'd ``billing_bq_dataset`` (or the SA lacks
    ``bigquery.datasets.get`` on it). The pre-fix silent-skip class:

      per_account scoped:  every ``get_table(<typo>.<acc>_…)`` →
                           NotFound → caught + return [] → cron ok,
                           60 scopes zero rows.
      per_month  scoped/tenant: every ``get_table(<typo>.aws_…)`` →
                           NotFound → caught + continue → picks=[] →
                           cron ok, zero rows.

    Both layouts must now raise loudly via the shared
    ``_bq_probe_dataset_exists`` helper at the top of
    ``_bq_resolve_tables``. The probe MUST happen BEFORE any per-table
    ``get_table`` so the operator-visible error reads as a clear
    dataset-level failure rather than a per-table miss.

    Parametrized over the three layout × scope combinations that hit
    the silent-skip class. (per_account tenant-wide is technically
    also covered by ``_bq_discover_account_tables`` raising on the
    INFORMATION_SCHEMA query, but the upfront probe is cheap and
    aligns the operator's error message across paths.)
    """
    from google.api_core.exceptions import NotFound

    driver = _bq_driver_v2() if layout == "per_month" else _bq_driver()
    client = MagicMock()
    client.get_dataset.side_effect = NotFound("dataset not found")
    scope = (
        CloudScopeRef(scope_type="aws_account", external_id=scope_ext)
        if scope_ext is not None else None
    )

    with pytest.raises(RuntimeError, match="was not found"):
        driver._bq_resolve_tables(
            client,
            layout=layout,
            dataset="typo-bi.aws_billings_no_payer",
            pattern=pattern,
            scope=scope,
            start=date(2026, 4, 1), end=date(2026, 5, 31),
        )
    # Probe MUST happen BEFORE the per-table get_table enumeration —
    # otherwise the per-table NotFound masks the real "dataset missing"
    # failure as a per-table-miss → silent skip. Verify the per-table
    # resolution path was never reached.
    client.get_table.assert_not_called()


def test_bq_resolve_tables_propagates_non_notfound_dataset_errors() -> None:
    """A 403 / 5xx from ``get_dataset`` (Forbidden, transient cloud
    error) must propagate unchanged so the operator gets the real
    failure surface instead of a flattened "dataset missing" message.
    Only NotFound becomes the RuntimeError wrapping; everything else
    passes through.
    """
    from google.api_core.exceptions import Forbidden

    driver = _bq_driver_v2()
    client = MagicMock()
    client.get_dataset.side_effect = Forbidden("permission denied")
    with pytest.raises(Forbidden):
        driver._bq_resolve_tables(
            client,
            layout="per_month",
            dataset="p.d",
            pattern="aws_billing_{yyyymmdd}",
            scope=CloudScopeRef(
                scope_type="aws_account", external_id="385595570414",
            ),
            start=date(2026, 5, 1), end=date(2026, 5, 31),
        )
    client.get_table.assert_not_called()


def test_bq_resolve_tables_dataset_probe_runs_before_enumeration() -> None:
    """Probe call-order invariant: ``get_dataset`` MUST run BEFORE any
    ``get_table`` for the per-month / per-account enumeration. If the
    order inverted, a per-month-table NotFound (legitimate gap before
    CUR-export start) could be raised by the per-table loop BEFORE the
    dataset-level probe gets to issue its clear-error message — the
    operator would see "no CUR table for month 2024-09" when the real
    cause is a dataset-name typo affecting every month.

    Asserts strict order: get_dataset is the FIRST BQ call, and every
    get_table call comes AFTER it (no get_table can precede the probe
    even by one position).
    """
    driver = _bq_driver_v2()
    client = MagicMock()
    call_order: list[str] = []
    client.get_dataset.side_effect = lambda *a, **kw: call_order.append("get_dataset")
    client.get_table.side_effect = lambda *a, **kw: call_order.append("get_table")
    driver._bq_resolve_tables(
        client,
        layout="per_month",
        dataset="p.d",
        pattern="aws_billing_{yyyymmdd}",
        scope=CloudScopeRef(
            scope_type="aws_account", external_id="385595570414",
        ),
        start=date(2026, 5, 1), end=date(2026, 5, 31),
    )
    # Strict ordering: first call get_dataset, then EVERY subsequent
    # call must be get_table (no other ordering acceptable).
    assert call_order, "no BQ calls captured — test setup defect"
    assert call_order[0] == "get_dataset"
    assert all(c == "get_table" for c in call_order[1:]), (
        f"calls after the dataset probe must all be get_table — got: "
        f"{call_order!r}"
    )


def test_bq_resolve_tables_dataset_probe_caches_within_driver_instance() -> None:
    """Per-tick efficiency invariant: the daily cron fans out ~60
    scopes through the SAME driver instance. Issuing 60 redundant
    ``get_dataset`` round trips for the same dataset string would
    burn metadata-API quota and tick latency. Verify the probe caches
    on the driver instance — second + subsequent ``_bq_resolve_tables``
    calls with the same dataset MUST NOT re-invoke ``get_dataset``.

    Cache key is the dataset STRING; success-only caching means a
    transient Forbidden / 5xx still re-probes on the next call (so a
    permission grant lands without a worker restart).
    """
    driver = _bq_driver_v2()
    client = MagicMock()
    # get_dataset returns truthy MagicMock by default; that's a "probe
    # passed" signal.
    client.get_table.return_value = _stub_table(())

    # First call — probe issued.
    driver._bq_resolve_tables(
        client,
        layout="per_month",
        dataset="p.d",
        pattern="aws_billing_{yyyymmdd}",
        scope=CloudScopeRef(
            scope_type="aws_account", external_id="385595570414",
        ),
        start=date(2026, 5, 1), end=date(2026, 5, 31),
    )
    assert client.get_dataset.call_count == 1

    # Subsequent calls (different scope, same dataset) — probe should
    # NOT re-fire.
    for scope_ext in ("123456789012", "987654321098"):
        driver._bq_resolve_tables(
            client,
            layout="per_month",
            dataset="p.d",
            pattern="aws_billing_{yyyymmdd}",
            scope=CloudScopeRef(
                scope_type="aws_account", external_id=scope_ext,
            ),
            start=date(2026, 5, 1), end=date(2026, 5, 31),
        )
    assert client.get_dataset.call_count == 1, (
        f"Probe re-fired on subsequent same-dataset calls — got "
        f"{client.get_dataset.call_count} total invocations; expected 1."
    )

    # Different dataset → probe runs again.
    driver._bq_resolve_tables(
        client,
        layout="per_month",
        dataset="different.dataset",
        pattern="aws_billing_{yyyymmdd}",
        scope=CloudScopeRef(
            scope_type="aws_account", external_id="385595570414",
        ),
        start=date(2026, 5, 1), end=date(2026, 5, 31),
    )
    assert client.get_dataset.call_count == 2


def test_bq_resolve_tables_dataset_probe_only_caches_on_success() -> None:
    """A transient Forbidden / 5xx must NOT poison the cache — a
    permission grant or quota recovery should take effect on the NEXT
    call without requiring a worker restart.
    """
    from google.api_core.exceptions import Forbidden

    driver = _bq_driver_v2()
    client = MagicMock()
    # First call: 403, second call: success.
    client.get_dataset.side_effect = [Forbidden("denied"), None]
    client.get_table.return_value = _stub_table(())

    with pytest.raises(Forbidden):
        driver._bq_resolve_tables(
            client,
            layout="per_month",
            dataset="p.d",
            pattern="aws_billing_{yyyymmdd}",
            scope=CloudScopeRef(
                scope_type="aws_account", external_id="385595570414",
            ),
            start=date(2026, 5, 1), end=date(2026, 5, 31),
        )

    # Second call must re-issue the probe — cache must NOT have been
    # populated by the failed first call.
    driver._bq_resolve_tables(
        client,
        layout="per_month",
        dataset="p.d",
        pattern="aws_billing_{yyyymmdd}",
        scope=CloudScopeRef(
            scope_type="aws_account", external_id="385595570414",
        ),
        start=date(2026, 5, 1), end=date(2026, 5, 31),
    )
    assert client.get_dataset.call_count == 2


@pytest.mark.asyncio
async def test_list_billing_daily_propagates_dataset_missing_through_to_thread() -> None:
    """End-to-end integration: a missing-dataset RuntimeError raised
    inside ``_bq_resolve_tables`` must propagate cleanly through
    ``asyncio.to_thread`` to the awaiting caller (the cron handler
    that aggregates into ``scheduled_job_runs.errors[]``). If the
    RuntimeError got swallowed at the to-thread boundary, a typo'd
    dataset would land as a silent ok-with-zero-data run again.
    """
    from google.api_core.exceptions import NotFound

    driver = _bq_driver_v2()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="385595570414",
    )
    client = MagicMock()
    client.get_dataset.side_effect = NotFound("dataset not found")
    with patch.object(driver, "_bq_client", return_value=client):
        with pytest.raises(RuntimeError, match="was not found"):
            await driver.list_billing_daily(
                date(2026, 5, 1), date(2026, 5, 3),
                scope=scope, bq_billing=_BQ_BILLING_CUR_V2,
            )
    # FD-leak guard: client must be closed even on this error path.
    client.close.assert_called_once()


# ── End-to-end CUR 2.0 + per_month: cost_facts ────────────────


@pytest.mark.asyncio
async def test_list_cost_facts_bq_per_month_unions_month_tables_and_filters_account() -> None:
    """Single-calendar-month window (April) in per_month + nested mode:
    the SQL must UNION ALL the month-table AND the next month-table
    (the ``_bq_resolve_tables`` +2-day end pad pulls the May table in
    so boundary-spanning rows stay visible), inject
    ``line_item_usage_account_id = @scope_acct`` in each leg, and
    project tags via correlated subqueries against the nested array.

    The window stays inside ONE calendar month, so ``_bq_cost_facts_windows``
    leaves it as one chunk → exactly one BQ query — UNION-of-two-tables
    semantics are orthogonal to the multi-month chunking covered by
    ``test_list_cost_facts_bq_chunks_long_per_month_window``."""
    driver = _bq_driver_v2()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="385595570414",
    )
    captured_sqls: list[str] = []
    captured_params: list[list[object]] = []

    # One row from each month-table (April + May 2026) for the same
    # (day, account, product, resource, currency, usage_type) — the
    # Python post-aggregator must sum them into a single CostFact.
    row_april = SimpleNamespace(
        day=date(2026, 5, 1),  # row's start_date late-Apr; partition end_date Apr
        account_id="385595570414",
        product_code="AmazonEC2",
        service_name="Amazon Elastic Compute Cloud",
        resource_id="i-0abc1234",
        currency="USD",
        region="us-west-2",
        usage_type="BoxUsage:m5.large",
        usage_unit="Hrs",
        # CUR 2.0 nested tags: the per_month/nested data query projects
        # ``ANY_VALUE(resource_tags.key_value) AS tag_kv`` and Python
        # parses the ARRAY<STRUCT<key,value>> (stripping the ``user_``
        # prefix). The old flat ``tag_owner=`` attributes are ignored
        # by the nested parser.
        tag_kv=[{"key": "user_owner", "value": "alice@example.com"}],
        native_cost=2.0, usage_amount=2.0,
    )
    row_may = SimpleNamespace(
        day=date(2026, 5, 1),
        account_id="385595570414",
        product_code="AmazonEC2",
        service_name="Amazon Elastic Compute Cloud",
        resource_id="i-0abc1234",
        currency="USD",
        region="us-west-2",
        usage_type="BoxUsage:m5.large",
        usage_unit="Hrs",
        # CUR 2.0 nested tags: the per_month/nested data query projects
        # ``ANY_VALUE(resource_tags.key_value) AS tag_kv`` and Python
        # parses the ARRAY<STRUCT<key,value>> (stripping the ``user_``
        # prefix). The old flat ``tag_owner=`` attributes are ignored
        # by the nested parser.
        tag_kv=[{"key": "user_owner", "value": "alice@example.com"}],
        native_cost=8.0, usage_amount=8.0,
    )

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        captured_params.append(kw.get("job_config").query_parameters)
        job = MagicMock()
        job.result.return_value = iter([row_april, row_may])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(())  # nested schema uniform
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        facts = await driver.list_cost_facts(
            date(2026, 4, 15), date(2026, 4, 30),
            scope=scope, bq_billing=_BQ_BILLING_CUR_V2,
        )

    assert len(captured_sqls) == 1, "single UNION ALL query expected"
    sql = captured_sqls[0]
    # Both month-tables referenced.
    assert "aws_billing_20260401" in sql
    assert "aws_billing_20260501" in sql
    # Account filter injected in each leg.
    assert sql.count("line_item_usage_account_id = @scope_acct") == 2
    # Nested tag projection — flat column should NEVER appear. The data
    # query surfaces ALL user tags via a single ``ANY_VALUE`` of the
    # nested array (parsed in Python), NOT a per-key UNNEST subquery —
    # that per-key form is only used by the dimension/group-by path.
    assert "resource_tags_user_owner" not in sql
    assert "ANY_VALUE(resource_tags.key_value) AS tag_kv" in sql
    # Nested product fields.
    assert "UNNEST(product.key_value)" in sql
    assert "v.key = 'product_name'" in sql
    # @scope_acct bound once with the right value.
    scope_params = [
        p for params in captured_params for p in params
        if p.name == "scope_acct"
    ]
    assert len(scope_params) == 1
    assert scope_params[0].value == "385595570414"

    # Cross-month aggregation: 2.0 (Apr) + 8.0 (May) = 10.0
    assert len(facts) == 1
    assert facts[0].amount_native == pytest.approx(10.0)
    assert facts[0].tags == {"owner": "alice@example.com"}


@pytest.mark.asyncio
async def test_list_cost_facts_bq_chunks_long_per_month_window() -> None:
    """A multi-month per_month (CUR 2.0) window is split on CALENDAR-
    MONTH boundaries into one BQ query per month, so no single query
    trips ``maximum_bytes_billed`` — the prod AWS month-close regression
    where a one-month re-pull scanned ~68.5 GiB > the old 50 GiB cap and
    aborted with ``bytesBilledLimitExceeded`` (#1197/#1200 follow-up).
    Sub-windows partition the range by start-day, so per-window facts
    concatenate with no fact_key clash. Month boundaries (not arbitrary
    day counts) keep each chunk identical to a standalone monthly close,
    avoiding the spanning-row drop a mid-month boundary would cause."""
    driver = _bq_driver_v2()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="385595570414",
    )
    seen_windows: list[tuple[date, date]] = []

    def _capture_query(sql, **kw):
        params = {
            p.name: p.value for p in kw["job_config"].query_parameters
        }
        w_start, w_end = params["start"], params["end"]
        seen_windows.append((w_start, w_end))
        # One distinct fact per chunk (day = chunk start) so the
        # concatenated result carries one row per sub-query.
        job = MagicMock()
        job.result.return_value = iter([
            SimpleNamespace(
                day=w_start, account_id="385595570414",
                product_code="AmazonEC2",
                service_name="Amazon Elastic Compute Cloud",
                resource_id=f"i-{w_start.isoformat()}",
                currency="USD", region="us-west-2",
                usage_type="BoxUsage:m5.large", usage_unit="Hrs",
                tag_kv=[], native_cost=1.0, usage_amount=1.0,
            ),
        ])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        facts = await driver.list_cost_facts(
            date(2026, 4, 15), date(2026, 5, 14),  # spans Apr + May
            scope=scope, bq_billing=_BQ_BILLING_CUR_V2,
        )

    # Window spans two calendar months → one query per month, each
    # window the slice of that month inside [start, end].
    assert seen_windows == [
        (date(2026, 4, 15), date(2026, 4, 30)),
        (date(2026, 5, 1), date(2026, 5, 14)),
    ]
    # Per-window facts concatenate — one per chunk here.
    assert {f.date for f in facts} == {date(2026, 4, 15), date(2026, 5, 1)}


@pytest.mark.asyncio
async def test_list_cost_facts_bq_per_account_does_not_chunk_long_window() -> None:
    """CUR 1.0 ``per_account`` tables are MONTH-partitioned with one
    table per account: chunking a multi-month window gives no byte
    savings, so the window stays whole — a single query with the full
    ``[start, end]``."""
    driver = _bq_driver()  # per_account (default)
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    seen_windows: list[tuple[date, date]] = []

    def _capture_query(sql, **kw):
        params = {
            p.name: p.value for p in kw["job_config"].query_parameters
        }
        seen_windows.append((params["start"], params["end"]))
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = _stub_bq_client_with_rows([])
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            date(2026, 4, 15), date(2026, 5, 14),  # 30-day window
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )

    assert seen_windows == [(date(2026, 4, 15), date(2026, 5, 14))]


@pytest.mark.asyncio
async def test_list_cost_facts_bq_per_month_routes_to_cur_v2_helpers() -> None:
    """Smoke check that the driver dispatches to ``_bq_resolve_tables``
    + per_month helpers when ``table_layout=per_month``. Guards against
    a future routing inversion silently falling back to per_account."""
    driver = _bq_driver_v2()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="385595570414",
    )
    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    job = MagicMock()
    job.result.return_value = iter([])
    client.query.return_value = job

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_CUR_V2,
        )
    # The driver MUST resolve to exactly one month-table (May) with
    # the account filter — `_bq_resolve_tables` is the gate.
    # `client.query` ran one UNION-of-1 SQL — verify its FROM clause.
    sql = client.query.call_args[0][0]
    assert "aws_billing_20260501" in sql
    assert "{account_id}" not in sql  # no leftover placeholder


# ── End-to-end CUR 2.0 + per_month: billing_daily ─────────────


@pytest.mark.asyncio
async def test_list_billing_daily_bq_per_month_owner_breakdown_uses_nested() -> None:
    """CUR 2.0 owner_breakdown source is a correlated subquery against
    ``resource_tags.key_value`` (key='user_owner'), not the flat
    ``resource_tags_user_owner`` column. The schema-drop logic that
    flat needs (drop UNION legs over tables lacking the column) is a
    noop here — every per-month table has the nested array on every
    row, so every pick contributes."""
    driver = _bq_driver_v2()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="385595570414",
    )
    captured_sqls: list[str] = []

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_billing_daily(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, include_owner_breakdown=True,
            bq_billing=_BQ_BILLING_CUR_V2,
        )

    # 3 queries fire: totals, services, owner_breakdown.
    assert len(captured_sqls) == 3
    owner_sql = captured_sqls[2]
    assert "resource_tags_user_owner" not in owner_sql
    assert "UNNEST(resource_tags.key_value)" in owner_sql
    assert "v.key = 'user_owner'" in owner_sql
    # owner_breakdown WHERE filter on tag value uses the same subquery.
    assert owner_sql.count("v.key = 'user_owner'") >= 2


# ── End-to-end CUR 2.0 + per_month: billing_by_dimension ─────


@pytest.mark.asyncio
async def test_list_billing_by_dimension_bq_per_month_tag_dimension_uses_nested() -> None:
    """``tag:Owner`` group_by in CUR 2.0 mode projects the dimension
    as ``UNNEST(resource_tags.key_value) WHERE key='user_owner'`` —
    not as a flat column reference. No per-table schema-drop runs
    (nested arrays are present uniformly)."""
    driver = _bq_driver_v2()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="385595570414",
    )
    captured_sqls: list[str] = []

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_billing_by_dimension(
            date(2026, 5, 14), date(2026, 5, 14),
            group_by=["tag:Owner"], scope=scope, filters={},
            bq_billing=_BQ_BILLING_CUR_V2,
        )

    sql = captured_sqls[0]
    assert "resource_tags_user_owner" not in sql
    assert "UNNEST(resource_tags.key_value)" in sql
    assert "v.key = 'user_owner'" in sql
    # ``_bq_resolve_table_columns`` MUST NOT have been called — the
    # schema-drop logic only applies to flat. Verify the
    # INFORMATION_SCHEMA path didn't run.
    information_schema_calls = [
        c for c in client.query.call_args_list
        if "INFORMATION_SCHEMA" in str(c)
    ]
    assert not information_schema_calls, (
        f"per_month nested must skip the per-table schema probe; "
        f"unexpected INFORMATION_SCHEMA queries: {information_schema_calls}"
    )


@pytest.mark.asyncio
async def test_list_billing_by_dimension_bq_per_month_service_dimension_uses_nested() -> None:
    """``service`` group_by in CUR 2.0 mode uses
    ``UNNEST(product.key_value) WHERE key='product_name'`` (not the
    flat ``product_product_name`` column that doesn't exist in CUR 2.0).
    """
    driver = _bq_driver_v2()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="385595570414",
    )
    captured_sqls: list[str] = []

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_billing_by_dimension(
            date(2026, 5, 14), date(2026, 5, 14),
            group_by=["service"], scope=scope, filters={},
            bq_billing=_BQ_BILLING_CUR_V2,
        )

    sql = captured_sqls[0]
    assert "product_product_name" not in sql
    assert "UNNEST(product.key_value)" in sql
    assert "v.key = 'product_name'" in sql


# ── End-to-end CUR 2.0 + per_month: legacy CE/CUR-1.0 paths still work ──


@pytest.mark.asyncio
async def test_per_account_flat_path_byte_for_byte_unchanged_for_existing_aws_primary() -> None:
    """Sanity guard: a driver configured with the legacy defaults
    (table_layout=per_account, tag_schema=flat — what aws-primary used
    before this PR) still produces the CUR 1.0 SQL shape with Hive
    year/month partition predicate, per-account table routing, and
    flat ``resource_tags_user_<key>`` projections. Existing prod
    integrations should be noop-compatible through the deploy of this
    refactor; the per_month flip is an explicit opt-in via config."""
    driver = _bq_driver()  # legacy CUR 1.0 config
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="385595570414",
    )
    captured_sqls: list[str] = []

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(_DEFAULT_CUR_COLUMNS)
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,
        )

    sql = captured_sqls[0]
    # Hive partition predicate, NOT native-DAY range.
    assert "year = @bq_yr_0 AND month = @bq_mo_0" in sql
    # Per-account table name routing (not month-table enumeration).
    assert "385595570414_aws_billing_data" in sql
    assert "aws_billing_20260501" not in sql
    # Flat tag column references (NOT nested array subqueries).
    assert "resource_tags_user_owner" in sql
    assert "UNNEST(resource_tags.key_value)" not in sql
    # Flat product field (NOT nested).
    assert "product_product_name" in sql
    assert "UNNEST(product.key_value)" not in sql
    # No account filter — per_account routes by table name.
    assert "line_item_usage_account_id = @scope_acct" not in sql


# ── fact_grain (CUR 2.0 row-volume bound) ─────────────────────


_BQ_BILLING_CUR_V2_PER_RESOURCE = {
    **_BQ_BILLING_CUR_V2,
    "fact_grain": "per_resource",
}


def test_bq_resolve_options_rejects_invalid_fact_grain() -> None:
    with pytest.raises(RuntimeError, match="fact_grain"):
        AwsCloudDriver._bq_resolve_options({
            "credentials_json": _FAKE_SA,
            "dataset": "p.d",
            "fact_grain": "per_universe",
        })


def test_bq_resolve_options_default_fact_grain_is_per_product() -> None:
    """Default keeps cost_facts row volume bounded across the dataset
    cutover. The CUR 1.0 per_account+flat path effectively behaved as
    per_product for accounts whose CUR table lacked
    ``line_item_resource_id`` (the synthetic-id branch fired) —
    explicit per_product default preserves that contract for the
    CUR 2.0 nested path too."""
    _, _, _, _, _, fact_grain = AwsCloudDriver._bq_resolve_options({
        "credentials_json": _FAKE_SA,
        "dataset": "p.d",
    })
    assert fact_grain == "per_product"


@pytest.mark.asyncio
async def test_list_cost_facts_bq_per_product_nulls_resource_id_in_select() -> None:
    """per_product grain MUST project ``CAST(NULL AS STRING) AS
    resource_id`` regardless of layout / tag_schema, so the GROUP BY
    collapses every row of the same product into one bucket and the
    Python synthesises the stable ``aws-account:<acc>:<product>`` id
    downstream. Without this, CUR 2.0 (which always has
    ``line_item_resource_id`` populated) would emit one fact per
    resource × usage_type — 272x more rows than CUR 1.0 in PingCAP
    prod, exhausting TiDB upsert throughput.
    """
    driver = _bq_driver_v2()  # per_month + nested (CUR 2.0 config)
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="385595570414",
    )
    captured_sqls: list[str] = []

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_CUR_V2,
        )

    sql = captured_sqls[0]
    # Must NEVER reference the actual line_item_resource_id column —
    # otherwise GROUP BY would re-introduce the 272x explosion.
    assert "line_item_resource_id AS resource_id" not in sql
    assert "CAST(NULL AS STRING) AS resource_id" in sql


@pytest.mark.asyncio
async def test_list_cost_facts_bq_per_resource_projects_real_resource_id() -> None:
    """per_resource grain (explicit opt-in) projects the real
    ``line_item_resource_id`` column so cost-allocation rules can match
    individual resource ARNs. Caller MUST size the cron timeout for
    the resulting volume.
    """
    driver = _bq_driver_v2()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="385595570414",
    )
    captured_sqls: list[str] = []

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_CUR_V2_PER_RESOURCE,
        )

    sql = captured_sqls[0]
    assert "line_item_resource_id AS resource_id" in sql
    assert "CAST(NULL AS STRING) AS resource_id" not in sql


@pytest.mark.asyncio
async def test_list_cost_facts_bq_per_resource_on_cur_v1_flat_uses_real_resource_id_when_present() -> None:
    """fact_grain=per_resource opt-in on legacy CUR 1.0 (per_account +
    flat) projects the real ``line_item_resource_id`` column when the
    table's schema carries it. Mirrors the prior CUR 1.0 default
    behaviour for accounts whose CUR was built with
    "Include resource IDs" enabled.
    """
    driver = _bq_driver()  # per_account + flat (CUR 1.0)
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    bq_billing_per_resource = {
        **_BQ_BILLING_OK,
        "fact_grain": "per_resource",
    }
    captured_sqls: list[str] = []

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(_DEFAULT_CUR_COLUMNS)
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=bq_billing_per_resource,
        )

    sql = captured_sqls[0]
    assert "line_item_resource_id AS resource_id" in sql
    assert "CAST(NULL AS STRING) AS resource_id" not in sql


@pytest.mark.asyncio
async def test_list_cost_facts_bq_per_resource_on_cur_v1_flat_falls_back_to_null_when_absent() -> None:
    """fact_grain=per_resource opt-in on CUR 1.0 with a per-account
    table that LACKS ``line_item_resource_id`` (the PR #311 schema-
    heterogeneity case) MUST still project ``CAST(NULL AS STRING) AS
    resource_id`` so the SELECT is well-formed, and the Python
    synthetic-id path picks up the missing rows. Combined-axes
    coverage: fact_grain doesn't override the per-pick column-absence
    guard for legacy CUR 1.0 tables.
    """
    driver = _bq_driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="470438934515",
    )
    bq_billing_per_resource = {
        **_BQ_BILLING_OK,
        "fact_grain": "per_resource",
    }
    columns = tuple(
        c for c in _DEFAULT_CUR_COLUMNS if c != "line_item_resource_id"
    )
    captured_sqls: list[str] = []

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(columns)
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=bq_billing_per_resource,
        )

    sql = captured_sqls[0]
    assert "line_item_resource_id AS resource_id" not in sql
    assert "CAST(NULL AS STRING) AS resource_id" in sql


@pytest.mark.asyncio
async def test_list_cost_facts_bq_per_product_on_cur_v1_flat_nulls_resource_id_even_when_present() -> None:
    """fact_grain=per_product on CUR 1.0 with a per-account table that
    HAS ``line_item_resource_id`` MUST still NULL it out — the
    per_product contract is "aggregate at product level, never emit
    per-resource facts" regardless of layout / tag_schema /
    column-presence. Pre-fix the driver always projected the real
    column when present; this test pins the per_product override.
    """
    driver = _bq_driver()  # per_account + flat (CUR 1.0); default fact_grain=per_product
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
    )
    captured_sqls: list[str] = []

    def _capture_query(sql, **kw):
        captured_sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(_DEFAULT_CUR_COLUMNS)
    client.query.side_effect = _capture_query

    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_OK,  # default fact_grain=per_product
        )

    sql = captured_sqls[0]
    # Column is present in the schema (full _DEFAULT_CUR_COLUMNS) but
    # per_product MUST NULL it anyway.
    assert "line_item_resource_id AS resource_id" not in sql
    assert "CAST(NULL AS STRING) AS resource_id" in sql


@pytest.mark.asyncio
async def test_list_cost_facts_bq_per_product_synthesises_resource_id_in_python() -> None:
    """End-to-end: per_product SELECT projects NULL resource_id; BQ
    row arrives with ``resource_id=None``; Python synthesises
    ``aws-account:<acc>:<product>``; the emitted CostFact's
    resource_id is that synthetic value and resource_type is
    ``aws-aggregated``. This is the exact contract CUR 1.0 already
    relied on via the absent-column branch.
    """
    driver = _bq_driver_v2()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="385595570414",
    )
    row = SimpleNamespace(
        day=date(2026, 5, 14),
        account_id="385595570414",
        product_code="AmazonEC2",
        service_name="Amazon Elastic Compute Cloud",
        resource_id=None,  # per_product → NULL
        currency="USD",
        region="us-west-2",
        usage_type="BoxUsage:m5.large",
        usage_unit="Hrs",
        # CUR 2.0 nested tags: the per_month/nested data query projects
        # ``ANY_VALUE(resource_tags.key_value) AS tag_kv`` and Python
        # parses the ARRAY<STRUCT<key,value>> (stripping the ``user_``
        # prefix). The old flat ``tag_owner=`` attributes are ignored
        # by the nested parser.
        tag_kv=[{"key": "user_owner", "value": "alice@example.com"}],
        native_cost=42.0,
        usage_amount=24.0,
    )
    with patch.object(
        driver, "_bq_client",
        return_value=_stub_bq_client_with_rows([row], columns=()),
    ):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope, bq_billing=_BQ_BILLING_CUR_V2,
        )
    assert len(facts) == 1
    assert facts[0].resource_id == "aws-account:385595570414:AmazonEC2"
    assert facts[0].resource_type == "aws-aggregated"
    assert facts[0].tags == {"owner": "alice@example.com"}


# ── Batched all-accounts cost-facts path (per_month, scan-once) ──
#
# For CUR 2.0 (per_month) every linked account shares one DAY-partitioned
# month-table, and the per-account ``WHERE line_item_usage_account_id``
# filter does NOT prune bytes (the table is partitioned by date, not
# clustered by account). Issuing one query per account therefore
# re-scans the same partitions N× — the prod cost bleed this path fixes.
# When cost_service hands the driver the full target scope set via
# ``_cost_facts_all_account_scopes``, the FIRST scope's call scans the
# window ONCE for all accounts and the rest are served from a per-run
# cache. These tests pin: scan-once, the SQL drops the account filter,
# each fact carries the scope matching its row's account, unregistered
# accounts are dropped, and every fall-back (no target set / span too
# wide / per_resource grain) keeps the proven per-account behaviour.

_ACC_A = "111111111111"
_ACC_B = "222222222222"
_ACC_UNREG = "999999999999"


def _v2_row(
    account_id: str, *, cost: float = 10.0, resource: str = "i-0abc",
) -> SimpleNamespace:
    return SimpleNamespace(
        day=date(2026, 5, 14), account_id=account_id,
        product_code="AmazonEC2",
        service_name="Amazon Elastic Compute Cloud",
        resource_id=resource, currency="USD", region="us-west-2",
        usage_type="BoxUsage:m5.large", usage_unit="Hrs",
        tag_kv=[], native_cost=cost, usage_amount=1.0,
    )


@pytest.mark.asyncio
async def test_cost_facts_batched_scans_once_serves_all_scopes() -> None:
    driver = _bq_driver_v2()
    scope_a = CloudScopeRef(scope_type="aws_account", external_id=_ACC_A)
    scope_b = CloudScopeRef(scope_type="aws_account", external_id=_ACC_B)
    driver._cost_facts_all_account_scopes = [scope_a, scope_b]
    client = _stub_bq_client_with_rows([
        _v2_row(_ACC_A, cost=10.0), _v2_row(_ACC_B, cost=20.0),
    ])
    with patch.object(driver, "_bq_client", return_value=client):
        facts_a = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope_a, bq_billing=_BQ_BILLING_CUR_V2,
        )
        facts_b = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope_b, bq_billing=_BQ_BILLING_CUR_V2,
        )
    # ONE scan total — the second scope is served from the per-run cache.
    assert client.query.call_count == 1
    # The per-account filter is gone (its whole point: it bills identical
    # bytes, so one all-accounts scan replaces N per-account scans).
    sql = client.query.call_args.args[0]
    assert "@scope_acct" not in sql
    assert "line_item_usage_account_id = @scope_acct" not in sql
    # Each scope gets exactly its own account's fact, with the matching
    # scope object stamped (compute_fact_key keys on scope external id,
    # so this must be the row's account — not the call's scope).
    assert [f.scope.external_id for f in facts_a] == [_ACC_A]
    assert [f.scope.external_id for f in facts_b] == [_ACC_B]
    assert facts_a[0].resource_group == _ACC_A
    assert facts_a[0].amount_usd == pytest.approx(10.0)
    assert facts_b[0].amount_usd == pytest.approx(20.0)


@pytest.mark.asyncio
async def test_cost_facts_batched_drops_unregistered_account() -> None:
    # A row for an account the cron never registered as a scope (another
    # tenant sharing the month-table) must not surface under any scope.
    driver = _bq_driver_v2()
    scope_a = CloudScopeRef(scope_type="aws_account", external_id=_ACC_A)
    driver._cost_facts_all_account_scopes = [scope_a]
    client = _stub_bq_client_with_rows([
        _v2_row(_ACC_A, cost=10.0), _v2_row(_ACC_UNREG, cost=99.0),
    ])
    with patch.object(driver, "_bq_client", return_value=client):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope_a, bq_billing=_BQ_BILLING_CUR_V2,
        )
    assert [f.scope.external_id for f in facts] == [_ACC_A]
    assert all(f.amount_usd != pytest.approx(99.0) for f in facts)


@pytest.mark.asyncio
async def test_cost_facts_per_month_without_target_set_keeps_account_filter() -> None:
    # No target set supplied → per-account path: SQL keeps @scope_acct and
    # the batch cache is never populated (backward-compat for callers like
    # the Bedrock-profile / monthclose paths that don't opt in).
    driver = _bq_driver_v2()
    scope_a = CloudScopeRef(scope_type="aws_account", external_id=_ACC_A)
    client = _stub_bq_client_with_rows([_v2_row(_ACC_A)])
    with patch.object(driver, "_bq_client", return_value=client):
        facts = await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=scope_a, bq_billing=_BQ_BILLING_CUR_V2,
        )
    sql = client.query.call_args.args[0]
    assert "line_item_usage_account_id = @scope_acct" in sql
    assert driver._bq_cost_facts_batch_cache == {}
    assert len(facts) == 1


@pytest.mark.asyncio
async def test_cost_facts_batched_falls_back_when_span_too_wide() -> None:
    # Span beyond the guard → per-account streaming (memory safety on a
    # giant single-shot backfill); batch cache stays empty.
    driver = _bq_driver_v2()
    scope_a = CloudScopeRef(scope_type="aws_account", external_id=_ACC_A)
    scope_b = CloudScopeRef(scope_type="aws_account", external_id=_ACC_B)
    driver._cost_facts_all_account_scopes = [scope_a, scope_b]
    span = driver._BQ_COST_FACTS_BATCH_MAX_SPAN_DAYS + 10
    start = date(2026, 1, 1)
    end = start + timedelta(days=span)
    sqls: list[str] = []

    def _q(sql, **kw):
        sqls.append(sql)
        job = MagicMock()
        job.result.return_value = iter([])
        return job

    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    client.query.side_effect = _q
    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            start, end, scope=scope_a, bq_billing=_BQ_BILLING_CUR_V2,
        )
    assert sqls, "expected per-account queries"
    assert all("line_item_usage_account_id = @scope_acct" in s for s in sqls)
    assert driver._bq_cost_facts_batch_cache == {}


@pytest.mark.asyncio
async def test_cost_facts_per_resource_grain_not_batched() -> None:
    # per_resource grain has unbounded row cardinality (one fact per
    # resource), so the batched path is gated off to keep the per-run
    # cache memory bounded — falls back to per-account streaming.
    driver = _bq_driver_v2()
    scope_a = CloudScopeRef(scope_type="aws_account", external_id=_ACC_A)
    driver._cost_facts_all_account_scopes = [
        scope_a, CloudScopeRef(scope_type="aws_account", external_id=_ACC_B),
    ]
    client = _stub_bq_client_with_rows([_v2_row(_ACC_A)])
    with patch.object(driver, "_bq_client", return_value=client):
        await driver.list_cost_facts(
            date(2026, 5, 14), date(2026, 5, 14), scope=scope_a,
            bq_billing={**_BQ_BILLING_CUR_V2, "fact_grain": "per_resource"},
        )
    sql = client.query.call_args.args[0]
    assert "line_item_usage_account_id = @scope_acct" in sql
    assert driver._bq_cost_facts_batch_cache == {}


# ── Batched all-accounts billing_daily path (per_month, scan-once) ──
#
# Same scan-once optimisation as cost_facts, applied to the daily
# billing sync (which runs up to 3 per-account queries — totals /
# service / owner — on the same DAY-partitioned, non-account-clustered
# CUR month-tables). When the caller hands the driver the target set via
# `_billing_daily_all_account_scopes`, the first scope scans all
# accounts once and the rest are served from a per-run cache; each
# CloudDailyCost already self-identifies its account, and the caller
# still upserts per loop-scope so the transparent cache keeps behaviour
# identical. per_account layout (the `_bq_driver()` billing tests above)
# is never eligible — confirming backward-compat.

def _bd_totals(account_id: str, cost: float) -> SimpleNamespace:
    return SimpleNamespace(
        day=date(2026, 5, 14), account_id=account_id,
        currency="USD", native_cost=cost,
    )


def _bd_service(account_id: str, service: str, cost: float) -> SimpleNamespace:
    return SimpleNamespace(
        day=date(2026, 5, 14), account_id=account_id,
        currency="USD", service=service, native_cost=cost,
    )


@pytest.mark.asyncio
async def test_billing_daily_batched_scans_once_serves_all_scopes() -> None:
    driver = _bq_driver_v2()
    scope_a = CloudScopeRef(scope_type="aws_account", external_id=_ACC_A)
    scope_b = CloudScopeRef(scope_type="aws_account", external_id=_ACC_B)
    driver._billing_daily_all_account_scopes = [scope_a, scope_b]

    totals_job, svc_job, owner_job = MagicMock(), MagicMock(), MagicMock()
    totals_job.result.return_value = iter([
        _bd_totals(_ACC_A, 100.0), _bd_totals(_ACC_B, 200.0),
    ])
    svc_job.result.return_value = iter([
        _bd_service(_ACC_A, "Amazon EC2", 100.0),
        _bd_service(_ACC_B, "Amazon S3", 200.0),
    ])
    owner_job.result.return_value = iter([])
    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    client.query.side_effect = [totals_job, svc_job, owner_job]

    with patch.object(driver, "_bq_client", return_value=client):
        rows_a = await driver.list_billing_daily(
            date(2026, 5, 14), date(2026, 5, 14), scope=scope_a,
            include_owner_breakdown=True, bq_billing=_BQ_BILLING_CUR_V2,
        )
        rows_b = await driver.list_billing_daily(
            date(2026, 5, 14), date(2026, 5, 14), scope=scope_b,
            include_owner_breakdown=True, bq_billing=_BQ_BILLING_CUR_V2,
        )
    # 3 queries total (totals/service/owner) — NOT 6; second scope cached.
    assert client.query.call_count == 3
    # Account filter is gone from the totals query (one all-accounts scan).
    totals_sql = client.query.call_args_list[0].args[0]
    assert "line_item_usage_account_id = @scope_acct" not in totals_sql
    # Each scope gets exactly its own account's day-row.
    assert [r.scope.external_id for r in rows_a] == [_ACC_A]
    assert [r.scope.external_id for r in rows_b] == [_ACC_B]
    assert rows_a[0].total_cost_usd == pytest.approx(100.0)
    assert rows_a[0].cost_by_service == {"Amazon EC2": 100.0}
    assert rows_b[0].total_cost_usd == pytest.approx(200.0)
    assert rows_b[0].cost_by_service == {"Amazon S3": 200.0}


@pytest.mark.asyncio
async def test_billing_daily_batched_drops_unregistered_account() -> None:
    driver = _bq_driver_v2()
    scope_a = CloudScopeRef(scope_type="aws_account", external_id=_ACC_A)
    driver._billing_daily_all_account_scopes = [scope_a]

    totals_job, svc_job, owner_job = MagicMock(), MagicMock(), MagicMock()
    totals_job.result.return_value = iter([
        _bd_totals(_ACC_A, 100.0), _bd_totals(_ACC_UNREG, 999.0),
    ])
    svc_job.result.return_value = iter([
        _bd_service(_ACC_A, "Amazon EC2", 100.0),
        _bd_service(_ACC_UNREG, "Amazon EC2", 999.0),
    ])
    owner_job.result.return_value = iter([])
    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    client.query.side_effect = [totals_job, svc_job, owner_job]

    with patch.object(driver, "_bq_client", return_value=client):
        rows_a = await driver.list_billing_daily(
            date(2026, 5, 14), date(2026, 5, 14), scope=scope_a,
            include_owner_breakdown=True, bq_billing=_BQ_BILLING_CUR_V2,
        )
        # Unregistered account is not a target scope → cache has no slice.
        rows_unreg = await driver.list_billing_daily(
            date(2026, 5, 14), date(2026, 5, 14),
            scope=CloudScopeRef(scope_type="aws_account", external_id=_ACC_UNREG),
            include_owner_breakdown=True, bq_billing=_BQ_BILLING_CUR_V2,
        )
    assert [r.scope.external_id for r in rows_a] == [_ACC_A]
    assert rows_a[0].total_cost_usd == pytest.approx(100.0)
    assert rows_unreg == []


@pytest.mark.asyncio
async def test_billing_daily_per_month_without_target_set_keeps_account_filter() -> None:
    # No target set → per-account path: totals SQL keeps @scope_acct and
    # the batch cache is never populated (backward-compat).
    driver = _bq_driver_v2()
    scope_a = CloudScopeRef(scope_type="aws_account", external_id=_ACC_A)

    totals_job, svc_job, owner_job = MagicMock(), MagicMock(), MagicMock()
    totals_job.result.return_value = iter([_bd_totals(_ACC_A, 100.0)])
    svc_job.result.return_value = iter([_bd_service(_ACC_A, "Amazon EC2", 100.0)])
    owner_job.result.return_value = iter([])
    client = MagicMock()
    client.get_table.return_value = _stub_table(())
    client.query.side_effect = [totals_job, svc_job, owner_job]

    with patch.object(driver, "_bq_client", return_value=client):
        rows = await driver.list_billing_daily(
            date(2026, 5, 14), date(2026, 5, 14), scope=scope_a,
            include_owner_breakdown=True, bq_billing=_BQ_BILLING_CUR_V2,
        )
    totals_sql = client.query.call_args_list[0].args[0]
    assert "line_item_usage_account_id = @scope_acct" in totals_sql
    assert driver._bq_billing_daily_batch_cache == {}
    assert len(rows) == 1