import datetime import json from collections import namedtuple from sme_logger import get_logger from apollo_main_db.push_notifications.models import PushMessage, TopicEnum from apple_charts_notifications import utils from apple_charts_notifications.config import Config from apple_charts_notifications.constants import CHART_NAME from apple_charts_notifications.handler import process_added_tracks, process_dropped_tracks, \ process_major_moves_of_tracks from apollo_push_client import PushClient from tests import factories Song = namedtuple('Song', ['name', 'id', 'artist_name', 'isrc']) def get_formatted_today_yesterday(): today = datetime.date.today() yesterday = today - datetime.timedelta(days=1) return today.strftime("%Y-%m-%d"), yesterday.strftime("%Y-%m-%d") def test_get_added_tracks(db_session): today, yesterday = get_formatted_today_yesterday() market = 'us' yesterday_chart = factories.AppleMusicChartFactory.create( storefront=market, chart_name=CHART_NAME, date=yesterday) today_chart = factories.AppleMusicChartFactory.create( storefront=market, chart_name=CHART_NAME, date=today) existing_song = Song(name='song 1', id=1, artist_name='artist 1', isrc='USRC11901801') new_song = Song(name='song 2', id=2, artist_name='artist 2', isrc='USRC11901802') factories.AppleMusicSongFactory.create(id=existing_song.id, isrc=existing_song.isrc) factories.AppleMusicSongFactory.create(id=new_song.id, isrc=new_song.isrc) # create charts entities factories.AppleMusicChartSongEntryFactory.create( chart_id=yesterday_chart.id, position=1, song_id=existing_song.id, artist_name=existing_song.artist_name, name=existing_song.name ) factories.AppleMusicChartSongEntryFactory.create( chart_id=today_chart.id, position=1, song_id=existing_song.id, artist_name=existing_song.artist_name, name=existing_song.name ) factories.AppleMusicChartSongEntryFactory.create( chart_id=today_chart.id, position=2, song_id=new_song.id, artist_name=new_song.artist_name, name=new_song.name ) added_tracks = utils.get_diff_tracks_by_dates(db_session, market, today, yesterday).all() assert len(added_tracks) == 1 assert added_tracks[0].song_id == new_song.id assert added_tracks[0].position == 3 assert added_tracks[0].isrc == new_song.isrc def test_get_dropped_tracks(db_session): today, yesterday = get_formatted_today_yesterday() market = 'us' yesterday_chart = factories.AppleMusicChartFactory.create( storefront=market, chart_name=CHART_NAME, date=yesterday) today_chart = factories.AppleMusicChartFactory.create( storefront=market, chart_name=CHART_NAME, date=today) existing_song = Song(name='song 1', id=1, artist_name='artist 1', isrc='USRC11901801') dropped_song = Song(name='song 2', id=2, artist_name='artist 2', isrc='USRC11901802') factories.AppleMusicSongFactory.create(id=existing_song.id, isrc=existing_song.isrc) factories.AppleMusicSongFactory.create(id=dropped_song.id, isrc=dropped_song.isrc) # create charts entities factories.AppleMusicChartSongEntryFactory.create( chart_id=yesterday_chart.id, position=1, song_id=existing_song.id, artist_name=existing_song.artist_name, name=existing_song.name ) factories.AppleMusicChartSongEntryFactory.create( chart_id=yesterday_chart.id, position=2, song_id=dropped_song.id, artist_name=dropped_song.artist_name, name=dropped_song.name ) factories.AppleMusicChartSongEntryFactory.create( chart_id=today_chart.id, position=1, song_id=existing_song.id, artist_name=existing_song.artist_name, name=existing_song.name ) dropped_song_tracks = utils.get_diff_tracks_by_dates(db_session, market, start_date=yesterday, end_date=today).all() assert len(dropped_song_tracks) == 1 assert dropped_song_tracks[0].song_id == dropped_song.id assert dropped_song_tracks[0].position == 3 assert dropped_song_tracks[0].isrc == dropped_song.isrc def test_get_major_moves_of_tracks(db_session): today, yesterday = get_formatted_today_yesterday() market = 'us' yesterday_chart = factories.AppleMusicChartFactory.create( storefront=market, chart_name=CHART_NAME, date=yesterday) today_chart = factories.AppleMusicChartFactory.create( storefront=market, chart_name=CHART_NAME, date=today) song = Song(name='song 1', id=1, artist_name='artist 1', isrc='USRC11901902') factories.AppleMusicSongFactory.create(id=song.id, isrc=song.isrc) # create charts entities factories.AppleMusicChartSongEntryFactory.create( chart_id=yesterday_chart.id, position=5, song_id=song.id, artist_name=song.artist_name, name=song.name ) factories.AppleMusicChartSongEntryFactory.create( chart_id=today_chart.id, position=0, song_id=song.id, artist_name=song.artist_name, name=song.name ) major_moves = utils.get_major_moves(db_session, market, start_date=today, end_date=yesterday).all() assert len(major_moves) == 1 assert major_moves[0].song_id == song.id assert major_moves[0].position == 1 assert major_moves[0].moves == -5 assert major_moves[0].isrc == song.isrc def test_get_only_starred_tracks(db_session): song = Song(name='song 1', id=1, artist_name='artist 1', isrc='USRC11901902') starred_song = Song(name='song 2', id=2, artist_name='artist 2', isrc='usrc11901901') factories.AppleMusicSongFactory.create(id=song.id, isrc=song.isrc) factories.AppleMusicSongFactory.create(id=starred_song.id, isrc=starred_song.isrc) market = factories.MarketFactory.create(market_name='us') today, _ = get_formatted_today_yesterday() # create chart and chart entities chart = factories.AppleMusicChartFactory.create( storefront=market.market_name, chart_name=CHART_NAME, date=today) factories.AppleMusicChartSongEntryFactory.create( chart_id=chart.id, position=1, song_id=song.id, artist_name=song.artist_name, name=song.name ) factories.AppleMusicChartSongEntryFactory.create( chart_id=chart.id, position=0, song_id=starred_song.id, artist_name=starred_song.artist_name, name=starred_song.name ) # mark track as starred user_id = 'user_id' factories.UserMarketFactory.create( user_id=user_id, market_id=market.market_id) factories.StarredContentFactory.create( user_id=user_id, uri=starred_song.id, isrc=starred_song.isrc) tracks_query = utils.get_base_query(db_session, market.market_name, date=today).subquery() factories.UserDeviceFactory.create_batch( size=2, user_id=user_id, is_active=False) factories.UserDeviceFactory.create_batch( size=3, user_id=user_id, is_active=True) starred_tracks = utils.get_only_starred_tracks_with_device_tokens( db_session, tracks_query, market.market_name).all() assert len(starred_tracks) == 1 assert starred_tracks[0].user_id == user_id assert starred_tracks[0].song_id == starred_song.id assert starred_tracks[0].position == 1 assert len(starred_tracks[0].device_tokens.split(',')) == 3 def test_process_added_tracks(db_session, mocked_sqs, redis_client): user_id = 'user_id' market = factories.MarketFactory.create(market_name='us') factories.UserMarketFactory.create( user_id=user_id, market_id=market.market_id) factories.UserDeviceFactory.create_batch( size=3, user_id=user_id, is_active=True) # create charts and chart entities today, yesterday = get_formatted_today_yesterday() factories.AppleMusicChartFactory.create( storefront=market.market_name, chart_name=CHART_NAME, date=yesterday) chart = factories.AppleMusicChartFactory.create( storefront=market.market_name, chart_name=CHART_NAME, date=today) # set up un starred song song = Song(name='song 1', id=1, artist_name='artist 1', isrc='USRC11901903') factories.AppleMusicSongFactory.create(id=song.id, isrc=song.isrc) factories.AppleMusicChartSongEntryFactory.create( chart_id=chart.id, position=1, song_id=song.id, artist_name=song.artist_name, name=song.name ) # setup starred songs for i in range(2, 22): starred_song = Song(name=f'song {i}', id=i, artist_name=f'artist {i}', isrc=f'usrc119018{i:02d}') factories.AppleMusicSongFactory.create(id=starred_song.id, isrc=starred_song.isrc) factories.AppleMusicChartSongEntryFactory.create( chart_id=chart.id, position=i, song_id=starred_song.id, artist_name=starred_song.artist_name, name=starred_song.name ) # mark track as starred factories.StarredContentFactory.create( user_id=user_id, uri=starred_song.id, isrc=starred_song.isrc) logger = get_logger('test') config = Config() push_client = PushClient( logger_client=logger, sqs_client=mocked_sqs, db_session=db_session, config=config ) process_added_tracks(market.market_name, today, yesterday, push_client, redis_client, db_session) push_messages = db_session.query(PushMessage).all() assert len(push_messages) == 20 for push in push_messages: assert push.topic == TopicEnum.CHART_ADDITIONS def test_process_dropped_tracks(db_session, mocked_sqs, redis_client): user_id = 'user_id' market = factories.MarketFactory.create(market_name='us') factories.UserMarketFactory.create( user_id=user_id, market_id=market.market_id) factories.UserDeviceFactory.create_batch( size=3, user_id=user_id, is_active=True) # create charts and chart entities today, yesterday = get_formatted_today_yesterday() yesterday_chart = factories.AppleMusicChartFactory.create( storefront=market.market_name, chart_name=CHART_NAME, date=yesterday) today_chart = factories.AppleMusicChartFactory.create( storefront=market.market_name, chart_name=CHART_NAME, date=today) # set up un starred song song = Song(name='song 1', id=1, artist_name='artist 1', isrc='USRC11901903') factories.AppleMusicSongFactory.create(id=song.id, isrc=song.isrc) factories.AppleMusicChartSongEntryFactory.create( chart_id=today_chart.id, position=1, song_id=song.id, artist_name=song.artist_name, name=song.name ) # setup starred songs for i in range(2, 22): starred_song = Song(name=f'song {i}', id=i, artist_name=f'artist {i}', isrc=f'USRC119018{i:02d}') factories.AppleMusicSongFactory.create(id=starred_song.id, isrc=starred_song.isrc) factories.AppleMusicChartSongEntryFactory.create( chart_id=yesterday_chart.id, position=i, song_id=starred_song.id, artist_name=starred_song.artist_name, name=starred_song.name ) # mark track as starred factories.StarredContentFactory.create( user_id=user_id, uri=starred_song.id, isrc=starred_song.isrc) logger = get_logger('test') config = Config() push_client = PushClient( logger_client=logger, sqs_client=mocked_sqs, db_session=db_session, config=config ) process_dropped_tracks(market.market_name, today, yesterday, push_client, redis_client, db_session) push_messages = db_session.query(PushMessage).all() assert len(push_messages) == 20 for push in push_messages: assert push.topic == TopicEnum.CHART_REMOVALS def test_process_major_moves(db_session, mocked_sqs, redis_client): user_id = 'user_id' market = factories.MarketFactory.create(market_name='us') factories.UserMarketFactory.create( user_id=user_id, market_id=market.market_id) factories.UserDeviceFactory.create_batch( size=3, user_id=user_id, is_active=True) # create charts and chart entities today, yesterday = get_formatted_today_yesterday() yesterday_chart = factories.AppleMusicChartFactory.create( storefront=market.market_name, chart_name=CHART_NAME, date=yesterday) today_chart = factories.AppleMusicChartFactory.create( storefront=market.market_name, chart_name=CHART_NAME, date=today) # set up un starred song song = Song(name='song 1', id=1, artist_name='artist 1', isrc='USRC11901802') factories.AppleMusicSongFactory.create(id=song.id, isrc=song.isrc) factories.StarredContentFactory.create(user_id=user_id, uri=song.id, isrc=song.isrc) factories.AppleMusicChartSongEntryFactory.create( chart_id=yesterday_chart.id, position=1, song_id=song.id, artist_name=song.artist_name, name=song.name ) factories.AppleMusicChartSongEntryFactory.create( chart_id=today_chart.id, position=10, song_id=song.id, artist_name=song.artist_name, name=song.name ) logger = get_logger('test') config = Config() push_client = PushClient( logger_client=logger, sqs_client=mocked_sqs, db_session=db_session, config=config ) process_major_moves_of_tracks( market.market_name, today, yesterday, push_client, redis_client, db_session) push_messages = db_session.query(PushMessage).all() assert len(push_messages) == 1 for push in push_messages: data = json.loads(push.data) assert push.topic == TopicEnum.CHART_MAJOR_MOVES assert data['change']