"""Unit tests for Shazam Statistics Workflow.""" import asyncio import datetime from asynctest import CoroutineMock, MagicMock, patch from unittest.mock import AsyncMock import nest_asyncio nest_asyncio.apply() import pytest from unittest import mock from freezegun import freeze_time import shazam_with_shazamio as shazam from conftest import artist_last_stats, artist_statistics_from_table, processed_links, \ select_artists_ids, select_tracks_ids, track_statistics_from_table class TestShazamTracker: """Test Shazam Tracker related methods.""" def test_get_id_from_link(self): """Test get_id_from_link method with reverse parameter also.""" id = shazam.get_id_from_link(processed_links()['shazam']) assert id == 207950953 @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_artists_ids', select_artists_ids) @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_tracks_ids', select_tracks_ids) @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_artists_stats', artist_statistics_from_table) @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_tracks_stats', track_statistics_from_table) def test_create_tables_for_the_first_time( self, mock_delete_from_staging_raw, mock_create_staging_raw_shazam_statistics, mock_create_shazam_statistics, mock_create_shazam_tracks_statistics, mock_create_staging_raw_shazam_tracks_statistics, mock_executor): """Test create_tables method working correctly during the very first run.""" mock_create_shazam_tracks_statistics.return_value, \ mock_create_staging_raw_shazam_tracks_statistics.return_value, \ mock_create_shazam_statistics.return_value, \ mock_create_staging_raw_shazam_statistics.return_value = [['']] * 4 tracker = shazam.ShazamTracker() tracker.create_tables() assert mock_create_shazam_statistics.called assert mock_delete_from_staging_raw.called assert mock_create_staging_raw_shazam_statistics.called assert tracker.__first_time__ assert 'cijikin' in tracker.get_profiles().keys() assert tracker.get_profiles()['cijikin']['id'] == 207950953 assert 3553207950953 not in tracker.get_tracks() @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_artists_ids', select_artists_ids) @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_tracks_ids', select_tracks_ids) @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_artists_stats', artist_statistics_from_table) @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_tracks_stats', track_statistics_from_table) def test_create_tables_not_for_the_first_time( self, mock_delete_from_staging_raw, mock_create_staging_raw_shazam_statistics, mock_create_shazam_statistics, mock_create_shazam_tracks_statistics, mock_create_staging_raw_shazam_tracks_statistics, mock_executor): """Test create_tables method working correctly if the run is not first.""" mock_create_shazam_tracks_statistics.return_value, \ mock_create_staging_raw_shazam_tracks_statistics.return_value, \ mock_create_shazam_statistics.return_value, \ mock_create_staging_raw_shazam_statistics.return_value = [['already exists']] * 4 tracker = shazam.ShazamTracker() tracker.create_tables() assert mock_create_shazam_statistics.called assert mock_create_staging_raw_shazam_statistics.called assert mock_delete_from_staging_raw.called assert mock_create_shazam_tracks_statistics.called assert mock_create_staging_raw_shazam_tracks_statistics.called assert not tracker.__first_time__ assert 'cijikin' in tracker.get_profiles().keys() assert tracker.get_profiles()['cijikin']['id'] == 207950953 assert 3553207950953 in tracker.get_tracks() @freeze_time('1946-10-01') @mock.patch('shazam_with_shazamio.ShazamStatsSFExecutor.select_weekly_change') def test_compare_current_and_last_week_stats_for_weekly_statistics( self, mock_select_weekly_change, mock_executor): """Test correctness of compare_current_and_last_week_stats method calculations.""" def weekly_side_effect(*args, **kwargs): if -6 in list(kwargs.values()): return {'shazams': 34, 'tracks_count': 0} else: return {'shazams': 51, 'tracks_count': 0} mock_select_weekly_change.side_effect = weekly_side_effect current_shazams = {'shazams': 215} tracker = shazam.ShazamTracker() calculations = tracker.compare_current_and_last_week_stats( current_shazams, 'shazams', artist_last_stats(), 'cijikin') artist_lasts = artist_last_stats()['cijikin'] calculations_to_assert = [ calculations['last_week_shazams'] == artist_lasts['last_week_shazams'], calculations['daily_change_in_shazams'] == current_shazams['shazams'] - artist_lasts['current_shazams'], calculations['weekly_change_in_shazams'] == calculations['daily_change_in_shazams'] + weekly_side_effect('cijikin')['shazams'], calculations['percentage_change_in_shazams'] == artist_lasts['percentage_change_in_shazams'], calculations['acceleration_shazams'] == \ (calculations['daily_change_in_shazams'] + weekly_side_effect('cijikin')['shazams'] - weekly_side_effect(first_day_of_period=-6)['shazams']) / weekly_side_effect(first_day_of_period=-6)[ 'shazams'], calculations['last_processing_date'] == datetime.date(1946, 10, 1), calculations['last_week_processing_date'] == artist_lasts['last_week_processing_date']] for calculation in calculations_to_assert: assert calculation @freeze_time('1946-10-01') @mock.patch('shazam_with_shazamio.ShazamStatsSFExecutor.select_weekly_change') def test_compare_current_and_last_week_stats_for_week_period_statistics( self, mock_select_weekly_change, mock_executor): """Test correctness of compare_current_and_last_week_stats method calculations for shazams.""" current_shazams = {'shazams': 215} tracker = shazam.ShazamTracker() def weekly_side_effect(*args, **kwargs): if -6 in list(kwargs.values()): return {'shazams': 34, 'tracks_count': 0} else: return {'shazams': 51, 'tracks_count': 0} mock_select_weekly_change.side_effect = weekly_side_effect calculations = tracker.compare_current_and_last_week_stats( current_shazams, 'shazams', artist_last_stats( last_week_processing_date=datetime.date.today() - datetime.timedelta(days=7)), 'cijikin') artist_lasts = artist_last_stats( last_week_processing_date=datetime.date.today() - datetime.timedelta(days=7))['cijikin'] # checking the calculation when week period ends assert mock_select_weekly_change.called assert calculations == { 'current_shazams': current_shazams['shazams'], 'last_week_shazams': artist_lasts['current_shazams'], 'daily_change_in_shazams': current_shazams['shazams'] - artist_lasts[ 'current_shazams'], 'weekly_change_in_shazams': calculations['daily_change_in_shazams'] + weekly_side_effect('cijikin')['shazams'], 'percentage_change_in_shazams': (calculations['daily_change_in_shazams'] + weekly_side_effect('cijikin')[ 'shazams']) / artist_lasts['last_week_shazams'], 'acceleration_shazams': \ (calculations['daily_change_in_shazams'] + weekly_side_effect('cijikin')['shazams'] - weekly_side_effect(first_day_of_period=-6)['shazams']) / weekly_side_effect(first_day_of_period=-6)[ 'shazams'], 'last_processing_date': datetime.date(1946, 10, 1), 'last_week_processing_date': artist_lasts['last_week_processing_date']} @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_artists_ids', select_artists_ids) @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_tracks_ids', select_tracks_ids) @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_artists_stats', artist_statistics_from_table) @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_tracks_stats', track_statistics_from_table) @mock.patch("shazam_with_shazamio.ShazamTracker.get_artist_top_tracks") @pytest.mark.asyncio async def test_get_tracks_(self, mock_get_artist_top_tracks, mock_delete_from_staging_raw, mock_create_shazam_tracks_statistics, mock_create_shazam_statistics, mock_create_staging_raw_shazam_tracks_statistics, mock_create_staging_raw_shazam_statistics, mock_executor): tracks_ = [ {'name': 'Holy Water', 'id': 510117575}, {'name': 'Smoke Break', 'id': 565446159} ] f = asyncio.Future() f.set_result(tracks_) mock_get_artist_top_tracks.return_value = f mock_create_shazam_tracks_statistics.return_value, \ mock_create_staging_raw_shazam_tracks_statistics.return_value, \ mock_create_shazam_statistics.return_value, \ mock_create_staging_raw_shazam_statistics.return_value = [['already exists']] * 4 tracker = shazam.ShazamTracker() tracker.create_tables() tracks, artist_id = tracker.get_tracks_('cijikin', log=True) assert artist_id == 207950953 assert tracks == tracks_ @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_artists_ids', select_artists_ids) @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_tracks_ids', select_tracks_ids) @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_artists_stats', artist_statistics_from_table) @mock.patch('snowflake_executor.ShazamStatsSFExecutor.select_tracks_stats', track_statistics_from_table) @mock.patch('shazam_with_shazamio.ShazamStatsSFExecutor.select_weekly_change') @mock.patch("shazam_with_shazamio.ShazamTracker.get_listenings_count") @pytest.mark.asyncio async def test_get_tracks_(self, mock_get_listenings_count, mock_select_weekly_change, mock_delete_from_staging_raw, mock_create_shazam_tracks_statistics, mock_create_shazam_statistics, mock_insert_tracks_stats, mock_create_staging_raw_shazam_tracks_statistics, mock_create_staging_raw_shazam_statistics, mock_compare_current_and_last_week_stats, mock_executor): f = asyncio.Future() f.set_result(91572) mock_get_listenings_count.return_value = f mock_create_shazam_tracks_statistics.return_value, \ mock_create_staging_raw_shazam_tracks_statistics.return_value, \ mock_create_shazam_statistics.return_value, \ mock_create_staging_raw_shazam_statistics.return_value = [['already exists']] * 4 def weekly_side_effect(*args, **kwargs): if -6 in list(kwargs.values()): return {'shazams': 34, 'tracks_count': 0} else: return {'shazams': 51, 'tracks_count': 0} mock_select_weekly_change.side_effect = weekly_side_effect tracker = shazam.ShazamTracker() tracker.create_tables() tracker.get_common_stats('cijikin') assert mock_compare_current_and_last_week_stats.call_count == 8 assert mock_get_listenings_count.called