"""Regression tests for ``OffboardingService.list_my_team_leavers``.

This is the non-admin lookout backing ``/my/team-leavers``, reached
from the Feishu SLA card's "Handle in Helix" button when the recipient
does not hold ``global_admin``. Three viewer relationships qualify
(mirroring the SLA dispatcher's non-admin recipient set):

* ``hrbp_person_id == viewer`` (URGENT / BREACH tier)
* ``manager_person_id == viewer`` (WARN tier, direct manager)
* Skip-level: ``manager_person_id`` of the leaver's manager equals the
  viewer (WARN tier, manager chain second hop)

The tests below pin each path in isolation plus the negative case
(unrelated viewer sees an empty list) and the terminal-state filter
(completed / cancelled processes excluded). Without these, the bare
SQL was a single backend round-trip nobody had ever exercised — Pass-3
review caught a MySQL-incompatible ``NULLS LAST`` clause this way; the
``test_orders_by_first_blocked_at_nullslast_safe`` assertion below
locks the workaround in.
"""
from __future__ import annotations

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

import pytest

from app.domains.offboarding.models import (
    PROC_BLOCKED,
    PROC_CANCELLED,
    PROC_COMPLETED,
    PROC_SCANNING,
    OffboardingProcess,
)
from app.domains.offboarding.service import OffboardingService
from app.identity.person_models import IdentityPerson


async def _mkperson(
    db,
    *,
    email: str,
    hrbp_person_id: uuid.UUID | None = None,
    manager_person_id: uuid.UUID | None = None,
) -> 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-{email}",
        hrbp_person_id=hrbp_person_id,
        manager_person_id=manager_person_id,
    )
    db.add(p)
    await db.flush()
    return p


async def _mkproc(
    db,
    *,
    person: IdentityPerson,
    status: str = PROC_BLOCKED,
    first_blocked_at: datetime | None = None,
) -> OffboardingProcess:
    proc = OffboardingProcess(
        id=uuid.uuid4(),
        person_id=person.id,
        leaver_date=date.today() - timedelta(days=1),
        status=status,
        first_blocked_at=first_blocked_at,
    )
    db.add(proc)
    await db.flush()
    return proc


@pytest.mark.asyncio
async def test_lists_leaver_where_viewer_is_hrbp(db_session):
    viewer = await _mkperson(db_session, email="hrbp@x.com")
    leaver = await _mkperson(
        db_session, email="leaver@x.com", hrbp_person_id=viewer.id,
    )
    await _mkproc(db_session, person=leaver)
    await db_session.commit()

    svc = OffboardingService(db_session)
    items, total = await svc.list_my_team_leavers(viewer.id)
    assert total == 1
    assert items[0].person_email == "leaver@x.com"


@pytest.mark.asyncio
async def test_lists_leaver_where_viewer_is_direct_manager(db_session):
    viewer = await _mkperson(db_session, email="mgr@x.com")
    leaver = await _mkperson(
        db_session, email="leaver@x.com", manager_person_id=viewer.id,
    )
    await _mkproc(db_session, person=leaver)
    await db_session.commit()

    svc = OffboardingService(db_session)
    items, total = await svc.list_my_team_leavers(viewer.id)
    assert total == 1
    assert items[0].person_email == "leaver@x.com"


@pytest.mark.asyncio
async def test_excludes_leaver_where_viewer_is_skip_level(db_session):
    """A skip-level (level+2) manager must NOT see an indirect report's
    offboarding row. The list mirrors the SLA dispatcher's non-admin
    recipients, and ``_dm_manager_chain`` notifies the DIRECT manager
    only — skip-level is deliberately excluded per the user decision on
    the lifecycle visualization ("经理链上级 / level+2 不需要知道这个事情").
    An earlier revision leaked these rows via a bogus "walks 2 levels"
    justification; this guards the closed hole.
    """
    viewer = await _mkperson(db_session, email="skip@x.com")
    direct = await _mkperson(
        db_session, email="direct@x.com", manager_person_id=viewer.id,
    )
    leaver = await _mkperson(
        db_session, email="leaver@x.com", manager_person_id=direct.id,
    )
    await _mkproc(db_session, person=leaver)
    await db_session.commit()

    svc = OffboardingService(db_session)
    items, total = await svc.list_my_team_leavers(viewer.id)
    assert total == 0
    assert items == []


@pytest.mark.asyncio
async def test_unrelated_viewer_sees_empty_list(db_session):
    unrelated = await _mkperson(db_session, email="random@x.com")
    leaver_hrbp = await _mkperson(db_session, email="hrbp-target@x.com")
    leaver = await _mkperson(
        db_session, email="leaver@x.com", hrbp_person_id=leaver_hrbp.id,
    )
    await _mkproc(db_session, person=leaver)
    await db_session.commit()

    svc = OffboardingService(db_session)
    items, total = await svc.list_my_team_leavers(unrelated.id)
    assert total == 0
    assert items == []


@pytest.mark.asyncio
async def test_excludes_completed_and_cancelled_processes(db_session):
    viewer = await _mkperson(db_session, email="hrbp@x.com")
    leaver_done = await _mkperson(
        db_session, email="done@x.com", hrbp_person_id=viewer.id,
    )
    leaver_cancel = await _mkperson(
        db_session, email="cancel@x.com", hrbp_person_id=viewer.id,
    )
    leaver_active = await _mkperson(
        db_session, email="active@x.com", hrbp_person_id=viewer.id,
    )
    await _mkproc(db_session, person=leaver_done, status=PROC_COMPLETED)
    await _mkproc(db_session, person=leaver_cancel, status=PROC_CANCELLED)
    await _mkproc(db_session, person=leaver_active, status=PROC_BLOCKED)
    await db_session.commit()

    svc = OffboardingService(db_session)
    items, total = await svc.list_my_team_leavers(viewer.id)
    assert total == 1
    assert items[0].person_email == "active@x.com"


@pytest.mark.asyncio
async def test_orders_by_first_blocked_at_with_nulls_last_emulation(db_session):
    """The ``NULLS LAST`` emulation must (a) execute on MySQL without
    1064, and (b) place null ``first_blocked_at`` rows after the dated
    ones. SQLAlchemy's ``.nullslast()`` would emit literal
    ``NULLS LAST`` syntax that MySQL/TiDB reject, hence the
    ``is_(None).asc() THEN .desc()`` pair in the service.
    """
    viewer = await _mkperson(db_session, email="hrbp@x.com")
    leaver_unblocked = await _mkperson(
        db_session, email="unblocked@x.com", hrbp_person_id=viewer.id,
    )
    leaver_old_block = await _mkperson(
        db_session, email="old@x.com", hrbp_person_id=viewer.id,
    )
    leaver_recent_block = await _mkperson(
        db_session, email="recent@x.com", hrbp_person_id=viewer.id,
    )
    now = datetime.now(UTC)
    await _mkproc(
        db_session, person=leaver_unblocked,
        status=PROC_SCANNING, first_blocked_at=None,
    )
    await _mkproc(
        db_session, person=leaver_old_block,
        status=PROC_BLOCKED, first_blocked_at=now - timedelta(hours=72),
    )
    await _mkproc(
        db_session, person=leaver_recent_block,
        status=PROC_BLOCKED, first_blocked_at=now - timedelta(hours=2),
    )
    await db_session.commit()

    svc = OffboardingService(db_session)
    items, total = await svc.list_my_team_leavers(viewer.id)
    assert total == 3
    # Dated rows first, most-recent-first; null last.
    assert items[0].person_email == "recent@x.com"
    assert items[1].person_email == "old@x.com"
    assert items[2].person_email == "unblocked@x.com"


@pytest.mark.asyncio
async def test_does_not_list_own_offboarding(db_session):
    """Defensive: even if mis-configured data points hrbp_person_id at
    the viewer themselves, the viewer's own offboarding does not appear
    in their lookout."""
    viewer = await _mkperson(db_session, email="self@x.com")
    # Mis-config: viewer is their own HRBP
    viewer.hrbp_person_id = viewer.id
    await db_session.flush()
    await _mkproc(db_session, person=viewer)
    await db_session.commit()

    svc = OffboardingService(db_session)
    items, total = await svc.list_my_team_leavers(viewer.id)
    assert total == 0
    assert items == []