"""Governance API — /api/v1/identity/consumers CRUD + token lifecycle.

Pins:

* RBAC: employee → 403 on every governance endpoint
* create validates: unknown scope, pii-without-base-scope, bad
  filter_config, non-active owner, duplicate slug (incl. soft-deleted)
* slug is immutable (PATCH has no slug field at all)
* token mint returns plaintext exactly once; listing shows prefix only
* active-token cap (3) enforced; revoke is idempotent
* delete soft-deletes + revokes every live token
* admin mutations land audit rows
"""
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.identity.consumers.models import (
    IdentityConsumer,
    IdentityConsumerToken,
)
from app.identity.consumers.token_service import SVC_TOKEN_PREFIX
from app.identity.person_models import IdentityPerson

pytestmark = pytest.mark.asyncio

BASE = "/api/v1/identity/consumers"


async def _owner(
    db: AsyncSession, *, status: str = "active", employee_no: str | None = None,
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"owner-{uuid.uuid4().hex[:8]}@pingcap.com",
        display_name="Owner",
        employee_no=employee_no,
        employment_status=status,
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:10]}",
    )
    db.add(p)
    await db.flush()
    return p


def _create_body(owner_id, **overrides) -> dict:
    body = {
        "name": "Huilianyi expense",
        "slug": f"huilianyi-{uuid.uuid4().hex[:6]}",
        "description": "expense platform pull",
        "owner_person_id": str(owner_id),
        "scopes": ["directory.persons:read"],
        "filter_config": {},
    }
    body.update(overrides)
    return body


async def test_employee_cannot_touch_governance(client, employee_headers, db_session):
    owner = await _owner(db_session)
    await db_session.commit()
    r = await client.get(BASE, headers=employee_headers)
    assert r.status_code == 403
    r = await client.post(
        BASE, headers=employee_headers, json=_create_body(owner.id),
    )
    assert r.status_code == 403


async def test_create_list_detail_roundtrip(client, admin_headers, db_session):
    owner = await _owner(db_session)
    await db_session.commit()
    body = _create_body(owner.id)
    r = await client.post(BASE, headers=admin_headers, json=body)
    assert r.status_code == 200, r.text
    created = r.json()["data"]
    assert created["slug"] == body["slug"]
    assert created["status"] == "active"
    assert created["scopes"] == ["directory.persons:read"]
    # No explicit granted_fields → full persons bundle default, with the
    # always-on keys present.
    assert "work_email" in created["granted_fields"]
    assert "id" in created["granted_fields"]
    assert "mobile_phone" not in created["granted_fields"]  # pii not granted
    assert created["owner_work_email"] == owner.work_email
    assert created["active_token_count"] == 0

    r = await client.get(BASE, headers=admin_headers)
    assert r.status_code == 200
    slugs = [c["slug"] for c in r.json()["data"]]
    assert body["slug"] in slugs

    r = await client.get(f"{BASE}/{created['id']}", headers=admin_headers)
    assert r.status_code == 200
    detail = r.json()["data"]
    assert detail["tokens"] == []
    assert detail["requests_30d"] == 0


async def test_create_validations(client, admin_headers, db_session):
    owner = await _owner(db_session)
    left = await _owner(db_session, status="left")
    await db_session.commit()

    # unknown scope
    r = await client.post(BASE, headers=admin_headers, json=_create_body(
        owner.id, scopes=["directory.persons:read", "nope:read"],
    ))
    assert r.status_code == 422

    # pii without base scope
    r = await client.post(BASE, headers=admin_headers, json=_create_body(
        owner.id, scopes=["directory.pii:read"],
    ))
    assert r.status_code == 422

    # bad filter_config shape
    r = await client.post(BASE, headers=admin_headers, json=_create_body(
        owner.id, filter_config={"mode": "xor"},
    ))
    assert r.status_code == 422

    # non-active owner
    r = await client.post(BASE, headers=admin_headers, json=_create_body(left.id))
    assert r.status_code == 422

    # bad slug shape
    r = await client.post(BASE, headers=admin_headers, json=_create_body(
        owner.id, slug="Bad_Slug!",
    ))
    assert r.status_code == 422

    # granted_field whose bundle isn't granted (pii field, persons-only scope)
    r = await client.post(BASE, headers=admin_headers, json=_create_body(
        owner.id, granted_fields=["work_email", "mobile_phone"],
    ))
    assert r.status_code == 422

    # unknown field key
    r = await client.post(BASE, headers=admin_headers, json=_create_body(
        owner.id, granted_fields=["work_email", "not_a_field"],
    ))
    assert r.status_code == 422


async def test_field_level_grant_create_and_update(client, admin_headers, db_session):
    owner = await _owner(db_session)
    await db_session.commit()

    # Create with an explicit narrowed grant.
    r = await client.post(BASE, headers=admin_headers, json=_create_body(
        owner.id, granted_fields=["work_email", "display_name"],
    ))
    assert r.status_code == 200, r.text
    created = r.json()["data"]
    # Narrowed set + always-on (id, updated_at) forced in.
    assert set(created["granted_fields"]) == {
        "id", "updated_at", "work_email", "display_name",
    }

    # Widen scopes to pii AND grant a pii field via PATCH.
    r = await client.patch(
        f"{BASE}/{created['id']}", headers=admin_headers, json={
            "scopes": ["directory.persons:read", "directory.pii:read"],
            "granted_fields": ["work_email", "mobile_phone"],
        },
    )
    assert r.status_code == 200, r.text
    updated = r.json()["data"]
    assert "mobile_phone" in updated["granted_fields"]

    # Dropping the pii scope without sending fields reconciles the grant
    # (orphaned mobile_phone removed automatically).
    r = await client.patch(
        f"{BASE}/{created['id']}", headers=admin_headers, json={
            "scopes": ["directory.persons:read"],
        },
    )
    assert r.status_code == 200, r.text
    assert "mobile_phone" not in r.json()["data"]["granted_fields"]


async def test_slug_conflict_includes_soft_deleted(client, admin_headers, db_session):
    owner = await _owner(db_session)
    await db_session.commit()
    body = _create_body(owner.id)
    r = await client.post(BASE, headers=admin_headers, json=body)
    assert r.status_code == 200
    consumer_id = r.json()["data"]["id"]

    r = await client.delete(f"{BASE}/{consumer_id}", headers=admin_headers)
    assert r.status_code == 200

    # Same slug again → still conflict (soft-deleted slugs are reserved).
    r = await client.post(BASE, headers=admin_headers, json=body)
    assert r.status_code == 409


async def test_token_mint_revoke_lifecycle(client, admin_headers, db_session):
    owner = await _owner(db_session)
    await db_session.commit()
    r = await client.post(
        BASE, headers=admin_headers, json=_create_body(owner.id),
    )
    consumer_id = r.json()["data"]["id"]

    r = await client.post(
        f"{BASE}/{consumer_id}/tokens", headers=admin_headers,
        json={"name": "initial", "expires_in_days": 90},
    )
    assert r.status_code == 200, r.text
    minted = r.json()["data"]
    assert minted["plaintext"].startswith(SVC_TOKEN_PREFIX)
    assert minted["token"]["is_active"] is True
    token_id = minted["token"]["id"]

    # plaintext never appears in detail
    r = await client.get(f"{BASE}/{consumer_id}", headers=admin_headers)
    detail = r.json()["data"]
    assert len(detail["tokens"]) == 1
    assert "plaintext" not in detail["tokens"][0]
    assert detail["tokens"][0]["token_prefix"] == minted["token"]["token_prefix"]

    # invalid expiry preset
    r = await client.post(
        f"{BASE}/{consumer_id}/tokens", headers=admin_headers,
        json={"name": "weird", "expires_in_days": 17},
    )
    assert r.status_code == 422

    # cap: 3 active tokens max
    for i in range(2):
        r = await client.post(
            f"{BASE}/{consumer_id}/tokens", headers=admin_headers,
            json={"name": f"extra-{i}", "expires_in_days": None},
        )
        assert r.status_code == 200
    r = await client.post(
        f"{BASE}/{consumer_id}/tokens", headers=admin_headers,
        json={"name": "over-cap", "expires_in_days": None},
    )
    assert r.status_code == 409

    # revoke + idempotent re-revoke
    r = await client.post(
        f"{BASE}/{consumer_id}/tokens/{token_id}/revoke", headers=admin_headers,
    )
    assert r.status_code == 200
    assert r.json()["data"]["is_active"] is False
    r = await client.post(
        f"{BASE}/{consumer_id}/tokens/{token_id}/revoke", headers=admin_headers,
    )
    assert r.status_code == 200


async def test_patch_kill_switch_and_immutable_slug(
    client, admin_headers, db_session,
):
    owner = await _owner(db_session)
    await db_session.commit()
    body = _create_body(owner.id)
    r = await client.post(BASE, headers=admin_headers, json=body)
    consumer_id = r.json()["data"]["id"]

    r = await client.patch(
        f"{BASE}/{consumer_id}", headers=admin_headers,
        json={"status": "disabled", "slug": "evil-rename"},
    )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["status"] == "disabled"
    # slug field in the body is simply not part of the schema → ignored
    assert data["slug"] == body["slug"]

    r = await client.patch(
        f"{BASE}/{consumer_id}", headers=admin_headers,
        json={"status": "active", "rate_limit_per_minute": 60},
    )
    assert r.status_code == 200
    assert r.json()["data"]["rate_limit_per_minute"] == 60

    # Explicit null resets back to the platform default. ``None`` had to
    # be distinguished from "field omitted" via model_fields_set —
    # without it a custom limit was permanently sticky (review fix).
    r = await client.patch(
        f"{BASE}/{consumer_id}", headers=admin_headers,
        json={"rate_limit_per_minute": None},
    )
    assert r.status_code == 200
    assert r.json()["data"]["rate_limit_per_minute"] is None

    # Omitting the field leaves the (now-null) value untouched — and a
    # subsequent set still works, proving omit ≠ null.
    r = await client.patch(
        f"{BASE}/{consumer_id}", headers=admin_headers,
        json={"description": "unrelated change"},
    )
    assert r.json()["data"]["rate_limit_per_minute"] is None
    r = await client.patch(
        f"{BASE}/{consumer_id}", headers=admin_headers,
        json={"rate_limit_per_minute": 120},
    )
    assert r.json()["data"]["rate_limit_per_minute"] == 120


async def test_delete_revokes_all_tokens(client, admin_headers, db_session):
    owner = await _owner(db_session)
    await db_session.commit()
    r = await client.post(BASE, headers=admin_headers, json=_create_body(owner.id))
    consumer_id = r.json()["data"]["id"]
    for i in range(2):
        await client.post(
            f"{BASE}/{consumer_id}/tokens", headers=admin_headers,
            json={"name": f"t{i}", "expires_in_days": None},
        )

    r = await client.delete(f"{BASE}/{consumer_id}", headers=admin_headers)
    assert r.status_code == 200
    assert r.json()["data"]["tokens_revoked"] == 2

    db_session.expire_all()
    row = await db_session.get(IdentityConsumer, uuid.UUID(consumer_id))
    assert row.deleted_at is not None
    tokens = (await db_session.execute(
        select(IdentityConsumerToken).where(
            IdentityConsumerToken.consumer_id == uuid.UUID(consumer_id),
        )
    )).scalars().all()
    assert all(t.revoked_at is not None for t in tokens)

    # detail of a deleted consumer → 404
    r = await client.get(f"{BASE}/{consumer_id}", headers=admin_headers)
    assert r.status_code == 404


async def test_admin_mutations_write_audit_rows(client, admin_headers, db_session):
    owner = await _owner(db_session)
    await db_session.commit()
    r = await client.post(BASE, headers=admin_headers, json=_create_body(owner.id))
    consumer_id = r.json()["data"]["id"]
    await client.post(
        f"{BASE}/{consumer_id}/tokens", headers=admin_headers,
        json={"name": "initial", "expires_in_days": None},
    )

    db_session.expire_all()
    actions = [a for (a,) in (await db_session.execute(
        select(AuditLog.action).where(AuditLog.resource_id == consumer_id)
    )).all()]
    assert "identity_consumer.create" in actions
    assert "identity_consumer.token_mint" in actions


async def test_admin_status_change_clears_recert_lapsed_flag(client, admin_headers, db_session):
    """Review regression: an admin status change clears recert_lapsed_at so
    a later recert `keep` can't re-enable against the admin kill switch."""
    from datetime import UTC, datetime

    from app.identity.consumers.models import IdentityConsumer

    owner = await _owner(db_session)
    await db_session.commit()
    r = await client.post(BASE, headers=admin_headers, json=_create_body(owner.id))
    cid = r.json()["data"]["id"]
    # Simulate a prior lapse auto-disable.
    consumer = await db_session.get(IdentityConsumer, uuid.UUID(cid))
    consumer.status = "disabled"
    consumer.recert_lapsed_at = datetime.now(UTC)
    await db_session.commit()

    # Admin re-enables via PATCH → recert_lapsed_at must clear.
    r = await client.patch(f"{BASE}/{cid}", headers=admin_headers, json={"status": "active"})
    assert r.status_code == 200, r.text
    db_session.expire_all()
    refreshed = await db_session.get(IdentityConsumer, uuid.UUID(cid))
    assert refreshed.status == "active"
    assert refreshed.recert_lapsed_at is None


async def test_scope_catalog(client, admin_headers):
    r = await client.get(f"{BASE}/scopes", headers=admin_headers)
    assert r.status_code == 200
    data = r.json()["data"]
    assert "directory.persons:read" in data["scopes"]
    assert "directory.pii:read" in data["scopes"]
    assert 90 in data["token_expiry_days"]
    # Field catalog with sensitivity tiers for the applicant/admin UI.
    fields = {f["key"]: f for f in data["fields"]}
    assert fields["mobile_phone"]["sensitivity"] == "pii"
    assert fields["mobile_phone"]["bundle"] == "directory.pii:read"
    assert fields["display_name"]["sensitivity"] == "public"
    assert fields["id"]["always"] is True


async def test_owner_leaver_indicators_populated(client, admin_headers, db_session):
    """List + detail expose owner_employee_no / owner_employment_status so the
    frontend renders the owner cell through <PersonTag> with the 已离职
    treatment once an owner offboards. The create endpoint rejects a left
    owner, so we model the "owner left after creation" state by inserting
    the consumer row directly against a left owner."""
    from app.identity.consumers.models import IdentityConsumer

    owner = await _owner(db_session, status="left", employee_no="P0777")
    cid = uuid.uuid4()
    db_session.add(IdentityConsumer(
        id=cid,
        name="Huilianyi expense",
        slug=f"huilianyi-{uuid.uuid4().hex[:6]}",
        owner_person_id=owner.id,
        scopes=["directory.persons:read"],
        granted_fields=["id", "work_email"],
        filter_config={},
        status="active",
    ))
    await db_session.commit()

    listed = (await client.get(BASE, headers=admin_headers)).json()["data"]
    row = next(c for c in listed if c["id"] == str(cid))
    assert row["owner_employee_no"] == "P0777"
    assert row["owner_employment_status"] == "left"

    detail = (await client.get(f"{BASE}/{cid}", headers=admin_headers)).json()["data"]
    assert detail["owner_employee_no"] == "P0777"
    assert detail["owner_employment_status"] == "left"