"""Batch revoke for software permission grants.

POST /api/v1/permissions/grants/batch-revoke

Reuses the single-row ``revoke_user_permission`` per row (so each
succeeded grant gets its own revoke provisioning task and the status
flip stays deferred to the task — Helix never claims REVOKED before the
driver ran). Already-revoked / terminal grants → ``skipped``; one
aggregated ``permission.grants.batch_revoke`` audit row per call.
"""
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.domains.asset.models import AssetSoftwareCatalog
from app.domains.permission.models import (
    PERM_ACTIVE,
    PERM_REVOKED,
    SoftwarePermission,
    UserPermission,
)
from app.domains.provisioning.models import ACTION_REVOKE, ProvisioningTask


async def _seed_grant(
    db: AsyncSession, user_id: uuid.UUID, *, status: str = PERM_ACTIVE,
) -> UserPermission:
    software = AssetSoftwareCatalog(name=f"BatchRevoke SW {uuid.uuid4().hex[:6]}")
    db.add(software)
    await db.flush()
    permission = SoftwarePermission(
        software_id=software.id,
        permission_key=f"perm_{uuid.uuid4().hex[:8]}",
        display_name="Batch revoke test permission",
    )
    db.add(permission)
    await db.flush()
    grant = UserPermission(
        user_id=user_id,
        permission_id=permission.id,
        status=status,
    )
    db.add(grant)
    await db.flush()
    return grant


@pytest.mark.asyncio
async def test_batch_revoke_all_success(client, admin_user, employee_user, db_session):
    _, token = admin_user
    employee, _ = employee_user
    grant_a = await _seed_grant(db_session, employee.id)
    grant_b = await _seed_grant(db_session, employee.id)
    await db_session.commit()

    resp = await client.post(
        "/api/v1/permissions/grants/batch-revoke",
        json={"grant_ids": [str(grant_a.id), str(grant_b.id)], "reason": "batch test"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["succeeded"] == 2
    assert data["skipped"] == 0
    assert data["failed"] == 0

    # Each succeeded row got its own revoke provisioning task; status
    # stays ACTIVE until the task succeeds (single-row invariant).
    for grant in (grant_a, grant_b):
        row = await db_session.get(UserPermission, grant.id)
        assert row.provisioning_task_id is not None
        task = await db_session.get(ProvisioningTask, row.provisioning_task_id)
        assert task.action == ACTION_REVOKE
        assert row.status == PERM_ACTIVE


@pytest.mark.asyncio
async def test_batch_revoke_mixed_rows_and_audit(
    client, admin_user, employee_user, db_session,
):
    _, token = admin_user
    employee, _ = employee_user
    good = await _seed_grant(db_session, employee.id)
    already_revoked = await _seed_grant(db_session, employee.id, status=PERM_REVOKED)
    missing_id = str(uuid.uuid4())
    await db_session.commit()

    resp = await client.post(
        "/api/v1/permissions/grants/batch-revoke",
        json={"grant_ids": [str(good.id), str(already_revoked.id), missing_id]},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["succeeded"] == 1
    assert data["skipped"] == 1
    assert data["failed"] == 1
    by_id = {r["id"]: r for r in data["results"]}
    assert by_id[str(good.id)]["status"] == "succeeded"
    assert by_id[str(already_revoked.id)]["status"] == "skipped"
    assert by_id[missing_id]["status"] == "failed"
    assert by_id[missing_id]["error"] == "not_found"

    audit_row = (
        await db_session.execute(
            select(AuditLog)
            .where(AuditLog.action == "permission.grants.batch_revoke")
            .order_by(AuditLog.created_at.desc())
        )
    ).scalars().first()
    assert audit_row is not None
    assert audit_row.detail["total"] == 3
    assert audit_row.detail["succeeded"] == 1
    assert audit_row.detail["skipped"] == 1
    assert audit_row.detail["failed"] == 1
    assert str(good.id) in audit_row.detail["ids"]


@pytest.mark.asyncio
async def test_batch_revoke_employee_403(client, employee_user):
    _, token = employee_user
    resp = await client.post(
        "/api/v1/permissions/grants/batch-revoke",
        json={"grant_ids": [str(uuid.uuid4())]},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_batch_revoke_over_cap_422(client, admin_user):
    _, token = admin_user
    resp = await client.post(
        "/api/v1/permissions/grants/batch-revoke",
        json={"grant_ids": [str(uuid.uuid4()) for _ in range(501)]},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_batch_revoke_reraises_transaction_aborting_error(
    client, admin_user, employee_user, db_session, monkeypatch,
):
    """Regression: a per-row lock-wait/deadlock (errno 1205/1213) rolls
    back the WHOLE transaction server-side. Swallowing it inside the
    per-row SAVEPOINT loop turns a transient, retryable deadlock into a
    "SAVEPOINT does not exist" cascade that silently marks every
    remaining grant failed and commits a partial batch. The endpoint must
    re-raise so the caller restarts on a fresh transaction (HTTP 500),
    not return 200 with a fabricated per-row failure.
    """
    from sqlalchemy.exc import OperationalError

    from app.domains.permission.service import PermissionService

    _, token = admin_user
    employee, _ = employee_user
    grant = await _seed_grant(db_session, employee.id)
    await db_session.commit()

    class _LockWaitOrig(Exception):
        def __init__(self):
            super().__init__(1205, "Lock wait timeout exceeded")
            self.args = (1205, "Lock wait timeout exceeded")

    async def _boom(self, *args, **kwargs):
        raise OperationalError("UPDATE ...", {}, _LockWaitOrig())

    monkeypatch.setattr(
        PermissionService, "revoke_user_permission", _boom,
    )

    # The txn-aborting error must propagate (→ 500 / fresh-tx retry),
    # NOT be swallowed into a 200 partial-failure report. Under the test
    # ASGI transport an unhandled endpoint exception surfaces to the
    # caller rather than being rendered as a 500 body.
    with pytest.raises(OperationalError):
        await client.post(
            "/api/v1/permissions/grants/batch-revoke",
            json={"grant_ids": [str(grant.id)], "reason": "deadlock test"},
            headers={"Authorization": f"Bearer {token}"},
        )


@pytest.mark.asyncio
async def test_batch_revoke_unexpected_error_returns_stable_code(
    client, admin_user, employee_user, db_session, monkeypatch, caplog,
):
    from app.domains.permission.service import PermissionService

    secret = "api_key=permission-secret at /srv/private/permission.py:99"

    async def _boom(*args, **kwargs):
        raise RuntimeError(secret)

    monkeypatch.setattr(PermissionService, "revoke_user_permission", _boom)
    employee, _ = employee_user
    grant = await _seed_grant(db_session, employee.id)
    await db_session.commit()
    _, token = admin_user

    response = await client.post(
        "/api/v1/permissions/grants/batch-revoke",
        json={"grant_ids": [str(grant.id)], "reason": "test"},
        headers={"Authorization": f"Bearer {token}"},
    )

    assert response.status_code == 200
    assert response.json()["data"]["results"][0] == {
        "id": str(grant.id),
        "status": "failed",
        "error": "INTERNAL_ERROR",
    }
    assert secret not in response.text
    assert secret not in caplog.text