"""Tests for the post-failure side-effect hook + dispatcher per-user
serialization.

Closes the bug class caught 2026-05-08 where Helix flipped a seat to
``revoked`` immediately on enqueue but the dispatcher's vendor call
failed (Zoom 1107 managed-domain), leaving Helix-says-revoked /
vendor-still-licensed divergence.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime

import pytest

from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.domains.provisioning.models import (
    ACTION_GRANT,
    ACTION_REVOKE,
    EXEC_AUTO_API,
    EXEC_MANUAL_NOTIFY,
    SOURCE_MANUAL,
    SOURCE_REQUEST,
    STATUS_FAILED,
    STATUS_PENDING,
    ProvisioningTask,
    SoftwareConnector,
)
from app.integrations.driver_base import ProvisioningResult
from app.integrations.models import Integration


async def _make_software(db_session, name: str = "TestSaaS") -> AssetSoftwareCatalog:
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(),
        name=name,
        asset_type="SaaS",
        status="active",
    )
    db_session.add(sw)
    await db_session.flush()
    return sw


async def _make_plan(
    db_session, software_id: uuid.UUID, name: str = "Standard",
) -> SoftwareLicensePlan:
    from decimal import Decimal as _Decimal
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(),
        software_id=software_id,
        name=name,
        price=_Decimal("0.00"),
        billing_cycle="monthly",
    )
    db_session.add(plan)
    await db_session.flush()
    return plan


async def _make_active_seat(
    db_session, plan_id: uuid.UUID, user_id: uuid.UUID,
) -> SoftwareSeatAllocation:
    seat = SoftwareSeatAllocation(
        id=uuid.uuid4(),
        plan_id=plan_id,
        user_id=user_id,
        status="active",
        allocated_at=datetime.now(UTC),
    )
    db_session.add(seat)
    await db_session.flush()
    return seat


async def _make_revoke_task(
    db_session,
    user_id: uuid.UUID,
    software_id: uuid.UUID,
    seat_id: uuid.UUID,
    *,
    status: str = STATUS_PENDING,
    revert_seat_on_failure: bool = True,
) -> ProvisioningTask:
    """Helper: creates an admin-style revoke task. Defaults to
    ``revert_seat_on_failure=True`` because the admin endpoint opts
    in by default; tests that exercise the leaver-blocked / change-
    plan path explicitly pass ``False``."""
    ctx = {}
    if revert_seat_on_failure:
        ctx["revert_seat_on_failure"] = True
    task = ProvisioningTask(
        id=uuid.uuid4(),
        user_id=user_id,
        software_id=software_id,
        action=ACTION_REVOKE,
        target_type="seat",
        target_id=seat_id,
        source=SOURCE_MANUAL,
        execution_mode=EXEC_AUTO_API,
        status=status,
        context_data=ctx,
    )
    db_session.add(task)
    await db_session.flush()
    return task


# ── _post_failure_side_effects ───────────────────────────────────


@pytest.mark.asyncio
async def test_revoke_auto_api_failed_reverts_seat_to_active(
    db_session, employee_user, monkeypatch,
):
    """When dispatcher gets ok=False from a revoke driver call, the
    optimistically-revoked seat must flip back to active so the admin
    sees the seat is still live and can retry. Closes 2026-05-08
    Zoom managed-domain bug."""
    employee, _ = employee_user
    sw = await _make_software(db_session)
    integ = Integration(
        id=uuid.uuid4(),
        kind="software_provisioning",
        provider="manual",
        name="Manual",
        slug="manual-failure-revert",
        enabled=True,
        is_primary=False,
        config_public={},
    )
    db_session.add(integ)
    await db_session.flush()
    db_session.add(SoftwareConnector(
        id=uuid.uuid4(),
        software_id=sw.id,
        integration_id=integ.id,
        binding_config={},
        is_active=True,
    ))
    plan = await _make_plan(db_session, sw.id)
    seat = await _make_active_seat(db_session, plan.id, employee.id)
    # Mirror the admin-revoke endpoint's optimistic flip — the seat is
    # already revoked locally when the dispatcher picks up the task.
    seat.status = "revoked"
    seat.revoked_at = datetime.now(UTC)
    await db_session.flush()
    task = await _make_revoke_task(db_session, employee.id, sw.id, seat.id)
    await db_session.commit()

    async def _failing_revoke(self, ctx):
        return ProvisioningResult(
            ok=False,
            error_message="Zoom delete user 400: 1107 managed domain",
        )

    from app.integrations.drivers.manual_software import ManualSoftwareDriver
    monkeypatch.setattr(ManualSoftwareDriver, "revoke_seat", _failing_revoke)

    from app.domains.provisioning.service import ProvisioningService
    svc = ProvisioningService(db_session)
    updated = await svc.execute_task(
        task, trace_id="test-revert", actor_user_id=None,
    )
    await db_session.commit()

    assert updated.status == STATUS_FAILED
    assert "1107" in (updated.error_message or "")

    await db_session.refresh(seat)
    assert seat.status == "active"
    assert seat.revoked_at is None
    assert seat.reason is not None
    assert "task" in seat.reason and "failed" in seat.reason


@pytest.mark.asyncio
async def test_revoke_failure_revert_idempotent_on_already_active_seat(
    db_session, employee_user, monkeypatch,
):
    """If the seat is already active when post_failure_side_effects
    runs (e.g. the optimistic flip was rolled back by another path),
    don't trample the reason field with a duplicate marker."""
    employee, _ = employee_user
    sw = await _make_software(db_session)
    integ = Integration(
        id=uuid.uuid4(), kind="software_provisioning", provider="manual",
        name="Manual", slug="manual-revert-idempotent", enabled=True,
        is_primary=False, config_public={},
    )
    db_session.add(integ)
    await db_session.flush()
    db_session.add(SoftwareConnector(
        id=uuid.uuid4(), software_id=sw.id, integration_id=integ.id,
        binding_config={}, is_active=True,
    ))
    plan = await _make_plan(db_session, sw.id)
    seat = await _make_active_seat(db_session, plan.id, employee.id)  # stays active
    task = await _make_revoke_task(db_session, employee.id, sw.id, seat.id)
    await db_session.commit()

    async def _failing_revoke(self, ctx):
        return ProvisioningResult(ok=False, error_message="boom")

    from app.integrations.drivers.manual_software import ManualSoftwareDriver
    monkeypatch.setattr(ManualSoftwareDriver, "revoke_seat", _failing_revoke)

    from app.domains.provisioning.service import ProvisioningService
    svc = ProvisioningService(db_session)
    await svc.execute_task(task, trace_id="t", actor_user_id=None)
    await db_session.commit()

    await db_session.refresh(seat)
    assert seat.status == "active"
    # Reason untouched (no failure marker appended) since seat wasn't revoked
    assert seat.reason is None or "failed" not in (seat.reason or "")


@pytest.mark.asyncio
async def test_revoke_without_revert_flag_does_not_revert_seat(
    db_session, employee_user, monkeypatch,
):
    """Leaver-blocked guard / change-plan paths pre-flip seats but
    do NOT opt into the ``revert_seat_on_failure`` flag. Their
    seats must stay revoked on task failure — the Helix-side
    revoke is either the billing-correctness invariant (leaver-
    blocked guard) or paired with sibling state (change-plan's
    new seat) that a unilateral revert would break."""
    from app.domains.provisioning.models import SOURCE_OFFBOARDING

    employee, _ = employee_user
    sw = await _make_software(db_session, name="NoRevertFlag")
    integ = Integration(
        id=uuid.uuid4(), kind="software_provisioning", provider="manual",
        name="Manual", slug="manual-no-revert-flag", enabled=True,
        is_primary=False, config_public={},
    )
    db_session.add(integ)
    await db_session.flush()
    db_session.add(SoftwareConnector(
        id=uuid.uuid4(), software_id=sw.id, integration_id=integ.id,
        binding_config={}, is_active=True,
    ))
    plan = await _make_plan(db_session, sw.id)
    seat = await _make_active_seat(db_session, plan.id, employee.id)
    seat.status = "revoked"  # caller (e.g. leaver-blocked guard) pre-flipped
    seat.revoked_at = datetime.now(UTC)
    await db_session.flush()
    task = ProvisioningTask(
        id=uuid.uuid4(),
        user_id=employee.id,
        software_id=sw.id,
        action=ACTION_REVOKE,
        target_type="seat",
        target_id=seat.id,
        source=SOURCE_OFFBOARDING,
        execution_mode=EXEC_AUTO_API,
        status=STATUS_PENDING,
        context_data={},  # ← key: no revert_seat_on_failure flag
    )
    db_session.add(task)
    await db_session.commit()

    async def _failing_revoke(self, ctx):
        return ProvisioningResult(ok=False, error_message="vendor down")

    from app.integrations.drivers.manual_software import ManualSoftwareDriver
    monkeypatch.setattr(ManualSoftwareDriver, "revoke_seat", _failing_revoke)

    from app.domains.provisioning.service import ProvisioningService
    svc = ProvisioningService(db_session)
    await svc.execute_task(task, trace_id="no-flag", actor_user_id=None)
    await db_session.commit()

    await db_session.refresh(seat)
    # No opt-in flag: seat stays revoked even on task failure
    assert seat.status == "revoked"
    assert seat.revoked_at is not None


@pytest.mark.asyncio
async def test_grant_failure_does_not_touch_seat(
    db_session, employee_user, monkeypatch,
):
    """The post-failure hook only reverts REVOKE failures. A failed
    GRANT shouldn't mutate the seat (the seat is still pending/un-
    granted in that case — the grant failure path leaves it alone)."""
    employee, _ = employee_user
    sw = await _make_software(db_session)
    integ = Integration(
        id=uuid.uuid4(), kind="software_provisioning", provider="manual",
        name="Manual", slug="manual-grant-failure", enabled=True,
        is_primary=False, config_public={},
    )
    db_session.add(integ)
    await db_session.flush()
    db_session.add(SoftwareConnector(
        id=uuid.uuid4(), software_id=sw.id, integration_id=integ.id,
        binding_config={}, is_active=True,
    ))
    plan = await _make_plan(db_session, sw.id)
    seat = SoftwareSeatAllocation(
        id=uuid.uuid4(),
        plan_id=plan.id,
        user_id=employee.id,
        status="pending",  # grant hasn't completed
        allocated_at=None,
    )
    db_session.add(seat)
    await db_session.flush()
    task = ProvisioningTask(
        id=uuid.uuid4(),
        user_id=employee.id,
        software_id=sw.id,
        action=ACTION_GRANT,
        target_type="seat",
        target_id=seat.id,
        source=SOURCE_REQUEST,
        execution_mode=EXEC_AUTO_API,
        status=STATUS_PENDING,
    )
    db_session.add(task)
    await db_session.commit()

    async def _failing_grant(self, ctx):
        return ProvisioningResult(ok=False, error_message="vendor 500")

    from app.integrations.drivers.manual_software import ManualSoftwareDriver
    monkeypatch.setattr(ManualSoftwareDriver, "grant_seat", _failing_grant)

    from app.domains.provisioning.service import ProvisioningService
    svc = ProvisioningService(db_session)
    await svc.execute_task(task, trace_id="t", actor_user_id=None)
    await db_session.commit()

    await db_session.refresh(seat)
    # Grant failure: seat stays pending, reason untouched
    assert seat.status == "pending"
    assert seat.reason is None or "failed" not in (seat.reason or "")


# ── execute_pending_batch per-(user, software) serialization ──────


@pytest.mark.asyncio
async def test_dispatcher_serializes_same_user_software_per_tick(
    db_session, employee_user, monkeypatch,
):
    """Multiple revoke tasks for the same (user, software) in one
    pending batch must NOT all fire in the same tick — vendor APIs
    rate-limit per user. The leading task fires; the rest defer to
    the next tick. Closes 2026-05-08 Zoom 429 cascade."""
    employee, _ = employee_user
    sw = await _make_software(db_session, name="ManySeats")
    integ = Integration(
        id=uuid.uuid4(), kind="software_provisioning", provider="manual",
        name="Manual", slug="manual-serialization", enabled=True,
        is_primary=False, config_public={},
    )
    db_session.add(integ)
    await db_session.flush()
    db_session.add(SoftwareConnector(
        id=uuid.uuid4(), software_id=sw.id, integration_id=integ.id,
        binding_config={}, is_active=True,
    ))
    plan = await _make_plan(db_session, sw.id)
    # 3 seats, 3 pending revoke tasks for same (user, software) — same
    # shape as the kevin.lu Zoom incident.
    seats = []
    tasks = []
    for i in range(3):
        s = SoftwareSeatAllocation(
            id=uuid.uuid4(),
            plan_id=plan.id,
            user_id=employee.id,
            status="revoked",
            allocated_at=datetime.now(UTC),
            revoked_at=datetime.now(UTC),
        )
        db_session.add(s)
        seats.append(s)
    await db_session.flush()
    for s in seats:
        t = await _make_revoke_task(db_session, employee.id, sw.id, s.id)
        tasks.append(t)
    await db_session.commit()

    call_log: list[uuid.UUID] = []

    async def _logging_revoke(self, ctx):
        call_log.append(uuid.uuid4())  # any tick-fire counts
        return ProvisioningResult(ok=True, external_ref="manual:done")

    from app.integrations.drivers.manual_software import ManualSoftwareDriver
    monkeypatch.setattr(ManualSoftwareDriver, "revoke_seat", _logging_revoke)

    from app.domains.provisioning.service import ProvisioningService
    svc = ProvisioningService(db_session)
    summary = await svc.execute_pending_batch(limit=50, trace_id="serial-test")
    await db_session.commit()

    # Only ONE driver call this tick — the other two deferred.
    assert len(call_log) == 1, (
        f"Expected 1 driver call (per-(user,software) serialization), "
        f"got {len(call_log)}"
    )
    assert summary["picked"] == 3
    assert summary["completed"] == 1
    assert summary["deferred_per_user_serial"] == 2

    # Two tasks remain pending for the next tick
    from sqlalchemy import select as _select
    remaining_pending = (
        await db_session.execute(
            _select(ProvisioningTask).where(
                ProvisioningTask.user_id == employee.id,
                ProvisioningTask.status == STATUS_PENDING,
            )
        )
    ).scalars().all()
    assert len(remaining_pending) == 2


@pytest.mark.asyncio
async def test_dispatcher_does_not_serialize_across_different_softwares(
    db_session, employee_user, monkeypatch,
):
    """Per-user but cross-software tasks have NO shared rate limit
    (different vendors). They must all fire in the same tick rather
    than artificially serialize and waste 30s ticks."""
    employee, _ = employee_user
    sw_a = await _make_software(db_session, name="A")
    sw_b = await _make_software(db_session, name="B")
    integ = Integration(
        id=uuid.uuid4(), kind="software_provisioning", provider="manual",
        name="Manual", slug="manual-cross-software", enabled=True,
        is_primary=False, config_public={},
    )
    db_session.add(integ)
    await db_session.flush()
    for sw_id in (sw_a.id, sw_b.id):
        db_session.add(SoftwareConnector(
            id=uuid.uuid4(), software_id=sw_id, integration_id=integ.id,
            binding_config={}, is_active=True,
        ))
    plan_a = await _make_plan(db_session, sw_a.id)
    plan_b = await _make_plan(db_session, sw_b.id)
    seat_a = await _make_active_seat(db_session, plan_a.id, employee.id)
    seat_b = await _make_active_seat(db_session, plan_b.id, employee.id)
    seat_a.status = "revoked"
    seat_b.status = "revoked"
    await db_session.flush()
    await _make_revoke_task(db_session, employee.id, sw_a.id, seat_a.id)
    await _make_revoke_task(db_session, employee.id, sw_b.id, seat_b.id)
    await db_session.commit()

    call_count = 0

    async def _logging_revoke(self, ctx):
        nonlocal call_count
        call_count += 1
        return ProvisioningResult(ok=True, external_ref="manual:done")

    from app.integrations.drivers.manual_software import ManualSoftwareDriver
    monkeypatch.setattr(ManualSoftwareDriver, "revoke_seat", _logging_revoke)

    from app.domains.provisioning.service import ProvisioningService
    svc = ProvisioningService(db_session)
    summary = await svc.execute_pending_batch(limit=50, trace_id="cross-sw")
    await db_session.commit()

    assert call_count == 2
    assert summary["completed"] == 2
    assert summary.get("deferred_per_user_serial", 0) == 0


@pytest.mark.asyncio
async def test_dispatcher_serialization_skips_manual_notify_tasks(
    db_session, employee_user, monkeypatch,
):
    """manual_notify tasks settle locally (audit log + notification)
    and don't make outbound vendor calls — they don't need
    serialization. Two manual_notify tasks for the same (user,
    software) should both run in the same tick."""
    employee, _ = employee_user
    sw = await _make_software(db_session, name="ManualNotifySoft")
    integ = Integration(
        id=uuid.uuid4(), kind="software_provisioning", provider="manual",
        name="Manual", slug="manual-notify-no-serial", enabled=True,
        is_primary=False, config_public={},
    )
    db_session.add(integ)
    await db_session.flush()
    db_session.add(SoftwareConnector(
        id=uuid.uuid4(), software_id=sw.id, integration_id=integ.id,
        binding_config={}, is_active=True,
    ))
    plan = await _make_plan(db_session, sw.id)

    for _ in range(2):
        seat = await _make_active_seat(db_session, plan.id, employee.id)
        seat.status = "revoked"
        await db_session.flush()
        task = ProvisioningTask(
            id=uuid.uuid4(),
            user_id=employee.id,
            software_id=sw.id,
            action=ACTION_REVOKE,
            target_type="seat",
            target_id=seat.id,
            source=SOURCE_MANUAL,
            execution_mode=EXEC_MANUAL_NOTIFY,
            status=STATUS_PENDING,
        )
        db_session.add(task)
    await db_session.commit()

    from app.domains.provisioning.service import ProvisioningService
    svc = ProvisioningService(db_session)
    summary = await svc.execute_pending_batch(limit=50, trace_id="no-serial")
    await db_session.commit()

    # Both tasks notified, none deferred
    assert summary["notified"] == 2
    assert summary.get("deferred_per_user_serial", 0) == 0


# ── Round-22 (2026-05-13) — connector raising AppException ──


@pytest.mark.asyncio
async def test_connector_raising_appexception_captures_structured_message(
    db_session, employee_user, monkeypatch,
):
    """Round-22 regression: when a software connector raises an
    ``AppException``-derived exception (e.g. credential-decrypt
    failure surfaces ``AppException(code="CREDENTIAL_DECRYPT_FAILED",
    message="...")``), the dispatcher used to record
    ``task.error_message = "connector raised exception: "`` (empty
    string after ``str(AppException)``) — admins reading the failed
    task row had no idea WHY. The Round-22 fix routes the format
    through ``format_exception_for_log`` which surfaces the
    ``.code`` + ``.message``.
    """
    from app.core.exceptions import AppException

    employee, _ = employee_user
    sw = await _make_software(db_session, name="AppExcSaaS")
    integ = Integration(
        id=uuid.uuid4(), kind="software_provisioning", provider="manual",
        name="Manual", slug="manual-appexception", enabled=True,
        is_primary=False, config_public={},
    )
    db_session.add(integ)
    await db_session.flush()
    db_session.add(SoftwareConnector(
        id=uuid.uuid4(), software_id=sw.id, integration_id=integ.id,
        binding_config={}, is_active=True,
    ))
    plan = await _make_plan(db_session, sw.id)
    seat = await _make_active_seat(db_session, plan.id, employee.id)
    seat.status = "revoked"
    seat.revoked_at = datetime.now(UTC)
    await db_session.flush()
    task = await _make_revoke_task(db_session, employee.id, sw.id, seat.id)
    await db_session.commit()

    async def _raising_revoke(self, ctx):
        raise AppException(
            code="CREDENTIAL_DECRYPT_FAILED",
            message="Could not decrypt SaaS API key — rotation needed.",
            status_code=500,
        )

    from app.integrations.drivers.manual_software import ManualSoftwareDriver
    monkeypatch.setattr(ManualSoftwareDriver, "revoke_seat", _raising_revoke)

    from app.domains.provisioning.service import ProvisioningService
    svc = ProvisioningService(db_session)
    updated = await svc.execute_task(
        task, trace_id="round22-appexception", actor_user_id=None,
    )
    await db_session.commit()

    assert updated.status == STATUS_FAILED
    # The pre-Round-22 sentinel — "connector raised exception: "
    # (empty body) — must never appear again.
    assert updated.error_message != "connector raised exception: "
    # The structured AppException fields must be present.
    assert "AppException" in updated.error_message
    assert "CREDENTIAL_DECRYPT_FAILED" in updated.error_message
    assert "Could not decrypt SaaS API key" in updated.error_message