"""Cross-month per-software cost summary — ``billing_service.summarize_range``
plus its two HTTP surfaces (JSON + XLSX).

Why this exists (2026-07-30): finance could see "software X cost this much in
month M" (one month at a time) but had no way to ask "these software, this
date range, how much each" — the standing question at renewal / budget time.
The range summary answers it, and these tests pin the properties that make
the answer trustworthy:

  * a CLOSED month reads the frozen ``software_monthly_bills`` snapshot, so a
    later plan-price edit can never retroactively move it (the exact bug the
    monthly snapshot was introduced for — the range view must not reopen it
    by recomputing from live plans);
  * an unclosed month is live-previewed, and months beyond the live-preview
    cap are REPORTED in ``excluded_periods`` rather than silently dropped
    (a short total that looks complete is worse than a flagged gap);
  * currencies are bucketed, never summed across (CNY + USD is not a number);
  * ``distinct_users`` de-duplicates one person across months AND across the
    frozen/live sources, and never counts add-on charge rows as seats;
  * scope + software filters narrow the same way the monthly view does, and
    ``view=all`` is gated on the software read scope.
"""
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.exceptions import AppException
from app.domains.asset import billing_service
from app.domains.asset.billing_service import (
    RANGE_MAX_LIVE_PERIODS,
    _current_period,
    close_month,
    summarize_range,
)
from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareMonthlyBill,
    SoftwareSeatAllocation,
)
from app.identity.person_models import IdentityPerson

pytestmark = pytest.mark.asyncio


def _shift(period: str, months: int) -> str:
    """``period`` shifted by ``months`` (negative = back)."""
    idx = int(period[:4]) * 12 + (int(period[5:]) - 1) + months
    year, month = divmod(idx, 12)
    return f"{year:04d}-{month + 1:02d}"


def _month_start(period: str) -> datetime:
    return datetime(int(period[:4]), int(period[5:]), 1, tzinfo=UTC)


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 _software_with_plan(
    db: AsyncSession, name: str, *, price: str, currency: str,
) -> tuple[AssetSoftwareCatalog, SoftwareLicensePlan]:
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(), name=name, asset_type="SaaS", status="active",
    )
    db.add(sw)
    await db.flush()
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=sw.id, name=f"{name}-Pro",
        price=Decimal(price), billing_cycle="monthly", currency=currency,
        status="active", max_seats=None,
    )
    db.add(plan)
    await db.flush()
    return sw, plan


async def _seat(
    db: AsyncSession, plan: SoftwareLicensePlan, user_id: uuid.UUID,
    *, allocated_at: datetime,
) -> SoftwareSeatAllocation:
    a = SoftwareSeatAllocation(
        id=uuid.uuid4(), plan_id=plan.id, user_id=user_id,
        status="active", allocated_at=allocated_at,
    )
    db.add(a)
    await db.flush()
    return a


async def _two_software_two_months(db: AsyncSession):
    """Fixture body: two software (USD + CNY), two holders, seats granted
    before the previous month. Previous month gets CLOSED; the current
    month stays live. Returns (prev_period, curr_period, sw_usd, sw_cny,
    holder_a, holder_b)."""
    curr = _current_period()
    prev = _shift(curr, -1)
    sw_usd, plan_usd = await _software_with_plan(
        db, f"RangeUSD-{uuid.uuid4().hex[:6]}", price="10.00", currency="USD",
    )
    sw_cny, plan_cny = await _software_with_plan(
        db, f"RangeCNY-{uuid.uuid4().hex[:6]}", price="100.00", currency="CNY",
    )
    holder_a = await _person(db, f"range-a-{uuid.uuid4().hex[:6]}@x.com")
    holder_b = await _person(db, f"range-b-{uuid.uuid4().hex[:6]}@x.com")
    granted = _month_start(prev) - timedelta(days=5)
    # A holds both software; B holds only the USD one → USD range total
    # covers 2 people, CNY covers 1.
    await _seat(db, plan_usd, holder_a.id, allocated_at=granted)
    await _seat(db, plan_usd, holder_b.id, allocated_at=granted)
    await _seat(db, plan_cny, holder_a.id, allocated_at=granted)
    await db.commit()
    await close_month(db, prev, actor_id=None)
    await db.commit()
    return prev, curr, sw_usd, sw_cny, holder_a, holder_b


async def test_range_totals_per_software_per_currency(db_session: AsyncSession):
    """Each software gets its own range total, bucketed per currency, with a
    per-month breakdown — closed month frozen, current month live."""
    prev, curr, sw_usd, sw_cny, _a, _b = await _two_software_two_months(db_session)

    data = await summarize_range(
        db_session, prev, curr,
        software_ids=[sw_usd.id, sw_cny.id],
    )

    assert data["periods"] == [prev, curr]
    assert data["period_status"] == {prev: "finalized", curr: "live"}
    assert data["live_periods"] == [curr]
    assert data["excluded_periods"] == []

    by_id = {row["software_id"]: row for row in data["software"]}
    usd = by_id[str(sw_usd.id)]
    cny = by_id[str(sw_cny.id)]
    # 2 seats × $10 × 2 months
    assert usd["by_currency"] == {"USD": "40.00"}
    assert usd["months"] == {prev: {"USD": "20.00"}, curr: {"USD": "20.00"}}
    assert usd["distinct_users"] == 2
    # 1 seat × ¥100 × 2 months
    assert cny["by_currency"] == {"CNY": "200.00"}
    assert cny["distinct_users"] == 1

    # Never summed across currencies — the grand total stays split, and the
    # mixed-currency flag is on so the UI refuses to print one number.
    assert data["totals"]["by_currency"] == {"CNY": "200.00", "USD": "40.00"}
    assert data["currency_warning"] is True
    # A person holding two software is ONE billed person overall.
    assert data["distinct_users"] == 2


async def test_closed_month_is_frozen_against_later_price_edit(
    db_session: AsyncSession,
):
    """The caliber invariant: raising a plan's price today must not move the
    already-closed month inside the range. Only the live month follows the
    new price."""
    prev, curr, sw_usd, _cny, _a, _b = await _two_software_two_months(db_session)

    plan = (
        await db_session.execute(
            select(SoftwareLicensePlan).where(
                SoftwareLicensePlan.software_id == sw_usd.id,
            )
        )
    ).scalar_one()
    plan.price = Decimal("99.00")
    await db_session.commit()

    data = await summarize_range(
        db_session, prev, curr, software_ids=[sw_usd.id],
    )
    row = data["software"][0]
    # Closed month keeps the frozen $10 × 2 seats; the live month recomputes
    # at the new $99 × 2.
    assert row["months"][prev] == {"USD": "20.00"}
    assert row["months"][curr] == {"USD": "198.00"}


async def test_scope_and_software_filters_narrow(db_session: AsyncSession):
    """``user_ids`` narrows to one person's seats; ``software_ids`` narrows to
    one product. Both apply to the frozen AND the live month."""
    prev, curr, sw_usd, sw_cny, holder_a, _b = await _two_software_two_months(
        db_session,
    )

    scoped = await summarize_range(db_session, prev, curr, user_ids=[holder_a.id])
    by_id = {row["software_id"]: row for row in scoped["software"]}
    # A holds one USD seat (not two) and the single CNY seat.
    assert by_id[str(sw_usd.id)]["by_currency"] == {"USD": "20.00"}
    assert by_id[str(sw_cny.id)]["by_currency"] == {"CNY": "200.00"}
    assert scoped["distinct_users"] == 1

    only_cny = await summarize_range(
        db_session, prev, curr, software_ids=[sw_cny.id],
    )
    assert [r["software_id"] for r in only_cny["software"]] == [str(sw_cny.id)]
    assert only_cny["totals"]["by_currency"] == {"CNY": "200.00"}
    assert only_cny["currency_warning"] is False


async def test_endpoints_in_either_order_and_single_month(
    db_session: AsyncSession,
):
    """start/end may arrive reversed (pickers vary); a one-month range is a
    valid degenerate case."""
    prev, curr, sw_usd, _cny, _a, _b = await _two_software_two_months(db_session)

    reversed_range = await summarize_range(
        db_session, curr, prev, software_ids=[sw_usd.id],
    )
    assert reversed_range["periods"] == [prev, curr]
    assert reversed_range["start"] == prev
    assert reversed_range["end"] == curr

    one_month = await summarize_range(
        db_session, prev, prev, software_ids=[sw_usd.id],
    )
    assert one_month["periods"] == [prev]
    assert one_month["software"][0]["by_currency"] == {"USD": "20.00"}


async def test_unclosed_months_beyond_cap_are_reported_not_dropped(
    db_session: AsyncSession,
):
    """A range of never-closed months live-previews the most recent
    ``RANGE_MAX_LIVE_PERIODS`` and lists the rest in ``excluded_periods`` so
    the reader knows the total is short."""
    curr = _current_period()
    start = _shift(curr, -(RANGE_MAX_LIVE_PERIODS + 2))

    data = await summarize_range(db_session, start, curr)

    assert len(data["live_periods"]) == RANGE_MAX_LIVE_PERIODS
    assert data["live_periods"] == data["periods"][-RANGE_MAX_LIVE_PERIODS:]
    assert data["excluded_periods"] == data["periods"][:-RANGE_MAX_LIVE_PERIODS]
    assert all(
        data["period_status"][p] == "excluded" for p in data["excluded_periods"]
    )


async def test_future_months_are_never_computed(db_session: AsyncSession):
    """A month that hasn't happened yet must contribute NOTHING.

    ``preview_current_or_draft`` bills every currently-active seat for any
    period handed to it, so live-previewing a future month would (a) invent
    cost that hasn't been incurred and label it "实时", and (b) burn the
    live-preview budget the real current month needs. Pinned because a
    RangePicker slip (end = December) is a one-click mistake.
    """
    prev, curr, sw_usd, _cny, _a, _b = await _two_software_two_months(db_session)
    future_1 = _shift(curr, 1)
    future_2 = _shift(curr, 2)

    data = await summarize_range(
        db_session, prev, future_2, software_ids=[sw_usd.id],
    )

    assert data["future_periods"] == [future_1, future_2]
    assert data["period_status"][future_1] == "future"
    assert data["period_status"][future_2] == "future"
    # The real months are untouched: closed month frozen + current month live.
    assert data["live_periods"] == [curr]
    row = data["software"][0]
    assert set(row["months"]) == {prev, curr}
    assert row["by_currency"] == {"USD": "40.00"}   # 2 seats × $10 × 2 months
    # Nothing was invented for the future months.
    assert future_1 not in data["totals"]["months"]
    assert data["totals"]["by_currency"] == {"USD": "40.00"}


async def test_currency_drift_between_frozen_label_and_live_plan_is_reported(
    db_session: AsyncSession,
):
    """A closed month whose currency label disagrees with what the software
    charges TODAY must be named, not left for a human to spot.

    Real incident (2026-07-30): migration 117 blanket-backfilled every plan
    and bill row to ``CNY``, mislabelling the USD-priced Claude / OPEN AI
    seats until the next close wrote ``USD``. The range table then showed ¥
    and $ on the same seats at the same price, with nothing explaining why.
    """
    prev, curr, sw_usd, _cny, _a, _b = await _two_software_two_months(db_session)

    # Simulate the mislabel: the frozen month says CNY, the live plan says USD.
    await db_session.execute(
        SoftwareMonthlyBill.__table__.update()
        .where(
            SoftwareMonthlyBill.software_id == sw_usd.id,
            SoftwareMonthlyBill.period == prev,
        )
        .values(currency="CNY")
    )
    await db_session.commit()

    data = await summarize_range(
        db_session, prev, curr, software_ids=[sw_usd.id],
    )

    assert len(data["currency_drift"]) == 1
    entry = data["currency_drift"][0]
    assert entry["software_id"] == str(sw_usd.id)
    assert entry["plan_currency"] == "USD"
    assert entry["periods"] == [{"period": prev, "currency": "CNY"}]
    # Reported, NEVER silently rewritten — the frozen row still reads CNY and
    # the split total still shows both currencies.
    assert data["software"][0]["months"][prev] == {"CNY": "20.00"}
    assert data["currency_warning"] is True


async def test_currency_drift_stays_quiet_on_a_genuinely_multi_currency_product(
    db_session: AsyncSession,
):
    """A product whose live plans really do bill in two currencies has no
    single "right" label, so a differing frozen month is not evidence of a
    mislabel — staying quiet is what keeps the banner worth reading."""
    prev, curr, sw_usd, _cny, _a, _b = await _two_software_two_months(db_session)
    # Add a second, CNY-priced live plan to the same software.
    db_session.add(SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=sw_usd.id, name="CNY-tier",
        price=Decimal("70.00"), billing_cycle="monthly", currency="CNY",
        status="active", max_seats=None,
    ))
    await db_session.execute(
        SoftwareMonthlyBill.__table__.update()
        .where(
            SoftwareMonthlyBill.software_id == sw_usd.id,
            SoftwareMonthlyBill.period == prev,
        )
        .values(currency="CNY")
    )
    await db_session.commit()

    data = await summarize_range(
        db_session, prev, curr, software_ids=[sw_usd.id],
    )
    assert data["currency_drift"] == []


async def test_range_wider_than_cap_is_rejected(db_session: AsyncSession):
    curr = _current_period()
    with pytest.raises(AppException) as exc:
        await summarize_range(
            db_session, _shift(curr, -billing_service.RANGE_MAX_MONTHS), curr,
        )
    assert exc.value.code == "RANGE_TOO_WIDE"

    with pytest.raises(AppException) as exc:
        await summarize_range(db_session, "2026-1", "2026-03")
    assert exc.value.code == "INVALID_PERIOD"


async def test_api_finance_sees_company_wide_employee_does_not(
    client, db_session: AsyncSession, finance_headers, employee_headers,
):
    """``view=all`` is the company-wide lens — finance (software read scope)
    gets it, a plain employee is refused and must use ``view=my``."""
    prev, curr, sw_usd, _cny, _a, _b = await _two_software_two_months(db_session)

    ok = await client.get(
        "/api/v1/assets/software-billing/range-summary",
        params={"start": prev, "end": curr, "view": "all",
                "software_id": str(sw_usd.id)},
        headers=finance_headers,
    )
    assert ok.status_code == 200
    payload = ok.json()["data"]
    assert payload["software"][0]["by_currency"] == {"USD": "40.00"}

    denied = await client.get(
        "/api/v1/assets/software-billing/range-summary",
        params={"start": prev, "end": curr, "view": "all"},
        headers=employee_headers,
    )
    assert denied.status_code == 403

    own = await client.get(
        "/api/v1/assets/software-billing/range-summary",
        params={"start": prev, "end": curr, "view": "my"},
        headers=employee_headers,
    )
    assert own.status_code == 200
    # The employee fixture holds none of these seats — an empty result, not
    # an error, and the month legend still renders.
    assert own.json()["data"]["software"] == []
    assert own.json()["data"]["period_status"][prev] == "finalized"


async def test_api_xlsx_export_has_one_row_per_software_currency(
    client, db_session: AsyncSession, finance_headers,
):
    """The export carries currency as its own column (never a mixed sum) and
    one column per month."""
    prev, curr, sw_usd, sw_cny, _a, _b = await _two_software_two_months(db_session)

    res = await client.get(
        "/api/v1/assets/software-billing/range-summary/export.xlsx",
        params={"start": prev, "end": curr,
                "software_id": [str(sw_usd.id), str(sw_cny.id)]},
        headers=finance_headers,
    )
    assert res.status_code == 200
    assert "spreadsheet" in res.headers["content-type"]

    import io

    from openpyxl import load_workbook

    ws = load_workbook(io.BytesIO(res.content)).active
    header = [c.value for c in ws[1]]
    assert header[:2] == ["软件", "币种"]
    assert any(prev in str(h) for h in header)
    assert any(curr in str(h) for h in header)
    # Row 2 is ``build_xlsx``'s machine-key row; data starts at row 3.
    assert [c.value for c in ws[2]][:2] == ["software_name", "currency"]
    body = [[c.value for c in row] for row in ws.iter_rows(min_row=3)]
    currencies = {row[1] for row in body}
    assert currencies == {"USD", "CNY"}
    usd_row = next(row for row in body if row[1] == "USD")
    assert usd_row[2] == "40.00"