"""Integration test for the manager (approver) auto-include + notification.

The rescue keeps ANY active manager of an in-scope report in the target,
even when the manager falls outside the filter (different entity /
country). Historically it was gated to ``is_china_employee=True``
managers; that gate was dropped 2026-07-07 (see
``test_non_cn_out_of_scope_manager_pulled_in``) because the flag is
unreliable and silently dropped real approvers.


End-to-end through ``IdentitySyncService._sweep`` against a real MySQL
test DB. A stub runner replaces the SCIM HTTP layer so the test
exercises the orchestrator logic (filter → auto-include → state row
marking → notification enqueue) without making outbound calls.

Scenarios:

1. **First sweep**: foreign employee in scope, their CN manager not
   matching the filter, gets pulled in. State row carries
   ``auto_included_reason='manager_of_foreign'``, ``notified_at`` is
   set, a ``notification_outbox`` row exists for the manager.

2. **Second sweep**: rerun the same target. Manager's state row
   already has ``notified_at`` non-NULL → no second notification
   enqueued (would amount to spam).

3. **No foreign-report fallout**: if every in-scope person is
   Chinese, no auto-include fires (gating heuristic preserved).

4. **Permanent in-scope**: in the second sweep we remove the foreign
   report from scope (mark them ``leaver_pending`` so they drop out
   of the active-employment filter). The CN manager state row
   carries ``auto_included_reason`` → must NOT transition to
   ``deactivated`` via the out-of-scope sweep.

Test DB: requires the standard pytest fixture suite (db_engine /
db_session). Schema includes the migration-202 columns
(``auto_included_reason`` + ``notified_at``).
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.identity_sync.models import (
    IdentitySyncState,
    IdentitySyncTarget,
)
from app.domains.identity_sync.runners.base import PersonOutcome, SyncRunner
from app.domains.identity_sync.service import IdentitySyncService
from app.domains.leaver_blocked.listener import (
    register_listener as _register_leaver_listener,
)
from app.identity.person_models import IdentityPerson
from app.notifications.models import NotificationOutbox


@pytest.fixture(autouse=True)
def _wire_prod_listeners():
    """Run this module under the production LeaverBlocked flush listener.

    In a full-suite run the listener gets registered as a side effect of
    ``tests/test_core/test_worker_model_preload.py`` (it calls
    ``worker._load_all_models()``, which wires the listener globally on
    the SQLAlchemy ``Session`` class for the rest of the process).
    Registering it here too makes the standalone run match the
    full-suite run — regression guard for the 2026-06-11
    order-dependent failure where a bare ``employment_status = 'left'``
    write in this module passed standalone but was trapped (correctly)
    by the listener when ``tests/test_core`` ran first.
    """
    _register_leaver_listener()


class _StubRunner(SyncRunner):
    """Records every upsert / deactivate / delete call; returns success.

    The orchestrator owns state-row updates — the runner only signals
    outcome. We capture call args so the test can assert who got
    pushed (and crucially, that auto-included managers ARE pushed,
    and out-of-scope auto-included rows are NOT deactivated).
    """

    supports_deactivate = True
    supports_delete = True

    def __init__(self) -> None:
        self.upserts: list[tuple[str, dict]] = []
        self.deactivates: list[str] = []
        self.deletes: list[str] = []

    async def upsert(self, *, person_email, payload, prior_external_id):
        self.upserts.append((person_email, payload))
        return PersonOutcome(
            status="success",
            external_id=f"ext-{person_email}",
        )

    async def deactivate(self, *, person_email, prior_external_id):
        self.deactivates.append(person_email)
        return PersonOutcome(status="deactivated")

    async def delete(self, *, person_email, prior_external_id):
        self.deletes.append(person_email)
        return PersonOutcome(status="deleted")

    def describe(self) -> str:
        return "stub"


async def _make_person(
    db: AsyncSession,
    *,
    email: str,
    is_china: bool,
    employment_status: str = "active",
    manager_id: uuid.UUID | None = None,
    company_name: str = "01_PingCAP",
) -> IdentityPerson:
    pid = uuid.uuid4()
    person = IdentityPerson(
        id=pid,
        work_email=email,
        display_name=email.split("@")[0].replace(".", " ").title(),
        employment_status=employment_status,
        source_system="successfactors",
        hris_person_id=f"sf-{email}",
        is_china_employee=is_china,
        country="CHN" if is_china else "USA",
        company_name=company_name,
        manager_person_id=manager_id,
    )
    db.add(person)
    await db.flush()
    return person


async def _make_navan_target(
    db: AsyncSession, *, slug: str = "test-navan",
) -> IdentitySyncTarget:
    """Build a foreigners-only Navan target (countries filter
    excludes CHN). Schedule is 'manual' so the dispatcher doesn't
    own it; we drive _sweep directly.
    """
    from app.domains.identity_sync.presets import NAVAN_MAPPING_RULES

    tgt = IdentitySyncTarget(
        id=uuid.uuid4(),
        name="Test Navan",
        slug=slug,
        kind="navan",
        enabled=True,
        on_offboard="deactivate",
        mapping_config={"rules": list(NAVAN_MAPPING_RULES)},
        filter_config={
            "include_offboarded": False,
            "departments": [],
            "include_subtree": True,
            "countries": ["USA", "JPN", "SGP", "GBR"],
            "employment_statuses": ["active"],
            "job_title_includes": [],
            "person_ids": [],
            "mode": "and",
        },
        schedule_config={
            "trigger": "manual", "cron_expr": None, "timezone": "UTC",
        },
        runner_options={
            "base_url": "https://app.navan.com/universal/scim/v2",
            "auth_scheme": "bearer",
            "extra_headers": {},
            "verify_ssl": True,
            "timeout_seconds": 30,
            "rate_limit_per_minute": 60,
        },
    )
    db.add(tgt)
    await db.flush()
    return tgt


async def _drive_sweep(
    svc: IdentitySyncService,
    target: IdentitySyncTarget,
    runner: SyncRunner,
):
    """Invoke ``_sweep`` directly + commit so the state rows survive
    the test session boundary."""
    summary, status = await svc._sweep(target=target, runner=runner)
    await svc.db.commit()
    return summary, status


@pytest.mark.asyncio
async def test_cn_manager_pulled_in_and_notified(db_session: AsyncSession):
    """A US employee whose manager is a CN employee → CN manager is
    auto-included into the Navan target. State row marked. A
    notification outbox row is created exactly once.
    """
    # CN manager — would NOT match the filter (country=CHN, filter
    # countries=[USA, JPN, SGP, GBR]).
    mgr = await _make_person(
        db_session, email="cn.boss@pingcap.com", is_china=True,
    )
    # US report — matches the filter and reports to the CN manager.
    await _make_person(
        db_session,
        email="us.report@pingcap.com",
        is_china=False,
        manager_id=mgr.id,
    )
    target = await _make_navan_target(db_session)
    await db_session.commit()

    runner = _StubRunner()
    svc = IdentitySyncService(db_session)
    summary, status = await _drive_sweep(svc, target, runner)

    assert status == "ok"
    assert summary.persons_in_scope == 2  # report + auto-pulled manager
    assert summary.auto_included == 1
    assert summary.auto_included_notified == 1
    # Both got upserted.
    emails_pushed = {e for e, _ in runner.upserts}
    assert "us.report@pingcap.com" in emails_pushed
    assert "cn.boss@pingcap.com" in emails_pushed

    # Manager's state row carries the auto-include flag + notified_at.
    state = (await db_session.execute(
        select(IdentitySyncState).where(
            IdentitySyncState.target_id == target.id,
            IdentitySyncState.person_id == mgr.id,
        )
    )).scalars().first()
    assert state is not None
    assert state.auto_included_reason == "manager_of_foreign"
    assert state.notified_at is not None
    assert state.last_status == "success"

    # Exactly one notification outbox row, addressed to the manager.
    notifs = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.recipient_person_id == mgr.id,
        )
    )).scalars().all()
    assert len(notifs) == 1
    assert notifs[0].related_resource_type == "identity_sync_target"
    # ``related_resource_id`` is CHAR(36) on the model so re-reads
    # come back as a string — compare via str() to dodge the
    # UUID-vs-str type mismatch.
    assert str(notifs[0].related_resource_id) == str(target.id)
    assert notifs[0].template_key == "identity_sync.auto_included_manager"
    assert "外籍" in notifs[0].body_text  # CN body line


@pytest.mark.asyncio
async def test_non_cn_out_of_scope_manager_pulled_in(db_session: AsyncSession):
    """The Ed-Huang case (2026-07-07): a manager who is NOT flagged
    ``is_china_employee`` and does NOT match the filter (e.g. a founder
    on the Beijing entity, country=CHN) but who manages an in-scope
    report is STILL an approver → must be auto-included. The old gate
    required ``mgr.is_china_employee=True`` and would have dropped
    (deactivated) this manager's account.
    """
    # Manager: NOT is_china_employee, country=CHN → outside the
    # foreigners-only filter, and NOT rescued by the old CN-only gate.
    mgr = await _make_person(
        db_session, email="founder.cn@pingcap.com", is_china=False,
    )
    # ``_make_person`` sets country=USA when is_china=False; force CHN so
    # the manager genuinely falls outside the countries filter.
    mgr.country = "CHN"
    await db_session.flush()
    # US report matches the filter and reports to the manager.
    await _make_person(
        db_session,
        email="us.ic@pingcap.com",
        is_china=False,
        manager_id=mgr.id,
    )
    target = await _make_navan_target(db_session, slug="noncn-mgr-1")
    await db_session.commit()

    runner = _StubRunner()
    svc = IdentitySyncService(db_session)
    summary, status = await _drive_sweep(svc, target, runner)

    assert status == "ok"
    assert summary.auto_included == 1
    emails_pushed = {e for e, _ in runner.upserts}
    assert "founder.cn@pingcap.com" in emails_pushed  # kept, not dropped
    assert "founder.cn@pingcap.com" not in runner.deactivates

    state = (await db_session.execute(
        select(IdentitySyncState).where(
            IdentitySyncState.target_id == target.id,
            IdentitySyncState.person_id == mgr.id,
        )
    )).scalars().first()
    assert state is not None
    assert state.auto_included_reason == "manager_of_foreign"
    assert state.last_status == "success"


@pytest.mark.asyncio
async def test_second_sweep_does_not_renotify(db_session: AsyncSession):
    """Re-running the sweep against the same target must not enqueue
    a second notification — ``notified_at`` is the dedup signal."""
    mgr = await _make_person(
        db_session, email="cn.lead@pingcap.com", is_china=True,
    )
    await _make_person(
        db_session,
        email="uk.eng@pingcap.com",
        is_china=False,
        manager_id=mgr.id,
    )
    target = await _make_navan_target(db_session, slug="dup-test-1")
    await db_session.commit()

    runner = _StubRunner()
    svc = IdentitySyncService(db_session)
    await _drive_sweep(svc, target, runner)
    # Bust the payload-hash cache so the second sweep actually re-pushes
    # (otherwise the payload-hash skip would short-circuit before the
    # notification path is reachable — that path is gated on
    # ``was_new``, but defensive against false positives).
    # The second sweep should HIT the notification dedup path either
    # way: ``notified_at`` is non-NULL.
    runner2 = _StubRunner()
    summary2, _ = await _drive_sweep(svc, target, runner2)

    assert summary2.auto_included == 1  # manager still auto-included
    assert summary2.auto_included_notified == 0  # but not renotified

    notifs = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.recipient_person_id == mgr.id,
        )
    )).scalars().all()
    assert len(notifs) == 1  # still just one


@pytest.mark.asyncio
async def test_left_auto_included_manager_not_rehydrated(
    db_session: AsyncSession,
):
    """Regression (2026-07-08): a departed auto-included manager must NOT be
    re-upserted on later sweeps. The rehydration loop pulls in ANY
    ``auto_included_reason`` state row; without an employment gate a manager
    who leaves gets re-created / reactivated on the vendor every nightly sweep,
    undoing the deactivation offboarding's event push just made. The fresh-pull
    path already gates on active — this closes the rehydration path.
    """
    mgr = await _make_person(
        db_session, email="cn.tl2@pingcap.com", is_china=True,
    )
    await _make_person(
        db_session,
        email="jp.eng@pingcap.com",
        is_china=False,
        manager_id=mgr.id,
    )
    target = await _make_navan_target(db_session, slug="left-mgr-1")
    await db_session.commit()

    runner = _StubRunner()
    svc = IdentitySyncService(db_session)
    await _drive_sweep(svc, target, runner)
    # Sweep 1: manager auto-included + upserted.
    assert "cn.tl2@pingcap.com" in {e for e, _ in runner.upserts}

    # Manager leaves (leaver_pending — a bare flip to 'left' with no
    # offboarding process would trip the LeaverBlocked listener trap; both
    # statuses are equally != 'active' and exercise the same gate).
    mgr.employment_status = "leaver_pending"
    await db_session.commit()

    runner2 = _StubRunner()
    await _drive_sweep(svc, target, runner2)
    # Sweep 2: the departed manager must NOT be re-upserted (no resurrection).
    assert "cn.tl2@pingcap.com" not in {e for e, _ in runner2.upserts}, (
        "a departed auto-included manager was rehydrated + re-upserted — "
        "this re-creates the vendor account offboarding just deactivated"
    )


@pytest.mark.asyncio
async def test_no_auto_include_when_no_foreign_reports(
    db_session: AsyncSession,
):
    """Filter shape gates the auto-include heuristic — if every
    in-scope person is foreign and has a FOREIGN manager (so the
    manager already matches the filter), no auto-include fires.
    Also: a target whose in-scope set is entirely CN doesn't pull
    in CN managers (they're already covered or out of scope by
    design)."""
    # Two foreign employees, one is the other's manager — both in
    # scope on the foreigners-only filter, no CN manager to pull.
    boss = await _make_person(
        db_session, email="us.boss@pingcap.com", is_china=False,
    )
    await _make_person(
        db_session,
        email="us.report@pingcap.com",
        is_china=False,
        manager_id=boss.id,
    )
    target = await _make_navan_target(db_session, slug="no-pull-1")
    await db_session.commit()

    runner = _StubRunner()
    svc = IdentitySyncService(db_session)
    summary, _ = await _drive_sweep(svc, target, runner)

    assert summary.auto_included == 0
    assert summary.auto_included_notified == 0
    assert summary.persons_in_scope == 2  # both foreigners, nothing extra


@pytest.mark.asyncio
async def test_auto_included_row_skips_out_of_scope_deactivate(
    db_session: AsyncSession,
):
    """A CN manager that was auto-included MUST remain in scope
    forever — even after the foreign report is removed from the
    target's filter. Without this guard the manager would yo-yo
    between Navan-enabled and Navan-deactivated as team
    composition shifts.
    """
    mgr = await _make_person(
        db_session, email="cn.tl@pingcap.com", is_china=True,
    )
    report = await _make_person(
        db_session,
        email="kr.eng@pingcap.com",
        is_china=False,
        manager_id=mgr.id,
    )
    target = await _make_navan_target(db_session, slug="permanent-1")
    await db_session.commit()

    runner = _StubRunner()
    svc = IdentitySyncService(db_session)
    await _drive_sweep(svc, target, runner)

    # Now mark the foreign report as terminating so they drop out of
    # the active-employment filter. The CN manager should stay,
    # since their state row carries ``auto_included_reason``.
    # ``leaver_pending`` (the HRIS-sync output status) — NOT ``left``:
    # an unguarded flip straight to ``left`` is exactly what the
    # LeaverBlocked listener trap blocks (no offboarding process →
    # LeaverBlockedError), and the listener is registered here. Both
    # statuses are equally outside the target's
    # ``employment_statuses=['active']`` filter.
    report.employment_status = "leaver_pending"
    await db_session.commit()

    runner2 = _StubRunner()
    summary2, _ = await _drive_sweep(svc, target, runner2)

    # Manager should still be in scope (auto_included survives).
    assert summary2.auto_included == 1
    # Runner must NOT receive a deactivate for the CN manager.
    assert "cn.tl@pingcap.com" not in runner2.deactivates
    # The state row stays at success/skipped, never deactivated.
    state = (await db_session.execute(
        select(IdentitySyncState).where(
            IdentitySyncState.target_id == target.id,
            IdentitySyncState.person_id == mgr.id,
        )
    )).scalars().first()
    assert state.last_status in ("success", "skipped")
    assert state.auto_included_reason == "manager_of_foreign"