"""End-to-end tests for ``GET /api/v1/assets/export.xlsx``.

The export endpoint mirrors the list endpoint's filter signature and
RBAC; the only differences are (a) no pagination — up to
``ASSET_EXPORT_ROW_LIMIT`` rows in one shot — and (b) it returns an
XLSX binary plus three ``X-Asset-Export-*`` response headers carrying
the matched/returned/truncated metadata.

Coverage:

  - admin can export full inventory; the workbook contains a header
    row, a machine-name row, and one data row per asset.
  - filters (status, category, location, usage_company, search) match
    the list endpoint's behaviour; the file contains only matching rows.
  - employee scope: an employee can only export their own assets — the
    file never leaks teammates' rows even if the caller passes
    ``assigned_to`` of someone else.
  - manager scope: returns own + direct reports' assets.
  - audit log row written with ``asset.export`` action and accurate
    row_count / total_matched / truncated flag.
  - sort ``sort_by`` whitelist is enforced (matches list endpoint 400).
"""
from __future__ import annotations

import io
import uuid

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

from app.audit.models import AuditLog
from app.domains.asset.models import AssetItem, AssetOfficeAddress
from app.identity.person_models import IdentityPerson


def _read_workbook(content: bytes) -> list[list]:
    """Return all rows from the active sheet as a list of lists."""
    wb = load_workbook(io.BytesIO(content), read_only=True)
    ws = wb.active
    return [list(row) for row in ws.iter_rows(values_only=True)]


@pytest.mark.asyncio
async def test_admin_export_includes_all_filtered_rows(
    client: AsyncClient,
    admin_headers: dict,
    admin_user,
    db_session: AsyncSession,
):
    """Admin filters by category=laptop and gets exactly the laptops."""
    laptops = [
        AssetItem(
            asset_tag=f"EXP-LAP-{i:03d}",
            asset_name=f"Test laptop {i}",
            category="laptop",
            status="idle",
            brand="TestBrand",
            location="北京",
        )
        for i in range(3)
    ]
    monitor = AssetItem(
        asset_tag="EXP-MON-001",
        asset_name="Test monitor",
        category="monitor",
        status="idle",
        brand="TestBrand",
        location="北京",
    )
    db_session.add_all([*laptops, monitor])
    await db_session.commit()

    resp = await client.get(
        "/api/v1/assets/export.xlsx?category=laptop",
        headers=admin_headers,
    )
    assert resp.status_code == 200
    assert resp.headers["content-type"].startswith(
        "application/vnd.openxmlformats-officedocument."
        "spreadsheetml.sheet"
    )
    cd = resp.headers.get("content-disposition", "")
    assert "attachment" in cd and "assets_" in cd and ".xlsx" in cd

    rows = _read_workbook(resp.content)
    # row 1 = label, row 2 = machine name, row 3+ = data
    assert len(rows) >= 5  # 2 header rows + 3 laptops
    machine_row = rows[1]
    assert "asset_tag" in machine_row
    tag_idx = machine_row.index("asset_tag")
    cat_idx = machine_row.index("category")
    data_rows = rows[2:]
    tags = {r[tag_idx] for r in data_rows}
    cats = {r[cat_idx] for r in data_rows}
    # All exported rows are laptops; the monitor must not leak.
    assert "EXP-MON-001" not in tags
    assert {f"EXP-LAP-{i:03d}" for i in range(3)} <= tags
    assert cats == {"laptop"}

    # Metadata headers reflect what we returned.
    assert int(resp.headers["x-asset-export-returned"]) == len(data_rows)
    assert resp.headers["x-asset-export-truncated"] == "false"


@pytest.mark.asyncio
async def test_export_writes_audit_with_filters(
    client: AsyncClient,
    admin_headers: dict,
    admin_user,
    db_session: AsyncSession,
):
    asset = AssetItem(
        asset_tag="EXP-AUDIT-001",
        category="laptop",
        status="idle",
        location="上海",
    )
    db_session.add(asset)
    await db_session.commit()

    resp = await client.get(
        "/api/v1/assets/export.xlsx?location=上海&search=AUDIT",
        headers=admin_headers,
    )
    assert resp.status_code == 200

    audit_user, _ = admin_user
    audits = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "asset.export",
                AuditLog.actor_user_id == audit_user.id,
            )
        )
    ).scalars().all()
    assert len(audits) >= 1
    detail = audits[-1].detail
    assert detail["row_count"] >= 1
    assert detail["filters"]["location"] == ["上海"]
    assert detail["filters"]["search"] == "AUDIT"


@pytest.mark.asyncio
async def test_employee_export_only_includes_own_assets(
    client: AsyncClient,
    employee_user,
    employee_headers,
    db_session: AsyncSession,
):
    """An employee MUST NOT be able to read other people's assets via
    export, even if they pass ``assigned_to=<someone else>``."""
    employee, _ = employee_user
    # FK on AssetItem.assigned_to → identity_persons.id, so the
    # "other employee" we're trying to peek at must be a real row.
    other = IdentityPerson(
        id=uuid.uuid4(),
        work_email="other@test.com",
        display_name="Other Person",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db_session.add(other)
    await db_session.commit()
    other_id = other.id

    own = AssetItem(
        asset_tag="EXP-OWN-001",
        category="laptop",
        status="claimed",
        assigned_to=employee.id,
    )
    other_asset = AssetItem(
        asset_tag="EXP-OTHER-001",
        category="laptop",
        status="claimed",
        assigned_to=other_id,
    )
    db_session.add_all([own, other_asset])
    await db_session.commit()

    # No assigned_to filter → employee gets own only.
    resp = await client.get(
        "/api/v1/assets/export.xlsx",
        headers=employee_headers,
    )
    assert resp.status_code == 200
    rows = _read_workbook(resp.content)
    machine_row = rows[1]
    tag_idx = machine_row.index("asset_tag")
    tags = {r[tag_idx] for r in rows[2:]}
    assert "EXP-OWN-001" in tags
    assert "EXP-OTHER-001" not in tags

    # Trying to spoof assigned_to=<someone else> → 403.
    resp_spoof = await client.get(
        f"/api/v1/assets/export.xlsx?assigned_to={other_id}",
        headers=employee_headers,
    )
    assert resp_spoof.status_code == 403


@pytest.mark.asyncio
async def test_export_neutralizes_formula_injection_including_tab_cr(
    client: AsyncClient,
    admin_headers: dict,
    admin_user,
    db_session: AsyncSession,
):
    """Free-text / HRIS-sourced cells must be neutralized against CSV /
    Excel formula injection — INCLUDING the ``\\t`` and ``\\r`` DDE
    prefixes that the previous hand-rolled prefix list (``= + - @`` only)
    missed. The export now routes every cell through the shared
    ``app.core.exporting.to_cell``, the single source of truth for the
    full prefix set. Regression for the 2026-06-14 code-review fix.
    """
    payloads = {
        "EXP-INJ-EQ": '=HYPERLINK("http://evil","x")',
        "EXP-INJ-PLUS": "+1+1",
        "EXP-INJ-MINUS": "-2+3",
        "EXP-INJ-AT": "@SUM(A1:A9)",
        "EXP-INJ-TAB": "\t=cmd|'/c calc'!A1",
        "EXP-INJ-CR": "\r=1+1",
    }
    for tag, name in payloads.items():
        db_session.add(AssetItem(
            asset_tag=tag,
            asset_name=name,
            category="laptop",
            status="idle",
            location="北京",
        ))
    await db_session.commit()

    resp = await client.get(
        "/api/v1/assets/export.xlsx",
        headers=admin_headers,
    )
    assert resp.status_code == 200
    rows = _read_workbook(resp.content)
    machine_row = rows[1]
    tag_idx = machine_row.index("asset_tag")
    name_idx = machine_row.index("asset_name")
    by_tag = {r[tag_idx]: r[name_idx] for r in rows[2:]}

    for tag, original in payloads.items():
        assert tag in by_tag, f"{tag} missing from export"
        cell = by_tag[tag]
        # Neutralised: a leading apostrophe forces Excel to treat the
        # cell as text, so the formula / DDE payload never executes.
        assert isinstance(cell, str) and cell.startswith("'"), (
            f"{tag}: cell {cell!r} was not neutralized "
            f"(original payload {original!r})"
        )


@pytest.mark.asyncio
async def test_invalid_sort_by_rejected(
    client: AsyncClient,
    admin_headers: dict,
):
    resp = await client.get(
        "/api/v1/assets/export.xlsx?sort_by=__nope__",
        headers=admin_headers,
    )
    assert resp.status_code == 400
    body = resp.json()
    assert body["errors"][0]["code"] == "INVALID_SORT_FIELD"


@pytest.mark.asyncio
async def test_export_with_holder_kind_and_holder_office_filter(
    client: AsyncClient,
    admin_headers: dict,
    db_session: AsyncSession,
):
    """Export honours ``holder_kind`` + ``holder_office_id`` filters
    and writes both into the audit detail. Regression for PR 2026-05-26
    where the export endpoint applied the filters to the query but
    omitted them from the audit log — admins exporting "office-held
    only" left no audit trail of the narrowing.
    """
    office = AssetOfficeAddress(
        city="Tokyo",
        office_name="Tokyo HQ",
        is_active=True,
    )
    db_session.add(office)
    await db_session.commit()
    await db_session.refresh(office)

    # Three assets: office-held at Tokyo (matches),
    # office-held elsewhere (only kind matches),
    # personal-held (matches nothing).
    other_office = AssetOfficeAddress(
        city="Shanghai", office_name="Shanghai", is_active=True,
    )
    db_session.add(other_office)
    await db_session.commit()
    await db_session.refresh(other_office)

    db_session.add_all([
        AssetItem(
            asset_tag="EXP-HOLDER-TOKYO",
            category="laptop",
            status="idle",
            holder_kind="office",
            holder_office_id=office.id,
        ),
        AssetItem(
            asset_tag="EXP-HOLDER-SHANGHAI",
            category="laptop",
            status="idle",
            holder_kind="office",
            holder_office_id=other_office.id,
        ),
        AssetItem(
            asset_tag="EXP-HOLDER-PERSONAL",
            category="laptop",
            status="idle",
            holder_kind="personal",
        ),
    ])
    await db_session.commit()

    # holder_office_id alone — should pick exactly the Tokyo row.
    resp = await client.get(
        f"/api/v1/assets/export.xlsx?holder_office_id={office.id}",
        headers=admin_headers,
    )
    assert resp.status_code == 200
    rows = _read_workbook(resp.content)
    # Two metadata rows (header + machine name) + N data rows.
    data_rows = rows[2:]
    tags = {r[0] for r in data_rows}
    assert tags == {"EXP-HOLDER-TOKYO"}

    # holder_kind alone — picks office-held only (two rows).
    resp_kind = await client.get(
        "/api/v1/assets/export.xlsx?holder_kind=office",
        headers=admin_headers,
    )
    assert resp_kind.status_code == 200
    data_rows = _read_workbook(resp_kind.content)[2:]
    tags = {r[0] for r in data_rows}
    assert tags == {"EXP-HOLDER-TOKYO", "EXP-HOLDER-SHANGHAI"}

    # Both filters together — narrows to Tokyo only.
    resp_both = await client.get(
        f"/api/v1/assets/export.xlsx"
        f"?holder_kind=office&holder_office_id={office.id}",
        headers=admin_headers,
    )
    assert resp_both.status_code == 200
    data_rows = _read_workbook(resp_both.content)[2:]
    tags = {r[0] for r in data_rows}
    assert tags == {"EXP-HOLDER-TOKYO"}

    # Audit log: the export with BOTH filters MUST have both in its
    # detail. We can't rely on ``ORDER BY created_at DESC LIMIT 1``
    # because the three calls here landed within the same second and
    # MySQL's DATETIME ties on insert order — so locate the specific
    # row by its detail shape instead.
    audits = (
        await db_session.execute(
            select(AuditLog).where(AuditLog.action == "asset.export")
        )
    ).scalars().all()
    matching = [
        a for a in audits
        if a.detail.get("filters", {}).get("holder_kind") == ["office"]
        and a.detail.get("filters", {}).get("holder_office_id") == [str(office.id)]
    ]
    assert matching, (
        "expected an asset.export audit row with both holder_kind and "
        "holder_office_id captured in detail.filters"
    )

    # Also assert the holder_kind-only call captured holder_kind but
    # NOT holder_office_id (regression: previously omitted entirely).
    kind_only = [
        a for a in audits
        if a.detail.get("filters", {}).get("holder_kind") == ["office"]
        and a.detail.get("filters", {}).get("holder_office_id") is None
    ]
    assert kind_only


@pytest.mark.asyncio
async def test_export_holder_office_id_rejects_invalid_uuid(
    client: AsyncClient,
    admin_headers: dict,
):
    """FastAPI's ``list[UUID]`` Query parser must reject malformed
    holder_office_id values up-front, not silently filter to zero
    rows."""
    resp = await client.get(
        "/api/v1/assets/export.xlsx?holder_office_id=not-a-uuid",
        headers=admin_headers,
    )
    assert resp.status_code == 422