# import logging # # import pandas # from parameterized import parameterized # 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' # # # def f(dat, c='red'): # colors = [] # for i in dat: # if i: # c = 'green' # else: # c = 'red' # colors.append(f'background-color: {c}') # return colors # # # chart_date = 'chart_date' # json_keys_top_charts = ['streams', 'chart_date', 'is_starred_track', 'image_url', 'position', 'isrc', 'change', 'trend', 'id', # 'name', 'is_new', 'is_sony', 'is_re_enter', 'artist', 'artists'] # json_keys_top_charts_apple = ['is_new', 'artists', 'image_url', 'isrc', 'position', 'name', 'trend', 'id', 'is_starred_track', # 'is_sony', 'is_re_enter', 'chart_date', 'artist', 'change'] # json_keys_top_charts_date_range = ['max_date', 'min_date'] # request_url = 'Request URL' # @parameterized.expand( # [ # ('us', '2021-09-20', 'daily'), # ('global', '2021-09-17', 'daily'), # ('se', '2021-09-16', 'weekly'), # ('us', '2021-09-02', 'weekly'), # ('de', '2020-01-31', 'daily'), # ('pl', '2020-01-30', 'weekly'), # ('de', '2019-05-17', 'daily'), # ('es', '2020-03-05', 'weekly'), # ('jp', '2020-01-16', 'daily'), # ('es', '2020-01-23', 'weekly'), # ('cl', '2019-10-26', 'daily'), # ] # ) # def test_top_charts_spotify(market, date, chart_type): # old_endpoint = f'/apollo-api/v1/charts/top/?vendor=spotify&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&date={date}' \ # f'&type={chart_type}&end=200' # 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&date={date}' \ # f'&type={chart_type}&end=200' # # 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'] # log.info(f"Old resp: {len(old_resp_tracks)}") # # get new response # 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"New resp: {len(new_resp_tracks)}") # isrcs_old = [i['isrc'] for i in old_resp_tracks] # isrcs_new = [i['isrc'] for i in new_resp_tracks] # log.info(f'old isrc: {isrcs_old}') # log.info(f'new isrc: {isrcs_new}') # isrcs_intersected = list(set(isrcs_new).intersection(isrcs_old)) # new_resp_tracks = [i for i in new_resp_tracks if i['isrc'] in isrcs_intersected] # old_resp_tracks = [i for i in old_resp_tracks if i['isrc'] in isrcs_intersected] # new_resp_tracks = sorted(new_resp_tracks, key=lambda d: d['isrc']) # old_resp_tracks = sorted(old_resp_tracks, key=lambda d: d['isrc']) # log.info(f"\n\n\n\n\n\n\n\n\new resp: {new_resp_tracks}\n\n\n\n\n\n\n\n\'") # log.info(f"\n\n\n\n\n\n\n\n\old resp: {old_resp_tracks}\n\n\n\n\n\n\n\n\'") # log.info(f'old len: {len(old_resp_tracks)}' # f'\nnew len: {len(new_resp_tracks)}' # f'\nnew isrc: {list({x for x in new_resp_tracks if new_resp_tracks.count(x) > 1})}' # f'\ninter len: {len(isrcs_intersected)}' # f'\ndiff_old: {[item for item in isrcs_old if item not in isrcs_new]}' # f'\ndiff_new: {[item for item in isrcs_new if item not in isrcs_old]}') # # indexes_all = [request_url] + [chart_date] + json_keys_top_charts_apple * len(isrcs_intersected) # 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_top_charts_apple]) # for t in new_resp_tracks: # new_endpoint_values.extend([t[k] for k in json_keys_top_charts_apple]) # for to, tn in zip(old_resp_tracks, new_resp_tracks): # diffs.extend([to[k] == tn[k] for k in json_keys_top_charts_apple]) # # 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'{date}_spotify_{market}_top_chart_{chart_type}.xlsx') # @parameterized.expand( # [#'us', 'global', 'de', 'ca', # 'se', # #'pl', 'es', 'jp', 'cl' # ] # ) # def test_top_charts_apple(market): # old_endpoint = f'/apollo-api/v1/charts/top/?vendor=apple&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&end=100' # if market == 'global': # market = 'worldwide' # new_endpoint = f'/gate-api/charts/top/?vendor=apple&market={market}&fields=is_sony&fields=is_re_enter&fields=trends&end=100' # # 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'] # log.info(f"Old resp: {len(old_resp_tracks)}") # log.info(f"Apple keys: {list(old_resp['tracks'][0].keys())}") # # get new response # 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"New resp: {len(new_resp_tracks)}") # isrcs_old = [i['isrc'] for i in old_resp_tracks] # isrcs_new = [i['isrc'] for i in new_resp_tracks] # log.info(f'old isrc: {isrcs_old}') # log.info(f'new isrc: {isrcs_new}') # isrcs_intersected = list(set(isrcs_new).intersection(isrcs_old)) # new_resp_tracks = [i for i in new_resp_tracks if i['isrc'] in isrcs_intersected] # old_resp_tracks = [i for i in old_resp_tracks if i['isrc'] in isrcs_intersected] # new_resp_tracks = sorted(new_resp_tracks, key=lambda d: d['isrc']) # old_resp_tracks = sorted(old_resp_tracks, key=lambda d: d['isrc']) # log.info(f"\n\n\n\n\n\n\n\n\new resp: {new_resp_tracks}\n\n\n\n\n\n\n\n\'") # log.info(f"\n\n\n\n\n\n\n\n\old resp: {old_resp_tracks}\n\n\n\n\n\n\n\n\'") # log.info(f'old len: {len(old_resp_tracks)}' # f'\nnew len: {len(new_resp_tracks)}' # f'\nnew isrc: {list({x for x in new_resp_tracks if new_resp_tracks.count(x) > 1})}' # f'\ninter len: {len(isrcs_intersected)}' # f'\ndiff_old: {[item for item in isrcs_old if item not in isrcs_new]}' # f'\ndiff_new: {[item for item in isrcs_new if item not in isrcs_old]}') # # indexes_all = [request_url] + [chart_date] + json_keys_top_charts_apple * len(isrcs_intersected) # 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_top_charts_apple]) # for t in new_resp_tracks: # new_endpoint_values.extend([t[k] for k in json_keys_top_charts_apple]) # for to, tn in zip(old_resp_tracks, new_resp_tracks): # diffs.extend([to[k] == tn[k] for k in json_keys_top_charts_apple]) # # 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'vendor_apple_{market}_top_chart.xlsx') # # @parameterized.expand( # [ # ('apple', 'global', 'daily'), # ('apple', 'global', 'weekly'), # ('apple', 'ua', 'daily'), # ('apple', 'ua', 'weekly'), # ] # ) # def test_top_charts_date_range(vendor, market, chart_type): # old_endpoint = f'/apollo-api/charts/date-range/?vendor={vendor}&market={market}&type={chart_type}' # if market == 'global': # new_endpoint = f'/gate-api/charts/date-range/?vendor={vendor}&market=worldwide&type={chart_type}' # else: # new_endpoint = f'/gate-api/charts/date-range/?vendor={vendor}&market={market}&type={chart_type}' # # get old response # old_resp = api_client_old.get_response(endpoint=old_endpoint, user=CoreConfig.USER_EMAIL, env=old_env, token=token_old)['body'] # # get new response # new_resp = api_client_old.get_response(endpoint=new_endpoint, user=CoreConfig.USER_EMAIL, env=new_env)['body'] # indexes_all = [request_url] + json_keys_top_charts_date_range # old_endpoint_values = [] # new_endpoint_values = [] # diffs = [] # # old_endpoint_values.append(old_endpoint) # new_endpoint_values.append(new_endpoint) # diffs.append('') # old_endpoint_values.extend(old_resp[k] for k in json_keys_top_charts_date_range) # new_endpoint_values.extend(new_resp[k] for k in json_keys_top_charts_date_range) # diffs.extend(old_resp[k] == new_resp[k] for k in json_keys_top_charts_date_range) # # 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'date_range_spotify_{market}_top_chart_{chart_type}.xlsx')