"""Transfer-ownership tests for the /top-* leaderboard endpoints. These three leaderboards aggregate whole catalogues, so no query param can isolate product 5244974's transfer. The FF-pair gives each one a real check anyway — fire the request with the FF-ON profile and its FF-OFF twin (same account access, flag off — see conftest.PROFILE_PAIRS) and compare: * **/top-metrics** accepts a `upc` filter, so `upc=198000137366` *does* isolate product 5244974 — its rows become a precise per-ISRC transfer probe: a former owner sees nothing without the flag and the time-sliced slice with it. * **/top-accounts-metrics** is keyed by account. A former-owner account regains its transferred-away catalogue under the flag, so its `streams_all_time` can only grow — the differential asserts the v2 path never drops an account's data (monotonicity). * **/top-sound-recordings** is a windowed global leaderboard with no all-time column; the transfer signal is sub-noise there, but the employee — who sees current attribution under both flags — still gives an exact no-op oracle. """ 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, PROFILE_ACCOUNTS, TRANSFER_PROFILE_PAIRS, UPC, assert_ff_monotonic, assert_ff_noop, assert_isolated_transfer, find_row, sum_metric, ) # The shared `PAGINATION` constant sorts by `streams`, but the /top-metrics and # /top-sound-recordings leaderboards expose no bare `streams` column — only # windowed `streams__days`. Sort by one of those instead. _PAGINATION = "limit=50&order_by=streams_1_day&order_dir=desc" # (account_id, account_type) for the two accounts in the transfer chain. _TRANSFER_ACCOUNTS = ( ("81790", "vendor"), ("797716", "vendor"), ) def _accounts_for(pair) -> tuple[tuple[str, str], ...]: """The transfer accounts a profile's leaderboard should surface.""" if pair.key == "employee": return _TRANSFER_ACCOUNTS return tuple((str(aid), atype) for aid, atype in PROFILE_ACCOUNTS[pair.key]) def _top_accounts_url(account_id: str, account_type: str) -> str: """A /top-accounts-metrics URL scoped to a single transfer account. The endpoint excludes subaccount rows unless include_subaccounts=true, and a request mixing label_ids with subaccount_ids ANDs the two filters and returns nothing — so each transfer account is queried on its own. """ if account_type == "subaccount": scope = f"subaccount_ids={account_id}&include_subaccounts=true" else: scope = f"label_ids={account_id}" return f"{config.TOP_ACCOUNTS_METRICS_URL}?{scope}" class TestTopAccountsMetrics: """/top-accounts-metrics — cross-account leaderboard, keyed by account. A former-owner account's `streams_all_time` regains every product it transferred away once the flag routes through PRODUCT_OWNERSHIP_ACCESS, so the v2 figure can only be >= the legacy figure. The differential asserts that monotonicity for each transfer account's own row — catching a v2 path that silently drops an account's data. Each transfer account is fetched with its own scoped request (see _top_accounts_url). """ @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS) def test_account_rows(self, pair): """Each transfer account's all-time streams never shrink under the flag.""" checked = 0 for account_id, account_type in _accounts_for(pair): url = _top_accounts_url(account_id, account_type) on = assert_endpoint( url, headers=pair.ff_on, expect_payload_keys=[ "items", "total_count", "labels_count", "subaccounts_count", ], ) off = assert_endpoint( url, headers=pair.ff_off, expect_payload_keys=["items"] ) id_key = "subaccount_id" if account_type == "subaccount" else "label_id" on_row = find_row(on["items"], id_key, account_id) off_row = find_row(off["items"], id_key, account_id) if on_row is None and off_row is None: continue checked += 1 assert_ff_monotonic( (on_row or {}).get("streams_all_time") or 0, (off_row or {}).get("streams_all_time") or 0, label=f"{pair.key} /top-accounts-metrics account {account_id}", ) assert checked, ( f"{pair.key}: no transfer account surfaced on /top-accounts-metrics " f"— cannot verify the differential" ) class TestTopMetrics: """/top-metrics — global metrics leaderboard, isolated by `upc`. `upc=198000137366` filters the leaderboard down to product 5244974's own ISRCs, turning a catalogue leaderboard into a precise per-product transfer probe. Summed `streams_all_time` across those rows is product 5244974's figure: a former owner sees 0 of it without the flag (legacy hides the transferred-away product) and its frozen historical slice with the flag. """ @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS) def test_upc_isolates_transfer(self, pair): """Product 5244974's ISRC rows are correctly time-sliced for the FF-pair.""" url = f"{config.TOP_METRICS_URL}?upc={UPC}&limit=50" on = assert_endpoint(url, headers=pair.ff_on) off = assert_endpoint(url, headers=pair.ff_off) assert_isolated_transfer( sum_metric(on["items"], "streams_all_time"), sum_metric(off["items"], "streams_all_time"), role_key=pair.key, label=f"{pair.key} /top-metrics?upc product 5244974 streams_all_time", ) class TestTopSoundRecordings: """/top-sound-recordings — global sound-recording leaderboard. Under FF-ON the SQL swaps to `STREAMS_BY_TRACK_FEED_DISTRIBUTOR_ROLLUP_ PRODUCT_TRANSFER` and uses the v2 rollup permission filter. The leaderboard orders by `streams_1_day` (a recent window) and the response column `streams` is `SUM({{order_by}})` — i.e. the same recent metric. For a former owner the transferred-away product's recent slice is empty (~all streams predate the transfer-boundary day), so it never ranks in. What the FF-pair pins: with full access the v2 path reads `is_current=TRUE` rows only, so the employee's leaderboard must agree with legacy current-ownership attribution up to FF_DRIFT. For scoped non-employee viewers the legacy and `_PRODUCT_TRANSFER` rollups are independent dbt builds, and their per-row drift on `streams_1_day` can reshuffle the top-N membership over the drift — summed-streams monotonicity does not hold there. Only structural well-formedness is asserted for them. """ @pytest.mark.parametrize("pair", TRANSFER_PROFILE_PAIRS) def test_leaderboard(self, pair): """Employee FF-noop on summed streams; non-employee structural check.""" query = _PAGINATION url = f"{config.TOP_SOUND_RECORDINGS_URL}?{query}" on = assert_endpoint(url, headers=pair.ff_on) off = assert_endpoint(url, headers=pair.ff_off) label = f"{pair.key} /top-sound-recordings" if pair.key == "employee": assert_ff_noop( sum_metric(on["items"], "streams"), sum_metric(off["items"], "streams"), label=f"{label} streams", ) else: assert on["items"], f"{label}: v2 returned an empty leaderboard" for row in on["items"]: assert row.get("isrc"), f"{label}: leaderboard row missing isrc" # =========================================================================== # Participation-axis probes — the originating & destination artists # =========================================================================== # # Only the two *viewer-scoped* leaderboards apply to artists. /top-metrics and # /top-sound-recordings gate on the viewer's `permissions_filter` (the # participation branch for an artist), so the originating/destination artists # isolate product 5244974 exactly as ala_vendor/sued_vendor do. /top-accounts- # metrics is intentionally NOT mirrored here: it is account-scoped (it filters # directly on label_ids/subaccount_ids, not the viewer's participation), so an # artist cannot meaningfully exercise it — it stays on the ownership axis. class TestTopMetricsArtist: """/top-metrics?upc= — global leaderboard isolated to product 5244974. `upc=198000137366` filters to product 5244974's ISRCs, so summed `streams_all_time` is the product's figure for the artist viewer: the originating artist sees 0 without the flag (legacy hides the transferred-away product) and its frozen historical slice with it; the destination artist sees the product's full lifetime either way. """ @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS) def test_upc_isolates_transfer(self, pair): """Product 5244974's ISRC rows are time-sliced for the artist FF-pair.""" url = f"{config.TOP_METRICS_URL}?upc={UPC}&limit=50" on = assert_endpoint(url, headers=pair.ff_on) off = assert_endpoint(url, headers=pair.ff_off) assert_isolated_transfer( sum_metric(on["items"], "streams_all_time"), sum_metric(off["items"], "streams_all_time"), role_key=pair.key, label=f"{pair.key} /top-metrics?upc product 5244974 streams_all_time", ) class TestTopSoundRecordingsArtist: """/top-sound-recordings — windowed global leaderboard, participation axis. Both artists are *scoped* viewers (neither is full-access), so — as for the scoped ownership profiles (README §8) — the legacy and `_PRODUCT_TRANSFER` rollups are independent dbt builds whose per-row drift on the recent window can reshuffle the top-N; summed-streams monotonicity/no-op does not hold. Only structural well-formedness is asserted. The destination artist (live recent participation) returns a non-empty board; the originating artist's recent window is frozen before W2, so it may legitimately be empty. """ @pytest.mark.parametrize("pair", ARTIST_PROFILE_PAIRS) def test_leaderboard(self, pair): """Structural well-formedness; destination board non-empty.""" url = f"{config.TOP_SOUND_RECORDINGS_URL}?{_PAGINATION}" on = assert_endpoint(url, headers=pair.ff_on) off = assert_endpoint(url, headers=pair.ff_off) label = f"{pair.key} /top-sound-recordings" assert isinstance(on["items"], list) and isinstance(off["items"], list) for row in on["items"]: assert row.get("isrc"), f"{label}: leaderboard row missing isrc" if pair.key == "destination_artist": assert on["items"], ( f"{label}: the current participant got an empty leaderboard" )