"""Integration tests for participant endpoints. Endpoints tested: - /participant//summary (TestParticipantSummary) - /participant//timeseries (TestParticipantTimeseries) - /participant//track-streams-all (TestParticipantTrackStreams) - /participant//track-streams-by-store (TestParticipantTrackStreams) - /participant//aggregated-streams (TestParticipantAggregatedStreams) - /participant//demographics (TestParticipantDemographics) - /participant/metrics (TestParticipantMetrics) Every test parametrizes on ALL_HEADERS (6 auth profiles) to verify access control. """ import pytest from analytics import config from analytics.config import BASE_URL from analytics.constants.countries import ISO_ALPHA_2 from tests.integration.endpoints.conftest import ( ALL_HEADERS, INSIGHTS_EMPLOYEE, PAGINATION, assert_endpoint, ) # --------------------------------------------------------------------------- # Participant-specific constants # --------------------------------------------------------------------------- PARTICIPANT_METRICS_FIELDS = [ "id", "streams_1_day", "growth_percentage_1_day", "streams_7_days", "growth_percentage_7_days", "streams_28_days", "growth_percentage_28_days", "streams_all_time", ] TIMESERIES_STREAMS_TYPES = [ "TRACK_STREAMS", "TRACK_STREAMS_BY_COUNTRY", "TRACK_STREAMS_BY_SOS", "TRACK_STREAMS_BY_SOUND_RECORDING", "TRACK_STREAMS_BY_STORE", "TRACK_STREAMS_BY_PRODUCT", ] TIMESERIES_DOWNLOADS_TYPES = [ "TRACK_DOWNLOADS", "TRACK_DOWNLOADS_BY_COUNTRY", "TRACK_DOWNLOADS_BY_SOUND_RECORDING", "TRACK_DOWNLOADS_BY_STORE", "TRACK_DOWNLOADS_BY_PRODUCT", "ALBUM_DOWNLOADS", "ALBUM_DOWNLOADS_BY_COUNTRY", "ALBUM_DOWNLOADS_BY_STORE", "ALBUM_DOWNLOADS_BY_PRODUCT", ] # Store IDs supported for SOS_DETAILED (YouTube excluded) SOS_DETAILED_STORE_IDS = { "spotify": "286", "apple": "1", "amazon": "187", } # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _assert_participant_timeseries_keys(keys, ts_type): """Assert expected keys on a participant timeseries item. Unlike account timeseries, participant timeseries always includes 'id' and streaming types include 'skips' alongside 'skip_rate'. """ assert "id" in keys assert "date" in keys assert "value" in keys if ts_type == "TRACK_STREAMS_BY_SOS" or ts_type in TIMESERIES_DOWNLOADS_TYPES: return assert "saves" in keys assert "skips" in keys assert "skip_rate" in keys class TestParticipantSummary: """Integration tests for /participant//summary. Groups: 1. All summary types — every type returns correct shape (108 tests) 2. Deletion filtering — deleted ISRCs are excluded (6 tests) 3. Country filter — country param narrows results (108 tests) 4. SOS_DETAILED — store + stream-source variants (30 tests) 5. Store filter — stores param narrows results (36 tests) 6. All-time — no date params at all (36 tests) """ GP_ID = "e49ea9f9-0a74-4623-b3c6-ff819c7c0a4d" GP_IDS = [ GP_ID, "e90163a0-a3a5-433b-b768-aded5b7f18cd", "b9897857-0e83-4f23-a771-28c03acd42e5", ] START_DATE = "2022-07-26" END_DATE = "2022-07-27" SUMMARY_TYPES = ["SOUND_RECORDING", "PRODUCT", "COUNTRY", "SOS", "STORE", "TOTAL"] SUMMARY_FIELDS = [ "streams", "streams_start_date", "streams_end_date", "downloads_start_date", "downloads_end_date", "downloads", "saves", "skips", "skip_rate", ] def _url(self, gp_id, summary_type, extra=""): return ( f"{BASE_URL}/participant/{gp_id}/summary" f"?start_date={self.START_DATE}&end_date={self.END_DATE}" f"&type={summary_type}&{PAGINATION}{extra}" ) # ── Group 1: All summary types return correct shape ──────────── # # 6 headers × 3 gp_ids × 6 summary_types = 108 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("gp_id", GP_IDS) @pytest.mark.parametrize("summary_type", SUMMARY_TYPES) def test_summary(self, hdrs, gp_id, summary_type): """Every summary type returns items with all expected fields. Validates 9 response fields: streams, streams_start_date, streams_end_date, downloads_start_date, downloads_end_date, downloads, saves, skips, skip_rate. """ payload = assert_endpoint( self._url(gp_id, summary_type), headers=hdrs, ) if payload["items"]: item = payload["items"][0] for field in self.SUMMARY_FIELDS: assert field in item # ── Group 2: Deletion filtering ──────────────────────────────── # # 6 tests (one per header). # Verifies that deleted ISRCs are excluded from SOUND_RECORDING results. @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_filter_deletions(self, hdrs): """Deleted ISRC QZLGF2000009 must not appear in SOUND_RECORDING results. Uses a different gp_id (70525648...) and date range where the deletion is known to exist. """ payload = assert_endpoint( f"{BASE_URL}/participant/70525648-b8ab-490f-a654-e56d6d7e7f42/summary" f"?start_date=2024-07-23&end_date=2024-07-25" f"&type=SOUND_RECORDING&{PAGINATION}", headers=hdrs, ) if payload["items"]: assert "QZLGF2000009" not in [i["id"] for i in payload["items"]] # ── Group 3: Country-filtered summary ────────────────────────── # # 6 headers × 3 gp_ids × 6 summary_types = 108 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("gp_id", GP_IDS) @pytest.mark.parametrize("summary_type", SUMMARY_TYPES) def test_summary_country_filtered(self, hdrs, gp_id, summary_type): """Country filter narrows results; items still contain all fields.""" payload = assert_endpoint( self._url(gp_id, summary_type, "&countries=GB&countries=DE"), headers=hdrs, ) if payload["items"]: item = payload["items"][0] for field in self.SUMMARY_FIELDS: assert field in item # ── Group 4: SOS_DETAILED variants ───────────────────────────── # # 6 headers × 5 param_variants = 30 tests. # Covers: no store filter, per-store, and stream_source filtering. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "extra", [ pytest.param("", id="no_store_filter"), pytest.param( f"&stores={SOS_DETAILED_STORE_IDS['spotify']}", id="spotify_only", ), pytest.param( f"&stores={SOS_DETAILED_STORE_IDS['apple']}", id="apple_only", ), pytest.param( f"&stores={SOS_DETAILED_STORE_IDS['spotify']}" "&stream_sources=collection&stream_sources=search", id="spotify_with_stream_sources", ), pytest.param( f"&stores={SOS_DETAILED_STORE_IDS['apple']}" "&stream_sources=discovery", id="apple_with_stream_sources", ), ], ) def test_summary_sos_detailed(self, hdrs, extra): """SOS_DETAILED returns items with id and streams fields. Each item's id should be a known SOS detailed column name (e.g. streams_sos_spotify_collection). Validates the UNPIVOT output shape is consistent with the rest of the summary response. """ payload = assert_endpoint( self._url(self.GP_ID, "SOS_DETAILED", extra), headers=hdrs, ) if payload["items"]: item = payload["items"][0] assert "id" in item assert "streams" in item assert item["id"].startswith("streams_sos_") # ── Group 5: Store-filtered summary ──────────────────────────── # # 6 headers × 6 summary_types = 36 tests. # Exercises the `stores` query param across all types. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("summary_type", SUMMARY_TYPES) def test_summary_store_filtered(self, hdrs, summary_type): """Store filter narrows results; items still contain all fields.""" payload = assert_endpoint( self._url(self.GP_ID, summary_type, "&stores=286"), headers=hdrs, ) if payload["items"]: item = payload["items"][0] for field in self.SUMMARY_FIELDS: assert field in item # ── Group 6: All-time summary (no date params) ───────────────── # # 6 headers × 6 summary_types = 36 tests. # Omitting both start_date and end_date triggers the no-date-filter # branch in the SQL template (full participant history). @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("summary_type", SUMMARY_TYPES) def test_summary_all_time(self, hdrs, summary_type): """No start_date/end_date params returns full-history aggregation. Uses a smaller participant (Local Natives) so the all-time scan completes within QA's gateway window. """ payload = assert_endpoint( f"{BASE_URL}/participant/95ee273c-09cb-411c-ae00-8806e9ca938e/summary" f"?type={summary_type}&{PAGINATION}", headers=hdrs, ) if payload["items"]: item = payload["items"][0] for field in self.SUMMARY_FIELDS: assert field in item class TestParticipantTimeseries: """Integration tests for /participant//timeseries. Tests cover all streaming and download types across multiple date param styles and country filtering. Groups: 1. Streams with date variants — start/end and days_back (72 tests) 2. Streams country-filtered — countries param (36 tests) 3. Downloads — all download types (54 tests) 4. Downloads country-filtered — countries param (54 tests) 5. SOS_DETAILED variants — store + stream-source (30 tests) 6. Store-filtered (streams) — stores param (36 tests) 7. All-time (no date params) — full-history fallback (90 tests) 8. Resolution sweep — mid/low/ultralow downsampling (54 tests) """ ALL_TIME_GP_ID = "95ee273c-09cb-411c-ae00-8806e9ca938e" # Local Natives GP_ID = "e49ea9f9-0a74-4623-b3c6-ff819c7c0a4d" START_DATE = "2022-07-26" END_DATE = "2022-07-27" def _url(self, ts_type, date_params, extra=""): return ( f"{BASE_URL}/participant/{self.GP_ID}/timeseries" f"?{date_params}&type={ts_type}&{PAGINATION}{extra}" ) # ── Group 1: Streams with date variants ──────────────────────── # # 6 headers x 6 stream_types x 2 date_params = 72 tests. # Both start_date/end_date and days_back must produce valid results. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("ts_type", TIMESERIES_STREAMS_TYPES) @pytest.mark.parametrize( "date_params", [ pytest.param("start_date=2022-07-26&end_date=2022-07-27", id="date_range"), pytest.param("days_back=7", id="days_back"), ], ) def test_timeseries_streams(self, hdrs, ts_type, date_params): """Streaming timeseries returns correct keys for both date param styles. start_date/end_date and days_back exercise different date-handling code paths in the model layer. """ payload = assert_endpoint( self._url(ts_type, date_params), headers=hdrs, ) if payload["items"]: _assert_participant_timeseries_keys(payload["items"][0].keys(), ts_type) # ── Group 2: Streams country-filtered ────────────────────────── # # 6 headers x 6 stream_types = 36 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("ts_type", TIMESERIES_STREAMS_TYPES) def test_timeseries_streams_country_filtered(self, hdrs, ts_type): """Country filter activates the country-variant query path.""" payload = assert_endpoint( self._url( ts_type, f"start_date={self.START_DATE}&end_date={self.END_DATE}", "&countries=GB&countries=DE", ), headers=hdrs, ) if payload["items"]: _assert_participant_timeseries_keys(payload["items"][0].keys(), ts_type) # ── Group 3: Downloads ───────────────────────────────────────── # # 6 headers x 5 download_types = 30 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("ts_type", TIMESERIES_DOWNLOADS_TYPES) def test_timeseries_downloads(self, hdrs, ts_type): """Download timeseries returns correct keys (no saves/skip_rate).""" payload = assert_endpoint( self._url( ts_type, f"start_date={self.START_DATE}&end_date={self.END_DATE}", ), headers=hdrs, ) if payload["items"]: _assert_participant_timeseries_keys(payload["items"][0].keys(), ts_type) # ── Group 4: Downloads country-filtered ──────────────────────── # # 6 headers x 5 download_types = 30 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("ts_type", TIMESERIES_DOWNLOADS_TYPES) def test_timeseries_downloads_country_filtered(self, hdrs, ts_type): """Download timeseries with country filter narrows results.""" payload = assert_endpoint( self._url( ts_type, f"start_date={self.START_DATE}&end_date={self.END_DATE}", "&countries=GB&countries=DE", ), headers=hdrs, ) if payload["items"]: _assert_participant_timeseries_keys(payload["items"][0].keys(), ts_type) # ── Group 5: TRACK_STREAMS_BY_SOS_DETAILED variants ──────────── # # 6 headers × 5 param_variants = 30 tests. # Covers: no store filter, per-store, and stream_source filtering. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "extra", [ pytest.param("", id="no_store_filter"), pytest.param( f"&stores={SOS_DETAILED_STORE_IDS['spotify']}", id="spotify_only", ), pytest.param( f"&stores={SOS_DETAILED_STORE_IDS['apple']}", id="apple_only", ), pytest.param( f"&stores={SOS_DETAILED_STORE_IDS['spotify']}" "&stream_sources=collection&stream_sources=search", id="spotify_with_stream_sources", ), pytest.param( f"&stores={SOS_DETAILED_STORE_IDS['apple']}" "&stream_sources=discovery", id="apple_with_stream_sources", ), ], ) def test_timeseries_sos_detailed(self, hdrs, extra): """TRACK_STREAMS_BY_SOS_DETAILED returns items with id, date and value. Each item's id should be a known SOS detailed column name (e.g. streams_sos_spotify_collection). Validates the UNPIVOT output shape is consistent with the rest of the timeseries response. """ payload = assert_endpoint( self._url( "TRACK_STREAMS_BY_SOS_DETAILED", f"start_date={self.START_DATE}&end_date={self.END_DATE}", extra, ), headers=hdrs, ) if payload["items"]: item = payload["items"][0] assert "id" in item assert "date" in item assert "value" in item assert item["id"].startswith("streams_sos_") # ── Group 6: Store-filtered streams ──────────────────────────── # # 6 headers × 6 stream_types = 36 tests. The `stores` param exercises # the per-store IN-clause path in timeseries_streams.sql. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("ts_type", TIMESERIES_STREAMS_TYPES) def test_timeseries_streams_store_filtered(self, hdrs, ts_type): """Store filter narrows results; items still match expected shape.""" payload = assert_endpoint( self._url( ts_type, f"start_date={self.START_DATE}&end_date={self.END_DATE}", "&stores=286", ), headers=hdrs, ) if payload["items"]: _assert_participant_timeseries_keys(payload["items"][0].keys(), ts_type) # ── Group 7: All-time (no date params) ───────────────────────── # # 6 headers × (6 streams + 9 downloads) = 90 tests. # Omitting start_date/end_date/days_back triggers the no-date-filter # branch in timeseries SQL templates (full participant history). # Uses Local Natives — Bad Bunny's all-time scan times out in QA. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "ts_type", TIMESERIES_STREAMS_TYPES + TIMESERIES_DOWNLOADS_TYPES ) def test_timeseries_all_time(self, hdrs, ts_type): """No date params produces a full-history aggregation.""" payload = assert_endpoint( f"{BASE_URL}/participant/{self.ALL_TIME_GP_ID}/timeseries" f"?type={ts_type}&{PAGINATION}", headers=hdrs, ) if payload["items"]: _assert_participant_timeseries_keys(payload["items"][0].keys(), ts_type) # ── Group 8: Resolution sweep ────────────────────────────────── # # 6 headers x 3 resolutions x 3 query types = 54 tests. # Exercises the resolution_filter macro for each downsampling tier # (mid → DAYOFWEEK IN (1,3,6); low → (3,6); ultralow → DAYOFMONTH IN (14)). # Date ranges are sized so every tier emits non-empty data. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "resolution,start_date,end_date", [ pytest.param("mid", "2023-01-01", "2023-12-31", id="mid"), pytest.param("low", "2020-01-01", "2024-12-31", id="low"), pytest.param("ultralow", "2017-01-01", "2024-12-31", id="ultralow"), ], ) @pytest.mark.parametrize( "ts_type", ["TRACK_STREAMS", "TRACK_DOWNLOADS", "TRACK_STREAMS_BY_SOS"], ) def test_timeseries_resolution( self, hdrs, resolution, start_date, end_date, ts_type ): """Resolution downsampling — macro emits the right day-of-week filter. Uses Local Natives (smaller catalog) so multi-year ranges fit inside the 60s gateway window — Bad Bunny's 5-year scan exceeds it. """ payload = assert_endpoint( f"{BASE_URL}/participant/{self.ALL_TIME_GP_ID}/timeseries" f"?type={ts_type}&{PAGINATION}" f"&resolution={resolution}" f"&start_date={start_date}&end_date={end_date}", headers=hdrs, ) if payload["items"]: _assert_participant_timeseries_keys(payload["items"][0].keys(), ts_type) class TestParticipantTrackStreams: """Integration tests for /participant//track-streams-all and .../by-store. Both endpoints share the same query params and branching logic. Groups exercise each meaningful SQL branch in `analytics.queries.participant_track_streams`: - country_ids absent vs present → switches Snowflake view (V_STREAMS_BY_PARTICIPANT_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. Track streams all — base + filter variants (36 tests) 2. Track streams by store — base + filter variants (36 tests) """ GP_ID = "94602e2e-80ef-4061-ba61-81e08f328c91" 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: Track streams all ───────────────────────────────── # # 6 headers × 6 query variants = 36 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("query", QUERY_VARIANTS) def test_track_streams_all(self, hdrs, query): """Aggregated track streams returns global_participant_id 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). """ url = config.PARTICIPANT_TRACK_STREAMS_ALL_URL.replace( "", self.GP_ID ) assert_endpoint( url + query, headers=hdrs, items_key=None, expect_payload_keys=["global_participant_id", "items"], ) # ── Group 2: Track streams by store ──────────────────────────── # # 6 headers × 6 query variants = 36 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize("query", QUERY_VARIANTS) def test_track_streams_by_store(self, hdrs, query): """Per-store track streams returns global_participant_id 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). """ url = config.PARTICIPANT_TRACK_STREAMS_STORE_URL.replace( "", self.GP_ID ) assert_endpoint( url + query, headers=hdrs, items_key=None, expect_payload_keys=["global_participant_id", "stores"], ) class TestParticipantAggregatedStreams: """Integration tests for /participant//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) """ GP_ID = "e49ea9f9-0a74-4623-b3c6-ff819c7c0a4d" AGGREGATED_FIELDS = [ "all_other_rollup", "all_other_timeseries", "topn_timeseries", "topn_rollup", "total", ] def _url(self, dimension, extra=""): return ( config.PARTICIPANT_AGGREGATED_STREAMS_URL.replace( "", self.GP_ID ) + f"?dimension={dimension}{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_dimension(self, hdrs, dimension): """Each dimension returns all expected fields with 28-item timeseries. 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 if payload["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 ───────────────────────────────────────── # # 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["all_other_timeseries"]: assert len(payload["all_other_timeseries"]) == 7 for _, timeseries in payload["topn_timeseries"].items(): assert len(timeseries) == 7 class TestParticipantMetrics: """Integration tests for /participant/metrics. Verifies participant metrics listing with various filters, orderings, and participant ID lookups. Groups: 1. Default and sorted params — basic metrics response (12 tests) 2. IDs not null — all returned IDs are non-null (6 tests) 3. With participant IDs — filter by specific gp_ids (12 tests) 4. Non-existent artist — returns empty metrics (6 tests) """ # ── Group 1: Default and sorted params ───────────────────────── # # 6 headers x 2 param_variants = 12 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "params", [ pytest.param("", id="defaults"), pytest.param( "?limit=10&offset=0" "&order_by=growth_percentage_28_days&order_dir=ASC", id="sorted", ), pytest.param( "?limit=10&offset=0" "&order_by=growth_percentage_28_days&order_dir=desc", id="sorted_lowercase_order_dir", ), ], ) def test_metrics(self, hdrs, params): """Metrics endpoint returns list with all expected fields. Default params use server-side defaults; sorted params exercise the ORDER BY code path. The lowercase variant guards against regressions in case-insensitive order_dir handling — real clients send `desc`/`asc`, but the schema validator is strict. """ if hdrs == INSIGHTS_EMPLOYEE: pytest.skip( "Disabled: full-catalog full-access aggregation 60s " "Snowflake timeout for the employee profile." ) payload = assert_endpoint( config.PARTICIPANT_METRICS_URL + params, headers=hdrs, items_key=None, ) assert "metrics" in payload assert isinstance(payload["metrics"], list) if payload["metrics"]: for field in PARTICIPANT_METRICS_FIELDS: assert field in payload["metrics"][0] # ── Group 2: IDs not null ────────────────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_metrics_ids_not_none(self, hdrs): """Every returned metric item must have a non-null id. Catches regressions where NULL global_participant_ids leak into the response from JOINs with unmatched rows. """ if hdrs == INSIGHTS_EMPLOYEE: pytest.skip( "Disabled: full-catalog full-access aggregation 60s " "Snowflake timeout for the employee profile." ) payload = assert_endpoint( config.PARTICIPANT_METRICS_URL + "?limit=10&offset=0" "&order_by=growth_percentage_28_days&order_dir=ASC", headers=hdrs, items_key=None, ) assert all(item["id"] is not None for item in payload["metrics"]) # ── Group 3: With participant IDs ────────────────────────────── # # 6 headers x 2 filter_variants = 12 tests. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "params", [ pytest.param( "?global_participant_ids=1fd9086d-8a3d-4385-a9a1-a67f2d384979", id="gp_id_only", ), pytest.param( "?country_code=GB&country_code=US" "&global_participant_ids=1fd9086d-8a3d-4385-a9a1-a67f2d384979", id="gp_id_with_countries", ), ], ) def test_metrics_with_participants(self, hdrs, params): """Filtering by global_participant_ids returns total_participants count. With country_code filter, exercises the country-filtered SQL path. """ payload = assert_endpoint( config.PARTICIPANT_METRICS_URL + params, headers=hdrs, items_key=None, ) assert "metrics" in payload assert isinstance(payload["metrics"], list) assert "total_participants" in payload if payload["metrics"]: for field in PARTICIPANT_METRICS_FIELDS: assert field in payload["metrics"][0] # ── Group 4: Non-existent artist ─────────────────────────────── # # 6 tests (one per header). @pytest.mark.parametrize("hdrs", ALL_HEADERS) def test_non_existent_artist_id(self, hdrs): """Non-existent gp_id returns empty metrics list. Verifies graceful handling rather than 404/500. """ payload = assert_endpoint( config.PARTICIPANT_METRICS_URL + "?global_participant_ids=i-dont-exist-324534", headers=hdrs, items_key=None, ) assert "metrics" in payload assert len(payload["metrics"]) == 0 assert "total_participants" in payload # ── Group 5: Filter SQL-branch coverage ──────────────────────── # # 6 headers × 5 param_variants = 30 tests. Each variant exercises a # distinct conditional branch in participant/metrics.sql. @pytest.mark.parametrize("hdrs", ALL_HEADERS) @pytest.mark.parametrize( "params", [ pytest.param("?store_ids=286", id="store_filter"), pytest.param("?distributors=theorchard", id="distributors_filter"), pytest.param("?label_ids=26760", id="label_filter_rimas"), pytest.param("?subaccount_ids=46189", id="subaccount_filter"), pytest.param( "?order_by=streams_28_days&order_dir=DESC&limit=5", id="sort_28d" ), ], ) def test_metrics_filter_branches(self, hdrs, params): """Filter variants exercise distinct SQL conditional branches.""" if hdrs == INSIGHTS_EMPLOYEE and "distributors=" in params: pytest.skip( "Disabled: full-catalog full-access distributors_filter " "aggregation 60s Snowflake timeout for the employee profile." ) payload = assert_endpoint( config.PARTICIPANT_METRICS_URL + params, headers=hdrs, items_key=None, ) assert "metrics" in payload assert isinstance(payload["metrics"], list) assert "total_participants" in payload if payload["metrics"]: for field in PARTICIPANT_METRICS_FIELDS: assert field in payload["metrics"][0] PARTICIPANT_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 TestParticipantDemographics: """Integration tests for /participant//demographics. Parametrized by 7 SQL-branch variants × 6 auth headers × 2 gp_ids = 84 tests. The two gp_ids exercise different label-access paths: - e49ea9f9-... (Bad Bunny, RIMAS 26760) - 95ee273c-... (Local Natives, Frenchkiss 6971) 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( "gp_id", [ pytest.param("e49ea9f9-0a74-4623-b3c6-ff819c7c0a4d", id="bad_bunny"), pytest.param("95ee273c-09cb-411c-ae00-8806e9ca938e", id="local_natives"), ], ) @pytest.mark.parametrize("query", PARTICIPANT_DEMOGRAPHICS_QUERY_VARIANTS) def test_demographics(self, hdrs, gp_id, query): """Demographics returns global_participant_id, demographics, sources.""" assert_endpoint( f"{BASE_URL}/participant/{gp_id}/demographics{query}", headers=hdrs, items_key=None, expect_payload_keys=["global_participant_id", "demographics", "sources"], ) class TestParticipantMetricsCountryCsv: """Comma-separated country_code on GET /participant-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 = ["GB", "US", "DE"] # ── 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( config.PARTICIPANT_METRICS_URL + repeated_qs, headers=hdrs, items_key=None, ) csv = assert_endpoint( config.PARTICIPANT_METRICS_URL + f"?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( config.PARTICIPANT_METRICS_URL + f"?country_code={','.join(ISO_ALPHA_2)}&limit=10&offset=0", headers=hdrs, items_key=None, ) assert "metrics" in payload