"""Drift diff — Helix render vs vendor snapshot, field verdicts.

Pins the comparison semantics the admin's enable decision rides on:

* value mismatch → ``drift``; vendor-not-echoed field →
  ``not_comparable`` (NEVER drift — read-API shape isn't divergence)
* normalization: email case, scalar/str coercion, SCIM emails list
  multiset
* vendor record location: state.external_id → matched_person → email
* buckets: clean / drifted / vendor_missing / render_failed
* needs_snapshot short-circuit + google customSchemas remap + RBAC
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.identity_sync.models import (
    IdentitySyncState,
    IdentitySyncTarget,
    IdentitySyncVendorRecord,
)
from app.domains.identity_sync.service import IdentitySyncService
from app.identity.person_models import IdentityPerson

pytestmark = pytest.mark.asyncio


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


async def _target(db, *, slug: str, kind: str = "scim",
                  rules: list | None = None,
                  runner_options: dict | None = None):
    t = IdentitySyncTarget(
        id=uuid.uuid4(),
        name=f"Drift {slug}",
        slug=slug,
        kind=kind,
        enabled=False,
        on_offboard="deactivate",
        mapping_config={"rules": rules or [
            {"source_template": "{work_email}", "target_field": "userName"},
            {"source_template": "{job_title}", "target_field": "title"},
        ]},
        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=runner_options or {},
    )
    db.add(t)
    await db.flush()
    return t


def _vendor(target_id, *, key: str, email: str | None,
            raw: dict, person_id=None):
    return IdentitySyncVendorRecord(
        id=uuid.uuid4(), target_id=target_id, record_type="user",
        external_key=key, display_name=key, email=email,
        vendor_status="active", is_active=True,
        matched_person_id=person_id, raw=raw,
        fetched_at=datetime.now(UTC),
    )


async def test_needs_snapshot_short_circuit(db_session: AsyncSession):
    db = db_session
    target = await _target(db, slug="drift-empty")
    await db.commit()
    out = await IdentitySyncService(db).compute_drift(target_id=target.id)
    assert out["needs_snapshot"] is True
    assert out["rows"] == []


async def test_field_verdicts_and_buckets(db_session: AsyncSession):
    db = db_session
    # clean (email case-insensitive match), drifted, vendor-missing.
    clean = await _person(db, email="drift-clean@pingcap.com")
    clean.job_title = "Engineer"
    drifted = await _person(db, email="drift-bad@pingcap.com")
    drifted.job_title = "Engineer"
    await _person(db, email="drift-missing@pingcap.com")
    target = await _target(db, slug="drift-buckets")
    db.add_all([
        _vendor(
            target.id, key="u-clean", email="drift-clean@pingcap.com",
            raw={
                "userName": "DRIFT-CLEAN@pingcap.com",
                "title": "Engineer",
            },
            person_id=clean.id,
        ),
        _vendor(
            target.id, key="u-bad", email="drift-bad@pingcap.com",
            raw={
                "userName": "drift-bad@pingcap.com",
                "title": "Manager",  # ← drift
            },
            person_id=drifted.id,
        ),
    ])
    await db.commit()

    out = await IdentitySyncService(db).compute_drift(
        target_id=target.id, q="drift-",
    )
    assert out["needs_snapshot"] is False
    s = out["summary"]
    assert s["clean"] == 1
    assert s["drifted"] == 1
    assert s["vendor_missing"] == 1

    rows = {r["email"]: r for r in out["rows"]}
    bad = rows["drift-bad@pingcap.com"]
    assert bad["status"] == "drift"
    assert bad["drift_count"] == 1
    verdicts = {f["field"]: f["status"] for f in bad["fields"]}
    assert verdicts["title"] == "drift"
    assert verdicts["userName"] == "match"
    assert rows["drift-clean@pingcap.com"]["status"] == "clean"
    assert rows["drift-missing@pingcap.com"]["status"] == "vendor_missing"


async def test_vendor_not_echoed_is_not_comparable(
    db_session: AsyncSession,
):
    db = db_session
    p = await _person(db, email="drift-nc@pingcap.com")
    p.job_title = "Engineer"
    target = await _target(db, slug="drift-nc")
    db.add(_vendor(
        target.id, key="u-nc", email="drift-nc@pingcap.com",
        # Vendor read API doesn't echo title at all.
        raw={"userName": "drift-nc@pingcap.com"},
        person_id=p.id,
    ))
    await db.commit()
    out = await IdentitySyncService(db).compute_drift(
        target_id=target.id, q="drift-nc",
    )
    row = out["rows"][0]
    assert row["status"] == "clean"  # not_comparable ≠ drift
    verdicts = {f["field"]: f["status"] for f in row["fields"]}
    assert verdicts["title"] == "not_comparable"


async def test_external_id_lookup_beats_email(db_session: AsyncSession):
    db = db_session
    p = await _person(db, email="drift-ext@pingcap.com")
    target = await _target(db, slug="drift-ext", rules=[
        {"source_template": "{work_email}", "target_field": "userName"},
    ])
    db.add(IdentitySyncState(
        id=uuid.uuid4(), target_id=target.id, person_id=p.id,
        last_status="success", external_id="u-ext-1",
    ))
    # Vendor row has a STALE email — only external_id can find it.
    db.add(_vendor(
        target.id, key="u-ext-1", email="old-name@pingcap.com",
        raw={"userName": "old-name@pingcap.com"},
    ))
    await db.commit()
    out = await IdentitySyncService(db).compute_drift(
        target_id=target.id, q="drift-ext",
    )
    row = out["rows"][0]
    assert row["external_key"] == "u-ext-1"
    assert row["status"] == "drift"  # userName drifted (stale email)


@pytest.mark.asyncio
async def test_drift_row_carries_last_verified_at(db_session: AsyncSession):
    """The drift row surfaces the state's last_verified_at (drift-reverify
    cadence observability) — an ISO string when set, None when never verified."""
    import datetime as _dt

    db = db_session
    p = await _person(db, email="drift-lv@pingcap.com")
    target = await _target(db, slug="drift-lv", rules=[
        {"source_template": "{work_email}", "target_field": "userName"},
    ])
    lv = _dt.datetime(2026, 7, 10, 3, 0, tzinfo=_dt.UTC)
    db.add(IdentitySyncState(
        id=uuid.uuid4(), target_id=target.id, person_id=p.id,
        last_status="success", external_id="u-lv-1", last_verified_at=lv,
    ))
    db.add(_vendor(
        target.id, key="u-lv-1", email="drift-lv@pingcap.com",
        raw={"userName": "drift-lv@pingcap.com"},
    ))
    await db.commit()
    out = await IdentitySyncService(db).compute_drift(
        target_id=target.id, q="drift-lv",
    )
    row = out["rows"][0]
    assert row["last_verified_at"] is not None
    assert row["last_verified_at"].startswith("2026-07-10T03:00")


async def test_google_schema_remap(db_session: AsyncSession):
    db = db_session
    p = await _person(db, email="drift-g@pingcap.com")
    target = await _target(
        db, slug="drift-g", kind="google_schema",
        rules=[
            {"source_template": "{department_l1}", "target_field": "L1_Dept"},
        ],
        runner_options={"schema_id": "Department_Information"},
    )
    db.add(_vendor(
        target.id, key="drift-g@pingcap.com", email="drift-g@pingcap.com",
        raw={
            "primaryEmail": "drift-g@pingcap.com",
            "customSchemas": {
                "Department_Information": {"L1_Dept": "Engineering"},
            },
        },
        person_id=p.id,
    ))
    await db.commit()
    out = await IdentitySyncService(db).compute_drift(
        target_id=target.id, q="drift-g",
    )
    row = out["rows"][0]
    # Person has no org membership → payload renders without L1_Dept,
    # or renders empty — either way the row exists and the lookup path
    # proves the remap (no exception, vendor located via match).
    assert row["status"] in ("clean", "drift")


async def test_drift_http_rbac(client, admin_headers, employee_headers,
                               db_session: AsyncSession):
    db = db_session
    target = await _target(db, slug="drift-http")
    await db.commit()
    resp = await client.get(
        f"/api/v1/identity-sync/targets/{target.id}/drift",
        headers=admin_headers,
    )
    assert resp.status_code == 200
    assert resp.json()["data"]["needs_snapshot"] is True
    resp = await client.get(
        f"/api/v1/identity-sync/targets/{target.id}/drift",
        headers=employee_headers,
    )
    assert resp.status_code == 403


async def test_scim_urn_extension_paths_compare(db_session: AsyncSession):
    """REGRESSION (终轮复审 P1): the SCIM enterprise-extension URN key
    contains a literal dot ("…:2.0:User"); a string join/split
    round-trip sheared it apart and made every extension field
    not_comparable — drift on exactly the Authing-parity fields could
    never be detected. Segment-tuple flatten + walk must compare them.
    """
    db = db_session
    p = await _person(db, email="drift-urn@pingcap.com",
                      employee_no="P00042")
    urn = "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User"
    target = await _target(db, slug="drift-urn", rules=[
        {"source_template": "{work_email}", "target_field": "userName"},
        {"source_template": "{employee_id}",
         "target_field": f"{urn}.employeeNumber"},
    ])
    db.add(_vendor(
        target.id, key="u-urn", email="drift-urn@pingcap.com",
        raw={
            "userName": "drift-urn@pingcap.com",
            urn: {"employeeNumber": "P99999"},  # ← real drift
        },
        person_id=p.id,
    ))
    await db.commit()
    out = await IdentitySyncService(db).compute_drift(
        target_id=target.id, q="drift-urn",
    )
    row = out["rows"][0]
    assert row["status"] == "drift"
    verdicts = {f["field"]: f["status"] for f in row["fields"]}
    assert verdicts[f"{urn}.employeeNumber"] == "drift"


async def test_summary_counts_full_scope_even_with_q(
    db_session: AsyncSession,
):
    """REGRESSION (终轮复审 P2): the stat tiles must answer "is this
    target clean?" for the WHOLE scope — q only narrows the row list.
    """
    db = db_session
    a = await _person(db, email="qsum-a@pingcap.com")
    a.job_title = "Engineer"
    b = await _person(db, email="qsum-b@pingcap.com")
    b.job_title = "Engineer"
    target = await _target(db, slug="drift-qsum")
    db.add_all([
        _vendor(target.id, key="u-a", email="qsum-a@pingcap.com",
                raw={"userName": "qsum-a@pingcap.com",
                     "title": "Engineer"}, person_id=a.id),
        _vendor(target.id, key="u-b", email="qsum-b@pingcap.com",
                raw={"userName": "qsum-b@pingcap.com",
                     "title": "Manager"}, person_id=b.id),  # drift
    ])
    await db.commit()
    out = await IdentitySyncService(db).compute_drift(
        target_id=target.id, q="qsum-a",
    )
    # Rows narrowed to the q match…
    assert {r["email"] for r in out["rows"]} == {"qsum-a@pingcap.com"}
    # …but the summary still covers BOTH people (and counts b's drift),
    # plus everyone else in the unrestricted default-filter scope.
    assert out["summary"]["drifted"] >= 1
    assert out["summary"]["compared"] >= 2


async def test_huilianyi_entrytime_midnight_pad_is_not_drift(
    db_session: AsyncSession,
):
    """REGRESSION (终轮复审 P2): the vendor echoes ISO midnight for a date
    field — that shape may not read as drift against the bare
    "YYYY-MM-DD" render.

    Also pins the WRITE→READ key alias (2026-08-27): Helix writes 入职时间
    as ``entryTime`` (创建/更新员工V2) while the 增量查询员工 snapshot echoes
    it as ``entryDate``. Without the alias the comparator looks up the write
    name, misses, and downgrades 入职时间 to ``not_comparable`` — the field
    would stop being audited exactly when it started being pushed.
    """
    db = db_session
    p = await _person(db, email="drift-pad@pingcap.com")
    from datetime import date
    p.hire_date = date(2023, 4, 1)
    target = await _target(db, slug="drift-pad", kind="huilianyi", rules=[
        {"source_template": "{work_email}", "target_field": "email"},
        {"source_template": "{hire_date}", "target_field": "entryTime"},
    ])
    db.add(_vendor(
        target.id, key="P00077", email="drift-pad@pingcap.com",
        raw={
            "email": "drift-pad@pingcap.com",
            "entryDate": "2023-04-01T00:00:00Z",  # vendor echo shape
        },
        person_id=p.id,
    ))
    await db.commit()
    out = await IdentitySyncService(db).compute_drift(
        target_id=target.id, q="drift-pad",
    )
    row = out["rows"][0]
    verdicts = {f["field"]: f["status"] for f in row["fields"]}
    assert verdicts["entryTime"] == "match"
    assert row["status"] == "clean"