"""Integration tests for sound recording endpoints. Endpoints tested: - /sound-recording//timeseries (TestSoundRecordingTimeseries) - /sound-recording//summary (TestSoundRecordingSummary) - /sound-recording//tiktok/summary (TestSoundRecordingTikTokSummary) - /sound-recording//tiktok/aggregated-summary (TestSoundRecordingTikTokAggregated) - /sound-recording//tiktok/aggregated-timeseries (TestSoundRecordingTikTokAggregated) - /sound-recording//tiktok/timeseries (TestSoundRecordingTikTokTimeseries) - /sound-recording//metadata (TestSoundRecordingMetadata) - /sound-recording//streams-breakdown (TestStreamsBreakdown) - /sound-recording//streams-all (TestStreamsAllAndByStore) - /sound-recording//streams-by-store (TestStreamsAllAndByStore) - /sound-recording//demographics (TestSoundRecordingDemographics) - /sound-recording//top-markets (TestTopMarkets) - /sound-recording//aggregated-streams (TestSoundRecordingAggregatedStreams) - /sound-recording//related-videos-by-isrc (TestRelatedVideosByIsrc) - /sound-recording/aggregate-streams (TestAggregateStreams) - /sound-recording/streams (TestStreamsBulk) Every test parametrizes on ALL_HEADERS (6 auth profiles) to verify access control. Sound recording timeseries/summary endpoints also parametrize on DISTRIBUTOR_COMBOS to verify that the distributors filter determines data visibility. """ import os import pytest from analytics import config from analytics.config import BASE_URL from analytics.logic.sound_recording_timeseries import ( SOUND_RECORDING_TABLES_SUMMARY, SOUND_RECORDING_TABLES_TIMESERIES, ) from tests.integration.endpoints.conftest import ( ALL_HEADERS, INSIGHTS_ARTIST, INSIGHTS_D3, INSIGHTS_EMPLOYEE, INSIGHTS_LABEL, INSIGHTS_LABEL_AND_SUBACCOUNT, INSIGHTS_SUBACCOUNT, assert_endpoint, assert_timeseries_keys, ) # --------------------------------------------------------------------------- # Sound-recording-specific constants # --------------------------------------------------------------------------- SUMMARY_TYPES = [t for t in SOUND_RECORDING_TABLES_SUMMARY.keys() if t != "SOS_V2"] TIMESERIES_TYPES = [ t for t in SOUND_RECORDING_TABLES_TIMESERIES.keys() if t != "TRACK_STREAMS_BY_SOS_V2" ] DISTRIBUTOR_COMBOS = [ "distributors=THEORCHARD&distributors=AWAL", "distributors=THEORCHARD&distributors=SME&distributors=AWAL", ] SOS_FILTER_COMBOS = [ # (store_ids_param, stream_sources_param) ("&store_ids=1", "&stream_sources=discovery"), ("&store_ids=1", "&stream_sources=discovery&stream_sources=external"), ("&store_ids=1", ""), ("&store_ids=286", "&stream_sources=albumpage"), ("&store_ids=286", "&stream_sources=artistpage&stream_sources=albumpage"), ("&store_ids=286", ""), ("&store_ids=187", "&stream_sources=album"), ("&store_ids=187", "&stream_sources=album&stream_sources=artist"), ("&store_ids=1&store_ids=187", ""), ("&store_ids=1&store_ids=187", "&stream_sources=album&stream_sources=external"), ("", ""), ] # TikTok tests use two ISRCs because TikTok data access is label-scoped: # - Bad Bunny (QM6N21919851, subaccount 46189 / label 26760): d3, subaccount, label_and_subaccount # - Local Natives on Frenchkiss (GBZUZ1200056, label 6971): label, artist, label_and_subaccount # employee sees both; label_and_subaccount appears in both clusters (has access to both labels). _BAD_BUNNY_ISRC = "QM6N21919851" _LOCAL_NATIVES_ISRC = "GBZUZ1200056" TIKTOK_ISRC_HEADERS = [ pytest.param(_BAD_BUNNY_ISRC, INSIGHTS_EMPLOYEE, id="bad_bunny-employee"), pytest.param(_BAD_BUNNY_ISRC, INSIGHTS_D3, id="bad_bunny-d3"), pytest.param(_BAD_BUNNY_ISRC, INSIGHTS_SUBACCOUNT, id="bad_bunny-subaccount"), pytest.param( _BAD_BUNNY_ISRC, INSIGHTS_LABEL_AND_SUBACCOUNT, id="bad_bunny-label_and_subaccount", ), pytest.param(_LOCAL_NATIVES_ISRC, INSIGHTS_LABEL, id="local_natives-label"), pytest.param(_LOCAL_NATIVES_ISRC, INSIGHTS_ARTIST, id="local_natives-artist"), ] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _should_have_data(hdrs, distributors): """Label profile (INSIGHTS_LABEL) has no access; also needs SME in distributors. Sound recording endpoints require the ISRC's distributor (SME) to be in the distributors param. INSIGHTS_LABEL never has access to this ISRC. """ return hdrs != INSIGHTS_LABEL and "SME" in distributors def _assert_sr_summary_keys(keys, summary_type): """Assert expected keys on a sound recording summary item. Sound recording summary uses 'downloads' (not 'downloads_track'). SOS types omit saves, skip_rate, and downloads. """ assert "streams" in keys if "SOS" in summary_type: return assert "saves" in keys assert "skip_rate" in keys assert "downloads" in keys class TestSoundRecordingTimeseries: """Integration tests for /sound-recording//timeseries. Tests cover all timeseries types across distributor combos, with country, store, and SOS filter variants. Groups: 1. All types shape — every type returns correct keys 2. Country filter — countries param narrows results 3. Store filter — store_ids on BY_STORE 4. Data: country canaries — BY_COUNTRY contains expected codes 5. Country + SOS combos — BY_COUNTRY with SOS filter variants 6. SOS breakdown — BY_SOS returns expected categories 7. SOS V2 — BY_SOS_V2 with filter combos """ ISRC = "QZ9QQ2300457" START_DATE = "2023-09-01" END_DATE = "2023-09-02" def _url(self, ts_type, distributors, extra=""): base = config.SOUND_RECORDING_TIMESERIES_URL.replace("", self.ISRC) return ( f"{base}" f"?type={ts_type}&start_date={self.START_DATE}" f"&end_date={self.END_DATE}&{distributors}{extra}" ) # ── Group 1: All types return correct response shape ──────────── # # 6 headers × 2 distributor_combos × N timeseries_types. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) @pytest.mark.parametrize("ts_type", TIMESERIES_TYPES) def test_timeseries(self, hdrs, distributors, ts_type): """Every timeseries type returns data with correct keys.""" payload = assert_endpoint( self._url(ts_type, distributors), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: assert_timeseries_keys(payload["items"][0].keys(), ts_type) # ── Group 2: Country filter ──────────────────────────────────── # # 6 headers × 2 distributor_combos × N timeseries_types. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) @pytest.mark.parametrize("ts_type", TIMESERIES_TYPES) def test_timeseries_country_filtered(self, hdrs, distributors, ts_type): """Country filter activates the country-variant Snowflake table.""" payload = assert_endpoint( self._url(ts_type, distributors, "&countries=CA&countries=US"), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: assert_timeseries_keys(payload["items"][0].keys(), ts_type) # ── Group 3: Store filter for BY_STORE ───────────────────────── # # 6 headers × 2 distributor_combos. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) def test_timeseries_by_store_store_filtered(self, hdrs, distributors): """store_ids=1 filters BY_STORE to a single store. Verifies the WHERE store_id IN (...) SQL clause works. All returned items must have id=='1'. """ payload = assert_endpoint( self._url("TRACK_STREAMS_BY_STORE", distributors, "&store_ids=1"), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: assert_timeseries_keys(payload["items"][0].keys(), "TRACK_STREAMS_BY_STORE") assert payload["items"][0]["id"] == "1" # ── Group 4: Data canaries — BY_COUNTRY ──────────────────────── # # 6 headers × 2 distributor_combos × 2 methods. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) def test_streams_by_country(self, hdrs, distributors): """BY_COUNTRY includes US and CA for this ISRC. Canary test: catches data pipeline regressions where country data disappears. """ payload = assert_endpoint( self._url("TRACK_STREAMS_BY_COUNTRY", distributors), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: country_codes = [i["id"] for i in payload["items"]] assert "US" in country_codes assert "CA" in country_codes @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) def test_streams_by_country_country_filtered(self, hdrs, distributors): """countries=US&countries=CA returns exactly those two countries.""" payload = assert_endpoint( self._url( "TRACK_STREAMS_BY_COUNTRY", distributors, "&countries=US&countries=CA", ), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: assert sorted(["US", "CA"]) == sorted( [payload["items"][0]["id"], payload["items"][1]["id"]] ) # ── Group 5: BY_COUNTRY + SOS filter combinations ────────────── # # 6 headers × 2 distributor_combos × 11 SOS_FILTER_COMBOS. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) @pytest.mark.parametrize("store_ids,sos", SOS_FILTER_COMBOS) def test_streams_by_country_filtered_by_sos( self, hdrs, distributors, store_ids, sos ): """BY_COUNTRY with store_ids + stream_sources filter combos. Exercises the SOS-filtered country query path across all store/source combinations. """ payload = assert_endpoint( self._url( "TRACK_STREAMS_BY_COUNTRY", distributors, f"&countries=US&countries=CA{store_ids}{sos}", ), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: assert sorted(["US", "CA"]) == sorted( [payload["items"][0]["id"], payload["items"][1]["id"]] ) # ── Group 6: SOS breakdown ───────────────────────────────────── # # 6 headers × 2 distributor_combos. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) def test_streams_by_sos(self, hdrs, distributors): """BY_SOS breaks down into active, passive, collection, unknown. Verifies the SOS breakdown logic works end-to-end. """ payload = assert_endpoint( self._url("TRACK_STREAMS_BY_SOS", distributors), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: item_ids = list(dict.fromkeys([i["id"] for i in payload["items"]])) assert item_ids == ["active", "passive", "collection", "unknown"] # ── Group 7: SOS V2 with filter combinations ─────────────────── # # 6 headers × 2 distributor_combos × 11 SOS_FILTER_COMBOS. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) @pytest.mark.parametrize("store_ids,sos", SOS_FILTER_COMBOS) def test_streams_by_sos_v2(self, hdrs, distributors, store_ids, sos): """SOS_V2 with store_ids + stream_sources filter combos. Exercises the V2 SOS query path across all filter combinations. """ payload = assert_endpoint( self._url( "TRACK_STREAMS_BY_SOS_V2", distributors, f"{store_ids}{sos}", ), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: assert_timeseries_keys( payload["items"][0].keys(), "TRACK_STREAMS_BY_SOS_V2" ) class TestSoundRecordingSummary: """Integration tests for /sound-recording//summary. Tests cover all summary types across distributor combos, with country, store, and SOS filter variants. Groups: 1. All types shape — every type returns correct keys 2. Country filter — countries param narrows results 3. Store filter — store_ids on STORE type 4. Data: country canaries — COUNTRY type contains expected codes 5. Country + SOS combos — COUNTRY with SOS filter variants 6. SOS breakdown — SOS returns expected categories 7. SOS V2 — SOS_V2 with filter combos """ ISRC = "QZ9QQ2300457" START_DATE = "2023-09-01" END_DATE = "2023-09-02" def _url(self, summary_type, distributors, extra=""): base = config.SOUND_RECORDING_SUMMARY_URL.replace("", self.ISRC) return ( f"{base}" f"?type={summary_type}&start_date={self.START_DATE}" f"&end_date={self.END_DATE}&{distributors}{extra}" ) # ── Group 1: All types return correct response shape ──────────── # # 6 headers × 2 distributor_combos × N summary_types. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) @pytest.mark.parametrize("summary_type", SUMMARY_TYPES) def test_summary(self, hdrs, distributors, summary_type): """Every summary type returns data with correct keys.""" payload = assert_endpoint( self._url(summary_type, distributors), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: _assert_sr_summary_keys(payload["items"][0].keys(), summary_type) # ── Group 2: Country filter ──────────────────────────────────── # # 6 headers × 2 distributor_combos × N summary_types. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) @pytest.mark.parametrize("summary_type", SUMMARY_TYPES) def test_summary_country_filtered(self, hdrs, distributors, summary_type): """Country filter activates the country-variant query path.""" payload = assert_endpoint( self._url(summary_type, distributors, "&countries=CA&countries=US"), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: _assert_sr_summary_keys(payload["items"][0].keys(), summary_type) # ── Group 3: Store filter for STORE type ─────────────────────── # # 6 headers × 2 distributor_combos. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) def test_summary_by_store_store_filtered(self, hdrs, distributors): """store_ids=1 filters STORE summary to a single store. Verifies the WHERE store_id IN (...) SQL clause works. """ payload = assert_endpoint( self._url("STORE", distributors, "&store_ids=1"), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: _assert_sr_summary_keys(payload["items"][0].keys(), "STORE") assert payload["items"][0]["id"] == 1 # ── Group 4: Data canaries — COUNTRY type ────────────────────── # # 6 headers × 2 distributor_combos × 2 methods. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) def test_summary_by_country(self, hdrs, distributors): """COUNTRY summary includes US and CA for this ISRC. Canary test: catches data pipeline regressions where country data disappears. """ payload = assert_endpoint( self._url("COUNTRY", distributors), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: country_codes = [i["id"] for i in payload["items"]] assert "US" in country_codes assert "CA" in country_codes @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) def test_summary_by_country_country_filtered(self, hdrs, distributors): """countries=US&countries=CA returns exactly those two countries.""" payload = assert_endpoint( self._url("COUNTRY", distributors, "&countries=US&countries=CA"), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: assert sorted(["US", "CA"]) == sorted( [payload["items"][0]["id"], payload["items"][1]["id"]] ) # ── Group 5: COUNTRY + SOS filter combinations ───────────────── # # 6 headers × 2 distributor_combos × 11 SOS_FILTER_COMBOS. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) @pytest.mark.parametrize("store_ids,sos", SOS_FILTER_COMBOS) def test_summary_by_country_filtered_by_sos( self, hdrs, distributors, store_ids, sos ): """COUNTRY summary with store_ids + stream_sources filter combos. Exercises the SOS-filtered country query path across all store/source combinations. """ payload = assert_endpoint( self._url( "COUNTRY", distributors, f"&countries=US&countries=CA{store_ids}{sos}", ), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: assert sorted(["US", "CA"]) == sorted( [payload["items"][0]["id"], payload["items"][1]["id"]] ) # ── Group 6: SOS breakdown ───────────────────────────────────── # # 6 headers × 2 distributor_combos. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) def test_streams_by_sos(self, hdrs, distributors): """SOS breaks down into active, passive, collection, unknown. Verifies the SOS breakdown logic works end-to-end. """ payload = assert_endpoint( self._url("SOS", distributors), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: item_ids = list(dict.fromkeys([i["id"] for i in payload["items"]])) assert item_ids == ["active", "passive", "collection", "unknown"] # ── Group 7: SOS V2 with filter combinations ─────────────────── # # 6 headers × 2 distributor_combos × 11 SOS_FILTER_COMBOS. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("distributors", DISTRIBUTOR_COMBOS) @pytest.mark.parametrize("store_ids,sos", SOS_FILTER_COMBOS) def test_streams_by_sos_v2(self, hdrs, distributors, store_ids, sos): """SOS_V2 with store_ids + stream_sources filter combos. Exercises the V2 SOS query path across all filter combinations. """ payload = assert_endpoint( self._url("SOS_V2", distributors, f"{store_ids}{sos}"), headers=hdrs, expect_empty=not _should_have_data(hdrs, distributors), ) if _should_have_data(hdrs, distributors) and payload["items"]: _assert_sr_summary_keys(payload["items"][0].keys(), "SOS_V2") class TestSoundRecordingTikTokSummary: """Integration tests for /sound-recording//tiktok/summary. Tests cover default summary, by-country, and by-content-type variants. Uses two ISRCs to cover all profile clusters (see TIKTOK_ISRC_HEADERS). Groups: 1. Default summary — total summary (6 tests) 2. By country — BY_COUNTRY type and country filter (12 tests) 3. By content type — PGC/UGC breakdown (6 tests) """ START_DATE = "2024-01-01" END_DATE = "2024-01-02" def _url(self, isrc, extra=""): base = config.SOUND_RECORDING_TIKTOK_SUMMARY_URL.replace("", isrc) return ( f"{base}" f"?start_date={self.START_DATE}&end_date={self.END_DATE}{extra}" ) # ── Group 1: Default summary ─────────────────────────────────── # # 6 tests (one per isrc/header pair). @pytest.mark.parametrize("isrc,hdrs", TIKTOK_ISRC_HEADERS) def test_summary(self, isrc, hdrs): """Default TikTok summary returns exactly one item.""" payload = assert_endpoint( self._url(isrc), headers=hdrs, min_items=1, ) assert len(payload["items"]) == 1 # ── Group 2: By country ──────────────────────────────────────── # # 6 isrc/header pairs × 2 methods = 12 tests. @pytest.mark.parametrize("isrc,hdrs", TIKTOK_ISRC_HEADERS) def test_by_country(self, isrc, hdrs): """BY_COUNTRY returns items with id (country code).""" payload = assert_endpoint( self._url(isrc, "&type=BY_COUNTRY"), headers=hdrs, min_items=1, ) assert "id" in payload["items"][0] @pytest.mark.parametrize("isrc,hdrs", TIKTOK_ISRC_HEADERS) def test_by_country_country_filtered(self, isrc, hdrs): """BY_COUNTRY + countries=US returns only US.""" payload = assert_endpoint( self._url(isrc, "&type=BY_COUNTRY&countries=US"), headers=hdrs, min_items=1, ) assert payload["items"][0]["id"] == "US" # ── Group 3: By content type ─────────────────────────────────── # # 6 tests (one per isrc/header pair). @pytest.mark.parametrize("isrc,hdrs", TIKTOK_ISRC_HEADERS) def test_by_content_type(self, isrc, hdrs): """BY_CONTENT_TYPE breaks down into PGC and UGC.""" payload = assert_endpoint( self._url(isrc, "&type=BY_CONTENT_TYPE"), headers=hdrs, min_items=2, ) assert sorted(["PGC", "UGC"]) == sorted( [payload["items"][0]["id"], payload["items"][1]["id"]] ) class TestSoundRecordingTikTokAggregated: """Integration tests for TikTok aggregated-summary and aggregated-timeseries. Groups: 1. Aggregated summary — returns non-empty payload (6 tests) 2. Aggregated timeseries — returns non-empty payload (6 tests) """ ISRC = "QMDA62184373" # ── Group 1: Aggregated summary ──────────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_aggregated_summary(self, hdrs): """TikTok aggregated summary returns a non-empty payload.""" payload = assert_endpoint( config.SOUND_RECORDING_TIKTOK_AGGREGATED_SUMMARY_URL.replace( "", self.ISRC ), headers=hdrs, items_key=None, ) assert payload # ── Group 2: Aggregated timeseries ───────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_aggregated_timeseries(self, hdrs): """TikTok aggregated timeseries returns a non-empty payload.""" payload = assert_endpoint( config.SOUND_RECORDING_TIKTOK_AGGREGATED_TIMESERIES_URL.replace( "", self.ISRC ), headers=hdrs, items_key=None, ) assert payload # =========================================================================== # Sound recording metadata # =========================================================================== @pytest.mark.skip(reason="Brittle, need AWS creds in environment.") class TestSoundRecordingMetadata: """Integration tests for /sound-recording//metadata. Tests cover metadata retrieval across auth profiles, including deleted product visibility, artist sub-fields, product sub-fields, and the absence of the subaccount field. Groups: 1. Basic metadata — fields and ISRC value (12 tests) 2. No-products ISRC — products list is empty (6 tests) 3. Deleted products — include_deleted=true (6 tests) 4. Artist sub-fields — primary_artists contain expected keys (6 tests) 5. Product sub-fields — products contain expected keys (6 tests) 6. No subaccount field — subaccount not exposed (6 tests) """ ISRC_WITH_PRODUCTS = "QM6P41952433" ISRC_NO_PRODUCTS = "DED831000251" def _url(self, isrc, extra=""): base = config.SOUND_RECORDING_METADATA_URL.replace("", isrc) return f"{base}{extra}" # ── Group 1: Basic metadata ─────────────────────────────────────── @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_metadata_fields(self, hdrs): """Metadata returns isrc, name, primary_artists, products.""" assert_endpoint( self._url(self.ISRC_WITH_PRODUCTS), headers=hdrs, items_key=None, expect_payload_keys=["isrc", "name", "primary_artists", "products"], ) @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_metadata_returns_isrc(self, hdrs): """Metadata response contains the requested ISRC.""" payload = assert_endpoint( self._url(self.ISRC_NO_PRODUCTS), headers=hdrs, items_key=None, ) assert payload["isrc"] == self.ISRC_NO_PRODUCTS # ── Group 2: No-products ISRC ───────────────────────────────────── @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_no_products(self, hdrs): """ISRC with no active products returns empty products list.""" payload = assert_endpoint( self._url(self.ISRC_NO_PRODUCTS), headers=hdrs, items_key=None, ) assert len(payload["products"]) == 0 # ── Group 3: Deleted products ───────────────────────────────────── @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_include_deleted(self, hdrs): """include_deleted=true shows deleted products. DED831000251 has one deleted product that only appears with the include_deleted flag. """ payload = assert_endpoint( self._url(self.ISRC_NO_PRODUCTS, "?include_deleted=true"), headers=hdrs, items_key=None, ) assert len(payload["products"]) == 1 assert payload["products"][0]["is_deleted"] # ── Group 4: Artist sub-fields ──────────────────────────────────── @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_artist_fields(self, hdrs): """Primary artists contain artist_type and artist_name.""" payload = assert_endpoint( self._url(self.ISRC_WITH_PRODUCTS), headers=hdrs, items_key=None, ) for artist in payload["primary_artists"]: assert "artist_type" in artist assert "artist_name" in artist # ── Group 5: Product sub-fields ─────────────────────────────────── @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_product_fields(self, hdrs): """Products contain product_id, product_name, release_date, format, sale_start_date, and primary_artists.""" payload = assert_endpoint( self._url(self.ISRC_WITH_PRODUCTS), headers=hdrs, items_key=None, ) expected_fields = [ "product_id", "product_name", "release_date", "format", "sale_start_date", "primary_artists", ] for product in payload["products"]: for field in expected_fields: assert field in product # ── Group 6: No subaccount field ────────────────────────────────── @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_no_subaccount_field(self, hdrs): """Metadata does not expose a subaccount field.""" payload = assert_endpoint( self._url(self.ISRC_NO_PRODUCTS), headers=hdrs, items_key=None, ) assert "subaccount" not in payload # =========================================================================== # Streams breakdown # =========================================================================== class TestStreamsBreakdown: """Integration tests for /sound-recording//streams-breakdown. Tests cover streams breakdown data across all auth profiles, with store filters and various date parameter styles. Groups: 1. Basic breakdown — default request (6 tests) 2. With store filter — store_ids narrows breakdown (6 tests) 3. Date param styles — end_date, negative days, HIGHWATERMARK (30 tests) """ ISRC = "QM6P41952433" START_DATE = "2019-11-01" DAYS = "28" EXPECTED_FIELDS = ["isrc", "source_of_streams", "streams_by_subscription"] def _url(self, extra=""): base = config.STREAMS_BREAKDOWN_URL.replace("", self.ISRC) return ( f"{base}" f"?country_code=US&start_date={self.START_DATE}&days={self.DAYS}{extra}" ) # ── Group 1: Basic breakdown ────────────────────────────────────── @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_streams_breakdown(self, hdrs): """Streams breakdown returns isrc, source_of_streams, streams_by_subscription.""" assert_endpoint( self._url(), headers=hdrs, items_key=None, expect_payload_keys=self.EXPECTED_FIELDS, ) # ── Group 2: With store filter ──────────────────────────────────── @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_streams_breakdown_with_store_filter(self, hdrs): """Streams breakdown with store_ids filter still returns expected fields.""" assert_endpoint( self._url("&store_ids=1&store_ids=286"), headers=hdrs, items_key=None, expect_payload_keys=self.EXPECTED_FIELDS, ) # ── Group 3: Date parameter styles ──────────────────────────────── # # 6 headers × 5 date_param variants = 30 tests. # Exercises end_date, negative days, and HIGHWATERMARK code paths. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "date_params", [ pytest.param("start_date=2019-11-01&end_date=2019-12-01", id="end_date"), pytest.param("start_date=2019-11-01&days=30", id="positive_days"), pytest.param("start_date=2019-11-30&days=-30", id="negative_days"), pytest.param("start_date=HIGHWATERMARK&days=30", id="hwm_positive"), pytest.param("start_date=HIGHWATERMARK&days=-30", id="hwm_negative"), pytest.param("start_date=ALL_TIME&end_date=2019-12-01", id="all_time"), ], ) def test_date_param_styles(self, hdrs, date_params): """Streams breakdown accepts various date parameter styles. Exercises end_date (instead of days), negative days (backward range), HIGHWATERMARK (resolved server-side to latest available date), and start_date=ALL_TIME (omits start_date, SQL uses `<= end_date`). """ base = config.STREAMS_BREAKDOWN_URL.replace("", self.ISRC) url = f"{base}?country_code=US&{date_params}" assert_endpoint( url, headers=hdrs, items_key=None, expect_payload_keys=self.EXPECTED_FIELDS, ) @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_store_ids_intersection_empty_returns_empty_body(self, hdrs): """Only-unavailable store_ids → empty body (no invalid SQL, no 500).""" assert_endpoint( self._url("&store_ids=99999"), headers=hdrs, items_key=None, expect_payload_keys=self.EXPECTED_FIELDS, ) # =========================================================================== # Streams all + Streams by store # =========================================================================== class TestStreamsAllAndByStore: """Integration tests for /sound-recording//streams-all and .../streams-by-store. Both endpoints share the same query params and branching logic. Groups exercise each meaningful SQL branch in `analytics.queries.sound_recording_streams`: - country_ids absent vs present → switches Snowflake view (V_STREAMS_BY_TRACK_FEED_DISTRIBUTOR_DAILY vs ..._COUNTRY_...) - start_date == ALL_TIME → SQL omits start_date, uses `<= end_date` - store_ids provided → intersects with available stores - distributors non-default → tighter IN clause - no date params → falls back to HIGHWATERMARK − 28 days Groups: 1. Streams all — base + filter variants (36 tests) 2. Streams by store — base + filter variants (36 tests) """ ISRC = "DED831000251" QUERY_VARIANTS = [ pytest.param("?start_date=2021-02-24&days=7", id="base_date_range"), pytest.param("?start_date=all_time", id="all_time"), pytest.param( "?start_date=2021-02-24&days=7&country_code=US", id="country_filter", ), pytest.param( "?start_date=2021-02-24&days=7&store_ids=1&store_ids=286", id="store_filter", ), pytest.param( "?start_date=2021-02-24&days=7&distributors=theorchard", id="distributors_filter", ), pytest.param("", id="default_date_range"), ] # ── Group 1: Streams all ─────────────────────────────────────── # # 6 headers × 6 query variants = 36 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("query", QUERY_VARIANTS) def test_streams_all(self, hdrs, query): """Aggregate streams timeseries returns isrc and items. Each variant hits a distinct SQL branch in the new query class (country view switch, all_time clause, store/distributor filters, default date range). """ base = config.STREAMS_ALL_URL.replace("", self.ISRC) assert_endpoint( base + query, headers=hdrs, items_key=None, expect_payload_keys=["isrc", "items"], ) # ── Group 2: Streams by store ────────────────────────────────── # # 6 headers × 6 query variants = 36 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("query", QUERY_VARIANTS) def test_streams_by_store(self, hdrs, query): """Per-store streams returns isrc and stores. Each variant hits a distinct SQL branch in the new query class (country view switch, all_time clause, store/distributor filters, default date range). """ base = config.STREAMS_BY_STORE_URL.replace("", self.ISRC) assert_endpoint( base + query, headers=hdrs, items_key=None, expect_payload_keys=["isrc", "stores"], ) # =========================================================================== # Sound recording demographics # =========================================================================== SR_DEMOGRAPHICS_QUERY_VARIANTS = [ pytest.param( "?start_date=2023-01-01&end_date=2023-01-28", id="base_date_range", ), pytest.param( "?start_date=2023-01-01&end_date=2023-01-28&country_code=US", id="country_filter", ), pytest.param( "?start_date=2023-01-01&end_date=2023-01-28" "&country_code=US&country_code=MX", id="multi_country_filter", ), pytest.param( "?start_date=2023-01-01&end_date=2023-01-28&store_ids=286", id="store_filter", ), pytest.param( "?start_date=2023-01-01&end_date=2023-01-28&distributors=theorchard", id="distributors_filter", ), pytest.param("", id="default_date_range"), pytest.param("?days=0&start_date=HIGHWATERMARK", id="all_time_highwatermark"), ] class TestSoundRecordingDemographics: """Integration tests for /sound-recording//demographics. Parametrized by 7 SQL-branch variants × 6 auth headers × 2 ISRCs = 84 tests. The two ISRCs exercise different label-access paths: - QMDA72252608 (Bad Bunny, RIMAS 26760): employee, d3, label_and_subaccount - GBZUZ0900064 (Local Natives, Frenchkiss 6971): employee, label, label_and_subaccount, artist Each variant hits a distinct SQL branch: base date range, country filter, multi-country filter, store filter, distributors filter, the HIGHWATERMARK − 28 days default fallback, and the all-time graphql shape (?days=0&start_date=HIGHWATERMARK). """ @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "isrc", [ pytest.param("QMDA72252608", id="bad_bunny"), pytest.param("GBZUZ0900064", id="local_natives"), ], ) @pytest.mark.parametrize("query", SR_DEMOGRAPHICS_QUERY_VARIANTS) def test_demographics(self, hdrs, isrc, query): """Demographics returns isrc, demographics, and sources.""" base = config.DEMOGRAPHICS_URL.replace("", isrc) assert_endpoint( f"{base}{query}", headers=hdrs, items_key=None, expect_payload_keys=["isrc", "demographics", "sources"], ) # =========================================================================== # Top markets # =========================================================================== class TestTopMarkets: """Integration tests for /sound-recording//top-markets. Each variant exercises a distinct SQL branch in the new query class: - country_ids absent vs present → adds `country_code IN (...)` clause - store_ids provided → intersects with available stores - distributors non-default → tighter `distributor IN (...)` clause The endpoint reads from a 7-day rollup table (no start_date / end_date parameters), so there's no time-range branch to test. Groups: 1. Top markets — base + filter variants (24 tests) 2. Country filter narrows results — country_code=CA only (6 tests) """ ISRC = "DED831000251" EXPECTED_FIELDS = ["isrc", "sources", "items"] QUERY_VARIANTS = [ pytest.param("", id="base"), pytest.param("?country_code=US", id="country_filter"), pytest.param("?store_ids=1&store_ids=286", id="store_filter"), pytest.param("?distributors=theorchard", id="distributors_filter"), ] # ── Group 1: Top markets ────────────────────────────────────────── # # 6 headers × 4 query variants = 24 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("query", QUERY_VARIANTS) def test_top_markets(self, hdrs, query): """Top markets returns isrc, sources, and items. Each variant hits a distinct SQL branch in the query class (country WHERE clause, store filter intersection, distributor IN tightening). """ base = config.TOP_MARKETS_URL.replace("", self.ISRC) assert_endpoint( base + query, headers=hdrs, items_key=None, expect_payload_keys=self.EXPECTED_FIELDS, ) # ── Group 2: Country filter narrows results ─────────────────────── @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_top_markets_country_filtered(self, hdrs): """Top markets with country_code=CA returns only CA items.""" base = config.TOP_MARKETS_URL.replace("", self.ISRC) payload = assert_endpoint( f"{base}?country_code=CA", headers=hdrs, items_key=None, expect_payload_keys=self.EXPECTED_FIELDS, ) if payload.get("items"): assert all(item["country_code"] == "CA" for item in payload["items"]) # =========================================================================== # Sound recording aggregated streams # =========================================================================== class TestSoundRecordingAggregatedStreams: """Integration tests for /sound-recording//aggregated-streams. Tests cover aggregated streams across store, country, and SOS dimensions, including the days_back parameter. Groups: 1. All dimensions — store/country/sos return expected fields (18 tests) 2. Days back — days_back=7 limits timeseries length (6 tests) """ ISRC = "QM6P42334528" EXPECTED_FIELDS = [ "all_other_rollup", "all_other_timeseries", "topn_timeseries", "topn_rollup", "total", ] def _url(self, extra=""): base = config.SOUND_RECORDING_AGGREGATED_STREAMS_URL.replace( "", self.ISRC ) return f"{base}{extra}" # ── Group 1: All dimensions ─────────────────────────────────────── # # 6 headers × 3 dimensions = 18 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("dimension", ["store", "country", "sos"]) def test_aggregated_streams(self, hdrs, dimension): """Aggregated streams returns expected fields for each dimension. Timeseries length is verified only for the employee profile (guaranteed full data access). """ payload = assert_endpoint( self._url(f"?dimension={dimension}"), headers=hdrs, items_key=None, expect_payload_keys=self.EXPECTED_FIELDS, ) if hdrs == INSIGHTS_EMPLOYEE and payload.get("all_other_timeseries"): assert len(payload["all_other_timeseries"]) == 28 for _, timeseries in payload["topn_timeseries"].items(): assert len(timeseries) == 28 # ── Group 2: Days back ──────────────────────────────────────────── @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_aggregated_streams_days_back(self, hdrs): """days_back=7 limits timeseries to 7 items.""" payload = assert_endpoint( self._url("?dimension=country&days_back=7"), headers=hdrs, items_key=None, expect_payload_keys=self.EXPECTED_FIELDS, ) if hdrs == INSIGHTS_EMPLOYEE and payload.get("all_other_timeseries"): assert len(payload["all_other_timeseries"]) == 7 for _, timeseries in payload["topn_timeseries"].items(): assert len(timeseries) == 7 # =========================================================================== # TikTok timeseries # =========================================================================== class TestSoundRecordingTikTokTimeseries: """Integration tests for /sound-recording//tiktok/timeseries. Tests cover TikTok timeseries with different aggregation types (ALL, BY_COUNTRY, BY_CONTENT_TYPE) and date/country filters. Uses two ISRCs to cover all profile clusters (see TIKTOK_ISRC_HEADERS). Groups: 1. Default timeseries — ALL type (6 tests) 2. By country — BY_COUNTRY type with and without filter (12 tests) 3. By content type — BY_CONTENT_TYPE breakdown (6 tests) """ START_DATE = "2024-01-01" END_DATE = "2024-01-02" def _url(self, isrc, extra=""): base = config.SOUND_RECORDING_TIKTOK_TIMESERIES_URL.replace("", isrc) return ( f"{base}" f"?start_date={self.START_DATE}&end_date={self.END_DATE}{extra}" ) # ── Group 1: Default timeseries ─────────────────────────────────── # # 6 tests (one per isrc/header pair). @pytest.mark.parametrize("isrc,hdrs", TIKTOK_ISRC_HEADERS) def test_timeseries(self, isrc, hdrs): """Default TikTok timeseries returns items.""" assert_endpoint( self._url(isrc), headers=hdrs, min_items=1, ) # ── Group 2: By country ─────────────────────────────────────────── # # 6 isrc/header pairs × 2 methods = 12 tests. @pytest.mark.parametrize("isrc,hdrs", TIKTOK_ISRC_HEADERS) def test_by_country(self, isrc, hdrs): """BY_COUNTRY returns items with id (country code).""" payload = assert_endpoint( self._url(isrc, "&type=BY_COUNTRY"), headers=hdrs, min_items=1, ) assert "id" in payload["items"][0] @pytest.mark.parametrize("isrc,hdrs", TIKTOK_ISRC_HEADERS) def test_by_country_filtered(self, isrc, hdrs): """BY_COUNTRY + countries=US returns only US.""" payload = assert_endpoint( self._url(isrc, "&type=BY_COUNTRY&countries=US"), headers=hdrs, min_items=1, ) assert payload["items"][0]["id"] == "US" # ── Group 3: By content type ────────────────────────────────────── # # 6 tests (one per isrc/header pair). @pytest.mark.parametrize("isrc,hdrs", TIKTOK_ISRC_HEADERS) def test_by_content_type(self, isrc, hdrs): """BY_CONTENT_TYPE breaks down into PGC and UGC.""" payload = assert_endpoint( self._url(isrc, "&type=BY_CONTENT_TYPE"), headers=hdrs, min_items=2, ) assert sorted(["PGC", "UGC"]) == sorted( [payload["items"][0]["id"], payload["items"][1]["id"]] ) # =========================================================================== # Related videos by ISRC # =========================================================================== class TestRelatedVideosByIsrc: """Integration tests for /sound-recording//related-videos-by-isrc. Verifies related video IDs lookup for a sound recording. Groups: 1. Basic lookup — returns video_ids list (6 tests) """ ISRC = "QM6P41952433" @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_related_videos(self, hdrs): """Related videos returns a video_ids list.""" payload = assert_endpoint( config.SOUND_RECORDING_RELATED_VIDEOS_BY_ISRC_URL.replace( "", self.ISRC ), headers=hdrs, items_key=None, expect_payload_keys=["video_ids"], ) assert isinstance(payload["video_ids"], list) # =========================================================================== # Aggregate streams (POST) # =========================================================================== class TestAggregateStreams: """Integration tests for POST /sound-recording/aggregate-streams. Each variant hits a distinct SQL branch in the new bulk query class (country view switch, store intersect, distributor tightening). The endpoint has no date params — it always hits the all-time ROLLUP view. Groups: 1. Query variants — 6 variants × 6 headers = 36 tests 2. Empty ISRCs — empty list returns empty payload (6 tests) """ ISRC = "QM6P41952433" ISRC_2 = "DED831000251" QUERY_VARIANTS = [ pytest.param("", id="base_no_filters"), pytest.param("?country_code=US", id="country_filter"), pytest.param("?country_code=US&country_code=GB", id="multi_country_filter"), pytest.param("?store_ids=1&store_ids=286", id="store_filter"), pytest.param("?distributors=theorchard", id="distributors_filter"), pytest.param( "?country_code=US&store_ids=286&distributors=theorchard", id="all_filters", ), ] # ── Group 1: Query variants ──────────────────────────────────────── # # 6 headers × 6 variants = 36 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("query", QUERY_VARIANTS) def test_aggregate_streams(self, hdrs, query): """Aggregate all-time streams per ISRC returns a valid response. Each variant hits a distinct SQL branch (country view switch, store-id intersection, distributor IN tightening). """ payload = assert_endpoint( f"{BASE_URL}/sound-recording/aggregate-streams" + query, headers=hdrs, method="POST", json_body={"isrcs": [self.ISRC, self.ISRC_2]}, items_key=None, ) for isrc in (self.ISRC, self.ISRC_2): assert isrc in payload, f"Missing {isrc} in aggregate payload" assert "streams_all_time" in payload[isrc] assert "growth_percentage_7_days" in payload[isrc] # ── Group 2: Empty ISRCs ────────────────────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_empty_isrcs(self, hdrs): """Empty ISRC list returns empty payload.""" payload = assert_endpoint( f"{BASE_URL}/sound-recording/aggregate-streams", headers=hdrs, method="POST", json_body={"isrcs": []}, items_key=None, ) assert payload == {} # =========================================================================== # Streams bulk (POST) # =========================================================================== class TestStreamsBulk: """Integration tests for POST /sound-recording/streams. Each variant hits a distinct SQL branch in the new bulk query classes (daily view switch on country, HIGHWATERMARK − 28 days fallback when no dates, store-id intersection, distributor IN tightening). The endpoint runs a DAILY query (within start_date..end_date) and a ROLLUP query (all-time) in parallel. Groups: 1. Query variants — 7 variants × 6 headers = 42 tests 2. Empty ISRCs — empty list returns empty payload (6 tests) """ ISRC = "QM6P41952433" ISRC_2 = "DED831000251" QUERY_VARIANTS = [ pytest.param( "?start_date=2023-01-01&end_date=2023-01-07", id="base_date_range" ), pytest.param( "?start_date=2023-01-01&end_date=2023-01-07&country_code=US", id="country_filter", ), pytest.param( "?start_date=2023-01-01&end_date=2023-01-07" "&country_code=US&country_code=GB", id="multi_country_filter", ), pytest.param( "?start_date=2023-01-01&end_date=2023-01-07&store_ids=1&store_ids=286", id="store_filter", ), pytest.param( "?start_date=2023-01-01&end_date=2023-01-07&distributors=theorchard", id="distributors_filter", ), pytest.param("", id="default_date_range"), pytest.param("?days=0&start_date=HIGHWATERMARK", id="all_time_highwatermark"), ] # ── Group 1: Query variants ──────────────────────────────────────── # # 6 headers × 6 variants = 36 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("query", QUERY_VARIANTS) def test_streams_bulk(self, hdrs, query): """Bulk streams per ISRC returns a valid response. Each variant exercises a distinct SQL branch (country view switch, store-id intersection, distributor IN tightening, HIGHWATERMARK − 28 days fallback when no dates are provided). """ payload = assert_endpoint( f"{BASE_URL}/sound-recording/streams" + query, headers=hdrs, method="POST", json_body={"isrcs": [self.ISRC, self.ISRC_2]}, items_key=None, ) for isrc in (self.ISRC, self.ISRC_2): assert isrc in payload, f"Missing {isrc} in bulk payload" isrc_body = payload[isrc] assert isrc_body["isrc"] == isrc assert "aggregate" in isrc_body assert "stores" in isrc_body assert "sources" in isrc_body # ── Group 2: Empty ISRCs ────────────────────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_empty_isrcs(self, hdrs): """Empty ISRC list returns empty payload.""" payload = assert_endpoint( f"{BASE_URL}/sound-recording/streams" "?start_date=2023-01-01&end_date=2023-01-07", headers=hdrs, method="POST", json_body={"isrcs": []}, items_key=None, ) assert payload == {}