"""Unit tests for tasks of Spotify Workflow.""" from collections import defaultdict, namedtuple from itertools import product from unittest.mock import call, MagicMock from unittest.mock import patch from boto3.exceptions import S3UploadFailedError from garcon_contrib.dynamo_feed_status import garcon_feed_status import pytest from requests import ConnectionError from requests import HTTPError from requests import Response from data_landing_zone.flows.spotify import config from data_landing_zone.flows.spotify import tasks _date = '2017-11-16' @pytest.fixture def mock_common_tables(): """Return common table names.""" CommonTables = namedtuple( 'CommonTables', 'report_name, transitional_temp_table, staging_table') common_tables = [ CommonTables( 'users', 'transitional_temp_table_users_20171116', 'spotify_users'), CommonTables( 'tracks', 'transitional_temp_table_tracks_20171116', 'spotify_tracks')] return common_tables @pytest.fixture def mock_reports_status_names(): """Return reports and feed statuses for them.""" reports_status_names = defaultdict(dict) for licensor, report_name in product( config.spotify_api_licensors, config.reports): report_feed_name = '_'.join([config.feed_name, licensor, report_name]) reports_status_names[licensor][report_name] = report_feed_name return reports_status_names @pytest.fixture def mock_archive_paths(): """Return common table names.""" archive_paths = defaultdict(dict) for licensor, report_name in product( config.spotify_api_licensors, config.reports): archive_paths[licensor][report_name] = ( 'landing_zone/archives/2017-11-16/' '{report_name}/{licensor}/'.format( report_name=report_name, licensor=licensor)) return archive_paths @pytest.fixture def expected_bootstrap_response( mock_archive_paths, mock_reports_status_names): """Return expected response for bootstrap task.""" return { 'date': _date, 'date_as_in_uuid': _date, 'feed_name': config.feed_name, 'licensors_list': config.spotify_api_licensors, 'archive_paths': mock_archive_paths, 'reports_status_names': mock_reports_status_names} @pytest.yield_fixture def mock_task_status(): """Yield task status.""" task_status_path = 'data_landing_zone.flows.spotify.tasks.helpers' with patch(task_status_path) as task_status: task_status.is_completed_task.return_value = False task_status.mark_completed_task = MagicMock() yield task_status @pytest.yield_fixture def mock_set_overall_status(): """Yield overall status.""" overall_status_path = ( 'data_landing_zone.flows.spotify.tasks.garcon_feed_status.' 'set_overall_status') with patch(overall_status_path) as overall_status: yield overall_status @pytest.yield_fixture def mock_get_overall_status(): """Yield overall status.""" overall_status_path = ( 'data_landing_zone.flows.spotify.tasks.garcon_feed_status.' 'get_overall_status') with patch(overall_status_path) as overall_status: yield overall_status @pytest.yield_fixture def mock_delete_status(): """Yield delete status.""" delete_status_path = ( 'data_landing_zone.flows.spotify.tasks.garcon_feed_status.' 'delete_status') with patch(delete_status_path) as delete_status: yield delete_status @pytest.yield_fixture def mock_boto3(): """Mock boto3.""" boto3_path = 'data_landing_zone.flows.spotify.tasks.boto3' with patch(boto3_path) as boto3: mock_client = MagicMock() boto3.client.return_value = mock_client yield boto3 @patch('data_landing_zone.flows.spotify.tasks.check_report') def test_bootstrap( mock_check_report, expected_bootstrap_response, mock_delete_status): """Check that bootstrap returns expected results.""" context = { 'activity': MagicMock(), 'date': _date, 'reload': False, 'licensors': None, 'reports': None } response = tasks.bootstrap(**context) assert response == expected_bootstrap_response mock_delete_status.assert_not_called() assert mock_check_report.call_count == ( len(config.reports) * len(config.spotify_api_licensors)) @patch('data_landing_zone.flows.spotify.tasks.check_report') def test_bootstrap_if_one_of_the_reports_are_already_ingested( mock_check_report, expected_bootstrap_response): """Check that bootstrap returns expected results.""" def side_effect_status(report, report_feed_name, date): """Return status False for one of the reports.""" if report == report_name: return False return True context = { 'activity': MagicMock(), 'date': _date, 'reload': False, 'licensors': None, 'reports': None } mock_check_report.side_effect = side_effect_status for licensor in config.spotify_api_licensors: for report_name in set(config.reports): response = tasks.bootstrap(**context)['reports_status_names'] if report_name in config.common_reports: report_name in response[licensor] else: assert report_name not in response[licensor] @patch('data_landing_zone.flows.spotify.tasks.check_report') def test_bootstrap_if_one_of_the_licensor_is_already_ingested( mock_check_report, expected_bootstrap_response): """Check that bootstrap returns expected results.""" def side_effect_status(report, report_feed_name, date): """Return status False for one of the reports.""" return not (licensor == report_feed_name.split('_', 2)[1]) context = { 'activity': MagicMock(), 'date': _date, 'reload': False, 'licensors': None, 'reports': None } mock_check_report.side_effect = side_effect_status for licensor in config.spotify_api_licensors: response = tasks.bootstrap(**context) assert licensor not in response['reports_status_names'] @patch('data_landing_zone.flows.spotify.tasks.check_report') def test_bootstrap_reload_licensor( mock_check_report, expected_bootstrap_response, mock_delete_status): """Check that bootstrap returns expected results.""" for licensor in config.spotify_api_licensors: context = { 'activity': MagicMock(), 'date': _date, 'reload': 'True', 'licensors': licensor, 'reports': None } response = tasks.bootstrap(**context)['reports_status_names'] assert [licensor] == list(response.keys()) mock_delete_status.assert_has_calls( [call('_'.join([config.feed_name, licensor, report]), _date) for report in config.reports], any_order=True) @patch('data_landing_zone.flows.spotify.tasks.check_report') def test_bootstrap_reload_report( mock_check_report, expected_bootstrap_response, mock_delete_status): """Check that bootstrap returns expected results.""" for report in config.reports: context = { 'activity': MagicMock(), 'date': _date, 'reload': 'True', 'licensors': None, 'reports': report } response = tasks.bootstrap(**context)['reports_status_names'] for licensor in config.spotify_api_licensors: reports = set(config.common_reports + [report]) assert reports == set(response[licensor].keys()) mock_delete_status.assert_has_calls( [call('_'.join([config.feed_name, licensor, report]), _date) for licensor in config.spotify_api_licensors], any_order=True) @patch('data_landing_zone.flows.spotify.tasks.check_report') def test_bootstrap_reload_report_for_licensor( mock_check_report, expected_bootstrap_response, mock_delete_status): """Check that bootstrap returns expected results.""" for report, licensor in product( config.reports, config.spotify_api_licensors): context = { 'activity': MagicMock(), 'date': _date, 'reload': 'True', 'licensors': licensor, 'reports': report } response = tasks.bootstrap(**context)['reports_status_names'] reports = set(config.common_reports + [report]) assert [licensor] == list(response.keys()) assert reports == set(response[licensor].keys()) mock_delete_status.assert_called_with( '_'.join([config.feed_name, licensor, report]), _date) @patch('data_landing_zone.flows.spotify.tasks.check_report') def test_bootstrap_reload_several_reports( mock_check_report, expected_bootstrap_response, mock_delete_status): """Check that bootstrap returns expected results.""" reload_reports = list(config.reports.keys())[0:3] context = { 'activity': MagicMock(), 'date': _date, 'reload': 'True', 'licensors': None, 'reports': ','.join(reload_reports) } reports = set(config.common_reports + reload_reports) response = tasks.bootstrap(**context)['reports_status_names'] for licensor in config.spotify_api_licensors: assert reports == set(response[licensor].keys()) mock_delete_status.assert_has_calls( [call('_'.join([config.feed_name, licensor, report]), _date) for licensor in config.spotify_api_licensors for report in reload_reports], any_order=True) class TestGrabDropFiles(object): """Test grab_drop_files.""" NUMBER_OF_LICENSORS = len(config.spotify_api_licensors) AGGREGATED_FILE = 1 USERS_FILE = 1 TRACKS_FILE = 1 STREAMS_FILE_PER_COUNTRY = len(config.countries) SUB_30_SEC_STREAMS_FILE_PER_COUNTRY = len(config.countries) LICENSOR = 'test_licensor' @pytest.fixture def context_grab_drop_files(self, mock_archive_paths, monkeypatch): """Return context for grab_drop_files.""" spotify_api_credentials = { self.LICENSOR: { 'client_id': 'client_id', 'client_secret': 'client_secret', 'licensor': 'client_licensor', 'version': 'v2' } } monkeypatch.setattr( config, 'spotify_api_credentials', spotify_api_credentials) context = {} for report_name in config.reports: context[report_name] = dict( activity=MagicMock(), feed_name='_'.join( [config.feed_name, self.LICENSOR, report_name]), report_name=report_name, date=_date, archive_path='archive_path', licensor=self.LICENSOR) return context @pytest.yield_fixture def mock_spotify_api(self): """Spotify API Wrapper fixture.""" sa_path = ( 'data_landing_zone.flows.spotify.' 'tasks.SpotifyAPI') with patch(sa_path) as spotify_api: api_instance = spotify_api.return_value api_instance.get_aggregated_streams_to_file = MagicMock() yield api_instance @pytest.yield_fixture def mock_spotify_api_fail(self): """Spotify API Wrapper failing fixture.""" sa_path = ( 'data_landing_zone.flows.spotify.' 'tasks.SpotifyAPI') mock_response = MagicMock() mock_response.status_code = 500 err = HTTPError(response=mock_response) with patch(sa_path) as spotify_api: api_instance = spotify_api.return_value api_instance.get_aggregated_streams_to_file = ( MagicMock(side_effect=err)) api_instance.get_tracks_to_file = ( MagicMock(side_effect=err)) api_instance.get_users_to_file = ( MagicMock(side_effect=err)) api_instance.get_streams_to_file = ( MagicMock(side_effect=err)) api_instance.get_sub_30_sec_streams_to_file = ( MagicMock(side_effect=err)) yield api_instance @pytest.yield_fixture def mock_spotify_api_not_available(self): """Spotify API Wrapper resource not available fixture.""" sa_path = ( 'data_landing_zone.flows.spotify.' 'tasks.SpotifyAPI') mock_response = MagicMock() mock_response.status_code = 404 err = HTTPError(response=mock_response) with patch(sa_path) as spotify_api: api_instance = spotify_api.return_value api_instance.get_aggregated_streams_to_file = ( MagicMock(side_effect=err)) api_instance.get_tracks_to_file = ( MagicMock(side_effect=err)) api_instance.get_users_to_file = ( MagicMock(side_effect=err)) api_instance.get_streams_to_file = ( MagicMock(side_effect=err)) api_instance.get_sub_30_sec_streams_to_file = ( MagicMock(side_effect=err)) yield api_instance @pytest.yield_fixture def mock_spotify_api_connection_error(self): """Spotify API Wrapper resource connection error.""" sa_path = 'data_landing_zone.flows.spotify.tasks.SpotifyAPI' mock_response = MagicMock() mock_response.status_code = 404 err = ConnectionError(response=mock_response) with patch(sa_path) as spotify_api: api_instance = spotify_api.return_value api_instance.get_aggregated_streams_to_file = ( MagicMock(side_effect=err)) api_instance.get_tracks_to_file = ( MagicMock(side_effect=err)) api_instance.get_users_to_file = ( MagicMock(side_effect=err)) api_instance.get_streams_to_file = ( MagicMock(side_effect=err)) api_instance.get_sub_30_sec_streams_to_file = ( MagicMock(side_effect=err)) yield api_instance @pytest.yield_fixture def mock_spotify_api_streams_fail(self): """Spotify API Wrapper failing fixture.""" sa_path = ( 'data_landing_zone.flows.spotify.' 'tasks.SpotifyAPI') expected_countries = ['one', 'another'] def get_streams_to_file_mock(fd, date, country): if country in expected_countries: res = Response() res.status_code = 404 exc = HTTPError() exc.response = res raise res with patch(sa_path) as spotify_api: api_instance = spotify_api.return_value api_instance.get_streams_to_file = get_streams_to_file_mock yield api_instance @pytest.yield_fixture def mock_pool(self): """Mock multiprocessing pool.""" pool_path = 'data_landing_zone.flows.spotify.tasks.Pool' def starmap_mock(f, args_list): class StarMapResult(object): res = [] def __init__(self, f, args_list): for args in args_list: self.res.append(f(*args)) def get(self): return self.res return StarMapResult(f, args_list) with patch(pool_path) as pool: mock_pool = pool.return_value.__enter__.return_value mock_pool.starmap_async = starmap_mock yield mock_pool @pytest.yield_fixture def mock_boto3_fail(self): """Mock boto3.""" boto3_path = 'data_landing_zone.flows.spotify.tasks.boto3' with patch(boto3_path) as boto3: boto3.client.side_effect = S3UploadFailedError('Failed to upload') yield boto3 @pytest.yield_fixture def mock_remove_files_from_path(self): """Return mock remove_files_from_path.""" remove_files_path = ( 'data_landing_zone.flows.spotify.tasks.remove_files_from_path') with patch(remove_files_path) as remove_files: yield remove_files @pytest.fixture def grab_drop_files( self, context_grab_drop_files, mock_pool, mock_task_status, mock_set_overall_status): """Run grab_drop_files.""" for report_name, context in context_grab_drop_files.items(): tasks.grab_drop_files(**context) def test_call_tracks_api( self, mock_spotify_api, mock_boto3, mock_remove_files_from_path, grab_drop_files): """Should request tracks file from SpotifyAPI.""" assert mock_spotify_api.get_tracks_to_file.called def test_users_tracks_api( self, mock_spotify_api, mock_boto3, mock_remove_files_from_path, grab_drop_files): """Should request tracks file from SpotifyAPI.""" assert mock_spotify_api.get_users_to_file.called def test_streams_tracks_api( self, mock_spotify_api, mock_remove_files_from_path, mock_boto3, grab_drop_files): """Should request tracks file from SpotifyAPI.""" streams_call_count = mock_spotify_api.get_streams_to_file.call_count expected = self.STREAMS_FILE_PER_COUNTRY assert streams_call_count == expected def test_clear_s3( self, mock_spotify_api, mock_boto3, context_grab_drop_files, mock_remove_files_from_path, grab_drop_files): """Should clear archive destination folder before uploading to S3.""" assert mock_remove_files_from_path.called def test_upload_on_s3( self, mock_spotify_api, mock_boto3, mock_remove_files_from_path, grab_drop_files): """Should upload file to S3.""" expected_calls = ( self.STREAMS_FILE_PER_COUNTRY + self.USERS_FILE + self.SUB_30_SEC_STREAMS_FILE_PER_COUNTRY + self.TRACKS_FILE + self.AGGREGATED_FILE) assert ( mock_boto3.client.return_value.upload_file.call_count == expected_calls) def test_set_overall_status( self, mock_spotify_api, mock_task_status, mock_boto3, mock_remove_files_from_path, grab_drop_files): """Should set status after successful upload.""" assert mock_task_status.mark_completed_task.called def test_api_failure( self, context_grab_drop_files, mock_spotify_api_fail, mock_task_status, mock_set_overall_status, mock_boto3, mock_remove_files_from_path, mock_pool): """Test SpotifyAPI failure.""" for report_name, context in context_grab_drop_files.items(): tasks.grab_drop_files(**context) mock_set_overall_status.assert_called_with( '_'.join([config.feed_name, self.LICENSOR, report_name]), _date, garcon_feed_status.STATUS_NOT_AVAILABLE) mock_task_status.mark_completed_task.assert_not_called() def test_api_resource_not_avaiable( self, context_grab_drop_files, mock_spotify_api_not_available, mock_task_status, mock_set_overall_status, mock_boto3, mock_remove_files_from_path, mock_pool): """Test SpotifyAPI resource is not available.""" for report_name, context in context_grab_drop_files.items(): res = tasks.grab_drop_files(**context) assert res == {'stop': True} mock_set_overall_status.assert_called_with( '_'.join([config.feed_name, self.LICENSOR, report_name]), _date, garcon_feed_status.STATUS_NOT_AVAILABLE) mock_task_status.mark_completed_task.assert_not_called() def test_api_resource_connection_error( self, context_grab_drop_files, mock_spotify_api_connection_error, mock_task_status, mock_set_overall_status, mock_boto3, mock_pool, mock_remove_files_from_path): """Test SpotifyAPI resource returns connection error.""" for report_name, context in context_grab_drop_files.items(): res = tasks.grab_drop_files(**context) assert res == {'stop': True} mock_set_overall_status.assert_called_with( '_'.join([config.feed_name, self.LICENSOR, report_name]), _date, garcon_feed_status.STATUS_NOT_AVAILABLE) mock_task_status.mark_completed_task.assert_not_called() def test_expected_countries_api_failure( self, mock_spotify_api_streams_fail, mock_remove_files_from_path, mock_task_status, mock_boto3, grab_drop_files): """Should not fail if countriy is not in expected countries list.""" assert mock_task_status.mark_completed_task.called def test_s3_failure( self, context_grab_drop_files, mock_spotify_api, mock_task_status, mock_set_overall_status, mock_boto3_fail, mock_remove_files_from_path): """Test S3 failure.""" with pytest.raises(S3UploadFailedError): with pytest.raises(HTTPError): for report_name, context in context_grab_drop_files.items(): tasks.grab_drop_files(**context) mock_set_overall_status.assert_called_with( '_'.join( [config.feed_name, report_name]), _date, garcon_feed_status.STATUS_NOT_AVAILABLE) mock_task_status.mark_completed_task.assert_not_called()