"""Tests for ``ai_spending.cny_rate_refresh`` cron handler.

The cron drives the platform-wide CNY/USD rate that powers BOTH
AI-spending USD totals and cloud-cost USD totals. Two regression
classes pinned here:

1. **Source migration** — ``exchangerate.host`` went paid in 2026-Q2
   (returns 401 ``missing_access_key``), silently breaking refresh
   for ~11 days before anyone noticed because the handler returned
   ``refreshed=False`` (a non-failure status). New source is
   frankfurter.dev (ECB-published, free, no auth, same response
   shape). Pin: hitting the wrong URL would mean we missed the fix.

2. **Stale-FX hard-fail escalation** — a single transient outage
   soft-skips (back-off, retry next week), but if the stored rate
   is older than ``_RATE_STALE_AFTER_DAYS`` (14 = two missed
   weekly runs) the handler raises so failure_notifier alarms.
   Without this, a sustained upstream outage would silently drift
   AI-spending and cloud-cost USD numbers further apart from
   reality every week with no signal to the operator.
"""
from __future__ import annotations

from datetime import UTC, datetime, timedelta

import httpx
import pytest
import respx
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.ai_spending.service import (
    SETTING_CNY_RATE,
    SETTING_CNY_RATE_REFRESH_AT,
)
from app.integrations.models import PlatformSetting
from app.scheduled_jobs.ai_spending_handlers import (
    _FX_SOURCE_NAME,
    _FX_SOURCE_URL,
    _RATE_STALE_AFTER_DAYS,
    run_ai_spending_cny_rate_refresh,
)

# ── source pin ────────────────────────────────────────────────────


def test_fx_source_url_is_frankfurter_not_exchangerate_host() -> None:
    """exchangerate.host is paid since 2026-Q2 — pin frankfurter.dev
    so a future refactor doesn't accidentally regress the source."""
    assert "frankfurter.dev" in _FX_SOURCE_URL
    assert "exchangerate.host" not in _FX_SOURCE_URL
    assert _FX_SOURCE_NAME == "frankfurter.dev"
    # Must request CNY base + USD symbol, otherwise rates.USD won't
    # be in the response.
    assert "base=CNY" in _FX_SOURCE_URL
    assert "symbols=USD" in _FX_SOURCE_URL


# ── happy path ────────────────────────────────────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_refresh_persists_rate_from_frankfurter(
    db_session: AsyncSession,
) -> None:
    respx.get(_FX_SOURCE_URL).mock(return_value=httpx.Response(
        200,
        json={"amount": 1.0, "base": "CNY", "date": "2026-05-08",
              "rates": {"USD": 0.139}},
    ))
    result = await run_ai_spending_cny_rate_refresh(db_session, {})
    assert result["refreshed"] is True
    assert result["cny_to_usd_rate"] == pytest.approx(0.139)
    assert result["source"] == "frankfurter.dev"

    row = await db_session.get(PlatformSetting, SETTING_CNY_RATE)
    assert row is not None
    assert float(row.value) == pytest.approx(0.139)
    ts_row = await db_session.get(
        PlatformSetting, SETTING_CNY_RATE_REFRESH_AT,
    )
    assert ts_row is not None
    assert ts_row.value is not None  # populated, not JSON null


# ── soft-skip vs hard-fail decision ───────────────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_refresh_soft_skips_when_rate_is_recent(
    db_session: AsyncSession,
) -> None:
    """Source 5xx + recent stored rate → soft-skip, no exception.
    A single transient blip shouldn't page anyone."""
    # Seed a recent rate row.
    fresh = PlatformSetting(key=SETTING_CNY_RATE, value=0.14)
    db_session.add(fresh)
    await db_session.flush()

    respx.get(_FX_SOURCE_URL).mock(
        return_value=httpx.Response(503, text="Service Unavailable"),
    )
    result = await run_ai_spending_cny_rate_refresh(db_session, {})
    assert result["refreshed"] is False
    assert "skipped_reason" in result
    assert result["stale_days"] >= 0
    assert result["stale_threshold_days"] == _RATE_STALE_AFTER_DAYS


@pytest.mark.asyncio
@respx.mock
async def test_refresh_hard_fails_when_rate_is_stale(
    db_session: AsyncSession,
) -> None:
    """Source down + rate older than threshold → raise. Silent
    weekly drift on a value that drives every USD report is a real
    integrity issue."""
    # Seed a stale rate row by manually backdating updated_at.
    stale_when = datetime.now(UTC) - timedelta(
        days=_RATE_STALE_AFTER_DAYS + 1,
    )
    stale = PlatformSetting(key=SETTING_CNY_RATE, value=0.14)
    db_session.add(stale)
    await db_session.flush()
    # Force-update the row's updated_at via raw SQL — SQLAlchemy
    # ``onupdate`` would re-set it to now() on any attribute write.
    from sqlalchemy import text
    await db_session.execute(
        text(
            "UPDATE platform_settings SET updated_at = :ts "
            "WHERE `key` = :k",
        ),
        {"ts": stale_when, "k": SETTING_CNY_RATE},
    )
    await db_session.flush()
    # Expire the session-identity-map copy so the handler's
    # ``db.get`` re-fetches the row with the backdated updated_at.
    db_session.expire(stale)

    respx.get(_FX_SOURCE_URL).mock(
        return_value=httpx.Response(503, text="Service Unavailable"),
    )
    with pytest.raises(RuntimeError, match="stale"):
        await run_ai_spending_cny_rate_refresh(db_session, {})


@pytest.mark.asyncio
@respx.mock
async def test_refresh_hard_fails_when_no_existing_rate(
    db_session: AsyncSession,
) -> None:
    """No prior rate stored AND source failed → hard-fail. A fresh
    install with the migration not yet applied would land here;
    operator needs to know immediately."""
    # Make sure no row exists.
    existing = await db_session.get(PlatformSetting, SETTING_CNY_RATE)
    if existing is not None:
        await db_session.delete(existing)
        await db_session.flush()

    respx.get(_FX_SOURCE_URL).mock(
        return_value=httpx.Response(500, text="boom"),
    )
    with pytest.raises(RuntimeError, match="no existing rate"):
        await run_ai_spending_cny_rate_refresh(db_session, {})


# ── parse / plausibility ──────────────────────────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_refresh_skips_on_missing_rates_usd_field(
    db_session: AsyncSession,
) -> None:
    """If frankfurter ever changes shape and ``rates.USD`` is gone,
    the same staleness-aware soft-skip-or-hard-fail logic applies."""
    fresh = PlatformSetting(key=SETTING_CNY_RATE, value=0.14)
    db_session.add(fresh)
    await db_session.flush()

    respx.get(_FX_SOURCE_URL).mock(return_value=httpx.Response(
        200, json={"rates": {}},  # USD missing
    ))
    result = await run_ai_spending_cny_rate_refresh(db_session, {})
    assert result["refreshed"] is False
    assert "rates.USD" in result["skipped_reason"]


@pytest.mark.asyncio
@respx.mock
async def test_refresh_rejects_value_outside_plausibility_band(
    db_session: AsyncSession,
) -> None:
    """A hijacked source pushing 14.0 (CNY-per-USD by mistake) is 100×
    above the band — must NOT overwrite the stored value."""
    fresh = PlatformSetting(key=SETTING_CNY_RATE, value=0.14)
    db_session.add(fresh)
    await db_session.flush()

    respx.get(_FX_SOURCE_URL).mock(return_value=httpx.Response(
        200, json={"rates": {"USD": 14.0}},
    ))
    result = await run_ai_spending_cny_rate_refresh(db_session, {})
    assert result["refreshed"] is False
    assert "plausibility band" in result["skipped_reason"]
    # Stored value untouched.
    row = await db_session.get(PlatformSetting, SETTING_CNY_RATE)
    assert float(row.value) == pytest.approx(0.14)


@pytest.mark.asyncio
@respx.mock
async def test_refresh_skips_on_non_numeric_usd_field(
    db_session: AsyncSession,
) -> None:
    fresh = PlatformSetting(key=SETTING_CNY_RATE, value=0.14)
    db_session.add(fresh)
    await db_session.flush()

    respx.get(_FX_SOURCE_URL).mock(return_value=httpx.Response(
        200, json={"rates": {"USD": "not-a-number"}},
    ))
    result = await run_ai_spending_cny_rate_refresh(db_session, {})
    assert result["refreshed"] is False
    assert "not a number" in result["skipped_reason"]


# ── flag_modified regression — same-value re-refresh must tick `updated_at` ──


@pytest.mark.asyncio
@respx.mock
async def test_refresh_ticks_updated_at_even_when_rate_unchanged(
    db_session: AsyncSession,
) -> None:
    """ECB rates can be unchanged across consecutive Mondays
    (especially across CN holidays). When SQLAlchemy's JSON-column
    dirty-check sees ``row.value == new_value`` it skips the UPDATE
    → ``updated_at`` doesn't tick → the staleness clock keeps
    counting from whenever the value last actually changed → the
    next failed fetch escalates to a spurious hard-fail. The cron
    handler uses ``flag_modified`` to force the dirty marker so
    every successful refresh ticks ``updated_at``. Pin it."""
    from sqlalchemy import text

    # Seed at the value frankfurter will return.
    seed = PlatformSetting(key=SETTING_CNY_RATE, value=0.139)
    db_session.add(seed)
    await db_session.flush()
    # Backdate updated_at well into the staleness window so we can
    # observe it tick forward to "now" after the equal-value refresh.
    backdated = datetime.now(UTC) - timedelta(
        days=_RATE_STALE_AFTER_DAYS - 1,
    )
    await db_session.execute(
        text(
            "UPDATE platform_settings SET updated_at = :ts "
            "WHERE `key` = :k",
        ),
        {"ts": backdated, "k": SETTING_CNY_RATE},
    )
    await db_session.flush()
    db_session.expire(seed)

    respx.get(_FX_SOURCE_URL).mock(return_value=httpx.Response(
        200, json={"rates": {"USD": 0.139}},  # identical to seed
    ))
    result = await run_ai_spending_cny_rate_refresh(db_session, {})
    assert result["refreshed"] is True

    refreshed = await db_session.get(PlatformSetting, SETTING_CNY_RATE)
    assert refreshed is not None
    new_updated_at = refreshed.updated_at
    if new_updated_at.tzinfo is None:
        new_updated_at = new_updated_at.replace(tzinfo=UTC)
    # Must have ticked forward — not still at backdated time.
    assert (
        datetime.now(UTC) - new_updated_at
    ) < timedelta(minutes=5), (
        "updated_at did not tick on equal-value refresh; "
        "flag_modified missing → staleness check would later raise "
        "a spurious hard-fail."
    )