"""Cross-endpoint numeric invariants for the transfer scenario.

Each test relates two or more *live* endpoint responses for product 5244974,
so it is immune to the run-to-run data drift that makes absolute snapshots
fragile. If the v2 permission filter is correct, these relationships must
hold.

A test skips (rather than fails) when an endpoint returns zero streams — that
means the QA data / profile provisioning is not in place yet, which the
verification pass (README §9) addresses.
"""

from __future__ import annotations

import pytest

from analytics import config
from tests.integration.endpoints.conftest import PAGINATION, assert_endpoint
from tests.integration.transfer_ownership.conftest import (
    ALA_VENDOR,
    DESTINATION_ARTIST,
    EMPLOYEE,
    ORIGINATING_ACCESS_UNTIL,
    ORIGINATING_ARTIST,
    PRODUCT_ID,
    PRODUCT_ISRCS,
    SUED_VENDOR,
)

# Whole-history window. Product 5244974's earliest stream day is 2024-10-24
# (per FACTS.PROD.V_STREAMS_BY_TRACK_FEED_DISTRIBUTOR_DAILY); the upper bound
# trails today by a couple of days to absorb late-arriving fact-table writes.
_FULL = "start_date=2024-10-24&end_date=2026-05-25"


def _product_total(headers, params: str) -> int:
    """TOTAL streams from /product/5244974/summary for the given query params."""
    base = config.PRODUCT_SUMMARY_URL.replace("<product_id>", PRODUCT_ID)
    payload = assert_endpoint(
        f"{base}?type=TOTAL&{params}&{PAGINATION}", headers=headers
    )
    items = payload["items"]
    return int(items[0].get("streams") or 0) if items else 0


def _sr_total(headers, isrc: str, params: str) -> int:
    """TOTAL streams from /sound-recording/<isrc>/summary."""
    base = config.SOUND_RECORDING_SUMMARY_URL.replace("<isrc>", isrc)
    payload = assert_endpoint(
        f"{base}?type=TOTAL&{params}&{PAGINATION}", headers=headers
    )
    items = payload["items"]
    return int(items[0].get("streams") or 0) if items else 0


def _aggregate_all_time(headers) -> int:
    """streams_all_time from the ROLLUP-backed /product/5244974/aggregate-streams."""
    base = config.PRODUCT_AGGREGATE_STREAMS_URL.replace("<product_id>", PRODUCT_ID)
    payload = assert_endpoint(base, headers=headers, items_key=None)
    return int(payload.get("streams_all_time") or 0)


class TestNoForwardLeak:
    """A former owner sees nothing after their cutoff — so moving the query's
    end_date from their access_until_date (the inclusive last day of their
    tenure) up to today must add exactly zero streams."""

    def test_window_end_irrelevant(self):
        """former owner: total through its cutoff == total through today."""
        to_cutoff = _product_total(
            ALA_VENDOR,
            f"start_date=2024-10-24&end_date={ORIGINATING_ACCESS_UNTIL}",
        )
        if to_cutoff == 0:
            pytest.skip("former owner sees no streams — QA data/POA not in place")
        to_today = _product_total(ALA_VENDOR, _FULL)
        assert to_cutoff == to_today, (
            f"former owner leaked post-transfer streams: "
            f"to-cutoff={to_cutoff}, to-today={to_today}"
        )


class TestNestedContainment:
    """Each owner's window is a subset of the next: former ⊆ full.

    The destination (current) owner sees the full lifetime, matching the
    employee. Doubles as a no-fanout guard — if the current owner's total
    were inflated by the multi-row PRODUCT_OWNERSHIP_ACCESS join it would
    not stay within 2% of the full-access employee total.
    """

    def test_former_owner_sees_subset(self):
        """ala <= sued, and employee ≈ sued (current owner)."""
        ala = _product_total(ALA_VENDOR, _FULL)
        sued = _product_total(SUED_VENDOR, _FULL)
        emp = _product_total(EMPLOYEE, _FULL)
        if not all((ala, sued, emp)):
            pytest.skip("a profile sees no streams — QA data/POA not in place")
        assert ala <= sued, f"windows are not nested: former={ala}, sued={sued}"
        assert ala < sued, (
            f"former owner ({ala}) sees as much as the current owner ({sued}) — "
            f"streams are not being time-sliced at all"
        )
        assert abs(emp - sued) <= 0.02 * sued, (
            f"employee ({emp}) and current owner sued ({sued}) disagree on "
            f"the product's lifetime total"
        )


class TestNoRollupFanout:
    """The *_PRODUCT_TRANSFER rollup holds one row per owner (2 for product
    5244974 once POA reflects the transfer). A correct filter selects a
    single owner's row-set; a broken one could sum all of them. ROLLUP
    all-time must therefore match the (fanout-proof, EXISTS-filtered) DAILY
    total — checked for both the employee (whose `is_current` clamp picks one
    row) and the former owner (whose POA scope picks the other)."""

    def test_employee_rollup_matches_daily(self):
        """aggregate-streams streams_all_time ≈ summary DAILY total for employee."""
        daily = _product_total(EMPLOYEE, _FULL)
        rollup = _aggregate_all_time(EMPLOYEE)
        if not daily or not rollup:
            pytest.skip("employee sees no streams — QA data not in place")
        assert abs(rollup - daily) <= 0.03 * daily, (
            f"rollup all-time ({rollup}) diverges from daily total ({daily}) — "
            f"possible {rollup / daily:.2f}x fanout across owner rows"
        )

    def test_former_owner_rollup_matches_daily(self):
        """aggregate-streams streams_all_time ≈ summary DAILY total for ala.

        ala_vendor reads its time-sliced row-set from the _PRODUCT_TRANSFER
        rollup (cut at 2026-04-30); the v2 DAILY path is gated by the same
        EXISTS join on PRODUCT_OWNERSHIP_ACCESS. The two grains must agree
        on ala's W1 slice — a divergence means the dbt `transfer_time_slice`
        macro that builds the rollup and the v2 DAILY macro disagree on what
        `<= access_until_date` includes.
        """
        daily = _product_total(ALA_VENDOR, _FULL)
        rollup = _aggregate_all_time(ALA_VENDOR)
        if not daily or not rollup:
            pytest.skip("ala sees no streams — QA data/POA not in place")
        assert abs(rollup - daily) <= 0.03 * daily, (
            f"ala rollup all-time ({rollup}) diverges from daily total "
            f"({daily}) — dbt transfer_time_slice and v2 DAILY macro disagree "
            f"on the cutoff boundary"
        )


class TestWindowAdditivity:
    """For each owner the DAILY windows partition their view of the product:
    two contiguous halves must sum to the whole. Tested for both the current
    owner (whole lifetime) and the former owner (lifetime-up-to-cutoff)."""

    def test_sued_halves_sum_to_full(self):
        """sued: total(2024-10-24..2025-12-31) + total(2026-01-01..2026-05-25)
        ≈ total(full)."""
        first = _product_total(SUED_VENDOR, "start_date=2024-10-24&end_date=2025-12-31")
        second = _product_total(
            SUED_VENDOR, "start_date=2026-01-01&end_date=2026-05-25"
        )
        full = _product_total(SUED_VENDOR, _FULL)
        if not full:
            pytest.skip("sued sees no streams — QA data not in place")
        assert abs((first + second) - full) <= 0.01 * full, (
            f"window halves ({first} + {second} = {first + second}) do not sum "
            f"to the full-range total ({full})"
        )

    def test_ala_halves_sum_to_full(self):
        """ala: total(2024-10-24..2025-12-31) + total(2026-01-01..2026-04-30)
        ≈ total(full).

        The upper half ends at ala's cutoff because v2 truncates further dates
        away anyway; the `full` query (which extends past the cutoff) is also
        truncated by v2 to W1. So the explicit-cutoff halves and the
        implicit-truncated full must agree — a divergence means v2 lost (or
        double-counted) streams across the contiguous-window split.
        """
        first = _product_total(ALA_VENDOR, "start_date=2024-10-24&end_date=2025-12-31")
        second = _product_total(ALA_VENDOR, "start_date=2026-01-01&end_date=2026-04-30")
        full = _product_total(ALA_VENDOR, _FULL)
        if not full:
            pytest.skip("ala sees no streams — QA data/POA not in place")
        assert abs((first + second) - full) <= 0.01 * full, (
            f"former owner's halves ({first} + {second} = {first + second}) "
            f"do not sum to the full-range total ({full})"
        )


class TestSliceAdditivityAcrossOwners:
    """Disjoint owner slices reconstruct the lifetime — the canonical
    time-slicing identity across the handoff day.

    Under FF-ON the v2 path truncates ala at her access_until_date, so
    ala_vendor's `full` query returns sum(streams) over [start..2026-04-30].
    sued_vendor queried from 2026-05-01 onward returns sum(streams) over
    [2026-05-01..end]. The two disjoint date ranges abut at the handoff and
    together cover the full window — so their totals must add up to the
    employee's lifetime total.

    Integrates across the transfer boundary in a way no single-profile test
    does: a missed or doubled day on 2026-04-30 ↔ 2026-05-01 breaks this
    identity even if every per-profile window check still passes.
    """

    def test_ala_full_plus_sued_post_cutoff_eq_lifetime(self):
        """ala(full) + sued(2026-05-01..2026-05-25) ≈ employee(full)."""
        ala_full = _product_total(ALA_VENDOR, _FULL)
        sued_post = _product_total(
            SUED_VENDOR, "start_date=2026-05-01&end_date=2026-05-25"
        )
        lifetime = _product_total(EMPLOYEE, _FULL)
        if not all((ala_full, sued_post, lifetime)):
            pytest.skip("a profile sees no streams — QA data/POA not in place")
        recombined = ala_full + sued_post
        assert abs(recombined - lifetime) <= 0.02 * lifetime, (
            f"slices don't reconstruct the lifetime: ala_full ({ala_full}) + "
            f"sued_post_cutoff ({sued_post}) = {recombined}, but lifetime is "
            f"{lifetime} — a day was lost or double-counted across the "
            f"handoff (2026-04-30 ↔ 2026-05-01)"
        )


class TestIsrcSumMatchesProduct:
    """The product TOTAL equals the sum of its 13 ISRC TOTALs — the per-ISRC
    and per-product permission filters must agree."""

    def test_employee_isrc_sum(self):
        """sum over 13 ISRC summaries ≈ product summary, for the employee."""
        product = _product_total(EMPLOYEE, _FULL)
        if not product:
            pytest.skip("employee sees no streams — QA data not in place")
        isrc_sum = sum(_sr_total(EMPLOYEE, isrc, _FULL) for isrc in PRODUCT_ISRCS)
        assert abs(isrc_sum - product) <= 0.02 * product, (
            f"sum of {len(PRODUCT_ISRCS)} ISRC totals ({isrc_sum}) diverges "
            f"from the product total ({product})"
        )


# ===========================================================================
# Participation-axis invariants — the originating & destination artists
# ===========================================================================
#
# The same cross-endpoint identities, but for the two artist viewers (FF-ON, so
# the v2 participation path runs). The originating artist is the former
# participant (USED_TO grant, frozen at the move cutoff — same shape as
# ALA_VENDOR); the destination artist is the current participant (full lifetime
# — same shape as SUED_VENDOR). Each test skips when an endpoint returns zero
# streams (QA data / the subaccount-normalization macro not yet in the queried
# env — see test_used_to_participate.py module docstring).

ORIGINATING_ARTIST_ON = ORIGINATING_ARTIST.ff_on
DESTINATION_ARTIST_ON = DESTINATION_ARTIST.ff_on


class TestArtistNoForwardLeak:
    """Former participant sees nothing after its cutoff — so moving the query's
    end_date from its move cutoff up to today must add exactly zero streams."""

    def test_window_end_irrelevant(self):
        """originating artist: total through its cutoff == total through today."""
        to_cutoff = _product_total(
            ORIGINATING_ARTIST_ON,
            f"start_date=2024-10-24&end_date={ORIGINATING_ACCESS_UNTIL}",
        )
        if to_cutoff == 0:
            pytest.skip(
                "former participant sees no streams — QA data/macro not in place"
            )
        to_today = _product_total(ORIGINATING_ARTIST_ON, _FULL)
        assert to_cutoff == to_today, (
            f"former participant leaked post-transfer streams: "
            f"to-cutoff={to_cutoff}, to-today={to_today}"
        )


class TestArtistNestedContainment:
    """Former participant's window is a subset of the current participant's, and
    the current participant sees the full lifetime (== employee)."""

    def test_former_participant_sees_subset(self):
        """originating <= destination, and employee ≈ destination (lifetime)."""
        orig = _product_total(ORIGINATING_ARTIST_ON, _FULL)
        dest = _product_total(DESTINATION_ARTIST_ON, _FULL)
        emp = _product_total(EMPLOYEE, _FULL)
        if not all((orig, dest, emp)):
            pytest.skip("a profile sees no streams — QA data/macro not in place")
        assert orig <= dest, f"windows not nested: orig={orig}, dest={dest}"
        assert orig < dest, (
            f"former participant ({orig}) sees as much as the current "
            f"participant ({dest}) — streams are not being time-sliced at all"
        )
        assert abs(emp - dest) <= 0.02 * dest, (
            f"employee ({emp}) and current participant ({dest}) disagree on "
            f"the product's lifetime total"
        )


class TestArtistNoRollupFanout:
    """ROLLUP all-time must match the DAILY total for each artist — a broken
    participation filter could sum every owner row instead of selecting one."""

    def test_originating_rollup_matches_daily(self):
        """aggregate-streams all-time ≈ summary DAILY total for the originating artist."""
        daily = _product_total(ORIGINATING_ARTIST_ON, _FULL)
        rollup = _aggregate_all_time(ORIGINATING_ARTIST_ON)
        if not daily or not rollup:
            pytest.skip("originating artist sees no streams — QA data/macro not ready")
        assert abs(rollup - daily) <= 0.03 * daily, (
            f"originating rollup all-time ({rollup}) diverges from daily total "
            f"({daily}) — dbt transfer_time_slice and the v2 DAILY participation "
            f"macro disagree on the cutoff boundary"
        )

    def test_destination_rollup_matches_daily(self):
        """aggregate-streams all-time ≈ summary DAILY total for the destination artist."""
        daily = _product_total(DESTINATION_ARTIST_ON, _FULL)
        rollup = _aggregate_all_time(DESTINATION_ARTIST_ON)
        if not daily or not rollup:
            pytest.skip("destination artist sees no streams — QA data not ready")
        assert abs(rollup - daily) <= 0.03 * daily, (
            f"destination rollup all-time ({rollup}) diverges from daily total "
            f"({daily}) — possible {rollup / daily:.2f}x fanout across owner rows"
        )


class TestArtistSliceAdditivityAcrossOwners:
    """Disjoint participation slices reconstruct the lifetime across the handoff.

    The originating artist's `full` query is truncated by v2 to [start..cutoff];
    the destination artist queried from 2026-05-01 onward covers [cutoff+1..end].
    The two disjoint ranges abut at the handoff and together must add up to the
    employee's lifetime total — a missed or doubled handoff day breaks this even
    if every per-profile window check still passes.
    """

    def test_originating_full_plus_destination_post_cutoff_eq_lifetime(self):
        """originating(full) + destination(2026-05-01..2026-05-25) ≈ employee(full)."""
        orig_full = _product_total(ORIGINATING_ARTIST_ON, _FULL)
        dest_post = _product_total(
            DESTINATION_ARTIST_ON, "start_date=2026-05-01&end_date=2026-05-25"
        )
        lifetime = _product_total(EMPLOYEE, _FULL)
        if not all((orig_full, dest_post, lifetime)):
            pytest.skip("a profile sees no streams — QA data/macro not in place")
        recombined = orig_full + dest_post
        assert abs(recombined - lifetime) <= 0.02 * lifetime, (
            f"slices don't reconstruct the lifetime: orig_full ({orig_full}) + "
            f"dest_post_cutoff ({dest_post}) = {recombined}, but lifetime is "
            f"{lifetime} — a day was lost or double-counted across the handoff"
        )


class TestArtistIsrcSumMatchesProduct:
    """The product TOTAL equals the sum of its 13 ISRC TOTALs for the current
    participant — the per-ISRC and per-product participation filters must
    agree."""

    def test_destination_isrc_sum(self):
        """sum over 13 ISRC summaries ≈ product summary, for the destination artist."""
        product = _product_total(DESTINATION_ARTIST_ON, _FULL)
        if not product:
            pytest.skip("destination artist sees no streams — QA data not in place")
        isrc_sum = sum(
            _sr_total(DESTINATION_ARTIST_ON, isrc, _FULL) for isrc in PRODUCT_ISRCS
        )
        assert abs(isrc_sum - product) <= 0.02 * product, (
            f"sum of {len(PRODUCT_ISRCS)} ISRC totals ({isrc_sum}) diverges "
            f"from the product total ({product})"
        )