"""One failed revoke must not stop the expiry sweep — for real, not just
in the comment above the ``except``.

``sweep_lifecycle``'s expiry loop caught ``Exception`` per row so "one
permission must not stop the sweep". That promise does not survive the
commonest failure the loop can hit: a write that dies on **flush**
deactivates the whole transaction, so after the first such failure every
remaining expired permission raises ``PendingRollbackError`` and the
trailing ``db.flush()`` takes the cron down with it. The visible effect
is access staying live past its own expiry date, silently, with a single
warning line explaining a different problem.

The injection is deliberately a real failed flush and not a bare
``raise``: a bare ``raise`` is precisely the case the old code DID
survive, so a test built on one would pass with the fix removed.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select

from app.domains.asset.models import AssetSoftwareCatalog
from app.domains.permission.lifecycle import ApprovalSettings, sweep_lifecycle
from app.domains.permission.models import (
    PERM_ACTIVE,
    SoftwarePermission,
    UserPermission,
)
from app.identity.person_models import IdentityPerson

pytestmark = pytest.mark.asyncio


async def _person(db) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(), source_system="test",
        hris_person_id=f"P{uuid.uuid4().hex[:8]}",
        employee_no=f"E{uuid.uuid4().hex[:8]}",
        work_email=f"u{uuid.uuid4().hex[:8]}@x.test",
        display_name="Sweep Person", employment_status="active", country="CHN",
    )
    db.add(p)
    await db.flush()
    return p


async def _expired_grant(db, holder: IdentityPerson) -> UserPermission:
    sw = AssetSoftwareCatalog(name=f"SW {uuid.uuid4().hex[:6]}")
    db.add(sw)
    await db.flush()
    perm = SoftwarePermission(
        software_id=sw.id, permission_key=f"perm_{uuid.uuid4().hex[:8]}",
        display_name="Repo admin", requires_approval=False,
    )
    db.add(perm)
    await db.flush()
    row = UserPermission(
        user_id=holder.id, permission_id=perm.id, status=PERM_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=100),
        expires_at=datetime.now(UTC) - timedelta(hours=2),
    )
    db.add(row)
    await db.flush()
    return row


async def test_a_flush_failure_on_one_expiry_does_not_stop_the_others(
    db_session, monkeypatch,
):
    holder = await _person(db_session)
    first = await _expired_grant(db_session, holder)
    second = await _expired_grant(db_session, holder)
    # The sweep reads them in query order; make the failure hit whichever
    # comes first by keying on a counter rather than an id.
    seen: list[uuid.UUID] = []

    from app.domains.permission.service import PermissionService

    original = PermissionService.revoke_user_permission

    async def _maybe_poison(self, user_permission_id, **kwargs):
        seen.append(user_permission_id)
        if len(seen) == 1:
            # A duplicate-key insert — the exact shape any IntegrityError
            # from a queued provisioning task produces. Kills the whole
            # transaction, not just this statement.
            dup = IdentityPerson(
                id=uuid.uuid4(), source_system="test",
                hris_person_id=holder.hris_person_id,  # UNIQUE
                employee_no=f"E{uuid.uuid4().hex[:8]}",
                work_email=f"dup{uuid.uuid4().hex[:8]}@x.test",
                display_name="dup", employment_status="active", country="CHN",
            )
            self.db.add(dup)
            await self.db.flush()
            return None
        return await original(self, user_permission_id, **kwargs)

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

    summary = await sweep_lifecycle(
        db_session, now=datetime.now(UTC),
        settings=ApprovalSettings(), trace_id="t",
    )

    assert len(seen) == 2, "the second expired grant must still be attempted"
    assert summary["errors"] == 1
    assert summary["expired_revoke_tasks"] == 1, (
        "exactly one revoke landed: the poisoned one rolled back, "
        "the healthy one went through"
    )
    # The session survived: a query still works, and the healthy row got
    # its revoke task.
    ids = {first.id, second.id}
    rows = (
        await db_session.execute(
            select(UserPermission).where(UserPermission.id.in_(ids)),
        )
    ).scalars().all()
    tasked = [r for r in rows if r.provisioning_task_id is not None]
    assert len(tasked) == 1