"""Integration tests for the admin seat-revoke endpoint and its
revoke-preview sibling.

Covers the successor-validation branches introduced alongside the
Zoom transfer + Google Drive handoff plumbing:

  - Preview endpoint reports requires_successor based on the bound
    driver (Zoom delete → True, Slack → False, no connector → False).
  - Preview suggests the seat holder's manager when they hold an
    active seat on the same software.
  - POST /revoke without a successor falls back to the manager when
    required, 400s with SUCCESSOR_REQUIRED when no valid default.
  - 400s with SUCCESSOR_INVALID / SUCCESSOR_NO_SEAT when the picked
    successor doesn't satisfy the constraints.
  - No-successor-needed provider (Slack) revokes cleanly even when
    the request body is empty.

No real HTTP: the provisioning task that would fire is inspected at
the DB level. Driver-level transfer plumbing is covered by
test_zoom_driver and test_google_workspace_ops.
"""
from __future__ import annotations

import uuid
from decimal import Decimal

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

from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.domains.provisioning.models import (
    ACTION_REVOKE,
    ProvisioningTask,
    SoftwareConnector,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


async def _person(db: AsyncSession, email: str, *, manager: uuid.UUID | None = None) -> IdentityPerson:
    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}",
        manager_person_id=manager,
    )
    db.add(p)
    await db.flush()
    return p


async def _software_plan(
    db: AsyncSession, *, policy: str = "auto_revoke",
) -> tuple[AssetSoftwareCatalog, SoftwareLicensePlan]:
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(),
        name=f"Test-{uuid.uuid4().hex[:6]}",
        asset_type="SaaS",
        status="active",
        offboarding_policy=policy,
    )
    db.add(sw)
    await db.flush()
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=sw.id, name="Licensed",
        price=Decimal("0"), billing_cycle="monthly",
        status="active", max_seats=None,
    )
    db.add(plan)
    await db.flush()
    return sw, plan


async def _seat(
    db: AsyncSession, *, plan_id: uuid.UUID, user_id: uuid.UUID, status: str = "active",
) -> SoftwareSeatAllocation:
    seat = SoftwareSeatAllocation(plan_id=plan_id, user_id=user_id, status=status)
    db.add(seat)
    await db.flush()
    return seat


async def _connector(
    db: AsyncSession, *, software_id: uuid.UUID, provider: str,
    binding_config: dict | None = None,
) -> Integration:
    i = Integration(
        id=uuid.uuid4(),
        kind="software_provisioning",
        provider=provider,
        name=f"{provider} test",
        slug=f"{provider}-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        config_public={},
    )
    db.add(i)
    await db.flush()
    db.add(SoftwareConnector(
        software_id=software_id, integration_id=i.id, is_active=True,
        binding_config=binding_config or {},
    ))
    await db.flush()
    return i


@pytest.mark.asyncio
async def test_preview_reports_requires_successor_for_zoom_delete(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    sw, plan = await _software_plan(db_session)
    await _connector(
        db_session, software_id=sw.id, provider="zoom",
        binding_config={"action_on_offboard": "delete"},
    )
    manager = await _person(db_session, "mgr-zoom@x.com")
    holder = await _person(db_session, "zoom-holder@x.com", manager=manager.id)
    await _seat(db_session, plan_id=plan.id, user_id=manager.id)  # manager has seat
    seat = await _seat(db_session, plan_id=plan.id, user_id=holder.id)
    await db_session.commit()

    resp = await client.get(
        f"/api/v1/assets/seat-allocations/{seat.id}/revoke-preview",
        headers=admin_headers,
    )
    assert resp.status_code == 200
    data = resp.json()["data"]
    assert data["requires_successor"] is True
    assert data["provider"] == "zoom"
    assert data["suggested_successor_user_id"] == str(manager.id)


@pytest.mark.asyncio
async def test_preview_reports_no_successor_for_slack(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    sw, plan = await _software_plan(db_session)
    await _connector(db_session, software_id=sw.id, provider="slack")
    holder = await _person(db_session, "slack-holder@x.com")
    seat = await _seat(db_session, plan_id=plan.id, user_id=holder.id)
    await db_session.commit()

    resp = await client.get(
        f"/api/v1/assets/seat-allocations/{seat.id}/revoke-preview",
        headers=admin_headers,
    )
    assert resp.status_code == 200
    assert resp.json()["data"]["requires_successor"] is False


@pytest.mark.asyncio
async def test_revoke_slack_no_body_succeeds(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    sw, plan = await _software_plan(db_session)
    await _connector(db_session, software_id=sw.id, provider="slack")
    holder = await _person(db_session, "slack-revoke@x.com")
    seat = await _seat(db_session, plan_id=plan.id, user_id=holder.id)
    await db_session.commit()

    resp = await client.post(
        f"/api/v1/assets/seat-allocations/{seat.id}/revoke",
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()["data"]
    assert body["status"] == "revoked"
    # Successor key absent/null on no-transfer providers.
    assert body.get("successor_user_id") is None


@pytest.mark.asyncio
async def test_revoke_endpoint_stamps_revert_seat_on_failure_flag(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    """The admin revoke endpoint must opt the queued task into the
    dispatcher's failure-revert hook (``revert_seat_on_failure=True``).
    Without this opt-in, a downstream auto_api_failed (e.g. Zoom 1107
    managed-domain refusal) would leave the seat falsely-revoked
    while the vendor still holds the paid license — the bug class
    caught 2026-05-08."""
    sw, plan = await _software_plan(db_session)
    await _connector(db_session, software_id=sw.id, provider="slack")
    holder = await _person(db_session, "revert-flag-test@x.com")
    seat = await _seat(db_session, plan_id=plan.id, user_id=holder.id)
    await db_session.commit()

    resp = await client.post(
        f"/api/v1/assets/seat-allocations/{seat.id}/revoke",
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    task_id = resp.json()["data"]["provisioning_task_id"]
    assert task_id is not None

    task = await db_session.get(ProvisioningTask, uuid.UUID(task_id))
    assert task is not None
    assert (task.context_data or {}).get("revert_seat_on_failure") is True, (
        "admin revoke must stamp revert_seat_on_failure=True on the task "
        f"context_data; got {task.context_data!r}"
    )


@pytest.mark.asyncio
async def test_revoke_zoom_delete_requires_successor(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    """No successor in body, target has no manager → 400."""
    sw, plan = await _software_plan(db_session)
    await _connector(
        db_session, software_id=sw.id, provider="zoom",
        binding_config={"action_on_offboard": "delete"},
    )
    # Holder with no manager configured so default-pick fails.
    holder = await _person(db_session, "zoom-nomgr@x.com")
    seat = await _seat(db_session, plan_id=plan.id, user_id=holder.id)
    await db_session.commit()

    resp = await client.post(
        f"/api/v1/assets/seat-allocations/{seat.id}/revoke",
        headers=admin_headers, json={},
    )
    assert resp.status_code == 400
    assert resp.json()["errors"][0]["code"] == "SUCCESSOR_REQUIRED"


@pytest.mark.asyncio
async def test_revoke_zoom_delete_rejects_successor_without_seat(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    sw, plan = await _software_plan(db_session)
    await _connector(
        db_session, software_id=sw.id, provider="zoom",
        binding_config={"action_on_offboard": "delete"},
    )
    holder = await _person(db_session, "zoom-holder2@x.com")
    # Successor has NO seat on this software.
    successor = await _person(db_session, "zoom-no-seat@x.com")
    seat = await _seat(db_session, plan_id=plan.id, user_id=holder.id)
    await db_session.commit()

    resp = await client.post(
        f"/api/v1/assets/seat-allocations/{seat.id}/revoke",
        headers=admin_headers,
        json={"successor_user_id": str(successor.id)},
    )
    assert resp.status_code == 400
    assert resp.json()["errors"][0]["code"] == "SUCCESSOR_NO_SEAT"


@pytest.mark.asyncio
async def test_revoke_zoom_delete_defaults_to_manager(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    sw, plan = await _software_plan(db_session)
    await _connector(
        db_session, software_id=sw.id, provider="zoom",
        binding_config={"action_on_offboard": "delete"},
    )
    manager = await _person(db_session, "zoom-mgr@x.com")
    holder = await _person(db_session, "zoom-holder3@x.com", manager=manager.id)
    await _seat(db_session, plan_id=plan.id, user_id=manager.id)
    seat = await _seat(db_session, plan_id=plan.id, user_id=holder.id)
    await db_session.commit()

    resp = await client.post(
        f"/api/v1/assets/seat-allocations/{seat.id}/revoke",
        headers=admin_headers, json={},
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()["data"]
    assert body["successor_user_id"] == str(manager.id)
    # A ProvisioningTask with successor data must have been queued.
    task = (await db_session.execute(
        select(ProvisioningTask).where(
            ProvisioningTask.target_id == seat.id,
            ProvisioningTask.action == ACTION_REVOKE,
        )
    )).scalar_one_or_none()
    assert task is not None
    assert task.context_data["successor_user_id"] == str(manager.id)
    assert task.context_data["successor_email"] == manager.work_email


@pytest.mark.asyncio
async def test_revoke_rejects_self_successor(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    sw, plan = await _software_plan(db_session)
    await _connector(
        db_session, software_id=sw.id, provider="zoom",
        binding_config={"action_on_offboard": "delete"},
    )
    holder = await _person(db_session, "zoom-self@x.com")
    seat = await _seat(db_session, plan_id=plan.id, user_id=holder.id)
    await db_session.commit()

    resp = await client.post(
        f"/api/v1/assets/seat-allocations/{seat.id}/revoke",
        headers=admin_headers,
        json={"successor_user_id": str(holder.id)},
    )
    assert resp.status_code == 400
    assert resp.json()["errors"][0]["code"] == "SUCCESSOR_INVALID"


# ---------------------------------------------------------------------------
# Cross-department exfiltration guard (round 1 pre-launch hardening)
# ---------------------------------------------------------------------------
#
# Zoom ``action_on_offboard=delete`` and Google ``user_suspend`` move
# meetings / Drive content to the successor at revoke time. Without
# the cross-department guard, an asset_admin (or a stolen asset_admin
# token) could route a CEO's meetings to an attacker employee in a
# completely unrelated BU. The guard requires:
#   1. Same department → any admin (asset / global) may revoke.
#   2. Different department → request body MUST set
#      ``confirm_cross_department=True`` AND the caller MUST be a
#      global_admin (asset_admin is restricted to same-dept).
#
# These tests exercise (2). Same-dept paths are already covered by
# ``test_revoke_zoom_delete_defaults_to_manager`` above (manager and
# holder share no dept membership in that fixture, but neither has a
# membership row at all → both depts are NULL → ``cross_department``
# resolves to False, allowing the revoke).


async def _membership(
    db: AsyncSession, *, person_id: uuid.UUID, dept_name: str,
) -> None:
    from app.identity.person_models import IdentityOrgMembership
    db.add(IdentityOrgMembership(
        person_id=person_id,
        source_system="successfactors",
        org_node_source_id=f"DP-{dept_name.replace(' ', '-').upper()}",
        org_node_type="department",
        org_node_code=f"DP-{dept_name.replace(' ', '-').upper()}",
        org_node_name=dept_name,
        parent_org_node_source_id="DP-ROOT",
        org_path=f"PingCAP / {dept_name}",
        is_primary=True,
    ))
    await db.flush()


@pytest.mark.asyncio
async def test_revoke_cross_department_requires_confirm_flag(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    """Cross-dept successor without confirm_cross_department=True → 400."""
    sw, plan = await _software_plan(db_session)
    await _connector(
        db_session, software_id=sw.id, provider="zoom",
        binding_config={"action_on_offboard": "delete"},
    )
    holder = await _person(db_session, "eng-holder@x.com")
    successor = await _person(db_session, "sales-successor@x.com")
    await _membership(db_session, person_id=holder.id, dept_name="Engineering")
    await _membership(db_session, person_id=successor.id, dept_name="Sales")
    await _seat(db_session, plan_id=plan.id, user_id=successor.id)
    seat = await _seat(db_session, plan_id=plan.id, user_id=holder.id)
    await db_session.commit()

    resp = await client.post(
        f"/api/v1/assets/seat-allocations/{seat.id}/revoke",
        headers=admin_headers,
        json={"successor_user_id": str(successor.id)},
    )
    assert resp.status_code == 400, resp.text
    assert resp.json()["errors"][0]["code"] == "SUCCESSOR_CROSS_DEPARTMENT"


@pytest.mark.asyncio
async def test_revoke_cross_department_blocks_asset_admin(
    client: AsyncClient, db_session: AsyncSession, asset_admin_headers,
):
    """Even with confirm_cross_department=True, asset_admin can't do
    cross-dept revoke — only global_admin can."""
    sw, plan = await _software_plan(db_session)
    await _connector(
        db_session, software_id=sw.id, provider="zoom",
        binding_config={"action_on_offboard": "delete"},
    )
    holder = await _person(db_session, "eng-holder2@x.com")
    successor = await _person(db_session, "sales-successor2@x.com")
    await _membership(db_session, person_id=holder.id, dept_name="Engineering")
    await _membership(db_session, person_id=successor.id, dept_name="Sales")
    await _seat(db_session, plan_id=plan.id, user_id=successor.id)
    seat = await _seat(db_session, plan_id=plan.id, user_id=holder.id)
    await db_session.commit()

    resp = await client.post(
        f"/api/v1/assets/seat-allocations/{seat.id}/revoke",
        headers=asset_admin_headers,
        json={
            "successor_user_id": str(successor.id),
            "confirm_cross_department": True,
        },
    )
    assert resp.status_code == 403, resp.text


@pytest.mark.asyncio
async def test_revoke_cross_department_allows_global_admin_with_confirm(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    """Cross-dept revoke succeeds when caller is global_admin AND the
    confirm flag is set explicitly."""
    sw, plan = await _software_plan(db_session)
    await _connector(
        db_session, software_id=sw.id, provider="zoom",
        binding_config={"action_on_offboard": "delete"},
    )
    holder = await _person(db_session, "eng-holder3@x.com")
    successor = await _person(db_session, "sales-successor3@x.com")
    await _membership(db_session, person_id=holder.id, dept_name="Engineering")
    await _membership(db_session, person_id=successor.id, dept_name="Sales")
    await _seat(db_session, plan_id=plan.id, user_id=successor.id)
    seat = await _seat(db_session, plan_id=plan.id, user_id=holder.id)
    await db_session.commit()

    resp = await client.post(
        f"/api/v1/assets/seat-allocations/{seat.id}/revoke",
        headers=admin_headers,
        json={
            "successor_user_id": str(successor.id),
            "confirm_cross_department": True,
        },
    )
    assert resp.status_code == 200, resp.text
    assert resp.json()["data"]["status"] == "revoked"