"""Regression tests for the SLA escalation dispatch rewrite (PR 2026-05-02).

Three independent prod regressions sat on this code path simultaneously
and are all covered here so they cannot quietly come back:

1. **Recipient flood** — SLA URGENT/BREACH used to fan out to every
   active ``it_admin`` + ``hr`` role (33 people in prod, including the
   leaver themselves when they happened to hold ``hr``). The new
   ``_dm_hrbp_and_it_admins`` targets exactly: HRBP (one named person,
   resolved from Feishu custom_attrs) plus the small ``it_admin`` pool,
   never the leaver themselves.

2. **Empty Feishu card** — ``_load_template`` returned a truthy
   ``"(no template configured)"`` placeholder when an SLA template key
   had no entry in ``_DEFAULT_TEMPLATES`` and no platform-settings
   override. The renderers used the placeholder instead of the
   envelope's real title/body, shipping a useless card. We added the
   three missing SLA templates AND made ``_load_template`` return
   ``None`` (not a placeholder) so the envelope is the source of truth.

3. **No clickable button** — the dispatch set ``portal_url`` only in
   ``context``, leaving ``env.url`` ``None`` so the Feishu card built
   no action button. The button URL is now derived from ``env.url``.

All three are exercised here with monkeypatched Feishu/email channels
so the tests don't actually hit Feishu or SMTP.
"""
from __future__ import annotations

from datetime import UTC, date, datetime, timedelta
from typing import Any
from uuid import UUID, uuid4

import pytest
from sqlalchemy import select

from app.auth.models import AuthRole, AuthUserRole
from app.domains.leaver_blocked.sla_service import LeaverBlockedSlaService
from app.domains.offboarding.models import (
    PROC_BLOCKED,
    RES_API_KEY,
    RESO_PENDING,
    SLA_NONE,
    SLA_URGENT,
    OffboardingBlocker,
    OffboardingProcess,
)
from app.identity.person_models import IdentityPerson
from app.notifications.models import NotificationOutbox

# ── helpers ────────────────────────────────────────────────


async def _make_person(
    db_session,
    *,
    work_email: str,
    display_name: str,
    employment_status: str = "active",
    role_name: str | None = None,
    manager_person_id: UUID | None = None,
    hrbp_person_id: UUID | None = None,
) -> IdentityPerson:
    person = IdentityPerson(
        id=uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid4().hex[:8]}",
        work_email=work_email,
        display_name=display_name,
        employment_status=employment_status,
        manager_person_id=manager_person_id,
        hrbp_person_id=hrbp_person_id,
    )
    db_session.add(person)
    await db_session.flush()

    if role_name:
        role_id = (await db_session.execute(
            select(AuthRole.id).where(AuthRole.name == role_name)
        )).scalar_one()
        db_session.add(AuthUserRole(user_id=person.id, role_id=role_id))
        await db_session.flush()

    return person


async def _make_blocked_process(
    db_session,
    person: IdentityPerson,
    *,
    first_blocked_at: datetime,
    leaver_date: date,
) -> OffboardingProcess:
    proc = OffboardingProcess(
        id=uuid4(),
        person_id=person.id,
        leaver_date=leaver_date,
        status=PROC_BLOCKED,
        first_blocked_at=first_blocked_at,
        sla_breach_level=SLA_NONE,
    )
    db_session.add(proc)
    await db_session.flush()
    # 2026-05-30 — sla_service.run_tick now short-circuits the WARN-
    # tier manager DM when no non-hardware pending blocker exists
    # (per the "manager ≠ hardware" rule — laptop pickup is an
    # IT/Office Admin track). A PROC_BLOCKED process without any
    # blocker row would silently skip WARN. Attach one API-Key blocker
    # by default so the dispatch path stays under test.
    db_session.add(OffboardingBlocker(
        id=uuid4(),
        process_id=proc.id,
        resource_type=RES_API_KEY,
        resource_id=str(uuid4()),
        label="sla-fixture-api-key",
        resolution=RESO_PENDING,
    ))
    await db_session.flush()
    return proc


class _ChannelStubs:
    """Patch the two delivery channels so SLA dispatch can be unit-
    tested without reaching Feishu or SMTP. Each stub records the
    envelope it received for assertion-driven recipient-list checks.

    Returning success keeps every dispatch a "delivered" row on the
    outbox, which is fine — recipient-set assertions look at the rows
    ``dispatch`` created, not at delivery state. The feishu stub returns
    a 3-tuple ``(ok, error, message_id)`` and the email stub a 2-tuple:
    their contracts diverged with migration 571, where the Feishu send
    started reporting the card's message id so a reply's ``parent_id``
    can resolve back to it.
    """

    def __init__(self) -> None:
        self.feishu_envelopes: list[Any] = []
        self.email_envelopes: list[Any] = []

    async def fake_feishu(self, env):
        self.feishu_envelopes.append(env)
        return True, None, None

    async def fake_email(self, env):
        self.email_envelopes.append(env)
        return True, None


@pytest.fixture
def channel_stubs(monkeypatch):
    stubs = _ChannelStubs()
    from app.notifications import service as notif_service

    monkeypatch.setattr(
        notif_service.NotificationService,
        "_send_feishu",
        lambda self, env: stubs.fake_feishu(env),
    )
    monkeypatch.setattr(
        notif_service.NotificationService,
        "_send_email",
        lambda self, env: stubs.fake_email(env),
    )
    return stubs


@pytest.fixture
def stub_hrbp_resolver(monkeypatch):
    """Replace the live Feishu HRBP resolver with a deterministic stub.

    Tests can set ``stub_hrbp_resolver.next = HrbpResolution(...)`` to
    drive the next dispatch's HRBP outcome. The fixture intercepts the
    import inside ``sla_service`` (where ``resolve_hrbp_for_person`` is
    bound at import time) so the real Feishu round-trip never fires.
    """
    from app.integrations.feishu_hrbp import HrbpResolution

    class _Box:
        next: HrbpResolution = HrbpResolution(person_id=None, reason="not configured")

    box = _Box()

    async def fake_resolve(_db, _person):
        return box.next

    import app.domains.leaver_blocked.sla_service as svc_mod

    monkeypatch.setattr(svc_mod, "resolve_hrbp_for_person", fake_resolve)
    return box


# ── tests ──────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_urgent_dispatches_to_hrbp_and_it_admin_only(
    db_session, channel_stubs, stub_hrbp_resolver,
):
    """T+48h URGENT → HRBP (1) + it_admin pool (N), never the leaver,
    never the entire ``hr`` role."""
    from app.integrations.feishu_hrbp import HrbpResolution

    # HRBP: a real person, not the leaver, not in any admin role
    hrbp = await _make_person(
        db_session, work_email="hrbp@x.com", display_name="HRBP One",
    )
    # it_admin pool: 2 people
    admin_a = await _make_person(
        db_session, work_email="admin-a@x.com", display_name="Admin A",
        role_name="global_admin",
    )
    admin_b = await _make_person(
        db_session, work_email="admin-b@x.com", display_name="Admin B",
        role_name="global_admin",
    )
    # leaver, with hrbp pointer pre-set so the resolver short-circuits
    leaver = await _make_person(
        db_session, work_email="leaver@x.com", display_name="The Leaver",
        hrbp_person_id=hrbp.id,
    )
    # Decoy: a user with ``hr`` role — under the OLD policy they'd get
    # paged. Under the NEW policy they should not.
    role_id = (await db_session.execute(
        select(AuthRole.id).where(AuthRole.name == "global_admin")
    )).scalar_one_or_none()
    # ``hr`` role is seeded globally by conftest's ``_SYSTEM_ROLES`` (the
    # 2026-05-14 sync-history-privacy PR added it so hr_tools / HR-router
    # tests can run under the ``hr`` role). Look up the existing row
    # rather than INSERT — a fresh INSERT here would race the seed and
    # blow up with ``Duplicate entry 'hr' for key 'auth_roles.name'``.
    hr_role_id = (await db_session.execute(
        select(AuthRole.id).where(AuthRole.name == "hr")
    )).scalar_one()
    hr_only = await _make_person(
        db_session, work_email="hr-only@x.com", display_name="HR Only",
    )
    db_session.add(AuthUserRole(user_id=hr_only.id, role_id=hr_role_id))
    await db_session.flush()

    # Configure the fixture so the leaver's HRBP resolves to ``hrbp``
    stub_hrbp_resolver.next = HrbpResolution(
        person_id=hrbp.id, reason=None, hrbp_email=hrbp.work_email,
    )

    now = datetime.now(UTC)
    proc = await _make_blocked_process(
        db_session, leaver,
        first_blocked_at=now - timedelta(hours=49),
        leaver_date=date.today() - timedelta(days=2),
    )
    await db_session.commit()

    svc = LeaverBlockedSlaService(db_session)
    summary = await svc.run_tick(now=now)
    await db_session.commit()

    assert summary["urgent_sent"] == 1
    rows = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.template_key == "leaver_blocked_sla_urgent",
        )
    )).scalars().all()
    recipients = sorted(r.recipient_email for r in rows)

    # Exactly 3: HRBP + 2 admins
    assert recipients == sorted([hrbp.work_email, admin_a.work_email, admin_b.work_email])
    # Hard exclusions:
    assert leaver.work_email not in recipients
    assert hr_only.work_email not in recipients
    # And the dispatch carries an action URL, not a placeholder.
    # As of 2026-07-15 lifecycle-read recipients (global_admin / hr /
    # auditor / management) land on the per-process detail page
    # (/admin/offboarding/process/{id}) — it shows the blocker list
    # directly. This ``hrbp`` fixture holds NO role, so it still routes
    # to the counts-only /my/team-leavers lookout (the admin detail
    # route 403s for it). See test_urgent_hrbp_with_hr_role_gets_detail
    # for the HRBP-holds-hr case that this PR actually fixes.
    by_email = {row.recipient_email: row for row in rows}
    detail_url = f"/admin/offboarding/process/{proc.id}"
    assert by_email[admin_a.work_email].url == detail_url
    assert by_email[admin_b.work_email].url == detail_url
    assert by_email[hrbp.work_email].url == "/my/team-leavers"
    _ = proc, role_id, leaver  # keep linters quiet about unused fixture refs


@pytest.mark.asyncio
async def test_urgent_excludes_leaver_who_holds_global_admin(
    db_session, channel_stubs, stub_hrbp_resolver,
):
    """The leaver must never receive a notification about their own
    SLA breach, even if they hold ``global_admin``."""
    from app.integrations.feishu_hrbp import HrbpResolution

    other_admin = await _make_person(
        db_session, work_email="other@x.com", display_name="Other Admin",
        role_name="global_admin",
    )
    leaver = await _make_person(
        db_session, work_email="self-admin@x.com", display_name="Self Admin",
        role_name="global_admin",
    )
    stub_hrbp_resolver.next = HrbpResolution(
        person_id=None, reason="no HRBP configured",
    )

    now = datetime.now(UTC)
    await _make_blocked_process(
        db_session, leaver,
        first_blocked_at=now - timedelta(hours=49),
        leaver_date=date.today() - timedelta(days=2),
    )
    await db_session.commit()

    svc = LeaverBlockedSlaService(db_session)
    await svc.run_tick(now=now)
    await db_session.commit()

    rows = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.template_key == "leaver_blocked_sla_urgent",
        )
    )).scalars().all()
    recipients = [r.recipient_email for r in rows]
    assert recipients == [other_admin.work_email]
    assert leaver.work_email not in recipients


@pytest.mark.asyncio
async def test_urgent_falls_back_to_it_admin_when_hrbp_unresolved(
    db_session, channel_stubs, stub_hrbp_resolver,
):
    """HRBP resolution failure (no Feishu account, no HRBP attr, etc.)
    must NOT silently drop the SLA escalation — the it_admin pool still
    gets paged AND an audit row records why HRBP routing fell through.
    """
    from app.integrations.feishu_hrbp import HrbpResolution

    admin = await _make_person(
        db_session, work_email="admin@x.com", display_name="Admin One",
        role_name="global_admin",
    )
    leaver = await _make_person(
        db_session, work_email="leaver@x.com", display_name="The Leaver",
    )
    stub_hrbp_resolver.next = HrbpResolution(
        person_id=None, reason="leaver Feishu user_id resolve failed",
    )

    now = datetime.now(UTC)
    await _make_blocked_process(
        db_session, leaver,
        first_blocked_at=now - timedelta(hours=49),
        leaver_date=date.today() - timedelta(days=2),
    )
    await db_session.commit()

    svc = LeaverBlockedSlaService(db_session)
    await svc.run_tick(now=now)
    await db_session.commit()

    rows = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.template_key == "leaver_blocked_sla_urgent",
        )
    )).scalars().all()
    assert [r.recipient_email for r in rows] == [admin.work_email]

    from app.audit.models import AuditLog
    audit_rows = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "leaver_sla.hrbp_resolve_failed",
        )
    )).scalars().all()
    assert len(audit_rows) == 1
    assert audit_rows[0].status == "failure"
    assert (audit_rows[0].detail or {}).get("level") == "urgent"
    assert (audit_rows[0].detail or {}).get("reason") == (
        "leaver Feishu user_id resolve failed"
    )


@pytest.mark.asyncio
async def test_warn_at_24h_uses_direct_manager_only_not_skip_level(
    db_session, channel_stubs, stub_hrbp_resolver,
):
    """T+24h WARN still uses the manager chain; it does NOT touch the
    HRBP path or the it_admin pool. This guards against a later refactor
    accidentally changing WARN to use the URGENT recipient set.

    2026-05-30 revision: skip-level (manager's manager) was deliberately
    removed from the WARN recipient set per user feedback
    ("level+2 不需要知道这个事情"). Only the leaver's DIRECT manager
    is in the WARN cohort. Pinned so a refactor restoring the
    ``range(2)`` chain walk re-trips this test.
    """
    skip_level = await _make_person(
        db_session, work_email="skip@x.com", display_name="Skip Manager",
    )
    direct = await _make_person(
        db_session, work_email="direct@x.com", display_name="Direct Manager",
        manager_person_id=skip_level.id,
    )
    # An it_admin who would page under URGENT but must be SILENT under WARN
    await _make_person(
        db_session, work_email="admin@x.com", display_name="Admin Silent",
        role_name="global_admin",
    )
    leaver = await _make_person(
        db_session, work_email="leaver@x.com", display_name="The Leaver",
        manager_person_id=direct.id,
    )

    now = datetime.now(UTC)
    await _make_blocked_process(
        db_session, leaver,
        first_blocked_at=now - timedelta(hours=25),
        leaver_date=date.today() - timedelta(days=1),
    )
    await db_session.commit()

    svc = LeaverBlockedSlaService(db_session)
    summary = await svc.run_tick(now=now)
    await db_session.commit()

    assert summary["warn_sent"] == 1
    assert summary["urgent_sent"] == 0
    rows = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.template_key == "leaver_blocked_sla_warn",
        )
    )).scalars().all()
    recipients = sorted(r.recipient_email for r in rows)
    assert recipients == [direct.work_email], (
        "Skip-level must NOT be in the WARN recipient set "
        "(user feedback 2026-05-30: level+2 doesn't need to know)"
    )
    assert skip_level.work_email not in recipients
    # 2026-05-20: managers don't hold ``global_admin``, so the
    # ``/admin/offboarding`` URL 403s for them. The WARN tier dispatch
    # now routes them to ``/my/team-leavers`` — the read-only lookout
    # that surfaces leavers where the viewer is the HRBP or upline
    # manager. Without this assertion a future refactor that flips the
    # ``is_admin`` default back to True would silently re-break the
    # Feishu card button without anything catching it.
    for row in rows:
        assert row.url == "/my/team-leavers"


@pytest.mark.asyncio
async def test_urgent_hrbp_who_also_holds_global_admin_gets_admin_url(
    db_session, channel_stubs, stub_hrbp_resolver,
):
    """Rare-but-possible: an HRBP also holds ``global_admin``. The
    dispatcher's per-recipient URL switch must route them to the
    per-process detail page, not the non-admin lookout, because that
    page exposes the blocker list + (for global_admin) the override /
    advance flow they actually have permission to use.

    Guards the ``recipient.id in reader_ids`` routing in
    ``_dm_hrbp_and_it_admins`` — without it a future refactor that
    iterated HRBP-first and skipped the reader-set check would silently
    downgrade the HRBP-admin's UX.
    """
    from app.integrations.feishu_hrbp import HrbpResolution

    hrbp_admin = await _make_person(
        db_session, work_email="hrbp-admin@x.com",
        display_name="HRBP Who Is Also Admin",
        role_name="global_admin",
    )
    leaver = await _make_person(
        db_session, work_email="leaver@x.com", display_name="The Leaver",
        hrbp_person_id=hrbp_admin.id,
    )
    stub_hrbp_resolver.next = HrbpResolution(
        person_id=hrbp_admin.id, reason=None, hrbp_email=hrbp_admin.work_email,
    )

    now = datetime.now(UTC)
    proc = await _make_blocked_process(
        db_session, leaver,
        first_blocked_at=now - timedelta(hours=49),
        leaver_date=date.today() - timedelta(days=2),
    )
    await db_session.commit()

    svc = LeaverBlockedSlaService(db_session)
    summary = await svc.run_tick(now=now)
    await db_session.commit()

    assert summary["urgent_sent"] == 1
    rows = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.template_key == "leaver_blocked_sla_urgent",
            NotificationOutbox.recipient_email == hrbp_admin.work_email,
        )
    )).scalars().all()
    assert len(rows) == 1
    assert rows[0].url == f"/admin/offboarding/process/{proc.id}"


@pytest.mark.asyncio
async def test_urgent_hrbp_with_hr_role_routes_to_detail_page(
    db_session, channel_stubs, stub_hrbp_resolver,
):
    """The fix this PR ships: the leaver's HRBP typically holds ``hr``
    (not ``global_admin``). The per-process detail page was opened to
    ``hr`` on 2026-07-08 (``require_lifecycle_read``), but the SLA card
    kept routing non-global_admin recipients to the counts-only
    ``/my/team-leavers`` lookout — so the HRBP got an urgent "go handle
    it" card that dead-ended without showing WHICH resource was blocked.

    The card CTA now routes any lifecycle-read role (incl. ``hr``) to
    ``/admin/offboarding/process/{id}`` where the blocker list is
    visible, and the body itself names the outstanding resources.
    """
    from app.integrations.feishu_hrbp import HrbpResolution

    hrbp = await _make_person(
        db_session, work_email="hrbp-hr@x.com",
        display_name="HRBP With HR Role",
        role_name="hr",
    )
    leaver = await _make_person(
        db_session, work_email="leaver@x.com", display_name="The Leaver",
        hrbp_person_id=hrbp.id,
    )
    stub_hrbp_resolver.next = HrbpResolution(
        person_id=hrbp.id, reason=None, hrbp_email=hrbp.work_email,
    )

    now = datetime.now(UTC)
    proc = await _make_blocked_process(
        db_session, leaver,
        first_blocked_at=now - timedelta(hours=49),
        leaver_date=date.today() - timedelta(days=2),
    )
    await db_session.commit()

    svc = LeaverBlockedSlaService(db_session)
    await svc.run_tick(now=now)
    await db_session.commit()

    rows = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.template_key == "leaver_blocked_sla_urgent",
            NotificationOutbox.recipient_email == hrbp.work_email,
        )
    )).scalars().all()
    assert rows, "the hr-role HRBP must receive the URGENT card"
    # Routed to the detail page they can now read — NOT the counts-only lookout.
    assert all(r.url == f"/admin/offboarding/process/{proc.id}" for r in rows)
    # And the body names the specific outstanding resource so the logic
    # is legible without clicking through.
    assert any("sla-fixture-api-key" in (r.body_text or "") for r in rows)


@pytest.mark.asyncio
async def test_pending_resource_lines_lists_blocker_labels(db_session):
    """``_pending_resource_lines`` renders a bullet list of the pending
    blocker labels (mirrors the leaver_reminder ``pending_blockers``
    block) so the SLA card can name what is outstanding."""
    person = await _make_person(
        db_session, work_email="leaver@x.com", display_name="The Leaver",
    )
    now = datetime.now(UTC)
    proc = await _make_blocked_process(
        db_session, person,
        first_blocked_at=now - timedelta(hours=49),
        leaver_date=date.today() - timedelta(days=2),
    )
    # Add a hardware blocker alongside the default api-key one.
    from app.domains.offboarding.models import RES_HARDWARE
    db_session.add(OffboardingBlocker(
        id=uuid4(),
        process_id=proc.id,
        resource_type=RES_HARDWARE,
        resource_id=str(uuid4()),
        label="P-US-NB2024070003 / MacBook / Mac Air",
        resolution=RESO_PENDING,
    ))
    await db_session.flush()

    svc = LeaverBlockedSlaService(db_session)
    rendered = await svc._pending_resource_lines(proc)
    assert "  · sla-fixture-api-key" in rendered
    assert "  · P-US-NB2024070003 / MacBook / Mac Air" in rendered


def _has_cjk_only_line(text: str) -> bool:
    """True if any line contains CJK characters but no Latin letters —
    i.e. a Chinese-only sentence that would read as noise in the English
    email body."""
    import re

    for line in text.splitlines():
        if re.search(r"[一-鿿]", line) and not re.search(r"[A-Za-z]", line):
            return True
    return False


@pytest.mark.asyncio
async def test_pending_resource_lines_fallback_and_tail_are_bilingual(db_session):
    """Regression (PR #1945 follow-up): ``_pending_resource_lines`` output
    is injected verbatim into BOTH the zh and en template bodies (one
    shared context), so its fixed strings — the empty fallback and the
    truncation tail — must not be Chinese-only."""
    person = await _make_person(
        db_session, work_email="leaver-i18n@x.com", display_name="The Leaver",
    )
    now = datetime.now(UTC)
    proc = await _make_blocked_process(
        db_session, person,
        first_blocked_at=now - timedelta(hours=49),
        leaver_date=date.today() - timedelta(days=2),
    )
    svc = LeaverBlockedSlaService(db_session)

    # Truncation tail: exceed the render cap so the "…等 N 项" tail fires.
    for i in range(15):
        db_session.add(OffboardingBlocker(
            id=uuid4(),
            process_id=proc.id,
            resource_type=RES_API_KEY,
            resource_id=str(uuid4()),
            label=f"blocker-{i:02d}",
            resolution=RESO_PENDING,
        ))
    await db_session.flush()
    rendered = await svc._pending_resource_lines(proc)
    assert "items in total" in rendered  # bilingual tail
    assert not _has_cjk_only_line(rendered)

    # Empty fallback: resolve every blocker (benign race — the last one
    # cleared between the SLA decision and the render).
    from app.domains.offboarding.models import RESO_REVOKED_MANUAL
    for b in (await db_session.execute(
        select(OffboardingBlocker).where(OffboardingBlocker.process_id == proc.id)
    )).scalars().all():
        b.resolution = RESO_REVOKED_MANUAL
    await db_session.flush()
    fallback = await svc._pending_resource_lines(proc)
    assert "Helix" in fallback  # bilingual fallback, not zh-only
    assert not _has_cjk_only_line(fallback)


@pytest.mark.asyncio
async def test_en_sla_body_contains_no_cjk_only_lines(db_session):
    """End-to-end shape check: format the actual ``body_text_en`` SLA
    templates with a rendered ``pending_resources`` block (both the
    fallback and the truncated variants) and assert the English body
    carries no Chinese-only line."""
    from app.domains.leaver_blocked.sla_service import _level_label, _level_label_en
    from app.notifications.service import _DEFAULT_TEMPLATES

    person = await _make_person(
        db_session, work_email="leaver-en@x.com", display_name="The Leaver",
    )
    now = datetime.now(UTC)
    proc = await _make_blocked_process(
        db_session, person,
        first_blocked_at=now - timedelta(hours=73),
        leaver_date=date.today() - timedelta(days=3),
    )
    svc = LeaverBlockedSlaService(db_session)

    async def _render_variants() -> list[str]:
        pending = await svc._pending_resource_lines(proc)
        bodies = []
        for level in ("warn", "urgent", "breach"):
            tmpl = _DEFAULT_TEMPLATES[f"leaver_blocked_sla_{level}"]
            bodies.append(tmpl["body_text_en"].format(
                leaver_name=person.display_name,
                leaver_email=person.work_email,
                level_label_zh=_level_label(level),
                level_label_en=_level_label_en(level),
                pending_resources=pending,
                cta_line="(cta line)",
            ))
        return bodies

    # Variant 1: normal pending list (fixture blocker) — labels are
    # locale-neutral identifiers.
    for body in await _render_variants():
        assert not _has_cjk_only_line(body), body

    # Variant 2: empty fallback (all blockers resolved).
    from app.domains.offboarding.models import RESO_REVOKED_MANUAL
    for b in (await db_session.execute(
        select(OffboardingBlocker).where(OffboardingBlocker.process_id == proc.id)
    )).scalars().all():
        b.resolution = RESO_REVOKED_MANUAL
    await db_session.flush()
    for body in await _render_variants():
        assert not _has_cjk_only_line(body), body


@pytest.mark.asyncio
async def test_warn_direct_manager_who_also_holds_global_admin_gets_admin_url(
    db_session, channel_stubs, stub_hrbp_resolver,
):
    """A direct manager who ALSO holds ``global_admin`` (rare but
    possible — IT directors managing the IT team, lifecycle owners
    managing HRIS folks) must land on the admin console where they
    can act on the override flow, not the read-only ``/my/team-leavers``
    lookout. Guards the per-recipient ``admin_id_set`` lookup in
    ``_dm_manager_chain``; without this test a refactor that flipped
    the manager dispatch back to a hard-coded ``is_admin=False`` would
    silently downgrade an IT-director-manager's UX.

    2026-05-30 revision: the original version of this test wired the
    admin role onto the SKIP-LEVEL manager (since the old chain walked
    ``range(2)``). After the WARN-tier ``_dm_manager_chain`` was
    restricted to the direct manager only ("level+2 不需要知道"),
    the only manager-cohort person who can still trip this branch is
    the direct manager themselves. The test was rewritten to put the
    admin role on the direct manager and verify the admin URL routing
    for that path.
    """
    skip_level = await _make_person(
        db_session, work_email="skip@x.com",
        display_name="Skip-Level (Not Notified)",
    )
    admin_direct = await _make_person(
        db_session, work_email="director@x.com",
        display_name="IT Director Who Directly Manages",
        role_name="global_admin",
        manager_person_id=skip_level.id,
    )
    leaver = await _make_person(
        db_session, work_email="leaver@x.com", display_name="The Leaver",
        manager_person_id=admin_direct.id,
    )

    now = datetime.now(UTC)
    proc = await _make_blocked_process(
        db_session, leaver,
        first_blocked_at=now - timedelta(hours=25),
        leaver_date=date.today() - timedelta(days=1),
    )
    await db_session.commit()

    svc = LeaverBlockedSlaService(db_session)
    summary = await svc.run_tick(now=now)
    await db_session.commit()

    assert summary["warn_sent"] == 1
    rows = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.template_key == "leaver_blocked_sla_warn",
        )
    )).scalars().all()
    by_email = {row.recipient_email: row for row in rows}
    # Skip-level must NOT receive WARN (level+2 disabled).
    assert skip_level.work_email not in by_email, (
        "Skip-level must NOT be in the WARN recipient set"
    )
    # Direct manager who also holds global_admin → per-process detail page.
    assert admin_direct.work_email in by_email
    assert (
        by_email[admin_direct.work_email].url
        == f"/admin/offboarding/process/{proc.id}"
    )


@pytest.mark.asyncio
async def test_one_process_flush_failure_isolated_by_savepoint(
    db_session, channel_stubs, stub_hrbp_resolver, monkeypatch,
):
    """A single process's flush failure (e.g. a 1062 on the
    ``offboarding_reminder_log`` unique key under a lock-TTL-expiry race)
    must NOT poison the shared session and discard the whole tick — the
    other processes must still escalate and commit.

    Mirrors the onboarding escalation tick's per-process SAVEPOINT
    isolation (the leaver SLA tick was the asymmetric outlier). Without
    the SAVEPOINT, the unhandled error would propagate out of ``run_tick``
    and the trailing ``session.commit()`` would discard every level-up
    already applied this tick.
    """
    from app.integrations.feishu_hrbp import HrbpResolution

    # it_admin recipient pool (HRBP unresolved → falls back to admins)
    await _make_person(
        db_session, work_email="admin@x.com", display_name="Admin One",
        role_name="global_admin",
    )
    leaver_bad = await _make_person(
        db_session, work_email="bad@x.com", display_name="Bad Leaver",
    )
    leaver_ok = await _make_person(
        db_session, work_email="ok@x.com", display_name="OK Leaver",
    )
    stub_hrbp_resolver.next = HrbpResolution(
        person_id=None, reason="no HRBP configured",
    )

    now = datetime.now(UTC)
    proc_bad = await _make_blocked_process(
        db_session, leaver_bad,
        first_blocked_at=now - timedelta(hours=49),  # URGENT tier
        leaver_date=date.today() - timedelta(days=2),
    )
    proc_ok = await _make_blocked_process(
        db_session, leaver_ok,
        first_blocked_at=now - timedelta(hours=49),
        leaver_date=date.today() - timedelta(days=2),
    )
    await db_session.commit()

    svc = LeaverBlockedSlaService(db_session)
    orig_send_once = svc._send_once
    bad_person_id = leaver_bad.id

    async def flaky_send_once(proc, kind):
        # Simulate an unhandled flush error escaping for the bad process
        # only (the kind a 1062 on the reminder-log insert would raise).
        if proc.person_id == bad_person_id:
            raise RuntimeError("simulated reminder-log flush failure (1062)")
        return await orig_send_once(proc, kind)

    monkeypatch.setattr(svc, "_send_once", flaky_send_once)

    # Must NOT raise — the bad process is isolated by its SAVEPOINT.
    summary = await svc.run_tick(now=now)
    await db_session.commit()

    # The healthy process still escalated and its level-up committed.
    assert summary["urgent_sent"] == 1
    await db_session.refresh(proc_ok)
    assert proc_ok.sla_breach_level == SLA_URGENT

    # The bad process's in-SAVEPOINT level-up was rolled back (isolation),
    # so it stays at SLA_NONE and the next tick retries it cleanly.
    await db_session.refresh(proc_bad)
    assert proc_bad.sla_breach_level == SLA_NONE