"""Terminal Google-account DELETE sweep (2026-07-08 ADR「离职号最终删除」).

After the 2026-07-01 翻案 auto-SUSPENDS a leaver's Google account at
leaver_date+1, ``hr.offboarding_google_account_delete`` hard-DELETES it once
``leaver_date + delete_lead_days`` (default 2) local days have passed — for
accounts we confirmed suspended and that are NOT retention-exempt. Google's
~20-day recovery window is the mistake safety net; no data transfer.

These tests lock the gates: deletes only the eligible/suspended/due, is
idempotent (terminal ledger), and never touches retained / rehired /
not-yet-due / not-confirmed-suspended accounts.
"""
from __future__ import annotations

import uuid
from datetime import datetime, timedelta
from unittest.mock import AsyncMock, patch
from zoneinfo import ZoneInfo

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.asset.models import ExternalAccountRegistry
from app.domains.offboarding.models import (
    PROC_COMPLETED,
    OffboardingProcess,
)
from app.domains.offboarding.service import OffboardingService
from app.identity.person_models import IdentityPerson

_DELETE = (
    "app.core.coordinators.account_lifecycle_coordinator.delete_google_account"
)


def _today():
    # The delete sweep gates on ``_local_today_for_person`` (Asia/Shanghai
    # for a person with no location — see ``resolve_timezone``). Anchor the
    # test's day math to the SAME zone rather than the runner's system clock
    # (``date.today()`` = UTC on CI). Otherwise the boundary case
    # (leaver_date-1, lead 2 → due today+1) flips to "due" whenever CI runs
    # in the 16:00–24:00 UTC window (Beijing already on the next calendar
    # day), awaiting the delete mock and failing intermittently.
    return datetime.now(ZoneInfo("Asia/Shanghai")).date()


@pytest.fixture(autouse=True)
def _handover_satisfied():
    """Since 2026-08-13 the terminal delete is gated on a proven
    Drive/Calendar handover. These tests cover the OTHER gates (due-date,
    retention, rehire, suspend-confirmed, idempotency), so the handover is
    stubbed as already-completed here; the gate's own behaviour — refusing
    to delete without a successor, performing the transfer, resuming an
    in-flight one — lives in
    ``test_terminal_delete_handover_gate_2026_08_13.py``.
    """
    with patch(
        "app.core.coordinators.account_lifecycle_coordinator."
        "ensure_google_data_transfer",
        new=AsyncMock(
            return_value={"status": "completed", "transfer_id": "AKrEt-stub"},
        ),
    ):
        yield


async def _person(db: AsyncSession, *, status: str = "left") -> IdentityPerson:
    # Leavers get an ACTIVE manager so the handover gate resolves a
    # successor (see ``_handover_satisfied``).
    mid = uuid.uuid4()
    manager = IdentityPerson(
        id=mid,
        source_system="successfactors",
        hris_person_id=f"sf-{mid.hex[:8]}",
        work_email=f"mgr-{mid.hex[:8]}@x.io",
        display_name="Manager",
        employment_status="active",
    )
    db.add(manager)
    await db.flush()
    pid = uuid.uuid4()
    p = IdentityPerson(
        id=pid,
        source_system="successfactors",
        hris_person_id=f"sf-{pid.hex[:8]}",
        work_email=f"{pid.hex[:8]}@x.io",
        display_name="Leaver",
        employment_status=status,
        manager_person_id=mid,
        termination_date=_today() - timedelta(days=4),
    )
    db.add(p)
    await db.flush()
    return p


async def _process(
    db: AsyncSession,
    person_id: uuid.UUID,
    *,
    leaver_offset_days: int = -4,
    google_account_status: str | None = "suspended",
    deletion_status: str | None = None,
    status: str = PROC_COMPLETED,
) -> OffboardingProcess:
    enforcement: dict | None = None
    if google_account_status is not None or deletion_status is not None:
        enforcement = {}
        if google_account_status is not None:
            enforcement["google_account"] = {"status": google_account_status}
        if deletion_status is not None:
            enforcement["google_account_deletion"] = {"status": deletion_status}
    proc = OffboardingProcess(
        id=uuid.uuid4(),
        person_id=person_id,
        status=status,
        leaver_date=_today() + timedelta(days=leaver_offset_days),
        session_enforcement=enforcement,
    )
    db.add(proc)
    await db.flush()
    return proc


@pytest.mark.asyncio
async def test_sweep_deletes_eligible_suspended_account(
    db_session: AsyncSession,
) -> None:
    person = await _person(db_session)
    proc = await _process(db_session, person.id)  # leaver_date -4 → due (lead 2)
    await db_session.commit()

    svc = OffboardingService(db_session)
    with patch(
        _DELETE,
        new=AsyncMock(return_value={"status": "deleted", "already_absent": False}),
    ) as m:
        summary = await svc.sweep_google_account_deletions(trace_id="t")

    m.assert_awaited_once()
    assert summary["deleted"] == 1
    assert summary["scanned"] == 1
    assert not summary["errors"]

    await db_session.refresh(proc)
    led = proc.session_enforcement["google_account_deletion"]
    assert led["status"] == "deleted"

    # Idempotent: terminal ledger → second sweep does nothing.
    with patch(_DELETE, new=AsyncMock()) as m2:
        summary2 = await svc.sweep_google_account_deletions(trace_id="t")
    m2.assert_not_awaited()
    assert summary2["scanned"] == 0


@pytest.mark.asyncio
async def test_sweep_skips_not_yet_due(db_session: AsyncSession) -> None:
    # leaver_date = today-1, lead 2 → due today+1 → not yet due, but still
    # inside the UTC prefilter (today-1 <= today-1) so it IS scanned & skipped.
    person = await _person(db_session)
    await _process(db_session, person.id, leaver_offset_days=-1)
    await db_session.commit()

    svc = OffboardingService(db_session)
    with patch(_DELETE, new=AsyncMock()) as m:
        summary = await svc.sweep_google_account_deletions(trace_id="t")
    m.assert_not_awaited()
    assert summary["scanned"] == 0  # not_actionable is not counted


@pytest.mark.asyncio
async def test_sweep_never_deletes_retained_mailbox(
    db_session: AsyncSession,
) -> None:
    person = await _person(db_session)
    proc = await _process(db_session, person.id)
    db_session.add(
        ExternalAccountRegistry(
            id=uuid.uuid4(),
            email=person.work_email,
            classification="shared_service",
        )
    )
    await db_session.commit()

    svc = OffboardingService(db_session)
    with patch(_DELETE, new=AsyncMock()) as m:
        summary = await svc.sweep_google_account_deletions(trace_id="t")
    m.assert_not_awaited()  # retained → never reaches the vendor delete
    assert summary["retained"] == 1

    await db_session.refresh(proc)
    assert (
        proc.session_enforcement["google_account_deletion"]["status"] == "retained"
    )


@pytest.mark.asyncio
async def test_sweep_skips_rehired_active(db_session: AsyncSession) -> None:
    person = await _person(db_session, status="active")
    await _process(db_session, person.id)
    await db_session.commit()

    svc = OffboardingService(db_session)
    with patch(_DELETE, new=AsyncMock()) as m:
        summary = await svc.sweep_google_account_deletions(trace_id="t")
    m.assert_not_awaited()
    assert summary["scanned"] == 0


@pytest.mark.asyncio
async def test_sweep_skips_when_suspend_not_confirmed(
    db_session: AsyncSession,
) -> None:
    # Suspend failed / never confirmed → we must NOT delete (delete can never
    # outrun the suspend it depends on). Non-terminal skip → retried later.
    person = await _person(db_session)
    proc = await _process(db_session, person.id, google_account_status="failed")
    await db_session.commit()

    svc = OffboardingService(db_session)
    with patch(_DELETE, new=AsyncMock()) as m:
        summary = await svc.sweep_google_account_deletions(trace_id="t")
    m.assert_not_awaited()
    assert summary["skipped"] == 1

    await db_session.refresh(proc)
    led = proc.session_enforcement["google_account_deletion"]
    assert led["status"] == "skipped_not_suspended"  # non-terminal


@pytest.mark.asyncio
async def test_sweep_already_absent_account_is_terminal_no_call(
    db_session: AsyncSession,
) -> None:
    # Suspend step already found the account gone → mark deletion terminal
    # without a vendor call.
    person = await _person(db_session)
    proc = await _process(
        db_session, person.id, google_account_status="already_absent",
    )
    await db_session.commit()

    svc = OffboardingService(db_session)
    with patch(_DELETE, new=AsyncMock()) as m:
        summary = await svc.sweep_google_account_deletions(trace_id="t")
    m.assert_not_awaited()
    assert summary["already_absent"] == 1

    await db_session.refresh(proc)
    assert (
        proc.session_enforcement["google_account_deletion"]["status"]
        == "already_absent"
    )


@pytest.mark.asyncio
async def test_sweep_vendor_failure_is_non_terminal_and_trips_errors(
    db_session: AsyncSession,
) -> None:
    person = await _person(db_session)
    proc = await _process(db_session, person.id)
    await db_session.commit()

    svc = OffboardingService(db_session)
    with patch(
        _DELETE,
        new=AsyncMock(return_value={"status": "failed", "error": "403 scope"}),
    ):
        summary = await svc.sweep_google_account_deletions(trace_id="t")

    assert summary["failed"] == 1
    assert summary["errors"]  # → scheduled-job wrapper raises → run 'failed'

    await db_session.refresh(proc)
    # Non-terminal: a later sweep retries once the SA scope is fixed.
    assert (
        proc.session_enforcement["google_account_deletion"]["status"] == "failed"
    )
    with patch(
        _DELETE,
        new=AsyncMock(return_value={"status": "deleted", "already_absent": False}),
    ) as m2:
        await svc.sweep_google_account_deletions(trace_id="t")
    m2.assert_awaited_once()  # retried and succeeded