"""Tests for ``hris.absent_person_check`` — full-roster reconciliation.

Round-12 hardening 2026-05-08: detect Helix-active persons who are
missing from the live SuccessFactors roster (admin hard-deleted SF
user, webhook silently failed, tenant migration dropped rows).

Migration 222 upgrade 2026-05-30: detection now stamps a soft
tombstone (``hris_record_state='purged'``) and writes an
``hr_lifecycle_events`` row so the trail persists without
hard-deleting the IdentityPerson row. The brake / partial-failure /
percentage-ceiling behaviour is unchanged — those tests still cover
the safety net.
"""
from __future__ import annotations

import uuid
from typing import Any
from unittest.mock import patch

import pytest
from sqlalchemy import select

from app.domains.hr.models import HrLifecycleEvent
from app.identity.person_models import IdentityPerson
from app.scheduled_jobs.handlers import run_hris_absent_person_check


@pytest.mark.asyncio
async def test_no_instances_returns_empty_summary(db_session) -> None:
    """No HRIS integrations configured → handler returns the empty
    summary cleanly. (Same shape as ``run_hris_incremental_sync``.)"""
    with patch(
        "app.integrations.compat.list_hris_instances",
        return_value=[],
    ):
        summary = await run_hris_absent_person_check(db_session, {})
    assert summary["instances_checked"] == 0
    assert summary["candidates_total"] == 0
    assert summary["alerted_persons"] == []
    assert summary["brake_tripped"] == []
    assert summary["errors"] == []


@pytest.mark.asyncio
async def test_detects_helix_active_person_missing_from_sf(db_session) -> None:
    """Helix-active person whose ``hris_person_id`` is no longer in the
    SF roster gets flagged. Below mass-leaver threshold so the alert
    fires (not braked)."""
    instance_id = uuid.uuid4()
    present = IdentityPerson(
        id=uuid.uuid4(),
        work_email="present@x.io",
        display_name="Present",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-PRESENT",
        hris_instance_id=instance_id,
    )
    absent = IdentityPerson(
        id=uuid.uuid4(),
        work_email="absent@x.io",
        display_name="Absent",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-ABSENT",
        hris_instance_id=instance_id,
    )
    db_session.add_all([present, absent])
    await db_session.commit()

    class _FakeIntegrationRow:
        def __init__(self) -> None:
            self.id = instance_id
            self.slug = "test-instance"

    fake_instances = [
        (_FakeIntegrationRow(), {"base_url": "https://sf.example/odata"}),
    ]

    class _FakeClient:
        def __init__(self, *_, **__) -> None:
            pass

        async def list_current_jobs(self, *, limit: Any = None) -> list[dict[str, Any]]:
            # Only PRESENT remains in the SF roster; ABSENT is gone.
            return [{"userId": "sf-PRESENT"}]

    def _fake_auth(_config: dict[str, Any]) -> Any:
        return object()

    with (
        patch(
            "app.integrations.compat.list_hris_instances",
            return_value=fake_instances,
        ),
        patch(
            "app.identity.successfactors_client.SuccessFactorsClient",
            _FakeClient,
        ),
        patch(
            "app.identity.successfactors_auth.build_auth_from_config",
            _fake_auth,
        ),
    ):
        summary = await run_hris_absent_person_check(db_session, {})

    assert summary["instances_checked"] == 1
    assert summary["candidates_total"] == 1
    flagged_ids = {p["hris_person_id"] for p in summary["alerted_persons"]}
    assert flagged_ids == {"sf-ABSENT"}
    assert summary["brake_tripped"] == []
    # Migration 222: detection stamps the tombstone + lifecycle event.
    # ``purged_persons`` counter mirrors the number of rows we actually
    # flipped from ``present`` → ``purged``.
    assert summary["purged_persons"] == 1

    await db_session.refresh(absent)
    await db_session.refresh(present)
    assert absent.hris_record_state == "purged"
    assert absent.hris_purged_at is not None
    assert absent.hris_purged_reason == "absent_from_full_sync"
    # employment_status is intentionally NOT touched — that path runs
    # offboarding side-effects (asset transfer, SSO suspend, Feishu
    # notice) and must be a deliberate admin decision, not a side-effect
    # of reconciliation.
    assert absent.employment_status == "active"
    # The still-present person is untouched.
    assert present.hris_record_state == "present"
    assert present.hris_purged_at is None

    lifecycle_rows = (await db_session.execute(
        select(HrLifecycleEvent).where(
            HrLifecycleEvent.person_id == absent.id,
            HrLifecycleEvent.event_type == "hris_purged",
        )
    )).scalars().all()
    assert len(lifecycle_rows) == 1
    assert lifecycle_rows[0].detail["reason"] == "absent_from_full_sync"
    assert lifecycle_rows[0].detail["hris_person_id"] == "sf-ABSENT"


@pytest.mark.asyncio
async def test_already_purged_person_is_not_re_alerted(db_session) -> None:
    """Idempotency (migration 222): a person tombstoned in a prior run
    must NOT reappear as a fresh candidate. Without the
    ``hris_record_state='present'`` filter on the candidate query, every
    subsequent run would re-stamp the same person, re-write a lifecycle
    event, and re-audit — burying any genuinely new disappearance under
    accumulated noise."""
    from datetime import UTC, datetime
    instance_id = uuid.uuid4()
    # Pre-purged in a prior run.
    already_purged = IdentityPerson(
        id=uuid.uuid4(),
        work_email="ghost@x.io",
        display_name="Ghost",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-GHOST",
        hris_instance_id=instance_id,
        hris_record_state="purged",
        hris_purged_at=datetime.now(UTC),
        hris_purged_reason="absent_from_full_sync",
    )
    db_session.add(already_purged)
    await db_session.commit()

    class _FakeIntegrationRow:
        id = instance_id
        slug = "test-instance"

    class _FakeClient:
        def __init__(self, *_, **__) -> None:
            pass

        async def list_current_jobs(self, *, limit: Any = None) -> list[dict[str, Any]]:
            # SF still doesn't have the ghost — but the handler must
            # not look at her again.
            return [{"userId": "sf-OTHER"}]

    def _fake_auth(_config: dict[str, Any]) -> Any:
        return object()

    with (
        patch(
            "app.integrations.compat.list_hris_instances",
            return_value=[(_FakeIntegrationRow(), {"base_url": "https://sf.example/odata"})],
        ),
        patch(
            "app.identity.successfactors_client.SuccessFactorsClient",
            _FakeClient,
        ),
        patch(
            "app.identity.successfactors_auth.build_auth_from_config",
            _fake_auth,
        ),
    ):
        summary = await run_hris_absent_person_check(db_session, {})

    # The instance was checked but no fresh candidate appeared.
    assert summary["instances_checked"] == 1
    assert summary["candidates_total"] == 0
    assert summary["purged_persons"] == 0
    assert summary["alerted_persons"] == []
    # And the lifecycle table didn't grow — re-running must not create
    # a duplicate ``hris_purged`` row.
    lifecycle_rows = (await db_session.execute(
        select(HrLifecycleEvent).where(
            HrLifecycleEvent.person_id == already_purged.id,
            HrLifecycleEvent.event_type == "hris_purged",
        )
    )).scalars().all()
    assert lifecycle_rows == []


@pytest.mark.asyncio
async def test_brake_trips_on_percentage_ceiling_below_floor(db_session) -> None:
    """Round 47 hardening (2026-05-23): even when the absolute floor of
    25 isn't crossed, the brake MUST trip when > 25% of an instance has
    vanished in a single tick. A 10-person instance with 9 absent
    previously slipped the 25-floor and was about to mass-flag 90% of
    the roster as leavers."""
    instance_id = uuid.uuid4()
    # 10 Helix-active persons; SF returns only 1 (the other 9 are
    # "absent"). 9 < 25 (floor) but 9/10 = 90% > 25% (ceiling).
    persons = [
        IdentityPerson(
            id=uuid.uuid4(),
            work_email=f"small{i}@x.io",
            display_name=f"S{i}",
            employment_status="active",
            source_system="successfactors",
            hris_person_id=f"sf-S{i}",
            hris_instance_id=instance_id,
        )
        for i in range(10)
    ]
    db_session.add_all(persons)
    await db_session.commit()

    class _FakeIntegrationRow:
        id = instance_id
        slug = "tiny-instance"

    fake_instances = [
        (_FakeIntegrationRow(), {"base_url": "https://sf.example/odata"}),
    ]

    class _FakeClient:
        def __init__(self, *_, **__) -> None:
            pass

        async def list_current_jobs(self, *, limit: Any = None) -> list[dict[str, Any]]:
            # Only one of the ten persons remains in the SF roster.
            return [{"userId": "sf-S0"}]

    def _fake_auth(_config: dict[str, Any]) -> Any:
        return object()

    with (
        patch(
            "app.integrations.compat.list_hris_instances",
            return_value=fake_instances,
        ),
        patch(
            "app.identity.successfactors_client.SuccessFactorsClient",
            _FakeClient,
        ),
        patch(
            "app.identity.successfactors_auth.build_auth_from_config",
            _fake_auth,
        ),
    ):
        summary = await run_hris_absent_person_check(db_session, {})

    # Brake fired despite candidate_count (9) < absolute floor (25).
    assert summary["candidates_total"] == 0
    assert summary["alerted_persons"] == []
    assert len(summary["brake_tripped"]) == 1
    assert summary["brake_tripped"][0]["candidate_count"] == 9


@pytest.mark.asyncio
async def test_partial_failure_raises_for_failure_notifier(db_session) -> None:
    """Round 47 hardening (2026-05-23): even when 1 of N instances
    succeeds, the handler MUST raise if any instance errored. The
    platform-wide rule (``feedback_handler_status_must_raise``) is
    that ``maybe_notify_failure`` only fires on raised exceptions —
    a 'partial success' summary with errors hidden in a dict field
    silently lost the original ``positionNameEN`` 400 storm for 10
    days."""
    instance_ok = uuid.uuid4()
    instance_fail = uuid.uuid4()
    # One Helix person on each instance, both present in their
    # respective SF roster (the OK instance) or simulated to fail.
    db_session.add_all([
        IdentityPerson(
            id=uuid.uuid4(),
            work_email="ok@x.io",
            display_name="OK",
            employment_status="active",
            source_system="successfactors",
            hris_person_id="sf-OK",
            hris_instance_id=instance_ok,
        ),
        IdentityPerson(
            id=uuid.uuid4(),
            work_email="fail@x.io",
            display_name="FAIL",
            employment_status="active",
            source_system="successfactors",
            hris_person_id="sf-FAIL",
            hris_instance_id=instance_fail,
        ),
    ])
    await db_session.commit()

    class _OkRow:
        id = instance_ok
        slug = "ok-instance"

    class _FailRow:
        id = instance_fail
        slug = "fail-instance"

    fake_instances = [
        (_OkRow(), {"base_url": "https://ok.example/odata"}),
        (_FailRow(), {"base_url": "https://fail.example/odata"}),
    ]

    class _FakeClient:
        def __init__(self, *, base_url: str, **_: Any) -> None:
            self.base_url = base_url

        async def list_current_jobs(self, *, limit: Any = None) -> list[dict[str, Any]]:
            if "ok.example" in self.base_url:
                return [{"userId": "sf-OK"}]
            raise RuntimeError("simulated SF 503")

    def _fake_auth(_config: dict[str, Any]) -> Any:
        return object()

    with (
        patch(
            "app.integrations.compat.list_hris_instances",
            return_value=fake_instances,
        ),
        patch(
            "app.identity.successfactors_client.SuccessFactorsClient",
            _FakeClient,
        ),
        patch(
            "app.identity.successfactors_auth.build_auth_from_config",
            _fake_auth,
        ),
    ):
        with pytest.raises(RuntimeError) as exc:
            await run_hris_absent_person_check(db_session, {})

    msg = str(exc.value)
    assert "1/2 instances failed" in msg
    assert "fail-instance" in msg


@pytest.mark.asyncio
async def test_brake_trips_on_mass_absence(db_session) -> None:
    """If the candidate count exceeds ``max(25, 5%)`` of the instance's
    Helix-active count, the alert is suppressed and a brake row is
    recorded — protects against an SF outage / paginator truncation
    surfacing a thousand spurious leaver candidates."""
    instance_id = uuid.uuid4()
    # 30 Helix-active persons, all absent from SF → candidate count
    # 30 > brake floor 25.
    persons = [
        IdentityPerson(
            id=uuid.uuid4(),
            work_email=f"absent{i}@x.io",
            display_name=f"A{i}",
            employment_status="active",
            source_system="successfactors",
            hris_person_id=f"sf-A{i}",
            hris_instance_id=instance_id,
        )
        for i in range(30)
    ]
    db_session.add_all(persons)
    await db_session.commit()

    class _FakeIntegrationRow:
        id = instance_id
        slug = "outage-instance"

    fake_instances = [
        (_FakeIntegrationRow(), {"base_url": "https://sf.example/odata"}),
    ]

    class _FakeClient:
        def __init__(self, *_, **__) -> None:
            pass

        async def list_current_jobs(self, *, limit: Any = None) -> list[dict[str, Any]]:
            # SF returned 5 random user IDs (none matching ours) — looks
            # like a healthy non-empty roster but everyone we know is
            # missing. Should trip the brake.
            return [{"userId": f"sf-other-{i}"} for i in range(5)]

    def _fake_auth(_config: dict[str, Any]) -> Any:
        return object()

    with (
        patch(
            "app.integrations.compat.list_hris_instances",
            return_value=fake_instances,
        ),
        patch(
            "app.identity.successfactors_client.SuccessFactorsClient",
            _FakeClient,
        ),
        patch(
            "app.identity.successfactors_auth.build_auth_from_config",
            _fake_auth,
        ),
    ):
        summary = await run_hris_absent_person_check(db_session, {})

    assert summary["instances_checked"] == 1
    assert summary["candidates_total"] == 0
    assert summary["alerted_persons"] == []
    assert len(summary["brake_tripped"]) == 1
    assert summary["brake_tripped"][0]["candidate_count"] == 30


@pytest.mark.asyncio
async def test_absent_check_skips_day_one_hire_shanghai_boundary(
    db_session,
) -> None:
    """2026-07-16 PR-audit FIX 4: the future-hire exemption must use
    ``hire_date >= today`` anchored to the Asia/Shanghai date — the same
    boundary convention #1952 established for the full/incremental sync
    (and ``classify_inactive_sf_user``). The pre-fix strict
    ``hire_date > today_utc`` left a DAY-ONE hire unprotected: SF-DC/UTC
    date skew can keep their effective-today EmpJob row out of the
    roster on their first morning, and they'd be tombstoned + paged as
    "purged from HRIS" the day they start. A genuinely-absent past hire
    in the same batch must still be flagged."""
    from datetime import datetime, timedelta
    from zoneinfo import ZoneInfo

    instance_id = uuid.uuid4()
    shanghai_today = datetime.now(ZoneInfo("Asia/Shanghai")).date()
    day_one = IdentityPerson(
        id=uuid.uuid4(),
        work_email="dayone@x.io",
        display_name="Day One Hire",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-DAYONE",
        hris_instance_id=instance_id,
        hire_date=shanghai_today,  # starts TODAY (Shanghai business day)
    )
    past_absent = IdentityPerson(
        id=uuid.uuid4(),
        work_email="pastgone2@x.io",
        display_name="Past Gone 2",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-PASTGONE2",
        hris_instance_id=instance_id,
        hire_date=shanghai_today - timedelta(days=365),
    )
    db_session.add_all([day_one, past_absent])
    await db_session.commit()

    class _FakeIntegrationRow:
        def __init__(self) -> None:
            self.id = instance_id
            self.slug = "test-instance"

    class _FakeClient:
        def __init__(self, *_, **__) -> None:
            pass

        async def list_current_jobs(self, *, limit: Any = None) -> list[dict[str, Any]]:
            # Neither person appears — the day-one hire because of
            # SF-DC/UTC skew (structurally absent though alive), the
            # past hire because they were genuinely hard-deleted.
            return [{"userId": "sf-SOMEONE-ELSE"}]

    def _fake_auth(_config: dict[str, Any]) -> Any:
        return object()

    with (
        patch(
            "app.integrations.compat.list_hris_instances",
            return_value=[
                (_FakeIntegrationRow(), {"base_url": "https://sf.example/odata"}),
            ],
        ),
        patch(
            "app.identity.successfactors_client.SuccessFactorsClient",
            _FakeClient,
        ),
        patch(
            "app.identity.successfactors_auth.build_auth_from_config",
            _fake_auth,
        ),
    ):
        summary = await run_hris_absent_person_check(db_session, {})

    flagged = {p["hris_person_id"] for p in summary["alerted_persons"]}
    assert "sf-DAYONE" not in flagged, (
        "Day-one hire (hire_date == Shanghai today) was tombstoned — the "
        ">= / Asia/Shanghai boundary (#1952 convention) regressed to a "
        "strict UTC comparison."
    )
    assert flagged == {"sf-PASTGONE2"}
    assert summary["purged_persons"] == 1

    await db_session.refresh(day_one)
    assert day_one.hris_record_state == "present"
    assert day_one.hris_purged_at is None


# ── Future-hire cross-check (2026-07-17, P10019 post-mortem) ──────────
#
# A future hire (hire_date still ahead) has no effective-today EmpJob, so
# it is structurally absent from ``list_current_jobs`` whether alive OR
# cancelled in SF. The roster diff must NOT tombstone it (P10012/P10014).
# Instead each future hire is verified against SF by userId via
# ``get_users``; a userId SF no longer returns is a cancelled requisition
# (P10019) and — having no accounts/assets to offboard — is soft-deleted.

from datetime import UTC, date, datetime, timedelta  # noqa: E402

from app.identity.person_models import IdentityExternalIdentity  # noqa: E402


def _future_date() -> date:
    return date.today() + timedelta(days=5)


class _FutureHireIntegrationRow:
    def __init__(self, instance_id) -> None:
        self.id = instance_id
        self.slug = "test-instance"


def _fake_auth(_config):  # noqa: ANN001
    return object()


@pytest.mark.asyncio
async def test_cancelled_future_hire_soft_deleted_after_grace(db_session) -> None:
    """A future hire SF no longer returns is soft-deleted — but only AFTER the
    absence persists past the grace window (debounce, 2026-07-20 P10019). The
    FIRST absence merely records ``hris_absent_since``; a transient SF flicker
    that returns next run never gets deleted. A future hire SF still returns is
    left fully intact. Neither is ever surfaced by the roster diff."""
    instance_id = uuid.uuid4()
    cancelled = IdentityPerson(
        id=uuid.uuid4(),
        work_email="unknown@pingcap.com",
        display_name="Unknown",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="P10019",
        hris_instance_id=instance_id,
        hire_date=_future_date(),
        hris_record_state="present",
    )
    live_future = IdentityPerson(
        id=uuid.uuid4(),
        work_email="yihan.mo@pingcap.com",
        display_name="莫亦寒 Yihan Mo",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="P10014",
        hris_instance_id=instance_id,
        hire_date=_future_date(),
        hris_record_state="present",
    )
    # SF identity for the cancelled hire — must be soft-deleted in lockstep.
    cancelled_ident = IdentityExternalIdentity(
        id=uuid.uuid4(),
        person_id=cancelled.id,
        provider="successfactors",
        provider_subject="P10019",
        identity_status="active",
    )
    db_session.add_all([cancelled, live_future, cancelled_ident])
    await db_session.commit()

    class _FakeClient:
        def __init__(self, *_, **__) -> None:
            pass

        async def list_current_jobs(self, *, limit=None):  # noqa: ANN001
            # Neither future hire has an effective-today job row.
            return [{"userId": "sf-SOMEONE-ELSE"}]

        async def get_users(self, user_ids):  # noqa: ANN001
            # SF still knows P10014 (live) but NOT P10019 (cancelled).
            return [
                {"userId": "P10014", "displayName": "莫亦寒 Yihan Mo"}
            ]

    with (
        patch("app.integrations.compat.list_hris_instances",
              return_value=[(_FutureHireIntegrationRow(instance_id),
                             {"base_url": "https://sf.example/odata"})]),
        patch("app.identity.successfactors_client.SuccessFactorsClient",
              _FakeClient),
        patch("app.identity.successfactors_auth.build_auth_from_config",
              _fake_auth),
    ):
        # Run 1: FIRST absence → debounced. Grace recorded, NOT deleted.
        summary1 = await run_hris_absent_person_check(db_session, {})
        assert summary1["cancelled_future_hires"] == 0
        assert summary1["future_hire_absence_grace_started"] == 1
        await db_session.refresh(cancelled)
        assert cancelled.deleted_at is None
        assert cancelled.hris_absent_since is not None
        assert cancelled.hris_record_state == "present"

        # Backdate the first-seen-absent marker past the grace window, then
        # re-run (still absent) → NOW soft-deleted.
        cancelled.hris_absent_since = datetime.now(UTC) - timedelta(hours=21)
        await db_session.commit()
        summary = await run_hris_absent_person_check(db_session, {})

    # The roster diff itself flagged nobody (both are future hires).
    assert summary["candidates_total"] == 0
    assert summary["purged_persons"] == 0
    assert summary["cancelled_future_hires"] == 1

    await db_session.refresh(cancelled)
    await db_session.refresh(live_future)
    await db_session.refresh(cancelled_ident)

    # Cancelled hire: soft-deleted + tombstoned + identity soft-deleted.
    assert cancelled.deleted_at is not None
    assert cancelled.hris_record_state == "purged"
    assert cancelled.hris_purged_reason == "cancelled_future_hire_absent_from_sf"
    assert cancelled_ident.deleted_at is not None

    # Live future hire: completely untouched (never even entered grace).
    assert live_future.deleted_at is None
    assert live_future.hris_record_state == "present"
    assert live_future.hris_purged_at is None
    assert live_future.hris_absent_since is None

    lifecycle_rows = (await db_session.execute(
        select(HrLifecycleEvent).where(
            HrLifecycleEvent.person_id == cancelled.id,
            HrLifecycleEvent.event_type == "hris_purged",
        )
    )).scalars().all()
    assert len(lifecycle_rows) == 1
    assert lifecycle_rows[0].detail["reason"] == (
        "cancelled_future_hire_absent_from_sf"
    )
    assert lifecycle_rows[0].detail["soft_deleted"] is True


@pytest.mark.asyncio
async def test_future_hire_return_clears_absence_grace(db_session) -> None:
    """Transient SF-Onboarding flicker: a future hire absent once (grace
    recorded) that SF returns again next run has its ``hris_absent_since``
    marker cleared and is never soft-deleted (P10019 was this shape)."""
    instance_id = uuid.uuid4()
    fh = IdentityPerson(
        id=uuid.uuid4(),
        work_email="flicker@pingcap.com",
        display_name="Flicker Hire",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="P77020",
        hris_instance_id=instance_id,
        hire_date=_future_date(),
        hris_record_state="present",
    )
    db_session.add(fh)
    await db_session.commit()

    state = {"returns": False}

    class _FakeClient:
        def __init__(self, *_, **__) -> None:
            pass

        async def list_current_jobs(self, *, limit=None):  # noqa: ANN001
            return [{"userId": "sf-SOMEONE-ELSE"}]

        async def get_users(self, user_ids):  # noqa: ANN001
            if state["returns"]:
                return [{"userId": "P77020", "displayName": "Flicker Hire"}]
            return []  # transiently absent (lone future hire → not a blip)

    with (
        patch("app.integrations.compat.list_hris_instances",
              return_value=[(_FutureHireIntegrationRow(instance_id),
                             {"base_url": "https://sf.example/odata"})]),
        patch("app.identity.successfactors_client.SuccessFactorsClient",
              _FakeClient),
        patch("app.identity.successfactors_auth.build_auth_from_config",
              _fake_auth),
    ):
        # Run 1: absent → grace recorded, NOT deleted.
        s1 = await run_hris_absent_person_check(db_session, {})
        assert s1["future_hire_absence_grace_started"] == 1
        assert s1["cancelled_future_hires"] == 0
        await db_session.refresh(fh)
        assert fh.hris_absent_since is not None
        assert fh.deleted_at is None

        # Run 2: SF returns them → marker cleared, still never deleted.
        state["returns"] = True
        s2 = await run_hris_absent_person_check(db_session, {})
        assert s2["future_hire_absence_cleared"] == 1
        assert s2["cancelled_future_hires"] == 0

    await db_session.refresh(fh)
    assert fh.hris_absent_since is None
    assert fh.deleted_at is None
    assert fh.hris_record_state == "present"


@pytest.mark.asyncio
async def test_future_hire_cross_check_skips_on_sf_error(db_session) -> None:
    """``get_users`` raising must NOT delete anyone — an unverified
    signal is never actioned; the check retries next run."""
    instance_id = uuid.uuid4()
    fh = IdentityPerson(
        id=uuid.uuid4(),
        work_email="unknown@pingcap.com",
        display_name="Unknown",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="P10019",
        hris_instance_id=instance_id,
        hire_date=_future_date(),
        hris_record_state="present",
    )
    db_session.add(fh)
    await db_session.commit()

    class _FakeClient:
        def __init__(self, *_, **__) -> None:
            pass

        async def list_current_jobs(self, *, limit=None):  # noqa: ANN001
            return [{"userId": "sf-OTHER"}]

        async def get_users(self, user_ids):  # noqa: ANN001
            raise RuntimeError("SF 503")

    with (
        patch("app.integrations.compat.list_hris_instances",
              return_value=[(_FutureHireIntegrationRow(instance_id),
                             {"base_url": "https://sf.example/odata"})]),
        patch("app.identity.successfactors_client.SuccessFactorsClient",
              _FakeClient),
        patch("app.identity.successfactors_auth.build_auth_from_config",
              _fake_auth),
    ):
        summary = await run_hris_absent_person_check(db_session, {})

    assert summary["cancelled_future_hires"] == 0
    await db_session.refresh(fh)
    assert fh.deleted_at is None
    assert fh.hris_record_state == "present"


@pytest.mark.asyncio
async def test_future_hire_cross_check_blip_guard(db_session) -> None:
    """If ``get_users`` returns nothing for a set of ≥2 future hires,
    treat it as an SF scope/outage artifact and delete NObody."""
    instance_id = uuid.uuid4()
    people = []
    for i in range(3):
        p = IdentityPerson(
            id=uuid.uuid4(),
            work_email=f"fh{i}@pingcap.com",
            display_name=f"Future {i}",
            employment_status="active",
            source_system="successfactors",
            hris_person_id=f"P200{i}",
            hris_instance_id=instance_id,
            hire_date=_future_date(),
            hris_record_state="present",
        )
        people.append(p)
    db_session.add_all(people)
    await db_session.commit()

    class _FakeClient:
        def __init__(self, *_, **__) -> None:
            pass

        async def list_current_jobs(self, *, limit=None):  # noqa: ANN001
            return [{"userId": "sf-OTHER"}]

        async def get_users(self, user_ids):  # noqa: ANN001
            return []  # SF returned nothing for the whole set → blip

    with (
        patch("app.integrations.compat.list_hris_instances",
              return_value=[(_FutureHireIntegrationRow(instance_id),
                             {"base_url": "https://sf.example/odata"})]),
        patch("app.identity.successfactors_client.SuccessFactorsClient",
              _FakeClient),
        patch("app.identity.successfactors_auth.build_auth_from_config",
              _fake_auth),
    ):
        summary = await run_hris_absent_person_check(db_session, {})

    assert summary["cancelled_future_hires"] == 0
    assert any(
        b.get("kind") == "future_hire_cross_check"
        for b in summary["brake_tripped"]
    )
    for p in people:
        await db_session.refresh(p)
        assert p.deleted_at is None
        assert p.hris_record_state == "present"