"""Tests for the ai_spending_period_rollup materialisation (PR3).

Pins the read-through contract:
* A **closed** month is served from the rollup without recomputing.
* The rollup is written only by ``refresh_period_rollup`` (the cron);
  a window is replaced wholesale (DELETE + INSERT).
* A fingerprint mismatch (admin changed the whitelist) is a miss →
  live fallback.
* The **open** month never touches the rollup.
* The refresh cron materialises recent closed months every run and
  skips older already-materialised ones (bounded steady-state work).
"""
from __future__ import annotations

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

import pytest
from sqlalchemy import func, select

from app.domains.ai_spending.models import AiSpendingPeriodRollup
from app.domains.ai_spending.service import (
    BUCKET_BAILIAN,
    AiSpendingService,
    BucketRow,
)

# A window that is unambiguously closed relative to any 2026-H1+ clock.
CLOSED_START = date(2026, 1, 1)
CLOSED_END = date(2026, 2, 1)


@pytest.fixture(autouse=True)
def _stub_fingerprint():
    """``_rollup_fingerprint`` reads the migration-149-seeded whitelist
    platform_setting, which the create_all test schema doesn't carry.
    These tests exercise rollup *mechanics*, not fingerprinting, so pin a
    constant. Tests that assert fingerprint-mismatch behaviour override
    this per-instance inside their own ``with patch.object`` blocks."""
    with patch.object(
        AiSpendingService, "_rollup_fingerprint",
        AsyncMock(return_value="FP-TEST"),
    ):
        yield


def _rows() -> list[BucketRow]:
    return [
        BucketRow(
            bucket=BUCKET_BAILIAN, date=date(2026, 1, 5),
            cost_usd=Decimal("100"), cost_usd_gross=Decimal("250"),
            aliyun_account_id="acc1", user_id="u1",
            raw={"input_tokens": 42},
        ),
        BucketRow(
            bucket=BUCKET_BAILIAN, date=date(2026, 1, 6),
            cost_usd=Decimal("50"), aliyun_account_id="acc1",
        ),
    ]


async def _window_count(db, start: date, end: date) -> int:
    return (await db.execute(
        select(func.count()).select_from(AiSpendingPeriodRollup).where(
            AiSpendingPeriodRollup.period_start == start,
            AiSpendingPeriodRollup.period_end_excl == end,
        )
    )).scalar() or 0


@pytest.mark.asyncio
async def test_refresh_writes_rows_with_fidelity(db_session):
    svc = AiSpendingService(db_session)
    with patch.object(
        svc, "_compute_period_rows", AsyncMock(return_value=_rows()),
    ):
        n = await svc.refresh_period_rollup(CLOSED_START, CLOSED_END)
    assert n == 2
    assert await _window_count(db_session, CLOSED_START, CLOSED_END) == 2
    # gross + raw round-trip preserved.
    fp = await svc._rollup_fingerprint()
    rows = await svc._read_rollup(CLOSED_START, CLOSED_END, fp)
    assert rows is not None
    by_cost = {r.cost_usd: r for r in rows}
    assert by_cost[Decimal("100")].cost_usd_gross == Decimal("250")
    assert by_cost[Decimal("100")].raw == {"input_tokens": 42}
    assert by_cost[Decimal("50")].cost_usd_gross is None


@pytest.mark.asyncio
async def test_closed_month_served_from_rollup_no_recompute(db_session):
    svc = AiSpendingService(db_session)
    comp = AsyncMock(return_value=_rows())
    with patch.object(svc, "_compute_period_rows", comp):
        await svc.refresh_period_rollup(CLOSED_START, CLOSED_END)
        assert comp.await_count == 1
        # Closed window → read hits the rollup, does NOT recompute.
        rows = await svc._load_period_rows(CLOSED_START, CLOSED_END)
        assert comp.await_count == 1
    assert len(rows) == 2
    assert {r.cost_usd for r in rows} == {Decimal("100"), Decimal("50")}


@pytest.mark.asyncio
async def test_fingerprint_mismatch_falls_back_to_live(db_session):
    svc = AiSpendingService(db_session)
    with patch.object(
        svc, "_compute_period_rows", AsyncMock(return_value=_rows()),
    ), patch.object(
        svc, "_rollup_fingerprint", AsyncMock(return_value="FP-OLD"),
    ):
        await svc.refresh_period_rollup(CLOSED_START, CLOSED_END)
    # Read with a DIFFERENT fingerprint (whitelist changed) → miss → live.
    comp2 = AsyncMock(return_value=_rows())
    with patch.object(svc, "_compute_period_rows", comp2), patch.object(
        svc, "_rollup_fingerprint", AsyncMock(return_value="FP-NEW"),
    ):
        rows = await svc._load_period_rows(CLOSED_START, CLOSED_END)
        assert comp2.await_count == 1  # recomputed live
    assert len(rows) == 2


@pytest.mark.asyncio
async def test_open_month_served_from_rollup_when_materialised(db_session):
    """Perf fix 2026-06: the OPEN/current month is now ALSO served from the
    rollup (cron-maintained, ≤1-day stale) instead of always live-scanning
    cmdb_cost_facts — the default dashboard view IS the current month. A
    rollup hit short-circuits the live compute."""
    svc = AiSpendingService(db_session)
    open_start, open_end = date(2099, 1, 1), date(2099, 2, 1)  # always open
    comp = AsyncMock(return_value=[])
    read = AsyncMock(return_value=[_rows()[0]])  # materialised hit
    with patch.object(svc, "_compute_period_rows", comp), patch.object(
        svc, "_read_rollup", read,
    ):
        rows = await svc._load_period_rows(open_start, open_end)
    read.assert_awaited()           # open month consults the rollup
    assert comp.await_count == 0    # served from rollup, NOT recomputed
    assert len(rows) == 1


@pytest.mark.asyncio
async def test_open_month_falls_back_to_live_on_miss(db_session):
    """Open month with no materialisation yet → live compute (no error)."""
    svc = AiSpendingService(db_session)
    open_start, open_end = date(2099, 1, 1), date(2099, 2, 1)
    comp = AsyncMock(return_value=[])
    with patch.object(svc, "_compute_period_rows", comp), patch.object(
        svc, "_read_rollup", AsyncMock(return_value=None),
    ):
        rows = await svc._load_period_rows(open_start, open_end)
    assert comp.await_count == 1
    assert rows == []


@pytest.mark.asyncio
async def test_refresh_replaces_window(db_session):
    svc = AiSpendingService(db_session)
    with patch.object(
        svc, "_compute_period_rows", AsyncMock(return_value=_rows()),
    ):
        await svc.refresh_period_rollup(CLOSED_START, CLOSED_END)
    assert await _window_count(db_session, CLOSED_START, CLOSED_END) == 2
    # Re-refresh with a single row — the window must be REPLACED, not appended.
    with patch.object(
        svc, "_compute_period_rows", AsyncMock(return_value=_rows()[:1]),
    ):
        await svc.refresh_period_rollup(CLOSED_START, CLOSED_END)
    assert await _window_count(db_session, CLOSED_START, CLOSED_END) == 1


@pytest.mark.asyncio
async def test_empty_month_materialises_as_hit_not_recomputed(db_session):
    """A closed month that computes to ZERO rows still caches (via the
    sentinel marker): read returns [] as a HIT (not a miss → live), and
    the cron's exists-check skips it on the next run instead of
    recomputing forever."""
    svc = AiSpendingService(db_session)
    comp = AsyncMock(return_value=[])
    with patch.object(svc, "_compute_period_rows", comp):
        n = await svc.refresh_period_rollup(CLOSED_START, CLOSED_END)
        assert n == 0
        fp = await svc._rollup_fingerprint()
        # Read is a HIT (empty list), NOT None → no live fallback.
        assert await svc._read_rollup(CLOSED_START, CLOSED_END, fp) == []
        # Window registers as existing → cron will skip it.
        assert await svc._rollup_window_exists(CLOSED_START, CLOSED_END, fp) is True
        # _load_period_rows serves the empty window without recomputing.
        rows = await svc._load_period_rows(CLOSED_START, CLOSED_END)
        assert rows == []
        assert comp.await_count == 1  # only the initial refresh computed


@pytest.mark.asyncio
async def test_prune_removes_aged_out_windows(db_session):
    svc = AiSpendingService(db_session)
    old_start, old_end = date(2025, 1, 1), date(2025, 2, 1)
    with patch.object(
        svc, "_compute_period_rows", AsyncMock(return_value=_rows()),
    ):
        await svc.refresh_period_rollup(old_start, old_end)
        await svc.refresh_period_rollup(CLOSED_START, CLOSED_END)  # 2026-01
    # Prune windows ending on/before 2026-01-01: drops the 2025-01 window,
    # keeps 2026-01 (ends 2026-02-01).
    pruned = await svc.prune_rollup_before(date(2026, 1, 1))
    assert pruned == 2
    assert await _window_count(db_session, old_start, old_end) == 0
    assert await _window_count(db_session, CLOSED_START, CLOSED_END) == 2


@pytest.mark.asyncio
async def test_refresh_cron_materialises_then_skips(db_session):
    """First run materialises the current month + every requested closed
    month; the second run only recomputes the current + recent months and
    skips already-materialised older windows (bounded steady-state work).
    Windows = current(i=0) + months_back closed = months_back+1 total."""
    from app.scheduled_jobs.ai_spending_handlers import (
        run_ai_spending_period_rollup_refresh,
    )
    cfg = {"months_back": 4, "recent_months": 2}
    with patch.object(
        AiSpendingService, "_compute_period_rows",
        AsyncMock(return_value=_rows()),
    ):
        r1 = await run_ai_spending_period_rollup_refresh(db_session, cfg)
        # i = 0..4 → current month + 4 closed = 5 windows, all fresh.
        assert r1["periods_refreshed"] == 5
        assert r1["periods_skipped"] == 0
        r2 = await run_ai_spending_period_rollup_refresh(db_session, cfg)
        # always-refresh i=0,1,2 (current + recent 2) = 3; i=3,4 skip = 2.
        assert r2["periods_refreshed"] == 3
        assert r2["periods_skipped"] == 2