import logging import random import allure import pytest from assertpy import assert_that from ui_automation_framework.utils import logger as cl from app.mysql_db_client_custom import MySqlClient from app.src.api.facade import ApiClient from app.src.helpers.api_data_provider import ApiDataProvider logger = cl.Logger(logging.DEBUG) api_qa = ApiClient() api_qa.authorize(env="DEV") delphi_api = ApiClient() delphi_api.authorize(env="DELPHI") sql_client = MySqlClient() data_provider_qa = ApiDataProvider(api_qa) vendor = "spotify" market = "global" fields = [ "artist", "artists", "change", "image_url", "is_starred", "name", "position", "isrc", "is_sony", "is_new", "id", ] order_by = ["-is_starred_track", "position"] all_apple_charts = delphi_api.delphi.get_charts_apple(ignore_response_attaching=True) all_spotify_charts = delphi_api.delphi.get_charts_spotify(ignore_response_attaching=True) apple_markets = [("apple", i["country_code"]) for i in all_apple_charts] spotify_markets = [("spotify", i["country_code"]) for i in all_spotify_charts if i["country_code"] != "ru"] def equal_dicts(d1, d2, ignore_keys): d1_filtered = {k: v for k, v in d1.items() if k not in ignore_keys} d2_filtered = {k: v for k, v in d2.items() if k not in ignore_keys} assert d1_filtered == d2_filtered # @allure.severity(allure.severity_level.NORMAL) # @allure.title("Data comparison for Charts additions on QA and Prod") # @allure.description("Data comparison for Charts additions on QA and Prod") # @allure.testcase("AG-9502") # def test_data_comparison_for_charts_on_qa_and_prod_additions(): # # additions_data_qa = data_provider_qa.apollo_api_client.get_charts_additions( # vendor, market, order_by=order_by, fields=fields, limit=20)["body"]["tracks"] # additions_data_prod = data_provider_prod.apollo_api_client.get_charts_additions( # vendor, market, order_by=order_by, fields=fields, limit=20)["body"]["tracks"] # # for k, v in enumerate(additions_data_qa): # equal_dicts(additions_data_qa[k], additions_data_prod[k], ignore_keys=["image_url", "trend"]) # # # @allure.severity(allure.severity_level.NORMAL) # @allure.title("Data comparison for Charts removals on QA and Prod") # @allure.description("Data comparison for Charts removals on QA and Prod") # @allure.testcase("AG-9502") # def test_data_comparison_for_charts_on_qa_and_prod_removals(): # # removals_data_qa = data_provider_qa.apollo_api_client.get_charts_removals( # vendor, market, order_by=order_by, fields=fields, limit=20)["body"]["tracks"] # removals_data_prod = data_provider_prod.apollo_api_client.get_charts_removals( # vendor, market, order_by=order_by, fields=fields, limit=20)["body"]["tracks"] # # for k, v in enumerate(removals_data_qa): # equal_dicts(removals_data_qa[k], removals_data_prod[k], ignore_keys=["image_url", "trend"]) # # # @allure.severity(allure.severity_level.NORMAL) # @allure.title("Data comparison for Charts moves on QA and Prod") # @allure.description("Data comparison for Charts moves on QA and Prod") # @allure.testcase("AG-9502") # def test_data_comparison_for_charts_on_qa_and_prod_moves(): # # moves_data_qa = data_provider_qa.apollo_api_client.get_charts_moves( # vendor, market, order_by=["moves", "track_name"], fields=fields).tracks # moves_data_prod = data_provider_prod.apollo_api_client.get_charts_moves( # vendor, market, order_by=["moves", "track_name"], fields=fields).tracks # # for k, v in enumerate(moves_data_qa): # equal_dicts(moves_data_qa[k].__dict__, moves_data_prod[k].__dict__, ignore_keys=["image_url", "trend"]) # # # @allure.severity(allure.severity_level.NORMAL) # @allure.title("Data comparison for Charts top on QA and Prod") # @allure.description("Data comparison for Charts top on QA and Prod") # @allure.testcase("AG-9502") # def test_data_comparison_for_charts_on_qa_and_prod_top_data(): # # top_data_qa = data_provider_qa.apollo_api_client.get_charts_top( # vendor, market, fields=["image_url"], order_by=["position"])["body"]["tracks"] # top_data_prod = data_provider_prod.apollo_api_client.get_charts_top( # vendor, market, fields=["image_url"], order_by=["position"])["body"]["tracks"] # # for k, v in enumerate(top_data_qa): # equal_dicts(top_data_qa[k], top_data_prod[k], ignore_keys=["image_url", "trend"]) # # # @allure.severity(allure.severity_level.NORMAL) # @allure.title("Data comparison for Charts summary on QA and Prod") # @allure.description("Data comparison for Charts summary on QA and Prod") # @allure.testcase("AG-9502") # def test_data_comparison_for_charts_on_qa_and_prod_summary(): # # top_data_qa = data_provider_qa.apollo_api_client.get_charts_top( # vendor, market, fields=["image_url"], order_by=["position"])["body"]["tracks"] # top_isrcs = [track["isrc"] for track in top_data_qa] # for isrc in top_isrcs[:25]: # charts_data_qa = data_provider_qa.apollo_api_client.get_charts_summary(isrc, vendor, 20, 0)["body"]["items"] # charts_data_prod = data_provider_prod.apollo_api_client.get_charts_summary(isrc, vendor, 20, 0)["body"]["items"] # for k, v in enumerate(charts_data_qa): # equal_dicts(charts_data_qa[k], charts_data_prod[k], ignore_keys=["image_url", "trend"]) # @allure.severity(allure.severity_level.NORMAL) # @allure.title("Data comparison between market size in DB and rank in Delphi") # @allure.description("It’s required to compare market_size for Spotify vendor against rank " # "which provides Delphi and check if our market size matches with their rank") # @allure.testcase("AG-9355") # @pytest.mark.data_validation # def test_data_comparison_between_market_size_in_db_and_rank_in_delphi(): # # TODO: should be added to the separate job with data validation # number_of_days = 92 # # with allure.step("1. Get the latest date"): # latest_date = api_qa.dsp.get_latest_date_go(None) # latest_date = "2022-06-20" # latest_date_parsed_to_dt_obj = datetime.strptime(latest_date, "%Y-%m-%d") # twenty_eight_days_ago_to_latest_date = (latest_date_parsed_to_dt_obj - timedelta(days=number_of_days)).strftime("%Y-%m-%d") # # with allure.step("2. Take all Spotify charts from Delphi, ordered by rank"): # charts_summary = delphi_api.delphi.get_all_charts(dsp="spotify", sort_by="rank", sort_order="asc") # charts_summary_result = [i["country_code"] for i in charts_summary] # charts_summary_result.remove("worldwide") # charts_summary_result.remove("ru") # # with allure.step(f"3. Collect streams from DB per {number_of_days} days per market and range markets " # "from the highest to lowest starting from the latest date."): # market_and_total_streams = \ # sql_client.get_market_and_total_streams_for_a_specific_time_frame(twenty_eight_days_ago_to_latest_date, latest_date) # # # sum all values (total streams) that have the same key (market) # d = {} # for item in market_and_total_streams: # if item[0] in d: # d[item[0]] += item[1] # else: # d[item[0]] = item[1] # # convert dictionary to the list of tuples in order to sort by value (total streams) # r = [i for i in d.items()] # r.sort(key=lambda x: x[1], reverse=True) # # exclude values (total streams) by creating a new list with markets only # db_result_list = [x[0] for x in r] # # filter so that the list contains only the markets that are present in charts_summary_result # db_result_list_filtered = [i for i in db_result_list if i in charts_summary_result] # # try: # assert_that(charts_summary_result).is_equal_to(db_result_list_filtered) # except AssertionError as err: # is_correct = [] # for k, v in enumerate(charts_summary_result): # is_correct.append(charts_summary_result[k] == db_result_list_filtered[k]) # # 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 # # df = pd.DataFrame({"IS CORRECT": is_correct, # "From API": charts_summary_result, # "From DB": db_result_list_filtered}) # excel_path = os.getcwd() + "/comparison_between_db_and_delphi.xlsx" # df.reset_index().style.apply(f, subset=["IS CORRECT"]) \ # .to_excel(excel_path) # allure.attach.file(excel_path, name="comparison_between_db_and_delphi.xlsx", extension="xlsx") # raise err @allure.severity(allure.severity_level.NORMAL) @allure.title("Create end-to-end scenario to verify consistency between Chart Digest and Chart Placements") @allure.testcase("AG-9669") @pytest.mark.parametrize("vendor, market", apple_markets + spotify_markets) def test_consistency_between_chart_digest_and_chart_placements(vendor, market): charts_digest = data_provider_qa.apollo_api_client.get_charts_top(vendor, market)["body"]["tracks"] track = random.choice(charts_digest) isrc = track["isrc"] charts_summary = api_qa.apollo.get_charts_summary(isrc, vendor, 150, 0, market=market)["body"]["items"] chart_summary = next(i for i in charts_summary if i["country_code"] == market) # logger.info("#" * 150) # logger.info(f"ISRC: {isrc}") # logger.info(track) # logger.info(chart_summary) # logger.info("#" * 150) if track["chart_date"] == chart_summary["date"]: if not bool(assert_that(any([track["is_new"], track["is_re_enter"]])).is_equal_to(any([chart_summary["is_new"], chart_summary["is_re_enter"]]))): assert_that(track["change"] * -1).is_equal_to(chart_summary["change"]) assert_that(track["chart_date"]).is_equal_to(chart_summary["date"]) assert_that(track["position"]).is_equal_to(chart_summary["position"]) if vendor == "spotify": assert_that(track["streams"]).is_equal_to(chart_summary["streams"])