# import collections # import logging # from operator import itemgetter # # import pandas # import pytest # from ui_automation_framework.utils import logger as cl # from ui_automation_framework.utils import CoreConfig # # from app.api_client import ApiClient # # log = cl.Logger(logging.DEBUG) # # new_env = 'https://test-proxy.apollo.stream' # old_env = 'https://proxy.apollo.stream' # api_client_old = ApiClient(host_endpoint=old_env) # api_client_new = ApiClient(host_endpoint=new_env) # token_new = api_client_new.authorize() # prod_user = 'anna.andreyeva@dataart.com' # # request_url = 'Request URL' # # # # log.info('New token:' + token_new) # # def f(dat, c='red'): # colors = [] # for i in dat: # if i: # c = 'green' # elif i == '': # c = 'white' # else: # c = 'red' # colors.append(f'background-color: {c}') # return colors # # # def get_streams_per_country(isrc, market): # streams_per_country = api_client_old.get_response( # endpoint=f'/dsp-api/consumer_analytics/v1/track-per-country?isrc={isrc}&start_date=2021-11-24&end_date=2021-12-21&vendor=spotify&streams_only=false&combine_isrc=true', # user=CoreConfig.USER_EMAIL, env=new_env)['body'] # streams = 0 # for c in streams_per_country: # if market == 'worldwide': # market = '_gl' # if c['countryCode'] == market: # streams = sum(d['streams'] for d in c['data']) # return streams # # # json_keys_dashboard = ['id', 'is_sony', 'isrc', 'streams_per_country'] # gtp_keys = ['isrc'] # # markets = ['global', 'au'] # # # @pytest.mark.parametrize("market", markets) # def test_dashboard_e_1(market): # old_endpoint = f'/apollo-api/v1/charts/top/?vendor=spotify&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&end=10&date=2021-09-20' # if market == 'global': # market = 'worldwide' # new_endpoint = f'/gate-api/charts/top/?vendor=spotify&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&end=10&date=2021-09-20' # # get old response # old_resp = api_client_old.get_response(endpoint=old_endpoint, user=prod_user, env=old_env, token=token_old)['body'] # old_resp_tracks = old_resp['tracks'] # new_resp = api_client_old.get_response(endpoint=new_endpoint, user=CoreConfig.USER_EMAIL, env=new_env)['body'] # new_resp_tracks = new_resp['tracks'] # # indexes_all = [request_url] # log.info(f'indexes all: {indexes_all}') # old_endpoint_values = [] # new_endpoint_values = [] # diffs = [] # # old_endpoint_values.append(old_endpoint) # new_endpoint_values.append(new_endpoint) # diffs.append('') # # for to, tn in zip(old_resp_tracks, new_resp_tracks): # for k in json_keys_dashboard: # if to['is_sony'] != tn['is_sony'] and to['isrc'] not in old_endpoint_values: # indexes_all.extend(json_keys_dashboard) # diffs.extend([to[k] == tn[k] for k in json_keys_dashboard[:-1]] + ['']) # old_endpoint_values.extend([to[k] for k in json_keys_dashboard[:-1]] + [get_streams_per_country(to['isrc'], market)]) # new_endpoint_values.extend([tn[k] for k in json_keys_dashboard[:-1]] + ['']) # # df = pandas.DataFrame({ # 'IS CORRECT': diffs, # 'OLD API VALUE': old_endpoint_values, # 'NEW API VALUE': new_endpoint_values, # }, # index=indexes_all) # # df.reset_index().style.apply(f, subset=['IS CORRECT']) \ # .to_excel(f'IS_SONY_charts_top_{market}.xlsx') # # # @pytest.mark.parametrize("market", markets) # def test_dashboard_e_2(market): # old_endpoint = f'/apollo-api/v1/charts/out/?vendor=spotify&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&end=10&date=2021-09-20' # if market == 'global': # market = 'worldwide' # new_endpoint = f'/gate-api/charts/out/?vendor=spotify&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&end=10&date=2021-09-20' # # get response # old_resp = api_client_old.get_response(endpoint=old_endpoint, user=prod_user, env=old_env, token=token_old)['body'] # old_resp_tracks = old_resp['tracks'] # new_resp = api_client_old.get_response(endpoint=new_endpoint, user=CoreConfig.USER_EMAIL, env=new_env)['body'] # new_resp_tracks = new_resp['tracks'] # # log.info(f'old resp: {old_resp}') # log.info(f'new resp: {new_resp}') # # indexes_all = [request_url] # log.info(f'indexes all: {indexes_all}') # old_endpoint_values = [] # new_endpoint_values = [] # diffs = [] # # old_endpoint_values.append(old_endpoint) # new_endpoint_values.append(new_endpoint) # diffs.append('') # # for to, tn in zip(old_resp_tracks, new_resp_tracks): # for k in json_keys_dashboard: # if to['is_sony'] != tn['is_sony'] and to['isrc'] not in old_endpoint_values: # indexes_all.extend(json_keys_dashboard) # diffs.extend([to[k] == tn[k] for k in json_keys_dashboard[:-1]] + ['']) # old_endpoint_values.extend([to[k] for k in json_keys_dashboard[:-1]] + [get_streams_per_country(to['isrc'], market)]) # new_endpoint_values.extend([tn[k] for k in json_keys_dashboard[:-1]] + ['']) # # df = pandas.DataFrame({ # 'IS CORRECT': diffs, # 'OLD API VALUE': old_endpoint_values, # 'NEW API VALUE': new_endpoint_values, # }, # index=indexes_all) # # df.reset_index().style.apply(f, subset=['IS CORRECT']) \ # .to_excel(f'IS_SONY_charts_out_{market}.xlsx') # # # markets = ['_gl', 'au'] # # # @pytest.mark.parametrize("market", markets) # def test_dashboard_e_4(market): # old_endpoint = f'/apollo-api/v1/charts/moves/?market={market}&vendor=spotify&fields=is_sony&fields=is_re_enter&fields=trends&offset=0&limit=20&order_by=moves&date=2021-09-20' # if market == '_gl': # market = 'worldwide' # new_endpoint = f'/gate-api/charts/moves/?market={market}&vendor=spotify&fields=is_sony&fields=is_re_enter&fields=trends&offset=0&limit=20&order_by=moves&date=2021-09-20' # # get response # old_resp = api_client_old.get_response(endpoint=old_endpoint, user=CoreConfig.USER_EMAIL, env=old_env, token=token_old)['body'] # log.info(f'old resp: {old_resp}') # old_resp_tracks = old_resp['tracks'] # # old_resp_tracks = sorted(old_resp['tracks'], key=lambda d: (d['id'], d['isrc'])) # new_resp = api_client_old.get_response(endpoint=new_endpoint, user=CoreConfig.USER_EMAIL, env=new_env)['body'] # new_resp_tracks = new_resp['tracks'] # # new_resp_tracks = sorted(new_resp['tracks'], key=lambda d: (d['id'], d['isrc'])) # # indexes_all = [request_url] # # log.info(f'indexes all: {indexes_all}') # old_endpoint_values = [] # new_endpoint_values = [] # diffs = [] # # old_endpoint_values.append(old_endpoint) # new_endpoint_values.append(new_endpoint) # diffs.append('') # # for to, tn in zip(old_resp_tracks, new_resp_tracks): # for k in json_keys_dashboard: # if to['is_sony'] != tn['is_sony'] and to['isrc'] not in old_endpoint_values: # indexes_all.extend(json_keys_dashboard) # diffs.extend([to[k] == tn[k] for k in json_keys_dashboard[:-1]] + ['']) # old_endpoint_values.extend([to[k] for k in json_keys_dashboard[:-1]] + [get_streams_per_country(to['isrc'], market)]) # new_endpoint_values.extend([tn[k] for k in json_keys_dashboard[:-1]] + ['']) # # df = pandas.DataFrame({ # 'IS CORRECT': diffs, # 'OLD API VALUE': old_endpoint_values, # 'NEW API VALUE': new_endpoint_values, # }, # index=indexes_all) # # df.reset_index().style.apply(f, subset=['IS CORRECT']) \ # .to_excel(f'IS_SONY_charts_moves_{market}.xlsx') # # # @pytest.mark.parametrize("market", markets) # def test_dashboard_e_5(market): # old_endpoint = f'/apollo-api/v1/charts/additions/?vendor=spotify&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&date=2021-09-20' # if market == '_gl': # market = 'worldwide' # new_endpoint = f'/gate-api/charts/additions/?vendor=spotify&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&date=2021-09-20' # # get response # old_resp = api_client_old.get_response(endpoint=old_endpoint, user=CoreConfig.USER_EMAIL, env=old_env, token=token_old)['body'] # log.info(f'old resp: {old_resp}') # old_resp_tracks = old_resp['tracks'] # new_resp = api_client_old.get_response(endpoint=new_endpoint, user=CoreConfig.USER_EMAIL, env=new_env)['body'] # new_resp_tracks = new_resp['tracks'] # # indexes_all = [request_url] # # log.info(f'indexes all: {indexes_all}') # old_endpoint_values = [] # new_endpoint_values = [] # diffs = [] # # old_endpoint_values.append(old_endpoint) # new_endpoint_values.append(new_endpoint) # diffs.append('') # # for to, tn in zip(old_resp_tracks, new_resp_tracks): # for k in json_keys_dashboard: # if to['is_sony'] != tn['is_sony'] and to['isrc'] not in old_endpoint_values: # indexes_all.extend(json_keys_dashboard) # diffs.extend([to[k] == tn[k] for k in json_keys_dashboard[:-1]] + ['']) # old_endpoint_values.extend([to[k] for k in json_keys_dashboard[:-1]] + [get_streams_per_country(to['isrc'], market)]) # new_endpoint_values.extend([tn[k] for k in json_keys_dashboard[:-1]] + ['']) # # df = pandas.DataFrame({ # 'IS CORRECT': diffs, # 'OLD API VALUE': old_endpoint_values, # 'NEW API VALUE': new_endpoint_values, # }, # index=indexes_all) # # df.reset_index().style.apply(f, subset=['IS CORRECT']) \ # .to_excel(f'IS_SONY_charts_additions_{market}.xlsx') # # # @pytest.mark.parametrize("market", markets) # def test_dashboard_e_6(market): # old_endpoint = f'/apollo-api/v1/charts/removals/?vendor=spotify&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&date=2021-09-20' # if market == '_gl': # market = 'worldwide' # new_endpoint = f'/gate-api/charts/removals/?vendor=spotify&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&date=2021-09-20' # # get response # old_resp = api_client_old.get_response(endpoint=old_endpoint, user=CoreConfig.USER_EMAIL, env=old_env, token=token_old)['body'] # log.info(f'old resp: {old_resp}') # old_resp_tracks = old_resp['tracks'] # new_resp = api_client_old.get_response(endpoint=new_endpoint, user=CoreConfig.USER_EMAIL, env=new_env)['body'] # new_resp_tracks = new_resp['tracks'] # # indexes_all = [request_url] # # log.info(f'indexes all: {indexes_all}') # old_endpoint_values = [] # new_endpoint_values = [] # diffs = [] # # old_endpoint_values.append(old_endpoint) # new_endpoint_values.append(new_endpoint) # diffs.append('') # # for to, tn in zip(old_resp_tracks, new_resp_tracks): # for k in json_keys_dashboard: # if to['is_sony'] != tn['is_sony'] and to['isrc'] not in old_endpoint_values: # indexes_all.extend(json_keys_dashboard) # diffs.extend([to[k] == tn[k] for k in json_keys_dashboard[:-1]] + ['']) # old_endpoint_values.extend([to[k] for k in json_keys_dashboard[:-1]] + [get_streams_per_country(to['isrc'], market)]) # new_endpoint_values.extend([tn[k] for k in json_keys_dashboard[:-1]] + ['']) # # df = pandas.DataFrame({ # 'IS CORRECT': diffs, # 'OLD API VALUE': old_endpoint_values, # 'NEW API VALUE': new_endpoint_values, # }, # index=indexes_all) # # df.reset_index().style.apply(f, subset=['IS CORRECT']) \ # .to_excel(f'IS_SONY_charts_removals_{market}.xlsx') # # # @pytest.mark.parametrize("market", markets) # def test_dashboard_e_7(market): # old_endpoint = f'/apollo-api/v1/charts/top/?start=51&end=60&vendor=spotify&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&date=2021-09-20' # if market == '_gl': # market = 'worldwide' # new_endpoint = f'/gate-api/charts/top/?start=51&end=60&vendor=spotify&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&date=2021-09-20' # # get response # old_resp = api_client_old.get_response(endpoint=old_endpoint, user=CoreConfig.USER_EMAIL, env=old_env, token=token_old)['body'] # log.info(f'old resp: {old_resp}') # old_resp_tracks = old_resp['tracks'] # new_resp = api_client_old.get_response(endpoint=new_endpoint, user=CoreConfig.USER_EMAIL, env=new_env)['body'] # new_resp_tracks = new_resp['tracks'] # # indexes_all = [request_url] # # log.info(f'indexes all: {indexes_all}') # old_endpoint_values = [] # new_endpoint_values = [] # diffs = [] # # old_endpoint_values.append(old_endpoint) # new_endpoint_values.append(new_endpoint) # diffs.append('') # # for to, tn in zip(old_resp_tracks, new_resp_tracks): # for k in json_keys_dashboard: # if to['is_sony'] != tn['is_sony'] and to['isrc'] not in old_endpoint_values: # indexes_all.extend(json_keys_dashboard) # diffs.extend([to[k] == tn[k] for k in json_keys_dashboard[:-1]] + ['']) # old_endpoint_values.extend([to[k] for k in json_keys_dashboard[:-1]] + [get_streams_per_country(to['isrc'], market)]) # new_endpoint_values.extend([tn[k] for k in json_keys_dashboard[:-1]] + ['']) # # df = pandas.DataFrame({ # 'IS CORRECT': diffs, # 'OLD API VALUE': old_endpoint_values, # 'NEW API VALUE': new_endpoint_values, # }, # index=indexes_all) # # df.reset_index().style.apply(f, subset=['IS CORRECT']) \ # .to_excel(f'IS_SONY_charts_bubbling_under_{market}.xlsx')