"""Integration tests for API endpoints.""" from random import randint, choice import pytest import requests from charts.config import ENVIRONMENT, QA_ENVIRONMENT, HEALTH_CHECK from charts.constants.artist_mappings import artist_aggregated_rankings_order_by_mapping, \ artist_aggregated_rankings_order_direction_mapping from charts.constants.chart_columns import CHART_RANKINGS_COLUMN_NAMES, CHART_RANKINGS_AGG_VALUES_COLUMN_NAMES, CHART_COLUMN_NAMES, CHART_RANKINGS_VALUES_ON_DATE_COLUMN_NAMES if ENVIRONMENT == QA_ENVIRONMENT: BASE_URL = 'https://qa-ows-charts.theorchard.io' else: BASE_URL = 'http://localhost:5000' @pytest.fixture def admin_permission_headers(): # Profile is taken from Test user "insightsemployee" # Full list can be found in https://us-east-1.console.aws.amazon.com/secretsmanager/secret?name=qa%2Fe2e-test-secrets%2Fuser-data%2Finsights-user-data®ion=us-east-1 return { 'Orchard-Profile-Id': '422069', 'Orchard-Profile-Type': 'InsightsProfile', } class TestHello: """Test hello endpoint.""" @pytest.fixture def response(self): """Return response.""" return requests.get(BASE_URL + HEALTH_CHECK) def test_succeeds(self, response): """Test successful response.""" assert response.status_code == 200, 'Reason: {}, URL: {}'.format( response.reason, response.url) def test_returns_payload(self, response): """Test returns correct payload.""" assert response.json() == {'status': 'ok'} class TestCharts: """Test charts related endpoints.""" chart_id = '0abb8e9c-dfe9-4baf-9a8d-76566c3a09c4' # spotify viral50 GLOBAL product_chart_id = '0e80f8fe-38e1-444f-a38f-f4f015915170' chart_date = '2021-06-10' public_sound_recording_id = 'f1187d04-2101-4a28-8943-4a66c631107a' public_product_id = '06ca6732-e943-4eab-a06b-a2da8c897dfe' def test_charts(self): """Test /charts """ response = requests.get(BASE_URL + '/charts') assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list assert len(response_payload) > 0 assert all([entry['chartId'] for entry in response_payload]) map_by_id = {entry['chartId']: entry for entry in response_payload} assert self.chart_id in map_by_id def test_chart(self): """Test /chart """ response = requests.get(BASE_URL + f'/charts/{self.chart_id}') assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is dict assert response_payload['chartId'] == self.chart_id def test_available_dates(self): """Test /charts/{chart_id}/dates """ response = requests.get(BASE_URL + f'/charts/{self.chart_id}/dates') assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list assert len(response_payload) > 0 assert self.chart_date in response_payload def test_chart_rankings(self): """Test /charts/{chart_id}/dates/{a_date} """ response = requests.get(BASE_URL + f'/charts/{self.chart_id}/dates/{self.chart_date}') assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list assert len(response_payload) > 0 assert all(['position' in entry for entry in response_payload]) assert response_payload[0]['position'] == 1 assert response_payload[1]['position'] == 2 assert response_payload[2]['position'] == 3 def test_chart_rankings_with_limit(self): """Test /charts/{chart_id}/dates/{a_date} """ response = requests.get(BASE_URL + f'/charts/{self.chart_id}/dates/{self.chart_date}?limit=3&offset=2') assert response.status_code == 200 response_payload = response.json() assert len(response_payload) == 3 assert all(['position' in entry for entry in response_payload]) assert response_payload[0]['position'] == 3 assert response_payload[1]['position'] == 4 assert response_payload[2]['position'] == 5 def test_chart_song_history(self): """Test /charts/{chart_id}/song-history/{public_sound_recording_id} """ response = requests.get(BASE_URL + f'/charts/{self.chart_id}/song-history/{self.public_sound_recording_id}') assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list assert len(response_payload) > 0 assert 'position' in response_payload[0] check_response_ids = [ entry['publicSoundRecordingId'] == self.public_sound_recording_id for entry in response_payload] assert all(check_response_ids) def test_chart_product_history(self): """Test /charts/{chart_id}/product-history/{ public_product_id} """ response = requests.get(BASE_URL + f'/charts/' f'{self.product_chart_id}/product-history/' f'{self.public_product_id}') assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list assert len(response_payload) > 0 assert 'position' in response_payload[0] check_response_ids = [ entry['publicProductId'] == self.public_product_id for entry in response_payload] assert all(check_response_ids) @pytest.mark.parametrize( "params", [ {"platforms": ["spotify", "appleMusic"]}, {"platforms": ["spotify"]}, {"platforms": ["appleMusic"]}, ] ) def test_get_charts_by_platforms(self, params): """Test '/chart/by-platforms/' This API returns main charts information for only passed "platforms" """ response = requests.get(BASE_URL + "/charts/by-platforms/", params=params) assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list if response_payload: main_object_keys = ["platform", "chartName", "chartDate", "chartKey"] platforms = [pl.upper() for pl in params["platforms"]] for response_item in response_payload: assert all([key in response_item.keys() for key in main_object_keys]) assert response_item["platform"] in platforms @pytest.mark.parametrize( 'params,expected_status_code', [ [{}, 500], [{"platform": "spotify", "type": "regional", "frequency": "daily", "target": "track", "country_code": "global"}, 200], [{"platform": "spotify", "type": "viral", "frequency": "daily", "target": "track"}, 200], [{"platform": "spotify", "type": "viral", "frequency": "weekly", "target": "track"}, 200], [{"platform": "amazon", "genre": "blues & country", "frequency": "daily", "target": "track"}, 200], [{"platform": "amazon", "genre": "Blues_&_Country", "frequency": "daily", "target": "track"}, 200], [{"platform": "youtube", "type": "music videos", "frequency": "weekly", "target": "video"}, 200], [{"platform": "youtube", "type": "MUSIC_VIDEOS", "frequency": "weekly", "target": "video"}, 200], [{"platform": "youtube", "type": "tracks", "frequency": "weekly", "target": "track"}, 200], [{"platform": "tiktok", "type": "default", "frequency": "weekly", "target": "track"}, 200], [{"platform": "tiktok", "type": "default", "frequency": "daily", "target": "track"}, 200], [{"platform": "soundcloud", "type": "top", "frequency": "daily", "target": "track"}, 200], [{"platform": "shazam", "type": "default", "frequency": "daily", "target": "track"}, 200], [{"platform": "recochoku", "type": "single", "frequency": "daily", "target": "track"}, 200], [{"platform": "linemusic", "type": "top 100", "frequency": "daily", "target": "track"}, 200], [{"platform": "linemusic", "type": "top_100", "frequency": "daily", "target": "track"}, 200], [{"platform": "deezer", "type": "default", "frequency": "daily", "target": "track"}, 200], [{"platform": "applemusic", "type": "default", "frequency": "daily", "target": "track"}, 200], ] ) def test_get_charts_by_definition(self, params, expected_status_code): """Test '/charts/by-definition/' This API returns charts from definition (platform, type, frequency, target, genre, country_code) """ response = requests.get(BASE_URL + "/charts/by-definition/", params=params) assert response.status_code == expected_status_code if expected_status_code == 200: response_payload = response.json() assert type(response_payload) is list if response_payload: main_object_keys = ( "chartId", ) for response_item in response_payload: assert all([key in response_item.keys() for key in main_object_keys]) @pytest.mark.parametrize( 'params,expected_status_code', [ [{}, 500], [{"chart_ids_with_date": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "chart_date": "2024-04-17"}]}, 200], [{"chart_ids_with_d": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "chart_date": "2024-04-17"}]}, 500], [{"chart_ids_with_date": [{"chart": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "chart_date": "2024-04-17"}]}, 500], [{"chart_ids_with_date": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "date": "2024-04-17"}]}, 500], ] ) def test_get_chart_rankings_for_chart_ids(self, admin_permission_headers, params, expected_status_code): """Test '/charts/rankings-for-chart-ids/' This API returns chart rankings from post request json of list of chart_ids and dates """ response = requests.post( BASE_URL + "/charts/rankings-for-chart-ids/", json=params, headers=admin_permission_headers, ) assert response.status_code == expected_status_code if expected_status_code == 200: response_payload = response.json() assert type(response_payload) is list if response_payload: main_object_keys = ( "chartId", "chartDate", "uuids" ) for response_item in response_payload: data = response_item["data"] assert all([key in data.keys() for key in main_object_keys]) @pytest.mark.parametrize( 'params,expected_status_code', [ [{}, 500], [{"chart_rankings": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "chart_date": "2024-04-17", "uuid": "58aafccf-67b0-415e-a95f-1c63cb03f0ce"}]}, 200], [{"chart_rankin": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "chart_date": "2024-04-17", "uuid": "58aafccf-67b0-415e-a95f-1c63cb03f0ce"}]}, 500], [{"chart_rankings": [{"chart": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "chart_date": "2024-04-17", "uid": "58aafccf-67b0-415e-a95f-1c63cb03f0ce"}]}, 500], [{"chart_rankings": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "date": "2024-04-17", "uudi": "58aafccf-67b0-415e-a95f-1c63cb03f0ce"}]}, 500], [{"chart_rankings": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "chart_date": "2024-04-17", "uudi": "58aafccf-67b0-415e-a95f-1c63cb03f0ce"}]}, 500], ] ) def test_get_chart_rankings_dataloaded(self, params, expected_status_code): """Test '/chart-rankings/dataloaded/' This API returns chart rankings from post request json of list of chart_ids, dates, and positions """ response = requests.post(BASE_URL + "/chart-rankings/dataloaded/", json=params) assert response.status_code == expected_status_code if expected_status_code == 200: response_payload = response.json() assert type(response_payload) is list if response_payload: for response_item in response_payload: assert all([key in response_item['data'].keys() for key in CHART_RANKINGS_COLUMN_NAMES]) @pytest.mark.parametrize( 'params,expected_status_code', [ [{}, 500], [{"chart_rankings_aggregated": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "public_sound_recording_id": "caca9036-d4fd-4287-8e31-9889beab2144", "chartmetric_track_id": "1e9oZCCiX42nJl0AcqriVo", "public_product_id": None}]}, 200], [{"chart_rankin": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "public_sound_recording_id": None, "chartmetric_track_id": None, "public_product_id": None}]}, 500], [{"chart_rankings_aggregated": [{"chart": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "public_sound_recording_id": "07dee890-159b-4a98-b429-d55d46c1b953", "chartmetric_track_id": None, "public_product_id": None}]}, 500], [{"chart_rankings_aggregated": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "public_sound_recording": "07dee890-159b-4a98-b429-d55d46c1b953", "chartmetric_track_id": None, "public_product_id": None}]}, 500], [{"chart_rankings_aggregated": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "public_sound_recording_id": None, "chartmetric_track": None, "public_product_id": None}]}, 500], ] ) def test_get_chart_rankings_aggregated_values_dataloaded(self, params, expected_status_code): """Test '/chart-rankings/aggregated-values-dataloaded/' This API returns chart rankings aggregated values from post request json of list of chart_ids, dates, and positions """ response = requests.post(BASE_URL + "/chart-rankings/aggregated-values-dataloaded/", json=params) assert response.status_code == expected_status_code if expected_status_code == 200: response_payload = response.json() assert type(response_payload) is list if response_payload: for response_item in response_payload: assert all([key in response_item['data'].keys() for key in CHART_RANKINGS_AGG_VALUES_COLUMN_NAMES]) @pytest.mark.parametrize( 'params,expected_status_code', [ [{}, 500], [{"chart_rankings_aggregated_on_date": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "chart_date": "2019-11-20", "public_sound_recording_id": "caca9036-d4fd-4287-8e31-9889beab2144", "chartmetric_track_id": "1e9oZCCiX42nJl0AcqriVo", "public_product_id": None}]}, 200], [{"chart_rankin": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "chart_date": "2019-11-20", "public_sound_recording_id": "caca9036-d4fd-4287-8e31-9889beab2144", "chartmetric_track_id": "1e9oZCCiX42nJl0AcqriVo", "public_product_id": None}]}, 500], [{"chart_rankings_aggregated_on_date": [{"chart": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "chart_date": "2019-11-20", "public_sound_recording_id": "caca9036-d4fd-4287-8e31-9889beab2144", "chartmetric_track_id": "1e9oZCCiX42nJl0AcqriVo", "public_product_id": None}]}, 500], [{"chart_rankings_aggregated_on_date": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "chart_date": "2019-11-20", "public_sound_recording": "caca9036-d4fd-4287-8e31-9889beab2144", "chartmetric_track_id": "1e9oZCCiX42nJl0AcqriVo", "public_product_id": None}]}, 500], [{"chart_rankings_aggregated_on_date": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "chart_dat": "2019-11-20", "public_sound_recording_id": "caca9036-d4fd-4287-8e31-9889beab2144", "chartmetric_track_id": "1e9oZCCiX42nJl0AcqriVo", "public_product_id": None}]}, 500], ] ) def test_get_chart_rankings_aggregated_values_on_date_dataloaded(self, params, expected_status_code): """Test '/chart-rankings/aggregated-values-on-date-dataloaded/' This API returns chart rankings aggregated values from post request json of list of chart_ids, dates, and positions """ response = requests.post(BASE_URL + "/chart-rankings/aggregated-values-on-date-dataloaded/", json=params) assert response.status_code == expected_status_code if expected_status_code == 200: response_payload = response.json() assert type(response_payload) is list if response_payload: for response_item in response_payload: assert all([key in response_item['data'].keys() for key in CHART_RANKINGS_VALUES_ON_DATE_COLUMN_NAMES]) @pytest.mark.parametrize( 'params,expected_status_code', [ [{}, 500], [{"chart_ids": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5"}]}, 200], [{"chart_i": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5"}]}, 500], [{"chart_ids": [{"chart": "34f76bdb-4f5b-4917-a47e-02916f7291f5"}]}, 500], ] ) def get_chart_dates_for_chart_ids(self, params, expected_status_code): """Test '/charts/dates-for-chart-ids/' This API returns chart rankings from post request json of list of chart_ids, chart_dates """ response = requests.post(BASE_URL + "/charts/dates-for-chart-ids/", json=params) assert response.status_code == expected_status_code if expected_status_code == 200: response_payload = response.json() assert type(response_payload) is list if response_payload: for response_item in response_payload: assert all([key in response_item.keys() for key in ["chartId", "chartDates"]]) @pytest.mark.parametrize( 'params,expected_status_code', [ [{}, 500], [{"chart_ids": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5"}]}, 200], [{"chart_i": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5"}]}, 500], [{"chart_ids": [{"chart": "34f76bdb-4f5b-4917-a47e-02916f7291f5"}]}, 500], ] ) def get_metadata_for_chart_ids(self, params, expected_status_code): """Test '/charts/metadata/' This API returns chart rankings from post request json of list of chart_ids and its metadata """ response = requests.post(BASE_URL + "/charts/metadata/", json=params) assert response.status_code == expected_status_code if expected_status_code == 200: response_payload = response.json() assert type(response_payload) is list if response_payload: for response_item in response_payload: assert all([key in response_item.keys() for key in CHART_COLUMN_NAMES]) @pytest.mark.parametrize( 'body', [ ( { "chart_ids": [ "2934be11-59d5-488b-b94b-76151a1e1887" ], "date": "2024-11-25", "order_by": "FEATURING", "order_dir": "DESC", "filter": {} } ), ( { "chart_ids": [ "2934be11-59d5-488b-b94b-76151a1e1887" ], "date": "2024-11-25", "order_by": "MIN_POSITION", "order_dir": "DESC", "filter": {} } ), ( { "chart_ids": [ "2934be11-59d5-488b-b94b-76151a1e1887" ], "date": "2024-11-25", "order_by": "FEATURING", "order_dir": "DESC", "filter": {"brand_uuid_in": ["d25a4cd1-e820-45f2-be5c-56edcfeb8298"]} } ), ( { "chart_ids": [ "2934be11-59d5-488b-b94b-76151a1e1887" ], "date": "2024-11-25", "order_by": "MIN_POSITION", "order_dir": "DESC", "filter": {"brand_uuid_in": ["d25a4cd1-e820-45f2-be5c-56edcfeb8298"]} } ), ] ) def test_get_global_sound_recordings_for_charts_date(self, body, admin_permission_headers): """Test '/charts/gsrs-for-charts-date/' """ response = requests.post( BASE_URL + "/charts/gsrs-for-charts-date", json=body, headers=admin_permission_headers, ) assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is dict assert len(response_payload.get('globalSoundRecordingIsrcs')) > 0, 'expected non-empty response' @pytest.mark.parametrize( 'params,expected_status_code, result', [ [{}, 500, None], [{"chart_entries": [{ "chart_id": "2934be11-59d5-488b-b94b-76151a1e1887", "chart_date": "2024-11-25", "isrc": "USQX92405786", }]}, 200, 57], [{"chart_ids": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5"}]}, 500, None], [{"chart_entries": [{ "chart_id": "63190429-0bb9-48a6-9139-e2934526c939", "chart_date": "2024-11-25", "isrc": "USQX92405786", }]}, 200, None], ] ) def test_get_chart_date_entry_for_isrcs(self, params, expected_status_code, result): """Test '/charts/chart-date-entry/' This API returns chart entries from post request json of list of chart_ids, chart_dates, isrcs """ response = requests.post(BASE_URL + "/charts/chart-date-entry/", json=params) assert response.status_code == expected_status_code if expected_status_code == 200: response_payload = response.json() assert type(response_payload) is list if response_payload: data = response_payload[0]["data"] if data: assert data["position"] == result else: assert data == result @pytest.mark.parametrize( 'params,expected_status_code, result', [ [{}, 500, None], [ { "chart_entries": [{ "chart_date": "2024-11-25", "isrc": "USQX92405786", }], "chart_ids": ["2934be11-59d5-488b-b94b-76151a1e1887"] }, 200, { "feature": 1, "average_position": 57, "top_10_rank": 0 } ], [{"chart_ids": [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5"}]}, 500, None], [{ "chart_entries": [{ "chart_date": "2024-11-25", "isrc": "USQX92405786", }], "chart_ids": ["63190429-0bb9-48a6-9139-e2934526c939"] }, 200, None], ] ) def test_get_chart_date_entry_summary_for_isrcs(self, params, expected_status_code, result): """Test '/charts/chart-date-entry-summary/' This API returns chart entries summaries from post request json of list of chart_ids, chart_dates, isrcs """ response = requests.post(BASE_URL + "/charts/chart-date-entry-summary/", json=params) assert response.status_code == expected_status_code if expected_status_code == 200: response_payload = response.json() assert type(response_payload) is list if response_payload: data = response_payload[0]["data"] assert data == result class TestTracks: """Test tracks related endpoints.""" def test_get_track_aggregated_rankings(self): """Test '/tracks//aggregated' """ public_sound_recording_id = 'ff377c59-2800-4c20-a409-6de41b112973' response = requests.get(BASE_URL + f'/public-sound-recordings/{public_sound_recording_id}') assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list assert len(response_payload) > 0 def test_get_track_aggregated_rankings_with_appeared_on_date(self): """Test '/recordings/' """ public_sound_recording_id = 'ff377c59-2800-4c20-a409-6de41b112973' appeared_on_date = '2017-09-18' response = requests.get(BASE_URL + f'/public-sound-recordings/{public_sound_recording_id}?appeared_on_date={appeared_on_date}') assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list assert len(response_payload) > 0 @pytest.mark.parametrize( 'params', [ {"isrc": "USSM12209777", "platforms": ["spotify", "appleMusic"]}, {"isrc": "USSM12209777", "platforms": ["spotify"]}, {"isrc": "USSM12209777", "platforms": ["appleMusic"]}, {"isrc": "USSM12209777", "platforms": ["spotify"], "order_by": "current_position"}, {"isrc": "USSM12209777", "platforms": ["appleMusic"], "order_by": "current_position"}, ] ) def test_get_track_charts(self, params): """Test '/tracks/charts/' This API returns charts for only passed "platforms where the track by passed "isrc" is present At some point there can be a situation when the track won`t be present in any chart, so we can just provide some new & fresher "isrc" """ order_by_param = params.get("order_by") response = requests.get(BASE_URL + "/track/charts/", params=params) assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list if response_payload: main_object_keys = ["platform", "chartName", "chartDate", "chartCount", "countries"] countries_object_keys = [ "countryCode", "addedDate", "streamsNumber", "currentPosition", "positionChange", "newlyEntered" ] platforms = [pl.upper() for pl in params["platforms"]] for response_item in response_payload: assert all([key in response_item.keys() for key in main_object_keys]) item_countries = response_item["countries"] assert len(item_countries) >= 1 for item_country in item_countries: assert all([key in item_country.keys() for key in countries_object_keys]) assert response_item["platform"] in platforms test_position_order = 0 if order_by_param and order_by_param == "current_position": for country_data in item_countries: assert test_position_order <= country_data["currentPosition"] test_position_order = country_data["currentPosition"] class TestArtists: """Test artist related endpoints.""" artist_id = 3479 public_sound_recording_id = '43ccf7b1-4400-4c1b-89de-cd1b68908a95' isrc = 'USUM71607007' @pytest.mark.parametrize( 'limit, offset, order_by, order_dir, is_error', [ [1, 1, 'POSITION', 'ASC', False], # some error values to test [1, 1, 'pOSITION', 'ASC', True], [1, 1, 'POSITIONN', 'ASC', True], [1, 1, 'position', 'ASC', True], [1, 1, 'POSITION', 'ASK', True], ] + [ # for each supported 'order_by' generate random limit, offset, order_dir [ randint(1, 10), randint(0, 5), order_by, choice(list(artist_aggregated_rankings_order_direction_mapping)), # any of ordering direction False ] for order_by in artist_aggregated_rankings_order_by_mapping ] ) def test_get_artist_aggregated_rankings(self, limit, offset, order_by, order_dir, is_error): """Test '/artists//aggregated' """ params = { 'limit': limit, 'offset': offset, 'order_by': order_by, 'order_dir': order_dir, } response = requests.get(BASE_URL + f'/artists/{self.artist_id}/aggregated', params=params) if is_error: assert response.status_code == 500 else: assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list assert len(response_payload) == limit @pytest.mark.parametrize( 'params, response_ordered_by, order_ascending', [ [ {'order_by': 'POSITION', 'order_dir': 'ASC'}, 'position', True ], [ {'order_by': 'DAYS_ON_CHART', 'order_dir': 'asc'}, 'daysOnChart', True ], [ {'order_by': 'DAYS_ON_CHART', 'order_dir': 'DESC'}, 'daysOnChart', False ], # CHANGE property seems to have bug. Most values are null. will be fixed in DS-3766 # TODO: Uncomment after fixing of DS-3766 # [ # {'order_by': 'CHANGE', 'order_dir': 'ASC'}, # 'change', True # ], [ {'order_by': 'FIRST_POSITION', 'order_dir': 'DESC'}, 'firstPosition', False ], [ {'order_by': 'MOST_RECENT_POSITION', 'order_dir': 'ASC'}, 'mostRecentPosition', True ], [ {'order_by': 'PEAK_POSITION', 'order_dir': 'ASC'}, 'peakPosition', True ], [ {'order_by': 'PREVIOUS_POSITION', 'order_dir': 'ASC'}, 'previousPosition', True ], [ {'order_by': 'SOUND_RECORDING_NAME', 'order_dir': 'ASC'}, 'soundRecordingName', True ], ] ) def test_get_artist_aggregated_rankings_ordering(self, params, response_ordered_by, order_ascending): """Test '/artists//aggregated' """ params['limit'] = 20 # check only first N items url = BASE_URL + f'/artists/{self.artist_id}/aggregated' response = requests.get(url, params=params) assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list assert len(response_payload) > 0 prev_item = None for response_item in response_payload: if prev_item is not None: if params['order_by'] == 'CHANGE' and prev_item[response_ordered_by] is None: continue assert_message = f'response shall be ordered by "{response_ordered_by}" for {url} params:{params}' if order_ascending: assert prev_item[response_ordered_by] <= response_item[response_ordered_by], assert_message else: assert prev_item[response_ordered_by] >= response_item[response_ordered_by], assert_message prev_item = response_item @pytest.mark.parametrize( 'params, response_values', [ [ {'isrc': isrc}, {'isrc': isrc} ], [ {'public_sound_recording_id': public_sound_recording_id}, {'publicSoundRecordingId': public_sound_recording_id}, ], [ {'isrc': isrc, 'public_sound_recording_id': public_sound_recording_id}, {'isrc': isrc, 'publicSoundRecordingId': public_sound_recording_id}, ], ] ) def test_get_artist_aggregated_rankings_filter(self, params, response_values): """Test '/artists//aggregated' """ response = requests.get(BASE_URL + f'/artists/{self.artist_id}/aggregated', params=params) assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list assert len(response_payload) > 0, 'expected non-empty response' for response_item in response_payload: for key, value in response_values.items(): assert response_item[key] == value def test_get_artist_songs(self): """Test '/artists//songs' """ response = requests.get(BASE_URL + f'/artists/{self.artist_id}/songs') assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list assert len(response_payload) > 0 def test_get_artist_charts(self): """Test '/artists//charts' """ response = requests.get(BASE_URL + f'/artists/{self.artist_id}/charts') assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list assert len(response_payload) > 0 def test_get_artist_countries(self): """Test '/artists//countries' """ response = requests.get(BASE_URL + f'/artists/{self.artist_id}/countries') assert response.status_code == 200 response_payload = response.json() assert type(response_payload) is list assert len(response_payload) > 0 class Digest: """Tests for Chart Digest related endpoints.""" @pytest.mark.parametrize( 'params,expected_status_code', [ [{"definition_key": "spotify_regional_weekly", "chart_id": "123"}, 500], [{"definition_key": "spotify_regional_weekly", "filter": "EXIT"}, 500], [{}, 500], [{"definition_key": "spotify_regional_weekly", "country_code": "global"}, 200], [{"definition_key": "spotify_regional_daily", "country_code": "global"}, 200], [{"definition_key": "spotify_viral_daily", "country_code": "global"}, 200], [{"definition_key": "applemusic_default_daily", "country_code": "us"}, 200], [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5"}, 200], [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "chart_date": "2023-03-15"}, 200], [{"definition_key": "spotify_regional_weekly", "country_code": "global", "filter": "ENTRY"}, 200], [{"definition_key": "spotify_regional_daily", "country_code": "global", "filter": "MAJOR_MOVE"}, 200], [{"definition_key": "spotify_viral_daily", "country_code": "global", "filter": "ENTRY"}, 200], [{"definition_key": "applemusic_default_daily", "country_code": "us", "filter": "MAJOR_MOVE"}, 200], [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "filter": "ENTRY"}, 200], [{"chart_id": "34f76bdb-4f5b-4917-a47e-02916f7291f5", "filter": "MAJOR_MOVE"}, 200] ] ) def test_get_chart_digest_tracklist(self, params, expected_status_code): """Test '/chart/digest/tracks/' This API returns chart tracks Is a new version of chart rankings that supports more parameters """ is_latest_chart_tracklist = "chart_date" not in params.keys() filter_param = params.get("filter", "ALL") response = requests.get(BASE_URL + "/chart/digest/tracklist", params=params) assert response.status_code == expected_status_code if expected_status_code == 200: response_payload = response.json() assert type(response_payload) is list if response_payload: main_object_keys = ( "position", "positionChange", "streams", "totalStreams", "totalShazams", "views", "numTracks", "totalUnits", "trackId", "isrc", "publicSoundRecordingId", "upc", "publicProductId", "videoId", "channelId", "channelName", "chartDate", "peakTimestamp", "peakPosition", "daysOnChart", "artistNames", "trend", "newlyEntered" ) if filter_param == "ENTRY": assert all([item["newlyEntered"] is True for item in response_payload]) if filter_param == "MAJOR_MOVE": assert all([item["trend"] >= 5 for item in response_payload]) for response_item in response_payload: assert all([key in response_item.keys() for key in main_object_keys]) is_new = response_item["newlyEntered"] trend = response_item["trend"] if is_latest_chart_tracklist: if is_new: assert trend is None else: assert isinstance(trend, int)