"""The disposal → MDM-cleanup loop, and the departure that must not fire one.

The behaviour under test is a correction, so most of these assert an
absence. A Jamf administrator told us plainly that he removes a Mac when
the company stops owning it and at no other time — a returned laptop is
re-imaged and reassigned while staying enrolled the whole way through. So:

  * an employee leaving raises NO MDM work, even holding two enrolled Macs;
  * disposing of an asset whose serial is enrolled raises exactly one
    cleanup todo, routed to the MDM catalogue's owner;
  * disposing of anything the MDM has never heard of raises nothing, which
    is most disposals and must stay silent;
  * a request whose asset stops being disposed is withdrawn, because a
    card that outlives its reason gets a device deleted that the company
    still owns;
  * the reminder loop nudges weekly and widens its audience at three
    weeks, and stops the moment the todo closes.
"""
from __future__ import annotations

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

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.coordinators.mdm_device_retirement_coordinator import (
    CLEANUP_REASON_BUYBACK,
    REMINDER_INTERVAL_DAYS,
    due_cleanup_notifications,
    reminder_should_escalate,
    request_cleanup_for_disposed_asset,
    withdraw_stale_cleanups,
)
from app.domains.asset.models import (
    MDM_MATCH_MATCHED,
    MDM_STATE_ENROLLED,
    MDM_STATE_GONE,
    MDM_STATE_PENDING_CLEANUP,
    AssetItem,
    AssetSoftwareCatalog,
    MdmManagedDevice,
    SoftwareLicensePlan,
)
from app.domains.provisioning.models import (
    ACTION_REVOKE,
    EXEC_MANUAL_NOTIFY,
    SOURCE_ASSET_DISPOSAL,
    STATUS_CANCELLED,
    STATUS_NOTIFIED,
    TARGET_HARDWARE,
    ProvisioningTask,
    SoftwareConnector,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


async def _person(
    db: AsyncSession, email: str, status: str = "active",
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(), work_email=email, display_name=email.split("@")[0],
        employment_status=status, source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", country="cn",
    )
    db.add(p)
    await db.flush()
    return p


async def _mdm_catalogue(
    db: AsyncSession, *, owner: IdentityPerson | None,
) -> tuple[AssetSoftwareCatalog, Integration]:
    integ = Integration(
        id=uuid.uuid4(), kind="mdm", provider="jamf",
        name="Jamf", slug=f"jamf-{uuid.uuid4().hex[:6]}",
        enabled=True, is_primary=True, config_public={},
    )
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(), name="Jamf", asset_type="SaaS", status="active",
        owner_person_id=(owner.id if owner else None),
    )
    db.add_all([integ, sw])
    await db.flush()
    db.add(SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=sw.id, name="Jamf Seat",
        price=Decimal("0.00"), billing_cycle="yearly", status="active",
        max_seats=60,
    ))
    db.add(SoftwareConnector(
        software_id=sw.id, integration_id=integ.id, is_active=True,
        binding_config={},
    ))
    await db.flush()
    return sw, integ


async def _asset(
    db: AsyncSession, serial: str, *, holder: IdentityPerson | None = None,
) -> AssetItem:
    a = AssetItem(
        id=uuid.uuid4(), asset_tag=f"A-{serial}", serial_number=serial,
        status="assigned" if holder else "idle", category="laptop",
        assigned_to=(holder.id if holder else None),
    )
    db.add(a)
    await db.flush()
    return a


async def _device(
    db: AsyncSession, integ: Integration, serial: str,
    *, asset: AssetItem | None = None, holder: IdentityPerson | None = None,
) -> MdmManagedDevice:
    d = MdmManagedDevice(
        id=uuid.uuid4(), integration_id=integ.id, provider="jamf",
        device_type="computer", external_id=f"id-{serial}",
        serial_number=serial, serial_key=serial.lower(),
        device_name=f"MAC-{serial}", model="MacBook Pro", managed=True,
        asset_id=(asset.id if asset else None),
        assigned_person_id=(holder.id if holder else None),
        match_category=MDM_MATCH_MATCHED,
        enrollment_state=MDM_STATE_ENROLLED,
        first_seen_at=datetime.now(UTC), last_seen_at=datetime.now(UTC),
    )
    db.add(d)
    await db.flush()
    return d


# ──────────────────────────────────────────────────────────────────
# The absence: departures raise nothing
# ──────────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_offboarding_scan_raises_no_blocker_for_mdm_software(
    db_session: AsyncSession,
) -> None:
    """The gate, exercised through the real scanner.

    Even with a hand-created seat allocation on MDM-backed software — the
    shape migration 558 cleaned up, and the shape an admin could recreate
    — the leaver's scan must produce no software-seat blocker for it.
    """
    from app.domains.asset.models import SoftwareSeatAllocation
    from app.domains.offboarding.models import (
        RES_SOFTWARE_SEAT,
        OffboardingBlocker,
    )
    from app.domains.offboarding.service import OffboardingService

    leaver = await _person(db_session, "leaver@pingcap.com", "leaver_pending")
    sw, _integ = await _mdm_catalogue(db_session, owner=None)
    plan = (await db_session.execute(
        select(SoftwareLicensePlan).where(
            SoftwareLicensePlan.software_id == sw.id,
        )
    )).scalar_one()
    db_session.add(SoftwareSeatAllocation(
        id=uuid.uuid4(), plan_id=plan.id, user_id=leaver.id,
        status="active", allocated_at=datetime.now(UTC),
        reason="hand-created, the shape the gate must refuse",
    ))
    await db_session.flush()

    svc = OffboardingService(db_session)
    proc = await svc.start_process(
        leaver.id,
        (datetime.now(UTC) + timedelta(days=7)).date(),
        trace_id="t-mdm-gate",
        actor_user_id=None,
    )
    await db_session.flush()

    blockers = (await db_session.execute(
        select(OffboardingBlocker).where(
            OffboardingBlocker.process_id == proc.id,
            OffboardingBlocker.resource_type == RES_SOFTWARE_SEAT,
            OffboardingBlocker.software_id == sw.id,
        )
    )).scalars().all()
    assert blockers == [], (
        "an MDM enrolment is not access the leaver holds — the device "
        "stays enrolled through the handover"
    )


# ──────────────────────────────────────────────────────────────────
# The presence: disposal raises exactly one cleanup
# ──────────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_buyback_disposal_opens_one_cleanup_routed_to_the_owner(
    db_session: AsyncSession,
) -> None:
    owner = await _person(db_session, "mdmadmin@pingcap.com", "active")
    holder = await _person(db_session, "chris@pingcap.com", "left")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "GTQ90V4K34", holder=holder)
    device = await _device(
        db_session, integ, "gtq90v4k34", asset=asset, holder=holder,
    )

    result = await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="sold",
        operator_id=owner.id, last_holder_id=holder.id,
        via_buyback=True, trace_id="t-buyback",
    )
    assert result["tasks_created"] == 1

    await db_session.refresh(device)
    assert device.enrollment_state == MDM_STATE_PENDING_CLEANUP
    assert device.cleanup_reason == CLEANUP_REASON_BUYBACK
    assert device.cleanup_requested_at is not None
    assert device.cleanup_confirmed_at is None

    task = await db_session.scalar(
        select(ProvisioningTask).where(
            ProvisioningTask.id == device.cleanup_task_id,
        )
    )
    assert task is not None
    assert task.action == ACTION_REVOKE
    assert task.target_type == TARGET_HARDWARE
    assert task.target_id == device.id
    assert task.source == SOURCE_ASSET_DISPOSAL
    assert task.execution_mode == EXEC_MANUAL_NOTIFY
    # Created already notified: the generic dispatcher would announce this
    # with the software-seat template, which is the wrong sentence for a
    # laptop, and it only ever picks up PENDING rows.
    assert task.status == STATUS_NOTIFIED
    assert task.assignee_id == owner.id
    assert task.user_id == holder.id
    assert task.context_data["subject_role"] == "last_holder"
    assert task.context_data["holder_name"] == "chris"
    assert task.context_data["serial_number"] == "gtq90v4k34"


@pytest.mark.asyncio
async def test_disposal_of_unenrolled_asset_is_silent(
    db_session: AsyncSession,
) -> None:
    """Most disposals are not Apple hardware. They must raise nothing."""
    owner = await _person(db_session, "mdmadmin@pingcap.com", "active")
    await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "SOME-DELL-SERIAL")

    result = await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="scrapped",
        operator_id=owner.id, last_holder_id=None, trace_id="t-dell",
    )
    assert result == {"devices": 0, "tasks_created": 0, "skipped": []}
    assert (
        await db_session.execute(select(ProvisioningTask))
    ).scalars().all() == []


@pytest.mark.asyncio
async def test_serialless_asset_cannot_match_and_does_not_crash(
    db_session: AsyncSession,
) -> None:
    """A blank serial matches nothing rather than matching everything."""
    owner = await _person(db_session, "mdmadmin@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    await _device(db_session, integ, "realserial")
    asset = AssetItem(
        id=uuid.uuid4(), asset_tag="A-NOSERIAL", serial_number=None,
        status="idle", category="accessory",
    )
    db_session.add(asset)
    await db_session.flush()

    result = await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="scrapped",
        operator_id=owner.id, last_holder_id=None, trace_id="t-noserial",
    )
    assert result["devices"] == 0


@pytest.mark.asyncio
async def test_pool_device_disposal_names_no_holder(
    db_session: AsyncSession,
) -> None:
    """A conference-room iPad has no last holder.

    ``user_id`` is NOT NULL on the task, so the operator fills the slot —
    and ``subject_role`` records that the name is the operator's, so no
    surface renders an admin as the person who carried the device.
    """
    admin = await _person(db_session, "itadmin@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=None)
    asset = await _asset(db_session, "POOL1")
    device = await _device(db_session, integ, "pool1", asset=asset)

    await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="scrapped",
        operator_id=admin.id, last_holder_id=None, trace_id="t-pool",
    )

    await db_session.refresh(device)
    task = await db_session.scalar(
        select(ProvisioningTask).where(
            ProvisioningTask.id == device.cleanup_task_id,
        )
    )
    assert task.user_id == admin.id
    assert task.context_data["subject_role"] == "operator"
    assert task.context_data["holder_name"] is None
    # No live catalogue owner → the notification fans out to admins.
    assert task.assignee_id is None


@pytest.mark.asyncio
async def test_departed_catalogue_owner_is_not_assigned(
    db_session: AsyncSession,
) -> None:
    """A task assigned to someone who left looks handled and never is.

    Leaving the assignee empty is what routes the card to the IT-admin
    fan-out instead of into a mailbox nobody reads.
    """
    gone_owner = await _person(db_session, "exadmin@pingcap.com", "left")
    _sw, integ = await _mdm_catalogue(db_session, owner=gone_owner)
    asset = await _asset(db_session, "OWNERGONE")
    device = await _device(db_session, integ, "ownergone", asset=asset)

    await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="sold",
        operator_id=gone_owner.id, last_holder_id=None, trace_id="t-owner",
    )

    await db_session.refresh(device)
    task = await db_session.scalar(
        select(ProvisioningTask).where(
            ProvisioningTask.id == device.cleanup_task_id,
        )
    )
    assert task.assignee_id is None


@pytest.mark.asyncio
async def test_second_disposal_does_not_duplicate_the_card(
    db_session: AsyncSession,
) -> None:
    owner = await _person(db_session, "mdmadmin@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "DUP1")
    await _device(db_session, integ, "dup1", asset=asset)

    first = await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="sold",
        operator_id=owner.id, last_holder_id=None, trace_id="t-1",
    )
    second = await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="sold",
        operator_id=owner.id, last_holder_id=None, trace_id="t-2",
    )
    assert first["tasks_created"] == 1
    assert second["tasks_created"] == 0
    assert second["skipped"][0]["reason"] == "cleanup already open"
    tasks = (await db_session.execute(select(ProvisioningTask))).scalars().all()
    assert len(tasks) == 1


@pytest.mark.asyncio
async def test_gone_devices_are_not_re_requested(
    db_session: AsyncSession,
) -> None:
    """Nothing to remove from a tenant that no longer has it."""
    owner = await _person(db_session, "mdmadmin@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "OLD1")
    device = await _device(db_session, integ, "old1", asset=asset)
    device.enrollment_state = MDM_STATE_GONE
    device.gone_at = datetime.now(UTC)
    await db_session.flush()

    result = await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="scrapped",
        operator_id=owner.id, last_holder_id=None, trace_id="t-gone",
    )
    assert result["devices"] == 0


# ──────────────────────────────────────────────────────────────────
# Withdrawal, reminders, escalation
# ──────────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_cleanup_is_withdrawn_when_the_asset_is_no_longer_disposed(
    db_session: AsyncSession,
) -> None:
    """The card must not outlive its reason.

    A request standing against an asset the company still owns would have
    an administrator delete a live machine — the worst outcome this flow
    can produce. The queue re-checks its own premise rather than trusting
    every future writer of the asset status to remember it exists.
    """
    owner = await _person(db_session, "mdmadmin@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "REVERT1")
    device = await _device(db_session, integ, "revert1", asset=asset)
    await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="sold",
        operator_id=owner.id, last_holder_id=None, trace_id="t-rev",
    )
    task_id = device.cleanup_task_id
    # While the asset really is disposed, nothing is withdrawn.
    asset.status = "disposed"
    await db_session.flush()
    assert await withdraw_stale_cleanups(db_session) == 0

    asset.status = "idle"
    await db_session.flush()
    withdrawn = await withdraw_stale_cleanups(db_session)

    assert withdrawn == 1
    await db_session.refresh(device)
    assert device.enrollment_state == MDM_STATE_ENROLLED
    assert device.cleanup_task_id is None
    assert device.cleanup_reason is None
    task = await db_session.scalar(
        select(ProvisioningTask).where(ProvisioningTask.id == task_id)
    )
    assert task.status == STATUS_CANCELLED


@pytest.mark.asyncio
async def test_unlinked_device_cleanup_is_never_withdrawn(
    db_session: AsyncSession,
) -> None:
    """No asset link is not evidence the disposal was undone.

    Withdrawing on "no asset found" would silently cancel every cleanup
    for hardware the ledger has not matched to an asset row yet — which,
    on a fleet where a third of devices are unmatched, is most of them.
    """
    owner = await _person(db_session, "mdmadmin@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "UNLINKED1")
    device = await _device(db_session, integ, "unlinked1", asset=None)
    # Serial matches the asset, so the request is raised, but the ledger
    # row carries no asset_id (the nightly sync has not linked it yet).
    device.serial_key = "unlinked1"
    asset.serial_number = "UNLINKED1"
    await db_session.flush()
    await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="sold",
        operator_id=owner.id, last_holder_id=None, trace_id="t-unlinked",
    )
    await db_session.refresh(device)
    assert device.enrollment_state == MDM_STATE_PENDING_CLEANUP

    asset.status = "idle"
    await db_session.flush()

    assert await withdraw_stale_cleanups(db_session) == 0
    await db_session.refresh(device)
    assert device.enrollment_state == MDM_STATE_PENDING_CLEANUP


@pytest.mark.asyncio
async def test_first_notice_then_weekly_reminder(
    db_session: AsyncSession,
) -> None:
    """Never-announced is a first card; a week stale is a nudge."""
    owner = await _person(db_session, "mdmadmin@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "NAG1")
    device = await _device(db_session, integ, "nag1", asset=asset)
    await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="sold",
        operator_id=owner.id, last_holder_id=None, trace_id="t-nag",
    )

    due = await due_cleanup_notifications(db_session)
    assert [(d.id, is_reminder) for d, _t, is_reminder in due] == [
        (device.id, False),
    ]

    # Just announced → quiet until the interval elapses.
    device.cleanup_last_reminded_at = datetime.now(UTC)
    await db_session.flush()
    assert await due_cleanup_notifications(db_session) == []

    # A week later, still enrolled → nudge.
    device.cleanup_last_reminded_at = datetime.now(UTC) - timedelta(
        days=REMINDER_INTERVAL_DAYS + 1,
    )
    await db_session.flush()
    due = await due_cleanup_notifications(db_session)
    assert [is_reminder for _d, _t, is_reminder in due] == [True]


@pytest.mark.asyncio
async def test_reminder_stops_once_the_todo_is_closed(
    db_session: AsyncSession,
) -> None:
    owner = await _person(db_session, "mdmadmin@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "DONE1")
    device = await _device(db_session, integ, "done1", asset=asset)
    await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="sold",
        operator_id=owner.id, last_holder_id=None, trace_id="t-done",
    )
    task = await db_session.scalar(
        select(ProvisioningTask).where(
            ProvisioningTask.id == device.cleanup_task_id,
        )
    )
    task.status = "completed"
    await db_session.flush()

    assert await due_cleanup_notifications(db_session) == []


@pytest.mark.asyncio
async def test_naive_timestamp_does_not_kill_the_reminder_loop(
    db_session: AsyncSession,
) -> None:
    """MySQL/TiDB hand back naive datetimes for tz-aware columns.

    Comparing one to an aware cutoff raises instead of returning False,
    which would take the whole loop down rather than skip a row — the
    same class of bug that once silently stalled seat drift cleanup.
    """
    owner = await _person(db_session, "mdmadmin@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "NAIVE1")
    device = await _device(db_session, integ, "naive1", asset=asset)
    await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="sold",
        operator_id=owner.id, last_holder_id=None, trace_id="t-naive",
    )
    device.cleanup_last_reminded_at = datetime(2020, 1, 1)  # noqa: DTZ001
    await db_session.flush()

    due = await due_cleanup_notifications(db_session)
    assert [is_reminder for _d, _t, is_reminder in due] == [True]


@pytest.mark.asyncio
async def test_notify_enqueues_durably_without_talking_to_the_vendor(
    db_session: AsyncSession,
) -> None:
    """Phase one writes outbox rows and the stamp, and delivers nothing.

    An external round-trip inside this loop's open transaction is the
    dropped-connection shape this platform has paid for before, and a
    delivery that happened before a crash-before-commit would re-send the
    same card on the next tick. Both are avoided by the same split, so
    this test asserts the split itself: rows PENDING, stamp set, nothing
    sent.
    """
    from app.core.coordinators.mdm_device_retirement_coordinator import (
        notify_cleanup,
    )
    from app.notifications.models import NotificationOutbox

    owner = await _person(db_session, "mdmadmin@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "QUEUE1")
    device = await _device(db_session, integ, "queue1", asset=asset)
    await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="sold",
        operator_id=owner.id, last_holder_id=None, trace_id="t-q",
    )
    due = await due_cleanup_notifications(db_session)
    dev, task, is_reminder = due[0]

    result = await notify_cleanup(
        db_session, device=dev, task=task, trace_id="t-q",
        is_reminder=is_reminder,
    )

    assert result.problem is None
    assert len(result.queued) == 1
    outbox, _env = result.queued[0]
    assert outbox.recipient_person_id == owner.id
    # Durably recorded, not yet delivered — the retry worker is the
    # backstop for whatever phase two misses.
    row = await db_session.scalar(
        select(NotificationOutbox).where(NotificationOutbox.id == outbox.id)
    )
    assert row is not None
    assert row.status == "pending"
    # The stamp lands in the SAME transaction as the outbox row.
    await db_session.refresh(device)
    assert device.cleanup_last_reminded_at is not None
    # And with the stamp set, the device drops out of the due list until
    # the interval elapses — no second card this week.
    assert await due_cleanup_notifications(db_session) == []


@pytest.mark.asyncio
async def test_no_live_recipient_is_reported_and_retried_weekly(
    db_session: AsyncSession,
) -> None:
    """Nobody to tell is a gap the run has to report.

    Silence here used to be total: no card, no error, no stamp — so the
    device came back due on every tick of every working day while the run
    reported success. Now the attempt is stamped (weekly retry, not ten a
    day) and the gap is returned so the run records it.
    """
    from app.core.coordinators.mdm_device_retirement_coordinator import (
        notify_cleanup,
    )

    # No catalogue owner and no admin rows in this test's data.
    operator = await _person(db_session, "ops@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=None)
    asset = await _asset(db_session, "NOBODY1")
    device = await _device(db_session, integ, "nobody1", asset=asset)
    await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="sold",
        operator_id=operator.id, last_holder_id=None, trace_id="t-nb",
    )
    dev, task, is_reminder = (await due_cleanup_notifications(db_session))[0]

    result = await notify_cleanup(
        db_session, device=dev, task=task, trace_id="t-nb",
        is_reminder=is_reminder,
    )

    assert result.queued == []
    # Surfaced, not swallowed: a cleanup nobody can be told about is a
    # real gap, and the run must not look like it had nothing to do.
    assert result.problem is not None
    assert "no eligible recipient" in result.problem
    await db_session.refresh(device)
    # Stamped anyway — the stamp records the ATTEMPT. Without it this
    # device is due on every tick of every working day until somebody
    # gives the catalogue an owner; weekly is the honest cadence.
    assert device.cleanup_last_reminded_at is not None
    assert await due_cleanup_notifications(db_session) == []


@pytest.mark.asyncio
async def test_escalation_widens_the_audience_after_three_weeks(
    db_session: AsyncSession,
) -> None:
    owner = await _person(db_session, "mdmadmin@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "ESC1")
    device = await _device(db_session, integ, "esc1", asset=asset)
    await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="sold",
        operator_id=owner.id, last_holder_id=None, trace_id="t-esc",
    )

    assert reminder_should_escalate(device) is False
    device.cleanup_requested_at = datetime.now(UTC) - timedelta(days=22)
    assert reminder_should_escalate(device) is True


@pytest.mark.asyncio
async def test_a_reopened_todo_speaks_in_the_reminder_voice(
    db_session: AsyncSession,
) -> None:
    """Never a first notice — and the marker decides that and only that.

    A todo the sync re-raised has already been through a cycle: somebody
    was told, or somebody said they had done it. Opening with "please
    remove this device" pretends neither happened.

    Timing is the stamp's job, not the marker's. The sync clears the stamp
    when it overturns a claim, which is what makes the next tick speak;
    the marker is permanent and would, if it also waived the interval,
    produce a card every tick of every working day forever — see
    ``test_a_reopened_todo_is_chased_once_not_every_tick``.
    """
    owner = await _person(db_session, "mdmadmin@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "REOPEN1")
    device = await _device(db_session, integ, "reopen1", asset=asset)
    await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="sold",
        operator_id=owner.id, last_holder_id=None, trace_id="t-reopen",
    )
    task = await db_session.scalar(
        select(ProvisioningTask).where(
            ProvisioningTask.id == device.cleanup_task_id,
        )
    )
    # The state the sync leaves behind after overturning a claim: a task
    # carrying the provenance, and a cleared stamp.
    task.context_data = {
        **(task.context_data or {}),
        "reopened_from_task_id": str(uuid.uuid4()),
        "reopened_reason": "marked done, but the MDM still reports it",
    }
    device.cleanup_last_reminded_at = None
    await db_session.flush()

    due = await due_cleanup_notifications(db_session)

    assert len(due) == 1
    _dev_row, _task, is_reminder = due[0]
    assert is_reminder is True, (
        "a re-raised todo must not open as a first notice"
    )

    # And once spoken, it waits out the interval like any other — the
    # marker is still set, and must not keep it due.
    device.cleanup_last_reminded_at = datetime.now(UTC)
    await db_session.flush()
    assert await due_cleanup_notifications(db_session) == []


@pytest.mark.asyncio
async def test_english_card_has_no_chinese_left_in_it(
    db_session: AsyncSession,
) -> None:
    """The two producer-supplied lines resolve per RECIPIENT, like the rest.

    ``_localized_context`` exists because a production render of the
    45-day outbox found English cards ending in the producer's literal
    Chinese. This producer walked straight back into it: the template has
    ``body_text_en`` and ``body_text_ja``, but ``disposal_label`` and
    ``holder_line`` were plain Chinese strings, so two of the six bullet
    lines in every English and Japanese card came out Chinese. Most of
    this workforce reads the English one.

    Asserted against the RESOLVED context rather than the helpers'
    return types: a future author may reasonably change the shape, and
    what must not change is that an English reader gets English.
    """
    from app.notifications.service import _localized_context

    owner = await _person(db_session, "enreader@pingcap.com", "active")
    holder = await _person(db_session, "holder@pingcap.com", "left")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "ENCARD1")
    await _device(db_session, integ, "encard1", asset=asset)
    await request_cleanup_for_disposed_asset(
        db_session, asset=asset, disposal_type="buyback",
        operator_id=owner.id, last_holder_id=holder.id, trace_id="t-en",
    )
    due = await due_cleanup_notifications(db_session)
    dev, task, is_reminder = due[0]

    from app.core.coordinators.mdm_device_retirement_coordinator import (
        notify_cleanup,
    )

    result = await notify_cleanup(
        db_session, device=dev, task=task, trace_id="t-en",
        is_reminder=is_reminder,
    )
    assert result.problem is None
    _outbox, env = result.queued[0]

    def _cjk(text: str) -> str:
        return "".join(c for c in text if "一" <= c <= "鿿")

    en = _localized_context(env.context, "en")
    for field in ("disposal_label", "holder_line"):
        assert _cjk(str(en[field])) == "", (
            f"{field} rendered Chinese into an English card: {en[field]!r}"
        )
    # The English copy is real copy, not the enum value passed through.
    assert "uyback" in str(en["disposal_label"])
    # Japanese resolves too, and to something OTHER than the English —
    # a helper that returned the same string for both would satisfy the
    # CJK assertion above while shipping English to a Japanese reader.
    ja = _localized_context(env.context, "ja")
    assert str(ja["disposal_label"]) != str(en["disposal_label"])
    # And the Chinese reader is unaffected by the change.
    zh = _localized_context(env.context, "zh-CN")
    assert _cjk(str(zh["disposal_label"])) != ""

    # One tenant, one spelling. The title used ``provider.upper()`` while
    # the body interpolated the raw column, so a single card said 「JAMF」
    # and 「jamf」 four lines apart.
    assert en["provider"] == "Jamf"
    assert "Jamf" in env.title


@pytest.mark.asyncio
async def test_an_untranslated_disposal_type_still_says_something_true(
    db_session: AsyncSession,
) -> None:
    """An enum value nobody has translated renders verbatim, in every language.

    The alternative — falling back to a generic 「资产处置」 — is quietly
    less true than showing the raw word, and it hides from whoever added
    the enum that they never added its copy.
    """
    from app.core.coordinators.mdm_device_retirement_coordinator import (
        _disposal_label,
    )
    from app.notifications.service import resolve_card_text

    label = _disposal_label("repossessed_by_vendor")
    for lang in ("zh-CN", "en", "ja"):
        assert resolve_card_text(label, lang) == "repossessed_by_vendor"


@pytest.mark.asyncio
async def test_reconciler_raises_the_cleanup_the_hook_never_did(
    db_session: AsyncSession,
) -> None:
    """The loop has to converge in BOTH directions.

    ``withdraw_stale_cleanups`` retracts a request whose asset came back.
    Nothing opened one for an asset that was disposed while Helix was not
    looking — so every pre-feature buyback, every hook whose
    ``best_effort`` savepoint ate a DB error, and every writer that sets a
    disposed status outside ``AssetService`` left a sold Mac managed
    forever, silently. This is that backstop.
    """
    from app.core.coordinators.mdm_device_retirement_coordinator import (
        reconcile_disposed_assets_without_cleanup,
    )

    owner = await _person(db_session, "mdmowner-rec@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "BACKLOG1")
    device = await _device(db_session, integ, "backlog1", asset=asset)
    # The state the hook would have produced — minus the hook. This is a
    # disposal that predates the feature.
    asset.status = "disposed"
    asset.disposal_type = "sold"
    device.asset_id = asset.id
    await db_session.flush()

    out = await reconcile_disposed_assets_without_cleanup(
        db_session, trace_id="t-rec",
    )
    assert out["assets"] == 1
    assert out["tasks_created"] == 1
    assert out["unroutable"] == 0

    await db_session.refresh(device)
    assert device.enrollment_state == MDM_STATE_PENDING_CLEANUP
    assert device.cleanup_reason == "sold"
    assert device.cleanup_task_id is not None

    # Idempotent: a second pass must not put a second card in front of
    # the same administrator for one machine.
    again = await reconcile_disposed_assets_without_cleanup(
        db_session, trace_id="t-rec-2",
    )
    assert again["tasks_created"] == 0

    # And it is audited as a reconciler finding, not as a disposal that
    # just happened — the distinction an auditor needs to explain why a
    # cleanup appeared eight months after the sale.
    from app.audit.models import AuditLog

    actions = (
        await db_session.execute(
            select(AuditLog.action).where(
                AuditLog.resource_id == str(asset.id),
            )
        )
    ).scalars().all()
    assert "mdm.cleanup.reconciled" in actions


@pytest.mark.asyncio
async def test_reconciler_leaves_live_assets_and_gone_devices_alone(
    db_session: AsyncSession,
) -> None:
    """The predicate is narrow on both sides.

    A device the company still owns must never be queued for removal —
    that is an administrator deleting a machine still in service. And a
    device already gone from the tenant has nothing left to remove.
    """
    from app.core.coordinators.mdm_device_retirement_coordinator import (
        reconcile_disposed_assets_without_cleanup,
    )

    owner = await _person(db_session, "mdmowner-neg@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)

    live = await _asset(db_session, "LIVE1")
    live_dev = await _device(db_session, integ, "live1", asset=live)
    live_dev.asset_id = live.id

    sold = await _asset(db_session, "SOLDGONE1")
    gone_dev = await _device(db_session, integ, "soldgone1", asset=sold)
    sold.status = "disposed"
    sold.disposal_type = "sold"
    gone_dev.asset_id = sold.id
    gone_dev.enrollment_state = MDM_STATE_GONE
    await db_session.flush()

    out = await reconcile_disposed_assets_without_cleanup(
        db_session, trace_id="t-neg",
    )
    assert out["assets"] == 0
    assert out["tasks_created"] == 0

    await db_session.refresh(live_dev)
    await db_session.refresh(gone_dev)
    assert live_dev.enrollment_state == MDM_STATE_ENROLLED
    assert gone_dev.enrollment_state == MDM_STATE_GONE


@pytest.mark.asyncio
async def test_due_list_is_oldest_waiting_first(
    db_session: AsyncSession,
) -> None:
    """Ordering is what makes the burst cap fair rather than starving.

    The notify tick sends a bounded prefix of this list. Unordered, the
    devices the query happened to return first would be spoken about
    every tick while the tail is never reached — and the run summary
    would report a healthy "5 sent" forever.
    """
    owner = await _person(db_session, "mdmowner-ord@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    now = datetime.now(UTC)

    # Created in an order the expected result must PERMUTE. With the rows
    # inserted in their final order the assertion passes with the sort
    # deleted — the query returns insertion order and the test proves
    # nothing. (Caught by mutation-verifying this very test.)
    made: list[tuple[str, timedelta | None]] = [
        ("ORD-MID", timedelta(days=9)),
        ("ORD-NEW", None),          # never notified → must come first
        ("ORD-OLD", timedelta(days=30)),
    ]
    for tag, since in made:
        asset = await _asset(db_session, tag)
        dev = await _device(db_session, integ, tag, asset=asset)
        await request_cleanup_for_disposed_asset(
            db_session, asset=asset, disposal_type="sold",
            operator_id=owner.id, last_holder_id=None, trace_id=f"t-{tag}",
        )
        if since is not None:
            dev.cleanup_last_reminded_at = now - since
        await db_session.flush()

    due = await due_cleanup_notifications(db_session, now=now)
    order = [d.serial_number for d, _t, _r in due]
    assert order == ["ORD-NEW", "ORD-OLD", "ORD-MID"], order


@pytest.mark.asyncio
async def test_a_backlog_card_does_not_claim_the_sale_happened_today(
    db_session: AsyncSession,
) -> None:
    """``days_open`` measures the REQUEST, and the copy has to say so.

    Until the reconciler existed, ``cleanup_requested_at`` and the
    disposal were the same moment, so 「这台设备已处置 N 天」 was true by
    construction. A backlog item raised tonight for a Mac sold in January
    breaks that: the request is 0 days old and the sale is 200+. The card
    now reports the request's age as the request's age, and carries the
    real disposal date as its own line.
    """
    from app.core.coordinators.mdm_device_retirement_coordinator import (
        notify_cleanup,
        reconcile_disposed_assets_without_cleanup,
    )
    from app.notifications.service import _localized_context

    owner = await _person(db_session, "mdmowner-old@pingcap.com", "active")
    _sw, integ = await _mdm_catalogue(db_session, owner=owner)
    asset = await _asset(db_session, "OLDSALE1")
    device = await _device(db_session, integ, "oldsale1", asset=asset)
    long_ago = datetime.now(UTC) - timedelta(days=213)
    asset.status = "disposed"
    asset.disposal_type = "sold"
    asset.disposed_at = long_ago
    device.asset_id = asset.id
    await db_session.flush()

    await reconcile_disposed_assets_without_cleanup(
        db_session, trace_id="t-oldsale",
    )
    due = await due_cleanup_notifications(db_session)
    dev, task, _is_reminder = due[0]
    result = await notify_cleanup(
        db_session, device=dev, task=task, trace_id="t-oldsale",
        is_reminder=True,
    )
    _outbox, env = result.queued[0]
    ctx = _localized_context(env.context, "zh-CN")

    # The request is new; the sale is not. Both are stated, neither is
    # stated as the other.
    assert ctx["days_open"] == "0"
    assert ctx["disposed_on"] == long_ago.date().isoformat()
    assert "已处置 0 天" not in env.title
    assert "清理待办已开" in env.title

    # A task raised before this key existed renders a dash, never today.
    task.context_data = {
        k: v for k, v in (task.context_data or {}).items()
        if k != "disposed_at"
    }
    await db_session.flush()
    dev.cleanup_last_reminded_at = None
    await db_session.flush()
    again = await notify_cleanup(
        db_session, device=dev, task=task, trace_id="t-oldsale-2",
        is_reminder=True,
    )
    ctx2 = _localized_context(again.queued[0][1].context, "zh-CN")
    assert ctx2["disposed_on"] == "—"