"""Job unit tests.""" import math from collections import defaultdict from datetime import date, datetime, time, timedelta from typing import List, Tuple import pytest import requests_mock from apollo_main_db.apollo.models import ApolloKeyValueStorage from apollo_main_db.spotify.models import (MarketRankTypeEnum, SpotifyMarketRank, SpotifyNewMusicFridayDate, SpotifyNewMusicFridayPlaylistTrackHistory, SpotifyViewPlaylist, ViewPlaylistTypeEnum) from moto import mock_events import config import eventbridge import index from dsp_api import get_playlist_tracks from tests import factories @pytest.mark.parametrize( "current_date,expected_date", ( ((2021, 6, 17), (2021, 6, 18)), ((2021, 6, 18), (2021, 6, 18)), ((2021, 6, 16), (2021, 6, 11)), ((2021, 6, 15), (2021, 6, 11)), ((2021, 6, 14), (2021, 6, 11)), ((2021, 6, 19), (2021, 6, 18)), ((2021, 6, 20), (2021, 6, 18)), ), ) def test_get_friday(current_date, expected_date, mocker): mocked_datetime = mocker.patch("index.datetime") mocked_datetime.utcnow = mocker.Mock(return_value=datetime(*current_date)) assert index.get_friday() == date(*expected_date) def get_tracks(items: List[Tuple[int, str, int]]) -> List[SpotifyNewMusicFridayPlaylistTrackHistory]: return [ SpotifyNewMusicFridayPlaylistTrackHistory( playlist_index=i[0], track_id=i[1], added=datetime(2021, 6, i[2], 15, 21, 54) ) for i in items ] @pytest.mark.parametrize( "db_tracks,new_tracks,expected_result", ( ([], [], True), (get_tracks([(2, "t1", 15), (1, "t2", 14)]), get_tracks([(1, "t2", 14)]), False), (get_tracks([(2, "t1", 15), (1, "t2", 17)]), get_tracks([(1, "t3", 18)]), True), (get_tracks([(2, "t1", 15), (1, "t2", 17)]), get_tracks([(2, "t1", 15), (1, "t3", 18)]), False), (get_tracks([(2, "t1", 15), (1, "t2", 17)]), get_tracks([(2, "t4", 18), (1, "t3", 18)]), True), (get_tracks([(2, "t1", 15), (1, "t2", 17)]), get_tracks([(1, "t2", 18), (2, "t1", 18)]), True), ), ) def test_need_update(db_tracks, new_tracks, expected_result): friday = date(2021, 6, 18) assert index.need_update(db_tracks, new_tracks, friday) == expected_result def spotify_tracks(offset: int = 0, limit: int = 5, has_next: bool = False, day: int = 16) -> dict: return { "items": [ { "current": i, "spotify_track": {"track_id": f"tr_{i}"}, "isrc": f"isrc_{i}", "last_added_date_time": datetime(2021, 6, day, 15, 23, 11).isoformat(), } for i in range(offset, offset + limit) ], "next": has_next, } @pytest.mark.parametrize("page_size,total_count", ((4, 10), (4, 12), (8, 31))) @requests_mock.Mocker(kw="requests_mocker") def test_delphi_pagination(page_size, total_count, **kwargs): """Test for main.""" playlist_id = "pl_id_1" config.DSPAPI.PAGE_SIZE = page_size requests_mocker = kwargs["requests_mocker"] page_count = math.ceil(total_count / page_size) for i in range(page_count): not_last_page = i < page_count - 1 requests_mocker.register_uri( "GET", ( f"{config.DSPAPI.HOST}/api/delphi/public/track-positions/playlists/current-tracklist" + f"?offset={page_size * i}&limit={page_size}&include=track_info&playlist_id=spotify_{playlist_id}" ), json=spotify_tracks( offset=i * page_size, limit=page_size if not_last_page else total_count - page_size * i, has_next=not_last_page, ), ) result = get_playlist_tracks(playlist_id) assert result == spotify_tracks(0, total_count)["items"] @pytest.mark.parametrize("current_date,existing_dates", ((18, (4, 11)), (18, (4, 11, 18)), (18, (4, 18)))) @requests_mock.Mocker(kw="requests_mocker") def test_update_nmf_tracklist(current_date, existing_dates, mocker, db_session, **kwargs): current_date = date(2021, 6, current_date) for i in range(1, 6): factories.ApolloMarketFactory.create(id=i, code=f"m_{i}") factories.SpotifyViewPlaylistFactory.create( type=ViewPlaylistTypeEnum.HH, playlist_id=f"pl_{i}", market_code=f"m_{i}", last_date=date(2021, 2, 10), last_added_ts=datetime(2021, 2, 9, 10, 50, 31), ) factories.SpotifyMarketRankFactory.create(type=MarketRankTypeEnum.HH, market_code=f"m_{i}", rank=i) playlists = [ factories.SpotifyViewPlaylistFactory.create( type=ViewPlaylistTypeEnum.NMF, playlist_id=f"pl_{i}", market_code=f"m_{i}", last_date=date(2021, 6, 10), last_added_ts=datetime(2021, 6, 9, 10, 50, 31), ) for i in range(1, 6) ] ranks = [ factories.SpotifyMarketRankFactory.create(type=MarketRankTypeEnum.NMF, market_code=f"m_{i}", rank=i) for i in range(1, 6) ] mocked_get_friday = mocker.patch("index.get_friday") mocked_get_friday.return_value = current_date requests_mocker = kwargs["requests_mocker"] spotify_records_mapping = {} spotify_count = 0 for i, playlist in enumerate(playlists): spotify_count += 2 + i spotify_records = spotify_tracks(limit=(2 + i), day=(current_date.day - 1)) spotify_records_mapping[playlist.playlist_id] = { r["spotify_track"]["track_id"]: r for r in spotify_records["items"] } requests_mocker.register_uri( "GET", ( f"{config.DSPAPI.HOST}/api/delphi/public/track-positions/playlists/current-tracklist" + f"?offset=0&limit={config.DSPAPI.PAGE_SIZE}" + f"&include=track_info&playlist_id=spotify_{playlist.playlist_id}" ), json=spotify_records, ) history_records = defaultdict(lambda: defaultdict(dict)) history_count = 0 for i, day in enumerate(existing_dates): nmf_date = date(2021, 6, day) factories.SpotifyNewMusicFridayDateFactory(id=(i + 1), date=nmf_date) for j in range(i, len(playlists)): for k in range(1, 5): history_records[i + 1][playlists[j].playlist_id][k] = ( factories.SpotifyNewMusicFridayPlaylistTrackHistoryFactory.create( date_id=(i + 1), playlist_id=playlists[j].playlist_id, playlist_index=k, track_id=f"tr_{i + j + k}", added=datetime.combine(nmf_date, time(12, i + j + k)), isrc=f"isrc_{i + j + k}", ) ) if day != current_date.day: history_count += 1 index.update_nmf_tracklist() db_playlists = ( db_session.query(SpotifyViewPlaylist) .filter(SpotifyViewPlaylist.type == ViewPlaylistTypeEnum.NMF) .order_by(SpotifyViewPlaylist.market_code) .all() ) assert len(db_playlists) == len(playlists) for i in range(len(playlists)): db_playlist = db_playlists[i] playlist = playlists[i] assert db_playlist.playlist_id == playlist.playlist_id assert db_playlist.market_code == playlist.market_code assert db_playlist.last_date == current_date assert db_playlist.last_added_ts == datetime(2021, 6, current_date.day - 1, 15, 23, 11) db_ranks = ( db_session.query(SpotifyMarketRank) .filter(SpotifyMarketRank.type == MarketRankTypeEnum.NMF) .order_by(SpotifyMarketRank.market_code) .all() ) assert len(db_ranks) == len(ranks) for i in range(len(ranks)): db_rank = db_ranks[i] rank = ranks[i] assert db_rank.type == MarketRankTypeEnum.NMF assert db_rank.market_code == rank.market_code assert db_rank.rank == rank.rank db_dates = db_session.query(SpotifyNewMusicFridayDate).order_by(SpotifyNewMusicFridayDate.date).all() assert [i.date for i in db_dates] == [ date(2021, 6, i) for i in (list(sorted(existing_dates)) + ([current_date.day] if current_date.day not in existing_dates else [])) ] last_date_id = max(i.id for i in db_dates) db_history_records = ( db_session.query(SpotifyNewMusicFridayPlaylistTrackHistory) .order_by( SpotifyNewMusicFridayPlaylistTrackHistory.date_id, SpotifyNewMusicFridayPlaylistTrackHistory.playlist_id, SpotifyNewMusicFridayPlaylistTrackHistory.playlist_index, ) .all() ) assert len(db_history_records) == (history_count + spotify_count) for db_history in db_history_records: if db_history.date_id != last_date_id: history = history_records[db_history.date_id][db_history.playlist_id][db_history.playlist_index] assert db_history.track_id == history.track_id assert db_history.added == history.added assert db_history.isrc == history.isrc else: record = spotify_records_mapping[db_history.playlist_id][db_history.track_id] assert db_history.isrc == record["isrc"] assert db_history.added == datetime.fromisoformat(record["last_added_date_time"]) assert db_history.playlist_index == list(spotify_records_mapping[db_history.playlist_id].keys()).index( db_history.track_id ) def get_ts( year: int = 2021, month: int = 6, day: int = 14, hour: int = 15, minute: int = 34, second: int = 51 ) -> datetime: return datetime(year, month, day, hour, minute, second) @pytest.mark.parametrize( "updated_ts,current_ts,update_interval,expected_ts,expected_ranks", ( (get_ts(day=14), get_ts(day=16), 4, get_ts(day=14), {"es": 1, "mArKeT1": 2, "US": 3, "Market2": 4, "gb": 5}), (get_ts(day=14), get_ts(day=16), 1, get_ts(day=16), {"US": 1, "gb": 2, "es": 3, "mArKeT1": 51, "Market2": 52}), ), ) def test_update_nmf_ranks(updated_ts, current_ts, update_interval, expected_ts, expected_ranks, mocker, db_session): streams_markets = ["us", "gb", "ca", "es"] playlists_markets = ["es", "mArKeT1", "US", "Market2", "gb"] mocked_datetime = mocker.patch.object(index, "datetime", mocker.Mock(wraps=datetime)) mocked_datetime.utcnow.return_value = current_ts config.UPDATE_RANKS_INTERVAL = update_interval config.UNKNOWN_MARKET_RANK_FROM = 50 factories.ApolloKeyValueStorageFactory.create(key="nmf_rank_update_ts", value=updated_ts.isoformat()) for i, market in enumerate(playlists_markets): factories.ApolloMarketFactory.create(id=(i + 1), code=market) factories.SpotifyViewPlaylistFactory.create(type=ViewPlaylistTypeEnum.NMF, market_code=market) factories.SpotifyMarketRankFactory.create(type=MarketRankTypeEnum.NMF, market_code=market, rank=(i + 1)) factories.SpotifyMarketRankFactory.create(type=MarketRankTypeEnum.HH, market_code=market, rank=(i + 5)) for i in range(1, 20): for j, market in enumerate(streams_markets): factories.SpotifyAnalyticsAccountStreamInfoFactory.create( account=1, market=market, date=current_ts.date() - timedelta(days=i), total_streams=(1000 - j * 10 + i) ) index.update_nmf_ranks() result_value = ( db_session.query(ApolloKeyValueStorage.value).filter(ApolloKeyValueStorage.key == "nmf_rank_update_ts").first() )[0] assert result_value == expected_ts.isoformat() db_ranks = ( db_session.query(SpotifyMarketRank) .filter(SpotifyMarketRank.type == MarketRankTypeEnum.NMF) .order_by(SpotifyMarketRank.rank) .all() ) assert {i.market_code: i.rank for i in db_ranks} == expected_ranks @pytest.mark.parametrize( "eventbus_name,instance_id,playlists,ranks,expected_result", ( (None, None, None, None, None), ("", None, None, None, None), ("test_bus", None, None, None, True), ("test_bus", "", [], ([], []), True), ("test_bus", "test_id" * 10, ["pl_1", "pl_2"], (["pl_10", "pl_20"], ["pl_11"]), True), ), ) @mock_events def test_put_event(eventbus_name, instance_id, playlists, ranks, expected_result): config.AWS_EVENT_BUS_NAME = eventbus_name if eventbus_name: eventbridge.get_client().create_event_bus(Name=eventbus_name) result = eventbridge.put_event(instance_id, playlists, ranks) assert result == expected_result