"""Regression: the top "生效坐席" stat must count each vendor product
*family* ONCE, not once per catalog row / license plan that carries its
org-wide count.

Background — the bug this guards against (deferred from Round-56 #589,
re-confirmed Round-57 #592, fixed here):

``software_seat_sync_stat.vendor_license_holders`` is an ORG-WIDE,
site-agnostic count of a vendor product family keyed by ``product_key``
(the count side uses ``atlassian_activity.all_license_holders(family)`` —
no site filter), written once per *license plan*. The read/display layer
SUMmed it across plans and across software, so the same family got counted
multiple times. Two manifestations, both verified on prod 2026-05-30:

1. Cross-software — one family modelled as several catalog rows. The
   ``jira-software`` family is four catalogs (Jira Software + the Tempo /
   ScriptRunner / eazyBI marketplace addons), each carrying the identical
   org-wide 364 → the top card summed to 1456. Confluence is one product
   split per site (tidb + PingCAP US) → summed both. Top card read 3241
   vs the real 2119 (a 1122-seat / 53% inflation).
2. Same-software multi-plan — two active plans on one software_id share a
   product_key (allowed: ``SoftwareLicensePlan`` has no per-software
   uniqueness); each gets a stat row with the full family count, so the
   per-software SUM double-counts too.

Invariant under test: rows sharing a ``product_key`` carry the identical
family count and contribute it ONCE; distinct ``product_key`` families sum
normally; a never-synced catalog falls back to its distinct-user count.
"""
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 _current_period, list_period_bills
from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
    SoftwareSeatSyncStat,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration

pytestmark = pytest.mark.asyncio


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


async def _person(db: AsyncSession, tag: str) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"{tag}-{uuid.uuid4().hex[:6]}@x.com",
        display_name=tag,
        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(db: AsyncSession, name: str) -> AssetSoftwareCatalog:
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(), name=f"{name}-{uuid.uuid4().hex[:6]}",
        asset_type="SaaS", status="active",
    )
    db.add(sw)
    await db.flush()
    return sw


async def _plan(
    db: AsyncSession, software_id: uuid.UUID, max_seats: int | None = None,
) -> SoftwareLicensePlan:
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=software_id, name=f"Plan-{uuid.uuid4().hex[:4]}",
        price=Decimal("20.00"), billing_cycle="monthly", currency="USD",
        status="active", max_seats=max_seats,
    )
    db.add(plan)
    await db.flush()
    return plan


async def _allocate(
    db: AsyncSession, plan_id: uuid.UUID, period: str, n: int,
) -> list[IdentityPerson]:
    month_start = _month_start_dt(period)
    people = []
    for _ in range(n):
        person = await _person(db, "holder")
        db.add(SoftwareSeatAllocation(
            id=uuid.uuid4(), plan_id=plan_id, user_id=person.id,
            status="active", allocated_at=month_start + timedelta(days=1),
            revoked_at=None,
        ))
        people.append(person)
    await db.flush()
    return people


async def _integration(db: AsyncSession) -> Integration:
    integration = Integration(
        id=uuid.uuid4(), kind="software_provisioning", provider="atlassian",
        name="Test Atlassian", slug=f"atl-{uuid.uuid4().hex[:6]}",
        enabled=True, is_primary=False,
    )
    db.add(integration)
    await db.flush()
    return integration


async def _stat(
    db: AsyncSession, *, integration_id, software_id, plan_id,
    product_key: str | None, vendor_count: int, matched: int,
):
    db.add(SoftwareSeatSyncStat(
        integration_id=integration_id, software_id=software_id, plan_id=plan_id,
        provider="atlassian", product_key=product_key,
        vendor_license_holders=vendor_count,
        holders_matched=matched, holders_external=vendor_count - matched,
        holders_no_email=0,
    ))


async def test_two_catalogs_same_product_key_counted_once(
    db_session: AsyncSession,
):
    """Cross-software: the ``jira-software`` family modelled as two catalog
    rows (e.g. Jira Software + a Tempo marketplace addon), each carrying the
    identical org-wide 364, must contribute 364 to the top stat ONCE — not
    728. Each per-software cell still shows its own 364 (matches the drawer).
    """
    period = _current_period()
    integ = await _integration(db_session)
    sw_a = await _software(db_session, "Jira-Software")
    sw_b = await _software(db_session, "Jira-Tempo")
    plan_a = await _plan(db_session, sw_a.id)
    plan_b = await _plan(db_session, sw_b.id)
    await _allocate(db_session, plan_a.id, period, 2)
    await _allocate(db_session, plan_b.id, period, 1)
    await _stat(
        db_session, integration_id=integ.id, software_id=sw_a.id,
        plan_id=plan_a.id, product_key="jira-software", vendor_count=364,
        matched=2,
    )
    await _stat(
        db_session, integration_id=integ.id, software_id=sw_b.id,
        plan_id=plan_b.id, product_key="jira-software", vendor_count=364,
        matched=1,
    )
    await db_session.commit()

    data = await list_period_bills(db_session, period)
    # Family counted once for the top stat.
    assert data["active_seats"] == 364
    # Each per-software cell still surfaces the family's org-wide count.
    assert data["vendor_seats_by_software"][str(sw_a.id)] == 364
    assert data["vendor_seats_by_software"][str(sw_b.id)] == 364


async def test_two_plans_same_software_same_product_key_counted_once(
    db_session: AsyncSession,
):
    """Same-software multi-plan: two active plans on ONE software sharing a
    ``product_key`` (allowed config) each get a stat row with the full
    family count. The per-software total must dedup to the family count
    ONCE, not sum to double — both the per-software cell and the top stat.
    """
    period = _current_period()
    integ = await _integration(db_session)
    sw = await _software(db_session, "Confluence")
    plan_1 = await _plan(db_session, sw.id)
    plan_2 = await _plan(db_session, sw.id)
    await _allocate(db_session, plan_1.id, period, 3)
    await _allocate(db_session, plan_2.id, period, 2)
    await _stat(
        db_session, integration_id=integ.id, software_id=sw.id,
        plan_id=plan_1.id, product_key="confluence", vendor_count=107,
        matched=3,
    )
    await _stat(
        db_session, integration_id=integ.id, software_id=sw.id,
        plan_id=plan_2.id, product_key="confluence", vendor_count=107,
        matched=2,
    )
    await db_session.commit()

    data = await list_period_bills(db_session, period)
    # Per-software cell shows the family count once (not 214).
    assert data["vendor_seats_by_software"][str(sw.id)] == 107
    assert data["active_seats"] == 107


async def test_distinct_product_keys_sum_normally(db_session: AsyncSession):
    """Two software with DISTINCT families (``jira-software`` 364 +
    ``confluence`` 107) must sum to 471 — dedup only collapses WITHIN a
    family, never across genuinely different products.
    """
    period = _current_period()
    integ = await _integration(db_session)
    sw_a = await _software(db_session, "Jira-Software")
    sw_b = await _software(db_session, "Confluence")
    plan_a = await _plan(db_session, sw_a.id)
    plan_b = await _plan(db_session, sw_b.id)
    await _allocate(db_session, plan_a.id, period, 2)
    await _allocate(db_session, plan_b.id, period, 2)
    await _stat(
        db_session, integration_id=integ.id, software_id=sw_a.id,
        plan_id=plan_a.id, product_key="jira-software", vendor_count=364,
        matched=2,
    )
    await _stat(
        db_session, integration_id=integ.id, software_id=sw_b.id,
        plan_id=plan_b.id, product_key="confluence", vendor_count=107,
        matched=2,
    )
    await db_session.commit()

    data = await list_period_bills(db_session, period)
    assert data["active_seats"] == 364 + 107
    assert data["vendor_seats_by_software"][str(sw_a.id)] == 364
    assert data["vendor_seats_by_software"][str(sw_b.id)] == 107


async def test_never_synced_catalog_uses_distinct_user_fallback(
    db_session: AsyncSession,
):
    """A catalog with NO ``software_seat_sync_stat`` row falls back to the
    distinct ``user_id`` count among its billable rows — the dedup change
    must not disturb the fallback path. Three allocations across two
    distinct users → 2 seats.
    """
    period = _current_period()
    sw = await _software(db_session, "In-House-Tool")
    plan = await _plan(db_session, sw.id)
    p1 = await _person(db_session, "u1")
    p2 = await _person(db_session, "u2")
    month_start = _month_start_dt(period)
    for user_id in (p1.id, p2.id, p1.id):  # p1 appears twice (re-grant)
        db_session.add(SoftwareSeatAllocation(
            id=uuid.uuid4(), plan_id=plan.id, user_id=user_id,
            status="active", allocated_at=month_start + timedelta(days=1),
            revoked_at=None,
        ))
    await db_session.commit()

    data = await list_period_bills(db_session, period)
    assert data["active_seats"] == 2
    assert data.get("vendor_seats_by_software", {}) == {}


async def test_null_product_key_plans_stay_distinct(db_session: AsyncSession):
    """Rows with NULL ``product_key`` (e.g. Slack, in-house catalogs) have
    no family key to dedup on, so distinct plans must still sum — they key
    on plan id, never collapsing two unrelated unkeyed products together.
    """
    period = _current_period()
    integ = await _integration(db_session)
    sw_a = await _software(db_session, "Slack")
    sw_b = await _software(db_session, "OtherUnkeyed")
    plan_a = await _plan(db_session, sw_a.id)
    plan_b = await _plan(db_session, sw_b.id)
    await _allocate(db_session, plan_a.id, period, 1)
    await _allocate(db_session, plan_b.id, period, 1)
    await _stat(
        db_session, integration_id=integ.id, software_id=sw_a.id,
        plan_id=plan_a.id, product_key=None, vendor_count=423, matched=1,
    )
    await _stat(
        db_session, integration_id=integ.id, software_id=sw_b.id,
        plan_id=plan_b.id, product_key=None, vendor_count=10, matched=1,
    )
    await db_session.commit()

    data = await list_period_bills(db_session, period)
    # Two distinct unkeyed products sum (no spurious family collapse).
    assert data["active_seats"] == 423 + 10


async def test_access_clamped_to_contracted_max_seats(db_session: AsyncSession):
    """Under-licensing (prod Confluence 2026-05-31): the vendor product_access
    reports ~107 active holders, but the subscription is contracted for only
    ``max_seats=10`` (broad Confluence access that does not consume a paid
    license). The "生效坐席" headline must show the contracted paid count (10),
    NOT the raw 107 access count — clamp each plan to its ceiling.
    """
    period = _current_period()
    integ = await _integration(db_session)
    sw = await _software(db_session, "Confluence")
    plan = await _plan(db_session, sw.id, max_seats=10)
    await _allocate(db_session, plan.id, period, 4)
    await _stat(
        db_session, integration_id=integ.id, software_id=sw.id,
        plan_id=plan.id, product_key="confluence", vendor_count=107,
        matched=99,
    )
    await db_session.commit()

    data = await list_period_bills(db_session, period)
    # Clamped to the contracted ceiling, not the 107 raw access holders.
    assert data["vendor_seats_by_software"][str(sw.id)] == 10
    assert data["active_seats"] == 10


async def test_healthy_product_below_ceiling_not_clamped(
    db_session: AsyncSession,
):
    """Jira (prod): 372 holders against a 400-seat contract — holders ≤ ceiling,
    so the clamp is a no-op and the headline still shows the real 372, never
    inflated up to the 400 ceiling.
    """
    period = _current_period()
    integ = await _integration(db_session)
    sw = await _software(db_session, "Jira-Software")
    plan = await _plan(db_session, sw.id, max_seats=400)
    await _allocate(db_session, plan.id, period, 5)
    await _stat(
        db_session, integration_id=integ.id, software_id=sw.id,
        plan_id=plan.id, product_key="jira-software", vendor_count=372,
        matched=370,
    )
    await db_session.commit()

    data = await list_period_bills(db_session, period)
    assert data["vendor_seats_by_software"][str(sw.id)] == 372
    assert data["active_seats"] == 372


async def test_effective_seats_helper_unit():
    """Pure-function guard on the clamp itself."""
    from app.domains.asset.billing_service import _effective_seats

    assert _effective_seats(372, 400) == 372  # healthy: no-op
    assert _effective_seats(107, 10) == 10     # Confluence: clamp to contract
    assert _effective_seats(10, 10) == 10      # exactly at ceiling
    assert _effective_seats(30, None) == 30    # no contract on file: raw count
    assert _effective_seats(None, 10) == 0     # no holders