"""Access certification engine (GRC PR2, migration 251).

Covers: software_seat snapshot by department subtree (reviewer = manager,
out-of-dept excluded), decision lifecycle + guards, campaign completion
gate, the access_certification compliance-control evaluator transitions,
and HTTP RBAC.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta
from decimal import Decimal

import pytest
from httpx import AsyncClient
from sqlalchemy import select

from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.domains.auditor import access_certification_service as acs
from app.domains.auditor import compliance_service as compsvc
from app.domains.auditor.models import (
    AccessCertification,
    AccessCertificationItem,
    ComplianceControl,
    ComplianceControlStatus,
)
from app.identity.person_models import IdentityOrgMembership, IdentityPerson


def _person(**kw) -> IdentityPerson:
    return IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        work_email=kw.get("email", f"u-{uuid.uuid4().hex[:6]}@x.com"),
        display_name=kw.get("name", "Person"),
        employment_status="active",
        manager_person_id=kw.get("manager_id"),
    )


def _membership(person_id, dept: str) -> IdentityOrgMembership:
    return IdentityOrgMembership(
        id=uuid.uuid4(),
        person_id=person_id,
        source_system="successfactors",
        org_node_source_id=f"node-{dept}",
        org_node_type="department",
        org_node_name=dept,
        is_primary=True,
    )


async def _seat(db, person_id) -> SoftwareSeatAllocation:
    cat = AssetSoftwareCatalog(id=uuid.uuid4(), name=f"App-{uuid.uuid4().hex[:4]}")
    db.add(cat)
    await db.flush()
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=cat.id, name="Standard",
        price=Decimal("10.00"), billing_cycle="monthly",
    )
    db.add(plan)
    await db.flush()
    seat = SoftwareSeatAllocation(
        id=uuid.uuid4(), plan_id=plan.id, user_id=person_id, status="active",
    )
    db.add(seat)
    await db.flush()
    return seat


# ── snapshot builder ─────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_create_unsupported_type_rejected(db_session):
    with pytest.raises(Exception) as ei:
        await acs.create_campaign(
            db_session, label="x", campaign_type="nope",
            scope_config={}, due_at=datetime.now(UTC), actor_person_id=None,
        )
    assert "campaign_type" in ei.value.message


@pytest.mark.asyncio
async def test_software_seat_requires_department(db_session):
    with pytest.raises(Exception) as ei:
        await acs.create_campaign(
            db_session, label="x", campaign_type="software_seat",
            scope_config={}, due_at=datetime.now(UTC), actor_person_id=None,
        )
    assert "department" in ei.value.message


@pytest.mark.asyncio
async def test_software_seat_snapshot_by_department(db_session):
    manager = _person(name="Mgr")
    db_session.add(manager)
    await db_session.flush()
    # in-dept holder (has a seat) + out-of-dept holder (has a seat).
    inside = _person(name="In", manager_id=manager.id)
    outside = _person(name="Out", manager_id=manager.id)
    db_session.add_all([inside, outside])
    await db_session.flush()
    db_session.add_all([
        _membership(inside.id, "Platform"),
        _membership(outside.id, "Sales"),
    ])
    await _seat(db_session, inside.id)
    await _seat(db_session, outside.id)
    await db_session.commit()

    cert = await acs.create_campaign(
        db_session, label="Q2 Platform seat review",
        campaign_type="software_seat", scope_config={"department": "Platform"},
        due_at=datetime.now(UTC) + timedelta(days=14), actor_person_id=manager.id,
    )
    await db_session.commit()

    assert cert.snapshot_item_count == 1  # only the in-dept seat
    items, total = await acs.list_items(db_session, certification_id=cert.id)
    assert total == 1
    it = items[0]
    assert it["holder_email"] == inside.work_email
    assert it["access_type"] == "software_seat"
    assert it["assigned_reviewer_person_id"] == str(manager.id)
    assert it["decision"] == "pending"


# ── decision lifecycle ───────────────────────────────────────────────


async def _campaign_with_items(db, n=2) -> AccessCertification:
    cert = AccessCertification(
        id=uuid.uuid4(), label="lifecycle", campaign_type="software_seat",
        scope_config={"department": "X"}, status="in_progress",
        started_at=datetime.now(UTC).replace(tzinfo=None),
        due_at=datetime.now(UTC).replace(tzinfo=None) + timedelta(days=7),
        snapshot_item_count=n,
    )
    db.add(cert)
    await db.flush()
    for i in range(n):
        db.add(AccessCertificationItem(
            id=uuid.uuid4(), certification_id=cert.id,
            access_type="software_seat", access_label=f"App{i}",
            decision="pending",
            snapshot_at=datetime.now(UTC).replace(tzinfo=None),
        ))
    await db.commit()
    return cert


@pytest.mark.asyncio
async def test_decide_and_complete_lifecycle(db_session):
    actor = _person(name="Auditor")
    db_session.add(actor)
    cert = await _campaign_with_items(db_session, n=2)
    items = (await db_session.execute(
        select(AccessCertificationItem).where(
            AccessCertificationItem.certification_id == cert.id)
    )).scalars().all()

    await acs.decide_item(db_session, item_id=items[0].id, decision="keep",
                          note=None, actor_person_id=actor.id)
    await acs.decide_item(db_session, item_id=items[1].id, decision="revoke",
                          note="not needed", actor_person_id=actor.id)
    await db_session.commit()

    refreshed = await db_session.get(AccessCertification, cert.id)
    assert refreshed.decided_count == 2
    assert refreshed.revoked_count == 1
    revoked = await db_session.get(AccessCertificationItem, items[1].id)
    # These fixture items carry no access_ref (no resolvable seat), so the
    # coordinator routes the revoke to the terminal manual_required state
    # rather than enqueuing a (nonexistent) seat revoke.
    assert revoked.revocation_status == "manual_required"

    # double-decide rejected
    with pytest.raises(Exception):
        await acs.decide_item(db_session, item_id=items[0].id, decision="keep",
                              note=None, actor_person_id=actor.id)

    completed = await acs.complete_campaign(
        db_session, certification_id=cert.id, actor_person_id=actor.id)
    await db_session.commit()
    assert completed.status == "completed"

    # decide on completed campaign rejected
    extra = await _campaign_with_items(db_session, n=1)
    extra.status = "completed"
    await db_session.commit()
    its = (await db_session.execute(
        select(AccessCertificationItem).where(
            AccessCertificationItem.certification_id == extra.id)
    )).scalars().all()
    with pytest.raises(Exception):
        await acs.decide_item(db_session, item_id=its[0].id, decision="keep",
                              note=None, actor_person_id=actor.id)


@pytest.mark.asyncio
async def test_complete_blocked_while_pending(db_session):
    actor = _person(name="A")
    db_session.add(actor)
    cert = await _campaign_with_items(db_session, n=2)
    with pytest.raises(Exception) as ei:
        await acs.complete_campaign(
            db_session, certification_id=cert.id, actor_person_id=actor.id)
    assert "pending" in ei.value.message.lower()


# ── compliance control evaluator ─────────────────────────────────────


async def _eval_access_cert(db_session) -> str:
    await compsvc.recompute_control_status(db_session)
    ctrl = (await db_session.execute(
        select(ComplianceControl).where(
            ComplianceControl.key == "access_certification")
    )).scalar_one()
    st = (await db_session.execute(
        select(ComplianceControlStatus).where(
            ComplianceControlStatus.control_id == ctrl.id)
    )).scalar_one()
    return st.status


@pytest.mark.asyncio
async def test_control_not_configured_without_campaign(db_session):
    assert await _eval_access_cert(db_session) == "not_configured"


@pytest.mark.asyncio
async def test_control_pass_in_progress_and_fail_overdue(db_session):
    cert = AccessCertification(
        id=uuid.uuid4(), label="open", campaign_type="software_seat",
        scope_config={}, status="in_progress",
        started_at=datetime.now(UTC).replace(tzinfo=None),
        due_at=datetime.now(UTC).replace(tzinfo=None) + timedelta(days=5),
    )
    db_session.add(cert)
    await db_session.commit()
    assert await _eval_access_cert(db_session) == "pass"

    cert.due_at = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=1)
    await db_session.commit()
    assert await _eval_access_cert(db_session) == "fail"


# ── HTTP RBAC ────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_certifications_http_rbac(client: AsyncClient, admin_headers,
                                        employee_headers):
    # list — view open
    r = await client.get("/api/v1/compliance/certifications",
                         headers=admin_headers)
    assert r.status_code == 200, r.text
    assert "campaigns" in r.json()["data"]

    # create — manage only; software_seat on an empty dept yields 0 items
    # but proves the builder runs end-to-end without seeding.
    body = {"label": "audit", "campaign_type": "software_seat",
            "scope_config": {"department": "Nonexistent"}, "due_in_days": 7}
    ok = await client.post("/api/v1/compliance/certifications",
                           json=body, headers=admin_headers)
    assert ok.status_code == 200, ok.text
    assert ok.json()["data"]["snapshot_item_count"] == 0

    denied = await client.post("/api/v1/compliance/certifications",
                               json=body, headers=employee_headers)
    assert denied.status_code == 403


# ── privileged_role fold-in (GRC PR3) ────────────────────────────────


@pytest.mark.asyncio
async def test_privileged_role_snapshot_includes_helix_admins(db_session):
    """The privileged_role campaign builder snapshots Helix admins from
    auth_user_roles; external connectors degrade gracefully when not
    configured. Verifies the fold-in of the old privileged-review."""
    from sqlalchemy import text

    from app.auth.models import AuthUserRole

    person = _person(name="Admin")
    db_session.add(person)
    await db_session.flush()
    role_id = (await db_session.execute(
        text("SELECT id FROM auth_roles WHERE name = 'global_admin'")
    )).scalar_one()
    role_id = uuid.UUID(role_id) if isinstance(role_id, str) else role_id
    db_session.add(AuthUserRole(user_id=person.id, role_id=role_id))
    await db_session.commit()

    cert = await acs.create_campaign(
        db_session, label="Q-priv", campaign_type="privileged_role",
        scope_config={}, due_at=datetime.now(UTC) + timedelta(days=14),
        actor_person_id=None,
    )
    await db_session.commit()
    assert cert.campaign_type == "privileged_role"

    items, total = await acs.list_items(
        db_session, certification_id=cert.id, page_size=200)
    helix = [i for i in items if i["access_type"] == "privileged_role"
             and i["holder_email"] == person.work_email]
    assert len(helix) >= 1
    assert helix[0]["access_label"].startswith("helix/")


@pytest.mark.asyncio
async def test_privileged_snapshot_excludes_soft_deleted_holders(db_session):
    """A tombstone must not be certified as a sitting admin (#2163).

    Soft-deleted persons keep ``employment_status='active'`` here, so
    the join had no indirect filter either: a purged duplicate whose
    auth_user_roles row was never cleaned up showed up in the quarterly
    review as a live global_admin, and the reviewer's only sane answer
    (revoke) landed in the manual_required dead queue.
    """
    from sqlalchemy import text

    from app.auth.models import AuthUserRole

    ghost = _person(name="Ghost Admin")
    ghost.deleted_at = datetime.now(UTC).replace(tzinfo=None)
    live = _person(name="Live Admin")
    db_session.add_all([ghost, live])
    await db_session.flush()
    role_id = (await db_session.execute(
        text("SELECT id FROM auth_roles WHERE name = 'global_admin'")
    )).scalar_one()
    role_id = uuid.UUID(role_id) if isinstance(role_id, str) else role_id
    db_session.add(AuthUserRole(user_id=ghost.id, role_id=role_id))
    db_session.add(AuthUserRole(user_id=live.id, role_id=role_id))
    await db_session.commit()

    cert = await acs.create_campaign(
        db_session, label="Q-priv-tomb", campaign_type="privileged_role",
        scope_config={}, due_at=datetime.now(UTC) + timedelta(days=14),
        actor_person_id=None,
    )
    await db_session.commit()

    items, _total = await acs.list_items(
        db_session, certification_id=cert.id, page_size=500)
    emails = {i["holder_email"] for i in items}
    assert live.work_email in emails
    assert ghost.work_email not in emails, (
        "a soft-deleted person is being certified as an active "
        "privileged-role holder"
    )


@pytest.mark.asyncio
async def test_escalate_overdue_writes_incident_idempotent(db_session):
    from app.compliance.models import ComplianceIncident

    cert = AccessCertification(
        id=uuid.uuid4(), label="esc", campaign_type="software_seat",
        scope_config={}, status="in_progress",
        started_at=datetime.now(UTC).replace(tzinfo=None),
        due_at=datetime.now(UTC).replace(tzinfo=None) - timedelta(days=20),
        snapshot_item_count=1,
    )
    db_session.add(cert)
    await db_session.flush()
    db_session.add(AccessCertificationItem(
        id=uuid.uuid4(), certification_id=cert.id,
        access_type="software_seat", access_label="App",
        decision="pending",
        decided_at_deadline=datetime.now(UTC).replace(tzinfo=None)
        - timedelta(days=20),
        snapshot_at=datetime.now(UTC).replace(tzinfo=None),
    ))
    await db_session.commit()

    res = await acs.escalate_overdue(db_session, trace_id="t:test")
    await db_session.commit()
    assert res["escalated"] == 1
    n_incidents = int((await db_session.execute(
        select(__import__("sqlalchemy").func.count()).select_from(
            ComplianceIncident).where(
            ComplianceIncident.incident_type == "access_review_overdue")
    )).scalar_one())
    assert n_incidents >= 1

    # Idempotent — escalated_at set, second sweep skips it.
    res2 = await acs.escalate_overdue(db_session, trace_id="t:test2")
    await db_session.commit()
    assert res2["escalated"] == 0


# ── revoke fulfilment (decision → real downstream revoke) ─────────────


@pytest.mark.asyncio
async def test_decide_revoke_flips_seat_and_enqueues(db_session):
    """Deciding revoke on a software_seat item flips the seat to ``revoked``
    in Helix (authoritative) and enqueues a downstream provisioning revoke
    task, linked via revocation_trace_id — the attestation is now actually
    enforced, not just recorded. revocation_status stays ``pending`` until
    the task reaches a terminal state, then the fulfil reconciler closes it."""
    from app.core.coordinators.access_certification_coordinator import (
        fulfil_pending_revocations,
    )
    from app.domains.provisioning.models import ProvisioningTask

    mgr = _person(name="Mgr")
    db_session.add(mgr)
    await db_session.flush()
    holder = _person(name="Holder", manager_id=mgr.id)
    db_session.add(holder)
    await db_session.flush()
    db_session.add(_membership(holder.id, "Platform"))
    seat = await _seat(db_session, holder.id)
    await db_session.commit()

    cert = await acs.create_campaign(
        db_session, label="revoke me", campaign_type="software_seat",
        scope_config={"department": "Platform"},
        due_at=datetime.now(UTC) + timedelta(days=14), actor_person_id=mgr.id,
    )
    await db_session.commit()
    item = (await db_session.execute(
        select(AccessCertificationItem).where(
            AccessCertificationItem.certification_id == cert.id)
    )).scalar_one()
    assert item.access_ref == str(seat.id)

    await acs.decide_item(
        db_session, item_id=item.id, decision="revoke", note=None,
        actor_person_id=mgr.id)
    await db_session.commit()

    seat_after = await db_session.get(SoftwareSeatAllocation, seat.id)
    assert seat_after.status == "revoked"
    assert seat_after.revoked_at is not None
    item_after = await db_session.get(AccessCertificationItem, item.id)
    assert item_after.revocation_status == "pending"
    assert item_after.revocation_trace_id is not None
    task = await db_session.get(
        ProvisioningTask, item_after.revocation_trace_id)
    assert task is not None
    assert task.action == "revoke"
    assert task.source == "audit"

    # When that task reaches a terminal state, the fulfil reconciler closes
    # the item out.
    task.status = "notified"  # manual_notify delivered to a human
    await db_session.commit()
    await fulfil_pending_revocations(db_session, trace_id="t:fulfil")
    await db_session.commit()
    item_done = await db_session.get(AccessCertificationItem, item.id)
    assert item_done.revocation_status == "completed"
    assert item_done.revocation_completed_at is not None


@pytest.mark.asyncio
async def test_fulfil_mirrors_provisioning_task_terminal(db_session):
    """The fulfil reconciler mirrors a pending item's provisioning-task
    terminal state: completed→completed, failed→failed, in-flight→pending."""
    from app.core.coordinators.access_certification_coordinator import (
        fulfil_pending_revocations,
    )
    from app.domains.provisioning.models import ProvisioningTask

    actor = _person(name="A")
    db_session.add(actor)
    cert = await _campaign_with_items(db_session, n=3)
    items = (await db_session.execute(
        select(AccessCertificationItem).where(
            AccessCertificationItem.certification_id == cert.id)
        .order_by(AccessCertificationItem.access_label)
    )).scalars().all()

    def _task(status: str) -> ProvisioningTask:
        return ProvisioningTask(
            id=uuid.uuid4(), user_id=actor.id, action="revoke",
            target_type="seat", source="audit", execution_mode="auto_api",
            status=status,
        )
    t_done, t_fail, t_run = _task("completed"), _task("failed"), _task("in_progress")
    db_session.add_all([t_done, t_fail, t_run])
    await db_session.flush()

    for it, task in zip(items, [t_done, t_fail, t_run]):
        it.decision = "revoke"
        it.revocation_status = "pending"
        it.revocation_trace_id = task.id
    await db_session.commit()

    res = await fulfil_pending_revocations(db_session, trace_id="t:fulfil")
    await db_session.commit()

    assert (await db_session.get(AccessCertificationItem, items[0].id)).revocation_status == "completed"
    assert (await db_session.get(AccessCertificationItem, items[1].id)).revocation_status == "failed"
    assert (await db_session.get(AccessCertificationItem, items[2].id)).revocation_status == "pending"
    assert res["completed"] == 1
    assert res["failed"] == 1
    assert res["still_pending"] == 1


@pytest.mark.asyncio
async def test_revoke_non_software_seat_is_manual_required(db_session):
    """A revoke on a non-software_seat item (e.g. an external privileged
    role) has no automated enforcement path → it lands in the terminal
    ``manual_required`` state, NOT ``pending`` — so the fulfil cron does
    not spin on it forever and the operator sees it needs hands-on revoke."""
    from app.core.coordinators.access_certification_coordinator import (
        fulfil_pending_revocations,
    )

    actor = _person(name="Auditor")
    db_session.add(actor)
    cert = AccessCertification(
        id=uuid.uuid4(), label="priv", campaign_type="privileged_role",
        scope_config={}, status="in_progress",
        started_at=datetime.now(UTC).replace(tzinfo=None),
        due_at=datetime.now(UTC).replace(tzinfo=None) + timedelta(days=7),
        snapshot_item_count=1,
    )
    db_session.add(cert)
    await db_session.flush()
    item = AccessCertificationItem(
        id=uuid.uuid4(), certification_id=cert.id,
        access_type="privileged_role", access_label="AWS admin",
        access_ref=None, decision="pending",
        snapshot_at=datetime.now(UTC).replace(tzinfo=None),
    )
    db_session.add(item)
    await db_session.commit()

    await acs.decide_item(db_session, item_id=item.id, decision="revoke",
                          note=None, actor_person_id=actor.id)
    await db_session.commit()

    refreshed = await db_session.get(AccessCertificationItem, item.id)
    assert refreshed.revocation_status == "manual_required"
    assert refreshed.revocation_trace_id is None

    # The fulfil cron must NOT pick up a manual_required item (terminal).
    res = await fulfil_pending_revocations(db_session, trace_id="t:m")
    await db_session.commit()
    assert res["scanned"] == 0


@pytest.mark.asyncio
async def test_revoke_fulfil_cron_raises_when_items_failed(monkeypatch):
    """Regression: the 5-minute revoke-fulfil cron must RAISE when any
    revoke decision failed to reach downstream (``result['failed'] > 0``).
    Previously it returned the result verbatim, so the run landed
    ``status='ok'`` and the failure-notifier never fired — an unenforced
    revoke of access sat silently."""
    from app.scheduled_jobs import auditor_handlers

    async def fake_fulfil(session, *, trace_id, batch_size=100):
        return {"scanned": 3, "completed": 2, "failed": 1, "still_pending": 0}

    # The handler imports this symbol from the coordinator module at call time.
    monkeypatch.setattr(
        "app.core.coordinators.access_certification_coordinator."
        "fulfil_pending_revocations",
        fake_fulfil,
    )
    with pytest.raises(RuntimeError, match="revoke fulfil"):
        await auditor_handlers.run_access_certification_revoke_fulfil(
            None, {},  # db arg unused (handler opens its own session)
        )


@pytest.mark.asyncio
async def test_revoke_fulfil_cron_ok_when_no_failures(monkeypatch):
    """The cron returns normally (status='ok') when nothing failed."""
    from app.scheduled_jobs import auditor_handlers

    async def fake_fulfil(session, *, trace_id, batch_size=100):
        return {"scanned": 2, "completed": 2, "failed": 0, "still_pending": 0}

    monkeypatch.setattr(
        "app.core.coordinators.access_certification_coordinator."
        "fulfil_pending_revocations",
        fake_fulfil,
    )
    result = await auditor_handlers.run_access_certification_revoke_fulfil(
        None, {},
    )
    assert result["failed"] == 0
    assert "trace_id" in result


def test_cert_manual_upsert_normalizes_naive_datetime():
    """Regression: a naive ``expires_at`` (date-picker / offset-less ISO) must
    be stamped UTC so the service's ``expires_at - datetime.now(UTC)`` doesn't
    raise TypeError (naive - aware) → 500 on POST .../cert-expiries/manual."""
    from datetime import UTC, datetime

    from app.domains.auditor.schemas import CertManualUpsertRequest

    req = CertManualUpsertRequest(
        asset_kind="ssl_cert",
        external_id="cert-123",
        name="example.com",
        expires_at=datetime(2027, 1, 1, 0, 0, 0),  # naive
        issued_at=datetime(2026, 1, 1, 0, 0, 0),    # naive
    )
    assert req.expires_at.tzinfo is not None
    assert req.issued_at is not None and req.issued_at.tzinfo is not None
    # Already-aware values pass through unchanged.
    aware = datetime(2027, 6, 1, tzinfo=UTC)
    req2 = CertManualUpsertRequest(
        asset_kind="ssl_cert", external_id="c2", name="x", expires_at=aware,
    )
    assert req2.expires_at == aware


@pytest.mark.asyncio
async def test_generic_create_rejects_self_attestation(db_session):
    """#2164 — the generic opener has none of the three things this
    campaign type needs.

    ``open_self_attestation_campaign`` holds a Redis lock, is idempotent
    by period label, and notifies every holder. Opened through the
    generic endpoint instead: thousands of items snapshotted silently,
    a second campaign able to coexist and hide the first from
    ``my_open_self_attestation``, and 14 days past the deadline
    ``escalate_overdue`` chasing every employee about a self-check they
    were never told had started.
    """
    from app.core.exceptions import AppException

    with pytest.raises(AppException) as exc:
        await acs.create_campaign(
            db_session, label="sneaky-self-attest",
            campaign_type="self_attestation", scope_config={},
            due_at=datetime.now(UTC) + timedelta(days=14),
            actor_person_id=None,
        )
    assert exc.value.code == "USE_SELF_ATTESTATION_OPENER"


@pytest.mark.asyncio
async def test_dedicated_opener_still_works(db_session):
    """The gate must not break the sanctioned path."""
    result = await acs.open_self_attestation_campaign(
        db_session, period_year=2026, half="H2", actor_person_id=None,
    )
    assert result.get("certification_id"), result


@pytest.mark.asyncio
async def test_decide_item_takes_a_row_lock(db_session):
    """#2162 — the pending guard is a read-then-write over a counter.

    Two reviewers (or HTTP + MCP) deciding the same item both saw
    ``pending`` in their own snapshot, both passed the guard, and each
    applied ``decided_count + 1`` — pushing the counter past
    ``snapshot_item_count`` with no reconcile cron to heal it. Asserting
    on the emitted SQL rather than racing two sessions: the lock either
    is in the statement or it is not, and that is the whole fix.
    """
    from sqlalchemy import event

    actor = _person(name="LockAuditor")
    db_session.add(actor)
    cert = await _campaign_with_items(db_session, n=1)
    item = (await db_session.execute(
        select(AccessCertificationItem).where(
            AccessCertificationItem.certification_id == cert.id)
    )).scalars().one()
    await db_session.commit()

    seen: list[str] = []

    def _capture(_conn, _cursor, statement, *_a, **_kw):
        if "access_certification_items" in statement and "SELECT" in statement:
            seen.append(statement)

    sync_engine = db_session.get_bind().engine.sync_engine \
        if hasattr(db_session.get_bind().engine, 'sync_engine') \
        else db_session.get_bind().engine
    event.listen(sync_engine, "before_cursor_execute", _capture)
    try:
        await acs.decide_item(
            db_session, item_id=item.id, decision="keep",
            note=None, actor_person_id=actor.id,
        )
    finally:
        event.remove(sync_engine, "before_cursor_execute", _capture)

    assert any("FOR UPDATE" in s for s in seen), (
        "the item was read without a row lock:\n" + "\n".join(seen)
    )


@pytest.mark.asyncio
async def test_self_attestation_submit_locks_its_items(db_session):
    """#2162, sibling path — double-submit (two tabs) counted twice."""
    from sqlalchemy import event

    from app.domains.auditor.access_certification_service import (
        CAMPAIGN_TYPE_SELF_ATTESTATION,
    )

    holder = _person(name="Holder")
    db_session.add(holder)
    await db_session.flush()
    cert = AccessCertification(
        id=uuid.uuid4(), label="self-lock", campaign_type=(
            CAMPAIGN_TYPE_SELF_ATTESTATION
        ),
        scope_config={}, status="in_progress",
        started_at=datetime.now(UTC).replace(tzinfo=None),
        due_at=datetime.now(UTC).replace(tzinfo=None) + timedelta(days=7),
        snapshot_item_count=1,
    )
    db_session.add(cert)
    await db_session.flush()
    item = AccessCertificationItem(
        id=uuid.uuid4(), certification_id=cert.id, person_id=holder.id,
        access_type="software_seat", access_label="x/y",
        decision="pending",
        snapshot_at=datetime.now(UTC).replace(tzinfo=None),
        decided_at_deadline=cert.due_at,
    )
    db_session.add(item)
    await db_session.commit()

    seen: list[str] = []

    def _capture(_conn, _cursor, statement, *_a, **_kw):
        if "access_certification_items" in statement and "SELECT" in statement:
            seen.append(statement)

    sync_engine = db_session.get_bind().engine.sync_engine \
        if hasattr(db_session.get_bind().engine, 'sync_engine') \
        else db_session.get_bind().engine
    event.listen(sync_engine, "before_cursor_execute", _capture)
    try:
        await acs.submit_self_attestation(
            db_session, actor_person_id=holder.id,
            keep_item_ids=[item.id], remove_item_ids=[],
        )
    finally:
        event.remove(sync_engine, "before_cursor_execute", _capture)

    assert any("FOR UPDATE" in s for s in seen), (
        "self-attestation items were read without a row lock:\n"
        + "\n".join(seen)
    )