"""Transfer-ownership tests for /sound-recording//* endpoints. ISRC BX69Y2100018 (the canonical ISRC — see conftest.ISRC) belongs only to product 5244974, so /sound-recording/BX69Y2100018/* is a clean probe of the product's transfer scoping. The heavy parametrize axes use that ISRC; a visibility sweep at the W1 probe runs across all 13 product ISRCs to catch per-ISRC permission leaks. Endpoints covered: - /sound-recording//timeseries (DAILY, FF-threaded) - /sound-recording//summary (DAILY, FF-threaded) - /sound-recording//streams-breakdown (DAILY, FF-threaded) - /sound-recording//streams-all (DAILY, FF-threaded) - /sound-recording//streams-by-store (DAILY, FF-threaded) - /sound-recording//aggregated-streams (ROLLUP, last-28-day window) - /sound-recording//top-markets (ROLLUP, 7-day window) """ 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_PROFILES, DESTINATION_ARTIST, ISRC, PROBES, PRODUCT_ISRCS, PROFILE_PAIRS, SEES_RECENT_WINDOW, TRANSFER_PROFILES, WINDOW_PROBES, assert_ff_noop, assert_visibility, expected_visibility, profile_key_for, ) ISRC_PARAMS = [pytest.param(isrc, id=isrc) for isrc in PRODUCT_ISRCS] def _sr_url(template: str, isrc: str, *, params: str = "") -> str: """Substitute an ISRC into a sound-recording URL template.""" base = template.replace("", isrc) return f"{base}?{params}" if params else base class TestSoundRecordingTimeseries: """/sound-recording//timeseries — DAILY grain.""" @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_streams(self, hdrs, probe): """Stream timeseries for the canonical ISRC; visibility per matrix.""" url = _sr_url( config.SOUND_RECORDING_TIMESERIES_URL, ISRC, params=f"type=TRACK_STREAMS&{PROBES[probe]}", ) payload = assert_endpoint(url, headers=hdrs) assert_visibility( payload, profile_key=profile_key_for(hdrs), probe=probe, metric_key="value" ) class TestSoundRecordingSummary: """/sound-recording//summary — DAILY grain, TOTAL aggregation.""" @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_total(self, hdrs, probe): """TOTAL summary for the canonical ISRC; visibility per matrix.""" url = _sr_url( config.SOUND_RECORDING_SUMMARY_URL, ISRC, 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("isrc", ISRC_PARAMS) def test_total_per_isrc_w1(self, hdrs, isrc): """Per-ISRC W1 visibility sweep across all 13 product ISRCs. Every transfer profile owned product 5244974 during W1, so every ISRC on the product must surface streams under the `early` probe for every profile. Catches a per-ISRC permission leak that a single-ISRC test would miss. """ url = _sr_url( config.SOUND_RECORDING_SUMMARY_URL, isrc, params=f"type=TOTAL&{PROBES['early']}&{PAGINATION}", ) payload = assert_endpoint(url, headers=hdrs) assert_visibility(payload, profile_key=profile_key_for(hdrs), probe="early") class TestSoundRecordingStreamsBreakdown: """/sound-recording//streams-breakdown — DAILY source-of-streams split. Shape varies, so this is a directional check: a former owner must not surface breakdown rows for a window after their cutoff. """ @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_default(self, hdrs, probe): """Breakdown returns 200; former owners leak no post-cutoff rows.""" url = _sr_url(config.STREAMS_BREAKDOWN_URL, ISRC, params=PROBES[probe]) payload = assert_endpoint(url, headers=hdrs, items_key=None) items = payload.get("items") if isinstance(items, list): profile_key = profile_key_for(hdrs) if expected_visibility(profile_key, probe) == "empty": assert not items, ( f"{profile_key} leaked {len(items)} streams-breakdown rows " f"for probe {probe}" ) class TestSoundRecordingStreamsAll: """/sound-recording//streams-all — DAILY aggregate stream timeseries.""" @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_streams(self, hdrs, probe): """Aggregate stream timeseries for the canonical ISRC; visibility per matrix.""" url = _sr_url(config.STREAMS_ALL_URL, ISRC, params=PROBES[probe]) payload = assert_endpoint(url, headers=hdrs, items_key=None) assert_visibility( payload, profile_key=profile_key_for(hdrs), probe=probe, items_key="items", metric_key="streams", ) class TestSoundRecordingStreamsByStore: """/sound-recording//streams-by-store — DAILY streams split per store.""" @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_streams(self, hdrs, probe): """Per-store stream timeseries for the canonical ISRC; visibility per matrix.""" url = _sr_url(config.STREAMS_BY_STORE_URL, ISRC, params=PROBES[probe]) payload = assert_endpoint(url, headers=hdrs, items_key=None) # streams-by-store nests the daily points under stores[].items[]; flatten # to one series so the shared visibility helper can sum a single metric. flat = [ point for store in payload.get("stores", []) for point in store.get("items", []) ] assert_visibility( {"items": flat}, profile_key=profile_key_for(hdrs), probe=probe, items_key="items", metric_key="streams", ) class TestSoundRecordingAggregatedStreams: """/sound-recording//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 = _sr_url( config.SOUND_RECORDING_AGGREGATED_STREAMS_URL, ISRC, 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 = _sr_url( config.SOUND_RECORDING_AGGREGATED_STREAMS_URL, ISRC, 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 /sound-recording/{ISRC}/aggregated-streams " f"dimension={dimension} total.streams_28_days" ), ) class TestSoundRecordingTopMarkets: """/sound-recording//top-markets — ROLLUP grain, 7-day window. `streams` is the per-market 7-day rollup metric, so the window sits in W2. This is a "no forward leak" probe: a former owner, frozen before W2, must surface no market rows carrying streams. """ @pytest.mark.parametrize("hdrs", TRANSFER_PROFILES) def test_markets(self, hdrs): """Current owner / employee see recent streams; former owners leak none.""" url = _sr_url(config.TOP_MARKETS_URL, ISRC) payload = assert_endpoint(url, headers=hdrs, items_key=None) for field in ("isrc", "sources", "items"): assert field in payload, f"missing field: {field}" total = sum((item.get("streams") or 0) for item in payload["items"]) profile_key = profile_key_for(hdrs) if profile_key in SEES_RECENT_WINDOW: assert total > 0, ( f"{profile_key} sees no recent market streams for {ISRC} — " f"the v2 path dropped the current owner's 7-day window" ) else: assert total == 0, ( f"{profile_key} leaked {total} post-transfer streams " f"into top-markets (a 7-day rollup)" ) # =========================================================================== # Participation-axis probes — the originating & destination artists # =========================================================================== # # ISRC BX69Y2100018 belongs only to product 5244974, so /sound-recording//* # is a clean probe of the product's transfer scoping reached through the # viewer's `permission_label_participant_ids` branch. The two artists resolve to # the same visibility shapes as ala_vendor (originating) and sued_vendor # (destination), so these classes mirror the ownership-axis ones above on the # ARTIST axes. class TestSoundRecordingTimeseriesArtist: """/sound-recording//timeseries — DAILY, participation branch.""" @pytest.mark.parametrize("hdrs", ARTIST_PROFILES) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_streams(self, hdrs, probe): """Stream timeseries for the canonical ISRC; visibility per matrix.""" url = _sr_url( config.SOUND_RECORDING_TIMESERIES_URL, ISRC, params=f"type=TRACK_STREAMS&{PROBES[probe]}", ) payload = assert_endpoint(url, headers=hdrs) assert_visibility( payload, profile_key=profile_key_for(hdrs), probe=probe, metric_key="value" ) class TestSoundRecordingSummaryArtist: """/sound-recording//summary — DAILY, TOTAL aggregation.""" @pytest.mark.parametrize("hdrs", ARTIST_PROFILES) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_total(self, hdrs, probe): """TOTAL summary for the canonical ISRC; visibility per matrix.""" url = _sr_url( config.SOUND_RECORDING_SUMMARY_URL, ISRC, 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("isrc", ISRC_PARAMS) def test_total_per_isrc_w1(self, hdrs, isrc): """Per-ISRC W1 visibility sweep across all 13 product ISRCs. Participation is product-level, so every ISRC on product 5244974 must surface streams at W1 for both artists — catches a per-ISRC leak the single-ISRC test would miss. """ url = _sr_url( config.SOUND_RECORDING_SUMMARY_URL, isrc, params=f"type=TOTAL&{PROBES['early']}&{PAGINATION}", ) payload = assert_endpoint(url, headers=hdrs) assert_visibility(payload, profile_key=profile_key_for(hdrs), probe="early") class TestSoundRecordingStreamsBreakdownArtist: """/sound-recording//streams-breakdown — directional no-forward-leak.""" @pytest.mark.parametrize("hdrs", ARTIST_PROFILES) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_default(self, hdrs, probe): """Breakdown returns 200; former participant leaks no post-cutoff rows.""" url = _sr_url(config.STREAMS_BREAKDOWN_URL, ISRC, params=PROBES[probe]) payload = assert_endpoint(url, headers=hdrs, items_key=None) items = payload.get("items") if isinstance(items, list): profile_key = profile_key_for(hdrs) if expected_visibility(profile_key, probe) == "empty": assert not items, ( f"{profile_key} leaked {len(items)} streams-breakdown rows " f"for probe {probe}" ) class TestSoundRecordingStreamsAllArtist: """/sound-recording//streams-all — DAILY aggregate stream timeseries.""" @pytest.mark.parametrize("hdrs", ARTIST_PROFILES) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_streams(self, hdrs, probe): """Aggregate stream timeseries for the canonical ISRC; visibility per matrix.""" url = _sr_url(config.STREAMS_ALL_URL, ISRC, params=PROBES[probe]) payload = assert_endpoint(url, headers=hdrs, items_key=None) assert_visibility( payload, profile_key=profile_key_for(hdrs), probe=probe, items_key="items", metric_key="streams", ) class TestSoundRecordingStreamsByStoreArtist: """/sound-recording//streams-by-store — DAILY streams split per store.""" @pytest.mark.parametrize("hdrs", ARTIST_PROFILES) @pytest.mark.parametrize("probe", WINDOW_PROBES) def test_streams(self, hdrs, probe): """Per-store stream timeseries for the canonical ISRC; visibility per matrix.""" url = _sr_url(config.STREAMS_BY_STORE_URL, ISRC, params=PROBES[probe]) payload = assert_endpoint(url, headers=hdrs, items_key=None) flat = [ point for store in payload.get("stores", []) for point in store.get("items", []) ] assert_visibility( {"items": flat}, profile_key=profile_key_for(hdrs), probe=probe, items_key="items", metric_key="streams", ) class TestSoundRecordingAggregatedStreamsArtist: """/sound-recording//aggregated-streams — last-28-day no-forward-leak.""" 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 = _sr_url( config.SOUND_RECORDING_AGGREGATED_STREAMS_URL, ISRC, 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: # Watermark false-positive shape, same as the ownership axis (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.""" url = _sr_url( config.SOUND_RECORDING_AGGREGATED_STREAMS_URL, ISRC, 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 /sound-recording/{ISRC}/aggregated-streams " f"dimension={dimension} total.streams_28_days" ), ) class TestSoundRecordingTopMarketsArtist: """/sound-recording//top-markets — ROLLUP 7-day no-forward-leak.""" @pytest.mark.parametrize("hdrs", ARTIST_PROFILES) def test_markets(self, hdrs): """Destination artist sees recent streams; originating artist leaks none.""" url = _sr_url(config.TOP_MARKETS_URL, ISRC) payload = assert_endpoint(url, headers=hdrs, items_key=None) for field in ("isrc", "sources", "items"): assert field in payload, f"missing field: {field}" total = sum((item.get("streams") or 0) for item in payload["items"]) profile_key = profile_key_for(hdrs) if profile_key in SEES_RECENT_WINDOW: assert total > 0, ( f"{profile_key} sees no recent market streams for {ISRC} — " f"the v2 path dropped the current participant's 7-day window" ) else: assert total == 0, ( f"{profile_key} leaked {total} post-transfer streams " f"into top-markets (a 7-day rollup)" )