import logging import allure import pytest from assertpy import assert_that from dateutil.parser import parse from dateutil.relativedelta import relativedelta from ui_automation_framework.utils import AppConfig 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() # all_spotify_charts = delphi_api.delphi.get_charts_spotify() # 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("Compare YouTube data from dsp/gate-API and Delphi - Demographics") @allure.testcase("AG-9941") @pytest.mark.parametrize("playlist_id", AppConfig.get("playlists_spotify")) def test_comparison_youtube_data_between_dsp_gate_api_and_delphi(playlist_id): youtube_data = data_provider_qa.get_first_track_with_youtube_data_by_playlist(playlist_id) youtube_video_id = youtube_data[0]["video_id"] youtube_video_id_without_prefix = youtube_video_id.removeprefix("youtube_") youtube_latest_date = api_qa.dsp.get_youtube_latest_date()["body"]["date"] youtube_latest_date_2_weeks_ago = str(parse(youtube_latest_date) - relativedelta(days=13)).split()[0] demographics_data = api_qa.gate.get_youtube_demographics_data( start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, video_id=youtube_video_id_without_prefix, ) # Check for single YouTube id video_analytics = delphi_api.delphi.get_video_analytics_youtube(start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, video_id=youtube_video_id) if video_analytics: video_analytic_metrics = video_analytics[0]["metrics"] gender_with_the_biggest_percentage = sorted(video_analytic_metrics["gender_views_percentages"].items(), key=lambda item: item[1], reverse=True)[0][0] bands_with_the_popular_gender = {k: v for k, v in video_analytic_metrics["age_band_views_percentages"].items() if k.startswith(gender_with_the_biggest_percentage)} if all(i == 0.0 for i in bands_with_the_popular_gender.values()): popular_age_band = None else: popular_age_band = sorted(sorted(bands_with_the_popular_gender.items(), key=lambda x: x[0], reverse=True), key=lambda x: x[1], reverse=True)[0][0] else: popular_age_band = None assert_that(popular_age_band).is_equal_to(demographics_data["age_band_most_popular"]) # Check for bulk of YouTube ids youtube_video_ids_without_prefix = ",".join([i["video_id"] for i in youtube_data]) demographics_data = api_qa.gate.get_youtube_demographics_data( start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, video_id=youtube_video_ids_without_prefix, ) video_ids = ",".join([i["video_id"] for i in youtube_data]) all_gender_views_percentages = {} all_bands_with_the_popular_gender = {} video_analytics = delphi_api.delphi.get_video_analytics_youtube(start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, video_id=video_ids) if video_analytics: for video_analytic in video_analytics: video_analytic_metrics = video_analytic["metrics"] for gender, value in video_analytic_metrics["gender_views_percentages"].items(): if all_gender_views_percentages.get(gender): all_gender_views_percentages[gender] += value else: all_gender_views_percentages[gender] = value top_gender = sorted(all_gender_views_percentages.items(), key=lambda x: x[1], reverse=True)[0][0] for video_analytic in video_analytics: video_analytic_metrics = video_analytic["metrics"] for band, value in video_analytic_metrics["age_band_views_percentages"].items(): if band.startswith(top_gender): if all_bands_with_the_popular_gender.get(band): all_bands_with_the_popular_gender[band] += value else: all_bands_with_the_popular_gender[band] = value if all_bands_with_the_popular_gender: popular_age_band = sorted(all_bands_with_the_popular_gender.items(), key=lambda x: x[1], reverse=True)[0][0] if all(i == 0.0 for i in all_bands_with_the_popular_gender.values()): popular_age_band = None else: popular_age_band = None assert_that(popular_age_band).is_equal_to(demographics_data["age_band_most_popular"]) @allure.severity(allure.severity_level.NORMAL) @allure.title("Consistency of the YouTube default graph metrics with Delphi") @allure.testcase("AG-9977") @pytest.mark.parametrize("playlist_id", AppConfig.get("playlists_spotify"), ids=AppConfig.get("playlists_spotify")) def test_consistency_of_the_youtube_default_graph_metrics_with_delphi(playlist_id): youtube_data = data_provider_qa.get_first_track_with_youtube_data_by_playlist(playlist_id) youtube_video_id = youtube_data[0]["video_id"] youtube_video_id_without_prefix = youtube_video_id.removeprefix("youtube_") youtube_latest_date = api_qa.dsp.get_youtube_latest_date()["body"]["date"] youtube_latest_date_2_weeks_ago = str(parse(youtube_latest_date) - relativedelta(days=13)).split()[0] youtube_latest_date_4_weeks_ago = str(parse(youtube_latest_date_2_weeks_ago) - relativedelta(days=14)).split()[0] # Check trend data for the latest 2 weeks analytic_performance = api_qa.gate.get_youtube_analytic_performance(country_code="_gl", end_date=youtube_latest_date, video_id=youtube_video_id_without_prefix)["body"]["trends"] video_analytics_current_streams = delphi_api.delphi.get_video_analytics_youtube(start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, video_id=youtube_video_id) video_analytic_current_streams_metrics = video_analytics_current_streams if video_analytic_current_streams_metrics: video_analytic_current_streams_metrics = video_analytic_current_streams_metrics[0]["metrics"] else: video_analytic_current_streams_metrics = {"views": 0} assert_that(video_analytic_current_streams_metrics["views"]).is_equal_to(analytic_performance["weekly_views"]) # Check trend data for the previous 2 weeks video_analytics_past_streams = delphi_api.delphi.get_video_analytics_youtube( start_date=youtube_latest_date_4_weeks_ago, end_date=str(parse(youtube_latest_date_2_weeks_ago) - relativedelta(days=1)).split()[0], video_id=youtube_video_id, ) video_analytic_past_streams_metrics = video_analytics_past_streams if video_analytic_past_streams_metrics: video_analytic_past_streams_metrics = video_analytic_past_streams_metrics[0]["metrics"] else: video_analytic_past_streams_metrics = {"views": 0} if video_analytic_current_streams_metrics["views"] and not video_analytic_past_streams_metrics["views"]: rounded_calculated_trend = 100 elif video_analytic_past_streams_metrics["views"] == 0: rounded_calculated_trend = 0 else: calculated_trend = (video_analytic_current_streams_metrics["views"] - video_analytic_past_streams_metrics["views"]) / video_analytic_past_streams_metrics["views"] rounded_calculated_trend = round(calculated_trend * 100) assert_that(rounded_calculated_trend).is_equal_to(analytic_performance["weekly_trend"]) # Check trend data for the latest day video_analytics_for_the_latest_day_metrics = delphi_api.delphi.get_video_analytics_youtube(start_date=youtube_latest_date, end_date=youtube_latest_date, video_id=youtube_video_id) if video_analytics_for_the_latest_day_metrics: video_analytics_for_the_latest_day_metrics = video_analytics_for_the_latest_day_metrics[0]["metrics"] else: video_analytics_for_the_latest_day_metrics = {"views": 0} assert_that(video_analytics_for_the_latest_day_metrics["views"]).is_equal_to(analytic_performance["latest_date_views"]) # Check trend data for the previous day video_analytics_for_the_previous_day_metrics = delphi_api.delphi.get_video_analytics_youtube( start_date=str(parse(youtube_latest_date) - relativedelta(days=1)).split()[0], end_date=str(parse(youtube_latest_date) - relativedelta(days=1)).split()[0], video_id=youtube_video_id, ) if video_analytics_for_the_previous_day_metrics: video_analytics_for_the_previous_day_metrics = video_analytics_for_the_previous_day_metrics[0]["metrics"] else: video_analytics_for_the_previous_day_metrics = {"views": 0} if video_analytics_for_the_latest_day_metrics["views"] and not video_analytics_for_the_previous_day_metrics["views"]: rounded_calculated_trend = 100 elif video_analytics_for_the_previous_day_metrics["views"] == 0: rounded_calculated_trend = 0 else: calculated_trend = (video_analytics_for_the_latest_day_metrics["views"] - video_analytics_for_the_previous_day_metrics["views"]) / video_analytics_for_the_previous_day_metrics["views"] rounded_calculated_trend = round(calculated_trend * 100) assert_that(rounded_calculated_trend).is_equal_to(analytic_performance["daily_trend"]) @allure.severity(allure.severity_level.NORMAL) @allure.title("Consistency of the YouTube top 15 markets with Delphi") @allure.testcase("AG-9980") @pytest.mark.parametrize("playlist_id", AppConfig.get("playlists_spotify"), ids=AppConfig.get("playlists_spotify")) def test_consistency_of_the_youtube_top_markets_with_delphi(playlist_id): youtube_data = data_provider_qa.get_first_track_with_youtube_data_by_playlist(playlist_id) youtube_video_id = youtube_data[0]["video_id"] youtube_latest_date = api_qa.dsp.get_youtube_latest_date()["body"]["date"] youtube_latest_date_2_weeks_ago = str(parse(youtube_latest_date) - relativedelta(days=13)).split()[0] available_markets = api_qa.gate.get_list_of_available_markets_for_the_youtube_video(start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, video_id=youtube_video_id)["body"]["items"] youtube_top_markets = api_qa.gate.get_youtube_top_markets(video_id=youtube_video_id, start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, limit=15)["items"] youtube_top_market_codes = [i["market"] for i in youtube_top_markets] # Check for single YouTube id if available_markets: worldwide_views = delphi_api.delphi.get_video_analytics_youtube(start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, video_id=youtube_video_id)[0]["metrics"]["views"] for available_market in available_markets: specific_market_views = delphi_api.delphi.get_video_analytics_youtube( start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, video_id=youtube_video_id, country_code=available_market["code"])[0]["metrics"]["views"] if available_market["code"] in youtube_top_market_codes: top_market_data = next((i for i in youtube_top_markets if i["market"] == available_market["code"]), None) calculated_percentage_of_market_streams = round((specific_market_views / worldwide_views) * 100) assert_that(top_market_data["views"]).is_equal_to(specific_market_views) assert_that(top_market_data["percentage"]).is_equal_to(calculated_percentage_of_market_streams) else: assert_that(specific_market_views).is_less_than_or_equal_to(youtube_top_markets[14]["views"]) else: worldwide_views = delphi_api.delphi.get_video_analytics_youtube(start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, video_id=youtube_video_id)[0]["metrics"]["views"] assert_that(worldwide_views).is_zero() # Check for bulk of YouTube ids video_ids = ",".join([i["video_id"] for i in youtube_data]) available_markets = api_qa.gate.get_list_of_available_markets_for_the_youtube_video(start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, video_id=video_ids)["body"]["items"] youtube_top_markets = api_qa.gate.get_youtube_top_markets(video_id=video_ids, start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, limit=15)["items"] youtube_top_market_codes = [i["market"] for i in youtube_top_markets] if available_markets: worldwide_views = sum([i["metrics"]["views"] for i in delphi_api.delphi.get_video_analytics_youtube( start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, video_id=video_ids)]) for available_market in available_markets: specific_market_views = sum([i["metrics"]["views"] for i in delphi_api.delphi.get_video_analytics_youtube( start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, video_id=video_ids, country_code=available_market["code"])]) if available_market["code"] in youtube_top_market_codes: top_market_data = next((i for i in youtube_top_markets if i["market"] == available_market["code"]), None) calculated_percentage_of_market_streams = round((specific_market_views / worldwide_views) * 100) assert_that(top_market_data["views"]).is_equal_to(specific_market_views) assert_that(top_market_data["percentage"]).is_equal_to(calculated_percentage_of_market_streams) else: assert_that(specific_market_views).is_less_than_or_equal_to(youtube_top_markets[14]["views"]) else: worldwide_views = delphi_api.delphi.get_video_analytics_youtube(start_date=youtube_latest_date_2_weeks_ago, end_date=youtube_latest_date, video_id=video_ids)[0]["metrics"]["views"] assert_that(worldwide_views).is_zero()