"""Migration 461 repairs exactly the rows the leaver flip-flop wrote.

Runs the migration's own SQL against the real MySQL test schema (the
statements use MySQL multi-table ``UPDATE ... JOIN`` + ``UTC_TIMESTAMP()``,
so SQLite can't stand in) and pins both halves plus idempotency.
"""
from __future__ import annotations

import importlib.util
import sys
import uuid
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from pathlib import Path

import pytest
import sqlalchemy as sa

from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.domains.offboarding.models import OffboardingBlocker, OffboardingProcess
from app.domains.provisioning.models import ProvisioningTask
from app.identity.person_models import IdentityPerson

SERVER_DIR = Path(__file__).resolve().parents[2]
MIGRATION_461 = (
    SERVER_DIR
    / "alembic/versions/461_repair_stale_leaver_revoke_orphans.py"
)


def _load_migration():
    spec = importlib.util.spec_from_file_location(
        "migration_461_leaver_repair", MIGRATION_461,
    )
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


async def _seed(db, *, task_status: str, blocker_resolution: str, notes: str):
    person = 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="Repair Target",
        employment_status="active",
        termination_date=None,
    )
    db.add(person)
    await db.flush()
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(), name=f"SW-{uuid.uuid4().hex[:6]}",
        asset_type="software", status="active",
        offboarding_policy="auto_revoke",
    )
    db.add(sw)
    await db.flush()
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=sw.id, name="standard",
        price=Decimal("1.00"), billing_cycle="monthly",
    )
    db.add(plan)
    await db.flush()
    seat = SoftwareSeatAllocation(
        id=uuid.uuid4(), plan_id=plan.id, user_id=person.id,
        status="revoked", allocated_at=datetime.now(UTC),
        revoked_at=datetime.now(UTC),
    )
    db.add(seat)
    proc = OffboardingProcess(
        id=uuid.uuid4(), person_id=person.id,
        leaver_date=date.today() + timedelta(days=5), status="blocked",
    )
    db.add(proc)
    await db.flush()
    task = ProvisioningTask(
        id=uuid.uuid4(), user_id=person.id, software_id=sw.id,
        action="revoke", target_type="seat", target_id=seat.id,
        source="offboarding", execution_mode="manual_notify",
        status=task_status,
    )
    db.add(task)
    await db.flush()
    db.add(OffboardingBlocker(
        id=uuid.uuid4(), process_id=proc.id,
        resource_type="software_seat", resource_id=seat.id,
        label="seat", software_id=sw.id,
        resolution=blocker_resolution, notes=notes,
        provisioning_task_id=task.id,
    ))
    await db.flush()
    return person, proc, task


@pytest.mark.asyncio
async def test_cancels_orphan_and_restores_leaver(db_session):
    module = _load_migration()
    person, proc, task = await _seed(
        db_session,
        task_status="pending",
        blocker_resolution="skipped",
        notes="[stale_after_rescan] resource no longer present",
    )
    await db_session.flush()

    await db_session.execute(
        module._CANCEL_ORPHANS, {"note": module._CANCEL_NOTE},
    )
    await db_session.execute(module._RESTORE_LEAVERS)

    refreshed_task = await db_session.scalar(
        sa.select(ProvisioningTask).where(ProvisioningTask.id == task.id),
    )
    await db_session.refresh(refreshed_task)
    assert refreshed_task.status == "cancelled"
    assert refreshed_task.completed_at is not None
    assert "461" in (refreshed_task.error_message or "")

    refreshed_person = await db_session.get(IdentityPerson, person.id)
    await db_session.refresh(refreshed_person)
    assert refreshed_person.employment_status == "leaver_pending"
    assert refreshed_person.termination_date == proc.leaver_date, (
        "termination_date must come back from the open process's leaver_date "
        "— process_advance needs it to flip the person to 'left'"
    )


@pytest.mark.asyncio
async def test_is_idempotent(db_session):
    module = _load_migration()
    _, _, task = await _seed(
        db_session,
        task_status="pending",
        blocker_resolution="skipped",
        notes="[stale_after_rescan] resource no longer present",
    )
    await db_session.execute(
        module._CANCEL_ORPHANS, {"note": module._CANCEL_NOTE},
    )
    await db_session.execute(module._RESTORE_LEAVERS)
    second = await db_session.execute(
        module._CANCEL_ORPHANS, {"note": module._CANCEL_NOTE},
    )
    assert second.rowcount == 0
    third = await db_session.execute(module._RESTORE_LEAVERS)
    assert third.rowcount == 0
    assert task is not None


@pytest.mark.asyncio
async def test_leaves_live_task_and_in_progress_alone(db_session):
    module = _load_migration()
    # Blocker still pending → the task is the live one, must survive.
    _, _, live = await _seed(
        db_session, task_status="pending",
        blocker_resolution="pending", notes=None,
    )
    # Stale blocker but the dispatcher owns the row.
    _, _, inflight = await _seed(
        db_session, task_status="in_progress",
        blocker_resolution="skipped",
        notes="[stale_after_rescan] gone",
    )
    await db_session.execute(
        module._CANCEL_ORPHANS, {"note": module._CANCEL_NOTE},
    )

    for t in (live, inflight):
        await db_session.refresh(t)
    assert live.status == "pending"
    assert inflight.status == "in_progress"