"""API tests for POST /ai-spending/period-rollup/refresh — the
reconciliation cost-drift "重算物化" manual fix (PR4).

Closed-month-only (open months are computed live), IT-admin gated,
audited. Compute is stubbed — these pin the endpoint contract, not the
rollup mechanics (covered by test_period_rollup.py).
"""
from __future__ import annotations

from datetime import UTC, datetime
from unittest.mock import AsyncMock, patch
from zoneinfo import ZoneInfo

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

from app.audit.models import AuditLog
from app.domains.ai_spending.service import AiSpendingService


def _closed_period() -> str:
    """A month guaranteed strictly before the current Asia/Shanghai
    month (use the prior calendar year so the test is clock-stable)."""
    now = datetime.now(UTC)
    return f"{now.year - 1:04d}-01"


@pytest.mark.asyncio
async def test_refresh_closed_month_ok(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    with patch.object(
        AiSpendingService, "refresh_period_rollup",
        AsyncMock(return_value=3),
    ):
        resp = await client.post(
            "/api/v1/ai-spending/period-rollup/refresh",
            json={"period": _closed_period()},
            headers=admin_headers,
        )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["periods"] == [_closed_period()]
    assert data["rows_written"] == 3

    row = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "ai_spending.rollup.manual_refresh",
        ),
    )).scalars().first()
    assert row is not None


@pytest.mark.asyncio
async def test_refresh_open_month_rejected(
    client: AsyncClient, admin_headers,
):
    # The server computes "open month" in Asia/Shanghai (see _closed_period
    # + the rollup service). Compute the open period in the SAME tz, else at
    # the UTC↔Shanghai day/month boundary (UTC still prev month while
    # Shanghai has rolled over) the test's "open" period is already CLOSED
    # server-side → refresh allowed → 200 instead of 400.
    now = datetime.now(ZoneInfo("Asia/Shanghai"))
    open_period = f"{now.year:04d}-{now.month:02d}"
    with patch.object(
        AiSpendingService, "refresh_period_rollup",
        AsyncMock(return_value=0),
    ):
        resp = await client.post(
            "/api/v1/ai-spending/period-rollup/refresh",
            json={"period": open_period},
            headers=admin_headers,
        )
    assert resp.status_code == 400
    assert resp.json()["errors"][0]["code"] == "ROLLUP_OPEN_MONTH"


@pytest.mark.asyncio
async def test_refresh_bad_format_422(client: AsyncClient, admin_headers):
    resp = await client.post(
        "/api/v1/ai-spending/period-rollup/refresh",
        json={"period": "2026-13"},
        headers=admin_headers,
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_refresh_no_period_does_recent_two(
    client: AsyncClient, admin_headers,
):
    with patch.object(
        AiSpendingService, "refresh_period_rollup",
        AsyncMock(return_value=1),
    ):
        resp = await client.post(
            "/api/v1/ai-spending/period-rollup/refresh",
            json={},
            headers=admin_headers,
        )
    assert resp.status_code == 200, resp.text
    assert len(resp.json()["data"]["periods"]) == 2


@pytest.mark.asyncio
async def test_refresh_employee_forbidden(
    client: AsyncClient, employee_headers,
):
    resp = await client.post(
        "/api/v1/ai-spending/period-rollup/refresh",
        json={"period": _closed_period()},
        headers=employee_headers,
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_refresh_409_and_no_write_while_rollup_lock_held(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    """2026-07-28 review P1: the manual re-materialisation used to bypass
    ``ROLLUP_WRITE_LOCK`` entirely — an admin click overlapping the 06:30
    refresh cron or the 07:00 drift-heal left duplicate window rows every
    dashboard read SUMs. While another writer holds the lock the endpoint
    must answer 409 ``AI_ROLLUP_REFRESH_BUSY`` (a user-meaningful busy,
    not a stack trace — the MCP tool shares this exact service path and
    message) and perform NO rollup write and NO audit write."""
    from app.core.distributed_lock import distributed_lock
    from app.domains.ai_spending.locks import ROLLUP_WRITE_LOCK

    rebuild = AsyncMock(return_value=3)
    with patch.object(AiSpendingService, "refresh_period_rollup", rebuild):
        # Simulate the refresh cron / drift-heal holding the lock.
        async with distributed_lock(ROLLUP_WRITE_LOCK, ttl=60):
            resp = await client.post(
                "/api/v1/ai-spending/period-rollup/refresh",
                json={"period": _closed_period()},
                headers=admin_headers,
            )

    assert resp.status_code == 409, resp.text
    err = resp.json()["errors"][0]
    assert err["code"] == "AI_ROLLUP_REFRESH_BUSY"
    # The message is what the MCP wrapper surfaces verbatim — it must be
    # actionable prose, not an exception repr.
    assert "rebuilt" in err["message"] or "rebuild" in err["message"]
    rebuild.assert_not_awaited()

    row = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "ai_spending.rollup.manual_refresh",
        ),
    )).scalars().first()
    assert row is None, "busy refusal must not log a success audit row"


@pytest.mark.asyncio
async def test_refresh_succeeds_again_after_lock_released(
    client: AsyncClient, admin_headers,
):
    """The 409 is transient by design — once the lock holder finishes, the
    same request must go through (guards against the helper failing to
    release / leaking the busy state)."""
    from app.core.distributed_lock import distributed_lock
    from app.domains.ai_spending.locks import ROLLUP_WRITE_LOCK

    with patch.object(
        AiSpendingService, "refresh_period_rollup",
        AsyncMock(return_value=1),
    ):
        async with distributed_lock(ROLLUP_WRITE_LOCK, ttl=60):
            busy = await client.post(
                "/api/v1/ai-spending/period-rollup/refresh",
                json={"period": _closed_period()},
                headers=admin_headers,
            )
        assert busy.status_code == 409

        ok = await client.post(
            "/api/v1/ai-spending/period-rollup/refresh",
            json={"period": _closed_period()},
            headers=admin_headers,
        )
    assert ok.status_code == 200, ok.text
    assert ok.json()["data"]["periods"] == [_closed_period()]