"""Integration tests for product endpoints. Endpoints tested: - /product//timeseries (TestProductTimeseries) - /product//summary (TestProductSummary) - /product/ (TestProduct) - /product//metadata (TestProductMetadata) - /product//aggregate-streams (TestProductAggregateStreams) - /product//aggregated-streams (TestProductAggregatedStreams) - /product-metrics (TestProductMetrics) - /product//metrics-by-track (TestProductMetricsByTrack) - /product/growth-periods-bulk (TestProductGrowthPeriodsBulk) Every test parametrizes on ALL_HEADERS (6 auth profiles) to verify access control. """ import os import pytest from analytics import config from analytics.constants.countries import ISO_ALPHA_2 from analytics.logic.product_timeseries import ( AGGREGATION_FIELDS_SUMMARY, PRODUCT_TABLES_TIMESERIES, ) from tests.integration.endpoints.conftest import ( ALL_HEADERS, PAGINATION, assert_endpoint, assert_timeseries_keys, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- 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"), ("", ""), ] def _assert_product_summary_keys(item, summary_type): """Assert expected keys on a product summary item. Product summary items include download fields not present in account summaries: downloads_track, downloads, and (for non-SOS, non-SOUND_RECORDING types) downloads_product. """ assert "streams" in item if summary_type in ("SOS", "SOS_DETAILED"): return assert "downloads_track" in item assert "downloads" in item if summary_type != "SOUND_RECORDING": assert "downloads_product" in item assert "saves" in item assert "skip_rate" in item class TestProductTimeseries: """Integration tests for /product//timeseries. Tests are organized into 7 groups by concern. Every group parametrizes on ALL_HEADERS (6 auth profiles). Most also parametrize on multi_product (false/true) to exercise both single-product and multi-product SQL paths. Groups: 1. All types shape — every query_type returns correct keys (144 tests) 2. Country filter — country-variant table switching (144 tests) 3. Data: country codes — known country canaries (24 tests) 4. Data: SOS breakdown — SOS category ordering (24 tests) 5. ID and store filters — ids= and store_ids= params (36 tests) 6. Downloads data — download-specific assertions (18 tests) 7. Alternate product — different product ID exercises edge cases (12 tests) """ PRODUCT_ID = 3859857 START_DATE = "2022-05-05" END_DATE = "2022-06-05" def _url(self, ts_type, extra=""): base = config.PRODUCT_TIME_SERIES_URL.replace( "", str(self.PRODUCT_ID) ) return ( f"{base}" f"?type={ts_type}" f"&start_date={self.START_DATE}&end_date={self.END_DATE}{extra}" ) # ── Group 1: All types return correct response shape ──────────── # # 6 headers x 12 types x 2 multi_product = 144 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("ts_type", PRODUCT_TABLES_TIMESERIES.keys()) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_timeseries(self, hdrs, ts_type, multi_product): """Every timeseries type returns data with correct keys. Exercises both single-product and multi-product SQL paths. Key shape follows assert_timeseries_keys rules (date, value, etc.). """ payload = assert_endpoint( self._url(ts_type, f"&multi_product={multi_product}"), headers=hdrs, ) if payload["items"]: assert_timeseries_keys(payload["items"][0].keys(), ts_type) # ── Group 2: Country filter table switching ───────────────────── # # 6 headers x 12 types x 2 multi_product = 144 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("ts_type", PRODUCT_TABLES_TIMESERIES.keys()) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_timeseries_country_filtered(self, hdrs, ts_type, multi_product): """Country filter activates the country-variant Snowflake table. When countries= is non-empty, the model picks the "country" table variant from PRODUCT_TABLES_TIMESERIES. """ payload = assert_endpoint( self._url( ts_type, f"&countries=GB&countries=DE&multi_product={multi_product}", ), headers=hdrs, ) if payload["items"]: assert_timeseries_keys(payload["items"][0].keys(), ts_type) # ── Group 3: Data assertions — country codes ──────────────────── # # 6 headers x 2 multi_product x 2 tests = 24 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_streams_country(self, hdrs, multi_product): """BY_COUNTRY includes GB, DE, and NO for product 3859857. Canary test: catches data pipeline regressions where country data disappears. """ payload = assert_endpoint( self._url( "PRODUCT_STREAMS_BY_COUNTRY", f"&multi_product={multi_product}", ), headers=hdrs, ) if payload["items"]: country_codes = [i["id"] for i in payload["items"]] assert "GB" in country_codes assert "DE" in country_codes assert "NO" in country_codes @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_streams_country_by_country(self, hdrs, multi_product): """countries=GB&countries=DE returns exactly those two countries. Verifies the country filter narrows results to the requested set. """ payload = assert_endpoint( self._url( "PRODUCT_STREAMS_BY_COUNTRY", f"&countries=GB&countries=DE&multi_product={multi_product}", ), headers=hdrs, ) if payload["items"]: assert_timeseries_keys( payload["items"][0].keys(), "PRODUCT_STREAMS_BY_COUNTRY" ) assert sorted(["GB", "DE"]) == sorted( [payload["items"][0]["id"], payload["items"][1]["id"]] ) # ── Group 4: Data assertions — SOS breakdown ─────────────────── # # 6 headers x 2 multi_product x 2 tests = 24 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_streams_by_sos(self, hdrs, multi_product): """BY_SOS breaks down into active, passive, collection, unknown. The _breakdown_by function produces exactly these four categories in this order. Verifies the breakdown logic works end-to-end. """ payload = assert_endpoint( self._url( "PRODUCT_STREAMS_BY_SOS", f"&multi_product={multi_product}", ), headers=hdrs, ) if payload["items"]: item_ids = list(dict.fromkeys([i["id"] for i in payload["items"]])) assert item_ids == ["active", "passive", "collection", "unknown"] @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_streams_by_sos_country_filtered(self, hdrs, multi_product): """BY_SOS + country filter still produces all four categories. Exercises the most complex code path: country table switch + Python-side _breakdown_by. Verifies they compose correctly. """ payload = assert_endpoint( self._url( "PRODUCT_STREAMS_BY_SOS", f"&countries=GB&countries=DE&multi_product={multi_product}", ), headers=hdrs, ) if payload["items"]: item_ids = list(dict.fromkeys([i["id"] for i in payload["items"]])) assert item_ids == ["active", "passive", "collection", "unknown"] # ── Group 5: ID and store filters ─────────────────────────────── # # 6 headers x 2 multi_product x 3 tests = 36 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_streams_by_store_store_filtered(self, hdrs, multi_product): """store_ids=286 filters BY_STORE to a single store. Verifies the WHERE store_id IN (...) SQL clause works. """ payload = assert_endpoint( self._url( "PRODUCT_STREAMS_BY_STORE", f"&store_ids=286&multi_product={multi_product}", ), headers=hdrs, ) if payload["items"]: assert payload["items"][0]["id"] == 286 @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_streams_by_track_isrc_filtered(self, hdrs, multi_product): """ids=QM6MZ2214883 filters BY_TRACK to a single ISRC. BY_TRACK uses SQL-side WHERE clause for id filtering. All returned items must have that ISRC. """ payload = assert_endpoint( self._url( "PRODUCT_STREAMS_BY_TRACK", f"&ids=QM6MZ2214883&multi_product={multi_product}", ), headers=hdrs, ) if payload["items"]: assert set(i["id"] for i in payload["items"]) == {"QM6MZ2214883"} @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_streams_by_store_country_and_store_filtered(self, hdrs, multi_product): """Combined country + store filter narrows BY_STORE results. Exercises both the country table switch and the store_ids SQL clause simultaneously. """ payload = assert_endpoint( self._url( "PRODUCT_STREAMS_BY_STORE", f"&countries=GB&countries=DE&store_ids=286" f"&multi_product={multi_product}", ), headers=hdrs, ) if payload["items"]: assert set(i["id"] for i in payload["items"]) == {286} # ── Group 6: Downloads data assertions ────────────────────────── # # 6 headers x 3 tests = 18 tests. # Download types are not parametrized with multi_product. @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_downloads_country(self, hdrs): """PRODUCT_DOWNLOADS_BY_COUNTRY includes US for product 3859857. Canary test for download country data pipeline. """ payload = assert_endpoint( self._url("PRODUCT_DOWNLOADS_BY_COUNTRY"), headers=hdrs, ) if payload["items"]: country_codes = [i["id"] for i in payload["items"]] assert "US" in country_codes @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_downloads_country_by_country(self, hdrs): """countries=GB&countries=DE&countries=US returns those three countries. Verifies the country filter on download data. """ payload = assert_endpoint( self._url( "PRODUCT_DOWNLOADS_BY_COUNTRY", "&countries=GB&countries=DE&countries=US", ), headers=hdrs, ) if payload["items"]: country_codes = [i["id"] for i in payload["items"]] assert "US" in country_codes assert "DE" in country_codes assert "GB" in country_codes @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_track_downloads_by_store_store_filtered(self, hdrs): """store_ids=1 filters TRACK_DOWNLOADS_BY_STORE to store 1. Verifies the WHERE store_id IN (...) SQL clause for track downloads. """ payload = assert_endpoint( self._url("TRACK_DOWNLOADS_BY_STORE", "&store_ids=1"), headers=hdrs, ) if payload["items"]: assert payload["items"][0]["id"] == 1 # ── Group 7: Alternate product ID ─────────────────────────────── # # 6 headers x 2 tests = 12 tests. # Uses product 3636810 to verify behavior with a different product. @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_product_streams_sos(self, hdrs): """BY_SOS with ids=unknown on product 3636810 returns data. Exercises id filtering for SOS categories on a different product to catch product-specific edge cases. """ base = config.PRODUCT_TIME_SERIES_URL.replace("", "3636810") url = ( f"{base}" f"?type=PRODUCT_STREAMS_BY_SOS&ids=unknown" f"&start_date=2022-07-26&end_date=2022-08-22" ) assert_endpoint(url, headers=hdrs) @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_product_track_streams_isrc_filtered(self, hdrs): """BY_TRACK with ids=ARF412000306 on product 3636810 returns data. Exercises ISRC id filtering on a different product. """ base = config.PRODUCT_TIME_SERIES_URL.replace("", "3636810") url = ( f"{base}" f"?type=PRODUCT_STREAMS_BY_TRACK&ids=ARF412000306" f"&start_date=2022-07-26&end_date=2022-08-22" ) assert_endpoint(url, headers=hdrs) # ── Group 8: SOS Detailed with filter combinations ────────────── # # 6 headers × 11 SOS_FILTER_COMBOS × 2 multi_product = 132 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("store_ids,sos", SOS_FILTER_COMBOS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_streams_by_sos_detailed(self, hdrs, store_ids, sos, multi_product): """SOS_DETAILED with store_ids + stream_sources filter combos. Exercises the detailed SOS query path across all filter combinations. """ payload = assert_endpoint( self._url( "PRODUCT_STREAMS_BY_SOS_DETAILED", f"&multi_product={multi_product}{store_ids}{sos}", ), headers=hdrs, ) if payload["items"]: assert_timeseries_keys( payload["items"][0].keys(), "PRODUCT_STREAMS_BY_SOS_DETAILED" ) class TestProductSummary: """Integration tests for /product//summary. Tests cover all summary types across multi_product variants, country filtering, and specific data assertions. Groups: 1. All types shape — every summary_type returns correct keys (60 tests) 2. Country filter — country-filtered results (60 tests) 3. Data: country codes — known country canaries (24 tests) 4. Track/store filters — specific filter combinations (48 tests) 5. Alternate product — different product ID (6 tests) 6. SOS Detailed — SOS_DETAILED with filter combos (132 tests) """ PRODUCT_ID = 3859857 START_DATE = "2022-07-26" END_DATE = "2022-07-27" def _url(self, summary_type, extra=""): base = config.PRODUCT_SUMMARY_URL.replace("", str(self.PRODUCT_ID)) return ( f"{base}" f"?type={summary_type}" f"&start_date={self.START_DATE}&end_date={self.END_DATE}" f"&{PAGINATION}{extra}" ) # ── Group 1: All summary types return correct shape ───────────── # # 6 headers x 5 summary_types x 2 multi_product = 60 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("summary_type", AGGREGATION_FIELDS_SUMMARY.keys()) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_summary(self, hdrs, summary_type, multi_product): """Every summary type returns items with correct keys. Non-SOS types include downloads_track, downloads, saves, skip_rate. Non-SOS/non-SOUND_RECORDING also include downloads_product. """ payload = assert_endpoint( self._url(summary_type, f"&multi_product={multi_product}"), headers=hdrs, ) if payload["items"]: _assert_product_summary_keys(payload["items"][0], summary_type) # ── Group 2: Country-filtered summary ─────────────────────────── # # 6 headers x 5 summary_types x 2 multi_product = 60 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("summary_type", AGGREGATION_FIELDS_SUMMARY.keys()) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_summary_country_filtered(self, hdrs, summary_type, multi_product): """Country filter narrows results; items still contain correct keys.""" payload = assert_endpoint( self._url( summary_type, f"&countries=GB&countries=DE&multi_product={multi_product}", ), headers=hdrs, ) if payload["items"]: _assert_product_summary_keys(payload["items"][0], summary_type) # ── Group 3: Data assertions — country codes ──────────────────── # # 6 headers x 2 multi_product x 2 tests = 24 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_streams_country_summary(self, hdrs, multi_product): """COUNTRY summary includes US, GB, and DE for product 3859857. Canary test for country-level summary data pipeline. """ payload = assert_endpoint( self._url("COUNTRY", f"&multi_product={multi_product}"), headers=hdrs, ) if payload["items"]: country_codes = [i["id"] for i in payload["items"]] assert "US" in country_codes assert "GB" in country_codes assert "DE" in country_codes @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_streams_country_by_country_summary(self, hdrs, multi_product): """countries=GB&countries=DE returns exactly those two countries. Verifies the country filter narrows summary results. """ payload = assert_endpoint( self._url( "COUNTRY", f"&countries=GB&countries=DE&multi_product={multi_product}", ), headers=hdrs, ) if payload["items"]: assert sorted(["GB", "DE"]) == sorted( [payload["items"][0]["id"], payload["items"][1]["id"]] ) # ── Group 4: Track and store filters ──────────────────────────── # # 6 headers x 2 multi_product x 4 tests = 48 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_streams_by_track_summary(self, hdrs, multi_product): """SOUND_RECORDING summary includes isrc field. Verifies the ISRC-based aggregation path returns track identifiers and correct download/stream fields. """ payload = assert_endpoint( self._url("SOUND_RECORDING", f"&multi_product={multi_product}"), headers=hdrs, ) if payload["items"]: _assert_product_summary_keys(payload["items"][0], "SOUND_RECORDING") assert "isrc" in payload["items"][0] @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_streams_by_store_store_filtered_summary(self, hdrs, multi_product): """store_ids=286 filters STORE summary to a single store. Verifies the WHERE store_id IN (...) SQL clause in summary. """ payload = assert_endpoint( self._url("STORE", f"&store_ids=286&multi_product={multi_product}"), headers=hdrs, ) if payload["items"]: assert all(i["id"] == 286 for i in payload["items"]) @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_summary_store_with_store_ids(self, hdrs, multi_product): """STORE summary with multiple store_ids returns data without error. Verifies multiple store_ids compose in the SQL IN clause. """ assert_endpoint( self._url( "STORE", f"&store_ids=286&store_ids=1&multi_product={multi_product}", ), headers=hdrs, ) @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_summary_sound_recording_with_countries(self, hdrs, multi_product): """SOUND_RECORDING summary with country filter returns data with isrc. Exercises the country-filtered SQL path for track-level summary. """ payload = assert_endpoint( self._url( "SOUND_RECORDING", f"&countries=GB&countries=DE&multi_product={multi_product}", ), headers=hdrs, ) if payload["items"]: _assert_product_summary_keys(payload["items"][0], "SOUND_RECORDING") assert "isrc" in payload["items"][0] # ── Group 5: Alternate product ID ─────────────────────────────── # # 6 tests. # Uses product 3636810 to verify behavior with a different product. @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_streams_summary_sr(self, hdrs): """SOUND_RECORDING summary on product 3636810 returns data. Exercises summary with a different product to catch edge cases. """ base = config.PRODUCT_SUMMARY_URL.replace("", "3636810") url = ( f"{base}" f"?type=SOUND_RECORDING&start_date=2022-07-26" f"&end_date=2022-07-27&{PAGINATION}" ) assert_endpoint(url, headers=hdrs) # ── Group 6: SOS Detailed with filter combinations ────────────── # # 6 headers × 11 SOS_FILTER_COMBOS × 2 multi_product = 132 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("store_ids,sos", SOS_FILTER_COMBOS) @pytest.mark.parametrize( "multi_product", [pytest.param("false", id="single"), pytest.param("true", id="multi")], ) def test_summary_sos_detailed(self, hdrs, store_ids, sos, multi_product): """SOS_DETAILED with store_ids + stream_sources filter combos. Exercises the detailed SOS summary path across all filter combinations. """ payload = assert_endpoint( self._url( "SOS_DETAILED", f"&multi_product={multi_product}{store_ids}{sos}", ), headers=hdrs, ) if payload["items"]: _assert_product_summary_keys(payload["items"][0], "SOS_DETAILED") class TestProduct: """Integration tests for /product/. Verifies the product detail endpoint returns all expected fields. The endpoint does a parallel fan-out: - product_tracks.sql (recent track streams) - product_streams_all_time.sql (rollup all-time streams + 7d growth) - add_outage_error_to_stores (Python-side store outage info) Groups: 1. Product detail — field validation (6 tests) 2. Distributors filter — alternate distributors IN clause (6 tests) 3. Unknown product — empty-response path (6 tests) """ PRODUCT_ID = 3859857 EXPECTED_FIELDS = [ "product_id", "streams", "tracks", ] # ── Group 1: Product detail field validation ──────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_product(self, hdrs): """Product detail returns all expected fields and correct product_id. Validates 3 response fields: product_id, streams, tracks. Metadata fields (upc, artist_name, etc.) are returned by the separate /product//metadata endpoint, not this one. """ payload = assert_endpoint( config.PRODUCT_URL.replace("", str(self.PRODUCT_ID)), headers=hdrs, items_key=None, ) assert payload["product_id"] == str(self.PRODUCT_ID) for field in self.EXPECTED_FIELDS: assert field in payload, f"Missing field: {field}" # ── Group 2: Distributors filter ──────────────────────────────── # # 6 tests. Tightens the `distributor IN (...)` clause to a single # value to confirm the new template still keeps the # not_for_distribution = 'SwitchboardDummy' OR fallback. @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_product_distributors_filter(self, hdrs): """`distributors=theorchard` keeps the SQL valid and returns the same shape.""" base = config.PRODUCT_URL.replace("", str(self.PRODUCT_ID)) payload = assert_endpoint( f"{base}?distributors=theorchard", headers=hdrs, items_key=None, ) assert payload["product_id"] == str(self.PRODUCT_ID) for field in self.EXPECTED_FIELDS: assert field in payload, f"Missing field: {field}" # ── Group 3: Unknown product → empty payload ──────────────────── # # 6 tests. product_id=0 has no rows in either parallel call; logic # short-circuits and returns the empty product body with growth_percentage # and all_time set to None. @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_unknown_product(self, hdrs): """Non-existent product returns the empty body with the same keys.""" payload = assert_endpoint( config.PRODUCT_URL.replace("", "0"), headers=hdrs, items_key=None, ) assert payload["product_id"] == "0" for field in self.EXPECTED_FIELDS: assert field in payload, f"Missing field: {field}" assert payload["tracks"] == [] @pytest.mark.skip(reason="Brittle, need AWS creds in environment.") class TestProductMetadata: """Integration tests for /product//metadata. Verifies product metadata fields across auth profiles, including the use-artist-profiles query parameter. Groups: 1. Metadata fields — standard metadata response (6 tests) 2. Artist profile mode — use-artist-profiles=true (6 tests) """ METADATA_FIELDS = [ "product_id", "upc", "artist_name", "product_name", "format", "release_date", "sales_start_date", ] # ── Group 1: Standard metadata field validation ───────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_metadata(self, hdrs): """Product metadata returns all expected fields. Validates 7 response fields: product_id, upc, artist_name, product_name, format, release_date, sales_start_date. """ url = config.PRODUCT_METADATA_URL.replace("", "240038") payload = assert_endpoint( url, headers=hdrs, items_key=None, ) for field in self.METADATA_FIELDS: assert field in payload, f"Missing field: {field}" # ── Group 2: Artist profile mode ──────────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_metadata_artist_profiles(self, hdrs): """Product metadata with use-artist-profiles=true returns all fields. Exercises the artist-profile code path that resolves product ownership via global_participant_id instead of account. """ url = ( config.PRODUCT_METADATA_URL.replace("", "2665644") + "?use-artist-profiles=true" ) payload = assert_endpoint( url, headers=hdrs, items_key=None, ) for field in self.METADATA_FIELDS: assert field in payload, f"Missing field: {field}" class TestProductAggregateStreams: """Integration tests for /product//aggregate-streams. Verifies aggregate stream metrics (all-time, growth) for a product. Each variant exercises a distinct SQL branch: | variant | multi_product | country_ids | replaces SQL file | |--------------------|---------------|-------------|-------------------| | defaults | F | none | product_aggregate_streams.sql | | country_filter | F | yes | product_aggregate_streams_by_country.sql | | multi_product | T | none | multi_product_aggregate_streams.sql | | country+multi | T | yes | multi_product_aggregate_streams_by_country.sql | | distributors | F | none | product_aggregate_streams.sql + tighter IN (...) | Groups: 1. SQL-branch matrix — 5 variants × 6 headers = 30 tests 2. Unknown product — empty-payload defaults (6 tests) """ PRODUCT_ID = 3859857 EXPECTED_FIELDS = ["product_id", "streams_all_time", "growth_percentage_7_days"] QUERY_VARIANTS = [ pytest.param("", id="defaults"), pytest.param("?country_code=US", id="country_filter"), pytest.param("?multi_product=true", id="multi_product"), pytest.param("?multi_product=true&country_code=US", id="country+multi"), pytest.param("?distributors=theorchard", id="distributors"), ] # ── Group 1: SQL-branch matrix ────────────────────────────────── # # 6 headers × 5 variants = 30 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("variant", QUERY_VARIANTS) def test_aggregate_streams(self, hdrs, variant): """Aggregate streams returns product_id, streams_all_time, growth_percentage_7_days. Each variant exercises one of the four SQL files (product_aggregate_streams, _by_country, multi_product_aggregate_streams, _by_country) plus the distributor-tightening branch. """ base = config.PRODUCT_AGGREGATE_STREAMS_URL.replace( "", str(self.PRODUCT_ID) ) payload = assert_endpoint( f"{base}{variant}", headers=hdrs, items_key=None, ) for field in self.EXPECTED_FIELDS: assert field in payload, f"Missing field: {field}" assert payload["product_id"] == str(self.PRODUCT_ID) # ── Group 2: Unknown product → empty payload ──────────────────── # # 6 tests. Verifies the no-rows-found short-circuit in the logic # function: streams_all_time = 0, growth_percentage_7_days = None. @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_unknown_product(self, hdrs): """Non-existent product returns zero/None defaults.""" payload = assert_endpoint( config.PRODUCT_AGGREGATE_STREAMS_URL.replace("", "0"), headers=hdrs, items_key=None, ) for field in self.EXPECTED_FIELDS: assert field in payload, f"Missing field: {field}" assert payload["product_id"] == "0" assert payload["streams_all_time"] == 0 assert payload["growth_percentage_7_days"] is None class TestProductMetrics: """Integration tests for /product-metrics. Verifies product metrics listing with various filter combinations: countries, participant IDs, multi_product, company_brand, parent_company, pagination, and ordering. Groups: 1. All SQL-path combos — country x participant x multi_product (48 tests) 2. Other filter axes — company_brand, parent_company (12 tests) 3. Pagination / order — limit+offset, order_by, release_date sort (18 tests) """ GP_IDS = "global_participant_ids=f67b0892-f0bc-4575-b23e-59566ccb44bc" COUNTRY = "country_code=ES" # ── Group 1–3: All variants ───────────────────────────────────── # # 6 headers x 13 param combos = 78 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "request_params", [ # --- 8 SQL-path combos (countries x participant x multi_product) --- pytest.param("", id="defaults"), pytest.param(f"?{COUNTRY}", id="country"), pytest.param(f"?{GP_IDS}", id="gp_ids"), pytest.param(f"?{COUNTRY}&{GP_IDS}", id="country+gp_ids"), pytest.param("?multi_product=true", id="multi_product"), pytest.param(f"?{COUNTRY}&multi_product=true", id="country+multi"), pytest.param(f"?{GP_IDS}&multi_product=true", id="gp_ids+multi"), pytest.param( f"?{COUNTRY}&{GP_IDS}&multi_product=true", id="country+gp_ids+multi", ), # --- other filter axes --- pytest.param( "?company_brand=d25a4cd1-e820-45f2-be5c-56edcfeb8298", id="company_brand", ), pytest.param( f"?parent_company=955a1bbd-b623-4ea1-ab5f-8d6620c442fb&{GP_IDS}", id="parent_company+gp_ids", ), # --- pagination / ordering --- pytest.param("?limit=5&offset=10", id="limit_offset"), pytest.param( "?order_by=streams_28_days&order_dir=ASC", id="order_by_28d_asc", ), pytest.param( "?order_by=release_date&order_dir=desc", id="order_by_release_date", ), ], ) def test_product_metrics(self, hdrs, request_params): """Product metrics returns metrics list and total_products count. Each param combo exercises a different SQL path: country filtering, participant filtering, multi-product mode, org hierarchy filters, pagination, or ORDER BY column variants. """ payload = assert_endpoint( f"{config.PRODUCT_METRICS_URL}{request_params}", headers=hdrs, items_key=None, ) assert "metrics" in payload assert "total_products" in payload class TestProductMetricsByTrack: """Integration tests for /product//metrics-by-track. Verifies per-track metrics for a product. Each variant exercises a distinct SQL branch: | variant | country_ids | replaces SQL file | |--------------------|-------------|-------------------| | defaults | none | product_metrics_by_track.sql | | country_filter | yes | product_metrics_by_track_by_country.sql | | store_filter | none | product_metrics_by_track.sql + intersect(store_ids) | | country+store | yes | product_metrics_by_track_by_country.sql + intersect(store_ids) | | distributors | none | tighter distributor IN (...) | | order_by_28d | none | ORDER BY streams_28_days ASC | Groups: 1. SQL-branch matrix — 6 variants × 6 headers = 36 tests 2. Empty response — non-existent product returns empty tracks (6 tests) 3. Field validation — known product returns expected fields (6 tests) """ PRODUCT_ID = 3153050 EXPECTED_FIELDS = ["product_id", "tracks"] EXPECTED_TRACK_FIELDS = [ "track_id", "tuid", "streams_1_day", "growth_percentage_1_day", "streams_7_days", "growth_percentage_7_days", "streams_28_days", "growth_percentage_28_days", "streams_all_time", ] QUERY_VARIANTS = [ pytest.param("", id="defaults"), pytest.param("?country_code=US", id="country_filter"), pytest.param("?store_ids=1&store_ids=286", id="store_filter"), pytest.param("?country_code=US&store_ids=1", id="country+store"), pytest.param("?distributors=theorchard", id="distributors"), pytest.param("?order_by=streams_28_days&order_dir=ASC", id="order_by_28d"), ] # ── Group 1: SQL-branch matrix ────────────────────────────────── # # 6 headers × 6 variants = 36 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("variant", QUERY_VARIANTS) def test_metrics_by_track(self, hdrs, variant): """Each variant exercises a distinct SQL branch. Validates payload shape; tracks may be empty for restricted profiles but the response shape must match. """ base = config.PRODUCT_METRICS_BY_TRACK_URL.replace( "", str(self.PRODUCT_ID) ) payload = assert_endpoint( f"{base}{variant}", headers=hdrs, items_key=None, ) for field in self.EXPECTED_FIELDS: assert field in payload, f"Missing field: {field}" assert payload["product_id"] == str(self.PRODUCT_ID) if payload["tracks"]: for field in self.EXPECTED_TRACK_FIELDS: assert field in payload["tracks"][0] # ── Group 2: Empty response ───────────────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_empty_response(self, hdrs): """Non-existent product (id=0) returns empty tracks list. Verifies graceful handling rather than 404/500. """ payload = assert_endpoint( config.PRODUCT_METRICS_BY_TRACK_URL.replace("", "0"), headers=hdrs, items_key=None, ) assert payload["tracks"] == [] # ── Group 3: Field validation ─────────────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_fields(self, hdrs): """Product 3153050 returns all expected fields including tracks list. Validates 2 response fields: product_id, tracks. Metadata fields (upc, artist_name, etc.) belong to /metrics-by-track, not this endpoint. """ payload = assert_endpoint( config.PRODUCT_METRICS_BY_TRACK_URL.replace( "", str(self.PRODUCT_ID) ), headers=hdrs, items_key=None, ) for field in self.EXPECTED_FIELDS: assert field in payload, f"Missing field: {field}" class TestProductAggregatedStreams: """Integration tests for /product//aggregated-streams. Verifies aggregated streams with different dimensions (store, country, sos) and date range variants. Groups: 1. All dimensions — store, country, sos (18 tests) 2. Days back — shorter date range via days_back param (6 tests) """ PRODUCT_ID = 2799900 AGGREGATED_FIELDS = [ "all_other_rollup", "all_other_timeseries", "topn_timeseries", "topn_rollup", "total", ] def _url(self, dimension, extra=""): base = config.PRODUCT_AGGREGATED_STREAMS_URL.replace( "", str(self.PRODUCT_ID) ) return f"{base}?dimension={dimension}{extra}" # ── Group 1: All dimensions ──────────────────────────────────── # # 6 headers × 3 dimensions = 18 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("dimension", ["store", "sos", "country"]) def test_dimension(self, hdrs, dimension): """Each dimension returns all expected aggregated-streams fields. Default range is 28 days. Validates response shape and timeseries length for store, country, and sos dimensions. """ payload = assert_endpoint( self._url(dimension), headers=hdrs, items_key=None, ) for field in self.AGGREGATED_FIELDS: assert field in payload # ── Group 2: Days back ───────────────────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_days_back(self, hdrs): """days_back=7 returns 7-item timeseries instead of default 28. Exercises the custom date range code path in the model. """ payload = assert_endpoint( self._url("country", "&days_back=7"), headers=hdrs, items_key=None, ) for field in self.AGGREGATED_FIELDS: assert field in payload if payload.get("all_other_timeseries"): assert len(payload["all_other_timeseries"]) == 7 class TestProductGrowthPeriodsBulk: """Integration tests for /product/growth-periods-bulk. Verifies bulk growth period retrieval for multiple products. Groups: 1. Single product — one product_id (6 tests) 2. Multiple products — two product_ids (6 tests) 3. With filters — country and store filters (6 tests) """ PRODUCT_ID = "3859857" PRODUCT_ID_2 = "2799900" # ── Group 1: Single product ─────────────────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_single_product(self, hdrs): """Growth periods for a single product returns a valid response.""" assert_endpoint( config.PRODUCT_BULK_GROWTH_PERIODS_URL + f"?product_id={self.PRODUCT_ID}", headers=hdrs, items_key=None, ) # ── Group 2: Multiple products ──────────────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_multiple_products(self, hdrs): """Growth periods for multiple products returns a valid response.""" assert_endpoint( config.PRODUCT_BULK_GROWTH_PERIODS_URL + f"?product_id={self.PRODUCT_ID}&product_id={self.PRODUCT_ID_2}", headers=hdrs, items_key=None, ) # ── Group 3: With filters ───────────────────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_with_country_filter(self, hdrs): """Growth periods with country filter returns a valid response.""" assert_endpoint( config.PRODUCT_BULK_GROWTH_PERIODS_URL + f"?product_id={self.PRODUCT_ID}&countries=US&countries=GB", headers=hdrs, items_key=None, ) class TestProductMetricsCountryCsv: """Integration tests for comma-separated country_code on GET /product-metrics. The handler accepts the country filter as a comma-separated value (``?country_code=US,GB``) as well as the legacy repeated form. The comma form keeps the URL short enough to clear the edge query-string size limit that drops the repeated form (502) when many countries are sent. Groups: 1. CSV/repeated parity — the two forms yield identical payloads (6) 2. Large country list — the full ISO list as CSV still 200s (6) """ COUNTRIES = ["ES", "US", "GB"] # ── Group 1: CSV/repeated parity ────────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_csv_matches_repeated(self, hdrs): """Comma-separated and repeated country_code return the same payload.""" repeated_qs = "?" + "&".join(f"country_code={c}" for c in self.COUNTRIES) repeated = assert_endpoint( f"{config.PRODUCT_METRICS_URL}{repeated_qs}", headers=hdrs, items_key=None, ) csv = assert_endpoint( f"{config.PRODUCT_METRICS_URL}?country_code={','.join(self.COUNTRIES)}", headers=hdrs, items_key=None, ) assert csv == repeated # ── Group 2: Large country list ─────────────────────────────── # # 6 tests (one per header). @pytest.mark.skip( reason="Flaky: backend 504s computing the 249-country aggregation. " "CSV parsing is covered deterministically by " "tests/unit/test_metrics_country_csv.py (GO-4832)." ) @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_large_country_list_csv(self, hdrs): """The full ISO country list as a single comma-separated value 200s.""" payload = assert_endpoint( f"{config.PRODUCT_METRICS_URL}?country_code={','.join(ISO_ALPHA_2)}" "&limit=5&offset=0", headers=hdrs, items_key=None, ) assert "metrics" in payload assert "total_products" in payload