"""Vendor snapshot (对端数据) — refresh, matching, replace, query.

The snapshot is read-only on the vendor side and the basis for both
the 对端数据 tab and vendor-only detection. These tests pin:

* refresh persists normalized rows + matches persons by email AND by
  employee_no (汇联易 keys on 工号)
* wipe-and-replace semantics — a vanished vendor row does not survive
  the next refresh
* huilianyi collector normalization (resigned → inactive, dept 102 →
  inactive)
* scim collector tolerates a missing /Groups endpoint (warning, not
  failure)
* HTTP listing filters + RBAC
"""
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 (
    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 = "huilianyi"):
    t = IdentitySyncTarget(
        id=uuid.uuid4(),
        name=f"Vendor {slug}",
        slug=slug,
        kind=kind,
        enabled=False,
        on_offboard="deactivate",
        mapping_config={"rules": [
            {"source_template": "{work_email}", "target_field": "email"},
        ]},
        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={},
    )
    db.add(t)
    await db.flush()
    return t


async def test_refresh_matches_and_replaces(
    db_session: AsyncSession, monkeypatch,
):
    db = db_session
    alice = await _person(db, email="vs-alice@pingcap.com", employee_no="P00001")
    bob = await _person(db, email="vs-bob@pingcap.com", employee_no="P00002")
    target = await _target(db, slug="vs-replace")
    await db.commit()

    svc = IdentitySyncService(db)

    async def _fake_collect(t):
        return ([
            # email match
            {"record_type": "user", "external_key": "P00001",
             "email": "VS-Alice@pingcap.com", "display_name": "Alice",
             "vendor_status": "1001", "is_active": True, "raw": {"a": 1}},
            # employee_no match (vendor email stale)
            {"record_type": "user", "external_key": "P00002",
             "email": "old-bob@pingcap.com", "display_name": "Bob",
             "vendor_status": "1001", "is_active": True, "raw": {}},
            # vendor-only
            {"record_type": "user", "external_key": "C0013",
             "email": "legal@pingcap.com", "display_name": "Legal Bot",
             "vendor_status": "1001", "is_active": True, "raw": {}},
            {"record_type": "department", "external_key": "1809819",
             "display_name": "Finance", "vendor_status": "101",
             "is_active": True, "raw": {}},
        ], ["test warning"])

    monkeypatch.setattr(svc, "_collect_vendor_records", _fake_collect)
    result = await svc.refresh_vendor_snapshot(
        target_id=target.id, actor_id=None,
    )
    assert result["counts"] == {"user": 3, "department": 1}
    assert result["matched_users"] == 2
    assert result["vendor_only_users"] == 1
    assert result["warnings"] == ["test warning"]

    rows = (await db.execute(
        select(IdentitySyncVendorRecord).where(
            IdentitySyncVendorRecord.target_id == target.id,
        )
    )).scalars().all()
    by_key = {r.external_key: r for r in rows}
    assert by_key["P00001"].matched_person_id == alice.id
    assert by_key["P00001"].email == "vs-alice@pingcap.com"  # lowercased
    assert by_key["P00002"].matched_person_id == bob.id  # via employee_no
    assert by_key["C0013"].matched_person_id is None

    # Second refresh with a shrunken roster — stale rows must vanish.
    async def _fake_collect2(t):
        return ([
            {"record_type": "user", "external_key": "P00001",
             "email": "vs-alice@pingcap.com", "display_name": "Alice",
             "vendor_status": "1001", "is_active": True, "raw": {}},
        ], [])

    monkeypatch.setattr(svc, "_collect_vendor_records", _fake_collect2)
    await svc.refresh_vendor_snapshot(target_id=target.id, actor_id=None)
    rows = (await db.execute(
        select(IdentitySyncVendorRecord).where(
            IdentitySyncVendorRecord.target_id == target.id,
        )
    )).scalars().all()
    assert {r.external_key for r in rows} == {"P00001"}


async def test_huilianyi_collector_normalization(
    db_session: AsyncSession, monkeypatch,
):
    db = db_session
    target = await _target(db, slug="vs-hly")
    await db.commit()
    svc = IdentitySyncService(db)

    class _NullClient:
        async def aclose(self):
            return None

    async def _fake_client(t):
        return _NullClient()

    import app.domains.identity_sync.runners.huilianyi as hly

    async def _fake_users(client, **kw):
        return [
            {"employeeID": "P00001", "fullName": "Alice",
             "email": "a@pingcap.com", "status": 1001},
            {"employeeID": "P00009", "fullName": "Gone",
             "email": "g@pingcap.com", "status": 1003},
        ]

    async def _fake_depts(client, **kw):
        return [
            {"custDeptNumber": "1809819", "name": "Finance", "status": 101},
            {"custDeptNumber": "12", "name": "Legacy", "status": 102},
        ]

    monkeypatch.setattr(svc, "_build_huilianyi_client", _fake_client)
    monkeypatch.setattr(hly, "fetch_all_employees", _fake_users)
    monkeypatch.setattr(hly, "fetch_all_departments", _fake_depts)

    records, warnings = await svc._collect_huilianyi_records(target)
    by_key = {r["external_key"]: r for r in records}
    assert by_key["P00001"]["is_active"] is True
    assert by_key["P00009"]["is_active"] is False  # 1003 = resigned
    assert by_key["1809819"]["is_active"] is True
    assert by_key["12"]["is_active"] is False  # 102 = disabled
    assert warnings == []


async def test_scim_collector_tolerates_missing_groups(
    db_session: AsyncSession, monkeypatch,
):
    db = db_session
    target = await _target(db, slug="vs-scim", kind="scim")
    await db.commit()
    svc = IdentitySyncService(db)

    from app.domains.identity_sync.runners.scim import ScimError

    class _StubScim:
        async def list_users(self):
            return [
                {"id": "u-1", "userName": "a@pingcap.com",
                 "displayName": "Alice", "active": True},
                {"id": "u-2", "userName": "b@pingcap.com",
                 "displayName": "Bob", "active": False},
            ]

        async def list_groups(self):
            raise ScimError("GET /Groups → 501", status_code=501)

        async def aclose(self):
            return None

    async def _fake_build(t):
        return _StubScim()

    monkeypatch.setattr(svc, "_build_runner", _fake_build)
    records, warnings = await svc._collect_scim_records(target)
    assert {r["external_key"] for r in records} == {"u-1", "u-2"}
    assert next(
        r for r in records if r["external_key"] == "u-2"
    )["is_active"] is False
    assert len(warnings) == 1 and "Groups" in warnings[0]


async def test_vendor_records_http_and_rbac(
    client, admin_headers, employee_headers, db_session: AsyncSession,
):
    db = db_session
    p = await _person(db, email="vs-http@pingcap.com")
    target = await _target(db, slug="vs-http")
    from datetime import UTC, datetime
    now = datetime.now(UTC)
    db.add_all([
        IdentitySyncVendorRecord(
            id=uuid.uuid4(), target_id=target.id, record_type="user",
            external_key="P00001", display_name="Alice",
            email="vs-http@pingcap.com", vendor_status="1001",
            is_active=True, matched_person_id=p.id, raw={"x": 1},
            fetched_at=now,
        ),
        IdentitySyncVendorRecord(
            id=uuid.uuid4(), target_id=target.id, record_type="user",
            external_key="C0001", display_name="Ghost",
            email="ghost@pingcap.com", vendor_status="1001",
            is_active=True, matched_person_id=None, raw={},
            fetched_at=now,
        ),
        IdentitySyncVendorRecord(
            id=uuid.uuid4(), target_id=target.id, record_type="department",
            external_key="1809819", display_name="Finance",
            vendor_status="101", is_active=True, raw={},
            fetched_at=now,
        ),
    ])
    await db.commit()

    base = f"/api/v1/identity-sync/targets/{target.id}/vendor-records"
    resp = await client.get(
        base, params={"record_type": "user"}, headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert len(body["data"]) == 2
    assert body["meta"]["fetched_at"] is not None

    resp = await client.get(
        base, params={"record_type": "user", "matched": "false"},
        headers=admin_headers,
    )
    assert [r["external_key"] for r in resp.json()["data"]] == ["C0001"]

    resp = await client.get(
        base, params={"q": "fina"}, headers=admin_headers,
    )
    assert [r["external_key"] for r in resp.json()["data"]] == ["1809819"]

    resp = await client.get(base, headers=employee_headers)
    assert resp.status_code == 403