"""Tests for the production-grade FinOps surface (migration 139).

Pins down the contract every analytics surface depends on:

  * ``compute_fact_key`` is stable and collision-resistant on
    NULL-bearing natural keys (date + integration + scope +
    resource_id + usage_type).

  * ``_match_rule`` walks the rule ladder by priority + AND-s every
    non-NULL match condition + returns the first match's assignment.
    Wildcard rules (all match-* NULL) match anything.

  * ``ingest_cost_facts`` upserts idempotently — re-running with the
    same input window converges and doesn't duplicate.

  * ``breakdown`` / ``timeseries`` zero-fill missing days and bucket
    everything past top-N into "other".

  * ``forecast_month_end`` blends linear and trailing-mean projections
    correctly when MTD == 0 (early-month edge case).

  * ``detect_anomalies`` returns nothing when amount stays within
    median + z·IQR; flags only when above threshold AND impact
    >= ``min_impact_usd``.
"""
from __future__ import annotations

import uuid
from datetime import UTC, date, datetime, timedelta
from unittest.mock import patch

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

from app.domains.cmdb.cost_service import (
    _match_rule,
    breakdown,
    compute_fact_key,
    create_allocation_rule,
    delete_allocation_rule,
    detect_anomalies,
    forecast_month_end,
    ingest_cost_facts,
    recompute_allocation,
    timeseries,
    top_resources,
    update_allocation_rule,
)
from app.domains.cmdb.models import (
    CmdbCloudScope,
    CmdbCostAllocationRule,
    CmdbCostFact,
)
from app.integrations.driver_base import CloudScopeRef, CostFact
from app.integrations.models import Integration

# ─── Pure helpers (no DB) ───────────────────────────────────────────


def test_compute_fact_key_stable_and_disambiguates_nulls():
    """Same inputs → same hash; differing inputs → different hash;
    NULL components don't collapse rows incorrectly."""
    base = dict(
        integration_id="int-1", fact_date=date(2026, 5, 1),
        scope_external_id="sub-azure-1",
        resource_id="res-1", usage_type="vcpu",
    )
    h1 = compute_fact_key(**base)
    h2 = compute_fact_key(**base)
    assert h1 == h2

    # Different resource_id → different hash.
    h3 = compute_fact_key(**{**base, "resource_id": "res-2"})
    assert h1 != h3

    # Two different "no resource" rows on the same day with different
    # usage_type stay distinct (the NULL collapse failure mode).
    h_no_a = compute_fact_key(
        integration_id="int-1", fact_date=date(2026, 5, 1),
        scope_external_id=None, resource_id=None, usage_type="overhead-A",
    )
    h_no_b = compute_fact_key(
        integration_id="int-1", fact_date=date(2026, 5, 1),
        scope_external_id=None, resource_id=None, usage_type="overhead-B",
    )
    assert h_no_a != h_no_b


def test_compute_fact_key_stable_across_scopes_sync_catchup():
    """Pre-fix bug: ``compute_fact_key`` hashed the Helix DB UUID
    (``scope_db_id``), which is NULL until ``scopes_sync`` materialises
    the row. A fact ingested before scopes_sync caught up would hash
    with NULL and re-ingested after the catch-up would hash with the
    new UUID — different keys → duplicate row instead of an
    idempotent upsert.

    Post-fix, the natural key uses the cloud-side immutable
    ``external_id`` (Azure subId / GCP projectId / etc.), so the
    same logical fact converges to the same hash regardless of
    whether scopes_sync has run yet.

    This test pins the fix.
    """
    common = dict(
        integration_id="int-1", fact_date=date(2026, 5, 1),
        resource_id="res-A", usage_type="vcpu",
    )
    # Pre-catchup: cloud-side id is what the driver always knows.
    h_pre = compute_fact_key(
        scope_external_id="azure-sub-7bc8ab17", **common,
    )
    # Post-catchup: SAME cloud-side id (the Helix DB UUID is
    # different but the natural key doesn't see it).
    h_post = compute_fact_key(
        scope_external_id="azure-sub-7bc8ab17", **common,
    )
    assert h_pre == h_post


def _make_rule(
    *,
    name: str = "r",
    priority: int = 100,
    enabled: bool = True,
    match_provider: str | None = None,
    match_resource_group: str | None = None,
    match_service: str | None = None,
    match_tag_key: str | None = None,
    match_tag_value: str | None = None,
    match_integration_ids: list[str] | None = None,
    assign_person_id: str | None = None,
    assign_department: str | None = None,
    assign_cost_center: str | None = None,
) -> CmdbCostAllocationRule:
    return CmdbCostAllocationRule(
        id=str(uuid.uuid4()),
        name=name,
        priority=priority,
        enabled=enabled,
        match_provider=match_provider,
        match_resource_group=match_resource_group,
        match_service=match_service,
        match_tag_key=match_tag_key,
        match_tag_value=match_tag_value,
        match_integration_ids=match_integration_ids,
        assign_person_id=assign_person_id,
        assign_department=assign_department,
        assign_cost_center=assign_cost_center,
    )


def test_match_rule_returns_empty_when_no_rules():
    assert _match_rule(
        [],
        provider="aws",
        integration_id="i-1",
        resource_group="rg-prod",
        service="EC2",
        tags={},
    ) == {}


def test_match_rule_first_match_wins_by_priority_order():
    """Caller is responsible for ordering rules ascending by priority;
    _match_rule walks the list in order."""
    r_lower_priority = _make_rule(
        name="catchall", priority=999,
        assign_cost_center="DEFAULT",
    )
    r_higher = _make_rule(
        name="ec2-only", priority=10,
        match_service="EC2",
        assign_cost_center="ENG-INFRA",
    )
    out = _match_rule(
        [r_higher, r_lower_priority],  # already sorted
        provider="aws",
        integration_id="i-1",
        resource_group=None,
        service="EC2",
        tags={},
    )
    assert out["cost_center"] == "ENG-INFRA"


def test_match_rule_all_conditions_anded():
    """Provider + RG + service + tag are AND-ed; partial match misses."""
    rule = _make_rule(
        match_provider="azure",
        match_resource_group="rg-data",
        match_service="Storage",
        match_tag_key="Project",
        match_tag_value="tidb",
        assign_department="Data",
    )
    # Full match.
    assert _match_rule(
        [rule], provider="azure", integration_id="i-1",
        resource_group="rg-data", service="Storage",
        tags={"Project": "tidb"},
    ) == {
        "rule_id": str(rule.id),
        "person_id": None,
        "department": "Data",
        "cost_center": None,
    }
    # Wrong provider → miss.
    assert _match_rule(
        [rule], provider="aws", integration_id="i-1",
        resource_group="rg-data", service="Storage",
        tags={"Project": "tidb"},
    ) == {}
    # Wrong tag value → miss.
    assert _match_rule(
        [rule], provider="azure", integration_id="i-1",
        resource_group="rg-data", service="Storage",
        tags={"Project": "tikv"},
    ) == {}
    # Tag key absent → miss.
    assert _match_rule(
        [rule], provider="azure", integration_id="i-1",
        resource_group="rg-data", service="Storage", tags={},
    ) == {}


def test_match_rule_integration_id_allowlist():
    """match_integration_ids is treated as JSON list; only listed
    integrations match."""
    rule = _make_rule(
        match_integration_ids=["i-A", "i-C"],
        assign_cost_center="X",
    )
    assert _match_rule(
        [rule], provider="aws", integration_id="i-A",
        resource_group=None, service="EC2", tags={},
    )["cost_center"] == "X"
    assert _match_rule(
        [rule], provider="aws", integration_id="i-B",
        resource_group=None, service="EC2", tags={},
    ) == {}


def test_match_rule_tag_key_present_value_wildcard():
    """Setting tag_key without tag_value matches as long as the key
    is present (any value)."""
    rule = _make_rule(
        match_tag_key="Owner", assign_department="Owned",
    )
    assert _match_rule(
        [rule], provider="aws", integration_id="i-1",
        resource_group=None, service="EC2",
        tags={"Owner": "alice"},
    )["department"] == "Owned"
    assert _match_rule(
        [rule], provider="aws", integration_id="i-1",
        resource_group=None, service="EC2",
        tags={"Owner": "bob"},
    )["department"] == "Owned"
    assert _match_rule(
        [rule], provider="aws", integration_id="i-1",
        resource_group=None, service="EC2",
        tags={},
    ) == {}


def test_match_rule_resource_group_case_insensitive():
    """Round-19 fix: Azure RG names are case-insensitive on the cloud
    side (``MyResourceGroup`` and ``myresourcegroup`` are the same RG),
    but Cost Management may return either casing depending on tenant /
    API version. Pre-fix an admin who copied an RG name out of the
    Azure portal as ``MyRG`` would write a rule that silently never
    matched the lowercased fact emitted by Cost Management. The
    ``match_provider`` path already lowercases on both sides; extend
    the same defence in depth to RG / service / tag values."""
    rule = _make_rule(
        match_resource_group="MyResourceGroup",
        assign_cost_center="DATA",
    )
    # Mixed-case fact still matches the rule.
    out = _match_rule(
        [rule], provider="azure", integration_id="i-1",
        resource_group="myresourcegroup", service="VirtualMachines",
        tags={},
    )
    assert out["cost_center"] == "DATA"
    # All-uppercase variant also matches.
    out = _match_rule(
        [rule], provider="azure", integration_id="i-1",
        resource_group="MYRESOURCEGROUP", service="VirtualMachines",
        tags={},
    )
    assert out["cost_center"] == "DATA"


def test_match_rule_service_case_insensitive():
    """Cloud APIs return service names with inconsistent casing — pin
    case-insensitive comparison so an admin's ``Microsoft.Compute``
    rule matches Cost Management's ``microsoft.compute`` fact tag."""
    rule = _make_rule(
        match_service="Microsoft.Compute",
        assign_department="Infra",
    )
    out = _match_rule(
        [rule], provider="azure", integration_id="i-1",
        resource_group=None, service="microsoft.compute",
        tags={},
    )
    assert out["department"] == "Infra"


def test_match_rule_tag_value_case_insensitive():
    """Tag VALUES (not keys) are matched case-insensitively to honour
    operator intuition (admins type ``team-platform`` and
    ``Team-Platform`` interchangeably). Tag KEYS remain case-sensitive
    because clouds enforce case-sensitivity at the API level."""
    rule = _make_rule(
        match_tag_key="Owner", match_tag_value="team-platform",
        assign_department="Platform",
    )
    out = _match_rule(
        [rule], provider="aws", integration_id="i-1",
        resource_group=None, service="EC2",
        tags={"Owner": "Team-Platform"},  # casing differs
    )
    assert out["department"] == "Platform"


# ─── Forecast / anomaly math (no DB) ────────────────────────────────


@pytest.mark.asyncio
async def test_forecast_zero_mtd_returns_zeros(db_session: AsyncSession):
    """Edge case: no facts for the current month → all forecasts zero
    (no extrapolation from nothing)."""
    out = await forecast_month_end(
        db_session, today=date(2026, 5, 5),
    )
    assert out["mtd_usd"] == 0.0
    assert out["forecast_linear_usd"] == 0.0
    assert out["forecast_trailing_usd"] == 0.0
    assert out["forecast_blend_usd"] == 0.0
    assert out["days_in_month"] == 31
    # Round-22 fix anchors MTD on ``last_complete_day = today - 1`` so
    # today's partial-day row (ingested at 04:00 UTC the next morning)
    # doesn't drag the linear projection down. For today=2026-05-05
    # last_complete_day=2026-05-04 → days_elapsed = May 1..May 4 = 4.
    assert out["days_elapsed"] == 4


# ─── DB-bound tests (use the existing pytest-asyncio db_session) ───


@pytest_asyncio.fixture
async def cost_integration(db_session: AsyncSession) -> Integration:
    """A minimal cloud integration row to FK fact rows against."""
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="azure_cloud",
        name="Test Azure",
        slug="test-azure",
        enabled=True,
        config_public={},
    )
    db_session.add(integ)
    await db_session.commit()
    await db_session.refresh(integ)
    return integ


def _fact(
    *,
    integration_id: str,
    fact_date: date,
    resource_id: str = "res-A",
    service: str = "Compute",
    amount_usd: float = 10.0,
    resource_group: str | None = "rg-prod",
    usage_type: str | None = "vcpu",
    region: str | None = "eastus",
    tags: dict[str, str] | None = None,
) -> CostFact:
    return CostFact(
        date=fact_date,
        integration_id=integration_id,
        service=service,
        amount_usd=amount_usd,
        amount_native=amount_usd,
        currency="USD",
        scope=None,
        resource_group=resource_group,
        resource_id=resource_id,
        resource_name=resource_id,
        resource_type="vm",
        region=region,
        usage_type=usage_type,
        tags=tags or {},
    )


@pytest.mark.asyncio
async def test_ingest_idempotent_double_insert_keeps_one_row(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Same fact ingested twice — only one row remains, amount equals
    the latest value (ON DUPLICATE KEY UPDATE)."""
    integ_id = str(cost_integration.id)
    facts = [_fact(
        integration_id=integ_id, fact_date=date(2026, 5, 1),
        amount_usd=10.0,
    )]
    r1 = await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )
    assert r1["upserted"] == 1

    # Same key, different amount — second ingest must overwrite, not
    # double-insert.
    facts2 = [_fact(
        integration_id=integ_id, fact_date=date(2026, 5, 1),
        amount_usd=15.0,
    )]
    r2 = await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts2,
    )
    assert r2["upserted"] == 1

    rows = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.integration_id == integ_id,
        ),
    )).scalars().all()
    assert len(rows) == 1
    assert float(rows[0].amount_usd) == 15.0


@pytest.mark.asyncio
async def test_ingest_collapses_colliding_fact_keys_by_sum(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Facts sharing a fact_key (date, scope, resource_id, usage_type)
    but differing by product/service are SUMMED into one row, not
    dropped by the upsert.

    Regression for the cost_facts over-report (2026-06-17): AWS CUR
    EdpDiscount rows synthesise one bucket resource_id across 67
    product_codes, and Aliyun BSS credit days emit many adjustment
    lines on one resource_id — all colliding on fact_key. Pre-fix the
    ON DUPLICATE KEY UPDATE kept ONE and silently dropped the rest,
    and because the dropped lines are negative discounts/credits the
    persisted total over-reported spend. Pre-aggregation must net them.
    """
    integ_id = str(cost_integration.id)
    # Same (date, scope=None, resource_id, usage_type); three negative
    # discount lines that must net to -35, plus one positive usage line
    # on a DIFFERENT resource that must survive untouched.
    facts = [
        _fact(
            integration_id=integ_id, fact_date=date(2026, 5, 1),
            resource_id="aws-account:111:EDP Discount",
            usage_type="BoxUsage", service="EDP Discount",
            amount_usd=-20.0,
        ),
        _fact(
            integration_id=integ_id, fact_date=date(2026, 5, 1),
            resource_id="aws-account:111:EDP Discount",
            usage_type="BoxUsage", service="EDP Discount",
            amount_usd=-10.0,
        ),
        _fact(
            integration_id=integ_id, fact_date=date(2026, 5, 1),
            resource_id="aws-account:111:EDP Discount",
            usage_type="BoxUsage", service="EDP Discount",
            amount_usd=-5.0,
        ),
        _fact(
            integration_id=integ_id, fact_date=date(2026, 5, 1),
            resource_id="i-realresource", usage_type="BoxUsage",
            service="Compute", amount_usd=100.0,
        ),
    ]
    r = await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )
    # 3 discount lines collapse to 1 → 2 eliminated; 2 rows persisted.
    assert r["collapsed"] == 2
    assert r["upserted"] == 2

    rows = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.integration_id == integ_id,
        ),
    )).scalars().all()
    by_rid = {row.resource_id: row for row in rows}
    assert len(rows) == 2
    # Negatives netted, not dropped (pre-fix this was -20 / -10 / -5).
    assert float(by_rid["aws-account:111:EDP Discount"].amount_usd) == (
        pytest.approx(-35.0)
    )
    assert float(by_rid["i-realresource"].amount_usd) == pytest.approx(100.0)
    # Total reconciles: -35 + 100 = 65 (vs the pre-fix over-report where
    # only one discount line survived, e.g. -20 + 100 = 80).
    assert sum(float(r2.amount_usd) for r2 in rows) == pytest.approx(65.0)


@pytest.mark.asyncio
async def test_ingest_collision_metadata_from_largest_member(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """When colliding facts carry different metadata, the merged row
    takes the metadata of the largest-magnitude member (deterministic
    regardless of input order)."""
    integ_id = str(cost_integration.id)
    facts = [
        _fact(
            integration_id=integ_id, fact_date=date(2026, 5, 2),
            resource_id="shared", usage_type="u1",
            service="Minor", amount_usd=-3.0,
        ),
        _fact(
            integration_id=integ_id, fact_date=date(2026, 5, 2),
            resource_id="shared", usage_type="u1",
            service="Dominant", amount_usd=-99.0,
        ),
    ]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )
    rows = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.integration_id == integ_id,
            CmdbCostFact.resource_id == "shared",
        ),
    )).scalars().all()
    assert len(rows) == 1
    assert rows[0].service == "Dominant"
    assert float(rows[0].amount_usd) == pytest.approx(-102.0)


def test_preaggregate_gross_and_usage_none_semantics():
    """Pure unit: gross / usage_quantity sum across members that
    exposed a value and stay None only when NO member did (preserving
    the driver's 'None = provider didn't expose this' contract)."""
    from app.domains.cmdb.cost_service import (
        _preaggregate_facts_by_fact_key,
    )

    def mk(amount, gross, qty):
        return CostFact(
            date=date(2026, 5, 1), integration_id="i",
            service="s", amount_usd=amount, amount_native=amount,
            currency="USD", scope=None, resource_id="r",
            usage_type="u", amount_usd_gross=gross,
            amount_native_gross=gross, usage_quantity=qty,
        )

    # One member has gross+qty, the other has neither → summed values.
    # (Return entries are ``(fact, bucket_disambiguator)`` pairs since
    # the cross-bucket collision fix; same-bucket merges carry None.)
    out, eliminated = _preaggregate_facts_by_fact_key([
        mk(-10.0, 5.0, 2.0), mk(-4.0, None, None),
    ])
    assert eliminated == 1
    assert len(out) == 1
    merged, disambig = out[0]
    assert disambig is None
    assert merged.amount_usd == pytest.approx(-14.0)
    assert merged.amount_usd_gross == pytest.approx(5.0)
    assert merged.usage_quantity == pytest.approx(2.0)

    # All members None → stays None (not 0.0).
    out2, _ = _preaggregate_facts_by_fact_key([
        mk(-10.0, None, None), mk(-4.0, None, None),
    ])
    assert out2[0][0].amount_usd_gross is None
    assert out2[0][0].usage_quantity is None

    # Mixed-currency facts sharing one fact_key must fail loudly (the
    # grouping key omits currency; summing CNY+USD into one amount_native
    # would corrupt native analytics). Defends the invariant in-function.
    with pytest.raises(RuntimeError, match="multiple currencies"):
        _preaggregate_facts_by_fact_key([
            CostFact(
                date=date(2026, 5, 1), integration_id="i", service="s",
                amount_usd=10.0, amount_native=10.0, currency="USD",
                scope=None, resource_id="r", usage_type="u",
            ),
            CostFact(
                date=date(2026, 5, 1), integration_id="i", service="s",
                amount_usd=0.0, amount_native=70.0, currency="CNY",
                scope=None, resource_id="r", usage_type="u",
            ),
        ])

    # Distinct fact_keys → no collapse (no-op for well-behaved drivers).
    distinct = [
        CostFact(
            date=date(2026, 5, 1), integration_id="i", service="s",
            amount_usd=1.0, amount_native=1.0, currency="USD",
            scope=None, resource_id=f"r{n}", usage_type="u",
        )
        for n in range(3)
    ]
    out3, elim3 = _preaggregate_facts_by_fact_key(distinct)
    assert elim3 == 0
    assert len(out3) == 3


@pytest.mark.asyncio
async def test_breakdown_and_unallocated(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Breakdown sums + computes unallocated KPI — facts with NULL
    allocation_rule_id surface as unallocated even when some other
    facts ARE allocated."""
    integ_id = str(cost_integration.id)
    facts = [
        _fact(
            integration_id=integ_id, fact_date=date(2026, 5, 1),
            service="Compute", amount_usd=100.0, resource_id="r-A",
        ),
        _fact(
            integration_id=integ_id, fact_date=date(2026, 5, 1),
            service="Storage", amount_usd=40.0, resource_id="r-B",
        ),
        _fact(
            integration_id=integ_id, fact_date=date(2026, 5, 2),
            service="Compute", amount_usd=60.0, resource_id="r-A",
        ),
    ]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )

    out = await breakdown(
        db_session,
        start=date(2026, 5, 1), end=date(2026, 5, 2),
        group_by="service",
        integration_id=integ_id,
        top=10,
    )
    assert out["total_usd"] == pytest.approx(200.0)
    by_label = {g["label"]: g["cost_usd"] for g in out["groups"]}
    assert by_label["Compute"] == pytest.approx(160.0)
    assert by_label["Storage"] == pytest.approx(40.0)
    # No allocation rules in DB → everything counts as unallocated.
    assert out["unallocated_usd"] == pytest.approx(200.0)
    assert out["unallocated_pct"] == pytest.approx(100.0)


@pytest.mark.asyncio
async def test_breakdown_group_by_provider_joins_integration(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """``group_by='provider'`` groups on the joined ``integrations.provider``
    (provider is NOT a cost_facts column). Regression for the drilldown's
    'by cloud provider' dimension."""
    integ_id = str(cost_integration.id)  # provider='azure_cloud'
    await ingest_cost_facts(db_session, integration_id=integ_id, facts=[
        _fact(integration_id=integ_id, fact_date=date(2026, 5, 1),
              amount_usd=70.0, resource_id="r-1"),
        _fact(integration_id=integ_id, fact_date=date(2026, 5, 2),
              amount_usd=30.0, resource_id="r-2"),
    ])
    out = await breakdown(
        db_session, start=date(2026, 5, 1), end=date(2026, 5, 2),
        group_by="provider", integration_id=integ_id, top=10,
    )
    assert out["total_usd"] == pytest.approx(100.0)
    by_label = {g["label"]: g["cost_usd"] for g in out["groups"]}
    assert by_label == {"azure_cloud": pytest.approx(100.0)}


@pytest.mark.asyncio
async def test_breakdown_department_filter_exact_and_excludes_unallocated(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """``department=`` narrows breakdown to facts attributable to that
    department's people, and EXCLUDES NULL-owner (unallocated) facts —
    the filtered total is a true subset of the fleet total, not a fold-in
    of unattributed spend. Regression for the Cloud-Overview billing tab
    ignoring the header department filter.
    """
    integ_id = str(cost_integration.id)
    facts = [
        _fact(integration_id=integ_id, fact_date=date(2026, 5, 1),
              amount_usd=100.0, resource_id="r-eng"),
        _fact(integration_id=integ_id, fact_date=date(2026, 5, 1),
              amount_usd=40.0, resource_id="r-sales"),
        _fact(integration_id=integ_id, fact_date=date(2026, 5, 1),
              amount_usd=7.0, resource_id="r-unalloc"),
    ]
    await ingest_cost_facts(db_session, integration_id=integ_id, facts=facts)
    # ingest leaves cost_owner_department NULL (no rules / no scope owner).
    # Stamp departments directly so we exercise the read-path filter.
    from sqlalchemy import update
    await db_session.execute(
        update(CmdbCostFact)
        .where(CmdbCostFact.resource_id == "r-eng")
        .values(cost_owner_department="Engineering"),
    )
    await db_session.execute(
        update(CmdbCostFact)
        .where(CmdbCostFact.resource_id == "r-sales")
        .values(cost_owner_department="Sales"),
    )
    await db_session.commit()

    # No department → fleet total includes everything (140 + 7 unalloc).
    full = await breakdown(
        db_session, start=date(2026, 5, 1), end=date(2026, 5, 1),
        group_by="integration", integration_id=integ_id, top=10,
    )
    assert full["total_usd"] == pytest.approx(147.0)

    # department="Engineering" → only the 100 Engineering fact; the Sales
    # fact and the NULL-department unallocated fact are both excluded.
    eng = await breakdown(
        db_session, start=date(2026, 5, 1), end=date(2026, 5, 1),
        group_by="integration", integration_id=integ_id,
        department="Engineering", top=10,
    )
    assert eng["total_usd"] == pytest.approx(100.0)


@pytest.mark.asyncio
async def test_breakdown_department_filter_rolls_up_subtree(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Picking a PARENT department rolls up its whole subtree: a fact
    attributed to a child department is included when filtering on the
    parent name (matches the org-scope contract used by every other
    department filter)."""
    from app.identity.person_models import IdentityOrgNode

    # Parent "BO" with a child leaf "BO-Ops" (path joined by " / ").
    db_session.add(IdentityOrgNode(
        source_system="test", org_node_source_id="bo",
        org_node_type="department", org_node_name="BO", org_path="BO",
    ))
    db_session.add(IdentityOrgNode(
        source_system="test", org_node_source_id="bo-ops",
        org_node_type="department", org_node_name="BO-Ops",
        org_path="BO / BO-Ops",
    ))
    await db_session.commit()

    integ_id = str(cost_integration.id)
    facts = [
        _fact(integration_id=integ_id, fact_date=date(2026, 5, 1),
              amount_usd=55.0, resource_id="r-child"),
        _fact(integration_id=integ_id, fact_date=date(2026, 5, 1),
              amount_usd=999.0, resource_id="r-other"),
    ]
    await ingest_cost_facts(db_session, integration_id=integ_id, facts=facts)
    from sqlalchemy import update
    await db_session.execute(
        update(CmdbCostFact)
        .where(CmdbCostFact.resource_id == "r-child")
        .values(cost_owner_department="BO-Ops"),
    )
    await db_session.execute(
        update(CmdbCostFact)
        .where(CmdbCostFact.resource_id == "r-other")
        .values(cost_owner_department="Engineering"),
    )
    await db_session.commit()

    out = await breakdown(
        db_session, start=date(2026, 5, 1), end=date(2026, 5, 1),
        group_by="integration", integration_id=integ_id,
        department="BO", top=10,
    )
    # Only the child-department fact rolls up under "BO"; "Engineering"
    # is a different subtree and is excluded.
    assert out["total_usd"] == pytest.approx(55.0)


@pytest.mark.asyncio
async def test_timeseries_zero_fills_missing_days(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Timeseries returns one entry per day in [start,end] even when
    some days have no facts."""
    integ_id = str(cost_integration.id)
    facts = [
        _fact(
            integration_id=integ_id, fact_date=date(2026, 5, 1),
            service="Compute", amount_usd=10.0,
        ),
        # Skip 2026-05-02 entirely.
        _fact(
            integration_id=integ_id, fact_date=date(2026, 5, 3),
            service="Compute", amount_usd=20.0,
        ),
    ]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )

    out = await timeseries(
        db_session,
        start=date(2026, 5, 1), end=date(2026, 5, 3),
        group_by="service",
        integration_id=integ_id,
        top=5,
    )
    assert len(out["dates"]) == 3
    assert out["dates"] == [
        "2026-05-01", "2026-05-02", "2026-05-03",
    ]
    # The series is zero-filled on the missing day.
    compute_series = next(
        s for s in out["series"] if s["label"] == "Compute"
    )
    assert compute_series["points"] == [10.0, 0.0, 20.0]


@pytest.mark.asyncio
async def test_top_resources_with_spark(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Top resources sorted desc by cost; spark series is daily."""
    integ_id = str(cost_integration.id)
    today = date.today() - timedelta(days=1)
    facts = []
    # Resource A bigger spend.
    for i in range(5):
        facts.append(_fact(
            integration_id=integ_id,
            fact_date=today - timedelta(days=i),
            resource_id="r-A",
            amount_usd=100.0,
        ))
    # Resource B smaller spend.
    for i in range(5):
        facts.append(_fact(
            integration_id=integ_id,
            fact_date=today - timedelta(days=i),
            resource_id="r-B",
            amount_usd=10.0,
        ))
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )

    rows = await top_resources(
        db_session,
        start=today - timedelta(days=4),
        end=today,
        integration_id=integ_id,
        limit=10,
    )
    assert rows[0]["resource_id"] == "r-A"
    assert rows[0]["total_usd"] == pytest.approx(500.0)
    assert rows[1]["resource_id"] == "r-B"
    assert rows[1]["total_usd"] == pytest.approx(50.0)
    # Spark length covers the trailing 14 days (clipped to range).
    assert len(rows[0]["spark"]) >= 5
    # Every row must carry a ``tags`` dict (empty when no tags ingested);
    # the dashboard reads this field unconditionally.
    assert "tags" in rows[0]
    assert "tags" in rows[1]
    assert rows[0]["tags"] == {}
    assert rows[1]["tags"] == {}


@pytest.mark.asyncio
async def test_top_resources_surfaces_latest_non_empty_tags(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """``top_resources`` returns the most recent non-empty ``tags``
    dict per resource — older non-empty entries don't override later
    ones, and empty/missing tag rows don't blank out a preceding
    populated entry."""
    integ_id = str(cost_integration.id)
    today = date.today() - timedelta(days=1)
    facts = [
        # Day -3: r-A has Owner=alice
        _fact(
            integration_id=integ_id,
            fact_date=today - timedelta(days=3),
            resource_id="r-A",
            amount_usd=10.0,
            tags={"Owner": "alice"},
        ),
        # Day -2: r-A's tags updated to Owner=bob + Env=prod
        _fact(
            integration_id=integ_id,
            fact_date=today - timedelta(days=2),
            resource_id="r-A",
            amount_usd=10.0,
            usage_type="vcpu-2",
            tags={"Owner": "bob", "Env": "prod"},
        ),
        # Day -1: r-A's cost row has no tags (defensive — older
        # populated entry should still win)
        _fact(
            integration_id=integ_id,
            fact_date=today - timedelta(days=1),
            resource_id="r-A",
            amount_usd=10.0,
            usage_type="vcpu-3",
            tags={},
        ),
        # r-B never had tags at all.
        _fact(
            integration_id=integ_id,
            fact_date=today,
            resource_id="r-B",
            amount_usd=5.0,
            tags={},
        ),
    ]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )

    rows = await top_resources(
        db_session,
        start=today - timedelta(days=4),
        end=today,
        integration_id=integ_id,
        limit=10,
    )
    by_rid = {r["resource_id"]: r for r in rows}
    # Latest non-empty tags win: Owner=bob, Env=prod (day -2 entry).
    assert by_rid["r-A"]["tags"] == {"Owner": "bob", "Env": "prod"}
    # Never-tagged resource stays empty.
    assert by_rid["r-B"]["tags"] == {}


@pytest.mark.asyncio
async def test_top_resources_tag_filter_narrows_results(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """``tag_key`` + ``tag_value`` narrows results to rows whose
    latest tags dict matches; missing or partial filter raises
    ``ValueError`` (the router maps to 400)."""
    integ_id = str(cost_integration.id)
    today = date.today() - timedelta(days=1)
    facts = [
        _fact(
            integration_id=integ_id, fact_date=today,
            resource_id="r-alice", amount_usd=20.0,
            tags={"Owner": "alice"},
        ),
        _fact(
            integration_id=integ_id, fact_date=today,
            resource_id="r-bob", amount_usd=30.0,
            usage_type="vcpu-x",
            tags={"Owner": "bob"},
        ),
    ]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )

    only_alice = await top_resources(
        db_session,
        start=today - timedelta(days=2),
        end=today,
        integration_id=integ_id,
        tag_key="Owner", tag_value="alice",
        limit=10,
    )
    assert {r["resource_id"] for r in only_alice} == {"r-alice"}

    only_bob = await top_resources(
        db_session,
        start=today - timedelta(days=2),
        end=today,
        integration_id=integ_id,
        tag_key="Owner", tag_value="bob",
        limit=10,
    )
    assert {r["resource_id"] for r in only_bob} == {"r-bob"}

    # Half-set filter is a 400-class user error.
    with pytest.raises(ValueError):
        await top_resources(
            db_session,
            start=today - timedelta(days=2),
            end=today,
            integration_id=integ_id,
            tag_key="Owner", tag_value=None,
            limit=10,
        )

    # Invalid-charset tag key is rejected before any DB hit (the path
    # would otherwise interpolate JSON-path metacharacters into a
    # ``JSON_EXTRACT(tags, '$."<key>"')`` expression).
    with pytest.raises(ValueError):
        await top_resources(
            db_session,
            start=today - timedelta(days=2),
            end=today,
            integration_id=integ_id,
            tag_key='Owner";--', tag_value="alice",
            limit=10,
        )


def test_validate_tag_filter_rejects_overlong_key_and_value() -> None:
    """Regex limits + value-length cap are user-facing validation;
    a 129-char key or 257-char value must raise rather than slip
    into the SQL ``JSON_EXTRACT`` path expression."""
    from app.domains.cmdb.cost_service import (
        _TAG_VALUE_MAX,
        _validate_tag_filter,
    )

    # 129-char key — regex caps at 128.
    with pytest.raises(ValueError):
        _validate_tag_filter("a" * 129, "v")
    # 128-char key — at the boundary, must be accepted.
    assert _validate_tag_filter("a" * 128, "v") == ("a" * 128, "v")
    # Value > cap.
    with pytest.raises(ValueError):
        _validate_tag_filter("k", "x" * (_TAG_VALUE_MAX + 1))
    # Value at the cap — accepted.
    assert _validate_tag_filter("k", "x" * _TAG_VALUE_MAX) == (
        "k", "x" * _TAG_VALUE_MAX,
    )
    # Adversarial chars (JSON-path metacharacters) — rejected.
    for bad in ('"', "\\", "[", "]", "*", "$", " ", "\t", "\n"):
        with pytest.raises(ValueError):
            _validate_tag_filter(f"foo{bad}bar", "v")
    # None+None returns None (no filter); half-set raises.
    assert _validate_tag_filter(None, None) is None
    with pytest.raises(ValueError):
        _validate_tag_filter("k", None)
    with pytest.raises(ValueError):
        _validate_tag_filter(None, "v")


@pytest.mark.asyncio
async def test_top_resources_empty_dict_rows_dont_block_real_tags(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """``ingest_cost_facts`` writes ``{}`` for untagged facts (not
    NULL). The third-pass tag fetch's JSON_LENGTH > 0 predicate
    must skip those at the DB layer so they don't dominate the
    ``ORDER BY date DESC`` window and silently push real tag rows
    past the 50k LIMIT cap on tag-sparse tenants."""
    integ_id = str(cost_integration.id)
    today = date.today() - timedelta(days=1)
    facts = [
        # 5 days of EMPTY tag rows (latest by date).
        _fact(
            integration_id=integ_id,
            fact_date=today - timedelta(days=i),
            resource_id="r-A", amount_usd=1.0,
            usage_type=f"vcpu-{i}",
            tags={},
        )
        for i in range(5)
    ]
    # Older row WITH tags — must still surface.
    facts.append(_fact(
        integration_id=integ_id,
        fact_date=today - timedelta(days=10),
        resource_id="r-A", amount_usd=1.0,
        usage_type="vcpu-x",
        tags={"Owner": "alice"},
    ))
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )
    rows = await top_resources(
        db_session,
        start=today - timedelta(days=15), end=today,
        integration_id=integ_id, limit=10,
    )
    assert len(rows) == 1
    assert rows[0]["tags"] == {"Owner": "alice"}


@pytest.mark.asyncio
async def test_top_resources_tag_scan_cap_degrades_gracefully(
    db_session: AsyncSession,
    cost_integration: Integration,
    monkeypatch,
):
    """Lock in the contract: when the SQL ``LIMIT`` cap is hit
    before every topN resource has a non-empty tags dict, the
    response degrades gracefully — un-captured resources land with
    ``tags={}`` (no exception, no swap with the wrong resource's
    tags).

    Patches ``TOP_RESOURCES_TAG_SCAN_HARD_CAP`` down to 1 so the
    third-pass cursor returns at most one tag row. The ORDER BY
    date DESC + the row layout below force the cap-truncated
    result to capture r-A (recent) and miss r-B (older). The
    fallback ``tags_by_key.get(spark_key, {})`` in
    ``top_resources`` must paper over the gap.
    """
    import app.domains.cmdb.cost_service as cs

    monkeypatch.setattr(cs, "TOP_RESOURCES_TAG_SCAN_HARD_CAP", 1)

    integ_id = str(cost_integration.id)
    today = date.today() - timedelta(days=1)
    facts = [
        # r-A: recent tagged row — wins the cap-of-1 cursor slot.
        _fact(
            integration_id=integ_id, fact_date=today,
            resource_id="r-A", amount_usd=10.0,
            tags={"Owner": "alice"},
        ),
        # r-B: OLDER tagged row — ORDER BY date DESC pushes it past
        # the LIMIT 1 cap, so its tags must NOT surface.
        _fact(
            integration_id=integ_id,
            fact_date=today - timedelta(days=5),
            resource_id="r-B", amount_usd=5.0,
            usage_type="vcpu-B",
            tags={"Owner": "bob"},
        ),
    ]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )

    rows = await top_resources(
        db_session,
        start=today - timedelta(days=10), end=today,
        integration_id=integ_id, limit=10,
    )
    by_rid = {r["resource_id"]: r["tags"] for r in rows}
    # r-A captured (one cursor slot, most recent).
    assert by_rid["r-A"] == {"Owner": "alice"}
    # r-B missed by the cap — must return empty dict (the contract
    # is "graceful degradation", not "missing key in response").
    assert by_rid["r-B"] == {}


@pytest.mark.asyncio
async def test_list_tag_keys_ranks_by_fact_count(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """``list_tag_keys`` discovery returns observed tag keys ordered
    by per-key fact count (most-used first), used by the dashboard's
    tag-filter autocomplete."""
    from app.domains.cmdb.cost_service import list_tag_keys

    integ_id = str(cost_integration.id)
    today = date.today() - timedelta(days=1)
    facts = [
        # Owner appears on 3 facts; Env on 2; Project on 1.
        _fact(
            integration_id=integ_id, fact_date=today,
            resource_id="r-1", amount_usd=1.0,
            tags={"Owner": "alice", "Env": "prod"},
        ),
        _fact(
            integration_id=integ_id,
            fact_date=today - timedelta(days=1),
            resource_id="r-2", amount_usd=1.0,
            tags={"Owner": "bob", "Env": "staging", "Project": "helix"},
        ),
        _fact(
            integration_id=integ_id,
            fact_date=today - timedelta(days=2),
            resource_id="r-3", amount_usd=1.0,
            tags={"Owner": "carol"},
        ),
    ]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )

    out = await list_tag_keys(
        db_session,
        start=today - timedelta(days=3),
        end=today,
        integration_id=integ_id,
    )
    counts = {row["key"]: row["fact_count"] for row in out}
    assert counts == {"Owner": 3, "Env": 2, "Project": 1}
    # Most-used first.
    assert [row["key"] for row in out] == ["Owner", "Env", "Project"]


@pytest.mark.asyncio
async def test_anomaly_detection_flags_real_spike_only(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """A resource with a flat $10/day cost and one $1000 day should
    surface as an anomaly; a resource with a flat $10/day cost only
    should not."""
    integ_id = str(cost_integration.id)
    # Anchor on UTC like detect_anomalies — without this, CST tests
    # running after local midnight but before UTC midnight push the
    # spike day to date.today()-1 (local) which is OUTSIDE the
    # function's UTC-anchored 30-day window.
    today = datetime.now(UTC).date() - timedelta(days=1)

    facts = []
    # 14 stable days for r-A.
    for i in range(14):
        facts.append(_fact(
            integration_id=integ_id,
            fact_date=today - timedelta(days=i),
            resource_id="r-A",
            amount_usd=10.0,
        ))
    # Drop in a $1000 spike for the most recent day on r-A.
    facts[0] = _fact(
        integration_id=integ_id,
        fact_date=today,
        resource_id="r-A",
        amount_usd=1000.0,
    )
    # 14 stable days for r-B (no spike).
    for i in range(14):
        facts.append(_fact(
            integration_id=integ_id,
            fact_date=today - timedelta(days=i),
            resource_id="r-B",
            amount_usd=10.0,
        ))

    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )

    anomalies = await detect_anomalies(
        db_session,
        days=30,
        integration_id=integ_id,
        min_impact_usd=50.0,
        z_threshold=2.5,
    )
    flagged_ids = {a["resource_id"] for a in anomalies}
    assert "r-A" in flagged_ids
    assert "r-B" not in flagged_ids

    a = next(x for x in anomalies if x["resource_id"] == "r-A")
    assert a["amount_usd"] == pytest.approx(1000.0)
    assert a["impact_usd"] > 50.0

    filtered = await detect_anomalies(
        db_session,
        days=30,
        integration_id=integ_id,
        min_impact_usd=50.0,
        z_threshold=2.5,
        resource_keys={(integ_id, "r-B")},
    )
    assert filtered == []


@pytest.mark.asyncio
async def test_anomaly_does_not_merge_cross_integration_resource_id_collision(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Round-666 regression: two DIFFERENT cloud resources from two
    integrations that happen to share a ``resource_id`` must NOT be
    summed into one anomaly series. Pre-fix, ``detect_anomalies`` grouped
    by ``(resource_id, date)`` only, so integration B's flat data
    polluted integration A's spike series — wrong ``amount_usd`` (A+B
    merged) and arbitrary integration attribution (``func.max``). The fix
    keys on ``(integration_id, resource_id)`` (mirroring ``top_resources``).

    Reachable on the default dashboard / MCP anomalies path: no
    ``integration_id`` filter is passed, so colliding ids merge.
    """
    integ_a = str(cost_integration.id)
    integ_b_row = Integration(
        id=uuid.uuid4(), kind="cloud", provider="aliyun",
        name="Test Aliyun B", slug=f"test-b-{uuid.uuid4().hex[:6]}",
        enabled=True, config_public={},
    )
    db_session.add(integ_b_row)
    await db_session.commit()
    integ_b = str(integ_b_row.id)

    today = datetime.now(UTC).date() - timedelta(days=1)
    SHARED = "i-collision-001"

    # Integration A: shared id, flat $10 for 13 prior days + a $1000 spike
    # on the latest day.
    facts_a = [
        _fact(integration_id=integ_a, fact_date=today - timedelta(days=i),
              resource_id=SHARED, amount_usd=10.0)
        for i in range(1, 14)
    ]
    facts_a.append(_fact(integration_id=integ_a, fact_date=today,
                         resource_id=SHARED, amount_usd=1000.0))
    # Integration B: SAME id, flat $10 EVERY day including the latest — no
    # spike of its own. Under the old merged grain its $10 would inflate
    # A's spike-day amount to $1010 and corrupt the baseline.
    facts_b = [
        _fact(integration_id=integ_b, fact_date=today - timedelta(days=i),
              resource_id=SHARED, amount_usd=10.0)
        for i in range(0, 14)
    ]
    await ingest_cost_facts(db_session, integration_id=integ_a, facts=facts_a)
    await ingest_cost_facts(db_session, integration_id=integ_b, facts=facts_b)

    # Default path: NO integration filter — where the collision bites.
    anomalies = await detect_anomalies(
        db_session, days=30, min_impact_usd=50.0, z_threshold=2.5,
    )
    shared = [a for a in anomalies if a["resource_id"] == SHARED]
    # Exactly one anomaly for the shared id (B is flat → no spike), it is
    # attributed to integration A, and its amount is EXACTLY 1000 (A's own
    # series) — NOT 1010 (the A+B merged value the bug produced).
    assert len(shared) == 1, shared
    assert shared[0]["integration_id"] == integ_a
    assert shared[0]["amount_usd"] == pytest.approx(1000.0)


# ─── Round 1 hardening tests (added after first review pass) ────────


@pytest.mark.asyncio
async def test_ingest_raises_when_integration_missing(
    db_session: AsyncSession,
):
    """Ingest must fail loud when the integration row is gone — silent
    success would write facts with provider="" and break every
    provider-conditional rule for those facts."""
    bogus_id = str(uuid.uuid4())
    with pytest.raises(RuntimeError, match="not found"):
        await ingest_cost_facts(
            db_session,
            integration_id=bogus_id,
            facts=[_fact(
                integration_id=bogus_id, fact_date=date(2026, 5, 1),
            )],
        )


@pytest.mark.asyncio
async def test_ingest_tracks_missing_scope_lookup_misses(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """When the driver returns a CostFact with a scope external_id
    Helix doesn't have a scopes_sync row for, the row is still written
    (with scope_id=NULL) but the ingest summary surfaces the missing
    scope so an operator can re-run scopes_sync."""
    integ_id = str(cost_integration.id)
    fact = CostFact(
        date=date(2026, 5, 1),
        integration_id=integ_id,
        service="Compute",
        amount_usd=10.0,
        amount_native=10.0,
        currency="USD",
        scope=CloudScopeRef(
            scope_type="azure_subscription",
            external_id="missing-from-helix",
        ),
        resource_id="res-A",
    )
    summary = await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[fact],
        scope_lookup={},  # explicit miss
    )
    assert summary["upserted"] == 1
    assert "missing-from-helix" in summary["missing_scopes"]


@pytest.mark.asyncio
async def test_ingest_no_duplicate_after_scopes_sync_catchup(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Regression guard for the cost_service.py:200-220 collision bug.

    First ingest: scopes_sync hasn't run yet, ``scope_lookup={}``,
    fact is written with ``scope_id=NULL``.

    Second ingest (same logical fact, after scopes_sync catches up):
    ``scope_lookup`` resolves the external_id → Helix UUID. The
    UUID is stamped onto the row's ``scope_id`` column.

    The natural key (``fact_key``) MUST be the same across both
    ingests (since the cloud-side ``external_id`` is identical),
    so the second call hits ``ON DUPLICATE KEY UPDATE`` and we end
    up with EXACTLY ONE row — not two. Pre-fix, hashing
    ``scope_db_id`` (NULL → UUID) produced two distinct keys and
    silently duplicated the line item.
    """
    from app.domains.cmdb.models import CmdbCloudScope

    integ_id = str(cost_integration.id)
    fact = CostFact(
        date=date(2026, 5, 2),
        integration_id=integ_id,
        service="Compute",
        amount_usd=10.0,
        amount_native=10.0,
        currency="USD",
        scope=CloudScopeRef(
            scope_type="azure_subscription",
            external_id="sub-late-sync",
        ),
        resource_id="res-late-sync",
    )

    # Round 1: scopes_sync hasn't run yet — the lookup misses.
    r1 = await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[fact],
        scope_lookup={},
    )
    assert r1["upserted"] == 1

    # Round 2: scopes_sync caught up — materialise a real
    # cmdb_cloud_scopes row so the FK on cmdb_cost_facts.scope_id
    # is satisfied when the second ingest stamps the UUID.
    real_scope = CmdbCloudScope(
        integration_id=integ_id,
        scope_type="azure_subscription",
        external_id="sub-late-sync",
        display_name="Late-Sync Sub",
    )
    db_session.add(real_scope)
    await db_session.commit()

    r2 = await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[fact],
        scope_lookup={"sub-late-sync": str(real_scope.id)},
    )
    assert r2["upserted"] == 1
    # No "missing scope" warning the second time.
    assert "sub-late-sync" not in r2.get("missing_scopes", [])

    # Critical: still ONE row in the DB, not two.
    rows = (await db_session.execute(
        select(CmdbCostFact)
        .where(CmdbCostFact.resource_id == "res-late-sync"),
    )).scalars().all()
    assert len(rows) == 1, (
        "fact_key collided across scopes_sync catch-up — duplicate "
        "rows landed in cmdb_cost_facts. Bug: compute_fact_key was "
        "hashing the mutable Helix scope_id instead of the immutable "
        "cloud-side external_id."
    )
    # The surviving row carries the materialised scope_id (the
    # UPDATE branch of ON DUPLICATE KEY UPDATE refreshed it from
    # NULL to the real UUID).
    assert str(rows[0].scope_id) == str(real_scope.id)


@pytest.mark.asyncio
async def test_create_rule_rejects_unknown_person_id(
    db_session: AsyncSession,
):
    """Creating a rule with assign_person_id pointing at a non-existent
    person must raise — otherwise analytics would surface a phantom
    owner on every matched fact."""
    with pytest.raises(ValueError, match="does not refer to"):
        await create_allocation_rule(
            db_session,
            payload={
                "name": "phantom",
                "assign_person_id": str(uuid.uuid4()),
            },
            actor_id=None,
        )


@pytest.mark.asyncio
async def test_update_rule_rejects_unknown_person_id(
    db_session: AsyncSession,
):
    """PATCH guard mirrors create's existence check."""
    rule = await create_allocation_rule(
        db_session,
        payload={"name": "r", "assign_cost_center": "X"},
        actor_id=None,
    )
    with pytest.raises(ValueError, match="does not refer to"):
        await update_allocation_rule(
            db_session,
            rule_id=rule["id"],
            payload={"assign_person_id": str(uuid.uuid4())},
            actor_id=None,
        )


@pytest.mark.asyncio
async def test_delete_missing_rule_raises_lookup_error(
    db_session: AsyncSession,
):
    """DELETE on a non-existent rule must raise so the HTTP layer
    surfaces 404 rather than silently 200-OK."""
    with pytest.raises(LookupError, match="not found"):
        await delete_allocation_rule(
            db_session, rule_id=str(uuid.uuid4()),
        )


@pytest.mark.asyncio
async def test_breakdown_unallocated_kpi_is_subset_not_addend(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """``unallocated_usd`` is a SUBSET of ``total_usd``, not an
    additional bucket. ``total_usd`` is canonical:
    ``sum(groups) + other_usd``. Adding ``unallocated_usd`` would
    double-count.

    With no rules + 100 USD of facts, total=100, unallocated=100,
    sum(groups)+other=100. Adding all three would give 200."""
    integ_id = str(cost_integration.id)
    facts = [_fact(
        integration_id=integ_id, fact_date=date(2026, 5, 1),
        service="Compute", amount_usd=100.0, resource_id="r-A",
    )]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )
    out = await breakdown(
        db_session,
        start=date(2026, 5, 1), end=date(2026, 5, 1),
        group_by="service",
        integration_id=integ_id,
        top=10,
    )
    sum_groups = sum(g["cost_usd"] for g in out["groups"])
    assert out["total_usd"] == pytest.approx(sum_groups + out["other_usd"])
    assert out["unallocated_usd"] == pytest.approx(out["total_usd"])


@pytest.mark.asyncio
async def test_breakdown_labels_null_alloc_dim_as_unallocated(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """When ``group_by`` is one of the allocation-output columns
    (cost_center / department / owner), the NULL group is flagged
    with ``is_unallocated=True`` so the frontend can render an
    i18n'd "未分摊" / "Unallocated" label. Backend keeps user-facing
    strings out of API responses."""
    integ_id = str(cost_integration.id)
    facts = [_fact(
        integration_id=integ_id, fact_date=date(2026, 5, 1),
        service="Compute", amount_usd=42.0, resource_id="r-A",
    )]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )
    out = await breakdown(
        db_session,
        start=date(2026, 5, 1), end=date(2026, 5, 1),
        group_by="cost_center",
        integration_id=integ_id,
        top=10,
    )
    # Exactly one group, the NULL bucket, flagged as unallocated.
    assert any(
        g["is_unallocated"] and g["label"] == ""
        for g in out["groups"]
    )


@pytest.mark.asyncio
async def test_breakdown_unallocated_excludes_savings_plans(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """``group_by='service'`` regression for the SP pin: Savings Plans
    rows are pulled out into their own pinned group (``is_discount``,
    NOT unallocated) and never pollute the ``unallocated_usd`` scalar
    nor ``other_usd``, while ``total_usd`` still nets the SP discount.

    This is the baseline the allocation-dim generalisation mirrors —
    see ``test_breakdown_owner_pins_savings_plans_out_of_null_bucket``.
    """
    integ_id = str(cost_integration.id)
    facts = [
        _fact(
            integration_id=integ_id, fact_date=date(2026, 5, 1),
            service="Compute", amount_usd=100.0, resource_id="r-A",
        ),
        # AWS SP linked-account negation: ingest stamps this with
        # ``auto_owner_source='savings_plan'`` + NULL owner.
        _fact(
            integration_id=integ_id, fact_date=date(2026, 5, 1),
            service="Savings Plans", amount_usd=-30.0, resource_id="sp-1",
        ),
    ]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )
    out = await breakdown(
        db_session,
        start=date(2026, 5, 1), end=date(2026, 5, 1),
        group_by="service",
        integration_id=integ_id,
        top=10,
    )
    # Net total includes the SP discount (gross 100, net 70).
    assert out["total_usd"] == pytest.approx(70.0)
    by_label = {g["label"]: g for g in out["groups"]}
    sp = by_label["Savings Plans"]
    assert sp["cost_usd"] == pytest.approx(-30.0)
    assert sp["is_discount"] is True
    assert sp["is_unallocated"] is False
    # Compute is its own row, untouched by the pin.
    assert by_label["Compute"]["cost_usd"] == pytest.approx(100.0)
    # SP is NOT in the unallocated scalar (service isn't an alloc dim
    # so unallocated here = NULL-owner non-SP facts = 100, the Compute
    # row which has no scope owner).
    assert out["unallocated_usd"] == pytest.approx(100.0)
    # Canonical invariant: sum(groups) + other == total.
    assert sum(g["cost_usd"] for g in out["groups"]) + out["other_usd"] == (
        pytest.approx(out["total_usd"])
    )


@pytest.mark.asyncio
async def test_breakdown_owner_pins_savings_plans_out_of_null_bucket(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Generalised SP pin for an allocation dimension (``owner``).

    Mirrors ``test_breakdown_unallocated_excludes_savings_plans`` but
    for ``group_by='owner'``, where SP rows would otherwise fold into
    the NULL/"未分摊" bucket (SP has NULL cost_owner_person_id by
    design). The pin must:

    * surface SP as its own ``is_unallocated=False`` / ``is_discount``
      group,
    * subtract the SP amount out of the NULL bucket so the chart's
      未分摊 slice == the ``unallocated_usd`` KPI tile, and
    * keep ``total_usd`` netting the SP discount.

    Layout: 100 USD owner-attributed (via scope-owner fallback),
    40 USD unallocated (no scope), -30 USD SP. Net total = 110.
    """
    integ_id = str(cost_integration.id)

    person = await _make_person_with_dept(
        db_session,
        email="sp-owner@pingcap.com",
        department_leaf="Cloud Platform",
    )
    scope = CmdbCloudScope(
        integration_id=integ_id,
        scope_type="aws_account",
        external_id="sp-acct-001",
        display_name="sp-acct-001",
        owner_person_id=str(person.id),
    )
    db_session.add(scope)
    await db_session.commit()

    facts = [
        # Owner-attributed via the scope-owner fallback (no rule).
        CostFact(
            date=date(2026, 5, 1), integration_id=integ_id,
            service="Compute", amount_usd=100.0, amount_native=100.0,
            currency="USD",
            scope=CloudScopeRef(
                scope_type="aws_account", external_id="sp-acct-001",
            ),
            resource_id="r-owned",
        ),
        # Unallocated: no scope → NULL owner, NOT an SP row.
        CostFact(
            date=date(2026, 5, 1), integration_id=integ_id,
            service="Storage", amount_usd=40.0, amount_native=40.0,
            currency="USD", scope=None, resource_id="r-orphan",
        ),
        # SP negation: NULL owner + auto_owner_source='savings_plan'.
        CostFact(
            date=date(2026, 5, 1), integration_id=integ_id,
            service="Savings Plans", amount_usd=-30.0, amount_native=-30.0,
            currency="USD", scope=None, resource_id="sp-1",
        ),
    ]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
        scope_lookup={"sp-acct-001": str(scope.id)},
    )

    out = await breakdown(
        db_session,
        start=date(2026, 5, 1), end=date(2026, 5, 1),
        group_by="owner",
        integration_id=integ_id,
        top=10,
    )

    # Net total nets the SP discount: 100 + 40 - 30 = 110.
    assert out["total_usd"] == pytest.approx(110.0)

    # Pinned SP group: discount, NOT unallocated.
    sp_groups = [g for g in out["groups"] if g["label"] == "Savings Plans"]
    assert len(sp_groups) == 1
    sp = sp_groups[0]
    assert sp["cost_usd"] == pytest.approx(-30.0)
    assert sp["is_discount"] is True
    assert sp["is_unallocated"] is False

    # NULL/未分摊 bucket excludes SP — it carries ONLY the orphan 40,
    # not 40 + (-30) = 10.
    null_groups = [g for g in out["groups"] if g["is_unallocated"]]
    assert len(null_groups) == 1
    assert null_groups[0]["cost_usd"] == pytest.approx(40.0)

    # Owner-attributed row is its own group, never flagged discount.
    owner_groups = [
        g for g in out["groups"]
        if g["key"] == str(person.id)
    ]
    assert len(owner_groups) == 1
    assert owner_groups[0]["cost_usd"] == pytest.approx(100.0)
    assert owner_groups[0]["is_discount"] is False

    # KPI tile agrees with the chart's 未分摊 slice (both exclude SP).
    assert out["unallocated_usd"] == pytest.approx(40.0)
    assert out["unallocated_usd"] == pytest.approx(null_groups[0]["cost_usd"])

    # Canonical invariant survives the pin: sum(groups) + other == total.
    assert sum(g["cost_usd"] for g in out["groups"]) + out["other_usd"] == (
        pytest.approx(out["total_usd"])
    )


@pytest.mark.asyncio
async def test_timeseries_owner_excludes_savings_plans_and_pins_band(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Finding #12 (2026-06-04): ``timeseries()`` must mirror
    ``breakdown()``'s SP handling. Previously SP rows (NULL owner) folded
    into the timeseries 未分摊 band on the allocation dims while the
    breakdown KPI excluded them — two different unallocated numbers on the
    same Cost-Analysis screen. SP must now be pulled out of the unallocated
    band, pinned as its own ``is_discount`` series, and netted only into
    ``total_per_day``.

    Layout: 100 owner-attributed, 40 unallocated (non-SP), -30 SP, all on
    one day. Unallocated band == 40 (NOT 10), SP band == -30, total == 110.
    """
    integ_id = str(cost_integration.id)

    person = await _make_person_with_dept(
        db_session,
        email="sp-ts-owner@pingcap.com",
        department_leaf="Cloud Platform",
    )
    scope = CmdbCloudScope(
        integration_id=integ_id,
        scope_type="aws_account",
        external_id="sp-ts-acct-001",
        display_name="sp-ts-acct-001",
        owner_person_id=str(person.id),
    )
    db_session.add(scope)
    await db_session.commit()

    facts = [
        CostFact(
            date=date(2026, 5, 1), integration_id=integ_id,
            service="Compute", amount_usd=100.0, amount_native=100.0,
            currency="USD",
            scope=CloudScopeRef(
                scope_type="aws_account", external_id="sp-ts-acct-001",
            ),
            resource_id="r-owned",
        ),
        CostFact(
            date=date(2026, 5, 1), integration_id=integ_id,
            service="Storage", amount_usd=40.0, amount_native=40.0,
            currency="USD", scope=None, resource_id="r-orphan",
        ),
        CostFact(
            date=date(2026, 5, 1), integration_id=integ_id,
            service="Savings Plans", amount_usd=-30.0, amount_native=-30.0,
            currency="USD", scope=None, resource_id="sp-ts-1",
        ),
    ]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
        scope_lookup={"sp-ts-acct-001": str(scope.id)},
    )

    out = await timeseries(
        db_session,
        start=date(2026, 5, 1), end=date(2026, 5, 1),
        group_by="owner",
        integration_id=integ_id,
        top=10,
    )

    # Unallocated band carries ONLY the non-SP orphan 40 (not 40 + -30).
    unalloc = [s for s in out["series"] if s.get("is_unallocated")]
    assert len(unalloc) == 1
    assert unalloc[0]["points"] == [pytest.approx(40.0)]

    # SP is pinned as its own discount band, never unallocated.
    sp = [s for s in out["series"] if s.get("is_discount")]
    assert len(sp) == 1
    assert sp[0]["key"] == "Savings Plans"
    assert sp[0]["is_unallocated"] is False
    assert sp[0]["points"] == [pytest.approx(-30.0)]

    # The owner band is its own series, never flagged discount.
    owner = [
        s for s in out["series"]
        if s["key"] == str(person.id)
    ]
    assert len(owner) == 1
    assert owner[0]["points"] == [pytest.approx(100.0)]
    assert owner[0].get("is_discount") is False

    # total_per_day nets the SP discount back in: 100 + 40 - 30 = 110.
    assert out["total_per_day"] == [pytest.approx(110.0)]


@pytest.mark.asyncio
async def test_timeseries_no_savings_plans_band_when_no_sp_rows(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """GCP-only / SP-free tenants must NOT get a spurious all-zero
    'Savings Plans' band (the SP daily query is only appended when the
    window actually contains SP rows)."""
    integ_id = str(cost_integration.id)
    await ingest_cost_facts(
        db_session, integration_id=integ_id,
        facts=[_fact(
            integration_id=integ_id, fact_date=date(2026, 5, 1),
            service="Compute", amount_usd=10.0,
        )],
    )
    out = await timeseries(
        db_session,
        start=date(2026, 5, 1), end=date(2026, 5, 1),
        group_by="service", integration_id=integ_id, top=5,
    )
    assert not [s for s in out["series"] if s.get("is_discount")]


@pytest.mark.asyncio
@pytest.mark.asyncio
async def test_upsert_with_adaptive_split_halves_on_mem_quota():
    """When TiDB rejects a batch with ``tidb_mem_quota_query`` (error
    8175), the helper bisects until each sub-batch fits. A singleton
    that still fails is logged + dropped (return 0), not raised — the
    caller's outer loop must stay alive across rare oversize rows.

    Uses a stub session that fails the first execute (a 4-row batch)
    with the mem-quota error, then succeeds on the two 2-row halves,
    proving the recursion bisects rather than aborts.

    This is the regression test for the 2026-05-16 AWS BQ backfill
    incident: every batch over the historical window hit error 8175
    silently dropping ~6 month × N-account cost_facts before this
    helper was added.
    """
    from unittest.mock import AsyncMock

    from sqlalchemy.exc import OperationalError

    from app.domains.cmdb.cost_service import _upsert_with_adaptive_split

    rows = [
        {"fact_key": f"k{i}", "integration_id": "i", "resource_id": f"r{i}",
         "service": "ECS"} for i in range(4)
    ]

    calls: list[int] = []

    def execute_side_effect(stmt):
        # First call is the full 4-row batch — reject. Subsequent
        # calls (2 + 2) succeed.
        calls.append(len(stmt.parameters)
                     if hasattr(stmt, "parameters") else -1)
        if len(calls) == 1:
            raise OperationalError(
                "fake stmt", {},
                BaseException("(8175, 'tidb_mem_quota_query exceeded')"),
            )
        return None

    session = AsyncMock()
    session.execute.side_effect = execute_side_effect
    session.rollback = AsyncMock()
    session.commit = AsyncMock()

    inserted = await _upsert_with_adaptive_split(session, rows)
    assert inserted == 4, "all rows accounted for after recursive halving"
    # 1 (full) + 2 (halves) = 3 execute calls, plus rollback + commit
    # between halves.
    assert session.execute.await_count == 3
    assert session.rollback.await_count >= 1


@pytest.mark.asyncio
async def test_upsert_with_adaptive_split_drops_singleton_on_mem_quota():
    """A singleton row that still trips the memory quota gets logged
    + skipped (never infinite-recurses). The caller continues."""
    from unittest.mock import AsyncMock

    from sqlalchemy.exc import OperationalError

    from app.domains.cmdb.cost_service import _upsert_with_adaptive_split

    rows = [{"fact_key": "k0", "integration_id": "i", "resource_id": "r0",
             "service": "ECS"}]

    session = AsyncMock()
    session.execute.side_effect = OperationalError(
        "fake", {}, BaseException("(8175, 'tidb_mem_quota_query exceeded')"),
    )
    session.rollback = AsyncMock()

    inserted = await _upsert_with_adaptive_split(session, rows)
    assert inserted == 0
    assert session.execute.await_count == 1


@pytest.mark.asyncio
async def test_upsert_with_adaptive_split_reraises_non_quota_errors():
    """Deadlocks / connection drops / etc must bubble up so the cron
    handler can mark the run as failed — only the mem-quota error
    (8175) triggers the bisect path."""
    from unittest.mock import AsyncMock

    from sqlalchemy.exc import OperationalError

    from app.domains.cmdb.cost_service import _upsert_with_adaptive_split

    rows = [{"fact_key": "k0", "integration_id": "i", "resource_id": "r0",
             "service": "ECS"}]

    session = AsyncMock()
    session.execute.side_effect = OperationalError(
        "fake", {}, BaseException("(1213, 'Deadlock found')"),
    )

    with pytest.raises(OperationalError):
        await _upsert_with_adaptive_split(session, rows)


async def test_ingest_raw_payload_oversize_marker(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Provider payloads that would blow past the JSON column bound
    are replaced with a marker dict (NOT silently truncated to invalid
    JSON, NOT silently dropped).

    The date must stay INSIDE the ``raw`` retention window (2026-09-01):
    ``ingest_cost_facts`` deliberately writes ``raw = NULL`` for days
    already below it, so a hard-coded date would eventually age out of
    the window and null the very column this test asserts on — which is
    exactly how it broke when the retention guard landed (it was pinned
    to 2026-05-01, 123 days back). Relative-to-today keeps it pinned to
    the marker behaviour rather than to the calendar.
    """
    integ_id = str(cost_integration.id)
    huge_blob = "X" * 50_000  # well over the 16k bound
    fact = CostFact(
        date=datetime.now(UTC).date() - timedelta(days=1),
        integration_id=integ_id,
        service="Compute",
        amount_usd=1.0,
        amount_native=1.0,
        currency="USD",
        scope=None,
        resource_id="r-huge",
        raw={"huge_field": huge_blob, "more": list(range(1000))},
    )
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[fact],
    )
    row = (await db_session.execute(
        select(CmdbCostFact)
        .where(CmdbCostFact.resource_id == "r-huge"),
    )).scalar_one()
    assert row.raw is not None
    assert row.raw.get("__helix_truncated__") is True
    assert row.raw.get("__helix_raw_size_bytes__") > 16_000


def test_aliyun_driver_refuses_unprovided_fx_rate():
    """Aliyun list_cost_facts requires ``cny_per_usd`` from the
    caller (cost-ingestion service resolves it via
    :func:`app.core.fx_rate.resolve_cny_per_usd`). Calling without
    the kwarg must raise so a direct caller bypassing the service
    layer can't silently produce wrong USD aggregates."""
    import asyncio

    from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver

    driver = AliyunCloudDriver(
        integration_id="test-int", config={
            "access_key_id": "AK",
            "access_key_secret": "SK",
        },
    )
    with pytest.raises(RuntimeError, match="cny_per_usd"):
        asyncio.get_event_loop().run_until_complete(
            driver.list_cost_facts(
                date(2026, 5, 1), date(2026, 5, 1),
            ),
        )


def test_aliyun_driver_refuses_zero_or_negative_fx_rate():
    """Bad rate values fail loud — clamping would silently produce
    nonsense USD aggregates. Validation lives both here (driver
    self-defence for direct callers) and in the resolver (rejects
    invalid override values from the integration config)."""
    import asyncio

    from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver

    driver = AliyunCloudDriver(
        integration_id="test-int", config={
            "access_key_id": "AK",
            "access_key_secret": "SK",
        },
    )
    for bad in (0, -1.0, 0.0):
        # Round-41 follow-up: driver's CN-primary fallthrough now treats
        # zero/negative as "rate not provided" and raises the unified
        # ``cny_per_usd not provided`` message (single failure mode is
        # easier to action than two near-identical errors). Loosen the
        # regex to match the production message.
        with pytest.raises(RuntimeError, match="cny_per_usd"):
            asyncio.get_event_loop().run_until_complete(
                driver.list_cost_facts(
                    date(2026, 5, 1), date(2026, 5, 1),
                    cny_per_usd=bad,
                ),
            )


def test_tencent_driver_refuses_unprovided_fx_rate():
    """Tencent list_cost_facts requires ``cny_per_usd`` from the
    caller — same posture as Aliyun. Validate in the async wrapper
    so the error surface is identical regardless of whether the
    SDK is importable (which gates the actual API call path)."""
    import asyncio

    from app.integrations.drivers.tencent_cloud import TencentCloudDriver

    driver = TencentCloudDriver(
        integration_id="test-int", config={
            "secret_id": "AK",
            "secret_key": "SK",
        },
    )
    with pytest.raises(RuntimeError, match="cny_per_usd"):
        asyncio.get_event_loop().run_until_complete(
            driver.list_cost_facts(
                date(2026, 5, 1), date(2026, 5, 1),
            ),
        )


def test_tencent_driver_refuses_zero_or_negative_fx_rate():
    import asyncio

    from app.integrations.drivers.tencent_cloud import TencentCloudDriver

    driver = TencentCloudDriver(
        integration_id="test-int", config={
            "secret_id": "AK",
            "secret_key": "SK",
        },
    )
    for bad in (0, -7.0):
        # Round-41 follow-up: same posture as the Aliyun driver — zero
        # / negative is treated as "rate not provided" and surfaces the
        # ``cny_per_usd not provided`` error.
        with pytest.raises(RuntimeError, match="cny_per_usd"):
            asyncio.get_event_loop().run_until_complete(
                driver.list_cost_facts(
                    date(2026, 5, 1), date(2026, 5, 1),
                    cny_per_usd=bad,
                ),
            )


def test_aliyun_and_tencent_drivers_advertise_cny_fx_requirement():
    """The cost-ingestion service inspects ``requires_cny_fx`` to
    decide whether to resolve a CNY→USD rate before calling the
    driver. CNY-native providers must advertise True; flipping
    this to False would let the cron silently pass ``None`` and
    crash inside the driver — pin it with an assert."""
    from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver
    from app.integrations.drivers.tencent_cloud import TencentCloudDriver

    assert AliyunCloudDriver.requires_cny_fx is True
    assert TencentCloudDriver.requires_cny_fx is True


def test_tencent_billing_daily_raises_on_pagination_cap():
    """Round-19 audit regression: when ``DescribeBillDetail``
    pagination hits the per-cycle page cap, the driver MUST raise —
    never return the truncated ``by_day`` snapshot. Pre-fix the driver
    only logged a warning and silently understated the month's
    CNY-denominated spend in every downstream USD aggregate.
    """
    import asyncio
    from datetime import date as _date

    from app.integrations.drivers.tencent_cloud import TencentCloudDriver

    driver = TencentCloudDriver(
        integration_id="test-int",
        config={
            "secret_id": "AK", "secret_key": "SK", "region": "ap-shanghai",
        },
    )

    class _Row:
        # No ComponentSet / FeeBeginTime — the row contributes 0 and is
        # attributed to the cycle's first day; this test only exercises
        # the page-cap raise, not amount/day attribution.
        def __init__(self, product: str):
            self.BusinessCodeName = product

    class _Resp:
        def __init__(self, rows):
            self.DetailSet = rows

    class _StubClient:
        """Always returns 100 rows (full page) so the driver never
        sees ``len(detail) < limit`` and the loop must run all
        ``max_pages`` pages — exactly the truncation case under test.
        Rows carry no cost, so per-row work stays trivial even though
        the outer loop iterates to the cap.
        """

        _PAGE = [_Row(product="") for _ in range(100)]

        def DescribeBillDetail(self, _req):  # noqa: N802
            return _Resp(self._PAGE)

    # Bypass the per-service throttle (4.5 RPS for billing) so the
    # page-cap loop doesn't take minutes in CI just to prove the cap
    # is hit. The throttle is unrelated to the truncation behaviour
    # under test; pinning RPS to 100k effectively disables the sleep.
    from app.integrations import _tencent_throttle as _tt

    with patch.dict(
        _tt.SERVICE_RATES, {"billing": 100_000.0}, clear=False,
    ), patch.object(
        TencentCloudDriver, "_billing_client", return_value=_StubClient(),
    ):
        # Reset cached limiters so the patched rate takes effect.
        _tt._buckets.clear()
        _tt._sync_state.clear()
        with pytest.raises(RuntimeError, match="1000-page cap"):
            asyncio.get_event_loop().run_until_complete(
                driver.list_billing_daily(
                    _date(2026, 5, 1), _date(2026, 5, 31),
                ),
            )


def test_aliyun_and_tencent_declare_cny_per_usd_in_config_schema():
    """The integration admin form reads ``config_schema`` to render
    fields. Without ``cny_per_usd`` declared, ops can't set the
    optional override via the UI — they'd have to reach into
    config_public via SQL. PR #226 had this gap; pin it so a future
    refactor doesn't silently drop the field."""
    from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver
    from app.integrations.drivers.tencent_cloud import TencentCloudDriver

    for driver_cls in (AliyunCloudDriver, TencentCloudDriver):
        keys = {f.key for f in driver_cls.config_schema}
        assert "cny_per_usd" in keys, (
            f"{driver_cls.__name__}.config_schema missing cny_per_usd "
            f"override field — admins can't set it via the UI."
        )
        # Optional override, never required (platform value is the
        # default). Required=True would force every Aliyun/Tencent
        # integration to set it manually, defeating the platform-
        # value-driven design.
        for f in driver_cls.config_schema:
            if f.key == "cny_per_usd":
                assert f.required is False
                assert f.is_secret is False


def test_every_cloud_driver_accepts_cny_per_usd_kwarg():
    """The cost-ingestion service unconditionally passes
    ``cny_per_usd=...`` to every driver's ``list_cost_facts`` (resolved
    or None depending on ``requires_cny_fx``). USD-native drivers
    ignore the value but MUST accept the kwarg — Python rejects
    unknown kwargs at call time, so a missing parameter explodes the
    cron with ``unexpected keyword argument 'cny_per_usd'``.

    PR #229 introduced the kwarg on the CNY-native drivers but
    forgot to update the USD-native ones (GCP / AWS / Azure); the
    cost_facts cron blew up the next morning on every USD-native
    integration. Pin EVERY registered cloud driver — including any
    added in the future — by introspecting the registry rather than
    hard-coding the list, so a new driver can't reintroduce the gap.
    """
    import inspect

    # Force-import drivers package so every ``@register_driver``
    # runs and populates the registry.
    import app.integrations.drivers  # noqa: F401
    from app.integrations.driver_base import CloudDriver
    from app.integrations.registry import _DRIVERS

    cloud_driver_classes = [
        cls for cls in _DRIVERS.values()
        if issubclass(cls, CloudDriver)
    ]
    # Sanity floor — the registry should always carry the 5 we know
    # about. If this drops, drivers/__init__.py stopped importing
    # one and no driver-specific test will catch it.
    assert len(cloud_driver_classes) >= 5, (
        f"Expected ≥5 registered cloud drivers, got "
        f"{[c.__name__ for c in cloud_driver_classes]}"
    )

    for driver_cls in cloud_driver_classes:
        sig = inspect.signature(driver_cls.list_cost_facts)
        assert "cny_per_usd" in sig.parameters, (
            f"{driver_cls.__name__}.list_cost_facts is missing the "
            f"``cny_per_usd`` kwarg — the cost-ingestion service "
            f"passes it unconditionally and Python will reject the "
            f"call with TypeError. See cost_service._run_cost_facts_sync."
        )
        param = sig.parameters["cny_per_usd"]
        assert param.kind == inspect.Parameter.KEYWORD_ONLY, (
            f"{driver_cls.__name__}.list_cost_facts: ``cny_per_usd`` "
            f"must be keyword-only (after ``*,``)."
        )
        assert param.default is None, (
            f"{driver_cls.__name__}.list_cost_facts: ``cny_per_usd`` "
            f"default must be ``None`` to match the base-class contract."
        )


def test_csv_cell_defangs_formula_injection():
    """Cell values starting with a formula char (= + - @ tab CR) MUST
    be prefixed with ``'`` so spreadsheet apps render as text, not
    formula. Prevents CSV-injection chain provider→DB→export→Excel
    formula execution."""
    from app.domains.cmdb.cost_service import _csv_cell

    # Standard formula triggers
    for trigger in ("=", "+", "-", "@"):
        s = _csv_cell(f"{trigger}cmd|'/c calc'!A1")
        assert s.startswith("'") or s.startswith(f'"\'{trigger}'), (
            f"trigger {trigger!r} not defanged: {s!r}"
        )

    # Tab + CR are also formula-injection vectors per OWASP.
    assert _csv_cell("\thello").startswith("'")
    assert _csv_cell("\rhello").startswith("'")

    # Plain values pass through untouched.
    assert _csv_cell("hello") == "hello"
    assert _csv_cell("Compute") == "Compute"
    assert _csv_cell(None) == ""

    # Standard CSV quoting still applies on commas/newlines/quotes.
    assert _csv_cell("a,b") == '"a,b"'
    assert _csv_cell('she said "hi"') == '"she said ""hi"""'


def test_aliyun_tag_parser_handles_spaces_and_lists():
    """The Aliyun ``Tag`` field can be either a ``key:k value:v;``
    string OR an array of ``{TagKey, TagValue}`` objects. Both shapes
    must round-trip cleanly, including values that contain spaces."""
    from app.integrations.drivers.aliyun_cloud import _parse_aliyun_tags

    # String shape, value with spaces:
    out = _parse_aliyun_tags(
        "key:env value:staging prod; key:owner value:alice",
    )
    assert out == {"env": "staging prod", "owner": "alice"}

    # List shape:
    out = _parse_aliyun_tags([
        {"TagKey": "Project", "TagValue": "tidb"},
        {"TagKey": "Env", "TagValue": "prod"},
    ])
    assert out == {"Project": "tidb", "Env": "prod"}

    # Empty / None:
    assert _parse_aliyun_tags(None) == {}
    assert _parse_aliyun_tags("") == {}

    # Garbage type — not a string or list:
    assert _parse_aliyun_tags(42) == {}


def test_match_rule_empty_integration_ids_means_match_any():
    """``match_integration_ids=[]`` (empty list, e.g., from a UI that
    sends an empty multi-select) MUST be treated as 'match any
    integration', NOT 'match none'. Round 2 fix to a P0 silent-
    failure trap surfaced by a fresh-eye review."""
    rule = _make_rule(
        match_integration_ids=[],
        assign_cost_center="DEFAULT",
    )
    out = _match_rule(
        [rule], provider="aws", integration_id="any-id",
        resource_group=None, service="EC2", tags={},
    )
    assert out.get("cost_center") == "DEFAULT"


def test_create_rule_rejects_tag_value_without_key():
    """A rule with ``match_tag_value`` but no ``match_tag_key`` is a
    config trap — admin thinks they're filtering by tag value but the
    matcher silently ignores it. Reject at create."""
    import asyncio

    async def _run():
        from app.core.database import async_session_factory
        async with async_session_factory() as session:
            with pytest.raises(ValueError, match="match_tag_value"):
                await create_allocation_rule(
                    session,
                    payload={
                        "name": "broken",
                        "match_tag_value": "prod",
                        "assign_cost_center": "X",
                    },
                    actor_id=None,
                )
    asyncio.get_event_loop().run_until_complete(_run())


def test_create_rule_rejects_non_list_integration_ids():
    """Pydantic should catch this at the API layer too, but the
    service is the second line of defence (the cron-driven
    create_allocation_rule isn't behind Pydantic)."""
    import asyncio

    async def _run():
        from app.core.database import async_session_factory
        async with async_session_factory() as session:
            with pytest.raises(ValueError, match="match_integration_ids"):
                await create_allocation_rule(
                    session,
                    payload={
                        "name": "broken",
                        "match_integration_ids": "not-a-list",
                        "assign_cost_center": "X",
                    },
                    actor_id=None,
                )
    asyncio.get_event_loop().run_until_complete(_run())


def test_compute_fact_key_disambiguates_tilde_literal():
    """Round 2 hardening: the NULL marker switched from "~" to U+0000
    so a literal "~" in usage_type can't collide with a NULL row."""
    h_null = compute_fact_key(
        integration_id="i", fact_date=date(2026, 5, 1),
        scope_external_id=None, resource_id="r", usage_type=None,
    )
    h_tilde = compute_fact_key(
        integration_id="i", fact_date=date(2026, 5, 1),
        scope_external_id=None, resource_id="r", usage_type="~",
    )
    assert h_null != h_tilde


def test_aws_driver_refuses_scope_none():
    """AWS list_cost_facts requires LINKED_ACCOUNT pin. scope=None
    must raise NotImplementedError, NOT silently fall back to the
    caller's STS account (would return master-account spend on
    Org-master credentials, not member fanout)."""
    import asyncio

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

    driver = AwsCloudDriver(
        integration_id="test-int", config={
            "access_key_id": "AK",
            "secret_access_key": "SK",
            "region": "us-east-1",
        },
    )
    # Use a fresh window (yesterday) so the driver's 14-day-window
    # guard doesn't short-circuit BEFORE the scope=None refusal. The
    # original hard-coded 2026-05-01 went stale once test wallclock
    # moved past 14 days.
    today = datetime.now(UTC).date()
    yesterday = today - timedelta(days=1)
    with pytest.raises(NotImplementedError, match="explicit scope"):
        asyncio.get_event_loop().run_until_complete(
            driver.list_cost_facts(
                yesterday, yesterday, scope=None,
            ),
        )

    # Non-12-digit external_id is also rejected.
    bogus = CloudScopeRef(scope_type="aws_account", external_id="abc")
    with pytest.raises(RuntimeError, match="12-digit"):
        asyncio.get_event_loop().run_until_complete(
            driver.list_cost_facts(
                yesterday, yesterday, scope=bogus,
            ),
        )


@pytest.mark.asyncio
async def test_ingest_persists_tags_as_dict_not_double_encoded(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Tags must round-trip as a real JSON object, not a string-
    encoded JSON string. ``json.dumps`` into a JSON column would
    double-encode and break tag-based allocation rule matching."""
    integ_id = str(cost_integration.id)
    fact = _fact(
        integration_id=integ_id, fact_date=date(2026, 5, 1),
        resource_id="r-tagged",
        tags={"Project": "tidb", "Env": "prod"},
    )
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[fact],
    )
    row = (await db_session.execute(
        select(CmdbCostFact)
        .where(CmdbCostFact.resource_id == "r-tagged"),
    )).scalar_one()
    # Must be a dict, NOT a JSON-encoded string.
    assert isinstance(row.tags, dict)
    assert row.tags["Project"] == "tidb"
    assert row.tags["Env"] == "prod"


@pytest.mark.asyncio
async def test_ingest_accepts_large_usage_quantity_for_gcp_byte_seconds(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """GCP storage SKUs report ``usage.amount`` as cumulative
    byte-seconds at the daily-aggregate level — a single
    ``Storage Machine Image`` row for one ``us-central1`` project
    on 2026-05-09 came in at ``1,570,859,566,034,408 byte-seconds``
    (~1.57e15). Migration 144 widened ``usage_quantity`` from
    ``Numeric(20, 6)`` (max ~9.99e13) to ``Numeric(30, 6)``
    (max ~9.99e23) so this no longer overflows the column. Pre-fix
    the entire UPSERT batch raised ``Out of range value for column
    'usage_quantity'`` and every per-scope GCP cost-facts ingest
    failed.
    """
    from decimal import Decimal as _Dec

    integ_id = str(cost_integration.id)
    huge_quantity = _Dec("1570859566034408.0")  # 1.57e15 byte-seconds
    fact = _fact(
        integration_id=integ_id, fact_date=date(2026, 5, 9),
        resource_id="gcp-byte-seconds-overflow",
        service="Compute Engine",
        usage_type="A652-E97D-AB77@us-central1",
    )
    fact.usage_quantity = float(huge_quantity)
    fact.usage_unit = "byte-seconds"

    summary = await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[fact],
    )
    assert summary["upserted"] == 1
    row = (await db_session.execute(
        select(CmdbCostFact)
        .where(CmdbCostFact.resource_id == "gcp-byte-seconds-overflow"),
    )).scalar_one()
    assert row.usage_quantity == huge_quantity
    assert row.usage_unit == "byte-seconds"


# ─── Scope-owner attribution fallback (migration 145) ──────────────


async def _make_person_with_dept(
    db_session: AsyncSession,
    *,
    email: str,
    department_leaf: str,
    cost_center: str | None = None,
):
    """Spin up an IdentityPerson + a primary IdentityOrgMembership so
    ``_build_person_meta_cache`` resolves a real (department,
    cost_center) pair for them.

    The org_path mirrors the production shape — a leading "PingCAP"
    root segment that ``_build_person_meta_cache`` strips before
    taking the leaf node — so the test exercises the exact path the
    cache uses."""
    from app.identity.person_models import (
        IdentityOrgMembership,
        IdentityPerson,
    )

    pid = uuid.uuid4()
    person = IdentityPerson(
        id=pid,
        work_email=email,
        display_name=email.split("@")[0],
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{pid}",
        cost_center=cost_center,
    )
    db_session.add(person)
    await db_session.flush()
    db_session.add(
        IdentityOrgMembership(
            person_id=person.id,
            source_system="successfactors",
            org_node_source_id=f"NODE-{department_leaf}",
            org_node_type="department",
            org_node_code=department_leaf.replace(" ", "-").upper(),
            org_node_name=department_leaf,
            org_path=f"PingCAP / {department_leaf}",
            is_primary=True,
        ),
    )
    await db_session.commit()
    return person


@pytest.mark.asyncio
async def test_ingest_attributes_via_scope_owner_when_no_rule_matches(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """End-to-end of the production root cause: with zero allocation
    rules in the DB, a fact whose scope has an owner inherits that
    owner via the migration-145 fallback. Pre-fix, every such fact
    landed with ``cost_owner_*=NULL`` and surfaced as 100% 未分摊
    even though the chain fact → scope → owner → department was
    fully populated in the data."""
    integ_id = str(cost_integration.id)

    person = await _make_person_with_dept(
        db_session,
        email="dengshuan@pingcap.com",
        department_leaf="Dataplane Fabric",
    )
    scope = CmdbCloudScope(
        integration_id=integ_id,
        scope_type="azure_subscription",
        external_id="dbaas-prod-dataplane-001",
        display_name="dbaas-prod-dataplane-001",
        owner_person_id=str(person.id),
    )
    db_session.add(scope)
    await db_session.commit()

    fact = CostFact(
        date=date(2026, 5, 1),
        integration_id=integ_id,
        service="Compute",
        amount_usd=42.0,
        amount_native=42.0,
        currency="USD",
        scope=CloudScopeRef(
            scope_type="azure_subscription",
            external_id="dbaas-prod-dataplane-001",
        ),
        resource_id="vm-A",
    )
    summary = await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[fact],
        scope_lookup={"dbaas-prod-dataplane-001": str(scope.id)},
    )
    assert summary["upserted"] == 1

    row = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.resource_id == "vm-A",
        ),
    )).scalar_one()
    assert row.cost_owner_person_id == str(person.id)
    assert row.cost_owner_department == "Dataplane Fabric"
    assert row.allocation_rule_id is None
    assert row.auto_owner_source == "scope_owner"


@pytest.mark.asyncio
async def test_ingest_no_fallback_when_scope_has_no_owner(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """A fact whose scope exists but has no owner stays unattributed
    — ``auto_owner_source='none'`` — so the dashboard's 未分摊 bucket
    correctly counts it. Bug class: a previous design draft would
    have stamped the integration's "primary owner" as a last-ditch
    fallback; that's wrong because integration-level "owner" is
    Helix-side metadata (the person who registered the cloud
    account), not a billing chargeback target."""
    integ_id = str(cost_integration.id)
    scope = CmdbCloudScope(
        integration_id=integ_id,
        scope_type="azure_subscription",
        external_id="orphan-sub",
        display_name="Orphan Sub",
        owner_person_id=None,
    )
    db_session.add(scope)
    await db_session.commit()

    fact = CostFact(
        date=date(2026, 5, 1),
        integration_id=integ_id,
        service="Compute",
        amount_usd=10.0,
        amount_native=10.0,
        currency="USD",
        scope=CloudScopeRef(
            scope_type="azure_subscription",
            external_id="orphan-sub",
        ),
        resource_id="orphan-vm",
    )
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[fact],
        scope_lookup={"orphan-sub": str(scope.id)},
    )
    row = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.resource_id == "orphan-vm",
        ),
    )).scalar_one()
    assert row.cost_owner_person_id is None
    assert row.cost_owner_department is None
    assert row.auto_owner_source == "none"


@pytest.mark.asyncio
async def test_rule_match_takes_priority_over_scope_owner_fallback(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Rules win over the fallback. The Azure RG-split-out flow lives
    on this contract: an Azure subscription has a default owner
    (subscription level), but a specific RG that hosts an API-Key-
    provisioned workload has its own owner via an allocation rule.
    The rule MUST take precedence so the RG cost lands on the API-
    Key applicant, not the subscription owner."""
    integ_id = str(cost_integration.id)

    sub_owner = await _make_person_with_dept(
        db_session,
        email="sub-owner@pingcap.com",
        department_leaf="Cloud Platform",
    )
    rg_owner = await _make_person_with_dept(
        db_session,
        email="rg-owner@pingcap.com",
        department_leaf="AI Innovation",
    )
    scope = CmdbCloudScope(
        integration_id=integ_id,
        scope_type="azure_subscription",
        external_id="ai-sub",
        display_name="AI Sub",
        owner_person_id=str(sub_owner.id),
    )
    db_session.add(scope)
    await db_session.commit()

    rule = CmdbCostAllocationRule(
        id=str(uuid.uuid4()),
        name="Foundry RG → applicant",
        priority=10,
        match_provider="azure_cloud",
        match_resource_group="rg-foundry-001",
        assign_person_id=str(rg_owner.id),
    )
    db_session.add(rule)
    await db_session.commit()

    fact_in_rg = CostFact(
        date=date(2026, 5, 1),
        integration_id=integ_id,
        service="Cognitive Services",
        amount_usd=120.0,
        amount_native=120.0,
        currency="USD",
        scope=CloudScopeRef(
            scope_type="azure_subscription", external_id="ai-sub",
        ),
        resource_group="rg-foundry-001",
        resource_id="aoai-A",
    )
    fact_outside_rg = CostFact(
        date=date(2026, 5, 1),
        integration_id=integ_id,
        service="Compute",
        amount_usd=30.0,
        amount_native=30.0,
        currency="USD",
        scope=CloudScopeRef(
            scope_type="azure_subscription", external_id="ai-sub",
        ),
        resource_group="rg-default",
        resource_id="vm-default",
    )
    await ingest_cost_facts(
        db_session, integration_id=integ_id,
        facts=[fact_in_rg, fact_outside_rg],
        scope_lookup={"ai-sub": str(scope.id)},
    )

    rg_row = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.resource_id == "aoai-A",
        ),
    )).scalar_one()
    assert rg_row.cost_owner_person_id == str(rg_owner.id)
    assert rg_row.cost_owner_department == "AI Innovation"
    assert rg_row.allocation_rule_id == rule.id
    assert rg_row.auto_owner_source == "rule"

    default_row = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.resource_id == "vm-default",
        ),
    )).scalar_one()
    # No rule matched → falls back to subscription owner.
    assert default_row.cost_owner_person_id == str(sub_owner.id)
    assert default_row.cost_owner_department == "Cloud Platform"
    assert default_row.allocation_rule_id is None
    assert default_row.auto_owner_source == "scope_owner"


@pytest.mark.asyncio
async def test_recompute_attributes_existing_unallocated_facts(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """The "重算最近 90 天" button: pre-fix it returned ``命中 0 / 更新 0``
    on a tenant with no rules even though every fact had a scope
    with an owner. Post-fix the recompute walks the same fallback
    chain as ingest and re-stamps the historical rows with
    ``auto_owner_source='scope_owner'``."""
    integ_id = str(cost_integration.id)

    # Land a fact while the scope has NO owner — simulates the prod
    # state: facts ingested before we taught the engine the
    # fallback. ``cost_owner_*`` is NULL.
    scope = CmdbCloudScope(
        integration_id=integ_id,
        scope_type="azure_subscription",
        external_id="historical-sub",
        display_name="Historical",
        owner_person_id=None,
    )
    db_session.add(scope)
    await db_session.commit()

    fact = CostFact(
        date=date.today(),
        integration_id=integ_id,
        service="Compute",
        amount_usd=99.0,
        amount_native=99.0,
        currency="USD",
        scope=CloudScopeRef(
            scope_type="azure_subscription",
            external_id="historical-sub",
        ),
        resource_id="hist-vm",
    )
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[fact],
        scope_lookup={"historical-sub": str(scope.id)},
    )

    pre_row = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.resource_id == "hist-vm",
        ),
    )).scalar_one()
    assert pre_row.cost_owner_person_id is None
    assert pre_row.auto_owner_source == "none"

    # Admin assigns an owner to the scope retroactively (mirrors
    # the prod recovery action: scope-ownership is set via the CMDB
    # admin UI after the fact). Until recompute runs, the existing
    # fact rows still have NULL owner.
    person = await _make_person_with_dept(
        db_session,
        email="lu.xing@pingcap.com",
        department_leaf="O11Y",
    )
    scope.owner_person_id = str(person.id)
    db_session.add(scope)
    await db_session.commit()

    summary = await recompute_allocation(db_session, days=30)

    assert summary["scanned"] >= 1
    assert summary["matched_by_rule"] == 0
    assert summary["matched_by_scope"] >= 1
    assert summary["updated"] >= 1
    # Backwards-compat key the frontend i18n consumes.
    assert summary["matched"] == (
        summary["matched_by_rule"] + summary["matched_by_scope"]
    )

    post_row = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.resource_id == "hist-vm",
        ),
    )).scalar_one()
    assert post_row.cost_owner_person_id == str(person.id)
    assert post_row.cost_owner_department == "O11Y"
    assert post_row.auto_owner_source == "scope_owner"


@pytest.mark.asyncio
async def test_recompute_summary_reports_still_unallocated(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """The recompute summary surfaces the residual 未分摊 count so the
    operator can see "X facts still couldn't be attributed" without
    re-running a separate query. This is what powers the dashboard's
    'still unallocated' counter when the recompute toast lands."""
    integ_id = str(cost_integration.id)

    fact_no_scope = CostFact(
        date=date.today(),
        integration_id=integ_id,
        service="Compute",
        amount_usd=5.0,
        amount_native=5.0,
        currency="USD",
        scope=None,  # no scope — falls all the way through
        resource_id="rogue",
    )
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[fact_no_scope],
    )
    summary = await recompute_allocation(db_session, days=30)
    assert summary["still_unallocated"] >= 1
    assert summary["matched_by_rule"] == 0
    assert summary["matched_by_scope"] == 0


@pytest.mark.asyncio
async def test_recompute_scan_crosses_sparse_id_dead_zones(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Regression for the nightly 8176 kill: recompute must scan through
    the date-indexed in-window rows only, rather than range-scanning a
    historical id gap to fill one global keyset batch.

    This plants two in-window facts separated by old history and asserts
    both remain reachable without ever visiting the out-of-window row.
    """
    from decimal import Decimal

    integ_id = str(cost_integration.id)
    scope = CmdbCloudScope(
        integration_id=integ_id,
        scope_type="azure_subscription",
        external_id="sparse-sub",
        display_name="Sparse",
        owner_person_id=None,
    )
    db_session.add(scope)
    await db_session.commit()

    def _fact(fact_id: int, fact_date: date, rid: str) -> CmdbCostFact:
        return CmdbCostFact(
            id=fact_id,
            fact_key=compute_fact_key(
                fact_date=fact_date, integration_id=integ_id,
                scope_external_id="sparse-sub", resource_id=rid,
                usage_type=None,
            ),
            date=fact_date,
            integration_id=integ_id,
            scope_id=str(scope.id),
            resource_id=rid,
            service="Compute",
            amount_usd=Decimal("1"),
            amount_native=Decimal("1"),
            currency="USD",
        )

    today = date.today()
    stale = today - timedelta(days=400)
    # In-window facts on both sides of a historical id gap; the middle
    # row must be excluded by the accounting-day predicate.
    db_session.add_all([
        _fact(100, today, "vm-left"),
        _fact(800_000, stale, "vm-dead-zone"),
        _fact(1_600_000, today, "vm-right"),
    ])
    await db_session.commit()

    person = await _make_person_with_dept(
        db_session,
        email="sparse.owner@pingcap.com",
        department_leaf="Infra",
    )
    scope.owner_person_id = str(person.id)
    db_session.add(scope)
    await db_session.commit()

    summary = await recompute_allocation(db_session, days=30)

    # Only the two in-window rows are visited; the scan must not stop
    # at the empty band between them.
    assert summary["scanned"] == 2
    assert summary["matched_by_scope"] == 2
    assert summary["updated"] == 2

    stamped = (await db_session.execute(
        select(CmdbCostFact.resource_id, CmdbCostFact.auto_owner_source)
        .where(CmdbCostFact.integration_id == integ_id),
    )).all()
    by_rid = {r.resource_id: r.auto_owner_source for r in stamped}
    assert by_rid["vm-left"] == "scope_owner"
    assert by_rid["vm-right"] == "scope_owner"
    # Out-of-window history is untouched by the windowed recompute.
    assert by_rid["vm-dead-zone"] == "none"


@pytest.mark.asyncio
async def test_recompute_batches_writes_without_cross_contaminating_owners(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """Regression for the 2026-05-27 nightly timeout fix: ``recompute``
    no longer issues one ``UPDATE … WHERE id = :id`` per changed row —
    it groups changed rows by the target tuple they should END UP with
    and flushes one ``UPDATE … WHERE id IN (…)`` per distinct target per
    scan batch. The risk that grouping introduces is cross-contamination:
    a fact belonging to owner A getting stamped with owner B's tuple.

    This drives THREE distinct targets (three scopes → three different
    owners/departments) with MULTIPLE facts per target in a single
    recompute pass, then asserts every fact lands on its OWN owner. A
    re-run must report ``updated == 0`` — proving the batched path keeps
    the per-row change detection honest and converges (no always-write
    regression that would re-stamp the whole window every night)."""
    integ_id = str(cost_integration.id)

    p1 = await _make_person_with_dept(
        db_session, email="alloc.a@pingcap.com", department_leaf="Dept-A",
        cost_center="CC-A",
    )
    p2 = await _make_person_with_dept(
        db_session, email="alloc.b@pingcap.com", department_leaf="Dept-B",
        cost_center="CC-B",
    )
    p3 = await _make_person_with_dept(
        db_session, email="alloc.c@pingcap.com", department_leaf="Dept-C",
        cost_center="CC-C",
    )

    # Three scopes, each ingested while owner is NULL so the facts land
    # unallocated (the historical prod state recompute exists to fix).
    owners = {"sub-a": p1, "sub-b": p2, "sub-c": p3}
    scopes: dict[str, CmdbCloudScope] = {}
    for ext_id in owners:
        sc = CmdbCloudScope(
            integration_id=integ_id,
            scope_type="azure_subscription",
            external_id=ext_id,
            display_name=ext_id,
            owner_person_id=None,
        )
        db_session.add(sc)
        scopes[ext_id] = sc
    await db_session.commit()

    # 2 facts under sub-a, 2 under sub-b, 1 under sub-c — so the two
    # multi-fact targets prove the IN-list updates ALL ids of a group
    # (not just the first) and only those.
    plan = [("sub-a", "vm-a1"), ("sub-a", "vm-a2"),
            ("sub-b", "vm-b1"), ("sub-b", "vm-b2"),
            ("sub-c", "vm-c1")]
    for ext_id, res_id in plan:
        await ingest_cost_facts(
            db_session, integration_id=integ_id,
            facts=[CostFact(
                date=date.today(),
                integration_id=integ_id,
                service="Compute",
                amount_usd=10.0,
                amount_native=10.0,
                currency="USD",
                scope=CloudScopeRef(
                    scope_type="azure_subscription", external_id=ext_id,
                ),
                resource_id=res_id,
            )],
            scope_lookup={ext_id: str(scopes[ext_id].id)},
        )

    # Retroactively assign each scope a DIFFERENT owner.
    for ext_id, person in owners.items():
        scopes[ext_id].owner_person_id = str(person.id)
        db_session.add(scopes[ext_id])
    await db_session.commit()

    summary = await recompute_allocation(db_session, days=30)
    assert summary["matched_by_scope"] == 5
    assert summary["updated"] == 5

    rows = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.resource_id.in_([r for _, r in plan]),
        ),
    )).scalars().all()
    by_res = {r.resource_id: r for r in rows}
    expected = {
        "vm-a1": (p1, "Dept-A", "CC-A"),
        "vm-a2": (p1, "Dept-A", "CC-A"),
        "vm-b1": (p2, "Dept-B", "CC-B"),
        "vm-b2": (p2, "Dept-B", "CC-B"),
        "vm-c1": (p3, "Dept-C", "CC-C"),
    }
    for res_id, (person, dept, cc) in expected.items():
        row = by_res[res_id]
        assert row.cost_owner_person_id == str(person.id), res_id
        assert row.cost_owner_department == dept, res_id
        assert row.cost_center == cc, res_id
        assert row.auto_owner_source == "scope_owner", res_id

    # Idempotent: a second pass changes nothing.
    again = await recompute_allocation(db_session, days=30)
    assert again["updated"] == 0
    assert again["matched_by_scope"] == 5


@pytest_asyncio.fixture
async def patched_session_factory(db_engine):
    """Repoint ``app.core.database.async_session_factory`` at the test
    engine. Mirrors the ``test_failure_ack_and_dedup`` fixture — needed
    for service-layer tests that drive code which opens its own
    session (here: ``sync_cost_facts_for_integration``)."""
    from sqlalchemy.ext.asyncio import async_sessionmaker as _amsmaker

    from app.core import database as _db

    original = _db.async_session_factory
    _db.async_session_factory = _amsmaker(
        db_engine, class_=AsyncSession, expire_on_commit=False,
    )
    try:
        yield
    finally:
        _db.async_session_factory = original


@pytest.mark.asyncio
async def test_sync_cost_facts_skips_sub_tenant_scope_rows(
    db_session: AsyncSession,
    cost_integration: Integration,
    patched_session_factory,
):
    """Regression guard: ``cmdb.cost_facts_sync`` must NOT pass
    sub-tenant scope rows (today: ``azure_resource_group`` minted by
    the Azure OpenAI auto-grant flow) to the driver.

    Pre-fix, the cron iterated every active ``cmdb_cloud_scopes`` row
    and the Azure driver built ``/subscriptions/{rg_name}/...`` URLs
    that Cost Management rejected with ``InvalidSubscriptionId`` —
    flipping the integration into ``partial_failure`` on every tick
    even though the parent subscription's spend ingested fine. RG
    cost is attributed to the applicant via the allocation rule the
    auto-grant flow creates alongside the RG row, NOT via direct
    driver queries against the RG.
    """
    from app.domains.cmdb import cost_service as cost_service_mod
    from app.domains.cmdb.models import (
        SCOPE_TYPE_AZURE_RESOURCE_GROUP,
        SCOPE_TYPE_AZURE_SUBSCRIPTION,
    )

    integ_id = str(cost_integration.id)
    sub_scope = CmdbCloudScope(
        id=str(uuid.uuid4()),
        integration_id=integ_id,
        scope_type=SCOPE_TYPE_AZURE_SUBSCRIPTION,
        external_id="00000000-0000-0000-0000-000000000001",
        display_name="Test Sub",
        status="active",
    )
    rg_scope = CmdbCloudScope(
        id=str(uuid.uuid4()),
        integration_id=integ_id,
        scope_type=SCOPE_TYPE_AZURE_RESOURCE_GROUP,
        external_id="TestRG",  # RG name — NOT a subscription GUID
        display_name="Test RG",
        parent_scope_id=str(sub_scope.id),
        status="active",
    )
    db_session.add_all([sub_scope, rg_scope])
    await db_session.commit()

    called_with: list[CloudScopeRef | None] = []

    class _StubDriver:
        provider = "azure_cloud"
        requires_cny_fx = False

        async def list_cost_facts(
            self, start_date, end_date, *,
            scope=None, cny_per_usd=None,
            principal_tag_keys=None, bq_billing=None,
        ):
            called_with.append(scope)
            return []

        async def close(self) -> None:  # pragma: no cover — trivial
            return None

    async def _fake_loader(_session, _integration):
        return _StubDriver()

    with patch(
        "app.domains.cmdb.lifecycle_service._load_cloud_driver",
        new=_fake_loader,
    ):
        summary = await cost_service_mod.sync_cost_facts_for_integration(
            integ_id,
            start_date=date(2026, 5, 1),
            end_date=date(2026, 5, 1),
        )

    # Driver invoked exactly once — for the subscription scope only.
    assert len(called_with) == 1
    assert called_with[0] is not None
    assert called_with[0].scope_type == SCOPE_TYPE_AZURE_SUBSCRIPTION
    assert called_with[0].external_id == (
        "00000000-0000-0000-0000-000000000001"
    )
    # No driver errors — the RG row never reached the driver.
    assert summary["status"] == "ok"


@pytest.mark.asyncio
async def test_sync_cost_facts_default_window_is_three_day_rolling(
    db_session: AsyncSession,
    cost_integration: Integration,
    patched_session_factory,
):
    """Regression guard: when neither ``start_date`` nor ``end_date`` is
    supplied (the cron path), the service must call the driver with a
    3-day rolling window ending yesterday.

    This pins parity with ``sync_billing_daily_for_integration``
    (lifecycle_service.py: ``start_dt = end_dt - timedelta(days=2)``).
    Pre-fix the cost-facts default was a single-day window (start ==
    end == yesterday), which silently undercounted cmdb_cost_facts by
    5-12% vs cmdb_cloud_billing_daily because cloud billing exports
    keep ingesting late-arriving rows for ~2-3 days after the usage
    date — billing_daily caught them on its 3-day re-fetch, cost-facts
    did not. Direction is provider-agnostic (verified GCP + Azure
    2026-05-18). UPSERT is idempotent (keyed on ``fact_key``), so
    re-running for the same day overwrites stale values safely.
    """
    from datetime import UTC, datetime

    from app.domains.cmdb import cost_service as cost_service_mod
    from app.domains.cmdb.models import SCOPE_TYPE_AZURE_SUBSCRIPTION

    integ_id = str(cost_integration.id)
    db_session.add(CmdbCloudScope(
        id=str(uuid.uuid4()),
        integration_id=integ_id,
        scope_type=SCOPE_TYPE_AZURE_SUBSCRIPTION,
        external_id="00000000-0000-0000-0000-0000000000ff",
        display_name="Test Sub",
        status="active",
    ))
    await db_session.commit()

    seen: list[tuple[date, date]] = []

    class _StubDriver:
        provider = "azure_cloud"
        requires_cny_fx = False

        async def list_cost_facts(
            self, start_date, end_date, *,
            scope=None, cny_per_usd=None,
            principal_tag_keys=None, bq_billing=None,
        ):
            seen.append((start_date, end_date))
            return []

        async def close(self) -> None:  # pragma: no cover — trivial
            return None

    async def _fake_loader(_session, _integration):
        return _StubDriver()

    with patch(
        "app.domains.cmdb.lifecycle_service._load_cloud_driver",
        new=_fake_loader,
    ):
        # NO start_date / end_date — exercises the cron-default path.
        summary = await cost_service_mod.sync_cost_facts_for_integration(
            integ_id,
        )

    assert summary["status"] == "ok", f"unexpected summary: {summary}"
    assert len(seen) == 1, "driver should be called exactly once per scope"
    start_seen, end_seen = seen[0]
    expected_end = datetime.now(UTC).date() - timedelta(days=1)
    expected_start = expected_end - timedelta(days=2)
    assert end_seen == expected_end, (
        f"end_date default must be yesterday (UTC); got {end_seen} "
        f"expected {expected_end}"
    )
    assert start_seen == expected_start, (
        f"start_date default must be end_date - 2 days (3-day rolling); "
        f"got {start_seen} expected {expected_start} "
        f"(parity with sync_billing_daily_for_integration)"
    )


@pytest.mark.asyncio
async def test_sync_cost_facts_explicit_window_writes_service_audit(
    db_session: AsyncSession,
    cost_integration: Integration,
    patched_session_factory,
):
    """Regression guard for the 2026-07-28 prod forensics gap: a
    multi-month cost-facts reingest landed with NO audit_logs and NO
    scheduled_job_runs row because the caller invoked
    ``sync_cost_facts_for_integration`` directly (the repo's
    ``scripts/backfill_cost_facts.py`` path), bypassing the audited
    HTTP / MCP entry points.

    Contract pinned here:
      * explicit-window call → exactly one ``actor_type=system``
        ``cost.facts_sync.window_invoked`` row, detail carrying the
        window + best-effort caller;
      * a bogus inverted window still traces (the row lands BEFORE
        the start>end guard returns);
      * the cron default path (no window) writes NO such row — daily
        ticks must not spam audit_logs.
    """
    from app.audit.models import AuditLog
    from app.domains.cmdb import cost_service as cost_service_mod
    from app.domains.cmdb.models import SCOPE_TYPE_AZURE_SUBSCRIPTION

    integ_id = str(cost_integration.id)
    db_session.add(CmdbCloudScope(
        id=str(uuid.uuid4()),
        integration_id=integ_id,
        scope_type=SCOPE_TYPE_AZURE_SUBSCRIPTION,
        external_id="00000000-0000-0000-0000-0000000000aa",
        display_name="Audit Test Sub",
        status="active",
    ))
    await db_session.commit()

    class _StubDriver:
        provider = "azure_cloud"
        requires_cny_fx = False

        async def list_cost_facts(
            self, start_date, end_date, *,
            scope=None, cny_per_usd=None,
            principal_tag_keys=None, bq_billing=None,
        ):
            return []

        async def close(self) -> None:  # pragma: no cover — trivial
            return None

    async def _fake_loader(_session, _integration):
        return _StubDriver()

    async def _audit_rows() -> list[AuditLog]:
        # The audit row commits in its own session; end db_session's
        # current REPEATABLE READ snapshot so the select sees it.
        await db_session.commit()
        rows = (await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "cost.facts_sync.window_invoked",
                AuditLog.resource_id == integ_id,
            ).order_by(AuditLog.created_at),
        )).scalars().all()
        return list(rows)

    with patch(
        "app.domains.cmdb.lifecycle_service._load_cloud_driver",
        new=_fake_loader,
    ):
        # 1) Cron default path — no window, no service audit row.
        summary = await cost_service_mod.sync_cost_facts_for_integration(
            integ_id,
        )
        assert summary["status"] == "ok"
        assert await _audit_rows() == [], (
            "cron default (no explicit window) must NOT write the "
            "service-layer invocation audit row"
        )

        # 2) Explicit window — exactly one system trace row.
        summary = await cost_service_mod.sync_cost_facts_for_integration(
            integ_id,
            start_date=date(2026, 5, 1),
            end_date=date(2026, 5, 31),
        )
        assert summary["status"] == "ok"

    rows = await _audit_rows()
    assert len(rows) == 1, f"expected one audit row, got {len(rows)}"
    row = rows[0]
    assert row.actor_type == "system"
    assert row.status == "ok"
    assert row.resource_type == "integration"
    assert row.detail["start_date"] == "2026-05-01"
    assert row.detail["end_date"] == "2026-05-31"
    assert row.detail["window_days"] == 31
    assert row.detail["entrypoint"] == "sync_cost_facts_for_integration"
    # Caller attribution is best-effort but for a direct in-repo call
    # it must at least point outside cost_service.py — here, this
    # test module.
    assert row.detail["caller"] and "test_cost_service.py" in row.detail["caller"]

    # 3) Inverted window: guard returns "failed" without touching the
    #    driver, but the invocation trace must land anyway.
    result = await cost_service_mod.sync_cost_facts_for_integration(
        integ_id,
        start_date=date(2026, 6, 2),
        end_date=date(2026, 6, 1),
    )
    assert result["status"] == "failed"
    rows = await _audit_rows()
    assert len(rows) == 2, (
        "an inverted-window direct call must still leave an audit trace"
    )
    # created_at is second-precision — don't rely on row order.
    assert {r.detail["window_days"] for r in rows} == {31, 0}


@pytest.mark.asyncio
async def test_sync_cost_facts_evicts_stale_facts_before_upsert(
    db_session: AsyncSession,
    cost_integration: Integration,
    patched_session_factory,
):
    """Regression guard for the Round-36 staleness bug (2026-05-18):

    A prior run emitted a fact for natural-key K = (date, scope,
    resource, usage_type) when the driver's grouped SUM was positive
    (typical: an AWS SP CoveredUsage row whose paired SP Negation row
    has not yet been published by AWS). A later run, with the negation
    row now present, finds K's group sums to exactly $0 and the driver
    skips emission via ``if native == 0.0: continue``. Without
    pre-window DELETE, UPSERT never overwrites the stale row — it
    survives as a phantom positive cost forever.

    Reproducer: pre-seed a fact for the (scope, date) window that the
    second cron run would NOT re-emit. After ``sync_cost_facts_for_
    integration`` runs, the pre-seeded fact must be gone (DELETE
    swept it before the driver's fresh facts UPSERT-landed in its
    place). Verifies the eviction lands both on the non-empty-facts
    branch AND the empty-facts branch (a scope that legitimately
    dropped to zero cost gets cleaned up too).
    """
    from app.domains.cmdb import cost_service as cost_service_mod
    from app.domains.cmdb.models import SCOPE_TYPE_AZURE_SUBSCRIPTION

    integ_id = str(cost_integration.id)
    scope_a_id = str(uuid.uuid4())
    scope_b_id = str(uuid.uuid4())
    scope_a = CmdbCloudScope(
        id=scope_a_id,
        integration_id=integ_id,
        scope_type=SCOPE_TYPE_AZURE_SUBSCRIPTION,
        external_id="00000000-0000-0000-0000-0000000000aa",
        display_name="Scope A (non-empty facts)",
        status="active",
    )
    scope_b = CmdbCloudScope(
        id=scope_b_id,
        integration_id=integ_id,
        scope_type=SCOPE_TYPE_AZURE_SUBSCRIPTION,
        external_id="00000000-0000-0000-0000-0000000000bb",
        display_name="Scope B (driver returns empty)",
        status="active",
    )
    db_session.add_all([scope_a, scope_b])

    window_start = date(2026, 5, 10)
    window_end = date(2026, 5, 12)

    # Stale facts pre-seeded for BOTH scopes inside the window. These
    # simulate "the previous cron emitted them; a later BQ snapshot
    # makes the same fact_key sum to $0 so the driver no longer emits".
    stale_keys = [
        ("aa", "stale-resource-a", "stale-usage-a"),
        ("bb", "stale-resource-b", "stale-usage-b"),
    ]
    for tag, rid, ut in stale_keys:
        for offset in range(3):  # 5/10, 5/11, 5/12
            d = window_start + timedelta(days=offset)
            fk = compute_fact_key(
                integration_id=integ_id,
                fact_date=d,
                scope_external_id=(
                    "00000000-0000-0000-0000-0000000000" + tag
                ),
                resource_id=rid,
                usage_type=ut,
            )
            db_session.add(CmdbCostFact(
                fact_key=fk,
                date=d,
                integration_id=integ_id,
                scope_id=scope_a_id if tag == "aa" else scope_b_id,
                resource_id=rid,
                usage_type=ut,
                service="StaleService",
                amount_usd=999.99,  # absurd marker; real ingest never matches
                amount_native=999.99,
                currency="USD",
            ))
    # Also seed an out-of-window stale fact: must NOT be evicted.
    out_of_window = window_end + timedelta(days=5)
    fk_outside = compute_fact_key(
        integration_id=integ_id,
        fact_date=out_of_window,
        scope_external_id="00000000-0000-0000-0000-0000000000aa",
        resource_id="stale-outside",
        usage_type="stale-outside-usage",
    )
    db_session.add(CmdbCostFact(
        fact_key=fk_outside,
        date=out_of_window,
        integration_id=integ_id,
        scope_id=scope_a_id,
        resource_id="stale-outside",
        usage_type="stale-outside-usage",
        service="StaleService",
        amount_usd=42.0,
        amount_native=42.0,
        currency="USD",
    ))
    await db_session.commit()

    # Stub driver: scope A returns one fresh fact, scope B returns none.
    class _StubDriver:
        provider = "azure_cloud"
        requires_cny_fx = False

        async def list_cost_facts(
            self, start_date, end_date, *,
            scope=None, cny_per_usd=None,
            principal_tag_keys=None, bq_billing=None,
        ):
            if scope and scope.external_id.endswith("aa"):
                return [CostFact(
                    date=start_date,
                    integration_id=integ_id,
                    service="FreshService",
                    amount_usd=5.0,
                    amount_native=5.0,
                    currency="USD",
                    scope=scope,
                    resource_group=scope.external_id,
                    resource_id="fresh-resource",
                    resource_name="fresh-resource",
                    resource_type="compute",
                    region=None,
                    usage_type="fresh-usage",
                    tags={},
                    usage_quantity=None,
                    usage_unit=None,
                    raw={},
                )]
            return []

        async def close(self) -> None:  # pragma: no cover — trivial
            return None

    async def _fake_loader(_session, _integration):
        return _StubDriver()

    with patch(
        "app.domains.cmdb.lifecycle_service._load_cloud_driver",
        new=_fake_loader,
    ):
        summary = await cost_service_mod.sync_cost_facts_for_integration(
            integ_id,
            start_date=window_start,
            end_date=window_end,
        )

    assert summary["status"] == "ok"

    # Scope A: stale facts in window must be evicted; the fresh fact
    # must have landed on its own natural key.
    survivors_a = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.scope_id == scope_a_id,
            CmdbCostFact.date >= window_start,
            CmdbCostFact.date <= window_end,
        )
    )).scalars().all()
    assert all(f.service != "StaleService" for f in survivors_a), (
        "stale facts for scope A inside the window were not evicted"
    )
    assert any(f.service == "FreshService" for f in survivors_a), (
        "fresh fact for scope A did not land after eviction"
    )

    # Scope B: driver returned empty — stale facts MUST still be evicted
    # so a scope that legitimately went to zero cost stops showing
    # phantom spend on the dashboard.
    survivors_b = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.scope_id == scope_b_id,
            CmdbCostFact.date >= window_start,
            CmdbCostFact.date <= window_end,
        )
    )).scalars().all()
    assert survivors_b == [], (
        "empty-facts branch did not evict stale rows — scope B "
        "still shows phantom cost after a zero-cost window"
    )

    # Out-of-window stale fact MUST survive — eviction is bounded by
    # the requested [start_date, end_date], not a tenant-wide wipe.
    outside_row = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.scope_id == scope_a_id,
            CmdbCostFact.date == out_of_window,
        )
    )).scalars().first()
    assert outside_row is not None, (
        "eviction overshot the requested window — out-of-range "
        "stale fact was deleted"
    )
    assert outside_row.service == "StaleService"


@pytest.mark.asyncio
async def test_sync_cost_facts_per_scope_timeout_isolates_hung_scope(
    db_session: AsyncSession,
    cost_integration: Integration,
    patched_session_factory,
):
    """A driver call that hangs longer than the per-scope timeout MUST
    NOT block subsequent scopes. The cron records ``status='timeout'``
    in ``per_scope[]`` + an errors[] entry for the hung scope, and
    continues iterating.

    Pre-fix (2026-05-19 prod incident): a CUR 2.0 cost_facts_sync run
    hung at scope #43 (after 41 successful scopes) and stayed in
    ``running`` state indefinitely. No errors surfaced, the
    failure-notifier stayed silent. Other scopes never got a chance
    to run. This test pins the contract that prevents that class of
    incident.
    """
    import asyncio as _asyncio

    from app.domains.cmdb import cost_service as cost_service_mod
    from app.domains.cmdb.models import SCOPE_TYPE_AZURE_SUBSCRIPTION

    integ_id = str(cost_integration.id)
    # Two scopes — first hangs, second must still complete.
    # ``scope_type=SCOPE_TYPE_AZURE_SUBSCRIPTION`` chosen only because
    # the test fixture's ``cost_integration`` is Azure; the contract
    # under test (asyncio.wait_for around driver.list_cost_facts) is
    # provider-agnostic and would behave identically for AWS / Aliyun
    # / GCP scopes that hit the same per-scope iteration loop in
    # ``sync_cost_facts_for_integration``.
    hung_scope = CmdbCloudScope(
        id=str(uuid.uuid4()),
        integration_id=integ_id,
        scope_type=SCOPE_TYPE_AZURE_SUBSCRIPTION,
        external_id="00000000-0000-0000-0000-0000000000aa",
        display_name="Hung Scope",
        status="active",
    )
    fast_scope = CmdbCloudScope(
        id=str(uuid.uuid4()),
        integration_id=integ_id,
        scope_type=SCOPE_TYPE_AZURE_SUBSCRIPTION,
        external_id="00000000-0000-0000-0000-0000000000bb",
        display_name="Fast Scope",
        status="active",
    )
    db_session.add_all([hung_scope, fast_scope])
    await db_session.commit()

    class _PartialHangDriver:
        provider = "azure_cloud"
        requires_cny_fx = False

        async def list_cost_facts(
            self, start_date, end_date, *,
            scope=None, cny_per_usd=None,
            principal_tag_keys=None, bq_billing=None,
        ):
            # First scope hangs; second returns quickly.
            if scope is None or scope.external_id.endswith("aa"):
                await _asyncio.sleep(60)  # exceeds patched timeout
                return []
            return []

        async def close(self) -> None:  # pragma: no cover — trivial
            return None

    async def _fake_loader(_session, _integration):
        return _PartialHangDriver()

    # Shrink the per-scope timeout to ~0.2s so the hang fires
    # within the test window. Prod default is 300s; patched here
    # via the module-level ``_COST_FACTS_PER_SCOPE_TIMEOUT_SEC``.
    with patch(
        "app.domains.cmdb.lifecycle_service._load_cloud_driver",
        new=_fake_loader,
    ), patch.object(
        cost_service_mod, "_COST_FACTS_PER_SCOPE_TIMEOUT_SEC", 0.2,
    ):
        summary = await cost_service_mod.sync_cost_facts_for_integration(
            integ_id,
            start_date=date(2026, 5, 1),
            end_date=date(2026, 5, 1),
        )

    # Both scopes attempted; one timeout, one ok.
    per_scope = summary.get("per_scope") or []
    assert len(per_scope) == 2, (
        f"expected both scopes attempted, got {len(per_scope)}: {per_scope}"
    )
    statuses = sorted(s.get("status") for s in per_scope)
    assert statuses == ["ok", "timeout"], (
        f"expected [ok, timeout], got {statuses}: {per_scope}"
    )
    # Overall job status — the timeout pushes a string to errors[]
    # and the fast scope returns []  (no upserted facts), so the
    # service returns ``status='failed'`` (the partial_failure branch
    # requires at least one upsert AND at least one error). Pin the
    # exact value rather than ``in (…)`` so a future code refactor
    # that changes the failure-classification heuristic surfaces here.
    assert summary.get("status") == "failed", (
        f"expected 'failed' when timeout fires + no facts upserted, "
        f"got {summary.get('status')}: {summary}"
    )
    # Hung scope's per-scope entry must carry ``status='timeout'`` and
    # a ``error`` mentioning the timeout — operator UI surfaces this
    # to call out the specific scope that wedged.
    hung_entry = next(
        s for s in per_scope if s.get("status") == "timeout"
    )
    assert "timeout" in hung_entry.get("error", "").lower()


@pytest.mark.asyncio
async def test_sync_cost_facts_renders_empty_str_exception_with_type(
    db_session: AsyncSession,
    cost_integration: Integration,
    patched_session_factory,
):
    """A driver-side exception whose ``str(exc)`` is empty (httpx
    transport-level errors — ReadTimeout / RemoteProtocolError /
    some ConnectError paths often raise with no message) MUST still
    produce a meaningful per-scope error string with the exception
    class name. Without this, the failure-notifier message shows
    ``<scope_id>: `` with nothing after the colon and the operator
    is blind to the actual exception type.

    Pinning the 2026-05-21 prod incident: Azure subscription
    7bc8ab17-... cost_facts_sync produced exactly that empty-detail
    error, leaving ops unable to tell whether it was a timeout, a
    transport stall, an auth blip, or a 5xx. Same pattern was
    already used correctly in bailian_handlers.py — parity fix.
    """
    from app.domains.cmdb import cost_service as cost_service_mod
    from app.domains.cmdb.models import SCOPE_TYPE_AZURE_SUBSCRIPTION

    integ_id = str(cost_integration.id)
    scope_row = CmdbCloudScope(
        id=str(uuid.uuid4()),
        integration_id=integ_id,
        scope_type=SCOPE_TYPE_AZURE_SUBSCRIPTION,
        external_id="00000000-0000-0000-0000-0000000000cc",
        display_name="Empty Exc Scope",
        status="active",
    )
    db_session.add(scope_row)
    await db_session.commit()

    class _SilentTimeout(Exception):
        """Mimics httpx transport errors that ``str()`` to empty."""

        def __str__(self) -> str:  # noqa: D401
            return ""

    class _EmptyExcDriver:
        provider = "azure_cloud"
        requires_cny_fx = False

        async def list_cost_facts(
            self, start_date, end_date, *,
            scope=None, cny_per_usd=None,
            principal_tag_keys=None, bq_billing=None,
        ):
            raise _SilentTimeout()

        async def close(self) -> None:  # pragma: no cover — trivial
            return None

    async def _fake_loader(_session, _integration):
        return _EmptyExcDriver()

    with patch(
        "app.domains.cmdb.lifecycle_service._load_cloud_driver",
        new=_fake_loader,
    ):
        summary = await cost_service_mod.sync_cost_facts_for_integration(
            integ_id,
            start_date=date(2026, 5, 1),
            end_date=date(2026, 5, 1),
        )

    errors = summary.get("errors") or []
    assert errors, f"expected at least one error entry: {summary}"
    # Detail MUST include the class name; before the fix this was
    # just "<scope_id>: " with empty payload.
    joined = "; ".join(errors)
    assert "_SilentTimeout" in joined, (
        f"expected exception class in error string, got {joined!r}"
    )
    # The trailing detail must not be just whitespace / empty.
    assert not joined.rstrip().endswith(":"), (
        f"error string still ends with bare colon — empty exc not "
        f"rendered: {joined!r}"
    )
    # per_scope entry mirrors the same rendering.
    per_scope = summary.get("per_scope") or []
    failed_entry = next(
        (s for s in per_scope if s.get("status") == "failed"),
        None,
    )
    assert failed_entry is not None, (
        f"expected a per_scope[status=failed] entry: {per_scope}"
    )
    assert "_SilentTimeout" in (failed_entry.get("error") or ""), (
        f"per_scope error missing exc class: {failed_entry}"
    )


@pytest.mark.asyncio
async def test_breakdown_scope_id_isolates_one_subscription(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """``scope_id=`` narrows breakdown to ONE subscription's facts.

    Two Azure subscriptions under the same integration; two Foundry-
    style resources live under sub-A and one under sub-B. Filtering by
    sub-A's scope_id + group_by=resource_id must return only sub-A's
    two resources and a total equal to their sum — sub-B's spend is
    excluded entirely (this is the per-Foundry drilldown path)."""
    integ_id = str(cost_integration.id)
    sub_a = CmdbCloudScope(
        integration_id=integ_id, scope_type="azure_subscription",
        external_id="sub-a", display_name="Sub A",
    )
    sub_b = CmdbCloudScope(
        integration_id=integ_id, scope_type="azure_subscription",
        external_id="sub-b", display_name="Sub B",
    )
    db_session.add_all([sub_a, sub_b])
    await db_session.commit()
    await db_session.refresh(sub_a)
    await db_session.refresh(sub_b)

    def _foundry_fact(sub_ext: str, resource_id: str, usd: float) -> CostFact:
        return CostFact(
            date=date(2026, 5, 1), integration_id=integ_id,
            service="Foundry Models", amount_usd=usd, amount_native=usd,
            currency="USD",
            scope=CloudScopeRef(
                scope_type="azure_subscription", external_id=sub_ext,
            ),
            resource_group="rg", resource_id=resource_id,
            resource_name=resource_id, resource_type="account",
            region="eastus2", usage_type="tokens",
        )

    await ingest_cost_facts(
        db_session, integration_id=integ_id,
        facts=[
            _foundry_fact("sub-a", "tidbdocstrans-resource", 30.0),
            _foundry_fact("sub-a", "tidbclaw-resource", 12.0),
            _foundry_fact("sub-b", "other-resource", 99.0),
        ],
        scope_lookup={
            "sub-a": str(sub_a.id),
            "sub-b": str(sub_b.id),
        },
    )

    out = await breakdown(
        db_session,
        start=date(2026, 5, 1), end=date(2026, 5, 1),
        group_by="resource_id",
        integration_id=integ_id,
        scope_id=str(sub_a.id),
        top=10,
    )
    keys = {g["key"] for g in out["groups"]}
    assert keys == {"tidbdocstrans-resource", "tidbclaw-resource"}
    assert out["total_usd"] == pytest.approx(42.0)
    # Sub-B's $99 resource must NOT leak into the sub-A view.
    assert "other-resource" not in keys


@pytest.mark.asyncio
async def test_breakdown_unknown_scope_id_raises(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """An unknown ``scope_id`` raises ValueError (router maps to 400)
    rather than silently widening to the whole fleet."""
    with pytest.raises(ValueError, match="unknown scope_id"):
        await breakdown(
            db_session,
            start=date(2026, 5, 1), end=date(2026, 5, 1),
            group_by="resource_id",
            scope_id=str(uuid.uuid4()),
            top=10,
        )


@pytest.mark.asyncio
async def test_timeseries_scope_id_isolates_one_subscription(
    db_session: AsyncSession,
    cost_integration: Integration,
):
    """``scope_id=`` on the daily series mirrors breakdown's routing:
    only the filtered subscription's spend appears in the totals."""
    integ_id = str(cost_integration.id)
    sub_a = CmdbCloudScope(
        integration_id=integ_id, scope_type="azure_subscription",
        external_id="ts-sub-a", display_name="TS Sub A",
    )
    sub_b = CmdbCloudScope(
        integration_id=integ_id, scope_type="azure_subscription",
        external_id="ts-sub-b", display_name="TS Sub B",
    )
    db_session.add_all([sub_a, sub_b])
    await db_session.commit()
    await db_session.refresh(sub_a)
    await db_session.refresh(sub_b)

    await ingest_cost_facts(
        db_session, integration_id=integ_id,
        facts=[
            CostFact(
                date=date(2026, 5, 1), integration_id=integ_id,
                service="Foundry Models", amount_usd=25.0,
                amount_native=25.0, currency="USD",
                scope=CloudScopeRef(
                    scope_type="azure_subscription", external_id="ts-sub-a",
                ),
                resource_id="r-a", resource_name="r-a",
            ),
            CostFact(
                date=date(2026, 5, 1), integration_id=integ_id,
                service="Foundry Models", amount_usd=77.0,
                amount_native=77.0, currency="USD",
                scope=CloudScopeRef(
                    scope_type="azure_subscription", external_id="ts-sub-b",
                ),
                resource_id="r-b", resource_name="r-b",
            ),
        ],
        scope_lookup={
            "ts-sub-a": str(sub_a.id),
            "ts-sub-b": str(sub_b.id),
        },
    )

    out = await timeseries(
        db_session,
        start=date(2026, 5, 1), end=date(2026, 5, 1),
        group_by="resource_id",
        integration_id=integ_id,
        scope_id=str(sub_a.id),
        top=10,
    )
    assert out["total_per_day"] == [pytest.approx(25.0)]