"""Unmapped-department audit alert for the 汇联易 recode target.

Regression for the recurring "split" bug (prod 3×: Business Operations,
IT Services & Security, AI Lab): a Helix org node with NO
``huilianyi_dept_code_map`` entry passes its raw SF code through and can
create a parallel vendor department that doesn't match the NS/Authing-era
code — discovered weeks later by finance when cost attribution diverged.

:func:`maybe_notify_unmapped_depts` surfaces the gap the moment it
appears WITHOUT failing the sync: the run stays green, IT just gets an
actionable "map these depts" heads-up (user decision 2026-07-07 — a
non-blocking alert, not an error).

Pins:

* an active node absent from the map ⇒ one alert to configured
  recipients, carrying the unmapped code in ``context``;
* every node mapped (incl. mapped-to-itself) ⇒ no alert;
* empty map (passthrough-by-design, not a recode target) ⇒ no alert;
* ``alert_enabled=False`` ⇒ no alert;
* the same unmapped SET on a later run ⇒ dedup (no repeat push);
* no eligible recipients ⇒ nobody paged (no IT-admin fallback).
"""
from __future__ import annotations

import uuid
from unittest import mock

import pytest
import pytest_asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from app.domains.identity_sync.alert_notifier import (
    maybe_notify_unadopted_ns_depts,
    maybe_notify_unmapped_depts,
)
from app.domains.identity_sync.models import IdentitySyncTarget
from app.domains.identity_sync.service import IdentitySyncService
from app.identity.person_models import IdentityOrgNode, IdentityPerson
from app.notifications.models import NotificationOutbox

pytestmark = pytest.mark.asyncio

_TEMPLATE = "identity_sync.unmapped_depts"


@pytest_asyncio.fixture
async def patched_session_factory(db_engine):
    from app.core import database as _db

    original = _db.async_session_factory
    _db.async_session_factory = async_sessionmaker(
        db_engine, class_=AsyncSession, expire_on_commit=False,
    )
    try:
        yield
    finally:
        _db.async_session_factory = original


def patched_factory_session(_db_session):
    from app.core import database as _db

    return _db.async_session_factory()


async def _person(db: AsyncSession, *, email: str) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=email.split("@", 1)[0],
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{email}",
    )
    db.add(p)
    await db.flush()
    return p


async def _org_node(db: AsyncSession, *, src_id: str, name: str) -> None:
    db.add(IdentityOrgNode(
        id=uuid.uuid4(),
        source_system="successfactors",
        org_node_source_id=src_id,
        org_node_name=name,
        node_status="active",
    ))
    await db.flush()


async def _target(
    db: AsyncSession,
    *,
    slug: str,
    code_map: dict[str, str] | None,
    alert_recipient_ids: list[str] | None,
    alert_enabled: bool = True,
) -> IdentitySyncTarget:
    runner_options: dict = {}
    if code_map is not None:
        runner_options["huilianyi_dept_code_map"] = code_map
    t = IdentitySyncTarget(
        id=uuid.uuid4(),
        name=f"HLY {slug}",
        slug=slug,
        kind="huilianyi",
        enabled=True,
        on_offboard="deactivate",
        alert_enabled=alert_enabled,
        mapping_config={"rules": [
            {"source_template": "{employee_id}", "target_field": "employeeID"},
        ]},
        filter_config={
            "include_offboarded": False, "departments": [],
            "include_subtree": True, "countries": [],
            "employment_statuses": [], "job_title_includes": [],
            "email_suffixes": [], "person_ids": [], "mode": "and",
        },
        schedule_config={"trigger": "cron", "cron_expr": "0 2 * * *",
                         "timezone": "Asia/Shanghai"},
        runner_options=runner_options,
        alert_recipient_ids=alert_recipient_ids,
    )
    db.add(t)
    await db.flush()
    return t


async def _outbox(db: AsyncSession, *, person_id) -> list[NotificationOutbox]:
    rows = (
        await db.execute(
            select(NotificationOutbox).where(
                NotificationOutbox.template_key == _TEMPLATE,
                NotificationOutbox.recipient_person_id == person_id,
            )
        )
    ).scalars().all()
    return list(rows)


async def test_unmapped_node_alerts_recipient(
    db_session: AsyncSession, patched_session_factory,
):
    db = db_session
    recipient = await _person(db, email="sync-oncall@pingcap.com")
    await _org_node(db, src_id="D_MAPPED", name="Product")
    await _org_node(db, src_id="D_NEW", name="AI Lab")  # unmapped
    target = await _target(
        db, slug="hly-unmapped",
        code_map={"D_MAPPED": "20240101-1"},
        alert_recipient_ids=[str(recipient.id)],
    )
    await db.commit()

    await maybe_notify_unmapped_depts(target_id=target.id)

    async with patched_factory_session(db_session) as fresh:
        rows = await _outbox(fresh, person_id=recipient.id)
        assert len(rows) == 1
        ctx = rows[0].context or {}
        assert ctx.get("unmapped_count") == 1
        assert "D_NEW" in (ctx.get("unmapped_codes") or [])
        assert "D_MAPPED" not in (ctx.get("unmapped_codes") or [])


async def test_all_mapped_no_alert(
    db_session: AsyncSession, patched_session_factory,
):
    db = db_session
    recipient = await _person(db, email="sync-allmapped@pingcap.com")
    await _org_node(db, src_id="D1", name="Product")
    await _org_node(db, src_id="D2", name="New Dept")
    target = await _target(
        db, slug="hly-allmapped",
        # D2 mapped-to-itself = explicit "genuinely new" acknowledgement.
        code_map={"D1": "20240101-1", "D2": "D2"},
        alert_recipient_ids=[str(recipient.id)],
    )
    await db.commit()

    await maybe_notify_unmapped_depts(target_id=target.id)

    async with patched_factory_session(db_session) as fresh:
        assert await _outbox(fresh, person_id=recipient.id) == []


async def test_empty_map_no_alert(
    db_session: AsyncSession, patched_session_factory,
):
    """Empty map = passthrough-by-design, not a recode/cutover target —
    an unmapped node is expected there, so no alert."""
    db = db_session
    recipient = await _person(db, email="sync-nomap@pingcap.com")
    await _org_node(db, src_id="D_ANY", name="Whatever")
    target = await _target(
        db, slug="hly-nomap", code_map={},
        alert_recipient_ids=[str(recipient.id)],
    )
    await db.commit()

    await maybe_notify_unmapped_depts(target_id=target.id)

    async with patched_factory_session(db_session) as fresh:
        assert await _outbox(fresh, person_id=recipient.id) == []


async def test_switch_off_suppresses(
    db_session: AsyncSession, patched_session_factory,
):
    db = db_session
    recipient = await _person(db, email="sync-muted@pingcap.com")
    await _org_node(db, src_id="D_NEW", name="AI Lab")
    target = await _target(
        db, slug="hly-muted", code_map={"D_OTHER": "x"},
        alert_recipient_ids=[str(recipient.id)], alert_enabled=False,
    )
    await db.commit()

    await maybe_notify_unmapped_depts(target_id=target.id)

    async with patched_factory_session(db_session) as fresh:
        assert await _outbox(fresh, person_id=recipient.id) == []


async def test_same_set_dedups(
    db_session: AsyncSession, patched_session_factory,
):
    db = db_session
    recipient = await _person(db, email="sync-dedup@pingcap.com")
    await _org_node(db, src_id="D_NEW", name="AI Lab")
    target = await _target(
        db, slug="hly-dedup", code_map={"D_OTHER": "x"},
        alert_recipient_ids=[str(recipient.id)],
    )
    await db.commit()

    await maybe_notify_unmapped_depts(target_id=target.id)
    # Second sweep, SAME unmapped set → dedup, no repeat push.
    await maybe_notify_unmapped_depts(target_id=target.id)

    async with patched_factory_session(db_session) as fresh:
        assert len(await _outbox(fresh, person_id=recipient.id)) == 1


async def test_no_recipients_pages_nobody(
    db_session: AsyncSession, patched_session_factory,
):
    db = db_session
    admin = await _person(db, email="sync-norecip@pingcap.com")
    await _org_node(db, src_id="D_NEW", name="AI Lab")
    target = await _target(
        db, slug="hly-norecip", code_map={"D_OTHER": "x"},
        alert_recipient_ids=None,
    )
    await db.commit()

    # Must not raise, must page nobody (no IT-admin fallback).
    await maybe_notify_unmapped_depts(target_id=target.id)

    async with patched_factory_session(db_session) as fresh:
        assert await _outbox(fresh, person_id=admin.id) == []


# ── NetSuite unadopted-department audit ──────────────────────────────
# Same recurring bug on NS: an org node with no matching externalId is
# the split-risk seed once create_missing is flipped on. The audit fires
# from the (gated) NS dept phase and, before that flip, lists the depts
# that must be adopted in place. NS externalId set is read via SuiteQL —
# faked here so the test is hermetic.
_NS_TEMPLATE = "identity_sync.unadopted_ns_depts"


class _FakeNSClient:
    """Stand-in for NetSuiteClient: `suiteql` returns the adopted
    externalId rows (or raises to simulate an unreadable set)."""

    def __init__(self, adopted, *, raise_on_query: bool = False):
        self._adopted = adopted
        self._raise = raise_on_query

    async def suiteql(self, _q):
        if self._raise:
            raise RuntimeError("simulated SuiteQL failure")
        return [{"eid": c} for c in self._adopted]

    async def aclose(self):
        pass


def _patch_ns_client(adopted, *, raise_on_query: bool = False):
    async def _build(_self, _target):
        return _FakeNSClient(adopted, raise_on_query=raise_on_query)

    return mock.patch.object(
        IdentitySyncService, "_build_netsuite_client", _build,
    )


async def _ns_target(
    db: AsyncSession,
    *,
    slug: str,
    alert_recipient_ids: list[str] | None,
    create_missing: bool = False,
    alert_enabled: bool = True,
) -> IdentitySyncTarget:
    t = IdentitySyncTarget(
        id=uuid.uuid4(),
        name=f"NS {slug}",
        slug=slug,
        kind="netsuite",
        enabled=True,
        on_offboard="deactivate",
        alert_enabled=alert_enabled,
        mapping_config={"rules": [
            {"source_template": "{employee_id}", "target_field": "externalId"},
        ]},
        filter_config={
            "include_offboarded": False, "departments": [],
            "include_subtree": True, "countries": [],
            "employment_statuses": [], "job_title_includes": [],
            "email_suffixes": [], "person_ids": [], "mode": "and",
        },
        schedule_config={"trigger": "manual", "cron_expr": None,
                         "timezone": "UTC"},
        runner_options={"netsuite_create_missing": create_missing},
        alert_recipient_ids=alert_recipient_ids,
    )
    db.add(t)
    await db.flush()
    return t


async def _ns_outbox(db: AsyncSession, *, person_id) -> list[NotificationOutbox]:
    rows = (
        await db.execute(
            select(NotificationOutbox).where(
                NotificationOutbox.template_key == _NS_TEMPLATE,
                NotificationOutbox.recipient_person_id == person_id,
            )
        )
    ).scalars().all()
    return list(rows)


async def test_ns_unadopted_alerts(
    db_session: AsyncSession, patched_session_factory,
):
    db = db_session
    recipient = await _person(db, email="ns-oncall@pingcap.com")
    await _org_node(db, src_id="D_ADOPTED", name="Product")
    await _org_node(db, src_id="D_UNADOPTED", name="AI Lab")
    target = await _ns_target(
        db, slug="ns-unadopted", alert_recipient_ids=[str(recipient.id)],
    )
    await db.commit()

    with _patch_ns_client({"D_ADOPTED"}):
        await maybe_notify_unadopted_ns_depts(target_id=target.id)

    async with patched_factory_session(db_session) as fresh:
        rows = await _ns_outbox(fresh, person_id=recipient.id)
        assert len(rows) == 1
        ctx = rows[0].context or {}
        assert ctx.get("unadopted_count") == 1
        assert "D_UNADOPTED" in (ctx.get("unadopted_codes") or [])
        assert ctx.get("create_missing") is False


async def test_ns_all_adopted_no_alert(
    db_session: AsyncSession, patched_session_factory,
):
    db = db_session
    recipient = await _person(db, email="ns-alladopted@pingcap.com")
    await _org_node(db, src_id="D1", name="Product")
    await _org_node(db, src_id="D2", name="Eng")
    target = await _ns_target(
        db, slug="ns-alladopted", alert_recipient_ids=[str(recipient.id)],
    )
    await db.commit()

    with _patch_ns_client({"D1", "D2"}):
        await maybe_notify_unadopted_ns_depts(target_id=target.id)

    async with patched_factory_session(db_session) as fresh:
        assert await _ns_outbox(fresh, person_id=recipient.id) == []


async def test_ns_create_missing_flags_danger(
    db_session: AsyncSession, patched_session_factory,
):
    """With create_missing ON, an unadopted node WILL be created as a
    parallel dept — the card carries create_missing=True (danger tone)."""
    db = db_session
    recipient = await _person(db, email="ns-danger@pingcap.com")
    await _org_node(db, src_id="D_NEW", name="AI Lab")
    target = await _ns_target(
        db, slug="ns-danger", alert_recipient_ids=[str(recipient.id)],
        create_missing=True,
    )
    await db.commit()

    with _patch_ns_client(set()):  # nothing adopted
        await maybe_notify_unadopted_ns_depts(target_id=target.id)

    async with patched_factory_session(db_session) as fresh:
        rows = await _ns_outbox(fresh, person_id=recipient.id)
        assert len(rows) == 1
        assert (rows[0].context or {}).get("create_missing") is True


async def test_ns_suiteql_failure_no_alert(
    db_session: AsyncSession, patched_session_factory,
):
    """An unreadable NS externalId set must ABORT — never alert on an
    unreliable adopted set (would flag every node as unadopted)."""
    db = db_session
    recipient = await _person(db, email="ns-readfail@pingcap.com")
    await _org_node(db, src_id="D_NEW", name="AI Lab")
    target = await _ns_target(
        db, slug="ns-readfail", alert_recipient_ids=[str(recipient.id)],
    )
    await db.commit()

    with _patch_ns_client(set(), raise_on_query=True):
        await maybe_notify_unadopted_ns_depts(target_id=target.id)

    async with patched_factory_session(db_session) as fresh:
        assert await _ns_outbox(fresh, person_id=recipient.id) == []


async def test_ns_same_set_dedups(
    db_session: AsyncSession, patched_session_factory,
):
    db = db_session
    recipient = await _person(db, email="ns-dedup@pingcap.com")
    await _org_node(db, src_id="D_NEW", name="AI Lab")
    target = await _ns_target(
        db, slug="ns-dedup", alert_recipient_ids=[str(recipient.id)],
    )
    await db.commit()

    with _patch_ns_client(set()):
        await maybe_notify_unadopted_ns_depts(target_id=target.id)
        await maybe_notify_unadopted_ns_depts(target_id=target.id)

    async with patched_factory_session(db_session) as fresh:
        assert len(await _ns_outbox(fresh, person_id=recipient.id)) == 1