"""HTTP surface of the MDM device ledger.

Covers the parts of a read page that are easy to get subtly wrong and
expensive to notice: that the KPI strip counts the whole fleet rather
than the filtered page, that retired devices stay out of the default view
without becoming unreachable, that a dangling holder id renders as an
absence instead of a 500, and that confirming a cleanup is refused on a
device with nothing to confirm.

Also pins the end-to-end path the whole feature exists for: disposing an
asset whose serial is enrolled leaves a cleanup todo behind, in the same
transaction as the disposal.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime

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

from app.domains.asset.models import (
    MDM_MATCH_LEAVER_RESIDUAL,
    MDM_MATCH_MATCHED,
    MDM_MATCH_UNOWNED,
    MDM_STATE_ENROLLED,
    MDM_STATE_GONE,
    MDM_STATE_PENDING_CLEANUP,
    AssetItem,
    MdmManagedDevice,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration

BASE = "/api/v1/assets/mdm-devices"


async def _integration(db: AsyncSession) -> 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={},
    )
    db.add(integ)
    await db.flush()
    return integ


async def _device(
    db: AsyncSession,
    integ: Integration,
    serial: str,
    *,
    state: str = MDM_STATE_ENROLLED,
    category: str = MDM_MATCH_UNOWNED,
    person_id: uuid.UUID | None = None,
    asset_id: uuid.UUID | 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", os_version="14.5",
        managed=True, match_category=category, enrollment_state=state,
        assigned_person_id=person_id, asset_id=asset_id,
        first_seen_at=datetime.now(UTC), last_seen_at=datetime.now(UTC),
    )
    db.add(d)
    await db.flush()
    return d


@pytest.mark.asyncio
async def test_meta_counts_the_fleet_and_reports_write_capability(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
) -> None:
    integ = await _integration(db_session)
    await _device(db_session, integ, "M1", category=MDM_MATCH_MATCHED)
    await _device(
        db_session, integ, "M2", category=MDM_MATCH_LEAVER_RESIDUAL,
    )
    await _device(db_session, integ, "M3", state=MDM_STATE_GONE)
    await db_session.commit()

    r = await client.get(f"{BASE}/meta", headers=admin_headers)
    assert r.status_code == 200
    data = r.json()["data"]

    assert data["can_manage"] is True
    assert data["integration_configured"] is True
    assert data["enrollment_states"] == [
        MDM_STATE_ENROLLED, MDM_STATE_PENDING_CLEANUP, MDM_STATE_GONE,
    ]
    s = data["summary"]
    assert s["total"] == 3
    assert s["enrolled"] == 2
    assert s["gone"] == 1
    assert s["matched"] == 1
    assert s["leaver_residual"] == 1
    # A retired device is out of the fleet, so it must not drag the
    # asset-coverage number down forever.
    assert s["unlinked"] == 2


@pytest.mark.asyncio
async def test_every_kpi_counts_exactly_the_rows_its_click_shows(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
) -> None:
    """The house rule, enforced against the real endpoints.

    Each KPI tile applies a filter when clicked, and the list it lands on
    defaults to ``include_gone=false``. A counter computed over retired
    devices too would report a number the click then hides — which is how
    a ledger stops being believed. Caught in review with a retired
    ``unowned`` device inflating that tile by one.
    """
    integ = await _integration(db_session)
    await _device(db_session, integ, "K1", category=MDM_MATCH_MATCHED)
    await _device(
        db_session, integ, "K2", category=MDM_MATCH_LEAVER_RESIDUAL,
    )
    await _device(db_session, integ, "K3")  # unowned, enrolled
    # Retired devices carrying each category — these must not be counted.
    await _device(
        db_session, integ, "K4", state=MDM_STATE_GONE,
        category=MDM_MATCH_UNOWNED,
    )
    await _device(
        db_session, integ, "K5", state=MDM_STATE_GONE,
        category=MDM_MATCH_LEAVER_RESIDUAL,
    )
    await db_session.commit()

    s = (await client.get(f"{BASE}/meta", headers=admin_headers)).json()[
        "data"
    ]["summary"]

    # Every tile's count must equal the length of the list its click opens.
    for card, params in (
        ("enrolled", "enrollment_state=enrolled"),
        ("pending_cleanup", "enrollment_state=pending_cleanup"),
        ("leaver_residual", f"match_category={MDM_MATCH_LEAVER_RESIDUAL}"),
        ("unlinked", "unlinked=true"),
    ):
        listed = await client.get(f"{BASE}?{params}", headers=admin_headers)
        assert listed.json()["meta"]["total"] == s[card], (
            f"KPI '{card}' counts {s[card]} but its own filter returns "
            f"{listed.json()['meta']['total']}"
        )

    # And the retired rows are still counted where they belong.
    assert s["gone"] == 2
    assert s["total"] == 5


@pytest.mark.asyncio
async def test_retired_devices_are_hidden_by_default_but_reachable(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
) -> None:
    integ = await _integration(db_session)
    await _device(db_session, integ, "LIVE1")
    await _device(db_session, integ, "DEAD1", state=MDM_STATE_GONE)
    await db_session.commit()

    r = await client.get(BASE, headers=admin_headers)
    assert r.status_code == 200
    assert [d["serial_number"] for d in r.json()["data"]] == ["LIVE1"]
    assert r.json()["meta"]["total"] == 1

    # "When did this leave the fleet" is a real audit question.
    r = await client.get(f"{BASE}?include_gone=true", headers=admin_headers)
    assert sorted(d["serial_number"] for d in r.json()["data"]) == [
        "DEAD1", "LIVE1",
    ]


@pytest.mark.asyncio
async def test_search_and_filters(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
) -> None:
    integ = await _integration(db_session)
    person = IdentityPerson(
        id=uuid.uuid4(), work_email="frank@pingcap.com", display_name="Frank",
        employment_status="active", source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(person)
    asset = AssetItem(
        id=uuid.uuid4(), asset_tag="A-F1", serial_number="F1",
        status="assigned", category="laptop",
    )
    db_session.add(asset)
    await db_session.flush()
    await _device(
        db_session, integ, "F1", category=MDM_MATCH_MATCHED,
        person_id=person.id, asset_id=asset.id,
    )
    await _device(db_session, integ, "G2")
    await db_session.commit()

    r = await client.get(f"{BASE}?q=f1", headers=admin_headers)
    assert [d["serial_number"] for d in r.json()["data"]] == ["F1"]
    # The resolved joins are flattened onto the row for the table.
    row = r.json()["data"][0]
    assert row["assigned_person_name"] == "Frank"
    assert row["asset_tag"] == "A-F1"

    r = await client.get(f"{BASE}?unlinked=true", headers=admin_headers)
    assert [d["serial_number"] for d in r.json()["data"]] == ["G2"]

    r = await client.get(
        f"{BASE}?match_category={MDM_MATCH_MATCHED}", headers=admin_headers,
    )
    assert [d["serial_number"] for d in r.json()["data"]] == ["F1"]


@pytest.mark.asyncio
async def test_dangling_holder_id_renders_as_absence_not_a_500(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
) -> None:
    """The resolution columns carry no foreign keys, on purpose.

    A nightly batch rewrites them and TiDB locks the parent row on every
    FK-carrying insert, so the ledger trades referential integrity for a
    sync that does not serialise. The read path has to hold up its end of
    that trade: an id pointing at nothing renders empty.
    """
    integ = await _integration(db_session)
    await _device(
        db_session, integ, "ORPHAN1",
        category=MDM_MATCH_MATCHED,
        person_id=uuid.uuid4(),   # nobody
        asset_id=uuid.uuid4(),    # nothing
    )
    await db_session.commit()

    r = await client.get(BASE, headers=admin_headers)
    assert r.status_code == 200
    row = r.json()["data"][0]
    assert row["assigned_person_name"] is None
    assert row["asset_tag"] is None


@pytest.mark.asyncio
async def test_confirm_cleanup_refused_when_nothing_is_pending(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
) -> None:
    integ = await _integration(db_session)
    device = await _device(db_session, integ, "NOPE1")
    await db_session.commit()

    r = await client.post(
        f"{BASE}/{device.id}/confirm-cleanup", headers=admin_headers, json={},
    )
    assert r.status_code == 400
    assert r.json()["errors"][0]["code"] == "MDM_CLEANUP_NOT_PENDING"


@pytest.mark.asyncio
async def test_employee_cannot_read_the_ledger(
    client: AsyncClient, db_session: AsyncSession, employee_headers,
) -> None:
    integ = await _integration(db_session)
    await _device(db_session, integ, "PRIV1")
    await db_session.commit()

    r = await client.get(BASE, headers=employee_headers)
    assert r.status_code == 403


@pytest.mark.asyncio
async def test_auditor_reads_but_cannot_confirm(
    client: AsyncClient, db_session: AsyncSession, auditor_headers,
) -> None:
    integ = await _integration(db_session)
    device = await _device(db_session, integ, "AUD1")
    await db_session.commit()

    assert (await client.get(BASE, headers=auditor_headers)).status_code == 200
    meta = await client.get(f"{BASE}/meta", headers=auditor_headers)
    # The page must render without buttons rather than with buttons that 403.
    assert meta.json()["data"]["can_manage"] is False
    r = await client.post(
        f"{BASE}/{device.id}/confirm-cleanup",
        headers=auditor_headers, json={},
    )
    assert r.status_code == 403


@pytest.mark.asyncio
async def test_detail_404_for_unknown_device(
    client: AsyncClient, admin_headers,
) -> None:
    r = await client.get(f"{BASE}/{uuid.uuid4()}", headers=admin_headers)
    assert r.status_code == 404


@pytest.mark.asyncio
async def test_export_carries_the_truncation_headers(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
) -> None:
    integ = await _integration(db_session)
    await _device(db_session, integ, "X1")
    await _device(db_session, integ, "X2")
    await db_session.commit()

    r = await client.get(f"{BASE}/export.xlsx", headers=admin_headers)
    assert r.status_code == 200
    assert r.headers["X-Export-Returned"] == "2"
    assert r.headers["X-Export-Total-Matched"] == "2"
    assert r.headers["X-Export-Truncated"] == "false"


# ──────────────────────────────────────────────────────────────────
# End to end: disposal leaves a cleanup behind
# ──────────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_disposing_an_enrolled_asset_opens_a_cleanup(
    db_session: AsyncSession,
) -> None:
    """Through ``AssetService.dispose_asset``, not the coordinator.

    The hook lives in the service so every disposal path is covered by
    construction — the HTTP route, the buyback approval callback, and
    whatever is added next. A test that called the coordinator directly
    would pass with the hook deleted.
    """
    from app.domains.asset.models import AssetSoftwareCatalog
    from app.domains.asset.schemas import AssetDisposeRequest
    from app.domains.asset.service import AssetService
    from app.domains.provisioning.models import SoftwareConnector

    integ = await _integration(db_session)
    owner = IdentityPerson(
        id=uuid.uuid4(), work_email="mdmowner@pingcap.com",
        display_name="MDM Owner", employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(owner)
    await db_session.flush()
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(), name="Jamf", asset_type="SaaS", status="active",
        owner_person_id=owner.id,
    )
    db_session.add(sw)
    await db_session.flush()
    db_session.add(SoftwareConnector(
        software_id=sw.id, integration_id=integ.id, is_active=True,
        binding_config={},
    ))
    asset = AssetItem(
        id=uuid.uuid4(), asset_tag="A-E2E", serial_number="E2E-SERIAL",
        status="idle", category="laptop",
    )
    db_session.add(asset)
    await db_session.flush()
    device = await _device(
        db_session, integ, "e2e-serial", asset_id=asset.id,
    )

    await AssetService(db_session).dispose_asset(
        asset.id,
        AssetDisposeRequest(disposal_type="scrapped", notes="e2e"),
        owner.id,
    )
    await db_session.flush()

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

    from app.domains.provisioning.models import ProvisioningTask
    task = await db_session.scalar(
        select(ProvisioningTask).where(
            ProvisioningTask.id == device.cleanup_task_id,
        )
    )
    assert task is not None
    assert task.software_id == sw.id
    assert task.assignee_id == owner.id


@pytest.mark.asyncio
async def test_disposing_an_unenrolled_asset_stays_silent(
    db_session: AsyncSession,
) -> None:
    """Most disposals are not Apple hardware and must cost nothing."""
    from app.domains.asset.schemas import AssetDisposeRequest
    from app.domains.asset.service import AssetService
    from app.domains.provisioning.models import ProvisioningTask

    operator = IdentityPerson(
        id=uuid.uuid4(), work_email="ops@pingcap.com", display_name="Ops",
        employment_status="active", source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(operator)
    asset = AssetItem(
        id=uuid.uuid4(), asset_tag="A-DELL", serial_number="DELL-1",
        status="idle", category="laptop",
    )
    db_session.add(asset)
    await db_session.flush()

    await AssetService(db_session).dispose_asset(
        asset.id,
        AssetDisposeRequest(disposal_type="scrapped", notes="no mdm"),
        operator.id,
    )
    await db_session.flush()

    assert (
        await db_session.execute(select(ProvisioningTask))
    ).scalars().all() == []


@pytest.mark.asyncio
async def test_a_broken_cleanup_hook_does_not_take_down_the_disposal(
    db_session: AsyncSession,
) -> None:
    """The hook's docstring promises this; the SAVEPOINT has to deliver it.

    Disposal is the user's action and the cleanup request is a consequence
    of it. On the buyback approval callback a propagating failure here
    would surface as an approved buyback whose asset never got marked
    sold — money agreed, ledger silent. Asserted by making the hook throw
    rather than by reading the code, because "it is wrapped" and "the
    wrapper works" are different claims.
    """
    from unittest.mock import patch

    from app.domains.asset.schemas import AssetDisposeRequest
    from app.domains.asset.service import AssetService

    operator = IdentityPerson(
        id=uuid.uuid4(), work_email="ops2@pingcap.com", display_name="Ops",
        employment_status="active", source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(operator)
    asset = AssetItem(
        id=uuid.uuid4(), asset_tag="A-BOOM", serial_number="BOOM-1",
        status="idle", category="laptop",
    )
    db_session.add(asset)
    await db_session.flush()

    with patch(
        "app.core.coordinators.mdm_device_retirement_coordinator"
        ".request_cleanup_for_disposed_asset",
        side_effect=RuntimeError("cleanup hook exploded"),
    ):
        disposed, _op = await AssetService(db_session).dispose_asset(
            asset.id,
            AssetDisposeRequest(disposal_type="scrapped", notes="boom"),
            operator.id,
        )

    # The disposal stands, and the session is still usable afterwards.
    assert disposed.status == "disposed"
    await db_session.flush()
    fresh = await db_session.get(AssetItem, asset.id)
    assert fresh.status == "disposed"


@pytest.mark.asyncio
async def test_filtering_by_gone_actually_returns_the_gone_devices(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
) -> None:
    """Asking for the retired devices IS asking to include them.

    The state filter and ``include_gone`` were independent controls, so
    selecting 「已从 MDM 移除」 dropped every row before the state was even
    tested — an empty table, which reads as "there are none".
    """
    integ = await _integration(db_session)
    await _device(db_session, integ, "ALIVE1")
    await _device(db_session, integ, "RETIRED1", state=MDM_STATE_GONE)
    await _device(db_session, integ, "RETIRED2", state=MDM_STATE_GONE)
    await db_session.commit()

    r = await client.get(
        f"{BASE}?enrollment_state={MDM_STATE_GONE}", headers=admin_headers,
    )
    assert r.status_code == 200
    assert sorted(d["serial_number"] for d in r.json()["data"]) == [
        "RETIRED1", "RETIRED2",
    ]
    assert r.json()["meta"]["total"] == 2


@pytest.mark.asyncio
async def test_the_ledger_carries_the_disposal_date_the_card_names(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
) -> None:
    """The card and the page it links to must tell one story.

    The backlog reconciler makes ``cleanup_requested_at`` and the
    disposal two different dates for the first time — a Mac sold in
    January can be asked about tonight. The cleanup card states the sale
    date; this is the page that card deep-links to, so the fact has to
    survive the click.
    """
    integ = await _integration(db_session)
    sold_on = datetime(2026, 1, 14, 3, 0, tzinfo=UTC)
    asset = AssetItem(
        id=uuid.uuid4(), asset_tag="A-DISPDATE1", serial_number="DISPDATE1",
        status="disposed", category="laptop",
        disposal_type="sold", disposed_at=sold_on,
    )
    db_session.add(asset)
    await db_session.flush()
    device = await _device(
        db_session, integ, "DISPDATE1", category=MDM_MATCH_MATCHED,
        asset_id=asset.id,
    )
    await db_session.commit()

    r = await client.get(
        f"/api/v1/assets/mdm-devices/{device.id}", headers=admin_headers,
    )
    assert r.status_code == 200
    body = r.json()["data"]
    assert body["asset_disposed_at"] is not None
    assert body["asset_disposed_at"].startswith("2026-01-14")

    # And on the list, so the export (which serialises the same payload)
    # carries it too — the auditor's sheet is the third surface.
    rl = await client.get(
        "/api/v1/assets/mdm-devices?q=DISPDATE1", headers=admin_headers,
    )
    assert rl.status_code == 200
    row = rl.json()["data"][0]
    assert row["asset_disposed_at"].startswith("2026-01-14")