"""Terminal Zoom-account DELETE sweep (2026-08-04 — Zoom two-stage lifecycle).

Parity with the Google delete sweep: after ``_push_force_logout`` auto-
DEACTIVATES a leaver's Zoom account at leaver_date+1,
``hr.offboarding_zoom_account_delete`` hard-DELETES it once ``leaver_date +
delete_lead_days`` (default 2) local days have passed — for accounts we
confirmed deactivated and that are NOT retention-exempt. ⚠️ Zoom deletes are
PERMANENT (no undelete window), so these gates are the only safety net.

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_zoom_account"
)


def _today():
    # The delete sweep gates on ``_local_today_for_person`` (Asia/Shanghai for
    # a person with no location). Anchor the test's day math to the SAME zone
    # rather than the runner's system clock (UTC on CI) so the boundary case
    # doesn't flip in the 16:00–24:00 UTC window. See the Google delete test.
    return datetime.now(ZoneInfo("Asia/Shanghai")).date()


async def _person(db: AsyncSession, *, status: str = "left") -> IdentityPerson:
    # Every leaver gets an ACTIVE manager: since 2026-08-13 the terminal
    # Zoom delete refuses to run without a resolvable successor to receive
    # meetings / webinars / cloud recordings (Zoom deletes are permanent).
    # These tests exercise the *other* gates, so they need the handover gate
    # satisfied — see test_terminal_delete_handover_gate_2026_08_13.py for
    # the refusal path itself.
    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,
    zoom_account_status: str | None = "suspended",
    zoom_already_absent: bool = False,
    deletion_status: str | None = None,
    status: str = PROC_COMPLETED,
) -> OffboardingProcess:
    enforcement: dict | None = None
    if zoom_account_status is not None or deletion_status is not None:
        enforcement = {}
        if zoom_account_status is not None:
            enforcement["zoom_account"] = {
                "status": zoom_account_status,
                "already_absent": zoom_already_absent,
            }
        if deletion_status is not None:
            enforcement["zoom_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_zoom_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["zoom_account_deletion"]
    assert led["status"] == "deleted"

    # Idempotent: terminal ledger → second sweep does nothing.
    with patch(_DELETE, new=AsyncMock()) as m2:
        summary2 = await svc.sweep_zoom_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 so it IS scanned & skipped (not counted).
    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_zoom_account_deletions(trace_id="t")
    m.assert_not_awaited()
    assert summary["scanned"] == 0


@pytest.mark.asyncio
async def test_sweep_never_deletes_retained_account(
    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_zoom_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["zoom_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_zoom_account_deletions(trace_id="t")
    m.assert_not_awaited()
    assert summary["scanned"] == 0


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

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

    await db_session.refresh(proc)
    led = proc.session_enforcement["zoom_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:
    # Deactivate 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, zoom_already_absent=True,
    )
    await db_session.commit()

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

    await db_session.refresh(proc)
    assert (
        proc.session_enforcement["zoom_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": "500 zoom"}),
    ):
        summary = await svc.sweep_zoom_account_deletions(trace_id="t")

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

    await db_session.refresh(proc)
    assert (
        proc.session_enforcement["zoom_account_deletion"]["status"] == "failed"
    )
    with patch(
        _DELETE,
        new=AsyncMock(return_value={"status": "deleted", "already_absent": False}),
    ) as m2:
        await svc.sweep_zoom_account_deletions(trace_id="t")
    m2.assert_awaited_once()  # retried and succeeded


@pytest.mark.asyncio
async def test_sweep_deletes_without_transfer_when_recipient_lacks_webinar_license(
    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(side_effect=[
            {"status": "failed", "error": (
                "Zoom delete user 400: data receiver does not have Zoom "
                "Webinars licenses"
            )},
            {"status": "deleted", "already_absent": False},
        ]),
    ) as delete:
        summary = await svc.sweep_zoom_account_deletions(trace_id="t")

    assert summary["deleted"] == 1
    assert not summary["errors"]
    assert delete.await_count == 2
    assert delete.await_args_list[1].kwargs["successor_email"] is None
    await db_session.refresh(proc)
    ledger = proc.session_enforcement["zoom_account_deletion"]
    assert ledger["status"] == "deleted"
    assert ledger["data_transfer"] == "skipped_no_webinar_license"


# ── Recipient refusals Zoom raises on the transfer (2026-09-06) ───────
#
# Zoom validates ``transfer_email`` BEFORE it deletes, and refuses the
# whole call when the recipient cannot receive the content. Two members
# of that family, opposite treatment:
#
#   * no Webinars license → operator policy is delete-anyway, no
#     transfer (covered above, unchanged).
#   * anything else, chiefly "the data receiver does not exist" (a
#     successor with no Zoom account — 6 of 416 active people were in
#     that state in prod on 2026-09-06) → REFUSE. Deleting without a
#     transfer here is exactly the recordings loss DECISIONS 2026-08-13
#     reversed, and failing forever is the alert nobody reads.
#
# Before this, only the webinar sentence was recognised; every other
# recipient refusal fell into the generic ``failed`` branch and
# re-failed the cron nightly with no route to a terminal state.

_REMINDER_KIND_RECIPIENT = "del_blocked_zoom_recipient"


async def _admin(db: AsyncSession) -> IdentityPerson:
    """An active global_admin, so the page-out actually has a recipient.

    ``_notify_delete_blocked`` returns WITHOUT stamping its receipt when
    no admin exists (deliberately — so the page can still fire once one
    does), which would make a "paged once" assertion vacuously pass on
    an empty test DB.
    """
    from sqlalchemy import select

    from app.auth.models import AuthRole, AuthUserRole

    admin = await _person(db, status="active")
    role_id = await db.scalar(
        select(AuthRole.id).where(AuthRole.name == "global_admin"),
    )
    db.add(AuthUserRole(user_id=admin.id, role_id=role_id))
    await db.flush()
    return admin


async def _reminder_kinds(db: AsyncSession, proc_id: uuid.UUID) -> list[str]:
    from sqlalchemy import select

    from app.domains.offboarding.models import OffboardingReminderLog

    rows = await db.execute(
        select(OffboardingReminderLog.reminder_kind).where(
            OffboardingReminderLog.process_id == proc_id,
        )
    )
    return sorted(rows.scalars().all())


@pytest.mark.asyncio
async def test_sweep_refuses_delete_when_zoom_rejects_the_recipient(
    db_session: AsyncSession,
) -> None:
    """The driver-stamped tag is the contract, and a refusal is not a fault."""
    from app.integrations.drivers.zoom_software import (
        ZOOM_TAG_TRANSFER_RECIPIENT_INVALID,
    )

    await _admin(db_session)
    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": (
                'Zoom delete user 400 (code=200): {"code":200,"message":'
                '"Unable to transfer data because data receiver does not '
                'exist."} [transfer_email=mgr@x.io]'
            ),
            "transfer_refusal": ZOOM_TAG_TRANSFER_RECIPIENT_INVALID,
        }),
    ) as delete:
        summary = await svc.sweep_zoom_account_deletions(trace_id="t")

    # The irreversible call is made exactly ONCE — no no-transfer retry.
    # A second call here would delete the account and destroy the cloud
    # recordings the transfer exists to save.
    delete.assert_awaited_once()
    assert delete.await_args.kwargs["successor_email"] is not None

    # A safe refusal, counted apart from real faults so it neither fails
    # the cron nor drowns a genuine vendor error (DECISIONS 2026-08-13 d).
    assert summary["blocked_transfer_recipient"] == 1
    assert summary["failed"] == 0
    assert summary["deleted"] == 0
    assert not summary["errors"]

    await db_session.refresh(proc)
    led = proc.session_enforcement["zoom_account_deletion"]
    assert led["status"] == "blocked_transfer_recipient"
    assert led["transfer_refusal"] == ZOOM_TAG_TRANSFER_RECIPIENT_INVALID
    # Non-terminal: the account is still suspended and still deletable.
    assert led["status"] not in OffboardingService._DELETION_TERMINAL
    assert proc.session_enforcement["zoom_account"]["status"] == "suspended"

    # IT is paged, and under its own reminder kind so a process already
    # paged for "no successor" still gets told about this.
    assert await _reminder_kinds(db_session, proc.id) == [
        _REMINDER_KIND_RECIPIENT,
    ]


@pytest.mark.asyncio
async def test_recipient_refusal_recognised_without_the_tag(
    db_session: AsyncSession,
) -> None:
    """Message fallback: a result built by a path that predates the tag
    must still be refused, not deleted-without-transfer and not failed."""
    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": (
                "Zoom delete user 400 (code=200): Unable to transfer data "
                "because data receiver does not exist."
            ),
        }),
    ) as delete:
        summary = await svc.sweep_zoom_account_deletions(trace_id="t")

    delete.assert_awaited_once()
    assert summary["blocked_transfer_recipient"] == 1
    assert not summary["errors"]
    await db_session.refresh(proc)
    assert (
        proc.session_enforcement["zoom_account_deletion"]["status"]
        == "blocked_transfer_recipient"
    )


@pytest.mark.asyncio
async def test_recipient_refusal_pages_once_then_clears_itself(
    db_session: AsyncSession,
) -> None:
    """Repeats quietly, and needs no replay once IT fixes the successor."""
    await _admin(db_session)
    person = await _person(db_session)
    proc = await _process(db_session, person.id)
    await db_session.commit()

    svc = OffboardingService(db_session)
    refusal = AsyncMock(return_value={
        "status": "failed",
        "error": "Unable to transfer data because data receiver does not exist",
    })
    with patch(_DELETE, new=refusal):
        await svc.sweep_zoom_account_deletions(trace_id="t")
        second = await svc.sweep_zoom_account_deletions(trace_id="t")

    # Still retried every tick (non-terminal) …
    assert refusal.await_count == 2
    assert second["blocked_transfer_recipient"] == 1
    # … but IT is told once, not once per night.
    assert await _reminder_kinds(db_session, proc.id) == [
        _REMINDER_KIND_RECIPIENT,
    ]

    # IT names a successor Zoom accepts → the next tick just works.
    with patch(
        _DELETE,
        new=AsyncMock(return_value={"status": "deleted", "already_absent": False}),
    ) as ok:
        third = await svc.sweep_zoom_account_deletions(trace_id="t")
    ok.assert_awaited_once()
    assert third["deleted"] == 1
    await db_session.refresh(proc)
    assert proc.session_enforcement["zoom_account_deletion"]["status"] == "deleted"


@pytest.mark.asyncio
async def test_non_recipient_vendor_failure_still_fails_loudly(
    db_session: AsyncSession,
) -> None:
    """The refusal branch must not swallow a real fault: a 500, or a
    refusal about the LEAVER rather than the receiver, stays an error."""
    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": "Zoom delete user 500: internal error",
            "transfer_refusal": None,
        }),
    ):
        summary = await svc.sweep_zoom_account_deletions(trace_id="t")

    assert summary["failed"] == 1
    assert summary["blocked_transfer_recipient"] == 0
    assert summary["errors"]
    await db_session.refresh(proc)
    assert proc.session_enforcement["zoom_account_deletion"]["status"] == "failed"
    assert await _reminder_kinds(db_session, proc.id) == []