"""External API governance — /api/v1/api-clients CRUD + token lifecycle.

Pins:
* RBAC: employee → 403 on every governance endpoint
* create validates unknown scope, write-without-read, non-active owner,
  duplicate slug (incl. soft-deleted)
* scope catalog lists the cloud scopes
* token mint returns plaintext once (helix_api_ prefix); list shows
  prefix only; active-token cap (3); revoke idempotent
* delete soft-deletes + revokes live tokens; restore undoes it (deleted rows
  stay listable via include_deleted, tokens stay revoked, departed owner OK)
* mutations land audit rows (resource_type=api_client)
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import select
from sqlalchemy import update as sa_update
from sqlalchemy.ext.asyncio import AsyncSession

from app.api_clients.models import ApiClient
from app.api_clients.tokens import API_TOKEN_PREFIX
from app.audit.models import AuditLog
from app.identity.person_models import (
    IdentityOrgMembership,
    IdentityOrgNode,
    IdentityPerson,
)

pytestmark = pytest.mark.asyncio

BASE = "/api/v1/api-clients"


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.commit()
    return p


def _body(owner_id, **overrides) -> dict:
    body = {
        "name": "BI Platform",
        "slug": f"bi-{uuid.uuid4().hex[:6]}",
        "description": "reporting pull",
        "owner_person_id": str(owner_id),
        "scopes": ["cloud.subscriptions:read"],
    }
    body.update(overrides)
    return body


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


async def test_scope_catalog(client, admin_headers):
    r = await client.get(f"{BASE}/scopes", headers=admin_headers)
    assert r.status_code == 200, r.text
    keys = {s["key"] for s in r.json()["data"]["scopes"]}
    assert "cloud.subscriptions:read" in keys
    assert "cloud.subscriptions:write" in keys


async def test_create_list_detail_roundtrip(client, admin_headers, db_session):
    owner = await _owner(db_session)
    r = await client.post(BASE, headers=admin_headers, json=_body(
        owner.id, scopes=["cloud.subscriptions:read", "cloud.subscriptions:write"]))
    assert r.status_code == 200, r.text
    created = r.json()["data"]
    assert created["owner_work_email"] == owner.work_email
    assert set(created["scopes"]) == {"cloud.subscriptions:read", "cloud.subscriptions:write"}

    r = await client.get(BASE, headers=admin_headers)
    assert any(c["id"] == created["id"] for c in r.json()["data"])

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


async def test_person_visibility_roundtrip_and_audit(
    client, admin_headers, db_session,
):
    """The people-population ceiling survives create → detail → PATCH, a
    malformed filter is refused, and BOTH directions of a change are
    audited by shape — a widening (restricted → everyone) is the most
    consequential edit on this row and must never be inferable only from
    silence."""
    owner = await _owner(db_session)
    r = await client.post(BASE, headers=admin_headers, json=_body(
        owner.id,
        scopes=["people.github:read"],
        person_visibility={"countries": ["CN", "US"], "include_subtree": True},
    ))
    assert r.status_code == 200, r.text
    created = r.json()["data"]
    # Canonical form: sorted, empty/no-op dimensions dropped.
    assert created["person_visibility"] == {"countries": ["CN", "US"]}

    detail = await client.get(f"{BASE}/{created['id']}", headers=admin_headers)
    assert detail.json()["data"]["person_visibility"] == {"countries": ["CN", "US"]}

    # An unknown key is refused rather than silently ignored (which would
    # normalize to "no constraints" = the whole roster).
    bad = await client.patch(
        f"{BASE}/{created['id']}", headers=admin_headers,
        json={"person_visibility": {"country": ["CN"]}},
    )
    assert bad.status_code == 422, bad.text

    # Omitted field leaves the ceiling untouched...
    untouched = await client.patch(
        f"{BASE}/{created['id']}", headers=admin_headers, json={"name": "Renamed"},
    )
    assert untouched.json()["data"]["person_visibility"] == {"countries": ["CN", "US"]}

    # ...explicit null clears it (widening to everyone).
    widened = await client.patch(
        f"{BASE}/{created['id']}", headers=admin_headers,
        json={"person_visibility": None},
    )
    assert widened.status_code == 200, widened.text
    assert widened.json()["data"]["person_visibility"] is None

    rows = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.resource_type == "api_client",
            AuditLog.resource_id == created["id"],
        )
    )).scalars().all()
    created_row = next(r for r in rows if r.action == "api_client.create")
    assert created_row.detail["person_visibility"] == "restricted"
    assert created_row.detail["countries"] == ["CN", "US"]
    widen_row = next(
        r for r in rows
        if r.action == "api_client.update"
        and "person_visibility" in (r.detail.get("changes") or {})
    )
    change = widen_row.detail["changes"]["person_visibility"]
    assert change["from"]["person_visibility"] == "restricted"
    assert change["to"]["person_visibility"] == "unrestricted"


async def test_person_visibility_targets_are_existence_checked(
    client, admin_headers, db_session,
):
    """A ceiling naming a department or person that does not exist is not
    an error at read time — it is a filter that silently matches nobody,
    which the admin then debugs for a day. Caught at save time instead,
    the same way cloud visibility validates its subscription ids.

    The most likely mistake is identifier KIND: ``departments`` holds HRIS
    org-node source ids, not Helix row UUIDs.
    """
    owner = await _owner(db_session)

    ghost_dept = await client.post(BASE, headers=admin_headers, json=_body(
        owner.id, scopes=["people.github:read"],
        person_visibility={"departments": [str(uuid.uuid4())]},
    ))
    assert ghost_dept.status_code == 422, ghost_dept.text
    assert "org node" in str(ghost_dept.json()["errors"])

    ghost_person = await client.post(BASE, headers=admin_headers, json=_body(
        owner.id, scopes=["people.github:read"],
        person_visibility={"person_ids": [str(uuid.uuid4())]},
    ))
    assert ghost_person.status_code == 422, ghost_person.text

    # A real org node source id is accepted.
    node_source_id = f"node-{uuid.uuid4().hex[:8]}"
    db_session.add(IdentityOrgNode(
        id=uuid.uuid4(),
        org_node_source_id=node_source_id,
        org_node_name="Real Dept",
        org_path="PingCAP / Real Dept",
    ))
    await db_session.commit()
    ok = await client.post(BASE, headers=admin_headers, json=_body(
        owner.id, scopes=["people.github:read"],
        person_visibility={"departments": [node_source_id]},
    ))
    assert ok.status_code == 200, ok.text
    assert ok.json()["data"]["person_visibility"]["departments"] == [node_source_id]


async def test_malformed_person_visibility_is_visible_and_repairable(
    client, admin_headers, db_session,
):
    """A stored ceiling that isn't a JSON object makes the data plane
    resolve to nobody. Two things must therefore hold on the admin side:
    the console must not render that client as *unrestricted*, and the
    repair the integrator is told to ask for — clearing the ceiling —
    must actually clear it.

    Both failed at first: the response coerced the value to ``None``
    (reading as unrestricted), and the PATCH compared against that same
    coerced ``None``, so an explicit-null PATCH was a silent no-op that
    left the client reading nobody forever.
    """
    owner = await _owner(db_session)
    created = (await client.post(BASE, headers=admin_headers, json=_body(
        owner.id, scopes=["people.github:read"],
    ))).json()["data"]

    await db_session.execute(
        sa_update(ApiClient)
        .where(ApiClient.id == uuid.UUID(created["id"]))
        .values(person_visibility='"not-an-object"'),
    )
    await db_session.commit()

    detail = (await client.get(
        f"{BASE}/{created['id']}", headers=admin_headers,
    )).json()["data"]
    assert detail["person_visibility_misconfigured"] is True

    repaired = await client.patch(
        f"{BASE}/{created['id']}", headers=admin_headers,
        json={"person_visibility": None},
    )
    assert repaired.status_code == 200, repaired.text
    assert repaired.json()["data"]["person_visibility"] is None
    assert repaired.json()["data"]["person_visibility_misconfigured"] is False

    # The repair takes the client from "reads nobody" to "reads the whole
    # roster" — the largest widening on this surface. It must be legible
    # in the audit trail, not recorded as unrestricted → unrestricted.
    repair_row = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.resource_type == "api_client",
            AuditLog.resource_id == created["id"],
            AuditLog.action == "api_client.update",
        )
    )).scalars().all()
    change = next(
        r.detail["changes"]["person_visibility"] for r in repair_row
        if "person_visibility" in (r.detail.get("changes") or {})
    )
    assert change["from"]["person_visibility"] == "misconfigured"
    assert change["to"]["person_visibility"] == "unrestricted"

    row = await db_session.get(ApiClient, uuid.UUID(created["id"]))
    await db_session.refresh(row)
    assert row.person_visibility is None


async def test_non_uuid_person_id_is_a_422_not_a_500(
    client, admin_headers, db_session,
):
    """A malformed person id is a caller mistake, so it belongs in the
    same actionable 422 as an id that doesn't exist. Passing it through
    to the query instead makes the UUID TypeDecorator raise on bind,
    which SQLAlchemy wraps and the global handler turns into a 500."""
    owner = await _owner(db_session)
    r = await client.post(BASE, headers=admin_headers, json=_body(
        owner.id, scopes=["people.github:read"],
        person_visibility={"person_ids": ["alice"]},
    ))
    assert r.status_code == 422, r.text
    assert "alice" in str(r.json()["errors"])


async def test_person_visibility_departments_accept_membership_only_ids(
    client, admin_headers, db_session,
):
    """The runtime filter matches org-node source ids on MEMBERSHIPS, and
    seed ids survive subtree expansion whether or not a node row exists.
    So a source id that lives only on memberships yields a ceiling that
    demonstrably works — validating against the node table alone would
    reject a working grant (and would do so on every later PATCH, even
    one that narrows it)."""
    owner = await _owner(db_session)
    person = await _owner(db_session)
    orphan_source_id = f"node-{uuid.uuid4().hex[:8]}"
    db_session.add(IdentityOrgMembership(
        id=uuid.uuid4(),
        person_id=str(person.id),
        org_node_source_id=orphan_source_id,
        org_node_name="Restructured Team",
        org_path="PingCAP / Restructured Team",
        is_primary=True,
    ))
    await db_session.commit()

    r = await client.post(BASE, headers=admin_headers, json=_body(
        owner.id, scopes=["people.github:read"],
        person_visibility={"departments": [orphan_source_id]},
    ))
    assert r.status_code == 200, r.text

    # …and a genuinely unknown id is still refused on the PATCH path,
    # which the first version of this check never covered.
    bad = await client.patch(
        f"{BASE}/{r.json()['data']['id']}", headers=admin_headers,
        json={"person_visibility": {"departments": [f"ghost-{uuid.uuid4().hex[:6]}"]}},
    )
    assert bad.status_code == 422, bad.text

    # A SECONDARY-only membership is refused too: the runtime resolver
    # reads primary memberships only, so accepting it would bless a
    # ceiling that still matches nobody. (Without the is_primary
    # predicate on the membership half of the check, this would pass.)
    secondary_source_id = f"node-{uuid.uuid4().hex[:8]}"
    db_session.add(IdentityOrgMembership(
        id=uuid.uuid4(),
        person_id=str(person.id),
        org_node_source_id=secondary_source_id,
        org_node_name="Secondary Team",
        org_path="PingCAP / Secondary Team",
        is_primary=False,
    ))
    await db_session.commit()
    secondary = await client.post(BASE, headers=admin_headers, json=_body(
        owner.id, scopes=["people.github:read"],
        person_visibility={"departments": [secondary_source_id]},
    ))
    assert secondary.status_code == 422, secondary.text


async def test_create_validations(client, admin_headers, db_session):
    owner = await _owner(db_session)
    # unknown scope
    r = await client.post(BASE, headers=admin_headers, json=_body(owner.id, scopes=["bogus:read"]))
    assert r.status_code == 422
    # write without read
    r = await client.post(BASE, headers=admin_headers,
                          json=_body(owner.id, scopes=["cloud.subscriptions:write"]))
    assert r.status_code == 422
    # non-active owner
    left = await _owner(db_session, status="left")
    r = await client.post(BASE, headers=admin_headers, json=_body(left.id))
    assert r.status_code == 422
    # duplicate slug
    body = _body(owner.id)
    assert (await client.post(BASE, headers=admin_headers, json=body)).status_code == 200
    assert (await client.post(BASE, headers=admin_headers, json=body)).status_code == 409


async def test_deleted_slug_conflict_says_it_is_deleted(
    client, admin_headers, db_session,
):
    """A slug held by a DELETED client must say so, and name the holder.

    The module docstring has always claimed the duplicate check covers
    soft-deleted rows, but only the live case was pinned — so nobody noticed
    that the reader gets "already exists" for a client the list page hides by
    default (``include_deleted=false``). That combination reads as a platform
    bug: prod 2026-08-12, a client was created at 06:20, deleted at 06:25, and
    re-registering the same slug later failed with a message naming nothing the
    admin could find anywhere. The 409 must carry the deleted holder's id so
    the caller can restore it instead of being stuck.
    """
    owner = await _owner(db_session)
    body = _body(owner.id)
    cid = (
        await client.post(BASE, headers=admin_headers, json=body)
    ).json()["data"]["id"]
    assert (await client.delete(f"{BASE}/{cid}", headers=admin_headers)).status_code == 200

    r = await client.post(BASE, headers=admin_headers, json=body)
    assert r.status_code == 409
    detail = str(r.json()["errors"])
    assert body["slug"] in detail
    # Names the deleted state, the holder, and the way out — not merely
    # "already exists", and not a dead end.
    assert "DELETED" in detail
    assert cid in detail
    assert "restore" in detail


async def test_deleted_client_is_visible_and_restorable(
    client, admin_headers, db_session,
):
    """Soft delete must not be a one-way trip that burns the slug.

    The slug unique constraint counts deleted rows, so "invisible deleted
    client" + "slug can never be reused" was a deadlock with no exit from
    either the console or the API. Restore is the exit: same row, same id,
    same slug — audit trails stay unambiguous, which is exactly why handing
    the slug to a *different* client stays forbidden.
    """
    owner = await _owner(db_session)
    body = _body(owner.id)
    cid = (await client.post(BASE, headers=admin_headers, json=body)).json()["data"]["id"]
    assert (await client.post(
        f"{BASE}/{cid}/tokens", headers=admin_headers,
        json={"name": "t", "expires_in_days": 90},
    )).status_code == 200
    assert (await client.delete(f"{BASE}/{cid}", headers=admin_headers)).status_code == 200

    # Hidden by default, reachable (with the deleted_at marker) on demand.
    default_list = (await client.get(BASE, headers=admin_headers)).json()["data"]
    assert all(c["id"] != cid for c in default_list)
    with_deleted = (await client.get(
        f"{BASE}?include_deleted=true", headers=admin_headers,
    )).json()["data"]
    row = next(c for c in with_deleted if c["id"] == cid)
    assert row["deleted_at"] is not None

    r = await client.post(f"{BASE}/{cid}/restore", headers=admin_headers)
    assert r.status_code == 200, r.text
    assert r.json()["data"]["deleted_at"] is None
    # Tokens revoked by the delete stay revoked — restore grants no access.
    assert r.json()["data"]["active_token_count"] == 0
    detail = (await client.get(f"{BASE}/{cid}", headers=admin_headers)).json()["data"]
    assert all(tok["revoked_at"] is not None for tok in detail["tokens"])
    # Back in the default list, and idempotent.
    assert any(
        c["id"] == cid for c in (await client.get(BASE, headers=admin_headers)).json()["data"]
    )
    assert (await client.post(f"{BASE}/{cid}/restore", headers=admin_headers)).status_code == 200

    audit = (await db_session.execute(
        select(AuditLog).where(AuditLog.action == "api_client.restore")
    )).scalars().all()
    assert [a.resource_id for a in audit] == [cid]


async def test_restore_survives_a_departed_owner(client, admin_headers, db_session):
    """Restore must not re-run the active-owner check.

    Registration demands an active owner, but a client outlives its owner —
    if the owner leaves after the delete, an owner-validating restore would
    put the slug right back into the deadlock it was built to break.
    """
    owner = await _owner(db_session)
    cid = (
        await client.post(BASE, headers=admin_headers, json=_body(owner.id))
    ).json()["data"]["id"]
    assert (await client.delete(f"{BASE}/{cid}", headers=admin_headers)).status_code == 200

    # Core UPDATE, not an ORM attribute set: the LeaverBlocked guard refuses a
    # direct flip to 'left' and this test is about the API-client side.
    await db_session.execute(
        sa_update(IdentityPerson)
        .where(IdentityPerson.id == owner.id)
        .values(employment_status="left")
    )
    await db_session.commit()

    r = await client.post(f"{BASE}/{cid}/restore", headers=admin_headers)
    assert r.status_code == 200, r.text
    assert r.json()["data"]["owner_employment_status"] == "left"


async def test_restore_requires_admin(client, employee_headers):
    r = await client.post(f"{BASE}/{uuid.uuid4()}/restore", headers=employee_headers)
    assert r.status_code == 403


async def test_token_mint_revoke_and_audit(client, admin_headers, db_session):
    owner = await _owner(db_session)
    created = (await client.post(BASE, headers=admin_headers, json=_body(owner.id))).json()["data"]
    cid = created["id"]

    r = await client.post(f"{BASE}/{cid}/tokens", headers=admin_headers,
                          json={"name": "initial", "expires_in_days": 90})
    assert r.status_code == 200, r.text
    minted = r.json()["data"]
    plaintext = minted["plaintext"]
    assert plaintext.startswith(API_TOKEN_PREFIX)
    # Detail shows prefix only, never the plaintext.
    detail = (await client.get(f"{BASE}/{cid}", headers=admin_headers)).json()["data"]
    assert detail["tokens"][0]["token_prefix"] == minted["token"]["token_prefix"]
    assert plaintext not in str(detail)

    tid = minted["token"]["id"]
    r1 = await client.post(f"{BASE}/{cid}/tokens/{tid}/revoke", headers=admin_headers)
    r2 = await client.post(f"{BASE}/{cid}/tokens/{tid}/revoke", headers=admin_headers)
    assert r1.status_code == 200 and r2.status_code == 200  # idempotent

    rows = (await db_session.execute(
        select(AuditLog).where(AuditLog.resource_type == "api_client")
    )).scalars().all()
    actions = {r.action for r in rows}
    assert "api_client.create" in actions
    assert "api_client.token_mint" in actions
    assert "api_client.token_revoke" in actions


async def test_delete_soft_deletes_and_revokes_tokens(client, admin_headers, db_session):
    owner = await _owner(db_session)
    cid = (await client.post(BASE, headers=admin_headers, json=_body(owner.id))).json()["data"]["id"]
    await client.post(f"{BASE}/{cid}/tokens", headers=admin_headers,
                      json={"name": "t", "expires_in_days": None})
    r = await client.delete(f"{BASE}/{cid}", headers=admin_headers)
    assert r.status_code == 200
    assert r.json()["data"]["tokens_revoked"] == 1
    # Soft-deleted → 404 on detail.
    assert (await client.get(f"{BASE}/{cid}", headers=admin_headers)).status_code == 404


async def test_active_token_cap(client, admin_headers, db_session):
    owner = await _owner(db_session)
    cid = (await client.post(BASE, headers=admin_headers, json=_body(owner.id))).json()["data"]["id"]
    for i in range(3):
        r = await client.post(f"{BASE}/{cid}/tokens", headers=admin_headers,
                              json={"name": f"t{i}", "expires_in_days": None})
        assert r.status_code == 200
    # 4th over the cap → 409.
    r = await client.post(f"{BASE}/{cid}/tokens", headers=admin_headers,
                          json={"name": "t4", "expires_in_days": None})
    assert r.status_code == 409


async def test_owner_leaver_indicators_populated(client, admin_headers, db_session):
    """List + detail surface owner_employee_no / owner_employment_status so
    the frontend can render the owner cell via <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 client row directly against a left owner."""
    from app.api_clients.models import ApiClient

    owner = await _owner(db_session, status="left", employee_no="P0343")
    cid = uuid.uuid4()
    db_session.add(ApiClient(
        id=cid,
        name="BI Platform",
        slug=f"bi-{uuid.uuid4().hex[:6]}",
        owner_person_id=owner.id,
        scopes=["cloud.subscriptions:read"],
        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"] == "P0343"
    assert row["owner_employment_status"] == "left"

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