# import datetime # import logging # from collections import OrderedDict # from itertools import chain # # import pandas # import pytest # from ui_automation_framework.utils import logger as cl # # from app.api_client import ApiClient # log = cl.Logger(logging.DEBUG) # # new_env = 'https://staging-privateapi.filtr.com' # old_env = 'https://privateapi-auth0.filtr.com' # api_client_old = ApiClient(host_endpoint=old_env) # api_client_new = ApiClient(host_endpoint=new_env) # token_new = api_client_new.authorize() # #log.info('New token:' + token_new) # # # markets = ['ar'] # isrcs = ['USXDR1900703'] # dates = [("2020-01-01", "2020-02-29")] # last_dates = ["2020-10-16"] # playlist_ids = ['37i9dQZF1DWZUozJiHy44Y'] # containerIds = ["102834"] # @pytest.mark.parametrize("market", markets) # @pytest.mark.parametrize("isrc", isrcs) # def test_total_streams_count_bulk(isrc, market): # endpoint = f'/dsp-api/delphi/total-streams-count-bulk?isrc={isrc}&market={market}' # # get old response # old = api_client_old.get_response(endpoint, token_old, env=old_env)['body'] # # get new response # new = api_client_new.get_response(endpoint, token_new, env=new_env)['body'] # # log.info(f'endpoint: {endpoint}') # log.info(f'response old: {old}') # log.info(f'response new: {new}') # # json_keys = ['global', 'market', 'spotify', 'amazon', 'apple'] # old_endpoint_values = [] # new_endpoint_values = [] # diffs = [] # # old_endpoint_values.extend([old['global'], old['market'] # , old['first_stream_date']['spotify'] # , old['first_stream_date']['amazon'] # , old['first_stream_date']['apple']]) # new_endpoint_values.extend([new['global'], new['market'] # , new['first_stream_date']['spotify'] # , new['first_stream_date']['amazon'] # , new['first_stream_date']['apple']]) # # diffs.extend([new['global'] / old['global'] * 100, # new['market'] / old['market'] * 100, # # datetime.datetime.strptime(old['first_stream_date']['spotify'], "%Y-%m-%d").date() - # datetime.datetime.strptime(new['first_stream_date']['spotify'], "%Y-%m-%d").date(), # # datetime.datetime.strptime(old['first_stream_date']['amazon'], "%Y-%m-%d").date() - # datetime.datetime.strptime(new['first_stream_date']['amazon'], "%Y-%m-%d").date(), # # datetime.datetime.strptime(old['first_stream_date']['apple'], "%Y-%m-%d").date() - # datetime.datetime.strptime(new['first_stream_date']['apple'], "%Y-%m-%d").date(), # ]) # # df = pandas.DataFrame({'old_values': old_endpoint_values, # 'new_values': new_endpoint_values, # 'is_correct_100%/diff_days': diffs}, # index=json_keys) # # df.to_excel(f'total-streams-count-bulk_{isrc}_{market}.xlsx') # # @pytest.mark.parametrize("market", markets) # @pytest.mark.parametrize("isrc", isrcs) # @pytest.mark.parametrize(("start_date", "end_date"), dates) # def test_track_per_country_v1_spotify(isrc, market, start_date, end_date): # endpoint = f'/dsp-api/consumer_analytics/v1/track-per-country?start_date={start_date}&end_date={end_date}&market={market}&streams_only=false&vendor=spotify&isrc={isrc}' # # old = api_client_old.get_response(endpoint, token_old, env=old_env)['body'][0]['data'][0]['data'] # new = api_client_new.get_response(endpoint, token_new, env=new_env)['body'][0]['data'][0]['data'] # # keys_to_check = list(dict(old[0]).keys()) # keys_to_check.remove('date') # # log.info(f'endpoint: {endpoint}') # log.info(f'response old: {old}') # log.info(f'response new: {new}') # # old_values = [] # new_values = [] # is_correct_percentage = [] # # for k in keys_to_check: # old_v = sum(o[k] for o in old) # new_v = sum(n[k] for n in new) # if old_v != 0: # percent = new_v / old_v * 100 # elif old_v == new_v == 0: # percent = 100 # else: # percent = 0 # old_values.append(old_v) # new_values.append(new_v) # is_correct_percentage.append(percent) # # #Temprorary calculations while AP-4486 is fixed # sourceOtherTempOld = sum(o['sourceOther'] for o in old) + sum(o['sourceUnknown'] for o in old) # sourceOtherTempNew = sum(n['sourceOther'] for n in new) + sum(n['sourceUnknown'] for n in new) # # sourceOthersPlaylistTempOld = sum(o['sourceOthersPlaylist'] for o in old) + sum(o['sourceDiscoverWeekly'] for o in old) + sum(o['sourceReleaseRadar'] for o in old) # sourceOthersPlaylistTempNew = sum(n['sourceOthersPlaylist'] for n in new) + sum(n['sourceDiscoverWeekly'] for n in new) + sum(n['sourceReleaseRadar'] for n in new) # # sourceRadioTempOld = sum(o['sourceRadio'] for o in old) + sum(o['sourceDailyMix'] for o in old) # sourceRadioTempNew = sum(n['sourceRadio'] for n in new) + sum(n['sourceDailyMix'] for n in new) # # old_values.append(sourceOtherTempOld) # old_values.append(sourceOthersPlaylistTempOld) # old_values.append(sourceRadioTempOld) # new_values.append(sourceOtherTempNew) # new_values.append(sourceOthersPlaylistTempNew) # new_values.append(sourceRadioTempNew) # # is_correct_percentage.append(get_percentage(sourceOtherTempOld, sourceOtherTempNew)) # is_correct_percentage.append(get_percentage(sourceOthersPlaylistTempOld, sourceOthersPlaylistTempNew)) # is_correct_percentage.append(get_percentage(sourceRadioTempOld, sourceRadioTempNew)) # # keys_to_check.append('sourceOtherTemp') # keys_to_check.append('sourceOthersPlaylistTemp') # keys_to_check.append('sourceRadioTemp') # # #Day by day response # keys = list(dict(old[0]).keys()) # # for o, n in zip(old, new): # keys_to_check.extend(keys) # for k in keys: # old_values.append(o[k]) # new_values.append(n[k]) # if k == keys[0]: # perc = str(datetime.datetime.strptime(o[k], "%Y-%m-%d").date() - \ # datetime.datetime.strptime(n[k], "%Y-%m-%d").date()) + " days diff" # else: # if o[k] != 0: # perc = str(n[k] / o[k] * 100) # else: # perc = '0' # is_correct_percentage.append(perc) # # o['sourceOtherTempOld'] = o['sourceOther'] + o['sourceUnknown'] # n['sourceOtherTempNew'] = n['sourceOther'] + n['sourceUnknown'] # # o['sourceOthersPlaylistTempOld'] = o['sourceOthersPlaylist'] + o['sourceDiscoverWeekly'] + o['sourceReleaseRadar'] # n['sourceOthersPlaylistTempNew'] = n['sourceOthersPlaylist'] + n['sourceDiscoverWeekly'] + n['sourceReleaseRadar'] # # o['sourceRadioTempOld'] = o['sourceRadio'] + o['sourceDailyMix'] # n['sourceRadioTempNew'] = n['sourceRadio'] + n['sourceDailyMix'] # # old_values.append(o['sourceOtherTempOld']) # old_values.append(o['sourceOthersPlaylistTempOld']) # old_values.append(o['sourceRadioTempOld']) # new_values.append(n['sourceOtherTempNew']) # new_values.append(n['sourceOthersPlaylistTempNew']) # new_values.append(n['sourceRadioTempNew']) # keys_to_check.append('sourceOtherTemp') # keys_to_check.append('sourceOthersPlaylistTemp') # keys_to_check.append('sourceRadioTemp') # # is_correct_percentage.append(get_percentage(o['sourceOtherTempOld'], n['sourceOtherTempNew'])) # is_correct_percentage.append(get_percentage(o['sourceOthersPlaylistTempOld'], n['sourceOthersPlaylistTempNew'])) # is_correct_percentage.append(get_percentage(o['sourceRadioTempOld'], n['sourceRadioTempNew'])) # # df = pandas.DataFrame({'old_values': old_values, # 'new_values': new_values, # 'is_correct_100%': is_correct_percentage # }, index=keys_to_check) # df.to_excel(f'track_per_country_v1_spotify_{market}_{start_date}_{end_date}_{isrc}.xlsx') # # @pytest.mark.parametrize("market", markets) # @pytest.mark.parametrize("isrc", isrcs) # @pytest.mark.parametrize(("start_date", "end_date"), dates) # def test_track_per_country_v1_apple(isrc, market, start_date, end_date): # endpoint = f'/dsp-api/consumer_analytics/v1/track-per-country?start_date={start_date}&end_date={end_date}&market={market}&streams_only=false&vendor=apple&isrc={isrc}' # # old = api_client_old.get_response(endpoint, token_old, env=old_env)['body'][0]['data'][0]['data'] # new = api_client_new.get_response(endpoint, token_new, env=new_env)['body'][0]['data'][0]['data'] # streams = 'streams' # container = 'container' # source = 'source' # keys_container = list(dict(old[0][container]).keys()) # keys_source = list(dict(old[0][source]).keys()) # all_keys = [] # all_keys.append(streams) # for k in keys_container: # all_keys.append(f'{container}_{k}') # for k in keys_source: # all_keys.append(f'{source}_{k}') # # log.info(f'key streams: {streams}') # log.info(f'keys container: {keys_container}') # log.info(f'keys source: {keys_source}') # log.info(f'endpoint: {endpoint}') # log.info(f'response old: {old}') # log.info(f'response new: {new}') # # old_values = [] # new_values = [] # is_correct_percentage = [] # # # streams values # old_v = sum(o[streams] for o in old) # new_v = sum(n[streams] for n in new) # if old_v != 0: # percent = new_v / old_v * 100 # elif old_v == new_v == 0: # percent = 100 # else: # percent = 0 # old_values.append(old_v) # new_values.append(new_v) # is_correct_percentage.append(percent) # # # container values # for k in keys_container: # old_v = sum(o[container][k] for o in old) # new_v = sum(n[container][k] for n in new) # if old_v != 0: # percent = new_v / old_v * 100 # elif old_v == new_v == 0: # percent = 100 # else: # percent = 0 # old_values.append(old_v) # new_values.append(new_v) # is_correct_percentage.append(percent) # # # source values # for k in keys_source: # old_v = sum(o[source][k] for o in old) # new_v = sum(n[source][k] for n in new) # if old_v != 0: # percent = new_v / old_v * 100 # elif old_v == new_v == 0: # percent = 100 # else: # percent = 0 # old_values.append(old_v) # new_values.append(new_v) # is_correct_percentage.append(percent) # # #Day by day response # for o, n in zip(old, new): # # all_keys.append('date') # all_keys.append(streams) # for k in keys_container: # all_keys.append(f'{container}_{k}') # for k in keys_source: # all_keys.append(f'{source}_{k}') # # old_values.append(o['date']) # new_values.append(n['date']) # is_correct_percentage.append( # str(datetime.datetime.strptime(o['date'], "%Y-%m-%d").date() - datetime.datetime.strptime(n['date'], "%Y-%m-%d").date()) + " days diff") # old_values.append(o[streams]) # new_values.append(n[streams]) # if o[streams] != 0: # percent = str(n[streams] / o[streams] * 100) # else: # percent = 0 # is_correct_percentage.append(percent) # # for k in keys_container: # old_values.append(o[container][k]) # new_values.append(n[container][k]) # if o[container][k] != 0: # percent = str(n[container][k] / o[container][k] * 100) # else: # percent = 0 # is_correct_percentage.append(percent) # # for k in keys_source: # old_values.append(o[source][k]) # new_values.append(n[source][k]) # if o[source][k] != 0: # percent = str(n[source][k] / o[source][k] * 100) # else: # percent = 0 # is_correct_percentage.append(percent) # # df = pandas.DataFrame({'old_values': old_values, # 'new_values': new_values, # 'is_correct_100%': is_correct_percentage # }, index=all_keys) # df.to_excel(f'track_per_country_v1_apple_{market}_{start_date}_{end_date}_{isrc}.xlsx') # # @pytest.mark.parametrize("market", markets) # @pytest.mark.parametrize("isrc", isrcs) # @pytest.mark.parametrize(("start_date", "end_date"), dates) # def test_track_per_country_v1_amazon(isrc, market, start_date, end_date): # endpoint = f'/dsp-api/delphi/amazon/tracks/v1/streams?isrc={isrc}&start_date={start_date}&end_date={end_date}&group_by=date&combine_isrc=true' # # old_resp = api_client_old.get_response(endpoint, token_old, env=old_env)['body']['items'] # new_resp = api_client_new.get_response(endpoint, token_new, env=new_env)['body']['items'] # old = [] # new = [] # country_code = 'country_code' # for i in old_resp: # if i[country_code] == market: # old.append(i) # for i in new_resp: # if i[country_code] == market: # new.append(i) # # log.info(f'old response: {old}') # log.info(f'new response: {new}') # # device_type = 'device_type' # engagement = 'engagement' # operating_system = 'operating_system' # selection_source_type = 'selection_source_type' # subscription_type = 'subscription_type' # referral_source_type = 'referral_source_type' # # non dict # non_royalty_streams = 'non_royalty_streams' # streams = 'streams' # keys_dt = list(dict(new[0][device_type]).keys()) # keys_e = list(dict(new[0][engagement]).keys()) # keys_os = list(dict(new[0][operating_system]).keys()) # keys_sst = list(dict(new[0][selection_source_type]).keys()) # keys_st = list(dict(new[0][subscription_type]).keys()) # keys_rst = list(dict(new[0][referral_source_type]).keys()) # # all_keys = [] # all_keys.extend([streams, non_royalty_streams]) # all_keys.extend(keys_dt) # all_keys.extend(keys_e) # all_keys.extend(keys_os) # all_keys.extend(keys_sst) # all_keys.extend(keys_st) # all_keys.extend(keys_rst) # log.info(f'keys: {all_keys}') # # old_values = [] # new_values = [] # is_correct_percentage = [] # # values = get_from_dict_no_parent(keys=[streams, non_royalty_streams], resp_old=old, resp_new=new) # old_values.extend(values['old_values']) # new_values.extend(values['new_values']) # is_correct_percentage.extend(values['is_correct_percentage']) # # for v in [(device_type, keys_dt), # (engagement, keys_e), # (operating_system, keys_os), # (selection_source_type, keys_sst), # (subscription_type, keys_st), # (referral_source_type, keys_rst) # ]: # values = get_from_dict(parent_name=v[0], keys=v[1], resp_old=old, resp_new=new) # old_values.extend(values['old_values']) # new_values.extend(values['new_values']) # is_correct_percentage.extend(values['is_correct_percentage']) # # df = pandas.DataFrame({'old_values': old_values, # 'new_values': new_values, # 'is_correct_100%': is_correct_percentage # }, index=all_keys) # df.to_excel(f'track_per_country_v1_amazon_{market}_{start_date}_{end_date}_{isrc}.xlsx') # # # def get_from_dict(parent_name, keys, resp_old, resp_new): # old_vs = [] # new_vs = [] # is_corr_ps = [] # for k in keys: # old_v = sum(o[parent_name][k] for o in resp_old if o.get(parent_name).get(k)) # new_v = sum(n[parent_name][k] for n in resp_new if n.get(parent_name).get(k)) # if old_v != 0: # percent = new_v / old_v * 100 # elif old_v == new_v == 0: # percent = 100 # else: # percent = 0 # old_vs.append(old_v) # new_vs.append(new_v) # is_corr_ps.append(percent) # return {'old_values': old_vs, # 'new_values': new_vs, # 'is_correct_percentage': is_corr_ps} # # # def get_from_dict_no_parent(keys, resp_old, resp_new): # old_vs = [] # new_vs = [] # is_corr_ps = [] # for k in keys: # old_v = sum(o[k] for o in resp_old if o.get(k)) # new_v = sum(n[k] for n in resp_new if n.get(k)) # if old_v != 0: # percent = new_v / old_v * 100 # elif old_v == new_v == 0: # percent = 100 # else: # percent = 0 # old_vs.append(old_v) # new_vs.append(new_v) # is_corr_ps.append(percent) # return {'old_values': old_vs, # 'new_values': new_vs, # 'is_correct_percentage': is_corr_ps} # # @pytest.mark.parametrize("market", markets) # @pytest.mark.parametrize("isrc", isrcs) # @pytest.mark.parametrize(("start_date", "end_date"), dates) # def test_track_playlists_summary_spotify(isrc, market, start_date, end_date): # endpoint = f'/dsp-api/consumer_analytics/v1/playlists-summary?isrc={isrc}&vendor=spotify&start_date={start_date}&end_date={end_date}&market={market}' # old_resp = api_client_old.get_response(endpoint, token_old, env=old_env)['body']['items'] # new_resp = api_client_new.get_response(endpoint, token_new, env=new_env)['body']['items'] # playlistId = 'playlistId' # isrc = 'isrc' # old_resp.sort(key=lambda x: x[playlistId]) # new_resp.sort(key=lambda x: x[playlistId]) # all_keys = ['playlists_count'] # old_values = [len(old_resp)] # new_values = [len(new_resp)] # is_correct_percentage = [len(new_resp)/len(old_resp)*100] # # for o, n in zip(old_resp, new_resp): # keys_local = OrderedDict(o).keys() # all_keys.extend(keys_local) # old_values.extend(OrderedDict(o).values()) # new_values.extend([n[k] for k in keys_local]) # for k in keys_local: # if k in [isrc, playlistId]: # perc = (o[k] == n[k]) # else: # if o[k] == 0: # perc = 0 # else: # perc = n[k] / o[k] * 100 # is_correct_percentage.append(perc) # # df = pandas.DataFrame({'old_values': old_values, # 'new_values': new_values, # 'is_correct_100%': is_correct_percentage # }, index=all_keys) # df.to_excel(f'playlists_summary_v1_spotify_{isrc}_{market}_{start_date}_{end_date}.xlsx') # @pytest.mark.parametrize("market", markets) # @pytest.mark.parametrize("playlist_id", playlist_ids) # @pytest.mark.parametrize(("start_date", "end_date"), dates) # def test_tracks_in_playlist_summary_2(market, start_date, end_date, playlist_id): # endpoint_isrcs = f"/sonytracks/analyze?playlisturi=spotify%3Aplaylist%3A{playlist_id}®ion={market}" # isrcs_resp = api_client_old.get_response(endpoint_isrcs, token_old, env=old_env)['body']['tracks'] # playlist_isrcs = [i['isrc'] for i in isrcs_resp] # isrcs_part = ''.join([f'isrc={i}&' for i in playlist_isrcs]) # log.info(f'isrcs part: {isrcs_part}') # # endpoint = "/spotify-consumer-analytics-compat/tracks-in-playlist-summary-2?{isrcs}" \ # "playlistId={playlist_id}&startDate={start_date}&endDate={end_date}&countryCode={market}" \ # .format(start_date=start_date, end_date=end_date, market=market, playlist_id=playlist_id, isrcs='isrc=USUM71900764&isrc=USUG11901473&isrc=USUS11900405&isrc=USUM71914355&isrc=USUG11902591&isrc=USAT21903631&isrc=USAT21904228&isrc=USHR11939300&isrc=USUG11902801&isrc=USUG11902971&isrc=USAT21904201&isrc=USUM71916703&isrc=USAT21904473&isrc=QM6MZ1990263&isrc=USUM71913854&isrc=GBUM71901769&isrc=USUG11901474&isrc=CAM371902594&isrc=GBKPL1955263&isrc=USRC11901990&') # log.info(f'full endpoint: {endpoint}') # old_resp = api_client_old.get_response(endpoint, token_old, env=old_env)['body'] # new_resp = api_client_new.get_response(endpoint, token_new, env=new_env)['body'] # old_resp.sort(key=lambda x: x['isrc']) # new_resp.sort(key=lambda x: x['isrc']) # old_values = [] # new_values = [] # is_correct_percentage = [] # isrc_key = 'isrc' # values = 'values' # keys = ['globalStreams', 'marketStreams'] # keys_sum = ['sum_globalStreams', 'sum_marketStreams'] # keys_all = ['isrc_count'] # old_values.append(len(old_resp)) # new_values.append(len(new_resp)) # is_correct_percentage.append(len(new_resp)/len(old_resp)*100) # for o, n in zip(old_resp, new_resp): # log.info(f'from old: {o}') # log.info(f'from new: {n}') # # keys_all.append(isrc_key) # keys_all.extend(keys) # keys_all.extend(keys_sum) # # old_values.append(o[isrc_key]) # new_values.append(n[isrc_key]) # is_correct_percentage.append(o[isrc_key] == n[isrc_key]) # # # values local and global # for k in keys: # old_values.append(o[k]) # new_values.append(n[k]) # if o[k] != 0: # perc = n[k] / o[k] * 100 # else: # perc = 0 # is_correct_percentage.append(perc) # # # sums for all days # for k in keys: # old_v = sum(ov[k] for ov in dict(o)[values]) # new_v = sum(n[k] for n in dict(n)[values]) # if old_v != 0: # perc = new_v / old_v * 100 # else: # perc = 0 # old_values.append(old_v) # new_values.append(new_v) # is_correct_percentage.append(perc) # # df = pandas.DataFrame({'old_values': old_values, # 'new_values': new_values, # 'is_correct_100%': is_correct_percentage # }, index=keys_all) # df.to_excel(f'tracks_in_playlist_summary_2_spotify_{market}_{start_date}_{end_date}_{playlist_id}.xlsx') # # @pytest.mark.parametrize("market", markets) # @pytest.mark.parametrize("isrc", isrcs) # @pytest.mark.parametrize("playlist_id", playlist_ids) # @pytest.mark.parametrize(("start_date", "end_date"), dates) # def test_spotify_track_in_playlist_2(market, start_date, end_date, isrc, playlist_id): # endpoint = f"/spotify-consumer-analytics-compat/track-in-playlist-2?isrc={isrc}&playlistId={playlist_id}&countryCode={market}&startDate={start_date}&endDate={end_date}" # #endpoint = f"/spotify-consumer-analytics-compat/track-in-playlist-2?isrc={isrc}&playlistId={playlist_id}&startDate={start_date}&endDate={end_date}" # old_resp = list(api_client_old.get_response(endpoint, token_old, env=old_env)['body']['items']) # new_resp = list(api_client_new.get_response(endpoint, token_new, env=new_env)['body']['items']) # old_resp.sort(key=lambda x: x['date']) # new_resp.sort(key=lambda x: x['date']) # # old_values = [] # new_values = [] # is_correct_percentage = [] # all_keys = ['days_count', 'SUM_globalStreams', 'SUM_localStreams'] # keys = ['date', 'globalStreams', 'localStreams'] # old_values.append(len(old_resp)) # new_values.append(len(new_resp)) # old_values.append(sum(o[keys[1]] for o in old_resp)) # old_values.append(sum(o[keys[2]] for o in old_resp)) # new_values.append(sum(n[keys[1]] for n in new_resp)) # new_values.append(sum(n[keys[2]] for n in new_resp)) # # is_correct_percentage.append(len(new_resp)/len(old_resp)*100) # # is_correct_percentage.append(sum(n[keys[1]] for n in new_resp)/sum(o[keys[1]] for o in old_resp) * 100) # # is_correct_percentage.append(sum(n[keys[2]] for n in new_resp)/sum(o[keys[2]] for o in old_resp) * 100) # is_correct_percentage.append(get_percentage(len(old_resp), len(new_resp))) # is_correct_percentage.append(get_percentage(sum(o[keys[1]] for o in old_resp), sum(n[keys[1]] for n in new_resp))) # is_correct_percentage.append(get_percentage(sum(o[keys[2]] for o in old_resp), sum(n[keys[2]] for n in new_resp))) # for o, n in zip(old_resp, new_resp): # log.info(f'from old: {o}') # log.info(f'from new: {n}') # # all_keys.extend(keys) # for k in keys: # old_values.append(o[k]) # new_values.append(n[k]) # if k == keys[0]: # perc = str(datetime.datetime.strptime(o[k], "%Y-%m-%d").date() - \ # datetime.datetime.strptime(n[k], "%Y-%m-%d").date()) + " days diff" # else: # if o[k] != 0: # perc = str(n[k] / o[k] * 100) + '%' # else: # perc = '0%' # is_correct_percentage.append(perc) # # df = pandas.DataFrame({'old_values': old_values, # 'new_values': new_values, # 'is_correct_100%/diff_days': is_correct_percentage # }, index=all_keys) # df.to_excel(f'track_in_playlist_2_spotify_{market}_{start_date}_{end_date}_{playlist_id}_{isrc}.xlsx') # # @pytest.mark.parametrize("market", markets) # @pytest.mark.parametrize("isrc", isrcs) # @pytest.mark.parametrize("playlist_id", playlist_ids) # @pytest.mark.parametrize(("start_date", "end_date"), dates) # def test_spotify_latest_date(market, start_date, end_date, isrc, playlist_id): # endpoint = "/spotify-consumer-analytics-compat/latest-date" # key = 'date' # old_resp = api_client_old.get_response(endpoint, token_old, env=old_env)['body'][key] # new_resp = api_client_new.get_response(endpoint, token_new, env=new_env)['body'][key] # # old_values = [old_resp] # new_values = [new_resp] # diff_days = [datetime.datetime.strptime(old_resp, "%Y-%m-%d").date() - # datetime.datetime.strptime(new_resp, "%Y-%m-%d").date()] # # df = pandas.DataFrame({'old_values': old_values, # 'new_values': new_values, # 'diff_up_to_1_day': diff_days # }, index=[key]) # df.to_excel('spotify_latest_date.xlsx') # # @pytest.mark.parametrize("market", markets) # @pytest.mark.parametrize("isrc", isrcs) # @pytest.mark.parametrize("playlist_id", playlist_ids) # @pytest.mark.parametrize(("start_date", "end_date"), dates) # def test_spotify_stream_latest_date(market, start_date, end_date, isrc, playlist_id): # endpoint = "/dsp-api/consumer_analytics/streams-latest-date" # key = 'date' # old_resp = api_client_old.get_response(endpoint, token_old, env=old_env)['body'][key] # new_resp = api_client_new.get_response(endpoint, token_new, env=new_env)['body'][key] # # old_values = [old_resp] # new_values = [new_resp] # diff_days = [datetime.datetime.strptime(old_resp, "%Y-%m-%d").date() - # datetime.datetime.strptime(new_resp, "%Y-%m-%d").date()] # # df = pandas.DataFrame({'old_values': old_values, # 'new_values': new_values, # 'diff_up_to_1_day': diff_days # }, index=[key]) # df.to_excel('spotify_streams_latest_date.xlsx') # @pytest.mark.parametrize("market", markets) # @pytest.mark.parametrize("isrc", isrcs) # def test_total_streams_count(market, isrc): # endpoint = f"/dsp-api/analytics/v1/total-streams-count?isrc={isrc}&market={market}&per_vendors=true" # old_resp = api_client_old.get_response(endpoint, token_old, env=old_env)['body'] # new_resp = api_client_new.get_response(endpoint, token_new, env=new_env)['body'] # # stream_keys = ['global_streams', 'market_streams'] # vendors_key = 'vendors' # vs_keys = ['spotify', 'apple', 'amazon'] # all_keys = stream_keys.copy() # old_values = [] # new_values = [] # is_correct_percentage = [] # for s in stream_keys: # log.info(f'all keys: {all_keys}') # old_values.append(old_resp[s]) # new_values.append(new_resp[s]) # is_correct_percentage.append(new_resp[s]/old_resp[s]*100) # for s in stream_keys: # for v in vs_keys: # all_keys.append(f'{v}_{s}') # old_values.append(old_resp[vendors_key][v][s]) # new_values.append(new_resp[vendors_key][v][s]) # is_correct_percentage.append(new_resp[vendors_key][v][s]/old_resp[vendors_key][v][s]*100) # # df = pandas.DataFrame({'old_values': old_values, # 'new_values': new_values, # 'is_correct_100%': is_correct_percentage # }, index=all_keys) # df.to_excel(f'total_streams_count_{market}_{isrc}.xlsx') # @pytest.mark.parametrize("market", markets) # def test_v1_starred_tracks_detailed(market): # endpoint = f"/apollo-api/v1/starred-tracks/detailed/?limit=20&offset=0&related=true&market={market}" # old_resp = api_client_old.get_response(endpoint, token_old, env=old_env)['body'] # new_resp = api_client_new.get_response(endpoint, token_new, env=new_env)['body'] # count_key = 'count' # isrc_key = 'isrc' # items_key = 'items' # streams_key = 'streams' # old_items = old_resp[items_key] # new_items = new_resp[items_key] # old_items.sort(key=lambda x: x[isrc_key]) # new_items.sort(key=lambda x: x[isrc_key]) # all_keys = ['is count_10', 'isrcs are the same'] # old_values = [old_resp[count_key], ' \n'.join([o[isrc_key] for o in old_resp[items_key]])] # new_values = [new_resp[count_key], ' \n'.join([n[isrc_key] for n in new_resp[items_key]])] # is_correct_percentage = [new_resp[count_key] / old_resp[count_key] * 100, # [o[isrc_key] for o in old_resp[items_key]] == [n[isrc_key] for n in new_resp[items_key]]] # # for o, n, in zip(old_items, new_items): # all_keys.append(o[isrc_key]) # old_values.append(o[streams_key]) # new_values.append(n[streams_key]) # if o[streams_key] != 0: # perc = n[streams_key] / o[streams_key] * 100 # else: # perc = 0 # is_correct_percentage.append(perc) # # df = pandas.DataFrame({'old_values': old_values, # 'new_values': new_values, # 'is_correct_100%': is_correct_percentage # }, index=all_keys) # df.to_excel(f'v1_starred_tracks_detailed_{market}.xlsx') # @pytest.mark.parametrize("isrcs", isrcs) # def test_analytics_v1_streams_dates(isrcs): # endpoint = f"/dsp-api/analytics/v1/streams-dates?isrc={'%2C'.join(list(isrcs))}&include_last_dates=true" # log.info(f'endpoint: {endpoint}') # old_resp = api_client_old.get_response(endpoint, token_old, env=old_env)['body'] # new_resp = api_client_new.get_response(endpoint, token_new, env=new_env)['body'] # first_stream_date = 'first_stream_date' # last_stream_date = 'last_stream_date' # vendors = ['spotify', 'apple', 'amazon'] # all_keys = [] # old_values = [] # new_values = [] # diff_days = [] # # for v in vendors: # for i in isrcs: # all_keys.append(f'{v}_{i}_{first_stream_date}') # old_values.append(old_resp[first_stream_date][v][i]) # new_values.append(new_resp[first_stream_date][v][i]) # diff_days.append(datetime.datetime.strptime(old_resp[first_stream_date][v][i], "%Y-%m-%d").date() - # datetime.datetime.strptime(new_resp[first_stream_date][v][i], "%Y-%m-%d").date()) # # for v in vendors: # all_keys.append(f'{v}_{last_stream_date}') # new_values.append(new_resp[last_stream_date][v]) # if v != vendors[0]: # old_values.append('MISSED') # diff_days.append('MISSED') # else: # old_values.append(old_resp[last_stream_date][v]) # diff_days.append(datetime.datetime.strptime(old_resp[last_stream_date][v], "%Y-%m-%d").date() - # datetime.datetime.strptime(new_resp[last_stream_date][v], "%Y-%m-%d").date()) # # df = pandas.DataFrame({'old_values': old_values, # 'new_values': new_values, # 'diff_days': diff_days # }, index=all_keys) # df.to_excel(f"analytics_v1_streams_dates_{'_'.join(list(isrcs))}.xlsx") # @pytest.mark.parametrize("market", markets) # @pytest.mark.parametrize("playlistId", playlist_ids) # @pytest.mark.parametrize("isrc", isrcs) # @pytest.mark.parametrize(("start_date", "end_date"), dates) # def test_track_in_container_apple(isrc, market, playlistId, start_date, end_date): # endpoint = f'/apple-consumer-analytics/track-in-container?containerId={playlistId}&startDate={start_date}&endDate={end_date}&isrc={isrc}&countryCode={market}' # # get old response # old = api_client_old.get_response(endpoint, token_old, env=old_env)['body']['items'] # # get new response # new = api_client_new.get_response(endpoint, token_new, env=new_env)['body']['items'] # # get keys # keys = list(dict(old[0]).keys()) # # log.info(f'endpoint: {endpoint}') # log.info(f'response old: {old}') # log.info(f'response new: {new}') # log.info(f'response keys: {keys}') # # old_endpoint_values = [] # new_endpoint_values = [] # diffs = [] # keys_values = ['days_count', 'SUM_globalStreams', 'SUM_localStreams'] # sum_global_old = sum(o['globalStreams'] for o in old) # sum_local_old = sum(o['localStreams'] for o in old) # sum_global_new = sum(n['globalStreams'] for n in new) # sum_local_new = sum(n['localStreams'] for n in new) # old_endpoint_values.extend([len(old), sum_global_old, sum_local_old]) # new_endpoint_values.extend([len(new), sum_global_new, sum_local_new]) # diffs.extend([len(old_endpoint_values) - len(new_endpoint_values), # get_percentage(sum_global_old, sum_global_new), # get_percentage(sum_local_old, sum_local_new)]) # # keys_values.extend(keys * len(old)) # # for i in old: # old_endpoint_values.extend([old[list(old).index(i)]['date'], old[list(old).index(i)]['localStreams'], old[list(old).index(i)]['globalStreams']]) # # for j in new: # new_endpoint_values.extend([new[list(new).index(j)]['date'], new[list(new).index(j)]['localStreams'], new[list(new).index(j)]['globalStreams']]) # # for d in old: # diffs.extend([datetime.datetime.strptime(old[list(old).index(d)]['date'], "%Y-%m-%d").date() - # datetime.datetime.strptime(new[list(old).index(d)]['date'], "%Y-%m-%d").date(), # get_percentage(old[list(old).index(d)]['localStreams'], new[list(old).index(d)]['localStreams']), # get_percentage(old[list(old).index(d)]['globalStreams'], new[list(old).index(d)]['globalStreams']) # ]) # # df = pandas.DataFrame({'old_values': old_endpoint_values, # 'new_values': new_endpoint_values, # 'is_correct_100%/diff_days': diffs}, # index=keys_values) # # df.to_excel(f'track_in_container_apple_{playlistId}_{start_date}_{end_date}_{isrc}_{market}.xlsx') # # def get_percentage(old, new): # if old != 0: # perc = new / old * 100 # else: # perc = 0 # # return perc # @pytest.mark.parametrize("market", markets) # @pytest.mark.parametrize("isrc", isrcs) # def test_filtr_track_playlists_spotify(isrc, market): # endpoint = f'/gate-api/playlists/spotify/by-track/?isrc={isrc}&market={market}' # # get old response # old = api_client_old.get_response(endpoint, token_old, env=old_env)['body']['items'] # # get new response # new = api_client_new.get_response(endpoint, token_new, env=new_env)['body']['items'] # # keys # spotifyLink = 'spotifyLink' # keys = ['marketStreams7Days', 'globalStreams7Days'] # old.sort(key=lambda x: x[spotifyLink]) # new.sort(key=lambda x: x[spotifyLink]) # # log.info(f'endpoint: {endpoint}') # log.info(f'response old: {old}') # log.info(f'response new: {new}') # # keys_values = [] # old_endpoint_values = [] # new_endpoint_values = [] # diffs = [] # # for o, n in zip(old, new): # for k in keys: # keys_values.append(f"{str(o[spotifyLink]).split(':')[2]}_{k}") # old_endpoint_values.append(o[k]) # new_endpoint_values.append(n[k]) # diffs.append(get_percentage(o[k], n[k])) # log.info(f'values: {o[spotifyLink]}_{k}, \n {o[k]}, \n {n[k]}') # # df = pandas.DataFrame({'old_values': old_endpoint_values, # 'new_values': new_endpoint_values, # 'is_correct_100%': diffs}, # index=keys_values) # # df.to_excel(f'filtr_track_playlists_{isrc}_{market}.xlsx')