"""DB-backed tests for the unified expiry-alert engine: graduated cadence
(T-30 / T-14 once → daily ≤7 / expired), last-mile escalation
(本人 + 域名负责人 + lead), first-enable backlog suppression, and the
acknowledge / snooze loop."""
from __future__ import annotations

import uuid
from datetime import UTC, date, datetime, timedelta

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

from app.domains.cmdb.expiry_alert_engine import (
    BUCKET_7D,
    BUCKET_14D,
    BUCKET_30D,
    BUCKET_EXPIRED,
    ExpiryItem,
    acknowledge_expiry,
    run_expiry_alerts,
)
from app.domains.cmdb.models import CmdbExpiryAlertState
from app.identity.person_models import IdentityPerson

RT = "certificate"


class _Recorder:
    """A fake ``notify`` that records (resource_id, bucket, roles)."""

    def __init__(self, ok: bool = True):
        self.ok = ok
        self.calls: list[tuple[str, str, list[str]]] = []

    async def __call__(self, item, recipients, bucket) -> bool:
        self.calls.append(
            (item.resource_id, bucket, sorted(r.role for r in recipients)),
        )
        return self.ok


async def _person(db, name, *, manager_id=None, active=True) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(), source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", display_name=name,
        work_email=f"{name.lower()}-{uuid.uuid4().hex[:6]}@pingcap.com",
        employment_status="active" if active else "left",
        manager_person_id=manager_id,
    )
    db.add(p)
    await db.flush()
    return p


def _item(owner_id, *, days, domain_owner_id=None, fresh=False, rid=None) -> ExpiryItem:
    exp = datetime.now(UTC).date() + timedelta(days=days)
    return ExpiryItem(
        resource_type=RT, resource_id=rid or str(uuid.uuid4()),
        label="svc.example.com", expiry_date=exp,
        primary_owner_id=str(owner_id) if owner_id else None,
        domain_owner_id=str(domain_owner_id) if domain_owner_id else None,
        first_seen_fresh=fresh,
    )


async def _state(db, rid) -> CmdbExpiryAlertState | None:
    return (await db.execute(
        select(CmdbExpiryAlertState).where(
            CmdbExpiryAlertState.resource_type == RT,
            CmdbExpiryAlertState.resource_id == rid,
        )
    )).scalar_one_or_none()


@pytest.mark.asyncio
async def test_t30_t14_fire_once_owner_only(db_session: AsyncSession):
    owner = await _person(db_session, "Owner")
    await db_session.commit()
    # Pre-seed state at <=30d so the FIRST observation isn't suppressed —
    # this exercises the steady-state "fire once on entry" path.
    it30 = _item(owner.id, days=20, rid=str(uuid.uuid4()))
    db_session.add(CmdbExpiryAlertState(
        resource_type=RT, resource_id=it30.resource_id, last_bucket=BUCKET_30D,
        last_owner_person_id=str(owner.id), last_alerted_at=datetime.now(UTC),
    ))
    await db_session.commit()

    rec = _Recorder()
    # Same bucket, same owner → deduped, no send.
    s = await run_expiry_alerts(db_session, resource_type=RT, items=[it30], config={}, notify=rec)
    assert rec.calls == []
    assert s["deduped_unchanged"] == 1

    # Crosses into <=14d → fires once, owner ONLY (no escalation).
    it14 = _item(owner.id, days=12, rid=it30.resource_id)
    s = await run_expiry_alerts(db_session, resource_type=RT, items=[it14], config={}, notify=rec)
    assert len(rec.calls) == 1
    assert rec.calls[0][1] == BUCKET_14D
    assert rec.calls[0][2] == ["owner"]


@pytest.mark.asyncio
async def test_escalation_daily_in_last_7d(db_session: AsyncSession):
    lead = await _person(db_session, "Lead")
    owner = await _person(db_session, "Owner", manager_id=str(lead.id))
    dom_owner = await _person(db_session, "DomOwner")
    await db_session.commit()
    rid = str(uuid.uuid4())
    # Seed at <=14d (non-escalation) so crossing into <=7d is a go-forward
    # transition under the cron's watch.
    db_session.add(CmdbExpiryAlertState(
        resource_type=RT, resource_id=rid, last_bucket=BUCKET_14D,
        last_owner_person_id=str(owner.id), last_alerted_at=datetime.now(UTC),
    ))
    await db_session.commit()

    rec = _Recorder()
    it = _item(owner.id, days=5, domain_owner_id=dom_owner.id, rid=rid)
    # Day 1 in window → escalate to owner + domain_owner + lead.
    s = await run_expiry_alerts(db_session, resource_type=RT, items=[it], config={}, notify=rec)
    assert len(rec.calls) == 1
    assert rec.calls[0][2] == ["domain_owner", "lead", "owner"]
    assert s["escalated"] == 1
    st = await _state(db_session, rid)
    assert st.escalation_started_on is not None
    assert st.last_alerted_on == datetime.now(UTC).date()

    # Same day again → deduped (already alerted today).
    rec2 = _Recorder()
    await run_expiry_alerts(db_session, resource_type=RT, items=[it], config={}, notify=rec2)
    assert rec2.calls == []

    # Simulate "next day" by rewinding last_alerted_on, then re-run → daily resend.
    st.last_alerted_on = datetime.now(UTC).date() - timedelta(days=1)
    await db_session.commit()
    rec3 = _Recorder()
    await run_expiry_alerts(db_session, resource_type=RT, items=[it], config={}, notify=rec3)
    assert len(rec3.calls) == 1
    assert rec3.calls[0][2] == ["domain_owner", "lead", "owner"]


@pytest.mark.asyncio
async def test_owner_change_in_window_realerts_same_day(db_session: AsyncSession):
    """If the resolved owner changes mid-window, the NEW owner must be paged
    even if the old owner was already alerted today (else the new owner can
    miss the escalation before the cert lapses)."""
    old_owner = await _person(db_session, "OldOwner")
    new_owner = await _person(db_session, "NewOwner")
    await db_session.commit()
    rid = str(uuid.uuid4())
    db_session.add(CmdbExpiryAlertState(
        resource_type=RT, resource_id=rid, last_bucket=BUCKET_7D,
        last_owner_person_id=str(old_owner.id),
        escalation_started_on=date(2020, 1, 1),
        last_alerted_on=datetime.now(UTC).date(),  # already alerted TODAY
        last_alerted_at=datetime.now(UTC),
    ))
    await db_session.commit()
    rec = _Recorder()
    # Same day, but owner is now new_owner → must re-send.
    it = _item(new_owner.id, days=3, rid=rid)
    await run_expiry_alerts(db_session, resource_type=RT, items=[it], config={}, notify=rec)
    assert len(rec.calls) == 1
    assert rec.calls[0][2] == ["owner"]  # new owner, no lead (no manager set)


@pytest.mark.asyncio
async def test_first_enable_backlog_suppressed(db_session: AsyncSession):
    owner = await _person(db_session, "Owner")
    await db_session.commit()
    rec = _Recorder()
    # No prior state (first enable) + already in escalation window + NOT
    # fresh → suppressed (stamped, no send), escalation_started_on stays NULL.
    rid = str(uuid.uuid4())
    it = _item(owner.id, days=3, rid=rid, fresh=False)
    s = await run_expiry_alerts(db_session, resource_type=RT, items=[it], config={}, notify=rec)
    assert rec.calls == []
    assert s.get("suppressed_first_observation", 0) == 1
    st = await _state(db_session, rid)
    assert st is not None and st.escalation_started_on is None

    # Next day, still backlog (escalation_started_on NULL) → still suppressed.
    rec2 = _Recorder()
    await run_expiry_alerts(db_session, resource_type=RT, items=[it], config={}, notify=rec2)
    assert rec2.calls == []


@pytest.mark.asyncio
async def test_fresh_urgent_emits_on_first_sight(db_session: AsyncSession):
    owner = await _person(db_session, "Owner")
    await db_session.commit()
    rec = _Recorder()
    # Freshly discovered (fresh=True) + already expired → must alert on
    # first sight (it will never get a transition before lapsing).
    rid = str(uuid.uuid4())
    it = _item(owner.id, days=-1, rid=rid, fresh=True)
    _s = await run_expiry_alerts(db_session, resource_type=RT, items=[it], config={}, notify=rec)
    assert len(rec.calls) == 1
    assert rec.calls[0][1] == BUCKET_EXPIRED


@pytest.mark.asyncio
async def test_acknowledge_suppresses(db_session: AsyncSession):
    owner = await _person(db_session, "Owner")
    await db_session.commit()
    rid = str(uuid.uuid4())
    # Arm escalation first (seed already-in-window + armed).
    db_session.add(CmdbExpiryAlertState(
        resource_type=RT, resource_id=rid, last_bucket=BUCKET_7D,
        last_owner_person_id=str(owner.id),
        escalation_started_on=date(2020, 1, 1),
        last_alerted_on=datetime.now(UTC).date() - timedelta(days=1),
        last_alerted_at=datetime.now(UTC),
    ))
    await db_session.commit()
    # Owner acks → snooze.
    await acknowledge_expiry(db_session, resource_type=RT, resource_id=rid,
                             actor_person_id=str(owner.id), days=14)
    await db_session.commit()

    rec = _Recorder()
    it = _item(owner.id, days=3, rid=rid)
    s = await run_expiry_alerts(db_session, resource_type=RT, items=[it], config={}, notify=rec)
    assert rec.calls == []
    assert s.get("acknowledged_suppressed", 0) == 1


@pytest.mark.asyncio
async def test_ownerless_and_leaver_skipped(db_session: AsyncSession):
    leaver = await _person(db_session, "Leaver", active=False)
    await db_session.commit()
    rec = _Recorder()
    # Owner-less → skipped_no_owner.
    it_none = _item(None, days=5, rid=str(uuid.uuid4()))
    # Leaver owner in escalation → no active recipient → skipped_leaver.
    it_leaver = _item(leaver.id, days=5, rid=str(uuid.uuid4()), fresh=True)
    s = await run_expiry_alerts(db_session, resource_type=RT,
                                items=[it_none, it_leaver], config={}, notify=rec)
    assert rec.calls == []
    assert s["skipped_no_owner"] == 1
    assert s["skipped_leaver"] == 1