"""Regression: the billing_daily reconciliation pass (unregistered-
account, scope_id=NULL rows) must preserve the driver's currency label,
not hard-code ``currency="USD"``.

CN clouds (Aliyun / Tencent / UCloud) write the NATIVE CNY amount into
the legacy ``total_cost_usd`` column and tag the row ``currency='CNY'``;
downstream ``_to_usd`` converts. Pre-fix the reconciliation pass wrote
``currency="USD"`` unconditionally, so an unregistered CN sub-account's
spend was mislabeled CNY-as-USD and inflated the admin billing-overview
total ~7×. This test drives ``sync_billing_daily_for_integration`` with
a fake Aliyun driver whose tenant-wide (``scope=None``) call returns a
CNY row for an account that has no ``cmdb_cloud_scopes`` row, and
asserts the reconciliation row carries ``currency='CNY'``.
"""
from __future__ import annotations

import uuid
from contextlib import asynccontextmanager
from datetime import UTC, datetime
from datetime import date as _date

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

from app.domains.cmdb import lifecycle_service
from app.domains.cmdb.models import (
    SCOPE_STATUS_ACTIVE,
    SCOPE_TYPE_ALIYUN_ACCOUNT,
    CmdbCloudBillingDaily,
    CmdbCloudScope,
)
from app.integrations.driver_base import CloudDailyCost, CloudScopeRef
from app.integrations.models import Integration

_REGISTERED = "acct-registered-001"
_UNREGISTERED = "acct-unregistered-999"
_DAY = _date(2026, 5, 20)


class _FakeAliyunDriver:
    """Duck-typed CloudDriver for the billing-daily sync path. Returns
    native CNY rows. ``scope=None`` (reconciliation) includes one row
    for an account with NO scope row → exercises the scope_id=NULL path.
    """

    requires_cny_fx = True

    async def billing_daily_precheck(self) -> str | None:
        return None

    async def list_billing_daily(
        self, start_date, end_date, *, scope=None,
        include_owner_breakdown=False, bq_billing=None,
    ) -> list[CloudDailyCost]:
        if scope is not None:
            # Per-scope call for the registered account.
            return [CloudDailyCost(
                date=_DAY, total_cost_usd=700.0,
                scope=CloudScopeRef(
                    scope_type=SCOPE_TYPE_ALIYUN_ACCOUNT,
                    external_id=scope.external_id,
                ),
                currency="CNY",
            )]
        # Tenant-wide reconciliation call: registered + unregistered.
        return [
            CloudDailyCost(
                date=_DAY, total_cost_usd=700.0,
                scope=CloudScopeRef(
                    scope_type=SCOPE_TYPE_ALIYUN_ACCOUNT,
                    external_id=_REGISTERED,
                ),
                currency="CNY",
            ),
            CloudDailyCost(
                date=_DAY, total_cost_usd=350.0,
                scope=CloudScopeRef(
                    scope_type=SCOPE_TYPE_ALIYUN_ACCOUNT,
                    external_id=_UNREGISTERED,
                ),
                currency="CNY",
            ),
        ]

    async def close(self) -> None:
        return None


@pytest.mark.asyncio
async def test_reconciliation_row_preserves_cny_currency(
    db_session: AsyncSession, monkeypatch,
) -> None:
    integration = Integration(
        id=uuid.uuid4(), kind="cloud", provider="aliyun",
        name="Aliyun Recon Test", slug=f"aliyun-{uuid.uuid4().hex[:6]}",
        enabled=True, is_primary=False, capabilities=["billing"],
    )
    db_session.add(integration)
    await db_session.commit()  # integration must exist before scope FK
    scope = CmdbCloudScope(
        id=uuid.uuid4(), integration_id=str(integration.id),
        scope_type=SCOPE_TYPE_ALIYUN_ACCOUNT, external_id=_REGISTERED,
        display_name="Registered", status=SCOPE_STATUS_ACTIVE,
        first_seen_at=datetime.now(UTC),
    )
    db_session.add(scope)
    await db_session.commit()

    async def _fake_loader(session, integ):
        return _FakeAliyunDriver()

    # The sync helper opens its OWN session via async_session_factory;
    # point it at the test session (same pattern as
    # test_grants_inactivity_and_drift) so it sees the committed rows.
    @asynccontextmanager
    async def _factory():
        yield db_session

    monkeypatch.setattr(lifecycle_service, "async_session_factory", _factory)
    monkeypatch.setattr(lifecycle_service, "_load_cloud_driver", _fake_loader)

    result = await lifecycle_service.sync_billing_daily_for_integration(
        str(integration.id),
        start_date=_DAY, end_date=_DAY,
        dispatch_alerts=False,
    )
    assert result["status"] in ("ok", "partial_failure"), result

    # The unregistered account lands as a scope_id=NULL reconciliation
    # row. It MUST keep currency='CNY' (pre-fix it was hard-coded 'USD').
    recon = (await db_session.execute(
        select(CmdbCloudBillingDaily).where(
            CmdbCloudBillingDaily.integration_id == str(integration.id),
            CmdbCloudBillingDaily.scope_id.is_(None),
            CmdbCloudBillingDaily.linked_account_id == _UNREGISTERED,
            CmdbCloudBillingDaily.date == _DAY,
        )
    )).scalars().all()
    assert len(recon) == 1, recon
    assert recon[0].currency == "CNY"
    assert float(recon[0].total_cost_usd) == 350.0


@pytest.mark.asyncio
async def test_reconciliation_deadlock_reraises_not_swallowed(
    db_session: AsyncSession, monkeypatch,
) -> None:
    """A transaction-aborting error (deadlock 1213 / lock-wait 1205) during
    the reconciliation DELETE+insert must RE-RAISE, not be swallowed into a
    ``partial_failure`` envelope.

    On TiDB a 1213/1205 rolls the WHOLE transaction back and destroys the
    savepoint, so swallowing it would only defer the failure to the
    ``session.commit()`` right after (which then raises on the dead txn and
    bypasses the graceful envelope) — and it can't be recovered in-place.
    Re-raising lets the cron's fresh-session retry handle it; per-scope
    progress is already committed and the reconciliation is idempotent.
    Same-class fix as the #1066 batch endpoints / #1055 HRIS advance.
    """
    from sqlalchemy import Delete
    from sqlalchemy.exc import OperationalError

    from app.core.exceptions import is_transaction_aborting_db_error
    from app.domains.cmdb.models import CmdbCloudBillingDaily

    integration = Integration(
        id=uuid.uuid4(), kind="cloud", provider="aliyun",
        name="Aliyun Deadlock Test", slug=f"aliyun-{uuid.uuid4().hex[:6]}",
        enabled=True, is_primary=False, capabilities=["billing"],
    )
    db_session.add(integration)
    await db_session.commit()
    scope = CmdbCloudScope(
        id=uuid.uuid4(), integration_id=str(integration.id),
        scope_type=SCOPE_TYPE_ALIYUN_ACCOUNT, external_id=_REGISTERED,
        display_name="Registered", status=SCOPE_STATUS_ACTIVE,
        first_seen_at=datetime.now(UTC),
    )
    db_session.add(scope)
    await db_session.commit()

    async def _fake_loader(session, integ):
        return _FakeAliyunDriver()

    @asynccontextmanager
    async def _factory():
        yield db_session

    monkeypatch.setattr(lifecycle_service, "async_session_factory", _factory)
    monkeypatch.setattr(lifecycle_service, "_load_cloud_driver", _fake_loader)

    # Simulate a deadlock on the reconciliation DELETE (scope_id=NULL rows).
    class _Orig(Exception):
        pass

    orig_execute = db_session.execute

    async def _execute_deadlock_on_recon_delete(stmt, *args, **kwargs):
        if (
            isinstance(stmt, Delete)
            and getattr(getattr(stmt, "table", None), "name", "")
            == CmdbCloudBillingDaily.__tablename__
        ):
            raise OperationalError(
                "DELETE FROM cmdb_cloud_billing_daily ...",
                {},
                _Orig(1213, "Deadlock found when trying to get lock"),
            )
        return await orig_execute(stmt, *args, **kwargs)

    monkeypatch.setattr(db_session, "execute", _execute_deadlock_on_recon_delete)

    # Pre-fix: swallowed → doomed commit → confusing failure / no retry.
    # Post-fix: the 1213 propagates out so the cron can retry on a fresh
    # session.
    with pytest.raises(OperationalError) as ei:
        await lifecycle_service.sync_billing_daily_for_integration(
            str(integration.id),
            start_date=_DAY, end_date=_DAY,
            dispatch_alerts=False,
        )
    assert is_transaction_aborting_db_error(ei.value)


class _FakeCnyDriverTodayOnly:
    """CNY driver returning ONE row dated _DAY for the per-scope call (so
    ``today_row`` is set but the past-7-day baseline is empty → the spike
    block's FX guard is SKIPPED), and the same registered account on the
    reconciliation (``scope=None``) call (no unregistered rows)."""

    requires_cny_fx = True

    async def billing_daily_precheck(self) -> str | None:
        return None

    async def list_billing_daily(
        self, start_date, end_date, *, scope=None,
        include_owner_breakdown=False, bq_billing=None,
    ):
        acct = scope.external_id if scope is not None else _REGISTERED
        return [CloudDailyCost(
            date=_DAY, total_cost_usd=700.0,
            scope=CloudScopeRef(
                scope_type=SCOPE_TYPE_ALIYUN_ACCOUNT, external_id=acct,
            ),
            currency="CNY",
        )]

    async def close(self) -> None:
        return None


@pytest.mark.asyncio
async def test_rules_block_fx_failure_does_not_abort_run(
    db_session: AsyncSession, monkeypatch,
) -> None:
    """Round-666 regression: a CNY scope with < 7 baseline days AND
    user-defined alert rules must NOT abort the whole billing_daily run
    when the FX rate is missing/stale.

    The spike block wraps ``_to_usd`` in ``try/except RuntimeError`` only
    when ``len(past) >= 7``; for a freshly-registered CN scope that guard
    is skipped, so the parallel user-defined-rules MTD path (which also
    calls ``_to_usd``) needs its own guard. Pre-fix the unguarded
    RuntimeError propagated out of the loop (the only enclosing handler
    is ``finally``: driver.close), skipping this scope's commit AND every
    later scope AND the reconciliation pass, and raising instead of
    returning the partial_failure envelope. Post-fix: the scope is
    skipped gracefully, its already-upserted daily row is committed, and
    the run lands ``partial_failure``.
    """
    integration = Integration(
        id=uuid.uuid4(), kind="cloud", provider="aliyun",
        name="Aliyun Rules FX Test", slug=f"aliyun-{uuid.uuid4().hex[:6]}",
        enabled=True, is_primary=False, capabilities=["billing"],
    )
    db_session.add(integration)
    await db_session.commit()
    scope = CmdbCloudScope(
        id=uuid.uuid4(), integration_id=str(integration.id),
        scope_type=SCOPE_TYPE_ALIYUN_ACCOUNT, external_id=_REGISTERED,
        display_name="Registered", status=SCOPE_STATUS_ACTIVE,
        first_seen_at=datetime.now(UTC),
        # A user-defined monthly rule → the rules block runs (and hits the
        # unguarded _to_usd pre-fix).
        alert_rules=[{"type": "monthly", "threshold_usd": 100.0, "active": True}],
    )
    db_session.add(scope)
    await db_session.commit()

    async def _fake_loader(session, integ):
        return _FakeCnyDriverTodayOnly()

    @asynccontextmanager
    async def _factory():
        yield db_session

    async def _fx_raise(*a, **k):
        raise RuntimeError("CNY FX rate missing or stale")

    monkeypatch.setattr(lifecycle_service, "async_session_factory", _factory)
    monkeypatch.setattr(lifecycle_service, "_load_cloud_driver", _fake_loader)
    # _to_usd resolves FX lazily via ``from app.core.fx_rate import
    # resolve_cny_per_usd`` — patch the module attribute so the in-function
    # import picks up the raising stub.
    import app.core.fx_rate as _fx
    monkeypatch.setattr(_fx, "resolve_cny_per_usd", _fx_raise)

    # MUST NOT raise.
    result = await lifecycle_service.sync_billing_daily_for_integration(
        str(integration.id),
        start_date=_DAY, end_date=_DAY,
        dispatch_alerts=True,
    )
    assert result["status"] == "partial_failure", result
    assert any(
        "rules-fx-conversion-failed" in e for e in result["errors"]
    ), result["errors"]

    # The scope's daily row was still upserted + committed (no data loss).
    rows = (await db_session.execute(
        select(CmdbCloudBillingDaily).where(
            CmdbCloudBillingDaily.scope_id == str(scope.id),
            CmdbCloudBillingDaily.date == _DAY,
        )
    )).scalars().all()
    assert len(rows) == 1
    assert float(rows[0].total_cost_usd) == 700.0