"""Identity golden-record API — GET profile + login kill-switch PATCH.

Covers the PR1 Identity-Hub spine:
  * full aggregation shape (person / source / login_identities /
    projections / provenance / timeline)
  * downstream projection matrix reflects identity_sync_state rows
  * masked PII never leaks unmasked through the provenance block
  * admin kill-switch / suspend mutation persists + audits
  * RBAC: employee is refused read AND write (403)
  * 404 on a foreign / unknown external identity
"""
from __future__ import annotations

import uuid

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

from app.audit.models import AuditLog
from app.auth.models import AuthUserRole
from app.core.security import create_access_token
from app.domains.identity_sync.models import (
    IdentitySyncState,
    IdentitySyncTarget,
)
from app.identity.person_models import (
    IdentityExternalIdentity,
    IdentityPerson,
)


async def _role_id(db: AsyncSession, name: str) -> uuid.UUID:
    from app.auth.models import AuthRole

    row = (await db.execute(
        select(AuthRole.id).where(AuthRole.name == name)
    )).scalar_one()
    return row


@pytest.fixture
async def subject_person(db_session: AsyncSession):
    """A target person with a Feishu login binding + masked PII."""
    pid = uuid.uuid4()
    person = IdentityPerson(
        id=pid,
        work_email="liyang.zhang@pingcap.com",
        display_name="Liyang Zhang",
        employee_no="P0042",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-0042",
        job_title="Engineer",
        mobile_phone="+8613800001234",
        personal_email="liyang.personal@gmail.com",
    )
    db_session.add(person)
    await db_session.flush()

    ext = IdentityExternalIdentity(
        person_id=pid,
        provider="feishu",
        provider_subject="on_union_abcdef123456",
        provider_email="liyang.zhang@pingcap.com",
        is_login_identity=True,
        identity_status="active",
    )
    db_session.add(ext)
    await db_session.flush()
    await db_session.commit()
    return person, ext


async def test_get_identity_profile_shape(
    client, db_session, admin_headers, subject_person,
):
    person, ext = subject_person
    resp = await client.get(
        f"/api/v1/identity/persons/{person.id}/identity-profile",
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]

    # All five aggregation blocks present.
    assert set(data) >= {
        "person", "source", "login_identities", "projections",
        "provenance", "timeline",
    }
    assert data["person"]["work_email"] == "liyang.zhang@pingcap.com"
    assert data["source"]["source_system"] == "successfactors"
    assert data["source"]["hris_person_id"] == "sf-0042"

    # Login binding surfaces with a MASKED subject + kill-switch state.
    assert len(data["login_identities"]) == 1
    binding = data["login_identities"][0]
    assert binding["provider"] == "feishu"
    assert binding["is_login_identity"] is True
    assert binding["identity_status"] == "active"
    # The raw union_id must never appear verbatim.
    assert "on_union_abcdef123456" not in binding["provider_subject_masked"]
    assert "*" in binding["provider_subject_masked"]


async def test_profile_loads_manager_without_greenlet_error(
    client, db_session, admin_headers,
):
    """Regression: ``person.manager`` is lazy='select' — it MUST be
    eager-loaded or reading it in the async session raises MissingGreenlet.
    The base fixture has no manager, so this case exercises the load path.
    """
    mgr_id = uuid.uuid4()
    mgr = IdentityPerson(
        id=mgr_id, work_email="boss@pingcap.com", display_name="The Boss",
        employment_status="active", source_system="successfactors",
        hris_person_id="sf-boss",
    )
    db_session.add(mgr)
    await db_session.flush()
    report_id = uuid.uuid4()
    report = IdentityPerson(
        id=report_id, work_email="report@pingcap.com", display_name="Report",
        employment_status="active", source_system="successfactors",
        hris_person_id="sf-report", manager_person_id=mgr_id,
    )
    db_session.add(report)
    await db_session.commit()

    resp = await client.get(
        f"/api/v1/identity/persons/{report_id}/identity-profile",
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["person"]["manager"]["display_name"] == "The Boss"


async def test_provenance_masks_pii(
    client, admin_headers, subject_person,
):
    person, _ = subject_person
    resp = await client.get(
        f"/api/v1/identity/persons/{person.id}/identity-profile",
        headers=admin_headers,
    )
    data = resp.json()["data"]
    by_field = {p["field"]: p for p in data["provenance"]}

    # The mobile/personal-email provenance values must be masked — the
    # golden record never leaks raw PII without the audited reveal path.
    assert "13800001234" not in (by_field["mobile_phone"]["value"] or "")
    assert by_field["mobile_phone"]["source"] == "hris"
    assert "liyang.personal" not in (by_field["personal_email"]["value"] or "")
    # HR-managed fields flagged overridable.
    assert by_field["birthday"]["overridable"] is True
    assert by_field["work_email"]["overridable"] is False


async def test_projection_matrix_reflects_sync_state(
    client, db_session, admin_headers, subject_person,
):
    person, _ = subject_person
    target = IdentitySyncTarget(
        name="Navan Production",
        slug="navan-prod",
        kind="navan",
        enabled=True,
        mapping_config={"rules": []},
        filter_config={},
        schedule_config={},
    )
    db_session.add(target)
    await db_session.flush()
    db_session.add(IdentitySyncState(
        target_id=target.id,
        person_id=person.id,
        external_id="navan-user-99",
        last_status="success",
        last_payload_hash="deadbeef",
    ))
    await db_session.commit()

    resp = await client.get(
        f"/api/v1/identity/persons/{person.id}/identity-profile",
        headers=admin_headers,
    )
    data = resp.json()["data"]
    assert len(data["projections"]) == 1
    proj = data["projections"][0]
    assert proj["target_slug"] == "navan-prod"
    assert proj["external_id"] == "navan-user-99"
    assert proj["last_status"] == "success"
    # An empty mapping renders {} → hash differs from "deadbeef" → drift.
    assert proj["drift"] is True


async def test_drift_skipped_when_disabled(
    client, db_session, admin_headers, subject_person,
):
    person, _ = subject_person
    target = IdentitySyncTarget(
        name="Okta", slug="okta", kind="scim", enabled=True,
        mapping_config={"rules": []}, filter_config={}, schedule_config={},
    )
    db_session.add(target)
    await db_session.flush()
    db_session.add(IdentitySyncState(
        target_id=target.id, person_id=person.id,
        last_status="success", last_payload_hash="x",
    ))
    await db_session.commit()

    resp = await client.get(
        f"/api/v1/identity/persons/{person.id}/identity-profile?drift=false",
        headers=admin_headers,
    )
    data = resp.json()["data"]
    assert data["projections"][0]["drift"] is None


async def test_toggle_login_kill_switch_persists_and_audits(
    client, db_session, admin_headers, subject_person,
):
    person, ext = subject_person
    resp = await client.patch(
        f"/api/v1/identity/persons/{person.id}/external-identities/{ext.id}",
        headers=admin_headers,
        json={"is_login_identity": False, "identity_status": "suspended"},
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()["data"]
    assert body["is_login_identity"] is False
    assert body["identity_status"] == "suspended"

    # Persisted.
    refreshed = await db_session.get(IdentityExternalIdentity, ext.id)
    await db_session.refresh(refreshed)
    assert refreshed.is_login_identity is False
    assert refreshed.identity_status == "suspended"

    # Audited.
    logs = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "identity.external_identity.update_login",
            AuditLog.resource_id == str(ext.id),
        )
    )).scalars().all()
    assert len(logs) == 1
    assert logs[0].detail["before"]["is_login_identity"] is True
    assert logs[0].detail["after"]["is_login_identity"] is False


async def test_toggle_unknown_identity_404(
    client, admin_headers, subject_person,
):
    person, _ = subject_person
    resp = await client.patch(
        f"/api/v1/identity/persons/{person.id}/external-identities/{uuid.uuid4()}",
        headers=admin_headers,
        json={"is_login_identity": False},
    )
    assert resp.status_code == 404


async def test_employee_cannot_read_or_write(
    client, db_session, subject_person,
):
    person, ext = subject_person
    # A plain employee (no hr/admin/auditor) — self-mintable PAT parity.
    emp_id = uuid.uuid4()
    emp = IdentityPerson(
        id=emp_id, work_email="nobody@pingcap.com", display_name="Nobody",
        employment_status="active", source_system="successfactors",
        hris_person_id="sf-nobody",
    )
    db_session.add(emp)
    await db_session.flush()
    db_session.add(AuthUserRole(
        user_id=emp_id, role_id=await _role_id(db_session, "employee"),
    ))
    await db_session.commit()
    headers = {"Authorization": f"Bearer {create_access_token(str(emp_id), ['employee'])}"}

    read = await client.get(
        f"/api/v1/identity/persons/{person.id}/identity-profile",
        headers=headers,
    )
    assert read.status_code == 403

    write = await client.patch(
        f"/api/v1/identity/persons/{person.id}/external-identities/{ext.id}",
        headers=headers,
        json={"is_login_identity": False},
    )
    assert write.status_code == 403