"""Regression tests for ``ai_spending.monthly_summary_generate``. Two pins: 1. **Period TZ alignment** — the cron schedule is ``5 3 1 * *`` Asia/ Shanghai, but PR #443 had refactored the handler to compute "previous month" via ``datetime.now(UTC).date()``. At 03:05 SH on day 1 of month M (= 19:05 UTC on the last day of M-1), the UTC-anchored computation resolves the just-ended month as M-2 instead of M-1. In production this overwrote the existing 2026-04 summary on the 2026-06-01 cron run when it was supposed to generate 2026-05. Pinned here so a future refactor can't accidentally swap the TZ back. 2. **Skip-when-no-API-key path** — handler must early-return without blowing up when the operator hasn't configured the Bailian LLM key. """ from __future__ import annotations from datetime import UTC, datetime from typing import Any from zoneinfo import ZoneInfo import pytest from sqlalchemy.ext.asyncio import AsyncSession from app.domains.ai_spending.service import SETTING_LLM_API_KEY from app.integrations.models import PlatformSetting from app.scheduled_jobs import ai_spending_handlers from app.scheduled_jobs.ai_spending_handlers import ( _PERIOD_TZ, run_ai_spending_monthly_summary, ) def test_period_tz_is_shanghai_not_utc() -> None: """The handler MUST resolve "previous month" in the same timezone as the cron schedule. UTC anchoring breaks for SH-anchored crons fired at SH 00:00–07:59 of day 1 (UTC is still M-1 at that point). """ assert _PERIOD_TZ == ZoneInfo("Asia/Shanghai") @pytest.mark.asyncio async def test_skip_when_no_api_key(db_session: AsyncSession) -> None: """Without the LLM key configured the handler returns a skip-result instead of raising — this is the default state on a fresh install and shouldn't alarm via failure_notifier.""" # Ensure no key row exists. existing = await db_session.get(PlatformSetting, SETTING_LLM_API_KEY) if existing is not None: await db_session.delete(existing) await db_session.flush() result = await run_ai_spending_monthly_summary(db_session, {}) assert result["generated"] is False assert "skipped_reason" in result assert "api_key not configured" in result["skipped_reason"] @pytest.mark.asyncio async def test_skip_when_config_disabled(db_session: AsyncSession) -> None: """``config.enabled=false`` is a pause toggle (used during flaky LLM windows) — handler must respect it without trying to call the LLM.""" db_session.add(PlatformSetting( key=SETTING_LLM_API_KEY, value="sk-fake-disabled-path", )) await db_session.flush() result = await run_ai_spending_monthly_summary( db_session, {"enabled": False}, ) assert result["generated"] is False assert result["skipped_reason"] == "config.enabled=false" @pytest.mark.asyncio @pytest.mark.parametrize( "now_utc, expected_period", [ # Cron actually fires at SH 03:05 on day 1 = UTC 19:05 on the # LAST day of M-1. UTC-anchored "previous month" would resolve # to M-2 here — wrong. SH-anchored resolves to M-1 — right. (datetime(2026, 5, 31, 19, 5, tzinfo=UTC), "2026-05"), (datetime(2026, 4, 30, 19, 5, tzinfo=UTC), "2026-04"), (datetime(2025, 12, 31, 19, 5, tzinfo=UTC), "2025-12"), # And later in the day on day 1 (UTC has caught up), the # period is still M-1 — sanity check that anchoring on SH # doesn't accidentally double-shift. (datetime(2026, 6, 1, 12, 0, tzinfo=UTC), "2026-05"), ], ) async def test_period_resolves_to_just_ended_month_at_cron_fire( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, now_utc: datetime, expected_period: str, ) -> None: """The cron fires at SH 03:05 on day 1; the handler must summarise the month that just ended in SH local time. We stub out ``auto_generate_summary`` so the test only validates the period-resolution path — no actual LLM call. Verifies the period string passed downstream is correct at the SH/UTC calendar boundary that previously misfired in prod. """ db_session.add(PlatformSetting( key=SETTING_LLM_API_KEY, value="sk-fake-test-key", )) await db_session.flush() pinned = now_utc class _FrozenDatetime(datetime): @classmethod def now(cls, tz=None): # type: ignore[override] if tz is None: return pinned.replace(tzinfo=None) return pinned.astimezone(tz) monkeypatch.setattr(ai_spending_handlers, "datetime", _FrozenDatetime) captured: dict[str, Any] = {} async def fake_auto_generate(self, *, period, trace_id): # noqa: ANN001 captured["period"] = period captured["trace_id"] = trace_id class _Resp: model_used = "qwen3-max" input_token_count = None output_token_count = None return _Resp() monkeypatch.setattr( "app.domains.ai_spending.service.AiSpendingService.auto_generate_summary", fake_auto_generate, ) result = await run_ai_spending_monthly_summary(db_session, {}) assert result["generated"] is True assert captured["period"] == expected_period assert captured["trace_id"] == f"ai_spending.auto.{expected_period}"