import logging import random from datetime import datetime as dt 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() data_provider = ApiDataProvider(api) market = None no_response_from_third_request = False spotify_and_hh_playlists = (list(AppConfig.get("playlists_spotify")) + list(AppConfig.get("hot_hits_playlists_spotify")) + AppConfig.get("nmf_playlists_spotify")) class PlaylistTrendsTest: @allure.severity(allure.severity_level.NORMAL) @allure.title("Trend Consistency Validation Spotify") @allure.description("Verify consistency of position trends including Spotify and HH playlists within AGo app for Spotify") @allure.testcase("AG-9473") @pytest.mark.parametrize("playlist_id", spotify_and_hh_playlists) def test_consistency_of_position_trends_for_spotify_and_hot_hits_playlists(self, playlist_id): global no_response_from_third_request is_hh_playlist_id = True if playlist_id in AppConfig.get("hot_hits_playlists_spotify") else False is_nmf_playlist_id = True if playlist_id in AppConfig.get("nmf_playlists_spotify") else False log.debug("First request") tracklist = api.apollo.get_spotify_playlist_tracklist(playlist_id, market, "current_position")["body"]["items"] random_tracks = random.choices(tracklist, k=10) for track_info in random_tracks: track_isrc = track_info["isrc"] trend_from_response_1 = track_info["trend"] is_new_from_response_1 = track_info["is_new"] position_response_1 = track_info["current_position"] added_date_response_1 = track_info["added_date"] log.debug("Second request") track_data_in_playlist = data_provider.get_spotify_track_data_in_playlist(track_isrc, playlist_id, market) trend_from_response_2 = track_data_in_playlist.trend_position is_new_from_response_2 = track_data_in_playlist.is_new position_response_2 = track_data_in_playlist.current_position added_date_response_2 = track_data_in_playlist.added_date try: assert_that(trend_from_response_1).is_equal_to(trend_from_response_2) assert_that(is_new_from_response_1).is_equal_to(is_new_from_response_2) assert_that(position_response_1).is_equal_to(position_response_2) assert_that(added_date_response_1).is_equal_to(added_date_response_2) except AssertionError: print(f"First response: {track_info}\nSecond response: {track_data_in_playlist}\n") raise log.debug("Third request") playlists = api.apollo.get_spotify_playlists_positions_mobile(market, track_isrc, 5000, 0)["body"]["items"] try: playlist_info = next(playlist for playlist in playlists if playlist["id"] == playlist_id) trend_from_response_3 = playlist_info["trend"] is_new_from_response_3 = playlist_info["is_new"] position_response_3 = playlist_info["current_position"] added_date_response_3 = playlist_info["added_date"] except StopIteration: if trend_from_response_1 != 0: log.debug(f"There is no {playlist_id} playlist id") no_response_from_third_request = True else: trend_from_response_3 = 0 is_new_from_response_3 = False if is_hh_playlist_id and not no_response_from_third_request: log.debug("Fourth request") hh_playlists = api.apollo.get_hot_hits_playlists(isrc=track_isrc)["items"] try: hh_playlist_info = next(playlist for playlist in hh_playlists if playlist["playlist_id"] == playlist_id) trend_from_response_4 = -(hh_playlist_info["position_change"]) position_response_4 = hh_playlist_info["position"] added_date_response_4 = hh_playlist_info["entry_date"] if trend_from_response_4 == 0: trend_from_response_4 = None is_new_from_response_4 = hh_playlist_info["is_new"] except StopIteration as e: if trend_from_response_1 != 0: raise e else: trend_from_response_4 = 0 is_new_from_response_4 = False try: assert_that(trend_from_response_3).is_equal_to(trend_from_response_4) assert_that(is_new_from_response_3).is_equal_to(is_new_from_response_4) assert_that(position_response_3).is_equal_to(position_response_4) assert_that(added_date_response_3).is_equal_to(added_date_response_4) except AssertionError: print(f"Third response: {playlist_info}\nFourth response: {hh_playlist_info}") raise if is_nmf_playlist_id: now = dt.utcnow() if now.weekday() in (4, 5, 6) or (now.weekday() == 3 and now.hour >= 13): log.debug("Fifth request") nmf_playlists = api.gate.get_tracks_nmf_playlists(isrc=track_isrc) try: nmf_playlist_info = next( playlist for playlist in nmf_playlists if playlist["playlist_id"] == playlist_id) position_response_5 = nmf_playlist_info["position"] assert_that(position_response_1, f"{track_info}\n{nmf_playlist_info}").is_equal_to( position_response_5) except StopIteration: pass @allure.severity(allure.severity_level.NORMAL) @allure.title("Trend Consistency Validation Apple") @allure.description("Verify consistency of position trends within AGo app for Apple") @allure.testcase("AG-9472") @pytest.mark.parametrize("playlist_id", AppConfig.get("playlists_apple")) def test_consistency_of_position_trends_for_apple(self, playlist_id): market = "us" log.debug("First request") tracks_info = [track for track in api.apollo.get_v1_apple_music_playlist_tracklist( playlist_id, market, "current_position", 100)["body"]["items"] if track["isrc"] is not None] random_tracks = random.choices(tracks_info, k=10) for track_info in random_tracks: track_isrc = track_info["isrc"] trend_from_response_1 = track_info["trend"] is_new_from_response_1 = track_info["is_new"] added_date_from_response_1 = track_info["added_date"] log.debug("Second request") track_data_in_playlist = data_provider.get_apple_track_data_in_playlist(track_isrc, playlist_id, market) trend_from_response_2 = track_data_in_playlist.trend is_new_from_response_2 = track_data_in_playlist.is_new added_date_from_response_2 = track_data_in_playlist.added_date log.debug("Third request") playlists = api.apollo.get_apple_playlists_positions_mobile(market, track_isrc, 1000, 0, False)["body"]["items"] playlist_info = next(playlist for playlist in playlists if playlist["id"] == playlist_id) trend_from_response_3 = playlist_info["trend"] is_new_from_response_3 = playlist_info["is_new"] added_date_from_response_3 = playlist_info["added_date"] try: assert_that(trend_from_response_1).is_equal_to(trend_from_response_2).is_equal_to(trend_from_response_3) assert_that(is_new_from_response_1).is_equal_to(is_new_from_response_2).is_equal_to(is_new_from_response_3) assert_that(added_date_from_response_1).is_equal_to(added_date_from_response_2).is_equal_to(added_date_from_response_3) except AssertionError: print(f"First response: {track_info}\n" f"Second response: {track_data_in_playlist}\n" f"Third response: {playlist_info}") raise