"""Tests for ``run_billing_sync`` BQ-mode aggregation and error propagation.

Two regressions land together here:

1. **Status propagation** — every failure path in the original handler
   ``return``ed ``{"status": "failed", ...}`` instead of raising. The
   scheduled_job runner only marks ``scheduled_job_runs.status='failed'``
   when the handler raises, so credential errors / Cost Explorer
   AccessDenied / provider missing the method all landed as
   ``status='ok'`` with the truth buried in ``result_summary``. PingCAP
   prod aws-primary silently swallowed Cost Explorer AccessDenied for
   months (2026-05). Pin: every failure path now raises ``RuntimeError``.

2. **AWS billing_source=bigquery** — the AWS provider's
   ``get_monthly_billing`` always calls Cost Explorer regardless of the
   integration's ``billing_source``. Tenants on BQ CUR (PingCAP prod
   aws-primary) intentionally don't grant ``ce:GetCostAndUsage`` →
   every monthly run failed with AccessDenied. Pin: when
   ``provider='aws'`` AND ``config_public.billing_source == 'bigquery'``,
   the handler aggregates ``cmdb_cloud_billing_daily`` rows for the
   target month instead of calling the provider — same upstream feed
   (BQ CUR via ``cmdb.billing_daily_sync``), no second BQ scan, no CE
   permission needed.
"""
from __future__ import annotations

import uuid
from contextlib import asynccontextmanager
from datetime import date
from decimal import Decimal
from unittest.mock import AsyncMock, patch

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.models import (
    CmdbBillingMonthly,
    CmdbCloudBillingDaily,
    CmdbCloudScope,
)
from app.integrations.models import Integration


async def _create_cloud_integration(
    db: AsyncSession,
    *,
    provider: str = "aws",
    config_public: dict | None = None,
) -> Integration:
    row = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider=provider,
        name=f"{provider}-test",
        slug=f"{provider}-test-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["billing"],
        config_public=config_public or {},
    )
    db.add(row)
    await db.commit()
    return row


class _SessionFactoryAndAlertPatches:
    """Context manager bundling two patches that always go together.

    1. ``app.core.database.async_session_factory`` → yield the test's
       ``db_session`` so the handler's ``async with`` reuses it.
    2. ``maybe_send_budget_alert`` → no-op AsyncMock.

    Why bundle: the handler's tail-call to ``maybe_send_budget_alert``
    opens *nested* ``async with async_session_factory()`` blocks. Under
    our patched factory those nested blocks reuse ``db_session``
    directly (no real cleanup on context exit). With MySQL 8.0 + asyncmy
    this left a connection sleeping with metadata locks held — the
    next test's ``TRUNCATE`` then blocks indefinitely (CI hit the 6h
    workflow timeout, local run hung waiting for human intervention).
    The alert pipeline has its own coverage in ``test_billing_alerts``;
    here we only verify the upsert + status-propagation contract.
    """

    def __init__(self, db_session: AsyncSession):
        @asynccontextmanager
        async def _factory():
            yield db_session
        self._patches = [
            patch("app.core.database.async_session_factory", new=_factory),
            patch(
                "app.domains.cmdb.billing_alerts.maybe_send_budget_alert",
                new=AsyncMock(return_value={"status": "no-budget-configured"}),
            ),
        ]

    def __enter__(self):
        for p in self._patches:
            p.start()
        return self

    def __exit__(self, *exc):
        for p in reversed(self._patches):
            p.stop()
        return False


def _patch_session_factory(db_session: AsyncSession) -> _SessionFactoryAndAlertPatches:
    return _SessionFactoryAndAlertPatches(db_session)


@pytest.mark.asyncio
async def test_aws_bq_aggregates_from_daily_not_cost_explorer(
    db_session: AsyncSession,
) -> None:
    """AWS + billing_source=bigquery: handler must aggregate from
    ``cmdb_cloud_billing_daily`` and never touch the provider /
    Cost Explorer. Verifies the row written to ``cmdb_billing_monthly``
    matches the sum of daily rows for that month."""
    from app.scheduled_jobs.handlers import run_billing_sync

    integration = await _create_cloud_integration(
        db_session,
        provider="aws",
        config_public={"billing_source": "bigquery"},
    )
    month = date(2026, 5, 1)
    scope = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integration.id),
        scope_type="aws_account",
        external_id="000000000001",
        display_name="000000000001",
        status="active",
    )
    db_session.add(scope)
    await db_session.commit()
    # Seed three daily rows in the target month for this integration.
    for d, usd, svcs in [
        (date(2026, 5, 12), Decimal("21450.89"), {"EC2": 10000.0, "S3": 11450.89}),
        (date(2026, 5, 13), Decimal("19802.67"), {"EC2": 9000.0, "S3": 10802.67}),
        (date(2026, 5, 14), Decimal("8920.58"), {"EC2": 4000.0, "S3": 4920.58}),
    ]:
        db_session.add(CmdbCloudBillingDaily(
            id=str(uuid.uuid4()),
            integration_id=str(integration.id),
            scope_id=str(scope.id),
            linked_account_id="000000000001",
            date=d,
            total_cost_usd=usd,
            cost_by_service=svcs,
            currency="USD",
            raw_response={},
        ))
    # And one row OUTSIDE the target month — must not contribute.
    db_session.add(CmdbCloudBillingDaily(
        id=str(uuid.uuid4()),
        integration_id=str(integration.id),
        scope_id=str(scope.id),
        linked_account_id="000000000001",
        date=date(2026, 4, 30),
        total_cost_usd=Decimal("999999.99"),
        cost_by_service={"EC2": 999999.99},
        currency="USD",
        raw_response={},
    ))
    await db_session.commit()

    # If the BQ short-circuit works, the provider class is never even
    # constructed — patch ``get_provider_class`` to a sentinel that
    # would explode if called.
    with _patch_session_factory(db_session), patch(
        "app.domains.cmdb.providers.base.get_provider_class",
        side_effect=AssertionError("provider must not be loaded in BQ mode"),
    ):
        result = await run_billing_sync(
            db_session,
            {"integration_id": str(integration.id), "month": "2026-05-01"},
        )

    assert result["month"] == "2026-05-01"
    # 21450.89 + 19802.67 + 8920.58 = 50174.14
    assert result["total_cost_usd"] == pytest.approx(50174.14, abs=0.01)
    # Verify the row landed in cmdb_billing_monthly.
    from sqlalchemy import select
    row = (await db_session.execute(
        select(CmdbBillingMonthly).where(
            CmdbBillingMonthly.integration_id == str(integration.id),
            CmdbBillingMonthly.month == month,
        )
    )).scalar_one()
    assert float(row.total_cost_usd) == pytest.approx(50174.14, abs=0.01)
    # cost_by_service merges across days.
    assert row.cost_by_service["EC2"] == pytest.approx(23000.0, abs=0.01)
    assert row.cost_by_service["S3"] == pytest.approx(27174.14, abs=0.01)


@pytest.mark.asyncio
async def test_aws_cost_explorer_mode_still_calls_provider(
    db_session: AsyncSession,
) -> None:
    """AWS WITHOUT billing_source=bigquery (i.e. on Cost Explorer)
    must still flow through the provider — the BQ short-circuit is
    opt-in by config, not default."""
    from app.scheduled_jobs.handlers import run_billing_sync

    integration = await _create_cloud_integration(
        db_session,
        provider="aws",
        config_public={"billing_source": "cost_explorer"},
    )

    fake_provider = AsyncMock()
    fake_provider.get_monthly_billing = AsyncMock(return_value={
        "month": date(2026, 5, 1),
        "total_cost_usd": 1234.56,
        "cost_by_service": {"EC2": 1234.56},
        "currency": "USD",
        "raw_response": {},
    })
    fake_provider.close = AsyncMock()

    with _patch_session_factory(db_session), patch(
        "app.domains.cmdb.providers.base.get_provider_class",
        return_value=lambda creds: fake_provider,
    ), patch(
        "app.domains.cmdb.credential_crypto.decrypt_fields",
        return_value={},
    ):
        result = await run_billing_sync(
            db_session,
            {"integration_id": str(integration.id), "month": "2026-05-01"},
        )
    assert result["total_cost_usd"] == pytest.approx(1234.56)
    fake_provider.get_monthly_billing.assert_awaited_once()


@pytest.mark.asyncio
async def test_handler_raises_on_provider_failure(
    db_session: AsyncSession,
) -> None:
    """Provider's ``get_monthly_billing`` raising (e.g. CE AccessDenied
    on a misconfigured tenant) must propagate as ``RuntimeError`` so
    ``scheduled_job_runs.status='failed'``. Pre-fix the handler
    swallowed it into a return-dict and the run landed ``status='ok'``.
    """
    from app.scheduled_jobs.handlers import run_billing_sync

    integration = await _create_cloud_integration(
        db_session,
        provider="aws",
        config_public={"billing_source": "cost_explorer"},
    )

    fake_provider = AsyncMock()
    fake_provider.get_monthly_billing = AsyncMock(
        side_effect=Exception(
            "Cost Explorer error: AccessDeniedException — "
            "user not authorized for ce:GetCostAndUsage",
        ),
    )
    fake_provider.close = AsyncMock()

    with _patch_session_factory(db_session), patch(
        "app.domains.cmdb.providers.base.get_provider_class",
        return_value=lambda creds: fake_provider,
    ), patch(
        "app.domains.cmdb.credential_crypto.decrypt_fields",
        return_value={},
    ):
        with pytest.raises(RuntimeError, match="AccessDenied"):
            await run_billing_sync(
                db_session,
                {"integration_id": str(integration.id), "month": "2026-05-01"},
            )


@pytest.mark.asyncio
async def test_handler_raises_when_provider_missing_method(
    db_session: AsyncSession,
) -> None:
    """A provider class that doesn't implement ``get_monthly_billing``
    is a pipeline regression — must raise, not silently return failed."""
    from app.scheduled_jobs.handlers import run_billing_sync

    integration = await _create_cloud_integration(
        db_session,
        provider="aws",
        config_public={"billing_source": "cost_explorer"},
    )

    class _BrokenProvider:
        def __init__(self, creds): pass
        async def close(self): pass
    # Note: no get_monthly_billing method.

    with _patch_session_factory(db_session), patch(
        "app.domains.cmdb.providers.base.get_provider_class",
        return_value=_BrokenProvider,
    ), patch(
        "app.domains.cmdb.credential_crypto.decrypt_fields",
        return_value={},
    ):
        with pytest.raises(RuntimeError, match="get_monthly_billing"):
            await run_billing_sync(
                db_session,
                {"integration_id": str(integration.id), "month": "2026-05-01"},
            )


@pytest.mark.asyncio
async def test_handler_skips_on_disabled_integration(
    db_session: AsyncSession,
) -> None:
    """Disabled integration: ``skipped`` is a legitimate non-failure
    outcome — must NOT raise (avoid noisy failure-notifier pings on
    intentionally-quiesced tenants)."""
    from app.scheduled_jobs.handlers import run_billing_sync

    integration = await _create_cloud_integration(
        db_session,
        provider="aws",
        config_public={"billing_source": "bigquery"},
    )
    integration.enabled = False
    await db_session.commit()

    with _patch_session_factory(db_session):
        result = await run_billing_sync(
            db_session,
            {"integration_id": str(integration.id), "month": "2026-05-01"},
        )
    assert result["status"] == "skipped"


@pytest.mark.asyncio
async def test_cny_fx_fallback_persists_native_and_skips_budget_alert(
    db_session: AsyncSession,
) -> None:
    """Round-666 regression: when a CNY tenant's FX rebase FALLS BACK
    (weekly FX cron stale → ``resolve_cny_per_usd`` raises), the monthly
    row MUST be persisted unit-consistently — native CNY value tagged
    ``currency='CNY'`` so the billing-overview reader converts it exactly
    ONCE — and the USD budget alert MUST be skipped. Pre-fix, the Aliyun
    driver pre-multiplied by a stale 0.14 and the fallback stored a
    USD-ish value tagged CNY (reader double-converted → ~7× understated)
    AND the alert compared a native CNY figure to the USD budget."""
    from sqlalchemy import select

    from app.scheduled_jobs.handlers import run_billing_sync

    integration = await _create_cloud_integration(
        db_session, provider="aliyun", config_public={},
    )

    fake_provider = AsyncMock()
    # Round-666 driver contract: total_cost_usd carries the NATIVE value.
    fake_provider.get_monthly_billing = AsyncMock(return_value={
        "month": date(2026, 5, 1),
        "total_cost_usd": 7000.0,
        "total_cost_native": 7000.0,
        "cost_by_service": {"ECS": 7000.0},
        "currency": "CNY",
        "raw_response": {},
    })
    fake_provider.close = AsyncMock()

    @asynccontextmanager
    async def _factory():
        yield db_session

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

    alert_mock = AsyncMock(return_value={"status": "exceeded-alerted"})

    with patch("app.core.database.async_session_factory", new=_factory), patch(
        "app.domains.cmdb.providers.base.get_provider_class",
        return_value=lambda creds: fake_provider,
    ), patch(
        "app.domains.cmdb.credential_crypto.decrypt_fields", return_value={},
    ), patch(
        "app.core.fx_rate.resolve_cny_per_usd", new=_fx_raise,
    ), patch(
        "app.domains.cmdb.billing_alerts.maybe_send_budget_alert", new=alert_mock,
    ):
        result = await run_billing_sync(
            db_session,
            {"integration_id": str(integration.id), "month": "2026-05-01"},
        )

    row = (await db_session.execute(select(CmdbBillingMonthly).where(
        CmdbBillingMonthly.integration_id == str(integration.id),
        CmdbBillingMonthly.month == date(2026, 5, 1),
    ))).scalar_one()
    # Native value + CNY label — value and label agree (no double-convert).
    assert row.currency == "CNY"
    assert float(row.total_cost_usd) == pytest.approx(7000.0)
    assert result["currency"] == "CNY"
    # Budget alert skipped: no false USD-vs-CNY comparison.
    alert_mock.assert_not_awaited()
    assert result.get("budget_alert_skipped")


@pytest.mark.asyncio
async def test_cny_fx_resolves_rebases_to_usd_and_fires_budget_alert(
    db_session: AsyncSession,
) -> None:
    """Round-666: the budget-alert currency gate must NOT suppress the
    legitimate path. When FX resolves, a CNY tenant's total is rebased
    to USD, the row is stored as USD, and the budget alert fires."""
    from sqlalchemy import select

    from app.scheduled_jobs.handlers import run_billing_sync

    integration = await _create_cloud_integration(
        db_session, provider="aliyun", config_public={},
    )

    fake_provider = AsyncMock()
    fake_provider.get_monthly_billing = AsyncMock(return_value={
        "month": date(2026, 5, 1),
        "total_cost_usd": 7000.0,
        "total_cost_native": 7000.0,
        "cost_by_service": {"ECS": 7000.0},
        "currency": "CNY",
        "raw_response": {},
    })
    fake_provider.close = AsyncMock()

    @asynccontextmanager
    async def _factory():
        yield db_session

    async def _fx_ok(*a, **k):
        return 7.0  # 1 USD = 7 CNY → 7000 CNY = 1000 USD

    alert_mock = AsyncMock(return_value={"status": "within-budget"})

    with patch("app.core.database.async_session_factory", new=_factory), patch(
        "app.domains.cmdb.providers.base.get_provider_class",
        return_value=lambda creds: fake_provider,
    ), patch(
        "app.domains.cmdb.credential_crypto.decrypt_fields", return_value={},
    ), patch(
        "app.core.fx_rate.resolve_cny_per_usd", new=_fx_ok,
    ), patch(
        "app.domains.cmdb.billing_alerts.maybe_send_budget_alert", new=alert_mock,
    ):
        result = await run_billing_sync(
            db_session,
            {"integration_id": str(integration.id), "month": "2026-05-01"},
        )

    row = (await db_session.execute(select(CmdbBillingMonthly).where(
        CmdbBillingMonthly.integration_id == str(integration.id),
        CmdbBillingMonthly.month == date(2026, 5, 1),
    ))).scalar_one()
    assert row.currency == "USD"
    assert float(row.total_cost_usd) == pytest.approx(1000.0, abs=0.01)
    assert result["currency"] == "USD"
    alert_mock.assert_awaited_once()
    assert "budget_alert_skipped" not in result


@pytest.mark.asyncio
async def test_aws_bq_with_no_daily_rows_writes_zero(
    db_session: AsyncSession,
) -> None:
    """AWS-BQ when no daily rows exist for the target month: aggregate
    yields $0 and writes it. This is honest behaviour — daily-sync
    hasn't filled the month yet (or there genuinely was no spend).
    Distinct from ``skipped`` because the alert pipeline must still
    evaluate $0 against the budget threshold."""
    from sqlalchemy import select

    from app.scheduled_jobs.handlers import run_billing_sync

    integration = await _create_cloud_integration(
        db_session,
        provider="aws",
        config_public={"billing_source": "bigquery"},
    )
    with _patch_session_factory(db_session):
        result = await run_billing_sync(
            db_session,
            {"integration_id": str(integration.id), "month": "2026-05-01"},
        )
    assert result["total_cost_usd"] == 0.0
    row = (await db_session.execute(
        select(CmdbBillingMonthly).where(
            CmdbBillingMonthly.integration_id == str(integration.id),
            CmdbBillingMonthly.month == date(2026, 5, 1),
        )
    )).scalar_one()
    assert float(row.total_cost_usd) == 0.0