"""Regression: the procurement reminder cron actually dispatches.

PR #587 shipped ``procurement.reminders`` whose ``_push_inbox`` called a
nonexistent ``NotificationService.enqueue_inbox`` with a nonexistent
``coalesce_key`` kwarg — every call raised ``AttributeError``, which the
handler's blanket ``except`` swallowed. The cron ran green while NEVER
dispatching a single reminder (a silent dead feature). These tests pin
that a real ``notification_outbox`` row lands AND that a same-ISO-week
re-run is deduped (no storm).
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta
from decimal import Decimal

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

from app.domains.procurement.models import (
    ProcurementBudget,
    ProcurementBudgetConsumption,
    ProcurementOrder,
)
from app.identity.person_models import IdentityPerson
from app.notifications.models import NotificationOutbox
from app.scheduled_jobs.procurement_handlers import (
    _REMINDER_TEMPLATE_KEYS,
    _TEMPLATE_ORDER_LATE,
    run_procurement_reminders,
)


async def _person(db: AsyncSession, email: str) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=email.split("@")[0],
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db.add(p)
    await db.flush()
    return p


async def _outbox_rows(db: AsyncSession, person_id: uuid.UUID) -> list:
    return (
        await db.execute(
            select(NotificationOutbox).where(
                NotificationOutbox.recipient_person_id == person_id,
                # The three reminder kinds carry three template keys (one
                # per piece of copy) — match the whole family.
                NotificationOutbox.template_key.in_(_REMINDER_TEMPLATE_KEYS),
            )
        )
    ).scalars().all()


@pytest.mark.asyncio
async def test_budget_threshold_reminder_dispatches_and_dedups(
    db_session: AsyncSession,
):
    owner = await _person(db_session, "budget-owner@x.com")
    budget = ProcurementBudget(
        id=str(uuid.uuid4()),
        year=2026,
        department="IT",
        currency="CNY",
        total_amount=Decimal("1000.00"),
        owner_person_id=str(owner.id),
        is_active=True,
    )
    db_session.add(budget)
    await db_session.flush()
    # 90% locked → crosses the 80% threshold.
    db_session.add(ProcurementBudgetConsumption(
        id=str(uuid.uuid4()),
        budget_id=str(budget.id),
        event="lock",
        amount=Decimal("900.00"),
        currency="CNY",
    ))
    await db_session.commit()

    # First run — a real reminder is dispatched (NOT a silent no-op).
    counts = await run_procurement_reminders(db_session, {})
    assert counts["budget_alerts_dispatched"] == 1
    rows = await _outbox_rows(db_session, owner.id)
    assert len(rows) == 1
    assert rows[0].context["coalesce_key"].startswith(
        "procurement.budget.threshold:"
    )
    # In-app only — no Feishu/email push burst on first activation.
    assert rows[0].channels == []

    # Second run, same ISO week — deduped: no new row, counter reflects 0.
    counts2 = await run_procurement_reminders(db_session, {})
    assert counts2["budget_alerts_dispatched"] == 0
    assert len(await _outbox_rows(db_session, owner.id)) == 1


@pytest.mark.asyncio
async def test_budget_below_threshold_dispatches_nothing(
    db_session: AsyncSession,
):
    owner = await _person(db_session, "low-budget-owner@x.com")
    budget = ProcurementBudget(
        id=str(uuid.uuid4()),
        year=2026,
        department="Eng",
        currency="CNY",
        total_amount=Decimal("1000.00"),
        owner_person_id=str(owner.id),
        is_active=True,
    )
    db_session.add(budget)
    await db_session.flush()
    db_session.add(ProcurementBudgetConsumption(
        id=str(uuid.uuid4()),
        budget_id=str(budget.id),
        event="lock",
        amount=Decimal("500.00"),  # 50% — below the 80% floor
        currency="CNY",
    ))
    await db_session.commit()

    counts = await run_procurement_reminders(db_session, {})
    assert counts["budget_alerts_dispatched"] == 0
    assert await _outbox_rows(db_session, owner.id) == []


async def _budget_at(
    db: AsyncSession, owner: IdentityPerson, locked: str, total: str = "1000.00",
) -> ProcurementBudget:
    budget = ProcurementBudget(
        id=str(uuid.uuid4()),
        year=2026,
        department=f"DEPT-{uuid.uuid4().hex[:6]}",
        currency="CNY",
        total_amount=Decimal(total),
        owner_person_id=str(owner.id),
        is_active=True,
    )
    db.add(budget)
    await db.flush()
    db.add(ProcurementBudgetConsumption(
        id=str(uuid.uuid4()),
        budget_id=str(budget.id),
        event="lock",
        amount=Decimal(locked),
        currency="CNY",
    ))
    await db.commit()
    return budget


@pytest.mark.asyncio
async def test_budget_alert_is_edge_triggered_not_weekly(
    db_session: AsyncSession,
):
    """A pool steady above 80% must alert ONCE, not every week. The
    per-ISO-week coalesce key alone would re-fire on the next week; the
    last_alerted_threshold marker is what actually suppresses the re-nag."""
    owner = await _person(db_session, "steady-budget@x.com")
    budget = await _budget_at(db_session, owner, locked="850.00")  # 85%

    counts = await run_procurement_reminders(db_session, {})
    assert counts["budget_alerts_dispatched"] == 1
    await db_session.refresh(budget)
    assert budget.last_alerted_threshold == 80

    # Simulate "next ISO week": delete the outbox row so the week-bucket
    # dedup can NOT be what suppresses the second run. Only the
    # edge-trigger marker should stop it.
    for row in await _outbox_rows(db_session, owner.id):
        await db_session.delete(row)
    await db_session.commit()

    counts2 = await run_procurement_reminders(db_session, {})
    assert counts2["budget_alerts_dispatched"] == 0
    assert await _outbox_rows(db_session, owner.id) == []


@pytest.mark.asyncio
async def test_budget_alert_escalates_80_to_95(db_session: AsyncSession):
    owner = await _person(db_session, "escalate-budget@x.com")
    budget = await _budget_at(db_session, owner, locked="850.00")  # 85% → 80

    assert (await run_procurement_reminders(db_session, {}))[
        "budget_alerts_dispatched"] == 1
    await db_session.refresh(budget)
    assert budget.last_alerted_threshold == 80

    # Lock rises to 96% → tier 95 > last 80 → a fresh escalation alert.
    db_session.add(ProcurementBudgetConsumption(
        id=str(uuid.uuid4()),
        budget_id=str(budget.id),
        event="lock",
        amount=Decimal("110.00"),  # 850 + 110 = 960 / 1000 = 96%
        currency="CNY",
    ))
    await db_session.commit()

    assert (await run_procurement_reminders(db_session, {}))[
        "budget_alerts_dispatched"] == 1
    await db_session.refresh(budget)
    assert budget.last_alerted_threshold == 95
    rows = await _outbox_rows(db_session, owner.id)
    assert any(":95:" in r.context["coalesce_key"] for r in rows)


@pytest.mark.asyncio
async def test_budget_alert_resets_below_then_realerts(db_session: AsyncSession):
    owner = await _person(db_session, "reset-budget@x.com")
    budget = await _budget_at(db_session, owner, locked="850.00")  # 85% → 80
    assert (await run_procurement_reminders(db_session, {}))[
        "budget_alerts_dispatched"] == 1

    # Utilization drops below 80% (release) → marker resets, no alert.
    db_session.add(ProcurementBudgetConsumption(
        id=str(uuid.uuid4()),
        budget_id=str(budget.id),
        event="release",
        amount=Decimal("500.00"),  # net 350 / 1000 = 35%
        currency="CNY",
    ))
    await db_session.commit()
    assert (await run_procurement_reminders(db_session, {}))[
        "budget_alerts_dispatched"] == 0
    await db_session.refresh(budget)
    assert budget.last_alerted_threshold is None

    # Climbs back above 80% → a genuine re-crossing alerts again.
    db_session.add(ProcurementBudgetConsumption(
        id=str(uuid.uuid4()),
        budget_id=str(budget.id),
        event="lock",
        amount=Decimal("600.00"),  # net 950 / 1000 = 95%
        currency="CNY",
    ))
    await db_session.commit()
    assert (await run_procurement_reminders(db_session, {}))[
        "budget_alerts_dispatched"] == 1


@pytest.mark.asyncio
async def test_lingering_po_nag_dispatches_without_crashing(
    db_session: AsyncSession,
):
    """A submitted PO older than the 5-day cutoff must produce a nag.

    Regression: ``requested_at`` is ``DateTime(timezone=True)`` but the
    MySQL/TiDB round-trip strips tzinfo, so the freshly-loaded row is
    NAIVE. The handler computed ``datetime.now(UTC) - po.requested_at``
    (aware − naive) which raised ``TypeError`` — and since the handler is
    a registered scheduled job, that raise paged on-call on EVERY tick
    once a single old PO existed. We force the naive-load path with
    ``expire_all()`` and assert the nag dispatches cleanly.
    """
    owner = await _person(db_session, "po-owner@x.com")
    # Held as a plain local: ``expire_all()`` below detaches every
    # attribute, so reading ``po.order_number`` after the run would
    # trigger a lazy refresh (MissingGreenlet under the async session).
    order_number = f"PO-{uuid.uuid4().hex[:8]}"
    po = ProcurementOrder(
        id=str(uuid.uuid4()),
        order_number=order_number,
        order_type="software",
        status="submitted",
        currency="CNY",
        applicant_id=str(owner.id),
        procurement_owner_id=str(owner.id),
        external_po_number=None,
        # 10 days old (older than the 5-day cutoff). Stored as an AWARE
        # datetime; the round-trip + expire_all() below makes the reload
        # naive, exactly as production does.
        requested_at=datetime.now(UTC) - timedelta(days=10),
    )
    db_session.add(po)
    await db_session.commit()
    # Force the handler's SELECT to reload the row from the DB (naive
    # datetime) instead of returning the cached aware instance.
    db_session.expire_all()

    counts = await run_procurement_reminders(db_session, {})
    assert counts["order_nags_dispatched"] == 1
    rows = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.recipient_person_id == owner.id,
            NotificationOutbox.template_key == _TEMPLATE_ORDER_LATE,
        )
    )).scalars().all()
    assert len(rows) == 1
    assert rows[0].context["coalesce_key"].startswith(
        "procurement.order.helian_late:"
    )
    # The late-PO copy is localised through the template, so the
    # placeholders it interpolates must be on the frozen context.
    assert rows[0].context["order_number"] == order_number
    # Past the 5-day cutoff. Not pinned to exactly 10: ``requested_at``
    # round-trips through the driver as a NAIVE datetime and the handler
    # re-stamps it UTC, so the day count lands 9 or 10 depending on the
    # session's UTC offset — the same naive/aware quirk this test exists
    # to guard. What matters here is that ``days`` is on the context at
    # all (the template interpolates it) and is a sane age.
    assert rows[0].context["days"] >= 5


@pytest.mark.asyncio
async def test_reminder_skips_left_owner(db_session: AsyncSession):
    """A budget owner who has left must not keep getting bell reminders."""
    # Construct as 'left' at INSERT — flipping active→left would trip the
    # LeaverBlocked listener, which is unrelated to this cron's behaviour.
    owner = IdentityPerson(
        id=uuid.uuid4(),
        work_email="departed-owner@x.com",
        display_name="departed-owner",
        employment_status="left",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(owner)
    await db_session.flush()
    await _budget_at(db_session, owner, locked="900.00")  # 90%

    counts = await run_procurement_reminders(db_session, {})
    assert counts["budget_alerts_dispatched"] == 0
    assert await _outbox_rows(db_session, owner.id) == []