"""汇联易切换对账 dry-run — classification + roster pagination.

``dry_run_huilianyi`` is the read-only cutover gate: it must classify
every cutover risk class correctly (email conflicts, vendor-resigned
rows, unmapped company codes, vendor-only actives, the legacy dept
tree) because the admin's enable decision rides on this report.

The vendor roster fetch is stubbed at the module-level
``fetch_all_employees`` / ``fetch_all_departments`` helpers (the
service imports them at call time from the runners module), and
``_build_huilianyi_client`` is stubbed so no integration row / wire is
needed. Everything else — filter evaluation, dept scoping, mapping
inspection — runs the real code against a real DB.
"""
from __future__ import annotations

import uuid
from typing import Any

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

import app.domains.identity_sync.runners.huilianyi as hly_runners
from app.domains.identity_sync.models import IdentitySyncTarget
from app.domains.identity_sync.presets import (
    _HUILIANYI_USER_FILTER,
    HUILIANYI_USER_MAPPING_RULES,
)
from app.domains.identity_sync.runners.huilianyi import (
    HuilianyiClient,
    HuilianyiError,
    fetch_all_departments,
    fetch_all_employees,
)
from app.domains.identity_sync.service import IdentitySyncService
from app.identity.person_models import (
    IdentityOrgMembership,
    IdentityOrgNode,
    IdentityPerson,
)

pytestmark = pytest.mark.asyncio


# ── Pagination helper unit tests ─────────────────────────────────────
class _PagedClient:
    """Stub client mirroring the live server's pagination contract.

    ``pages`` is the ordered list of REAL pages (first page at list
    index 0). The live tenant is 1-indexed and treats ``page=0`` as an
    alias of ``page=1`` (verified 2026-06-11) — the stub reproduces
    that alias so the tests pin the truncation bug it used to cause.
    """

    def __init__(self, pages: list[list[dict[str, Any]]]):
        self.pages = pages
        self.calls: list[dict[str, Any]] = []

    async def request(self, method, path, *, json_body=None, params=None,
                      retry_on_429=2):
        self.calls.append({"path": path, "params": dict(params or {})})
        page = max(int((params or {}).get("page", "0")), 1)  # 0 → 1 alias
        idx = page - 1
        rows = self.pages[idx] if idx < len(self.pages) else []
        return 200, rows, ""

    async def aclose(self):
        return None


class TestRosterPagination:
    async def test_short_page_stops(self):
        full = [{"employeeID": f"P{i:05d}"} for i in range(100)]
        short = [{"employeeID": "P99999"}]
        client = _PagedClient([full, short])
        rows = await fetch_all_employees(client)  # type: ignore[arg-type]
        assert len(rows) == 101
        assert len(client.calls) == 2
        # Full-history window + leavers included.
        p = client.calls[0]["params"]
        assert p["startDate"].startswith("2000-01-01")
        assert p["containLeavedUser"] == "true"

    async def test_starts_at_page_one_no_alias_truncation(self):
        # REGRESSION (2026-06-11): the live server is 1-indexed and
        # serves page=0 as an alias of page=1. The old loop started at
        # 0, so page 1 came back as a full page of duplicates, the
        # fresh-row guard fired, and EVERY roster silently truncated
        # to exactly one page (100 of 286 real employees). Starting at
        # page=1 must walk the full roster.
        page_a = [{"custDeptNumber": f"A{i}"} for i in range(100)]
        page_b = [{"custDeptNumber": f"B{i}"} for i in range(100)]
        short = [{"custDeptNumber": "Z0"}]
        client = _PagedClient([page_a, page_b, short])
        rows = await fetch_all_departments(client)  # type: ignore[arg-type]
        assert len(rows) == 201
        # First request must be page=1 — page=0 would re-trigger the
        # alias truncation this test pins down.
        assert client.calls[0]["params"]["page"] == "1"
        assert [c["params"]["page"] for c in client.calls] == ["1", "2", "3"]

    async def test_server_ignoring_page_param_stops(self):
        # A server that returns the same full page regardless of the
        # page param must trip the fresh-row guard, not spin to
        # _MAX_PAGES.
        full = [{"custDeptNumber": str(i)} for i in range(100)]

        class _StuckClient(_PagedClient):
            async def request(self, method, path, *, json_body=None,
                              params=None, retry_on_429=2):
                self.calls.append({"path": path, "params": dict(params or {})})
                return 200, full, ""

        client = _StuckClient([])
        rows = await fetch_all_departments(client)  # type: ignore[arg-type]
        assert len(rows) == 100
        assert len(client.calls) == 2  # second page yields nothing new → stop

    async def test_non_list_200_body_raises(self):
        # An HTML login page / empty body parses to None — must raise,
        # not read as an empty roster ("everything safe to create").
        class _HtmlClient(_PagedClient):
            async def request(self, method, path, *, json_body=None,
                              params=None, retry_on_429=2):
                return 200, None, "<html>login</html>"

        with pytest.raises(HuilianyiError):
            await fetch_all_employees(_HtmlClient([]))  # type: ignore[arg-type]

    async def test_max_pages_truncation_raises(self, monkeypatch):
        import app.domains.identity_sync.runners.huilianyi as mod
        monkeypatch.setattr(mod, "_MAX_PAGES", 2)
        # Two FULL pages of distinct rows → guard exhausted → raise.
        p0 = [{"employeeID": f"A{i}"} for i in range(100)]
        p1 = [{"employeeID": f"B{i}"} for i in range(100)]
        client = _PagedClient([p0, p1, []])
        with pytest.raises(HuilianyiError, match="_MAX_PAGES"):
            await fetch_all_employees(client)  # type: ignore[arg-type]

    async def test_error_wrapped_page_raises(self):
        class _ErrClient(_PagedClient):
            async def request(self, method, path, *, json_body=None,
                              params=None, retry_on_429=2):
                return 200, {"errorCode": "403001", "message": "无权限"}, ""

        with pytest.raises(HuilianyiError):
            await fetch_all_employees(_ErrClient([]))  # type: ignore[arg-type]


# ── World builders ───────────────────────────────────────────────────
async def _node(db, *, src_id, name, parent=None, level=1):
    db.add(IdentityOrgNode(
        id=uuid.uuid4(),
        source_system="successfactors",
        org_node_source_id=src_id,
        org_node_name=name,
        parent_org_node_source_id=parent,
        org_level=level,
        node_status="active",
    ))
    await db.flush()


async def _person_in(db, *, email, dept_src, employee_no, company_code="DP00001"):
    pid = uuid.uuid4()
    db.add(IdentityPerson(
        id=pid,
        work_email=email,
        display_name=email.split("@")[0],
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{email}",
        employee_no=employee_no,
        company_code=company_code,
        is_china_employee=True,
        country="CHN",
    ))
    await db.flush()
    db.add(IdentityOrgMembership(
        id=uuid.uuid4(),
        person_id=pid,
        source_system="successfactors",
        org_node_source_id=dept_src,
        org_node_name=dept_src,
        is_primary=True,
        membership_status="active",
    ))
    await db.flush()
    return pid


async def _hly_target(db) -> IdentitySyncTarget:
    rules = [dict(r) for r in HUILIANYI_USER_MAPPING_RULES]
    for r in rules:
        if r.get("target_field") == "companyCode":
            # Narrow map so DP09999 below is provably unmapped.
            r["value_map"] = {"DP00001": "001"}
    tgt = IdentitySyncTarget(
        id=uuid.uuid4(),
        name="Test 汇联易",
        slug="test-hly-dryrun",
        kind="huilianyi",
        enabled=False,
        on_offboard="deactivate",
        mapping_config={"rules": rules},
        filter_config=dict(_HUILIANYI_USER_FILTER),
        schedule_config={"trigger": "manual", "cron_expr": None, "timezone": "UTC"},
        runner_options={},
    )
    db.add(tgt)
    await db.flush()
    return tgt


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


VENDOR_USERS = [
    # P00001 alice — exact match → would_update
    {"employeeID": "P00001", "email": "alice@pingcap.com",
     "fullName": "Alice", "status": 1001},
    # P00004 dave — resigned on vendor → resigned_on_vendor
    {"employeeID": "P00004", "email": "dave@pingcap.com",
     "fullName": "Dave", "status": 1003},
    # P00005 eve — vendor still has her OLD email → email_change
    {"employeeID": "P00005", "email": "eve.old@pingcap.com",
     "fullName": "Eve", "status": 1001},
    # P09999 — holds carol's email → carol's create collides
    {"employeeID": "P09999", "email": "carol@pingcap.com",
     "fullName": "Carol Ghost", "status": 1001},
    # P07777 — active on vendor, gate-shaped, missing from Helix
    {"employeeID": "P07777", "email": "ghost@pingcap.com",
     "fullName": "Ghost", "status": 1001},
    # C0013 — active, NON-gate 工号 → counted, not listed
    {"employeeID": "C0013", "email": "legal@pingcap.com",
     "fullName": "Legal Ext", "status": 1001},
]

VENDOR_DEPTS = [
    # ALPHA exists enabled → would_update
    {"custDeptNumber": "ALPHA", "departmentName": "Alpha 部", "status": 101},
    # GAMMA exists but disabled → disabled_on_vendor
    {"custDeptNumber": "GAMMA", "departmentName": "Gamma 部", "status": 102},
    # Legacy Beisen-coded dept (enabled, not Helix-managed) → untouched
    {"custDeptNumber": "12", "departmentName": "Finance(旧)", "status": 101},
]


@pytest.fixture()
def _stub_vendor(monkeypatch):
    async def fake_users(client, *, include_leavers=True):
        return [dict(r) for r in VENDOR_USERS]

    async def fake_depts(client):
        return [dict(r) for r in VENDOR_DEPTS]

    monkeypatch.setattr(hly_runners, "fetch_all_employees", fake_users)
    monkeypatch.setattr(hly_runners, "fetch_all_departments", fake_depts)

    async def fake_client(self, target):
        return _NullClient()

    monkeypatch.setattr(
        IdentitySyncService, "_build_huilianyi_client", fake_client,
    )


class TestDryRunHuilianyi:
    async def test_full_classification(
        self, db_session: AsyncSession, _stub_vendor,
    ):
        await _node(db_session, src_id="ROOT", name="PingCAP", level=1)
        await _node(db_session, src_id="ALPHA", name="Alpha", parent="ROOT", level=2)
        await _node(db_session, src_id="GAMMA", name="Gamma", parent="ROOT", level=2)
        await _node(db_session, src_id="NEW1", name="New Team", parent="ROOT", level=2)
        # Hollow shell — must be skipped, never created on the vendor.
        await _node(db_session, src_id="EMPTY", name="Empty", parent="ROOT", level=2)

        await _person_in(db_session, email="alice@pingcap.com",
                         dept_src="ALPHA", employee_no="P00001")
        await _person_in(db_session, email="bob@pingcap.com",
                         dept_src="NEW1", employee_no="P00002")
        await _person_in(db_session, email="carol@pingcap.com",
                         dept_src="ALPHA", employee_no="P00003")
        await _person_in(db_session, email="dave@pingcap.com",
                         dept_src="ALPHA", employee_no="P00004")
        await _person_in(db_session, email="eve@pingcap.com",
                         dept_src="GAMMA", employee_no="P00005")
        await _person_in(db_session, email="frank@pingcap.com",
                         dept_src="ALPHA", employee_no="P00006",
                         company_code="DP09999")
        # Non-gate service row → excluded by ^P\d+$. Uses a corporate
        # email so this row exercises the employee_no non-employee gate
        # specifically (a non-corporate email would be dropped one step
        # earlier by _load_candidates' corporate-email data-quality gate).
        await _person_in(db_session, email="svc@pingcap.com",
                         dept_src="ALPHA", employee_no="TimeAdmin")
        target = await _hly_target(db_session)
        await db_session.commit()

        svc = IdentitySyncService(db_session)
        plan = await svc.dry_run_huilianyi(target.id)

        u = plan["users"]
        empnos = lambda rows: {r["employee_no"] for r in rows}  # noqa: E731
        # alice exact-match update; eve email-change but still update.
        assert u["would_update_count"] == 2
        # bob (NEW1) is the only clean create — frank is blocked by the
        # unmapped company (still classified into create/update path
        # but listed in unmapped_company; carol is a conflict).
        assert "P00002" in empnos(u["would_create_sample"])
        assert "P00003" in empnos(u["email_conflicts"])
        assert u["email_conflicts"][0]["conflict_with_employee_no"] == "P09999"
        assert empnos(u["resigned_on_vendor"]) == {"P00004"}
        assert empnos(u["email_changes_sample"]) == {"P00005"}
        assert empnos(u["unmapped_company"]) == {"P00006"}
        # P07777 (pure ghost) AND P09999 (the stale row holding carol's
        # email) are both vendor-actives Helix doesn't manage — exactly
        # the review list the admin needs before killing the Authing pull.
        assert {r["employee_no"] for r in u["vendor_only_active"]} == {
            "P07777", "P09999",
        }
        assert u["vendor_nongate_active_count"] == 1  # C0013
        assert u["excluded_nonemployee_count"] == 1  # TimeAdmin
        assert u["missing_employee_no_count"] == 0
        # Conflict rows carry the holder's vendor status (active ghost).
        assert u["email_conflicts"][0]["holder_status"] == "1001"
        # Warnings fire for conflicts / unmapped / vendor-only.
        assert len(plan["warnings"]) >= 3

        d = plan["departments"]
        create_codes = {r["src_id"] for r in d["would_create"]}
        # NEW1 + ROOT are missing on the vendor → created; EMPTY is a
        # hollow shell → skipped, NOT created.
        assert "NEW1" in create_codes and "ROOT" in create_codes
        assert "EMPTY" not in create_codes
        assert {r["src_id"] for r in d["would_update_sample"]} == {"ALPHA"}
        assert {r["src_id"] for r in d["disabled_on_vendor"]} == {"GAMMA"}
        # The legacy Beisen-coded tree shows on the gauge.
        assert d["untouched_vendor_count"] == 1
        assert d["untouched_vendor_sample"][0]["cust_dept_number"] == "12"
        assert d["skipped_empty_count"] >= 1

    async def test_kind_guard(self, db_session: AsyncSession, _stub_vendor):
        tgt = IdentitySyncTarget(
            id=uuid.uuid4(),
            name="scim", slug="test-hly-guard", kind="scim",
            enabled=False, on_offboard="deactivate",
            mapping_config={"rules": []}, filter_config=dict(_HUILIANYI_USER_FILTER),
            schedule_config={"trigger": "manual", "cron_expr": None,
                             "timezone": "UTC"},
            runner_options={"base_url": "https://x.example", "auth_scheme": "bearer"},
        )
        db_session.add(tgt)
        await db_session.commit()
        from app.core.exceptions import AppException
        svc = IdentitySyncService(db_session)
        with pytest.raises(AppException) as ei:
            await svc.dry_run_huilianyi(tgt.id)
        assert ei.value.code == "IDENTITY_SYNC_NOT_HUILIANYI"


class TestClientArrayParsing:
    async def test_request_preserves_list_bodies(self, monkeypatch):
        client = HuilianyiClient(
            base_url="https://api.huilianyi.example",
            client_id="i", client_secret="s",
        )

        async def fake_token(*, force_refresh=False):
            return "tok"

        monkeypatch.setattr(client, "get_access_token", fake_token)

        class _Resp:
            status_code = 200
            text = '[{"employeeID": "P00001"}]'
            headers: dict[str, str] = {}

            def json(self):
                return [{"employeeID": "P00001"}]

        async def fake_request(method, url, headers=None, json=None,
                               params=None):
            return _Resp()

        monkeypatch.setattr(client._client, "request", fake_request)
        status, body, _raw = await client.request("GET", "/api/open/user")
        assert status == 200
        assert isinstance(body, list)
        assert body[0]["employeeID"] == "P00001"
        await client.aclose()