# 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() # # 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 # # # chart_date = 'chart_date' # chart_available = 'chart_available' # json_keys_dashboard = ['artist', 'artists', 'id', 'image_url', 'is_new', 'is_re_enter', 'is_sony', 'isrc', 'name', 'position', 'trend'] # gtp_keys = ['end_date', 'is_entered', 'isrc', 'market', 'start_date', 'trend'] # # markets = ['global', 'au'] # vendors = ['apple', 'spotify'] # spotify_date = "&date=2021-09-20" # # @pytest.mark.parametrize("vendor", vendors) # @pytest.mark.parametrize("market", markets) # def test_dashboard_top(market, vendor): # # old_endpoint = f'/apollo-api/v1/charts/top/?vendor={vendor}&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={vendor}&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&end=10&date=2021-09-20' # # get old response # if vendor == "apple": # old_endpoint = old_endpoint.replace(spotify_date, "") # new_endpoint = new_endpoint.replace(spotify_date, "") # 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, chart_date] + json_keys_dashboard * len(old_resp_tracks) # 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('') # old_endpoint_values.append(old_resp[chart_date]) # new_endpoint_values.append(new_resp[chart_date]) # diffs.append(new_resp[chart_date] == old_resp[chart_date]) # # for t in old_resp_tracks: # old_endpoint_values.extend([t[k] for k in json_keys_dashboard]) # for t in new_resp_tracks: # new_endpoint_values.extend([t[k] for k in json_keys_dashboard]) # for to, tn in zip(old_resp_tracks, new_resp_tracks): # diffs.extend([to[k] == tn[k] for k in json_keys_dashboard]) # # 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'charts_top_{market}_{vendor}.xlsx') # @pytest.mark.parametrize("vendor", vendors) # @pytest.mark.parametrize("market", markets) # def test_dashboard_out(market, vendor): # old_endpoint = f'/apollo-api/v1/charts/out/?vendor={vendor}&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={vendor}&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&end=10&date=2021-09-20' # # get response # if vendor == "apple": # old_endpoint = old_endpoint.replace(spotify_date, "") # new_endpoint = new_endpoint.replace(spotify_date, "") # 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, chart_date] + json_keys_dashboard * len(old_resp_tracks) # 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('') # # old_endpoint_values.append(old_resp[chart_date]) # new_endpoint_values.append(new_resp[chart_date]) # diffs.append(new_resp[chart_date] == old_resp[chart_date]) # # for t in old_resp_tracks: # old_endpoint_values.extend([t[k] for k in json_keys_dashboard]) # for t in new_resp_tracks: # new_endpoint_values.extend([t[k] for k in json_keys_dashboard]) # for to, tn in zip(old_resp_tracks, new_resp_tracks): # diffs.extend([to[k] == tn[k] for k in json_keys_dashboard]) # # 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'charts_out_{market}_{vendor}.xlsx') # # # markets = ['_gl', 'au'] # # # @pytest.mark.parametrize("market", markets) # def test_dashboard_gtp_charts(market): # old_endpoint = f'/apollo-api/gtp/charts/?isrc=GBCEN2100136&date_shift=4&market={market}&end=2020-11-20' # if market == '_gl': # market = 'worldwide' # new_endpoint = f'/gate-api/gtp/charts/?isrc=GBCEN2100136&date_shift=4&market={market}&end=2020-11-20' # 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 = sorted(list(old_resp['data']), key=itemgetter('isrc')) # new_resp = api_client_old.get_response(endpoint=new_endpoint, user=CoreConfig.USER_EMAIL, env=new_env)['body'] # new_resp_tracks = sorted(list(new_resp['data']), key=itemgetter('isrc')) # # indexes_all = [request_url, chart_available] # for t in old_resp_tracks: # indexes_all.extend([k for k in gtp_keys if k in t.keys()]) # # 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('') # old_endpoint_values.append(old_resp[chart_available]) # new_endpoint_values.append(new_resp[chart_available]) # diffs.append(new_resp[chart_available] == old_resp[chart_available]) # # for t in old_resp_tracks: # old_endpoint_values.extend([t[k] for k in gtp_keys if k in t.keys()]) # for t in new_resp_tracks: # new_endpoint_values.extend([t[k] for k in gtp_keys if k in t.keys()]) # for to, tn in zip(old_resp_tracks, new_resp_tracks): # diffs.extend([to[k] == tn[k] for k in gtp_keys if k in to.keys()]) # # 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'gtp_charts_{market}.xlsx') # # # @pytest.mark.parametrize("vendor", vendors) # @pytest.mark.parametrize("market", markets) # def test_dashboard_moves(market, vendor): # old_endpoint = f'/apollo-api/v1/charts/moves/?market={market}&vendor={vendor}&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={vendor}&fields=is_sony&fields=is_re_enter&fields=trends&offset=0&limit=20&order_by=moves&date=2021-09-20' # if vendor == "apple": # old_endpoint = old_endpoint.replace(spotify_date, "") # new_endpoint = new_endpoint.replace(spotify_date, "") # # 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['position'])) # 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['position'])) # # indexes_all = [request_url, chart_date, "count", "previous"] # for t in old_resp_tracks: # indexes_all.extend([k for k in json_keys_dashboard if k in t.keys()]) # # 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 i in [chart_date, "count", "previous"]: # old_endpoint_values.append(old_resp[i]) # new_endpoint_values.append(new_resp[i]) # diffs.append(new_resp[i] == old_resp[i]) # # for to, tn in zip(old_resp_tracks, new_resp_tracks): # old_endpoint_values.extend([to[k] for k in json_keys_dashboard if k in t.keys()]) # new_endpoint_values.extend([tn[k] for k in json_keys_dashboard if k in t.keys()]) # diffs.extend([to[k] == tn[k] for k in json_keys_dashboard if k in to.keys()]) # # 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'charts_moves_{market}_{vendor}.xlsx') # # # @pytest.mark.parametrize("vendor", vendors) # @pytest.mark.parametrize("market", markets) # def test_dashboard_additions(market, vendor): # old_endpoint = f'/apollo-api/v1/charts/additions/?vendor={vendor}&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={vendor}&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&date=2021-09-20' # if vendor == "apple": # old_endpoint = old_endpoint.replace(spotify_date, "") # new_endpoint = new_endpoint.replace(spotify_date, "") # # 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, chart_date, "tracks_count"] # for t in old_resp_tracks: # indexes_all.extend([k for k in json_keys_dashboard if k in t.keys()]) # # 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('') # # old_endpoint_values.append(old_resp[chart_date]) # new_endpoint_values.append(new_resp[chart_date]) # diffs.append(new_resp[chart_date] == old_resp[chart_date]) # # old_endpoint_values.append(len(old_resp_tracks)) # new_endpoint_values.append(len(new_resp_tracks)) # diffs.append(len(old_resp_tracks) == len(new_resp_tracks)) # # for t in old_resp_tracks: # old_endpoint_values.extend([t[k] for k in json_keys_dashboard if k in t.keys()]) # for t in new_resp_tracks: # new_endpoint_values.extend([t[k] for k in json_keys_dashboard if k in t.keys()]) # for to, tn in zip(old_resp_tracks, new_resp_tracks): # diffs.extend([to[k] == tn[k] for k in json_keys_dashboard if k in to.keys()]) # # 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'charts_additions_{market}_{vendor}.xlsx') # # # @pytest.mark.parametrize("vendor", vendors) # @pytest.mark.parametrize("market", markets) # def test_dashboard_removals(market, vendor): # old_endpoint = f'/apollo-api/v1/charts/removals/?vendor={vendor}&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={vendor}&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&date=2021-09-20' # if vendor == "apple": # old_endpoint = old_endpoint.replace(spotify_date, "") # new_endpoint = new_endpoint.replace(spotify_date, "") # # 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, chart_date, "tracks_count"] # for t in old_resp_tracks: # indexes_all.extend([k for k in json_keys_dashboard if k in t.keys()]) # # 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('') # # old_endpoint_values.append(old_resp[chart_date]) # new_endpoint_values.append(new_resp[chart_date]) # diffs.append(new_resp[chart_date] == old_resp[chart_date]) # # old_endpoint_values.append(len(old_resp_tracks)) # new_endpoint_values.append(len(new_resp_tracks)) # diffs.append(len(old_resp_tracks) == len(new_resp_tracks)) # # for t in old_resp_tracks: # old_endpoint_values.extend([t[k] for k in json_keys_dashboard if k in t.keys()]) # for t in new_resp_tracks: # new_endpoint_values.extend([t[k] for k in json_keys_dashboard if k in t.keys()]) # for to, tn in zip(old_resp_tracks, new_resp_tracks): # diffs.extend([to[k] == tn[k] for k in json_keys_dashboard if k in to.keys()]) # # 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'charts_removals_{market}_{vendor}.xlsx') # # @pytest.mark.parametrize("vendor", vendors) # @pytest.mark.parametrize("market", markets) # def test_dashboard_bu(market, vendor): # old_endpoint = f'/apollo-api/v1/charts/top/?start=51&end=60&vendor={vendor}&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={vendor}&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&date=2021-09-20' # if vendor == "apple": # old_endpoint = old_endpoint.replace(spotify_date, "") # new_endpoint = new_endpoint.replace(spotify_date, "") # # 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, chart_date, "tracks_count"] # for t in old_resp_tracks: # indexes_all.extend([k for k in json_keys_dashboard if k in t.keys()]) # # 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('') # # old_endpoint_values.append(old_resp[chart_date]) # new_endpoint_values.append(new_resp[chart_date]) # diffs.append(new_resp[chart_date] == old_resp[chart_date]) # # old_endpoint_values.append(len(old_resp_tracks)) # new_endpoint_values.append(len(new_resp_tracks)) # diffs.append(len(old_resp_tracks) == len(new_resp_tracks)) # # for t in old_resp_tracks: # old_endpoint_values.extend([t[k] for k in json_keys_dashboard if k in t.keys()]) # for t in new_resp_tracks: # new_endpoint_values.extend([t[k] for k in json_keys_dashboard if k in t.keys()]) # for to, tn in zip(old_resp_tracks, new_resp_tracks): # diffs.extend([to[k] == tn[k] for k in json_keys_dashboard if k in to.keys()]) # # 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'charts_bubbling_under_{market}_{vendor}.xlsx')