import logging import random from datetime import date 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 from app.src.helpers.api_data_provider import ApiDataProvider log = cl.Logger(logging.DEBUG) api = ApiClient() api.authorize() delphi_api = ApiClient() delphi_api.authorize(env="DELPHI") data_provider = ApiDataProvider(api) data = data_provider.get_sony_tracks_isrcs(market="us", ignore_response_attaching=True) isrcs = data[0][:10] playlists_spotify = list(AppConfig.get("playlists_spotify").keys()) playlists_apple = list(AppConfig.get("playlists_apple").keys()) def get_owner_category_info(playlist): try: category = playlist["playlist"]["owner"]["owner_category"] category_id = int(category["owner_category_id"]) category_name = category["name"] except (KeyError, TypeError): category_id, category_name = None, None return category_id, category_name def assert_owner_category_info_match(go_playlist, delphi_playlist): delphi_owner_category_id, delphi_owner_category_name = get_owner_category_info(delphi_playlist) assert_that(go_playlist["category_id"]).is_equal_to(delphi_owner_category_id) assert_that(go_playlist["category_name"]).is_equal_to(delphi_owner_category_name) def assert_added_date(go_playlist, delphi_playlist): if go_playlist["added_date"]: assert_that(go_playlist["added_date"]).is_equal_to(delphi_playlist["earliest_position_date_time"].split("T")[0]) else: assert_that(delphi_playlist["earliest_position_date_time"].split("T")[0]).is_equal_to("1970-01-01") def assert_is_new(go_playlist, delphi_playlist): if go_playlist["is_new"]: assert_that(delphi_playlist["is_top_track_for_isrc"] and delphi_playlist["num_days_on"] == 0).is_true() else: assert_that(not delphi_playlist["is_top_track_for_isrc"] or delphi_playlist["num_days_on"] != 0).is_true() def assert_playlist_updated_date(formatted_playlist_id, go_playlist, is_spotify_non_personalized): delphi_playlist_dates = delphi_api.delphi.get_playlist_dates(playlist_id=formatted_playlist_id) # log.debug(f"go_playlist: {go_playlist}") # log.debug(f"delphi_playlist_dates: {delphi_playlist_dates}") if is_spotify_non_personalized: today = date.today() assert_that(go_playlist["updated_date"]).is_equal_to(str(today)) else: if go_playlist["country_code"] != "_gl": delphi_updated_date = [i for i in delphi_playlist_dates["items"] if i["country_code"] == go_playlist["country_code"]][0]["dsp_sync_date_time"][0] else: delphi_updated_date = [i for i in delphi_playlist_dates["items"] if i["country_code"] == "global"][0]["dsp_sync_date_time"][0] assert_that(go_playlist["updated_date"]).is_equal_to(delphi_updated_date.split("T")[0]) def get_min_num_of_iterations(resp: dict): max_num_of_iterations = 50 num_of_items = len(resp["items"]) return min(num_of_items, max_num_of_iterations) @allure.severity(allure.severity_level.NORMAL) @allure.title("spotify/by-track AGo vs Delphi") @allure.description("Comparing spotify/by-track between AGo and Delphi") @allure.testcase("https://data-analytics.atlassian.net/browse/AG-10592") @pytest.mark.parametrize("isrc", isrcs) def test_consistency_spotify_by_track_ago_vs_delphi(isrc): start_date, end_date = data_provider.get_track_start_end_dates(isrc, 6) go = api.apollo.get_spotify_playlists_data_go( isrc=isrc, streams_market="_gl", start_date=start_date, end_date=end_date, streams_type="priority", keep_zero_streams=True, offset=0, limit=2000, order_by=["-current_streams", "-id"])["body"] delphi = delphi_api.delphi.get_all_playlists_positions( dsp="spotify", isrc={isrc}, limit=10000, include=["playlists", "streams_for_period", "meta_owner_categories"], streams_country_code="worldwide") for _ in range(50): go_playlist = random.choice(go["items"]) if go_playlist["personalized"]: continue go_playlist_id = go_playlist["id"] delphi_playlist = next(i for i in delphi["items"] if i["playlist_id"].endswith(go_playlist_id)) # log.debug(f"AGo playlist: {go_playlist}\nDelphi playlist: {delphi_playlist}") assert_owner_category_info_match(go_playlist, delphi_playlist) assert_that(go_playlist["name"]).is_equal_to(delphi_playlist["playlist"]["name"]) assert_added_date(go_playlist, delphi_playlist) assert_that(go_playlist["updated_date"]).is_equal_to(delphi_playlist["latest_position_date_time"].split("T")[0]) assert_that(go_playlist["country_code"]).is_equal_to(delphi_playlist["playlist"]["country_code"]) assert_that(go_playlist["track_id"]).is_equal_to(delphi_playlist["dsp_track_id"]) assert_that(go_playlist["followers"]).is_equal_to(delphi_playlist["playlist"]["followers"]) assert_that(go_playlist["personalized"]).is_equal_to(delphi_playlist["playlist"]["is_personalised"]) # assert_that(go_playlist["position_change_date"]).is_equal_to(delphi_playlist["last_date_change_14_days"]) assert_that(go_playlist["current_position"]).is_equal_to(delphi_playlist["current"] + 1) assert_that(go_playlist["trend"]).is_equal_to(delphi_playlist["trend_change_14_days"]) assert_that(delphi_playlist["is_top_track_for_isrc"]).is_true() assert_is_new(go_playlist, delphi_playlist) delphi_owner_categories = delphi["meta"]["owner_categories"] delphi_owner_categories_formatted = [{"category_name": d.get("name"), "category_id": int(d.get("owner_category_id"))} for d in delphi_owner_categories] assert_that(go["categories"]).is_equal_to(delphi_owner_categories_formatted) if go["streams_type"] == "track": key = "isrc_in_playlist" elif go["streams_type"] == "playlist": key = "playlist" else: raise ValueError("Incorrect streams_type value") try: assert_that(go_playlist["current_streams"]).is_equal_to(delphi_playlist["streaming_info"]["worldwide"]["7"][key]) except (TypeError, KeyError): assert_that(go_playlist["current_streams"]).is_none() try: assert_that(go_playlist["prev_streams"]).is_equal_to(delphi_playlist["streaming_info"]["worldwide"]["14"][key] - delphi_playlist["streaming_info"]["worldwide"]["7"][key]) except (TypeError, KeyError): assert_that(go_playlist["prev_streams"]).is_none() @allure.severity(allure.severity_level.NORMAL) @allure.title("apple/by-track AGo vs Delphi") @allure.description("Comparing apple/by-track between AGo and Delphi") @allure.testcase("https://data-analytics.atlassian.net/browse/AG-10593") @pytest.mark.parametrize("isrc", isrcs) def test_consistency_apple_by_track_ago_vs_delphi(isrc): start_date, end_date = data_provider.get_track_start_end_dates(isrc, 6) market = "us" go = api.apollo.get_apple_playlists_data_go( isrc=isrc, market=market, start_date=start_date, end_date=end_date, streams_type="priority", keep_zero_streams=True, offset=0, limit=2000, order_by=["-current_streams", "-id"])["body"] delphi = delphi_api.delphi.get_all_playlists_positions( dsp="apple", isrc=isrc, limit=10000, include=["playlists", "streams_for_period", "meta_owner_categories"], country_code=market, streams_country_code="local") for _ in range(50): go_playlist = random.choice(go["items"]) go_playlist_id = go_playlist["id"] print(go_playlist_id) delphi_playlist = next(i for i in delphi["items"] if i["playlist_id"].endswith(go_playlist_id)) # log.debug(f"AGo playlist: {go_playlist}\nDelphi playlist: {delphi_playlist}") assert_that(go["count"]).is_equal_to(delphi["count"]) assert_owner_category_info_match(go_playlist, delphi_playlist) assert_added_date(go_playlist, delphi_playlist) assert_that(go_playlist["updated_date"]).is_equal_to(delphi_playlist["latest_position_date_time"].split("T")[0]) assert_that(go_playlist["country_code"]).is_equal_to(delphi_playlist["playlist"]["country_code"]) assert_that(go_playlist["name"]).is_equal_to(delphi_playlist["playlist"]["name"]) assert_that(go_playlist["track_id"]).is_equal_to(delphi_playlist["dsp_track_id"]) assert_that(go_playlist["position_change_date"]).is_equal_to(delphi_playlist["last_date_change_14_days"]) assert_that(go_playlist["followers"]).is_equal_to(delphi_playlist["playlist"]["followers"]) assert_that(go_playlist["current_position"]).is_equal_to(delphi_playlist["current"] + 1) assert_that(delphi_playlist["is_top_track_for_isrc"]).is_true() assert_is_new(go_playlist, delphi_playlist) if delphi_playlist["is_top_track_for_isrc"]: assert_that(go_playlist["trend"]).is_equal_to(delphi_playlist["trend_change_14_days"]) else: assert_that(delphi_playlist["trend"]).is_none() if go["streams_type"] == "track": key = "isrc_in_playlist" elif go["streams_type"] == "playlist": key = "playlist" else: raise ValueError("Incorrect streams_type value") if delphi_playlist["streaming_info"]["local"]: if delphi_playlist["streaming_info"]["local"]["7"]: if delphi_playlist["streaming_info"]["local"]["7"].get(key): assert_that(go_playlist["current_streams"]).is_equal_to( delphi_playlist["streaming_info"]["local"]["7"][key]) assert_that(go_playlist["prev_streams"]).is_equal_to( delphi_playlist["streaming_info"]["local"]["14"][key] - delphi_playlist["streaming_info"]["local"]["7"][key]) else: assert_that(go_playlist["current_streams"]).is_none() assert_that(go_playlist["prev_streams"]).is_none() @allure.severity(allure.severity_level.NORMAL) @allure.title("spotify/by-track/history and apple/by-track/history AGo vs Delphi") @allure.description("Comparing spotify/by-track/history and apple/by-track/history between AGo and Delphi") @allure.testcase("https://data-analytics.atlassian.net/browse/AG-10594", "https://data-analytics.atlassian.net/browse/AG-10595") @pytest.mark.parametrize("vendor, api_call, market_parameter", [("spotify", api.apollo.get_v3_spotify_track_previous_playlists, "streams_market=_gl"), ("apple", api.apollo.get_v3_apple_music_track_previous_playlists, "market=us")]) @pytest.mark.parametrize("isrc", isrcs) def test_consistency_by_track_history_ago_vs_delphi(isrc, vendor, api_call, market_parameter): start_date, end_date = data_provider.get_track_start_end_dates(isrc, 6) go = api_call( f"isrc={isrc}", market_parameter, f"start_date={start_date}", f"end_date={end_date}", "offset=0", "limit=2000", "order_by=-removed_date,-id", "exclude_current_playlists=true")["body"] delphi = delphi_api.delphi.get_history_playlists_data( dsp=vendor, isrc=isrc, country_code="us" if vendor == "apple" else None, limit=10000, include=["playlists", "meta_owner_categories", "dsp_sync_date_time"], sort_by="removed_date_time", sort_order="desc") num_of_iterations = get_min_num_of_iterations(go) for _ in range(num_of_iterations): go_playlist = random.choice(go["items"]) go_playlist_id = go_playlist["id"] delphi_playlist = next(i for i in delphi["items"] if i["playlist_id"].endswith(go_playlist_id)) # log.debug(f"{vendor.capitalize()} AGo playlist: {go_playlist}\nDelphi playlist: {delphi_playlist}") assert_that(go["count"]).is_equal_to(delphi["count"]) assert_owner_category_info_match(go_playlist, delphi_playlist) assert_that(go_playlist["name"]).is_equal_to(delphi_playlist["playlist"]["name"]) assert_added_date(go_playlist, delphi_playlist) assert_that(go_playlist["updated_date"]).is_equal_to(delphi_playlist["playlist_dates"]["dsp_sync_date_time"].split("T")[0]) assert_that(go_playlist["removed_date"]).is_equal_to(delphi_playlist["removed_date_time"].split("T")[0]) assert_that(go_playlist["country_code"]).is_equal_to(delphi_playlist["playlist"]["country_code"]) delphi_owner_categories = delphi["meta"]["owner_categories"] delphi_owner_categories_formatted = [{"category_name": d.get("name"), "category_id": int(d.get("owner_category_id"))} for d in delphi_owner_categories] sorted_go_categories = sorted(go["categories"], key=lambda x: x["category_name"]) sorted_delphi_categories = sorted(delphi_owner_categories_formatted, key=lambda x: x["category_name"]) assert_that(sorted_go_categories).is_equal_to(sorted_delphi_categories) if vendor == "apple": assert_that(go_playlist["id"]).is_equal_to(delphi_playlist["playlist"]["playlist_id"].removeprefix(f"{vendor.lower()}_")) else: assert_that(go_playlist["personalized"]).is_equal_to(delphi_playlist["playlist"]["is_personalised"]) assert_that(go_playlist["followers"]).is_equal_to(delphi_playlist["playlist"]["followers"]) @allure.severity(allure.severity_level.NORMAL) @allure.title("playlists/spotify and playlists/apple AGo vs. Delphi") @allure.description("Comparing playlists/spotify and playlists/apple between AGo and Delphi") @allure.testcase("https://data-analytics.atlassian.net/browse/AG-10600", "https://data-analytics.atlassian.net/browse/AG-10601") @pytest.mark.parametrize("vendor, api_call, playlist_id", [("spotify", api.gate.get_playlists_spotify, i) for i in playlists_spotify] + [("apple", api.gate.get_playlists_apple, i) for i in playlists_apple]) def test_consistency_playlist_ago_vs_delphi(vendor, api_call, playlist_id): if vendor == "spotify": go_playlist = api_call(f"id={playlist_id}")["body"][0] elif vendor == "apple": go_playlist = api_call(f"id={playlist_id}&market=us")["body"][0] formatted_playlist_id = f"{vendor}_{playlist_id}" delphi_playlist = delphi_api.delphi.get_current_playlist_data(playlist_id=formatted_playlist_id) assert_that(go_playlist["id"]).is_equal_to(delphi_playlist["dsp_playlist_id"]) assert_that(go_playlist["owner"]).is_equal_to(delphi_playlist["owner"]["display_name"]) assert_that(go_playlist["name"]).is_equal_to(delphi_playlist["name"]) if vendor == "spotify": assert_that(go_playlist["personalized"]).is_equal_to(delphi_playlist["is_personalised"]) if go_playlist["country_code"] == "_gl": assert_that(delphi_playlist["country_code"]).is_equal_to("global") else: assert_that(go_playlist["country_code"]).is_equal_to(delphi_playlist["country_code"]) if go_playlist.get("personalized") or vendor == "apple": assert_playlist_updated_date(formatted_playlist_id, go_playlist, is_spotify_non_personalized=False) else: assert_playlist_updated_date(formatted_playlist_id, go_playlist, is_spotify_non_personalized=True) @allure.severity(allure.severity_level.NORMAL) @allure.title("spotify/tracklist and apple/tracklist AGo vs. Delphi") @allure.description("Comparing spotify/tracklist and apple/tracklist between AGo and Delphi") @allure.testcase("https://data-analytics.atlassian.net/browse/AG-10596", "https://data-analytics.atlassian.net/browse/AG-10597") @pytest.mark.parametrize("vendor, tracklist_api_call, track_api_call, playlist_id", [("spotify", api.apollo.get_spotify_playlist_tracklist, api.apollo.get_v2_spotify_playlist_track, i) for i in playlists_spotify] + [("apple", api.apollo.get_v1_apple_music_playlist_tracklist, api.apollo.get_v2_apple_music_playlist_track, i) for i in playlists_apple]) def test_consistency_spotify_and_apple_tracklist_ago_vs_delphi(vendor, tracklist_api_call, track_api_call, playlist_id): go = tracklist_api_call( playlist_id=playlist_id, market="_gl" if vendor == "spotify" else "us", order_by="current_position", limit=2000)["body"] delphi = delphi_api.delphi.get_current_tracklist_data( playlist_id=f"{vendor}_" + playlist_id, country_code=None if vendor == "spotify" else "us", sort_by="current", sort_order="asc", include="track_info") assert_that(go["count"]).is_equal_to(delphi["count"]) for go_tracklist_track, delphi_tracklist_track in zip(go["items"], delphi["items"]): if not delphi_tracklist_track["is_top_track_for_isrc"]: # log.debug(f"go_tracklist_track: {go_tracklist_track}\ndelphi_tracklist_track: {delphi_tracklist_track}") assert_that(go_tracklist_track["name"]).is_equal_to(delphi_tracklist_track[f"{vendor}_track"]["name"]) assert_that(go_tracklist_track["artists"]).is_equal_to([a["name"] for a in delphi_tracklist_track[f"{vendor}_track"]["artists"]]) assert_that(go_tracklist_track["current_position"]).is_equal_to(delphi_tracklist_track["current"] + 1) # assert_that(go_tracklist_track["position_change_date"]).is_equal_to(delphi_tracklist_track["last_date_change_14_days"]) assert_that(go_tracklist_track["added_date"]).is_equal_to(delphi_tracklist_track["earliest_position_date_time"].split("T")[0]) assert_that(go_tracklist_track["isrc"]).is_equal_to(delphi_tracklist_track["isrc"]) assert_is_new(go_tracklist_track, delphi_tracklist_track) if vendor == "spotify": assert_that(go_tracklist_track["track_id"]).is_equal_to(delphi_tracklist_track["track_id"]) elif vendor == "apple": assert_that(str(go_tracklist_track["track_id"])).is_equal_to(delphi_tracklist_track["track_id"]) # if go_tracklist_track["previous_position"] is None: # assert_that(delphi_tracklist_track["previous_position_change_14_days"]).is_none() # else: # assert_that(go_tracklist_track["previous_position"]).is_equal_to(delphi_tracklist_track["previous_position_change_14_days"] + 1) if delphi_tracklist_track["is_top_track_for_isrc"] and delphi_tracklist_track["trend_change_14_days"] != 0: assert_that(go_tracklist_track["trend"]).is_equal_to(delphi_tracklist_track["trend_change_14_days"]) elif not delphi_tracklist_track["is_top_track_for_isrc"] and delphi_tracklist_track["trend_change_14_days"] != 0: assert_that(go_tracklist_track["trend"]).is_none() elif delphi_tracklist_track["is_top_track_for_isrc"] and delphi_tracklist_track["trend_change_14_days"] == 0: assert_that(go_tracklist_track["trend"]).is_none() elif delphi_tracklist_track["is_top_track_for_isrc"]: # log.debug(f"go_tracklist_track: {go_tracklist_track}\ndelphi_tracklist_track: {delphi_tracklist_track}") with allure.step("Get info for Playlist placements header from AGo end-point"): track_data = track_api_call(go_tracklist_track["isrc"], playlist_id, "_gl" if vendor == "spotify" else "us")["body"] # log.debug(f"track_data: {track_data}") with allure.step("Get info for Playlist placements header from Delphi end-point"): delphi_playlist = delphi_api.delphi.get_current_playlist_data(playlist_id=f"{vendor}_{playlist_id}") delphi_playlist_dates = delphi_api.delphi.get_playlist_dates(playlist_id=f"{vendor}_{playlist_id}") delphi_updated_date = [i for i in delphi_playlist_dates["items"] if i["country_code"] == track_data["country_code"]][0]["dsp_sync_date_time"][0] # log.debug(f"delphi_playlist: {delphi_playlist}\ndelphi_playlist_dates: {delphi_playlist_dates}") with allure.step("Compare data from Playlist placements header and tracklist"): assert_that(go_tracklist_track["isrc"]).is_equal_to(track_data["isrc"]) assert_that(go_tracklist_track["added_date"]).is_equal_to(track_data["added_date"]) assert_that(go_tracklist_track["current_position"]).is_equal_to(track_data["current_position"]) assert_that(go_tracklist_track["previous_position"]).is_equal_to(track_data["previous_position"]) assert_that(go_tracklist_track["trend"]).is_equal_to(track_data["trend"]) assert_that(go_tracklist_track["is_new"]).is_equal_to(track_data["is_new"]) with allure.step("Compare data for Playlist placements header from Delphi and AGo"): assert_that(track_data["country_code"]).is_equal_to(delphi_playlist["country_code"]) assert_that(track_data["personalized"]).is_equal_to(delphi_playlist["is_personalised"]) assert_that(track_data["playlist_name"]).is_equal_to(delphi_playlist["name"]) if not track_data["personalized"] and vendor == "spotify": today = date.today() assert_that(track_data["changed_date"]).is_equal_to(str(today)) else: assert_that(track_data["changed_date"]).is_equal_to(delphi_updated_date.split("T")[0])