"""Transfer-ownership tests for POST /playlist/analytics-bulk.

ROLLUP grain, table-swap only. The PORT-5 work wrapped the FROM with
``transfer_product_ownership_table_name`` — when the flag is on, every
``STREAMS_BY_PLAYLIST_*_FEED_ROLLUP`` reference is swapped for its
``_PRODUCT_TRANSFER`` variant. The SQL has **no per-viewer permission
filter**, so the only thing the flag changes is which fact table the query
reads.

Per dbt construction the ``_PRODUCT_TRANSFER`` variant is a Pattern B drop-in
— same shape, similar magnitudes — but the POA-join filters out streams
whose product has no POA row, so the totals are not exactly equal (FACTS.QA
showed ~1% drift on `mix`). The assertion is therefore on the response
*shape*, not on the numeric metrics: the set of returned playlists must be
identical, and every input playlist must surface a row.

This is a regression-only guard: it confirms the FF-ON code path returns
200 + the same shape as FF-OFF. The dbt drop-in equivalence itself lives
in the dbt repo's test suite.
"""

from __future__ import annotations

import pytest
import requests

from tests.integration.config import QA_BASE_URL
from tests.integration.transfer_ownership.conftest import (
    CANONICAL_STORE_ID,
    CANONICAL_STORE_PLAYLIST_ID,
    TRANSFER_PROFILE_PAIRS,
)

URL = f"{QA_BASE_URL}/playlist/analytics-bulk"

# One canonical playlist with multi-window coverage on BX69Y2100018. The
# SQL aggregates over the whole playlist (not per-ISRC), so a single
# playlist body is enough to exercise the table swap.
BODY = {
    "playlists": [
        {
            "store_playlist_id": CANONICAL_STORE_PLAYLIST_ID,
            "store_id": CANONICAL_STORE_ID,
            "storefront": None,
        }
    ]
}


def _payload(headers: dict) -> dict:
    response = requests.post(URL, json=BODY, headers=headers)
    assert (
        response.status_code == 200
    ), f"POST {URL} -> {response.status_code}: {response.text}"
    return response.json()["data"]


def _playlist_keyset(payload: dict) -> set:
    """Project each returned playlist row to its (store_playlist_id, store_id)
    tuple. Used for shape comparison without binding to numeric metrics."""
    return {
        (row.get("store_playlist_id"), row.get("store_id"))
        for row in payload.get("playlists") or []
    }


class TestAnalyticsBulkTableSwap:
    """The ``_PRODUCT_TRANSFER`` rollup swap must preserve response shape.

    Numeric drift between the two table variants is expected (~1% on
    `mix`), so we don't compare streams totals here — only the set of
    playlists that surface. A regression where FF-ON returns fewer
    playlists, the wrong shape, or 500s would be caught.
    """

    @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS)
    def test_keyset_preserved(self, pair):
        ff_on = _payload(pair.ff_on)
        ff_off = _payload(pair.ff_off)
        on_keys = _playlist_keyset(ff_on)
        off_keys = _playlist_keyset(ff_off)
        assert on_keys == off_keys, (
            f"{pair.key} POST /playlist/analytics-bulk: the flag changed the "
            f"set of returned playlists — FF-ON={on_keys}, FF-OFF={off_keys}. "
            f"The table swap must preserve playlist coverage."
        )
        # The input playlist must always come back — a missing row is a
        # silent regression that the keyset-equality above would *not*
        # catch when both responses miss it.
        assert (CANONICAL_STORE_PLAYLIST_ID, int(CANONICAL_STORE_ID)) in on_keys, (
            f"{pair.key}: canonical playlist not in FF-ON response — "
            f"the table swap dropped the input row"
        )