"""Regression tests pinning ``billing_service`` revoke-mid-month behaviour.

The user-facing invariant under test: when an admin (or any other
revoke path) closes a seat mid-month, the company still owes the
vendor for the current month (monthly-paid SaaS doesn't refund
mid-cycle). Billing must:

  - Include the seat in the CURRENT period's bill at the full plan
    price, regardless of ``revoked_at`` falling inside the period.
  - Exclude the seat from the NEXT period's bill.
  - Exclude rejected-at-approval rows whose ``revoked_at = allocated_at``
    (zero billable window — the row was never actually live).
  - Never include ``status='pending'`` rows.

Predates 2026-05-16 these branches had ZERO test coverage —
``billing_service.py`` is the source of truth for finance numbers and
silent regressions here translate directly to over- / under-billing.
A grep across ``server/tests`` returned no hits for ``billing_service``,
``preview_current_or_draft``, or ``_fetch_allocations_overlapping`` prior to this
file. Adding here so future refactors of ``_fetch_allocations_overlapping``
trip a red light before reaching prod.

Each test calls the service directly (no HTTP) — the inclusion logic
lives in one private helper and the routing layer adds no relevant
state; testing through the router would obscure which branch is
broken when a regression lands.
"""
from __future__ import annotations

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

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.asset.billing_service import preview_current_or_draft
from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.identity.person_models import IdentityPerson

# ── Fixtures ──────────────────────────────────────────────────────


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


async def _seat(
    db: AsyncSession,
    *,
    plan_id: uuid.UUID,
    user_id: uuid.UUID,
    allocated_at: datetime,
    revoked_at: datetime | None = None,
    status: str = "active",
) -> SoftwareSeatAllocation:
    seat = SoftwareSeatAllocation(
        id=uuid.uuid4(),
        plan_id=plan_id,
        user_id=user_id,
        status=status,
        allocated_at=allocated_at,
        revoked_at=revoked_at,
    )
    db.add(seat)
    await db.flush()
    return seat


def _current_period() -> str:
    # Match billing_service._current_period (Asia/Shanghai) for the
    # primary assertions — tests should agree with the production
    # period anchor.
    from zoneinfo import ZoneInfo
    return datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y-%m")


def _next_period() -> str:
    cur = _current_period()
    y, m = int(cur[:4]), int(cur[5:])
    if m == 12:
        return f"{y + 1:04d}-01"
    return f"{y:04d}-{m + 1:02d}"


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


# ── Tests ─────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_revoke_mid_month_seat_still_billed_for_current_period(
    db_session: AsyncSession,
):
    """Core invariant: revoke mid-month → current month still billed.

    Allocation lives for >= 1 day in the period; revoked_at lands in
    the period; status='revoked'. The seat must appear in the current
    period bill at the full monthly price — the vendor doesn't refund
    mid-cycle for a monthly-paid plan.
    """
    period = _current_period()
    month_start = _month_start_dt(period)
    # Allocated on the 4th of the current month, revoked on the 16th.
    allocated_at = month_start + timedelta(days=3, hours=10)
    revoked_at = month_start + timedelta(days=15, hours=14)

    _, plan = await _plan(db_session, price=Decimal("20.00"), cycle="monthly")
    holder = await _person(db_session, "revoke-mid@x.com")
    await _seat(
        db_session,
        plan_id=plan.id,
        user_id=holder.id,
        allocated_at=allocated_at,
        revoked_at=revoked_at,
        status="revoked",
    )
    await db_session.commit()

    preview = await preview_current_or_draft(db_session, period)
    matching = [
        line for line in preview["items"]
        if line["user_id"] == holder.id
    ]
    assert len(matching) == 1, (
        "Revoked-mid-month seat must appear in the current period bill"
    )
    line = matching[0]
    assert Decimal(str(line["amount"])) == Decimal("20.00"), (
        "Monthly-paid seat charges the full price even when revoked "
        "mid-period — no proration"
    )
    assert line["status"] == "revoked"


@pytest.mark.asyncio
async def test_revoke_mid_month_seat_excluded_from_next_period(
    db_session: AsyncSession,
):
    """Companion to the above: the seat is OUT of next month's bill.

    Once revoked inside the current period, the seat must not carry
    forward into next period (revoked_at < next period start).
    """
    period = _current_period()
    nxt = _next_period()
    month_start = _month_start_dt(period)
    allocated_at = month_start + timedelta(days=3, hours=10)
    revoked_at = month_start + timedelta(days=15, hours=14)

    _, plan = await _plan(db_session, price=Decimal("20.00"))
    holder = await _person(db_session, "revoke-next-skip@x.com")
    await _seat(
        db_session,
        plan_id=plan.id,
        user_id=holder.id,
        allocated_at=allocated_at,
        revoked_at=revoked_at,
        status="revoked",
    )
    await db_session.commit()

    preview = await preview_current_or_draft(db_session, nxt)
    matching = [
        line for line in preview["items"]
        if line["user_id"] == holder.id
    ]
    assert len(matching) == 0, (
        "Seat revoked inside current period must not appear in next "
        "period's bill — revoked_at falls before next period_start"
    )


@pytest.mark.asyncio
async def test_rejected_at_approval_seat_excluded_from_bill(
    db_session: AsyncSession,
):
    """Approval-rejected row (revoked_at == allocated_at) → no bill.

    The Feishu approval callback sets ``revoked_at = allocated_at``
    for rejections to signal "zero billable window". Billing's
    ``revoked_at > allocated_at`` clause excludes these so the row
    never enters any period's bill.
    """
    period = _current_period()
    month_start = _month_start_dt(period)
    instant = month_start + timedelta(days=5, hours=11)

    _, plan = await _plan(db_session, price=Decimal("20.00"))
    holder = await _person(db_session, "rejected@x.com")
    await _seat(
        db_session,
        plan_id=plan.id,
        user_id=holder.id,
        allocated_at=instant,
        revoked_at=instant,  # equal → zero billable window
        status="revoked",
    )
    await db_session.commit()

    preview = await preview_current_or_draft(db_session, period)
    matching = [
        line for line in preview["items"]
        if line["user_id"] == holder.id
    ]
    assert len(matching) == 0, (
        "Allocations rejected at approval (revoked_at == allocated_at) "
        "must never enter a bill — zero-window guard in "
        "_fetch_allocations_overlapping"
    )


@pytest.mark.asyncio
async def test_pending_seat_excluded_from_bill(db_session: AsyncSession):
    """Pending allocations are not billable until approved.

    Hardened 2026-05-05: previously a pending row that got rejected
    via Feishu without flipping allocation.status='revoked' would
    still bill. Now the explicit status filter
    ``status IN ('active', 'revoked')`` keeps pending rows out.
    """
    period = _current_period()
    month_start = _month_start_dt(period)
    _, plan = await _plan(db_session, price=Decimal("20.00"))
    holder = await _person(db_session, "pending@x.com")
    await _seat(
        db_session,
        plan_id=plan.id,
        user_id=holder.id,
        allocated_at=month_start + timedelta(days=2),
        revoked_at=None,
        status="pending",
    )
    await db_session.commit()

    preview = await preview_current_or_draft(db_session, period)
    matching = [
        line for line in preview["items"]
        if line["user_id"] == holder.id
    ]
    assert len(matching) == 0, (
        "Pending-status seat must not appear in any bill — only "
        "active and revoked count toward billing"
    )


@pytest.mark.asyncio
async def test_active_unrevoked_seat_billed_for_current_period(
    db_session: AsyncSession,
):
    """Sanity baseline: active, never-revoked seat appears at full price."""
    period = _current_period()
    month_start = _month_start_dt(period)
    _, plan = await _plan(db_session, price=Decimal("20.00"))
    holder = await _person(db_session, "active@x.com")
    await _seat(
        db_session,
        plan_id=plan.id,
        user_id=holder.id,
        allocated_at=month_start + timedelta(days=1),
        revoked_at=None,
        status="active",
    )
    await db_session.commit()

    preview = await preview_current_or_draft(db_session, period)
    matching = [
        line for line in preview["items"]
        if line["user_id"] == holder.id
    ]
    assert len(matching) == 1
    assert Decimal(str(matching[0]["amount"])) == Decimal("20.00")
    assert matching[0]["status"] == "active"


@pytest.mark.asyncio
async def test_revoke_on_period_start_not_billed_that_period(
    db_session: AsyncSession,
):
    """Boundary: revoked at 00:00 of period_start → NOT billed that period.

    Security/billing review 2026-07-22 (finding #1): the predicate now
    uses a STRICT ``revoked_at > period_start``. A seat revoked at the
    first instant of a period had ZERO billable time in that period, so
    charging it a full un-prorated month is the extra-month over-bill the
    self-service "cancel effective end-of-month" feature promises to
    avoid (the auto-revoke stamps ``revoked_at`` = start of the next
    month). It is billed the PRIOR period (where it was active), not this
    one. This flips the previously-documented inclusive ``>=`` boundary on
    purpose — see ``billing_service.period_overlap_clauses``.
    """
    period = _current_period()
    month_start = _month_start_dt(period)
    # Allocated last month, revoked exactly at this month's start → it was
    # active through last month, gone at the first tick of this month.
    allocated_at = month_start - timedelta(days=10)
    revoked_at = month_start  # exact tick = period_start

    _, plan = await _plan(db_session, price=Decimal("20.00"))
    holder = await _person(db_session, "boundary-start@x.com")
    await _seat(
        db_session,
        plan_id=plan.id,
        user_id=holder.id,
        allocated_at=allocated_at,
        revoked_at=revoked_at,
        status="revoked",
    )
    await db_session.commit()

    preview = await preview_current_or_draft(db_session, period)
    matching = [
        line for line in preview["items"]
        if line["user_id"] == holder.id
    ]
    assert len(matching) == 0, (
        "Seat revoked exactly at period_start had zero billable time this "
        "period → must NOT bill (strict > boundary, finding #1)"
    )

    # And it MUST still bill the PRIOR period (where it was actually held)
    # — the fix must not drop the charge entirely, only move it off the
    # wrong month.
    _y, _m = int(period[:4]), int(period[5:])
    prior_period = f"{_y - 1:04d}-12" if _m == 1 else f"{_y:04d}-{_m - 1:02d}"
    prior_preview = await preview_current_or_draft(db_session, prior_period)
    prior_matching = [
        line for line in prior_preview["items"]
        if line["user_id"] == holder.id
    ]
    assert len(prior_matching) == 1, (
        "Seat active all of the prior month must bill that prior period"
    )


@pytest.mark.asyncio
async def test_yearly_seat_revoke_mid_month_bills_one_twelfth(
    db_session: AsyncSession,
):
    """Yearly cycle: per-month contribution = price/12 even on revoke."""
    period = _current_period()
    month_start = _month_start_dt(period)
    _, plan = await _plan(
        db_session, price=Decimal("240.00"), cycle="yearly",
    )
    holder = await _person(db_session, "yearly-revoke@x.com")
    await _seat(
        db_session,
        plan_id=plan.id,
        user_id=holder.id,
        allocated_at=month_start + timedelta(days=2),
        revoked_at=month_start + timedelta(days=20),
        status="revoked",
    )
    await db_session.commit()

    preview = await preview_current_or_draft(db_session, period)
    matching = [
        line for line in preview["items"]
        if line["user_id"] == holder.id
    ]
    assert len(matching) == 1
    # 240 / 12 = 20.00
    assert Decimal(str(matching[0]["amount"])) == Decimal("20.00"), (
        "Yearly plan revoked mid-month still bills price/12 for the "
        "month — no daily proration"
    )


@pytest.mark.asyncio
async def test_one_time_seat_revoke_mid_month_bills_only_alloc_month(
    db_session: AsyncSession,
):
    """One-time cycle: charged in allocation month, not in revoke-only
    months."""
    period = _current_period()
    nxt = _next_period()
    month_start = _month_start_dt(period)
    _, plan = await _plan(
        db_session, price=Decimal("99.00"), cycle="one_time",
    )
    holder = await _person(db_session, "onetime-revoke@x.com")
    # Allocated and revoked both inside the current month.
    await _seat(
        db_session,
        plan_id=plan.id,
        user_id=holder.id,
        allocated_at=month_start + timedelta(days=5),
        revoked_at=month_start + timedelta(days=10),
        status="revoked",
    )
    await db_session.commit()

    # Current period — one-time charge applies (allocation month).
    cur_preview = await preview_current_or_draft(db_session, period)
    cur_match = [
        line for line in cur_preview["items"]
        if line["user_id"] == holder.id
    ]
    assert len(cur_match) == 1
    assert Decimal(str(cur_match[0]["amount"])) == Decimal("99.00")

    # Next period — one-time outside allocation month → zero, dropped.
    nxt_preview = await preview_current_or_draft(db_session, nxt)
    nxt_match = [
        line for line in nxt_preview["items"]
        if line["user_id"] == holder.id
    ]
    assert len(nxt_match) == 0, (
        "One-time seat outside its allocation month must drop from the "
        "bill — zero-amount rows are omitted"
    )