"""Transfer-ownership tests for the /sound-recording/<isrc>/tiktok/* endpoints.

The four TikTok endpoints are transfer-aware: their SQL runs through the v2
`permissions_filter` and swaps to the `TIKTOK_BY_ISRC_PRODUCT_COUNTRY_ROLLUP`
`_PRODUCT_TRANSFER` variant when the flag is on. ISRC BX69Y2100018 belongs
only to product 5244974 (see conftest.ISRC), so every request here isolates
the transferred product.

Unlike streaming data, QA TikTok coverage for the product is not guaranteed
(README §9 originally deferred TikTok for this reason) — so these tests never
*require* TikTok data to exist. What they pin instead, via the FF-pair, are
the two facts that hold whether or not data is present:

* **no legacy leak** — a former owner sees nothing of the transferred-away
  product without the flag (FF-OFF total is 0);
* **no forward leak** — with the flag on, a former owner sees no TikTok
  activity dated after their transfer cutoff.

When TikTok data *does* exist the differential additionally proves the flag is
a no-op for the current owner / employee and a pure grant for a former owner.
"""

from __future__ import annotations

import pytest

from analytics import config
from tests.integration.endpoints.conftest import assert_endpoint
from tests.integration.transfer_ownership.conftest import (
    ARTIST_PROFILE_PAIRS,
    ISRC,
    PROBES,
    TRANSFER_PROFILE_PAIRS,
    WINDOW_PROBES,
    assert_ff_monotonic,
    assert_ff_noop,
    assert_isolated_transfer,
    expected_visibility,
    sum_metric,
)

# TikTok timeseries/summary rows carry creation / view counts; sum them all so
# the check is robust to which metric a given `type`/`order_by` surfaces.
_TIKTOK_METRICS = ("creations", "views", "value")


def _tiktok_total(items: list[dict]) -> float:
    """Sum every TikTok count metric across timeseries / summary items."""
    return sum(sum_metric(items, metric) for metric in _TIKTOK_METRICS)


def _tiktok_url(template: str, *, params: str = "") -> str:
    """Substitute the canonical ISRC into a TikTok URL template."""
    base = template.replace("<isrc>", ISRC)
    return f"{base}?{params}" if params else base


class TestTikTokTimeseries:
    """/sound-recording/<isrc>/tiktok/timeseries — DAILY grain, v2 macro."""

    @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS)
    @pytest.mark.parametrize("probe", WINDOW_PROBES)
    def test_streams(self, pair, probe):
        """TikTok timeseries is correctly time-sliced for the FF-pair."""
        url = _tiktok_url(
            config.SOUND_RECORDING_TIKTOK_TIMESERIES_URL, params=PROBES[probe]
        )
        on = assert_endpoint(url, headers=pair.ff_on)
        off = assert_endpoint(url, headers=pair.ff_off)
        on_total = _tiktok_total(on["items"])
        off_total = _tiktok_total(off["items"])
        assert_isolated_transfer(
            on_total,
            off_total,
            role_key=pair.key,
            label=f"{pair.key} tiktok/timeseries probe={probe}",
            require_data=False,
        )
        if expected_visibility(pair.key, probe) == "empty":
            assert on_total == 0, (
                f"{pair.key} leaked {on_total} TikTok activity into probe "
                f"{probe} — past the owner's transfer cutoff"
            )


class TestTikTokSummary:
    """/sound-recording/<isrc>/tiktok/summary — DAILY grain, windowed summary."""

    @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS)
    @pytest.mark.parametrize("probe", WINDOW_PROBES)
    def test_total(self, pair, probe):
        """TikTok summary is correctly time-sliced for the FF-pair."""
        url = _tiktok_url(
            config.SOUND_RECORDING_TIKTOK_SUMMARY_URL, params=PROBES[probe]
        )
        on = assert_endpoint(url, headers=pair.ff_on)
        off = assert_endpoint(url, headers=pair.ff_off)
        on_total = _tiktok_total(on["items"])
        off_total = _tiktok_total(off["items"])
        assert_isolated_transfer(
            on_total,
            off_total,
            role_key=pair.key,
            label=f"{pair.key} tiktok/summary probe={probe}",
            require_data=False,
        )
        if expected_visibility(pair.key, probe) == "empty":
            assert on_total == 0, (
                f"{pair.key} leaked {on_total} TikTok activity into probe "
                f"{probe} — past the owner's transfer cutoff"
            )


class TestTikTokAggregatedSummary:
    """/sound-recording/<isrc>/tiktok/aggregated-summary — ROLLUP, all-time."""

    @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS)
    def test_all_time(self, pair):
        """All-time TikTok aggregates obey the transfer scoping for the FF-pair."""
        url = _tiktok_url(config.SOUND_RECORDING_TIKTOK_AGGREGATED_SUMMARY_URL)
        on = assert_endpoint(url, headers=pair.ff_on, items_key=None)
        off = assert_endpoint(url, headers=pair.ff_off, items_key=None)
        assert isinstance(on, dict) and isinstance(off, dict)
        on_total = (on.get("creations_all_time") or 0) + (on.get("views_all_time") or 0)
        off_total = (off.get("creations_all_time") or 0) + (
            off.get("views_all_time") or 0
        )
        assert_isolated_transfer(
            on_total,
            off_total,
            role_key=pair.key,
            label=f"{pair.key} tiktok/aggregated-summary all-time",
            require_data=False,
        )


class TestTikTokAggregatedTimeseries:
    """/sound-recording/<isrc>/tiktok/aggregated-timeseries — ROLLUP, recent window.

    The aggregated-timeseries window is recent-only, so a former owner — frozen
    well before it — contributes nothing under either flag. The differential is
    therefore a no-op check: the v2 ROLLUP path must reproduce the legacy
    payload exactly for every profile.
    """

    @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS)
    def test_default(self, pair):
        """The v2 aggregated-timeseries path is a no-op over the recent window."""
        url = _tiktok_url(config.SOUND_RECORDING_TIKTOK_AGGREGATED_TIMESERIES_URL)
        on = assert_endpoint(url, headers=pair.ff_on, items_key=None)
        off = assert_endpoint(url, headers=pair.ff_off, items_key=None)
        assert isinstance(on, dict) and isinstance(off, dict)
        assert on.keys() == off.keys(), (
            f"{pair.key} tiktok/aggregated-timeseries: FF-ON / FF-OFF payload "
            f"keys differ — {sorted(on.keys())} vs {sorted(off.keys())}"
        )
        on_total = on.get("total") or {}
        off_total = off.get("total") or {}
        for metric in ("creations", "views"):
            on_value = on_total.get(metric) or 0
            off_value = off_total.get(metric) or 0
            label = f"{pair.key} tiktok/aggregated-timeseries {metric}"
            assert_ff_monotonic(on_value, off_value, label=label)
            if pair.key in ("employee", "sued_vendor"):
                assert_ff_noop(on_value, off_value, label=label)


# ===========================================================================
# Participation-axis probes — the originating & destination artists
# ===========================================================================
#
# The TikTok endpoints are viewer-scoped by `permissions_filter`, so an artist
# viewer is gated by the participation branch. As on the ownership axis, QA
# TikTok coverage for product 5244974 is not guaranteed, so these never require
# data: the FF-pair pins only the data-independent laws — no legacy leak
# (FF-OFF == 0 for the originating artist) and no forward leak (FF-ON shows
# nothing past the originating artist's move cutoff).


class TestTikTokTimeseriesArtist:
    """/sound-recording/<isrc>/tiktok/timeseries — DAILY, participation branch."""

    @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS)
    @pytest.mark.parametrize("probe", WINDOW_PROBES)
    def test_streams(self, pair, probe):
        """TikTok timeseries is correctly time-sliced for the artist FF-pair."""
        url = _tiktok_url(
            config.SOUND_RECORDING_TIKTOK_TIMESERIES_URL, params=PROBES[probe]
        )
        on = assert_endpoint(url, headers=pair.ff_on)
        off = assert_endpoint(url, headers=pair.ff_off)
        on_total = _tiktok_total(on["items"])
        off_total = _tiktok_total(off["items"])
        assert_isolated_transfer(
            on_total,
            off_total,
            role_key=pair.key,
            label=f"{pair.key} tiktok/timeseries probe={probe}",
            require_data=False,
        )
        if expected_visibility(pair.key, probe) == "empty":
            assert on_total == 0, (
                f"{pair.key} leaked {on_total} TikTok activity into probe "
                f"{probe} — past the artist's move cutoff"
            )


class TestTikTokSummaryArtist:
    """/sound-recording/<isrc>/tiktok/summary — DAILY windowed summary."""

    @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS)
    @pytest.mark.parametrize("probe", WINDOW_PROBES)
    def test_total(self, pair, probe):
        """TikTok summary is correctly time-sliced for the artist FF-pair."""
        url = _tiktok_url(
            config.SOUND_RECORDING_TIKTOK_SUMMARY_URL, params=PROBES[probe]
        )
        on = assert_endpoint(url, headers=pair.ff_on)
        off = assert_endpoint(url, headers=pair.ff_off)
        on_total = _tiktok_total(on["items"])
        off_total = _tiktok_total(off["items"])
        assert_isolated_transfer(
            on_total,
            off_total,
            role_key=pair.key,
            label=f"{pair.key} tiktok/summary probe={probe}",
            require_data=False,
        )
        if expected_visibility(pair.key, probe) == "empty":
            assert on_total == 0, (
                f"{pair.key} leaked {on_total} TikTok activity into probe "
                f"{probe} — past the artist's move cutoff"
            )


class TestTikTokAggregatedSummaryArtist:
    """/sound-recording/<isrc>/tiktok/aggregated-summary — ROLLUP all-time."""

    @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS)
    def test_all_time(self, pair):
        """All-time TikTok aggregates obey the participation scoping."""
        url = _tiktok_url(config.SOUND_RECORDING_TIKTOK_AGGREGATED_SUMMARY_URL)
        on = assert_endpoint(url, headers=pair.ff_on, items_key=None)
        off = assert_endpoint(url, headers=pair.ff_off, items_key=None)
        assert isinstance(on, dict) and isinstance(off, dict)
        on_total = (on.get("creations_all_time") or 0) + (on.get("views_all_time") or 0)
        off_total = (off.get("creations_all_time") or 0) + (
            off.get("views_all_time") or 0
        )
        assert_isolated_transfer(
            on_total,
            off_total,
            role_key=pair.key,
            label=f"{pair.key} tiktok/aggregated-summary all-time",
            require_data=False,
        )


class TestTikTokAggregatedTimeseriesArtist:
    """/sound-recording/<isrc>/tiktok/aggregated-timeseries — ROLLUP recent window.

    Recent-only window, so the originating artist (frozen before it) contributes
    nothing under either flag — a no-op check for the destination artist.
    """

    @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS)
    def test_default(self, pair):
        """The v2 aggregated-timeseries path is a no-op over the recent window."""
        url = _tiktok_url(config.SOUND_RECORDING_TIKTOK_AGGREGATED_TIMESERIES_URL)
        on = assert_endpoint(url, headers=pair.ff_on, items_key=None)
        off = assert_endpoint(url, headers=pair.ff_off, items_key=None)
        assert isinstance(on, dict) and isinstance(off, dict)
        assert on.keys() == off.keys(), (
            f"{pair.key} tiktok/aggregated-timeseries: FF-ON / FF-OFF payload "
            f"keys differ — {sorted(on.keys())} vs {sorted(off.keys())}"
        )
        on_total = on.get("total") or {}
        off_total = off.get("total") or {}
        for metric in ("creations", "views"):
            on_value = on_total.get(metric) or 0
            off_value = off_total.get(metric) or 0
            label = f"{pair.key} tiktok/aggregated-timeseries {metric}"
            assert_ff_monotonic(on_value, off_value, label=label)
            if pair.key == "destination_artist":
                assert_ff_noop(on_value, off_value, label=label)