"""Tests for the `apikey.expiry_check` cron handler (PR3-2).

Three properties:

1. ``evaluate_assignment`` fires on the 7-day window for a key
   expiring in 7 days (assignee + manager both receive cards),
   creates the dedup row, and a second call on the same day is a
   no-op (dedup-hit).

2. The 1-day threshold short-circuits the 7-day one: a key with
   ``days_left=1`` fires the 1-day card, not the 7-day card.

3. After admin extends the key (``valid_until`` changes), the next
   cron tick treats it as a fresh window and re-fires.

Notifications are routed through a fake :class:`NotificationService`
to avoid touching Feishu / SMTP.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from types import SimpleNamespace
from typing import Any

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

from app.domains.apikey.expiry_alerts import (
    evaluate_assignment,
    sweep_expiry_alerts,
)
from app.domains.apikey.models import (
    ApiKeyAssignment,
    ApiKeyExpiryAlertsDedup,
    ApiKeyProvider,
)
from app.identity.person_models import IdentityPerson


async def _make_person(
    db: AsyncSession,
    *,
    email: str,
    manager_id=None,
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name=email.split("@")[0],
        work_email=email,
        employment_status="active",
    )
    db.add(p)
    await db.commit()
    if manager_id is not None:
        # `IdentityPerson.manager_id` is a hybrid property; assign via
        # the setter which routes through the actual column.
        p.manager_id = manager_id
        await db.commit()
    return p


async def _make_provider(db: AsyncSession) -> ApiKeyProvider:
    p = ApiKeyProvider(
        id=uuid.uuid4(),
        name=f"Provider {uuid.uuid4().hex[:6]}",
        description="",
        is_active=True,
        sort_order=0,
    )
    db.add(p)
    await db.commit()
    return p


async def _make_active_assignment(
    db: AsyncSession,
    *,
    provider: ApiKeyProvider,
    applicant: IdentityPerson,
    valid_until,
) -> ApiKeyAssignment:
    a = ApiKeyAssignment(
        id=uuid.uuid4(),
        provider_id=provider.id,
        assigned_to=applicant.id,
        status="active",
        request_number=f"AK-{uuid.uuid4().hex[:8].upper()}",
        project_name="expiry-test",
        reason="expiry tests",
        monthly_budget_usd=Decimal("100.00"),
        key_value=None,
        valid_until=valid_until,
        detail={},
    )
    db.add(a)
    await db.commit()
    return a


@pytest.fixture
def fake_notifier(monkeypatch):
    """Capture every NotificationEnvelope passed to ``dispatch``."""
    sent: list[Any] = []

    async def fake_dispatch(self, env, *, trace_id="", actor_user_id=None):  # noqa: ANN001
        sent.append(env)
        return SimpleNamespace(delivered_channels=["feishu", "email"])

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "dispatch", fake_dispatch)
    return sent


@pytest.mark.asyncio
async def test_seven_day_window_fires_to_assignee_and_manager(
    db_session: AsyncSession, fake_notifier,
):
    """Key expiring in 7 days → 7-day card to assignee + manager,
    dedup row written, re-running is a no-op."""
    manager = await _make_person(db_session, email="mgr-7d@pingcap.com")
    employee = await _make_person(
        db_session, email="emp-7d@pingcap.com",
        manager_id=manager.id,
    )
    provider = await _make_provider(db_session)
    from datetime import date as date_type

    today = date_type(2026, 5, 19)
    assignment = await _make_active_assignment(
        db_session,
        provider=provider,
        applicant=employee,
        valid_until=today + timedelta(days=7),
    )

    result = await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    await db_session.commit()

    assert result["fired"]
    fired = result["fired"][0]
    assert fired["threshold_days"] == 7
    assert fired["days_left"] == 7
    # Two recipients dispatched: assignee + manager.
    assert len(fake_notifier) == 2
    titles = {env.title for env in fake_notifier}
    assert any("API Key 即将到期" in t for t in titles)
    # Dedup row landed.
    dedup_count = (
        await db_session.execute(
            select(func.count()).select_from(ApiKeyExpiryAlertsDedup),
        )
    ).scalar()
    assert dedup_count == 1

    # Second evaluation same day = no-op (dedup hit).
    fake_notifier.clear()
    result2 = await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    await db_session.commit()
    assert result2["fired"] == []
    assert result2["skipped"]
    assert "already-alerted" in result2["skipped"][0]["reason"]
    assert len(fake_notifier) == 0


@pytest.mark.asyncio
async def test_one_day_window_supersedes_seven_day(
    db_session: AsyncSession, fake_notifier,
):
    """``days_left=1`` fires the 1-day card. Even if the 7-day card
    was never sent (cron offline during the 7-day window), the
    1-day fire is what the assignee gets — not a redundant pair."""
    employee = await _make_person(db_session, email="emp-1d@pingcap.com")
    provider = await _make_provider(db_session)
    from datetime import date as date_type

    today = date_type(2026, 5, 19)
    assignment = await _make_active_assignment(
        db_session,
        provider=provider,
        applicant=employee,
        valid_until=today + timedelta(days=1),
    )

    result = await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    await db_session.commit()
    assert result["fired"]
    assert result["fired"][0]["threshold_days"] == 1
    # No 7-day row left lingering.
    dedup_rows = (
        await db_session.execute(select(ApiKeyExpiryAlertsDedup))
    ).scalars().all()
    assert len(dedup_rows) == 1
    assert dedup_rows[0].threshold_days == 1


@pytest.mark.asyncio
async def test_extending_validity_refires_on_next_window(
    db_session: AsyncSession, fake_notifier,
):
    """Admin extends ``valid_until`` → dedup key changes → cron re-
    fires when the new window crosses 7 days out."""
    employee = await _make_person(db_session, email="emp-ext@pingcap.com")
    provider = await _make_provider(db_session)
    from datetime import date as date_type

    today = date_type(2026, 5, 19)
    assignment = await _make_active_assignment(
        db_session,
        provider=provider,
        applicant=employee,
        valid_until=today + timedelta(days=7),
    )

    # First fire: 7-day card.
    await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    await db_session.commit()
    assert len(fake_notifier) == 1
    fake_notifier.clear()

    # Admin extends to 30 days out. Same-day re-eval: no fire (outside
    # any window).
    assignment.valid_until = today + timedelta(days=30)
    await db_session.commit()
    res_no_fire = await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    assert res_no_fire["fired"] == []

    # Fast-forward "today" — cron tick when days_left is 7 again.
    future_today = today + timedelta(days=23)  # 30 - 23 = 7 days left
    res_refire = await evaluate_assignment(
        db_session, assignment=assignment, today=future_today,
    )
    await db_session.commit()
    assert res_refire["fired"]
    assert res_refire["fired"][0]["threshold_days"] == 7
    # Two dedup rows: the original (valid_until=today+7) and the new
    # one (valid_until=today+30).
    dedup_count = (
        await db_session.execute(
            select(func.count()).select_from(ApiKeyExpiryAlertsDedup),
        )
    ).scalar()
    assert dedup_count == 2


@pytest.mark.asyncio
async def test_no_valid_until_skips_silently(
    db_session: AsyncSession, fake_notifier,
):
    """Perpetual keys (no valid_until) are not in scope; skip without
    error or dedup row."""
    employee = await _make_person(db_session, email="emp-none@pingcap.com")
    provider = await _make_provider(db_session)
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee, valid_until=None,
    )
    from datetime import date as date_type

    res = await evaluate_assignment(
        db_session, assignment=assignment, today=date_type(2026, 5, 19),
    )
    assert res["fired"] == []
    assert any(s["reason"] == "no-valid-until" for s in res["skipped"])
    assert len(fake_notifier) == 0


@pytest.mark.asyncio
async def test_leaver_skips_silently(
    db_session: AsyncSession, fake_notifier,
):
    """Don't pester a leaver to extend a key — offboarding owns that
    row already (PR1/PR2 cascade)."""
    employee = await _make_person(db_session, email="leaver-exp@pingcap.com")
    employee.employment_status = "leaver_pending"
    await db_session.commit()
    provider = await _make_provider(db_session)
    from datetime import date as date_type

    today = date_type(2026, 5, 19)
    assignment = await _make_active_assignment(
        db_session, provider=provider, applicant=employee,
        valid_until=today + timedelta(days=7),
    )

    res = await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    assert res["fired"] == []
    assert any(
        s["reason"].startswith("applicant-") for s in res["skipped"]
    )
    assert len(fake_notifier) == 0


@pytest.mark.asyncio
async def test_sweep_does_not_rollback_other_iterations_on_dedup_hit(
    db_session: AsyncSession, fake_notifier,
):
    """Round-45 self-review fix: an earlier shape used IntegrityError +
    session.rollback() to handle dedup hits, which would wipe ALL
    prior iterations' dedup rows in the same sweep. With the
    SELECT-then-INSERT fix, a dedup hit on row #2 must NOT roll back
    row #1's just-written dedup."""
    provider = await _make_provider(db_session)
    from datetime import date as date_type

    today = date_type(2026, 5, 19)

    # Row #1: fresh — will fire and write a dedup row.
    fresh_employee = await _make_person(
        db_session, email="fresh-pre@pingcap.com",
    )
    fresh = await _make_active_assignment(
        db_session, provider=provider, applicant=fresh_employee,
        valid_until=today + timedelta(days=7),
    )

    # Row #2: pre-existing dedup row (simulating a prior cron tick) →
    # will skip with "already-alerted" on the next sweep.
    repeat_employee = await _make_person(
        db_session, email="repeat-pre@pingcap.com",
    )
    repeat = await _make_active_assignment(
        db_session, provider=provider, applicant=repeat_employee,
        valid_until=today + timedelta(days=7),
    )
    db_session.add(ApiKeyExpiryAlertsDedup(
        id=uuid.uuid4(),
        assignment_id=repeat.id,
        valid_until_iso=repeat.valid_until,
        threshold_days=7,
        last_alerted_at=datetime(2026, 5, 18, tzinfo=_tz_utc()),
    ))
    await db_session.commit()

    # Run the sweep: row #1 should fire + write dedup; row #2 should
    # skip; row #1's dedup must SURVIVE (not be rolled back).
    summary = await sweep_expiry_alerts(db_session, today=today)
    await db_session.commit()

    assert summary["evaluated"] == 2
    assert summary["fired"] == 1
    assert summary["skipped"] == 1

    # Two dedup rows: the pre-seeded one + the newly-written one for
    # the fresh assignment.
    dedup_rows = (
        await db_session.execute(select(ApiKeyExpiryAlertsDedup))
    ).scalars().all()
    assert len(dedup_rows) == 2
    fresh_dedup = [
        r for r in dedup_rows if r.assignment_id == fresh.id
    ]
    assert len(fresh_dedup) == 1


def _tz_utc():

    return UTC


@pytest.mark.asyncio
async def test_sweep_aggregates_correctly(
    db_session: AsyncSession, fake_notifier,
):
    """``sweep_expiry_alerts`` returns a sane aggregate: counts
    evaluated / fired / skipped / errors. Verifies the cron handler's
    decision logic without invoking the handler directly (which would
    require an extra session_factory monkeypatch)."""
    provider = await _make_provider(db_session)
    from datetime import date as date_type

    today = date_type(2026, 5, 19)

    _a = await _make_active_assignment(
        db_session,
        provider=provider,
        applicant=await _make_person(db_session, email="a@pingcap.com"),
        valid_until=today + timedelta(days=7),
    )
    _b = await _make_active_assignment(
        db_session,
        provider=provider,
        applicant=await _make_person(db_session, email="b@pingcap.com"),
        valid_until=today + timedelta(days=1),
    )
    _c = await _make_active_assignment(
        db_session,
        provider=provider,
        applicant=await _make_person(db_session, email="c@pingcap.com"),
        valid_until=today + timedelta(days=30),  # outside both windows
    )

    summary = await sweep_expiry_alerts(db_session, today=today)
    await db_session.commit()

    assert summary["evaluated"] == 3
    assert summary["fired"] == 2  # a + b
    assert summary["skipped"] == 1  # c
    assert summary["errors"] == []