import logging from datetime import datetime, timedelta import allure import pytest from assertpy import assert_that from ui_automation_framework.utils import AppConfig from ui_automation_framework.utils import logger as cl from app.src.api.facade import ApiClient log = cl.Logger(logging.DEBUG) api = ApiClient() api.authorize() delphi_api = ApiClient() delphi_api.authorize(env="DELPHI") @allure.severity(allure.severity_level.NORMAL) @allure.title("TikTok API verifications - Trends") @allure.testcase("AG-10309") @pytest.mark.parametrize("playlist_id", AppConfig.get("playlists_spotify")) @pytest.mark.parametrize("tiktok_metrics", ["creations", "video_views"]) def test_tiktok_trends_verification(playlist_id, tiktok_metrics): market = "us" tracklist = api.apollo.get_spotify_playlist_tracklist(playlist_id, market, "current_position", 100)["body"]["items"] isrc = tracklist[0]["isrc"] # log.debug(isrc) dsp = api.dsp.get_analytics_v2_tiktok(f"isrc={isrc}") start_date = dsp[tiktok_metrics]["week"]["start_date"] end_date = dsp[tiktok_metrics]["week"]["end_date"] time_period = datetime.strptime(end_date, "%Y-%m-%d") - datetime.strptime(start_date, "%Y-%m-%d") start_date_for_previous_period = datetime.strftime(datetime.strptime(start_date, "%Y-%m-%d") - timedelta(days=time_period.days + 1), "%Y-%m-%d") end_date_for_previous_period = datetime.strftime(datetime.strptime(start_date, "%Y-%m-%d") - timedelta(days=1), "%Y-%m-%d") delphi = delphi_api.delphi.get_analytics_tiktok( metrics=["creations", "video_views"], isrc=isrc, start_date=start_date, end_date=end_date, expand_to="related_isrcs" ) delphi_previous_period = delphi_api.delphi.get_analytics_tiktok( metrics=["creations", "video_views"], isrc=isrc, expand_to="related_isrcs", start_date=start_date_for_previous_period, end_date=end_date_for_previous_period, ) if not delphi["breakdowns"]["daily"][tiktok_metrics]: assert_that(bool(dsp[tiktok_metrics]["latest"]["value"])).is_false() assert_that(bool(dsp[tiktok_metrics]["week"]["value"])).is_false() assert_that(bool(dsp[tiktok_metrics]["latest"]["trend"])).is_false() assert_that(bool(dsp[tiktok_metrics]["week"]["trend"])).is_false() else: assert_that(dsp[tiktok_metrics]["latest"]["value"]).is_equal_to(delphi["breakdowns"]["daily"][tiktok_metrics][-1]) delphi_breakdowns_daily_metrics = delphi["breakdowns"]["daily"][tiktok_metrics] delphi_breakdowns_daily_metrics_filtered = [i for i in delphi_breakdowns_daily_metrics if i is not None] assert_that(dsp[tiktok_metrics]["week"]["value"]).is_equal_to(sum(delphi_breakdowns_daily_metrics_filtered)) current_day = delphi["breakdowns"]["daily"][tiktok_metrics][-1] previous_day = delphi["breakdowns"]["daily"][tiktok_metrics][-2] if previous_day: calculated_latest_delphi_video_trend = round((current_day - previous_day) / previous_day * 100) assert_that(dsp[tiktok_metrics]["latest"]["trend"]).is_equal_to(calculated_latest_delphi_video_trend) else: assert_that(dsp[tiktok_metrics]["latest"]["trend"]).is_false() current_period = sum(delphi_breakdowns_daily_metrics_filtered) previous_period = sum(delphi_previous_period["breakdowns"]["daily"][tiktok_metrics]) if not previous_period: assert_that(dsp[tiktok_metrics]["week"]["trend"]).is_none() else: calculated_week_delphi_video_trend = round((current_period - previous_period) / previous_period * 100) assert_that(dsp[tiktok_metrics]["week"]["trend"]).is_equal_to(calculated_week_delphi_video_trend) @allure.severity(allure.severity_level.NORMAL) @allure.title("TikTok API verifications - Markets") @allure.testcase("AG-10323") @pytest.mark.parametrize("playlist_id", AppConfig.get("playlists_spotify")) def test_tiktok_markets_verification(playlist_id): market = "us" tracklist = api.apollo.get_spotify_playlist_tracklist(playlist_id, market, "current_position", 100)["body"]["items"] for track in tracklist[:5]: isrc = track["isrc"] # log.debug(isrc) dsp = api.dsp.get_analytics_v2_tiktok(f"isrc={isrc}") start_date = dsp["video_views"]["week"]["start_date"] end_date = dsp["video_views"]["week"]["end_date"] all_markets = ",".join([market["code"] for market in api.gate.get_all_available_markets_tiktok(isrc)]) delphi_analytics = delphi_api.delphi.get_analytics_tiktok( metrics=["creations", "video_views"], isrc=isrc, start_date=start_date, end_date=end_date, expand_to="related_isrcs", country_code=all_markets, breakdowns="country_totals", )[ "breakdowns" ]["country_totals"] if not delphi_analytics: assert_that(bool(api.dsp.get_analytics_v2_tiktok(f"isrc={isrc}")["video_views"]["week"]["value"])).is_false() continue else: break else: return # check that for all markets there is info (at least 1 creation or view) for metrics in delphi_analytics.items(): assert_that(any([metrics[1]["creations"] > 0, metrics[1]["video_views"] > 0])).is_true() # sort delphi_analytics by the "creations" key in descending order delphi_analytics_sorted_by_markets = dict(sorted(delphi_analytics.items())) delphi_analytics_sorted_by_creations = dict(sorted(delphi_analytics_sorted_by_markets.items(), key=lambda x: x[1]["creations"], reverse=True)) delphi_analytics_sorted_by_video_views = dict(sorted(delphi_analytics_sorted_by_markets.items(), key=lambda x: x[1]["video_views"], reverse=True)) dsp_markets = api.dsp.get_top_markets_tiktok(isrc=isrc) if dsp_markets["creations"]: for delphi, dsp in zip(delphi_analytics_sorted_by_creations.items(), dsp_markets["creations"]): assert_that(delphi[0]).is_equal_to(dsp["market"]) assert_that(delphi[1]["creations"]).is_equal_to(dsp["cnt"]) else: assert_that(any([i["creations"] for i in delphi_analytics_sorted_by_creations.values()])).is_false() for delphi, dsp in zip(delphi_analytics_sorted_by_video_views.items(), dsp_markets["video_views"]): assert_that(delphi[0]).is_equal_to(dsp["market"]) assert_that(delphi[1]["video_views"]).is_equal_to(dsp["cnt"]) @allure.severity(allure.severity_level.NORMAL) @allure.title("TikTok API verifications - Graph") @allure.testcase("AG-10308") @pytest.mark.parametrize("playlist_id", AppConfig.get("playlists_spotify")) def test_tiktok_graph_verification(playlist_id): market = "us" tracklist = api.apollo.get_spotify_playlist_tracklist(playlist_id, market, "current_position", 100)["body"]["items"] for track in tracklist[:5]: isrc = track["isrc"] # log.debug(isrc) dsp_graph_data = api.dsp.get_graph_tiktok(isrc) if dsp_graph_data["creations"]: dsp_graph_creations_data = [i["y"] for i in dsp_graph_data["creations"]["graph"] if i["y"] is not None] if dsp_graph_data["video_views"]: dsp_graph_video_views_data = [i["y"] for i in dsp_graph_data["video_views"]["graph"] if i["y"] is not None] dsp_analytics = api.dsp.get_analytics_v2_tiktok(f"isrc={isrc}") start_date = dsp_analytics["video_views"]["week"]["start_date"] end_date = dsp_analytics["video_views"]["week"]["end_date"] delphi_analytics = delphi_api.delphi.get_analytics_tiktok( metrics=["creations", "video_views"], isrc=isrc, start_date=start_date, end_date=end_date, expand_to="related_isrcs", ) if delphi_analytics["days_count"] < 14: continue delphi_analytics_daily_creations = delphi_analytics["breakdowns"]["daily"]["creations"] delphi_analytics_daily_video_views = delphi_analytics["breakdowns"]["daily"]["video_views"] if not all([delphi_analytics_daily_creations, delphi_analytics_daily_video_views]): assert_that(dsp_graph_data["creations"]).is_false() assert_that(dsp_graph_data["video_views"]).is_false() continue elif all([delphi_analytics_daily_creations, delphi_analytics_daily_video_views]): assert_that(delphi_analytics_daily_creations).is_equal_to(dsp_graph_creations_data) assert_that(delphi_analytics_daily_video_views).is_equal_to(dsp_graph_video_views_data) continue @allure.severity(allure.severity_level.NORMAL) @allure.title("TikTok API verifications - Insight") @allure.testcase("AG-10311") @pytest.mark.parametrize("playlist_id", AppConfig.get("playlists_spotify")) def test_tiktok_insight_verification(playlist_id): market = "us" tracklist = api.apollo.get_spotify_playlist_tracklist(playlist_id, market, "current_position", 100)["body"]["items"] for track in tracklist[:5]: isrc = track["isrc"] # log.debug(isrc) dsp_insight_data = api.dsp.get_insight_tiktok(isrc) dsp_analytics = api.dsp.get_analytics_v2_tiktok(f"isrc={isrc}") start_date = dsp_analytics["video_views"]["week"]["start_date"] end_date = dsp_analytics["video_views"]["week"]["end_date"] delphi_analytics = delphi_api.delphi.get_analytics_tiktok( metrics=["creations", "video_views"], isrc=isrc, start_date=start_date, end_date=end_date, expand_to="related_isrcs", ) if not dsp_insight_data["creations"] and not dsp_insight_data["views"]: assert_that(delphi_analytics["breakdowns"]["daily"]["creations"]).is_false() assert_that(delphi_analytics["breakdowns"]["daily"]["video_views"]).is_false() continue min_date_dt_obj = datetime.strptime(delphi_analytics["min_date"], "%Y-%m-%d") max_date_dt_obj = datetime.strptime(delphi_analytics["max_date"], "%Y-%m-%d") delphi_analytics_daily_creations = delphi_analytics["breakdowns"]["daily"]["creations"] delphi_analytics_daily_creations = [i for i in delphi_analytics_daily_creations if i is not None] delphi_analytics_daily_video_views = delphi_analytics["breakdowns"]["daily"]["video_views"] dates_range = [min_date_dt_obj + timedelta(n) for n in range(int((max_date_dt_obj - min_date_dt_obj).days) + 1)] dates_range_formatted = [datetime.strftime(date, "%Y-%m-%d") for date in dates_range] delphi_video_views_with_dates = list(zip(delphi_analytics_daily_video_views, dates_range_formatted)) max_delphi_daily_creations_value = max(delphi_analytics_daily_creations) assert_that(max_delphi_daily_creations_value).is_equal_to(dsp_insight_data["creations"]) max_delphi_daily_creations_value_indexes = [index for index, item in enumerate(delphi_analytics_daily_creations) if item == max_delphi_daily_creations_value] if len(max_delphi_daily_creations_value_indexes) == 1: delphi_video_views_value = next(i[0] for i in delphi_video_views_with_dates if i[1] == dsp_insight_data["peak_date"]) assert_that(delphi_video_views_value).is_equal_to(dsp_insight_data["views"]) else: delphi_video_views_values_without_none = [i for i in delphi_analytics_daily_video_views if i is not None] delphi_video_views_values = [delphi_video_views_values_without_none[index] for index in max_delphi_daily_creations_value_indexes] delphi_video_views_values = [i for i in delphi_video_views_values if i is not None] assert_that(max(delphi_video_views_values)).is_equal_to(dsp_insight_data["views"]) max_delphi_video_views_values_datetime = next(i for i in delphi_video_views_with_dates if i[0] == max(delphi_video_views_values))[1] assert_that(max_delphi_video_views_values_datetime).is_equal_to(dsp_insight_data["peak_date"])