"""Unit tests for the tasks.""" from datetime import datetime from datetime import timedelta from unittest.mock import call, MagicMock, patch import pytest from activity_detector.flows.spike_detector import config from activity_detector.flows.spike_detector import tasks from activity_detector.flows.spike_detector.notifications import SpikedTrack from activity_detector.utils import dynamodb from activity_detector.utils import ows_assets @pytest.yield_fixture def mock_sql_loader(): """Yield executor context.""" sf_sql_loader_path = ( 'activity_detector.flows.spike_detector.tasks.sql_loader') with patch(sf_sql_loader_path) as sql_loader: yield sql_loader @pytest.yield_fixture def mock_construct_id_filter(): """Yield construct id filter.""" with patch( 'activity_detector.flows.spike_detector.' 'tasks.construct_id_filter') as construct_id_filter: construct_id_filter.side_effect = \ lambda label_ids, subaccount_ids: { 'labelid_filter': label_ids, 'subaccountid_filter': subaccount_ids, 'logger_info': ''} yield construct_id_filter @pytest.yield_fixture def mock_config(): """Mock config.""" with patch( 'activity_detector.flows.spike_detector.tasks.config') as mconfig: mconfig.DEFAULT_WINDOW_DAYS = 91 mconfig.MIN_TRACK_STREAMS = 201 mconfig.MAX_TRACKS_PER_ACCOUNT = 5001 mconfig.MAX_TRACKS_PER_ACCOUNT = 11 mconfig.MIN_RELEASE_DATE_SHIFT_DAYS = 30 mconfig.Z_SCORE_THRESHOLD = 4.6 mconfig.ACTIVITY_TYPE_NAME = 'spite_detector_test' mconfig.STATUS_PROCESSED = 'oops' mconfig.STATUS_PROCESSED_NOTIF_SENT = 'oops_sent' mconfig.MIN_STREAM_GAIN = 2001 mconfig.PLAYLIST_PLACEMENTS_SHIFT_DAYS = 7 mconfig.MAX_PLAYLISTS_TO_SHOW = 10 mconfig.PLAYLIST_STREAMS_THRESHOLD = 0.1 mconfig.MAX_SPIKES_PER_ACCOUNT = 5 mconfig.STORES = { 1: 'apple', 2: 'spotify', 3: 'tiktok' } mconfig.env = 'prod' yield mconfig @pytest.yield_fixture def mock_session_context(): """Yield session context.""" pth = 'activity_detector.flows.spike_detector.tasks.get_session' with patch(pth) as gt_session: mock_session_context = \ gt_session.return_value.__enter__.return_value yield mock_session_context @pytest.yield_fixture def mock_send_notifications(): """Mock send_notifications.""" p = 'activity_detector.flows.spike_detector.tasks.send_notifications' with patch(p, autospec=True) as m: yield m @pytest.yield_fixture def mock_spike_notification(): """Mock send_notifications.""" p = 'activity_detector.flows.spike_detector.tasks.SpikeNotification' with patch(p) as m: yield m @pytest.fixture(params=['label', 'subaccount']) def account_level(request): """Parametrize account level.""" return request.param @pytest.mark.parametrize( 'label_ids,subaccount_ids', [(None, None), (7123, 134), (7123, None), (None, 134)]) def test_create_temp_activity_history_table( mock_session_context, mock_config, mock_sql_loader, mock_construct_id_filter, label_ids, subaccount_ids): """Test create_temp_activity_history_table function for all ids.""" def loader_side_effect(query_name): return query_name + '_{labelid_filter}_{subaccountid_filter}' target_date_str = '2018-06-29' start_date_str = '2018-06-19' run_id = 'x' source_type = 'streams' mock_sql_loader.load_query.side_effect = loader_side_effect tasks.create_temp_activity_history_table( MagicMock(), target_date_str, start_date_str, run_id, source_type, label_ids, subaccount_ids) expected_calls = [call( 'create_temp_activity_history_streams_{}_{}'.format( label_ids, subaccount_ids), { 'label_ids': label_ids, 'subaccount_ids': subaccount_ids, 'target_date': target_date_str, 'start_date': start_date_str, 'store_ids': list(mock_config.STORES), 'min_track_streams': mock_config.MIN_TRACK_STREAMS, 'max_tracks_per_account': mock_config.MAX_TRACKS_PER_ACCOUNT, 'z_score_threshold': mock_config.Z_SCORE_THRESHOLD, 'min_release_date_shift_days': ( mock_config.MIN_RELEASE_DATE_SHIFT_DAYS), 'min_stream_gain': mock_config.MIN_STREAM_GAIN, 'run_id': run_id })] mock_session_context.execute.assert_has_calls(expected_calls) def test_populate_activity_history_table(mock_session_context, mock_sql_loader): """Test populate_activity_history_table.""" def loader_side_effect(query_name): return query_name mock_sql_loader.load_query.side_effect = loader_side_effect tasks.populate_activity_history_table(MagicMock()) expected_calls = [call('load_activity_history')] mock_session_context.execute.assert_has_calls(expected_calls) def test_drop_temp_table(mock_session_context, mock_sql_loader): """Test populate_activity_history_table.""" def loader_side_effect(query_name): return query_name mock_sql_loader.load_query.side_effect = loader_side_effect tasks.drop_temp_table(MagicMock()) expected_calls = [call('drop_temp_table')] mock_session_context.execute.assert_has_calls(expected_calls) def test_construct_id_filter(): """Test construct_id_filter.""" assert tasks.construct_id_filter(7123, 134) == { 'labelid_filter': 'fa.labelid IN (:label_ids)', 'subaccountid_filter': 'fa.subaccountid IN (:subaccount_ids)', 'logger_info': 'Creating spikes table for label ids: {l_ids} and ' 'subaccount ids: {s_ids}'.format(l_ids=7123, s_ids=134) } assert tasks.construct_id_filter(7123, None) == { 'labelid_filter': 'fa.labelid IN (:label_ids)', 'subaccountid_filter': 'False', 'logger_info': 'Creating spikes table for label ids: ' '{}'.format(7123) } assert tasks.construct_id_filter(None, 134) == { 'labelid_filter': 'False', 'subaccountid_filter': 'fa.subaccountid IN (:subaccount_ids)', 'logger_info': 'Creating spikes table for subaccount ids: ' '{}'.format(134) } assert tasks.construct_id_filter(None, None) == { 'labelid_filter': 'True', 'subaccountid_filter': 'False', 'logger_info': 'Creating spikes table for all accounts' } def test_detect_spikes( monkeypatch, mock_session_context, mock_sql_loader, mock_config, mock_send_notifications, account_level): """Test detect_spikes function.""" def loader_side_effect(query_name): return query_name target_date_str = '2018-06-29' target_date = datetime.strptime(target_date_str, '%Y-%m-%d') window_days = mock_config.DEFAULT_WINDOW_DAYS dlt = timedelta(days=window_days) start_date = target_date - dlt start_date_str = start_date.strftime('%Y-%m-%d') ids = [1, 2, 3] run_id = 'x' monkeypatch.setattr(dynamodb, 'set_status', MagicMock(return_value=None)) mock_sql_loader.load_query.side_effect = loader_side_effect mock_session_context.execute.side_effect = [( MagicMock( storeid=1, artist_names='["oops"]', recent_playlist_placements='[]'), MagicMock( storeid=2, artist_names='["boo", "foo"]', recent_playlist_placements='[]'))] tasks.detect_spikes( MagicMock(), ids, target_date_str, start_date_str, account_level, run_id) expected_calls = [ call('get_spiked_tracks_{}'.format(account_level), { 'max_spikes_per_account': mock_config.MAX_SPIKES_PER_ACCOUNT, 'target_date': target_date_str, 'playlist_placement_shift_days': ( mock_config.PLAYLIST_PLACEMENTS_SHIFT_DAYS), 'max_playlists_to_show': mock_config.MAX_PLAYLISTS_TO_SHOW, 'playlist_streams_threshold': ( mock_config.PLAYLIST_STREAMS_THRESHOLD), 'run_id': run_id, 'store_ids': [1, 2, 3] }) ] mock_session_context.execute.assert_has_calls(expected_calls) assert mock_send_notifications.call_count == 1 def test_detect_spikes_no_labels_specified_default_window( monkeypatch, mock_session_context, mock_sql_loader, mock_config, mock_send_notifications, account_level): """Test detect_spikes function when no label_ids specified.""" def loader_side_effect(query_name): if query_name == ( 'create_summary_spike_detector_avg_table_{}'.format( account_level)): return query_name + '_{id_filter}' else: return query_name target_date_str = '2018-06-29' target_date = datetime.strptime(target_date_str, '%Y-%m-%d') window_days = mock_config.DEFAULT_WINDOW_DAYS dlt = timedelta(days=window_days) start_date = target_date - dlt start_date_str = start_date.strftime('%Y-%m-%d') run_id = 'x' mock_sql_loader.load_query.side_effect = loader_side_effect monkeypatch.setattr(dynamodb, 'set_status', MagicMock(return_value=None)) mock_session_context.execute.side_effect = [( MagicMock( storeid=1, artist_names='["foo"]', recent_playlist_placements='[]'), MagicMock( storeid=2, artist_names='["boo"]', recent_playlist_placements='[]'))] tasks.detect_spikes( MagicMock(), [], target_date_str, start_date_str, account_level, run_id) expected_calls = [ call('get_spiked_tracks_{}'.format(account_level), { 'max_spikes_per_account': mock_config.MAX_SPIKES_PER_ACCOUNT, 'target_date': target_date_str, 'playlist_placement_shift_days': ( mock_config.PLAYLIST_PLACEMENTS_SHIFT_DAYS), 'max_playlists_to_show': mock_config.MAX_PLAYLISTS_TO_SHOW, 'playlist_streams_threshold': ( mock_config.PLAYLIST_STREAMS_THRESHOLD), 'run_id': run_id, 'store_ids': [1, 2, 3] }) ] mock_session_context.execute.assert_has_calls(expected_calls) assert mock_send_notifications.call_count == 1 def test_detect_spikes_no_spikes( monkeypatch, mock_session_context, mock_sql_loader, mock_config, mock_send_notifications): """Test detect_spikes function when no spike detected.""" target_date_str = '2018-06-29' start_date_str = '2018-06-31' run_id = 'x' monkeypatch.setattr(dynamodb, 'set_status', MagicMock(return_value=None)) mock_session_context.execute.side_effect = [()] tasks.detect_spikes( MagicMock(), [], target_date_str, start_date_str, account_level, run_id) assert mock_send_notifications.call_count == 0 def test_send_nofitications_label(mock_spike_notification): """Test send_notifications functions (label).""" release_date = '1990-11-04' target_date = '2018-07-02' spiked_tracks = [ SpikedTrack( 1, 'ln', 'tn', 'an', 'isrc1', release_date, 100, 4, 500, 'iu', 'spotify', []), SpikedTrack( 2, 'ln', 'tn', 'an', 'isrc2', release_date, 100, 4, 500, 'iu', 'spotify', []), SpikedTrack( 2, 'ln', 'tn', 'an', 'isrc3', release_date, 100, 4, 500, 'iu', 'spotify', []), SpikedTrack( 3, 'ln', 'tn', 'an', 'isrc4', release_date, 100, 4, 500, 'iu', 'spotify', []), ] tasks.send_notifications( MagicMock(), target_date, spiked_tracks, level='label') expected_calls = [ call( target_date, 1, spiked_tracks[:1], 'ln', None, 'label'), call().send(), call( target_date, 2, spiked_tracks[1:3], 'ln', None, 'label'), call().send(), call(target_date, 3, spiked_tracks[3:], 'ln', None, 'label'), call().send() ] mock_spike_notification.assert_has_calls(expected_calls) def test_send_nofitications_subaccount(mock_spike_notification): """Test send_notifications functions (subaccount).""" release_date = '1990-11-04' target_date = '2018-07-02' spiked_tracks = [ SpikedTrack( 1, 'ln', 'tn', 'an', 'isrc1', release_date, 100, 4, 500, 'iu', 'spotify', [], subaccount_id=1, subaccount_name='sn'), SpikedTrack( 2, 'ln', 'tn', 'an', 'isrc2', release_date, 100, 4, 500, 'iu', 'spotify', [], subaccount_id=2, subaccount_name='sn'), SpikedTrack( 3, 'ln', 'tn', 'an', 'isrc3', release_date, 100, 4, 500, 'iu', 'spotify', [], subaccount_id=2, subaccount_name='sn'), SpikedTrack( 3, 'ln', 'tn', 'an', 'isrc4', release_date, 100, 4, 500, 'iu', 'spotify', [], subaccount_id=3, subaccount_name='sn'), ] tasks.send_notifications( MagicMock(), target_date, spiked_tracks, level='subaccount') expected_calls = [ call(target_date, 1, spiked_tracks[:1], 'ln', 'sn', 'subaccount'), call().send(), call(target_date, 2, spiked_tracks[1:3], 'ln', 'sn', 'subaccount'), call().send(), call(target_date, 3, spiked_tracks[3:], 'ln', 'sn', 'subaccount'), call().send() ] mock_spike_notification.assert_has_calls(expected_calls) def test_send_nofitications_not_sorted_label(mock_spike_notification): """Test send_notifications functions (label).""" release_date = '1990-11-04' target_date = '2018-07-02' spiked_tracks = [ SpikedTrack( 1, 'ln', 'tn', 'an', 'isrc1', release_date, 100, 4, 500, 'iu', 'spotify', []), SpikedTrack( 3, 'ln', 'tn', 'an', 'isrc2', release_date, 100, 4, 500, 'iu', 'spotify', []), SpikedTrack( 2, 'ln', 'tn', 'an', 'isrc3', release_date, 100, 4, 500, 'iu', 'spotify', []), SpikedTrack( 1, 'ln', 'tn', 'an', 'isrc4', release_date, 100, 4, 500, 'iu', 'spotify', []), ] with pytest.raises(Exception): tasks.send_notifications( MagicMock(), target_date, spiked_tracks, 'label') def test_send_nofitications_not_sorted_subaccount(mock_spike_notification): """Test send_notifications functions (subaccount).""" release_date = '1990-11-04' target_date = '2018-07-02' spiked_tracks = [ SpikedTrack( 1, 'ln', 'tn', 'an', 'isrc1', release_date, 100, 4, 500, 'iu', 'spotify', [], subaccount_id=3), SpikedTrack( 2, 'ln', 'tn', 'an', 'isrc2', release_date, 100, 4, 500, 'iu', 'spotify', [], subaccount_id=2), SpikedTrack( 3, 'ln', 'tn', 'an', 'isrc3', release_date, 100, 4, 500, 'iu', 'spotify', [], subaccount_id=4), SpikedTrack( 4, 'ln', 'tn', 'an', 'isrc4', release_date, 100, 4, 500, 'iu', 'spotify', [], subaccount_id=3), ] with pytest.raises(Exception): tasks.send_notifications( MagicMock(), target_date, spiked_tracks, 'subaccount') def test_send_nofitications_single_label(mock_spike_notification): """Test send_notifications functions.""" release_date = '1990-11-04' target_date = '2018-07-02' spiked_tracks = [ SpikedTrack( 1, 'ln', 'tn', 'an', 'isrc1', release_date, 200, 4, 500, 'iu', 'spotify', []), SpikedTrack( 1, 'ln', 'tn', 'an', 'isrc2', release_date, 100, 4, 500, 'iu', 'spotify', []), SpikedTrack( 1, 'ln', 'tn', 'an', 'isrc3', release_date, 400, 4, 500, 'iu', 'spotify', []), SpikedTrack( 1, 'ln', 'tn', 'an', 'isrc4', release_date, 300, 4, 500, 'iu', 'spotify', []), SpikedTrack( 1, 'ln', 'tn', 'an', 'isrc4', release_date, 600, 4, 500, 'iu', 'spotify', []), SpikedTrack( 1, 'ln', 'tn', 'an', 'isrc4', release_date, 600, 4, 500, 'iu', 'spotify', []), ] tasks.send_notifications(MagicMock(), target_date, spiked_tracks, 'label') # tracks should be sorted in desc by number of streams tracks_sorted = list(sorted( spiked_tracks, key=lambda t: t.num_streams, reverse=True)) expected_calls = [ call(target_date, 1, tracks_sorted, 'ln', None, 'label'), call().send(), ] mock_spike_notification.assert_has_calls(expected_calls) def test_send_nofitications_no_tracks(mock_spike_notification): """Test send_notifications functions.""" target_date = '2018-07-02' spiked_tracks = [] tasks.send_notifications(MagicMock(), target_date, spiked_tracks, 'label') assert mock_spike_notification.call_count == 0 def test_check_dynamo_status_should_run(monkeypatch): """Test checking the activity status in DynamoDB.""" last_processed_timestamp = datetime(2017, 12, 25, 0, 0, 0) date = '0000-00-01' monkeypatch.setattr( dynamodb, 'get_status', MagicMock(return_value={ 'status': 'PROCESSED', 'last_processed_timestamp': last_processed_timestamp})) result = tasks.check_dynamo_status(MagicMock(), date) dynamodb.get_status.assert_called_with( config.ACTIVITY_TYPE_NAME, date) assert result == {'should_run': True} def test_check_dynamo_status_none(monkeypatch): """Test checking the activity status in DynamoDB.""" date = '0000-00-01' monkeypatch.setattr( dynamodb, 'get_status', MagicMock(return_value=None)) result = tasks.check_dynamo_status(MagicMock(), date) dynamodb.get_status.assert_called_with( config.ACTIVITY_TYPE_NAME, date) assert result == {'should_run': True} def test_check_dynamo_status_already_processed(monkeypatch): """Test checking the activity status in DynamoDB.""" last_processed_timestamp = datetime(2017, 12, 25, 0, 0, 0) date = '0000-00-01' monkeypatch.setattr( dynamodb, 'get_status', MagicMock(return_value={ 'status': 'PROCESSED_NOTIF_SENT', 'last_processed_timestamp': last_processed_timestamp})) result = tasks.check_dynamo_status(MagicMock(), date) dynamodb.get_status.assert_called_with( config.ACTIVITY_TYPE_NAME, date) assert result == {'should_run': False} @pytest.mark.parametrize('test_input,expected', [ ((config.STATUS_PROCESSED_NOTIF_SENT, config.STATUS_PROCESSED_NOTIF_SENT), config.STATUS_PROCESSED_NOTIF_SENT), ((config.STATUS_PROCESSED_NOTIF_SENT, config.STATUS_PROCESSED), config.STATUS_PROCESSED_NOTIF_SENT), ((config.STATUS_PROCESSED, config.STATUS_PROCESSED_NOTIF_SENT), config.STATUS_PROCESSED_NOTIF_SENT), ((config.STATUS_PROCESSED, config.STATUS_PROCESSED), config.STATUS_PROCESSED)]) def test_set_dynamo_status(test_input, expected, monkeypatch): """Test setting the activity status in DynamoDB.""" date = '0000-00-01' monkeypatch.setattr(dynamodb, 'set_status', MagicMock(return_value=None)) status_label, status_subaccount = test_input tasks.set_dynamo_status( MagicMock(), date, status_label, status_subaccount) dynamodb.set_status.assert_called_with( config.ACTIVITY_TYPE_NAME, date, expected) def test_get_release_image_url(monkeypatch): """Test tasks.get_release_image_url function.""" monkeypatch.setattr( ows_assets, 'get_release_image_url', MagicMock(return_value='url_from_assets')) assert tasks.get_release_image_url(777, 'upc') == 'url_from_assets' def test_get_release_image_url_none(monkeypatch): """Test tasks.get_release_image_url function.""" monkeypatch.setattr( ows_assets, 'get_release_image_url', MagicMock(return_value=None)) assert tasks.get_release_image_url(777, 'upc') is None