"""Transfer-ownership tests for /product/<id>/* endpoints.

Product 5244974 ("Travessia") is slated to be transferred from Ala
Comunicação (vendor 81790) -> Sued Nunes Produções LTDA (vendor 797716).
Each test parametrizes on the three transfer-aware profiles and asserts the
response is correctly scoped in time by PRODUCT_OWNERSHIP_ACCESS — see
README.md and conftest._VISIBILITY.

DAILY endpoints (`/timeseries`, `/summary`) are checked against the
visibility matrix with the window probes. ROLLUP endpoints
(`/aggregate-streams`, `/metrics-by-track`) are checked structurally plus a
"no forward leak" assertion: a former owner's recent-window metrics are zero.
"""

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 (
    ARTIST_PROFILE_PAIRS,
    ARTIST_PROFILES,
    CURRENT_VIEW,
    DESTINATION_ARTIST,
    FORMER_OWNERS,
    GLOBAL_PARTICIPANT_ID,
    PROBES,
    PRODUCT_ID,
    PRODUCT_ISRCS,
    PROFILE_PAIRS,
    SEES_RECENT_WINDOW,
    TRANSFER_PROFILE_PAIRS,
    TRANSFER_PROFILES,
    WINDOW_PROBES,
    assert_ff_noop,
    assert_isolated_transfer,
    assert_visibility,
    find_row,
    profile_key_for,
)


def _product_url(template: str, *, params: str = "") -> str:
    """Substitute product 5244974 into a URL template and append a query string."""
    base = template.replace("<product_id>", PRODUCT_ID)
    return f"{base}?{params}" if params else base


class TestProductTimeseries:
    """/product/<id>/timeseries — DAILY grain, gated by the v2 permission macro."""

    @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES)
    @pytest.mark.parametrize("probe", WINDOW_PROBES)
    def test_streams(self, hdrs, probe):
        """Stream timeseries per profile x window probe; visibility per matrix."""
        url = _product_url(
            config.PRODUCT_TIME_SERIES_URL,
            params=f"type=PRODUCT_STREAMS&{PROBES[probe]}",
        )
        payload = assert_endpoint(url, headers=hdrs)
        assert_visibility(
            payload, profile_key=profile_key_for(hdrs), probe=probe, metric_key="value"
        )


class TestProductSummary:
    """/product/<id>/summary — DAILY grain, TOTAL and SOUND_RECORDING aggregations."""

    @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES)
    @pytest.mark.parametrize("probe", WINDOW_PROBES)
    def test_total(self, hdrs, probe):
        """TOTAL summary per profile x window probe; visibility per matrix."""
        url = _product_url(
            config.PRODUCT_SUMMARY_URL,
            params=f"type=TOTAL&{PROBES[probe]}&{PAGINATION}",
        )
        payload = assert_endpoint(url, headers=hdrs)
        assert_visibility(payload, profile_key=profile_key_for(hdrs), probe=probe)

    @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES)
    @pytest.mark.parametrize("probe", WINDOW_PROBES)
    def test_sound_recording_breakdown(self, hdrs, probe):
        """SOUND_RECORDING summary lists per-ISRC rows; visibility per matrix."""
        url = _product_url(
            config.PRODUCT_SUMMARY_URL,
            params=f"type=SOUND_RECORDING&{PROBES[probe]}&{PAGINATION}",
        )
        payload = assert_endpoint(url, headers=hdrs)
        assert_visibility(payload, profile_key=profile_key_for(hdrs), probe=probe)


class TestProductAggregateStreams:
    """/product/<id>/aggregate-streams — ROLLUP grain (all-time + growth windows).

    Every profile owned product 5244974 at some point, so all four see a
    non-zero `streams_all_time` time-sliced to their window. The FF-pair turns
    this single-product ROLLUP endpoint into a precise transfer probe: legacy
    (FF-OFF) hides the product from a former owner, so `streams_all_time = 0`
    there; v2 (FF-ON) grants the historical slice; the employee and current
    owner see the same lifetime total under both flags.
    """

    @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS)
    def test_transferred_product_all_time(self, pair):
        """Required fields present; streams_all_time obeys the transfer scoping."""
        url = _product_url(config.PRODUCT_AGGREGATE_STREAMS_URL)
        on = assert_endpoint(url, headers=pair.ff_on, items_key=None)
        off = assert_endpoint(url, headers=pair.ff_off, items_key=None)
        for field in ("product_id", "streams_all_time", "growth_percentage_7_days"):
            assert field in on, f"missing field: {field}"
        assert str(on["product_id"]) == PRODUCT_ID
        on_all_time = on.get("streams_all_time") or 0
        assert on_all_time > 0, (
            f"{pair.key} sees no lifetime streams for product {PRODUCT_ID} "
            f"under FF-ON — the v2 path lost the owner's slice"
        )
        assert_isolated_transfer(
            on_all_time,
            off.get("streams_all_time") or 0,
            role_key=pair.key,
            label=f"{pair.key} /product/{PRODUCT_ID}/aggregate-streams streams_all_time",
        )


class TestProductAggregatedStreams:
    """/product/<id>/aggregated-streams — last-28-day window by dimension.

    `days_back=28` lands entirely in W2 (the current owner's era) once the
    transfer cutoff (2026-04-30) is comfortably outside the 28-day window, so
    this is a "no forward leak" probe: only the current owner and employee
    see a non-zero windowed total; former owners see zero.
    """

    DIMENSIONS = ["store", "country", "sos"]

    @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES)
    @pytest.mark.parametrize("dimension", DIMENSIONS)
    def test_dimension(self, hdrs, dimension):
        """Shape always present; windowed total leaks no post-transfer streams."""
        url = _product_url(
            config.PRODUCT_AGGREGATED_STREAMS_URL,
            params=f"dimension={dimension}&days_back=28",
        )
        payload = assert_endpoint(url, headers=hdrs, items_key=None)
        for field in (
            "all_other_rollup",
            "all_other_timeseries",
            "topn_timeseries",
            "topn_rollup",
            "total",
        ):
            assert field in payload, f"missing field: {field}"

        # `total` carries the rollup window metrics, not a generic `value`
        # key — read the days_back=28 window directly.
        total = (payload.get("total") or {}).get("streams_28_days", 0) or 0
        if profile_key_for(hdrs) in SEES_RECENT_WINDOW:
            assert total > 0, "current owner / employee should see recent streams"
        else:
            # A non-zero figure for a former owner is often a false positive:
            # the rollup is anchored at get_max_available_streaming_stores_date(),
            # not today, so a watermark less than `access_until_date + 28` days
            # ahead legitimately covers W1's last days. See README §10 for the
            # diagnostic checklist (rollup table to query, decision criteria).
            assert total == 0, (
                f"{profile_key_for(hdrs)} leaked {total} post-transfer streams "
                f"into the last-28-day {dimension} total"
            )

    @pytest.mark.parametrize("dimension", DIMENSIONS)
    def test_employee_ff_noop(self, dimension):
        """Employee FF-ON == FF-OFF on the 28-day rollup `total.streams_28_days`.

        The 28-day rollup is `is_current = TRUE` for the employee under both
        flags, so the flag must not move this figure. A regression here would
        mean v2 is rewriting the employee's view of the recent-window rollup.
        """
        pair = PROFILE_PAIRS["employee"]
        url = _product_url(
            config.PRODUCT_AGGREGATED_STREAMS_URL,
            params=f"dimension={dimension}&days_back=28",
        )
        on = assert_endpoint(url, headers=pair.ff_on, items_key=None)
        off = assert_endpoint(url, headers=pair.ff_off, items_key=None)
        on_total = (on.get("total") or {}).get("streams_28_days") or 0
        off_total = (off.get("total") or {}).get("streams_28_days") or 0
        assert_ff_noop(
            on_total,
            off_total,
            label=(
                f"employee /product/{PRODUCT_ID}/aggregated-streams "
                f"dimension={dimension} total.streams_28_days"
            ),
        )


class TestProductMetricsByTrack:
    """/product/<id>/metrics-by-track — ROLLUP per-track breakdown."""

    @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES)
    def test_tracks(self, hdrs):
        """Per-track rows present for every owner; recent streams leak-free."""
        url = _product_url(config.PRODUCT_METRICS_BY_TRACK_URL)
        payload = assert_endpoint(url, headers=hdrs, items_key=None)
        assert str(payload["product_id"]) == PRODUCT_ID
        assert "tracks" in payload
        tracks = payload["tracks"]
        assert tracks, f"{profile_key_for(hdrs)} sees no tracks for the product"
        for field in (
            "track_id",
            "streams_1_day",
            "streams_7_days",
            "streams_28_days",
            "streams_all_time",
        ):
            assert field in tracks[0], f"missing per-track field: {field}"

        recent = sum((t.get("streams_28_days") or 0) for t in tracks)
        if profile_key_for(hdrs) in SEES_RECENT_WINDOW:
            assert recent > 0, "current owner / employee should see recent streams"
        else:
            # Non-zero here is often a false positive — same root cause as
            # TestProductAggregatedStreams above: the per-track `streams_28_days`
            # is anchored at the streaming watermark, not today, so W1's last
            # days sit inside the window until `max >= access_until_date + 28`.
            # See README §10 for the diagnostic SQL.
            assert recent == 0, (
                f"{profile_key_for(hdrs)} leaked {recent} post-transfer streams "
                f"into per-track streams_28_days"
            )

    @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES)
    def test_per_track_lifetime(self, hdrs):
        """Every track of product 5244974 has positive streams_all_time.

        All three profiles owned product 5244974 at some point, so each of
        the 13 tracks must carry positive all-time streams in the v2 rollup.
        A zero here is a per-track permission leak — the SQL truncated one
        ISRC's slice too aggressively while others passed. The existing
        aggregate check ("tracks non-empty") cannot catch a single-track
        drop; iterating per row does.
        """
        url = _product_url(config.PRODUCT_METRICS_BY_TRACK_URL)
        payload = assert_endpoint(url, headers=hdrs, items_key=None)
        tracks = payload["tracks"]
        profile_key = profile_key_for(hdrs)

        assert len(tracks) >= len(PRODUCT_ISRCS), (
            f"{profile_key}: only {len(tracks)} tracks visible, expected at "
            f"least {len(PRODUCT_ISRCS)} — per-track permission leak"
        )
        zero_track_ids = [
            track.get("track_id")
            for track in tracks
            if (track.get("streams_all_time") or 0) == 0
        ]
        assert not zero_track_ids, (
            f"{profile_key}: {len(zero_track_ids)} track(s) have zero "
            f"streams_all_time (track_ids={zero_track_ids}) — per-track "
            f"slice dropped streams"
        )

    def test_employee_ff_noop(self):
        """Employee FF-ON == FF-OFF on per-track `streams_28_days`.

        Per-track recent-window metrics for the employee resolve to the
        product's `is_current` rollup under both flags, so the flag must be a
        no-op for this sum. A regression here means v2 is rewriting the
        employee's view of the per-track recent window.
        """
        pair = PROFILE_PAIRS["employee"]
        url = _product_url(config.PRODUCT_METRICS_BY_TRACK_URL)
        on = assert_endpoint(url, headers=pair.ff_on, items_key=None)
        off = assert_endpoint(url, headers=pair.ff_off, items_key=None)
        on_recent = sum((t.get("streams_28_days") or 0) for t in on["tracks"])
        off_recent = sum((t.get("streams_28_days") or 0) for t in off["tracks"])
        assert_ff_noop(
            on_recent,
            off_recent,
            label=(
                f"employee /product/{PRODUCT_ID}/metrics-by-track "
                f"sum(tracks.streams_28_days)"
            ),
        )


class TestProductDetail:
    """/product/<id> — ROLLUP-backed all-time streams + 7-day growth.

    Every profile owned the product at some point, so all four see a positive
    `streams.aggregate.all_time` time-sliced to their window. The FF-pair pins
    that scoping precisely: legacy hides product 5244974 from a former owner
    so the all-time slot collapses to `None`/0; v2 grants the historical
    slice; employee and current owner see the same lifetime total under both
    flags.
    """

    @staticmethod
    def _all_time(payload: dict) -> float:
        return ((payload.get("streams") or {}).get("aggregate") or {}).get(
            "all_time"
        ) or 0

    @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS)
    def test_transferred_product_all_time(self, pair):
        """Detail payload shape present; all-time streams obey the transfer scoping."""
        url = _product_url(config.PRODUCT_URL)
        on = assert_endpoint(url, headers=pair.ff_on, items_key=None)
        off = assert_endpoint(url, headers=pair.ff_off, items_key=None)
        assert str(on["product_id"]) == PRODUCT_ID
        for field in ("streams", "tracks", "sources"):
            assert field in on, f"missing field: {field}"
        on_all_time = self._all_time(on)
        assert on_all_time > 0, (
            f"{pair.key} sees no lifetime streams for product {PRODUCT_ID} "
            f"under FF-ON — the v2 path lost the owner's slice"
        )
        assert_isolated_transfer(
            on_all_time,
            self._all_time(off),
            role_key=pair.key,
            label=f"{pair.key} /product/{PRODUCT_ID} streams.aggregate.all_time",
        )


class TestProductMetrics:
    """/product-metrics — ROLLUP listing, isolated by product row.

    `/product-metrics` returns one row per product, so product 5244974's own
    row is picked out of the participant's catalogue — the FF-pair then makes
    this listing a precise per-product transfer probe. A former owner has no
    5244974 row without the flag (legacy hides the transferred-away product)
    and a frozen, time-sliced one with it; the employee and current owner see
    the product's full lifetime either way.
    """

    @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS)
    def test_transferred_product_row(self, pair):
        """Product 5244974's product-metrics row obeys the transfer scoping."""
        url = (
            f"{config.PRODUCT_METRICS_URL}"
            # Large limit so product 5244974's row is never paginated away.
            f"?global_participant_ids={GLOBAL_PARTICIPANT_ID}&limit=100000"
        )
        on = assert_endpoint(url, headers=pair.ff_on, items_key=None)
        off = assert_endpoint(url, headers=pair.ff_off, items_key=None)
        for payload in (on, off):
            assert "metrics" in payload and isinstance(payload["metrics"], list)
            assert "total_products" in payload

        on_row = find_row(on["metrics"], "product_id", PRODUCT_ID)
        off_row = find_row(off["metrics"], "product_id", PRODUCT_ID)
        assert_isolated_transfer(
            (on_row or {}).get("streams_all_time") or 0,
            (off_row or {}).get("streams_all_time") or 0,
            role_key=pair.key,
            label=f"{pair.key} /product-metrics product {PRODUCT_ID} streams_all_time",
        )


class TestProductBulkGrowthPeriods:
    """/product/growth-periods-bulk — bulk ROLLUP per-product trailing windows.

    Goes through both the v2 permission macro (`rollup_grain=true`) and the
    `METRICS_BY_PRODUCT_FEED_DISTRIBUTOR_ROLLUP` → `*_PRODUCT_TRANSFER` table
    swap, and the `product_id` query param isolates product 5244974 — so the
    FF-pair is a precise per-product transfer probe. The response is a flat
    JSON list with one row per product whose only metrics are trailing
    windows (`streams_1_day` / `streams_7_days` / `streams_28_days`); there is
    no lifetime field, so two laws are asserted separately:

    - **row grant** — under FF-OFF, legacy hides the transferred-away product
      from a former owner (no row); under FF-ON, v2 grants the row from the
      owner's historical slice. The employee and current owner see the row
      under both flags.
    - **no forward leak** — under FF-ON, a former owner's slice is truncated
      at their `access_until_date`, so the trailing-window metrics are zero
      (the windows live in W2, after their cutoff). The current view sees
      current-era streams and the figure must match FF-OFF.
    """

    @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS)
    def test_transferred_product_row(self, pair):
        """Row presence + trailing-window metric obey the transfer scoping."""
        url = f"{config.PRODUCT_BULK_GROWTH_PERIODS_URL}?product_id={PRODUCT_ID}"
        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, list) and isinstance(off, list), (
            f"{pair.key} /product/growth-periods-bulk: expected a top-level "
            f"list, got {type(on).__name__} / {type(off).__name__}"
        )
        on_row = find_row(on, "id", PRODUCT_ID)
        off_row = find_row(off, "id", PRODUCT_ID)

        # FF-ON: every owner sees product 5244974's row — current natively,
        # former owners via the v2-granted historical slice.
        assert on_row is not None, (
            f"{pair.key} sees no product {PRODUCT_ID} row under FF-ON — "
            f"the v2 path lost the owner's slice"
        )

        if pair.key in CURRENT_VIEW:
            assert off_row is not None, (
                f"{pair.key} (current view) sees no product {PRODUCT_ID} row "
                f"under FF-OFF — must always be visible"
            )
            assert_ff_noop(
                on_row.get("streams_28_days") or 0,
                off_row.get("streams_28_days") or 0,
                label=(
                    f"{pair.key} /product/growth-periods-bulk product "
                    f"{PRODUCT_ID} streams_28_days"
                ),
            )
        elif pair.key in FORMER_OWNERS:
            assert off_row is None, (
                f"{pair.key} (former owner) LEAKED a row for transferred-away "
                f"product {PRODUCT_ID} under FF-OFF — legacy must hide it"
            )
            recent = on_row.get("streams_28_days") or 0
            # Same false-positive shape as TestProductAggregatedStreams: the
            # 28-day window is anchored at the streaming watermark, not today.
            # If `max - 27 <= access_until_date`, the former owner legitimately
            # carries W1 streams into this rolling figure. See README §10.
            assert recent == 0, (
                f"{pair.key} leaked {recent} post-transfer streams into "
                f"product {PRODUCT_ID} streams_28_days under FF-ON"
            )


# ===========================================================================
# Participation-axis probes — the originating & destination artists
# ===========================================================================
#
# /product/<id>/* is viewer-scoped by `permissions_filter`, so for an artist
# viewer it gates on the `permission_label_participant_ids` branch. The two
# artists reach product 5244974 through participation, not label ownership, but
# resolve to the SAME visibility shapes as the ownership profiles:
#
#   * destination_artist (current PARTICIPATED_IN row) == sued_vendor's shape:
#     full lifetime, flag no-op.
#   * originating_artist (USED_TO_PARTICIPATE_IN historical grant) == ala_vendor's
#     shape: W1 only, no W2 leak; FF-OFF == 0, FF-ON == the W1 slice.
#
# These classes mirror the ownership-axis classes above on the ARTIST axes,
# reusing the same URL builder, visibility matrix and FF-pair helpers. See
# test_used_to_participate.py for the participation axis the
# /participant/<gp>/* endpoints exercise; this covers the same scoping reached
# via /product/<id>/* instead.


class TestProductTimeseriesArtist:
    """/product/<id>/timeseries — DAILY, participation branch of the v2 macro."""

    @pytest.mark.parametrize("hdrs", ARTIST_PROFILES)
    @pytest.mark.parametrize("probe", WINDOW_PROBES)
    def test_streams(self, hdrs, probe):
        """Stream timeseries per artist x window probe; visibility per matrix."""
        url = _product_url(
            config.PRODUCT_TIME_SERIES_URL,
            params=f"type=PRODUCT_STREAMS&{PROBES[probe]}",
        )
        payload = assert_endpoint(url, headers=hdrs)
        assert_visibility(
            payload, profile_key=profile_key_for(hdrs), probe=probe, metric_key="value"
        )


class TestProductSummaryArtist:
    """/product/<id>/summary — DAILY, TOTAL and SOUND_RECORDING aggregations."""

    @pytest.mark.parametrize("hdrs", ARTIST_PROFILES)
    @pytest.mark.parametrize("probe", WINDOW_PROBES)
    def test_total(self, hdrs, probe):
        """TOTAL summary per artist x window probe; visibility per matrix."""
        url = _product_url(
            config.PRODUCT_SUMMARY_URL,
            params=f"type=TOTAL&{PROBES[probe]}&{PAGINATION}",
        )
        payload = assert_endpoint(url, headers=hdrs)
        assert_visibility(payload, profile_key=profile_key_for(hdrs), probe=probe)

    @pytest.mark.parametrize("hdrs", ARTIST_PROFILES)
    @pytest.mark.parametrize("probe", WINDOW_PROBES)
    def test_sound_recording_breakdown(self, hdrs, probe):
        """SOUND_RECORDING summary lists per-ISRC rows; visibility per matrix."""
        url = _product_url(
            config.PRODUCT_SUMMARY_URL,
            params=f"type=SOUND_RECORDING&{PROBES[probe]}&{PAGINATION}",
        )
        payload = assert_endpoint(url, headers=hdrs)
        assert_visibility(payload, profile_key=profile_key_for(hdrs), probe=probe)


class TestProductAggregateStreamsArtist:
    """/product/<id>/aggregate-streams — ROLLUP all-time, participation FF-pair.

    Legacy (FF-OFF) hides the transferred-away product from the originating
    artist (its PARTICIPATED_IN edge was deleted), so `streams_all_time = 0`;
    v2 (FF-ON) grants the USED_TO historical slice. The destination artist sees
    the same lifetime total under both flags.
    """

    @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS)
    def test_transferred_product_all_time(self, pair):
        """streams_all_time obeys the participation transfer scoping."""
        url = _product_url(config.PRODUCT_AGGREGATE_STREAMS_URL)
        on = assert_endpoint(url, headers=pair.ff_on, items_key=None)
        off = assert_endpoint(url, headers=pair.ff_off, items_key=None)
        for field in ("product_id", "streams_all_time", "growth_percentage_7_days"):
            assert field in on, f"missing field: {field}"
        assert str(on["product_id"]) == PRODUCT_ID
        on_all_time = on.get("streams_all_time") or 0
        assert on_all_time > 0, (
            f"{pair.key} sees no lifetime streams for product {PRODUCT_ID} "
            f"under FF-ON — the v2 participation path lost the artist's slice"
        )
        assert_isolated_transfer(
            on_all_time,
            off.get("streams_all_time") or 0,
            role_key=pair.key,
            label=f"{pair.key} /product/{PRODUCT_ID}/aggregate-streams streams_all_time",
        )


class TestProductAggregatedStreamsArtist:
    """/product/<id>/aggregated-streams — last-28-day window, participation axis.

    `days_back=28` lands in W2, so this is a no-forward-leak probe: the
    destination artist (current participant) sees a non-zero window; the
    originating artist (frozen at the move boundary) sees zero.
    """

    DIMENSIONS = ["store", "country", "sos"]

    @pytest.mark.parametrize("hdrs", ARTIST_PROFILES)
    @pytest.mark.parametrize("dimension", DIMENSIONS)
    def test_dimension(self, hdrs, dimension):
        """Shape always present; windowed total leaks no post-transfer streams."""
        url = _product_url(
            config.PRODUCT_AGGREGATED_STREAMS_URL,
            params=f"dimension={dimension}&days_back=28",
        )
        payload = assert_endpoint(url, headers=hdrs, items_key=None)
        for field in (
            "all_other_rollup",
            "all_other_timeseries",
            "topn_timeseries",
            "topn_rollup",
            "total",
        ):
            assert field in payload, f"missing field: {field}"

        total = (payload.get("total") or {}).get("streams_28_days", 0) or 0
        if profile_key_for(hdrs) in SEES_RECENT_WINDOW:
            assert total > 0, "destination artist should see recent streams"
        else:
            # Former-participant false positives have the same root cause as the
            # ownership axis: the 28-day rollup is anchored at the streaming
            # watermark, not today. See README §10.
            assert total == 0, (
                f"{profile_key_for(hdrs)} leaked {total} post-transfer streams "
                f"into the last-28-day {dimension} total"
            )

    @pytest.mark.parametrize("dimension", DIMENSIONS)
    def test_destination_ff_noop(self, dimension):
        """Destination artist FF-ON == FF-OFF on the 28-day rollup total.

        The current participation row is live under both flags, so the flag must
        not move this figure — the participation-axis analogue of the employee
        no-op on the ownership axis.
        """
        url = _product_url(
            config.PRODUCT_AGGREGATED_STREAMS_URL,
            params=f"dimension={dimension}&days_back=28",
        )
        on = assert_endpoint(url, headers=DESTINATION_ARTIST.ff_on, items_key=None)
        off = assert_endpoint(url, headers=DESTINATION_ARTIST.ff_off, items_key=None)
        on_total = (on.get("total") or {}).get("streams_28_days") or 0
        off_total = (off.get("total") or {}).get("streams_28_days") or 0
        assert_ff_noop(
            on_total,
            off_total,
            label=(
                f"destination_artist /product/{PRODUCT_ID}/aggregated-streams "
                f"dimension={dimension} total.streams_28_days"
            ),
        )


class TestProductMetricsByTrackArtist:
    """/product/<id>/metrics-by-track — ROLLUP per-track, participation axis."""

    @pytest.mark.parametrize("hdrs", ARTIST_PROFILES)
    def test_tracks(self, hdrs):
        """Per-track rows present for every artist; recent streams leak-free."""
        url = _product_url(config.PRODUCT_METRICS_BY_TRACK_URL)
        payload = assert_endpoint(url, headers=hdrs, items_key=None)
        assert str(payload["product_id"]) == PRODUCT_ID
        tracks = payload.get("tracks")
        assert tracks, f"{profile_key_for(hdrs)} sees no tracks for the product"

        recent = sum((t.get("streams_28_days") or 0) for t in tracks)
        if profile_key_for(hdrs) in SEES_RECENT_WINDOW:
            assert recent > 0, "destination artist should see recent streams"
        else:
            assert recent == 0, (
                f"{profile_key_for(hdrs)} leaked {recent} post-transfer streams "
                f"into per-track streams_28_days"
            )

    @pytest.mark.parametrize("hdrs", ARTIST_PROFILES)
    def test_per_track_lifetime(self, hdrs):
        """Every track of product 5244974 has positive streams_all_time.

        The participation grant is product-level (the v2 macro joins on
        product_id), so both artists see all 13 tracks with a positive all-time
        slice — a zero is a per-track leak in the participation rollup filter.
        """
        url = _product_url(config.PRODUCT_METRICS_BY_TRACK_URL)
        payload = assert_endpoint(url, headers=hdrs, items_key=None)
        tracks = payload["tracks"]
        profile_key = profile_key_for(hdrs)

        assert len(tracks) >= len(PRODUCT_ISRCS), (
            f"{profile_key}: only {len(tracks)} tracks visible, expected at "
            f"least {len(PRODUCT_ISRCS)} — per-track participation leak"
        )
        zero_track_ids = [
            track.get("track_id")
            for track in tracks
            if (track.get("streams_all_time") or 0) == 0
        ]
        assert not zero_track_ids, (
            f"{profile_key}: {len(zero_track_ids)} track(s) have zero "
            f"streams_all_time (track_ids={zero_track_ids}) — per-track slice "
            f"dropped streams"
        )

    def test_destination_ff_noop(self):
        """Destination artist FF-ON == FF-OFF on per-track streams_28_days."""
        url = _product_url(config.PRODUCT_METRICS_BY_TRACK_URL)
        on = assert_endpoint(url, headers=DESTINATION_ARTIST.ff_on, items_key=None)
        off = assert_endpoint(url, headers=DESTINATION_ARTIST.ff_off, items_key=None)
        on_recent = sum((t.get("streams_28_days") or 0) for t in on["tracks"])
        off_recent = sum((t.get("streams_28_days") or 0) for t in off["tracks"])
        assert_ff_noop(
            on_recent,
            off_recent,
            label=(
                f"destination_artist /product/{PRODUCT_ID}/metrics-by-track "
                f"sum(tracks.streams_28_days)"
            ),
        )


class TestProductDetailArtist:
    """/product/<id> — ROLLUP all-time streams, participation FF-pair."""

    @staticmethod
    def _all_time(payload: dict) -> float:
        return ((payload.get("streams") or {}).get("aggregate") or {}).get(
            "all_time"
        ) or 0

    @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS)
    def test_transferred_product_all_time(self, pair):
        """Detail payload shape present; all-time obeys the participation scoping."""
        url = _product_url(config.PRODUCT_URL)
        on = assert_endpoint(url, headers=pair.ff_on, items_key=None)
        off = assert_endpoint(url, headers=pair.ff_off, items_key=None)
        assert str(on["product_id"]) == PRODUCT_ID
        for field in ("streams", "tracks", "sources"):
            assert field in on, f"missing field: {field}"
        on_all_time = self._all_time(on)
        assert on_all_time > 0, (
            f"{pair.key} sees no lifetime streams for product {PRODUCT_ID} "
            f"under FF-ON — the v2 participation path lost the artist's slice"
        )
        assert_isolated_transfer(
            on_all_time,
            self._all_time(off),
            role_key=pair.key,
            label=f"{pair.key} /product/{PRODUCT_ID} streams.aggregate.all_time",
        )


class TestProductMetricsArtist:
    """/product-metrics — ROLLUP listing isolated by product row, participation axis.

    The participant's catalogue lists product 5244974's own row, so the FF-pair
    is a precise per-product probe: the originating artist has no row without
    the flag and a frozen, time-sliced one with it; the destination artist sees
    the full lifetime either way.
    """

    @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS)
    def test_transferred_product_row(self, pair):
        """Product 5244974's product-metrics row obeys the participation scoping."""
        url = (
            f"{config.PRODUCT_METRICS_URL}"
            f"?global_participant_ids={GLOBAL_PARTICIPANT_ID}&limit=100000"
        )
        on = assert_endpoint(url, headers=pair.ff_on, items_key=None)
        off = assert_endpoint(url, headers=pair.ff_off, items_key=None)
        for payload in (on, off):
            assert "metrics" in payload and isinstance(payload["metrics"], list)
            assert "total_products" in payload

        on_row = find_row(on["metrics"], "product_id", PRODUCT_ID)
        off_row = find_row(off["metrics"], "product_id", PRODUCT_ID)
        assert_isolated_transfer(
            (on_row or {}).get("streams_all_time") or 0,
            (off_row or {}).get("streams_all_time") or 0,
            role_key=pair.key,
            label=f"{pair.key} /product-metrics product {PRODUCT_ID} streams_all_time",
        )


class TestProductBulkGrowthPeriodsArtist:
    """/product/growth-periods-bulk — bulk ROLLUP trailing windows, participation axis.

    `product_id` isolates product 5244974, so the FF-pair is a precise probe:
    the originating artist gains the row under FF-ON (USED_TO grant) and has
    none under FF-OFF; its trailing windows live in W2 (after the cutoff), so
    they stay zero. The destination artist lists the row under both flags.
    """

    @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS)
    def test_transferred_product_row(self, pair):
        """Row presence + trailing-window metric obey the participation scoping."""
        url = f"{config.PRODUCT_BULK_GROWTH_PERIODS_URL}?product_id={PRODUCT_ID}"
        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, list) and isinstance(off, list), (
            f"{pair.key} /product/growth-periods-bulk: expected a top-level "
            f"list, got {type(on).__name__} / {type(off).__name__}"
        )
        on_row = find_row(on, "id", PRODUCT_ID)
        off_row = find_row(off, "id", PRODUCT_ID)

        assert on_row is not None, (
            f"{pair.key} sees no product {PRODUCT_ID} row under FF-ON — "
            f"the v2 participation path lost the artist's slice"
        )

        if pair.key in CURRENT_VIEW:
            assert off_row is not None, (
                f"{pair.key} (current participant) sees no product {PRODUCT_ID} "
                f"row under FF-OFF — must always be visible"
            )
            assert_ff_noop(
                on_row.get("streams_28_days") or 0,
                off_row.get("streams_28_days") or 0,
                label=(
                    f"{pair.key} /product/growth-periods-bulk product "
                    f"{PRODUCT_ID} streams_28_days"
                ),
            )
        elif pair.key in FORMER_OWNERS:
            assert off_row is None, (
                f"{pair.key} (former participant) LEAKED a row for transferred-"
                f"away product {PRODUCT_ID} under FF-OFF — legacy must hide it"
            )
            recent = on_row.get("streams_28_days") or 0
            # Same watermark false-positive shape as the ownership axis (README §10).
            assert recent == 0, (
                f"{pair.key} leaked {recent} post-transfer streams into "
                f"product {PRODUCT_ID} streams_28_days under FF-ON"
            )