"""Transfer-ownership tests for /market-ranks and /ugc-video-metrics.

Both endpoints are transfer-aware purely via the `*_PRODUCT_TRANSFER`
fact-table swap — *not* via a v2 permission macro (market-ranks swaps
`MARKET_SIZE_BY_STORE_COUNTRY`; ugc-video-metrics swaps the UGC rollup and
keeps the legacy permission filter). The flag therefore must be
**metric-neutral** for them: whatever the swapped table changes internally,
every viewer must see the same response with the flag on and off.

That is exactly what the FF-pair makes testable. Each test fires the request
with the FF-ON profile and its FF-OFF twin (identical account access — see
conftest.PROFILE_PAIRS) and asserts the two responses agree. A regression in
the `_PRODUCT_TRANSFER` rollup build (fanout, dropped rows, skew) surfaces as
a divergence — which a shape-only smoke test could never catch.
"""

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,
    CURRENT_VIEW,
    FORMER_OWNERS,
    PRODUCT_ISRCS,
    TRANSFER_PROFILE_PAIRS,
    assert_ff_noop,
    sum_metric,
)


def _ranks_by_store(items: list[dict]) -> dict:
    """Group /market-ranks rows into {store_id: (country set, sorted ranks)}."""
    by_store: dict = {}
    for row in items:
        countries, ranks = by_store.setdefault(row.get("store_id"), (set(), []))
        countries.add(row.get("country_code"))
        ranks.append(row.get("market_rank"))
    return {store: (cs, sorted(rs)) for store, (cs, rs) in by_store.items()}


def assert_market_ranks_equal(on: list[dict], off: list[dict], *, label: str) -> None:
    """Assert two /market-ranks responses match up to within-tie rank order.

    Countries of equal market size tie on `market_rank`, and the endpoint's
    ORDER BY breaks those ties unstably — so the two physically distinct fact
    tables (legacy vs `_PRODUCT_TRANSFER`) can assign a tied pair its ranks in
    either order (observed: ~105 country pairs swap). The invariant that holds
    is per store: the same set of ranked countries over the same multiset of
    rank values. A real regression — a country dropped, a rank value changed,
    rollup fanout — still breaks that; only a within-tie permutation passes.
    """
    on_by_store = _ranks_by_store(on)
    off_by_store = _ranks_by_store(off)
    assert on_by_store.keys() == off_by_store.keys(), (
        f"{label}: the flag changed the set of stores — "
        f"FF-ON {sorted(on_by_store)} vs FF-OFF {sorted(off_by_store)}."
    )
    for store, (on_countries, on_ranks) in on_by_store.items():
        off_countries, off_ranks = off_by_store[store]
        assert on_countries == off_countries, (
            f"{label}: store {store} — the flag changed the ranked country set "
            f"(FF-ON only={on_countries - off_countries}, "
            f"FF-OFF only={off_countries - on_countries})."
        )
        assert on_ranks == off_ranks, (
            f"{label}: store {store} — the flag changed the rank distribution "
            f"(rank multiset differs)."
        )


class TestMarketRanks:
    """/market-ranks — global market-size ranks, transfer-aware via table swap.

    /market-ranks is not ownership-scoped at all: the flag only swaps the
    `MARKET_SIZE_BY_STORE_COUNTRY` fact table. The swapped table must reproduce
    the legacy ranks — so every viewer's rank list must be identical with the
    flag on and off, up to the unstable tie-breaking of equal-market-size
    countries (see assert_market_ranks_equal).
    """

    @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS)
    def test_table_swap_is_neutral(self, pair):
        """The _PRODUCT_TRANSFER table swap leaves /market-ranks unchanged."""
        on = assert_endpoint(config.MARKET_RANKS_URL, headers=pair.ff_on)
        off = assert_endpoint(config.MARKET_RANKS_URL, headers=pair.ff_off)
        if on["items"]:
            for field in ("store_id", "country_code", "market_rank"):
                assert field in on["items"][0], f"missing field: {field}"
        assert_market_ranks_equal(
            on["items"], off["items"], label=f"{pair.key} /market-ranks"
        )


class TestUgcVideoMetrics:
    """/ugc-video-metrics — v2 transfer-aware UGC for product 5244974.

    Both the UGC rollup *and* the permission filter are flag-affected: the SQL
    swaps to `V_METRICS_UGC_BY_VIDEO_ASSET_*_FEED_DISTRIBUTOR_ROLLUP_PRODUCT_
    TRANSFER` and calls `permissions_filter(..., rollup_grain=true)`, which
    routes to the v2 POA gate under FF-ON. So pinned to product 5244974's 13
    ISRCs:

    * A current-view role (employee, sued_vendor) sees the product identically
      with the flag on and off — the employee reads `is_current=TRUE` rows
      only and sued_vendor is the current owner — so row count and view
      metrics must match.
    * A former owner (ala_vendor) has no access to product 5244974 under
      legacy current-ownership and is granted it under v2 (POA former-owner
      row): FF-OFF must return 0 rows, FF-ON must return rows.
    """

    @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS)
    def test_transfer_scoping(self, pair):
        """FF-pair behavior matches the viewer's ownership role."""
        track_isrcs = "&".join(f"track_isrcs={isrc}" for isrc in PRODUCT_ISRCS)
        url = f"{config.UGC_VIDEO_METRICS_URL}?{track_isrcs}&limit=25"
        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.get("items"), list), "missing items list"
        assert isinstance(off.get("items"), list), "missing items list"
        label = f"{pair.key} /ugc-video-metrics"

        if pair.key in CURRENT_VIEW:
            assert len(on["items"]) == len(off["items"]), (
                f"{label}: the flag changed the row count for a current-view "
                f"role — {len(on['items'])} FF-ON vs {len(off['items'])} FF-OFF"
            )
            assert_ff_noop(
                sum_metric(on["items"], "views"),
                sum_metric(off["items"], "views"),
                label=f"{label} views",
            )
            assert_ff_noop(
                sum_metric(on["items"], "premium_views"),
                sum_metric(off["items"], "premium_views"),
                label=f"{label} premium_views",
            )
        elif pair.key in FORMER_OWNERS:
            assert off["items"] == [], (
                f"{label}: legacy LEAKED product 5244974's UGC rows to a "
                f"former owner — FF-OFF returned {len(off['items'])} row(s), "
                f"must be 0 (legacy current-ownership hides the transferred-"
                f"away product)."
            )
            assert on["items"], (
                f"{label}: v2 did NOT grant the former owner its UGC rows "
                f"for product 5244974 — FF-ON returned 0 rows."
            )


# ===========================================================================
# Participation-axis probes — the originating & destination artists
# ===========================================================================
#
# Only /market-ranks is mirrored on the participation axis. It is not
# ownership- (or participation-) scoped at all — the flag merely swaps the
# MARKET_SIZE_BY_STORE_COUNTRY fact table — so the table swap must be neutral
# for *every* viewer, the artists included.
#
# /ugc-video-metrics is deliberately NOT mirrored for the artists: it is a
# video-surface endpoint (out of scope for this participation pass), and its
# participation-branch rollup join over the UGC fact table does not return
# within the suite's request budget for an artist viewer (observed >180s
# against local dev), so an artist probe here would be unreliable rather than
# informative. The ownership-axis TestUgcVideoMetrics above keeps the endpoint
# covered.


class TestMarketRanksArtist:
    """/market-ranks — table-swap neutrality, participation axis.

    /market-ranks is viewer-agnostic, so an artist viewer must get an identical
    rank list with the flag on and off, up to the unstable tie-breaking of
    equal-market-size countries (see assert_market_ranks_equal).
    """

    @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS)
    def test_table_swap_is_neutral(self, pair):
        """The _PRODUCT_TRANSFER table swap leaves /market-ranks unchanged."""
        on = assert_endpoint(config.MARKET_RANKS_URL, headers=pair.ff_on)
        off = assert_endpoint(config.MARKET_RANKS_URL, headers=pair.ff_off)
        if on["items"]:
            for field in ("store_id", "country_code", "market_rank"):
                assert field in on["items"][0], f"missing field: {field}"
        assert_market_ranks_equal(
            on["items"], off["items"], label=f"{pair.key} /market-ranks"
        )