"""API tests for GET /assets/seat-allocations/export.xlsx.

Pins the shared export-endpoint contract (see
``tests/test_provisioning/test_tasks_export.py`` for the canonical
version): RBAC mirrors the list endpoint (admin exports any slice,
an employee only their own ``user_id=<self>`` slice), filters pass
through the shared ``_query_seat_allocations`` helper, the response
carries the X-Export-* truncation headers, and the download leaves an
audit trail.

NOTE: this file is enumerated explicitly in the ``asset-b`` go-ci shard
(``.github/workflows/go-ci.yml``) — new sibling files must be added there
too or they silently never run.
"""
from __future__ import annotations

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

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 (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.identity.person_models import IdentityPerson

XLSX_CT = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"


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


async def _plan(
    db: AsyncSession, *, software_name: str | None = None,
) -> SoftwareLicensePlan:
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(),
        name=software_name or f"sw-{uuid.uuid4().hex[:6]}",
        asset_type="SaaS",
        status="active",
    )
    db.add(sw)
    await db.flush()
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(),
        software_id=sw.id,
        name=f"plan-{uuid.uuid4().hex[:6]}",
        price=Decimal("10"),
        billing_cycle="monthly",
        status="active",
    )
    db.add(plan)
    await db.flush()
    return plan


async def _seat(
    db: AsyncSession,
    *,
    plan: SoftwareLicensePlan,
    user: IdentityPerson,
    status: str = "active",
) -> SoftwareSeatAllocation:
    seat = SoftwareSeatAllocation(
        plan_id=plan.id,
        user_id=user.id,
        status=status,
        allocated_at=datetime.now(UTC),
    )
    db.add(seat)
    await db.flush()
    return seat


@pytest.mark.asyncio
async def test_admin_export_returns_xlsx_with_headers_and_audit(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    plan = await _plan(db_session, software_name="ExportSoft")
    holder = await _person(db_session, "seat-holder@test.com")
    other = await _person(db_session, "seat-other@test.com")
    await _seat(db_session, plan=plan, user=holder)
    await _seat(db_session, plan=plan, user=other, status="revoked")
    await db_session.commit()

    resp = await client.get(
        "/api/v1/assets/seat-allocations/export.xlsx",
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    assert resp.headers["content-type"].startswith(XLSX_CT)
    assert resp.headers["x-export-truncated"] == "false"
    assert int(resp.headers["x-export-returned"]) == 2
    assert "software_seat_allocations_" in resp.headers["content-disposition"]

    wb = load_workbook(io.BytesIO(resp.content))
    ws = wb.active
    # two-row header convention: labels then machine keys
    assert ws.cell(row=1, column=1).value == "软件"
    assert ws.cell(row=2, column=1).value == "software_name"
    email_cells = [
        ws.cell(row=r, column=4).value for r in range(3, ws.max_row + 1)
    ]
    assert "seat-holder@test.com" in email_cells
    assert "seat-other@test.com" in email_cells

    # the export itself must leave an audit row
    row = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "asset.seat_allocations.export",
            ).order_by(AuditLog.created_at.desc()),
        )
    ).scalars().first()
    assert row is not None
    assert row.resource_type == "software_seat_allocation"
    assert row.detail["row_count"] == 2
    assert row.detail["truncated"] is False


@pytest.mark.asyncio
async def test_status_filter_passes_through(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    plan = await _plan(db_session)
    active_user = await _person(db_session, "active-seat@test.com")
    revoked_user = await _person(db_session, "revoked-seat@test.com")
    await _seat(db_session, plan=plan, user=active_user, status="active")
    await _seat(db_session, plan=plan, user=revoked_user, status="revoked")
    await db_session.commit()

    resp = await client.get(
        "/api/v1/assets/seat-allocations/export.xlsx",
        params={"status": "revoked"},
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    assert int(resp.headers["x-export-returned"]) == 1

    wb = load_workbook(io.BytesIO(resp.content))
    ws = wb.active
    email_cells = [
        ws.cell(row=r, column=4).value for r in range(3, ws.max_row + 1)
    ]
    assert email_cells == ["revoked-seat@test.com"]

    audit = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "asset.seat_allocations.export",
            ).order_by(AuditLog.created_at.desc()),
        )
    ).scalars().first()
    assert audit is not None
    assert audit.detail["filters"]["status"] == "revoked"


@pytest.mark.asyncio
async def test_employee_cannot_export_other_users(
    client: AsyncClient, db_session: AsyncSession, employee_headers,
):
    # No user_id → full-universe request → 403, mirroring the list
    # endpoint's RBAC branch exactly.
    resp = await client.get(
        "/api/v1/assets/seat-allocations/export.xlsx",
        headers=employee_headers,
    )
    assert resp.status_code == 403

    # Someone else's user_id → 403 too.
    resp = await client.get(
        "/api/v1/assets/seat-allocations/export.xlsx",
        params={"user_id": str(uuid.uuid4())},
        headers=employee_headers,
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_employee_can_export_own_slice(
    client: AsyncClient,
    db_session: AsyncSession,
    employee_headers,
    employee_user,
):
    emp, _token = employee_user
    plan = await _plan(db_session)
    stranger = await _person(db_session, "stranger-seat@test.com")
    await _seat(db_session, plan=plan, user=emp)
    await _seat(db_session, plan=plan, user=stranger)
    await db_session.commit()

    resp = await client.get(
        "/api/v1/assets/seat-allocations/export.xlsx",
        params={"user_id": str(emp.id)},
        headers=employee_headers,
    )
    assert resp.status_code == 200, resp.text
    assert int(resp.headers["x-export-returned"]) == 1