"""Unit tests for tasks of Spotify Workflow.""" import copy from datetime import date as date_module, timedelta from itertools import product from unittest import mock from unittest.mock import ANY, call, MagicMock from unittest.mock import patch from boto3.exceptions import S3UploadFailedError import freezegun from garcon_contrib.dynamo_feed_status import garcon_feed_status import pytest from requests import ConnectionError # noqa:A004 from requests import HTTPError from requests import Response from feed_ingestion.flows.spotify import config from feed_ingestion.flows.spotify import tasks from feed_ingestion.flows.spotify.smart_downloader import DownloadTask _date = '2017-11-16' _new_date = '2021-02-10' @pytest.fixture def expected_bootstrap_response( mock_archive_paths, mock_reports_status_names, mock_temp_staging_raw_names, mock_drop_paths): """Response for bootstrap task.""" bootstrap_response = {} for licensor in config.spotify_api_licensors: bootstrap_response[licensor] = { 'date': _date, 'facts_feed_name': f'spotify_{licensor}_streams', 'aggregated_feed_name': f'spotify_{licensor}_aggregated_streams', 'date_as_in_uuid': _date, 'archive_paths': mock_archive_paths[licensor], 'dimension_tables': config.dimension_tables, 'reports_status_names': mock_reports_status_names[licensor], 'temp_staging_raw_names': mock_temp_staging_raw_names[licensor], 'use_s3': None, 'use_partitioned': 'True', 'licensor': licensor, 'drop_paths': mock_drop_paths[licensor], 'jenkins_config': config.jenkins_config, } return bootstrap_response @pytest.fixture def mock_executor_context(): """Yield executor context.""" sf_executor_class_path = 'feed_ingestion.flows.spotify.tasks.Spotify' with patch(sf_executor_class_path) as sf_executor: mock_executor_context = sf_executor.return_value.__enter__.return_value yield mock_executor_context @pytest.fixture def mock_executor_context_theorchard(): """Yield executor context.""" reg_executors_class_path = ( 'feed_ingestion.flows.spotify.tasks.registered_executors') with patch(reg_executors_class_path) as reg_executors: mock_executor = MagicMock() reg_executors.get = mock_executor yield mock_executor.return_value.return_value.__enter__.return_value @pytest.fixture def mock_task_status(): """Yield task status.""" task_status_path = 'feed_ingestion.tasks.task_status' 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.fixture def mock_spotify_task_status(): """Yield task status.""" task_status_path = 'feed_ingestion.flows.spotify.tasks.task_status' 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.fixture def mock_set_overall_status(): """Yield overall status.""" overall_status_path = ( 'feed_ingestion.flows.spotify.tasks.overall_status_tasks.' 'set_overall_status') with patch(overall_status_path) as overall_status: yield overall_status @pytest.fixture def mock_get_overall_status(): """Yield overall status.""" overall_status_path = ( 'feed_ingestion.flows.spotify.tasks.garcon_feed_status.' 'get_overall_status') with patch(overall_status_path) as overall_status: yield overall_status @pytest.fixture def mock_delete_status(): """Yield delete status.""" delete_status_path = ( 'feed_ingestion.flows.spotify.tasks.garcon_feed_status.' 'delete_status') with patch(delete_status_path) as delete_status: yield delete_status @pytest.fixture def mock_boto3(): """Mock boto3.""" boto3_path = 'feed_ingestion.flows.spotify.tasks.boto3' with patch(boto3_path) as boto3: mock_client = MagicMock() boto3.client.return_value = mock_client yield boto3 def test_check_feed_status_with_reload( mock_delete_status, mock_get_overall_status): """Test check_feed_status with reload.""" for licensor in config.spotify_api_licensors: feed_name = '_'.join([config.feed_name, licensor]) context = { 'activity': MagicMock(), 'date': _date, 'reload': 'True', 'licensor': licensor, } response = tasks.check_feed_status(**context) assert {'feed_name': feed_name} == response mock_delete_status.assert_any_call(feed_name, _date) mock_get_overall_status.assert_not_called() def test_check_feed_status_without_reload( mock_delete_status, mock_get_overall_status): """Test check_feed_status without reload.""" for licensor in config.spotify_api_licensors: feed_name = '_'.join([config.feed_name, licensor]) context = { 'activity': MagicMock(), 'date': _date, 'reload': None, 'licensor': licensor, } response = tasks.check_feed_status(**context) assert {'feed_name': feed_name} == response mock_get_overall_status.assert_any_call(feed_name, _date) mock_delete_status.assert_not_called() def test_check_feed_status_already_ingested( mock_delete_status, mock_get_overall_status): """Test check_feed_status without reload.""" mock_get_overall_status.return_value = garcon_feed_status.STATUS_INGESTED for licensor in config.spotify_api_licensors: feed_name = '_'.join([config.feed_name, licensor]) context = { 'activity': MagicMock(), 'date': _date, 'reload': None, 'licensor': licensor, } response = tasks.check_feed_status(**context) assert response == {'stop': True} mock_get_overall_status.assert_any_call(feed_name, _date) mock_delete_status.assert_not_called() @patch('feed_ingestion.flows.spotify.tasks.check_report') def test_bootstrap( 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': False, 'licensor': licensor, 'reports': None, 'use_s3': None, } response = tasks.bootstrap(**context) assert response == expected_bootstrap_response[licensor] mock_delete_status.assert_not_called() assert mock_check_report.call_count == ( len(config.reports) * len(config.spotify_api_licensors)) @patch('feed_ingestion.flows.spotify.tasks.check_report') def test_bootstrap_with_use_s3( 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': False, 'licensor': licensor, 'reports': None, 'use_s3': 'True', } response = tasks.bootstrap(**context) assert response['use_s3'] == 'True' @patch('feed_ingestion.flows.spotify.tasks.check_report') def test_bootstrap_with_reload( mock_check_report, expected_bootstrap_response, mock_delete_status, mock_reports_status_names): """Check that bootstrap returns expected results.""" for licensor in config.spotify_api_licensors: context = { 'activity': MagicMock(), 'date': _date, 'reload': 'True', 'licensor': licensor, 'reports': None, 'use_s3': None, } response = tasks.bootstrap(**context) assert response == expected_bootstrap_response[licensor] for report_name, feed_name in \ mock_reports_status_names[licensor].items(): mock_delete_status.assert_any_call( feed_name, _date) mock_check_report.assert_any_call( report_name, feed_name, _date) @patch('feed_ingestion.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 and report_name not in config.common_reports: return False return True mock_check_report.side_effect = side_effect_status for licensor in config.spotify_api_licensors: for report_name in set(config.reports): context = { 'activity': MagicMock(), 'date': _date, 'reload': False, 'licensor': licensor, 'reports': None, 'use_s3': None, } response = tasks.bootstrap(**context)['reports_status_names'] if report_name in config.common_reports: assert report_name in response else: assert report_name not in response @patch('feed_ingestion.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', 'licensor': licensor, 'reports': None, 'use_s3': None, } response = tasks.bootstrap(**context) assert response == expected_bootstrap_response[licensor] mock_delete_status.assert_has_calls( [call('_'.join([config.feed_name, licensor, report]), _date) for report in config.reports], any_order=True) @patch('feed_ingestion.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 licensor in config.spotify_api_licensors: for report in config.reports: context = { 'activity': MagicMock(), 'date': _date, 'reload': 'True', 'licensor': licensor, 'reports': report, 'use_s3': None, } response = tasks.bootstrap(**context)['reports_status_names'] reports = set(config.common_reports + [report]) assert reports == response.keys() mock_delete_status.assert_any_call( '_'.join([config.feed_name, licensor, report]), _date) @patch('feed_ingestion.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] for licensor in config.spotify_api_licensors: context = { 'activity': MagicMock(), 'date': _date, 'reload': 'True', 'licensor': licensor, 'reports': ','.join(reload_reports), 'use_s3': None, } reports = set(config.common_reports + reload_reports) response = tasks.bootstrap(**context)['reports_status_names'] assert reports == response.keys() mock_delete_status.assert_has_calls( [call('_'.join([config.feed_name, licensor, report]), _date) 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': 'v1' } } 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.fixture def mock_send_error_or_warning(self): """Mock function send_error_or_warning.""" path = 'feed_ingestion.flows.spotify.tasks.send_error_or_warning' with patch(path) as send_error_or_warning: yield send_error_or_warning @pytest.fixture def mock_spotify_api(self): """Spotify API Wrapper fixture.""" sa_path = ( 'feed_ingestion.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.fixture def mock_spotify_api_fail(self): """Spotify API Wrapper failing fixture.""" sa_path = ( 'feed_ingestion.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.fixture def mock_spotify_api_not_available(self): """Spotify API Wrapper resource not available fixture.""" sa_path = ( 'feed_ingestion.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.fixture def mock_spotify_api_connection_error(self): """Spotify API Wrapper resource connection error.""" sa_path = 'feed_ingestion.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.fixture def mock_spotify_api_streams_fail(self): """Spotify API Wrapper failing fixture.""" sa_path = ( 'feed_ingestion.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.fixture def mock_pool(self): """Mock multiprocessing pool.""" pool_path = 'feed_ingestion.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.fixture def mock_boto3_fail(self): """Mock boto3.""" boto3_path = 'feed_ingestion.flows.spotify.tasks.boto3' with patch(boto3_path) as boto3: boto3.client.side_effect = S3UploadFailedError('Failed to upload') yield boto3 @pytest.fixture def mock_remove_files_from_path(self): """Return mock remove_files_from_path.""" remove_files_path = ( 'feed_ingestion.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_send_error_or_warning, 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(mock.ANY, _date, '_'.join([config.feed_name, self.LICENSOR, report_name]), 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_send_error_or_warning, 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(mock.ANY, _date, '_'.join([config.feed_name, self.LICENSOR, report_name]), garcon_feed_status.STATUS_NOT_AVAILABLE) mock_task_status.mark_completed_task.assert_not_called() mock_send_error_or_warning.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, mock_send_error_or_warning): """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(mock.ANY, _date, '_'.join([config.feed_name, self.LICENSOR, report_name]), 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_pool, 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): 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() def test_drop_temp_table(mock_executor_context): """Test drop_temp_table task.""" sfdb_params = {'db': 'db', 'schema': 'schema'} tasks.drop_temp_table(MagicMock(), 'test_table', sfdb_params) (mock_executor_context.drop_table.assert_called_with( 'test_table')) def test_create_transitional_common_tables( mock_spotify_task_status, mock_executor_context, mock_common_tables): """Test create_transitional_common_tables.""" sfdb_params = {'db': 'db', 'schema': 'schema'} for record in mock_common_tables: tasks.create_transitional_common_tables( MagicMock(), _date, 'spotify_{}'.format(record.report_name), record.report_name, record.transitional_temp_table, sfdb_params) (mock_executor_context.create_transitional_temp_staging_raw_table. assert_called_with(record.report_name, record.transitional_temp_table)) def test_load_common_tables( mock_executor_context, mock_temp_staging_raw_names, mock_common_tables, mock_task_status): """Test load_common_tables.""" sfdb_params = {'db': 'db', 'schema': 'schema'} for licensor, record in product( config.spotify_api_licensors, mock_common_tables): tasks.load_common_tables( MagicMock(), _date, 'spotify_{}'.format(record.report_name), record.report_name, mock_temp_staging_raw_names[licensor][record.report_name], record.transitional_temp_table, record.staging_table, licensor, sfdb_params) mock_executor_context.load_transitional_common_tables. \ assert_called_with(_date, mock_temp_staging_raw_names[licensor][ record.report_name], record.transitional_temp_table, record.staging_table, record.report_name, licensor) mock_executor_context.load_common_staging_raw_table. \ assert_called_with(_date, record.transitional_temp_table, record.staging_table, record.report_name, licensor) mock_executor_context.drop_table.assert_called_with( record.transitional_temp_table) def test_check_available_reports_if_all_files_are_available( mock_spotify_task_status, mock_reports_status_names_without_download_only): """Test check_available_reports if all reports are downloaded.""" mock_spotify_task_status.is_completed_task.return_value = True for licensor in config.spotify_api_licensors: reports_status_names = mock_reports_status_names_without_download_only[ licensor] result = tasks.check_available_reports( MagicMock(), _date, reports_status_names) for report_name, feed_name in reports_status_names.items(): mock_spotify_task_status.is_completed_task.assert_any_call( feed_name, _date, 'grab_drop_files') necessary_activities = { 'load_fact_analytics': True, 'load_staging_raw': True, 'load_common_table': True } assert result['available_reports'] == reports_status_names assert result['necessary_activities'] == necessary_activities def test_check_available_reports_if_all_files_are_not_available( mock_spotify_task_status, mock_reports_status_names): """Test check_available_reports if all reports are not downloaded.""" for licensor in config.spotify_api_licensors: reports_status_names = mock_reports_status_names[licensor] result = tasks.check_available_reports( MagicMock(), _date, reports_status_names) for report_name, feed_name in reports_status_names.items(): mock_spotify_task_status.is_completed_task.assert_any_call( feed_name, _date, 'grab_drop_files') assert result == {'stop': True} def test_check_available_reports_if_common_files_are_not_available( mock_spotify_task_status, mock_reports_status_names): """Test check_available_reports if common reports are not downloaded.""" def side_effect_status(feed_name, date, task_id): """Return False for common reports otherwise True.""" report = feed_name.split('_', 2)[2] if report in config.common_reports: return False return True mock_spotify_task_status.is_completed_task.side_effect = ( side_effect_status) for licensor in config.spotify_api_licensors: reports_status_names = mock_reports_status_names[licensor] result = tasks.check_available_reports( MagicMock(), _date, reports_status_names) for report_name, feed_name in reports_status_names.items(): mock_spotify_task_status.is_completed_task.assert_any_call( feed_name, _date, 'grab_drop_files') assert result == {'stop': True} def test_check_available_reports_if_one_of_the_report_is_not_available( mock_spotify_task_status, mock_reports_status_names_without_download_only, mock_get_overall_status): """Test check_available_reports if one of reports is not downloaded. If one of the not common reports is not available, task should return dict of other available reports. """ def side_effect_status(feed_name, date, task_id): """Return False for one of not common report otherwise True.""" report = feed_name.split('_', 2)[2] if report == report_name: return False return True reports_for_next_process = ( set(config.reports) - set(config.common_reports) - set(config.download_only_reports)) for report_name in reports_for_next_process: mock_spotify_task_status.is_completed_task.side_effect = ( side_effect_status) for licensor in config.spotify_api_licensors: reports_status_names = ( mock_reports_status_names_without_download_only)[licensor] result = tasks.check_available_reports( MagicMock(), _date, reports_status_names) expected_result = copy.deepcopy(reports_status_names) expected_result.pop(report_name) assert result['available_reports'] == expected_result necessary_activities = { 'load_staging_raw': True, 'load_common_table': True} if report_name != config.fact_analytics_report: necessary_activities['load_fact_analytics'] = True assert result['necessary_activities'] == necessary_activities def test_check_available_reports_if_only_common_reports_are_available( mock_spotify_task_status, mock_reports_status_names, mock_get_overall_status): """Test check_available_reports if only common reports are available.""" def side_effect_status(feed_name, date, task_id): """Return False for one of not common report otherwise True.""" report = feed_name.split('_', 2)[2] if report not in config.common_reports: return False return True mock_spotify_task_status.is_completed_task.side_effect = ( side_effect_status) for licensor in config.spotify_api_licensors: result = tasks.check_available_reports( MagicMock(), _date, mock_reports_status_names[licensor]) assert result == {'stop': True} def test_check_available_reports_if_common_and_download_only_are_available( mock_spotify_task_status, mock_reports_status_names, mock_get_overall_status): """Test check_available_reports if only common and sub_30 are available.""" def side_effect_status(feed_name, date, task_id): """Return False for one of not common report otherwise True.""" report = feed_name.split('_', 2)[2] if report in config.common_reports + config.download_only_reports: return True return False mock_spotify_task_status.is_completed_task.side_effect = ( side_effect_status) for licensor in config.spotify_api_licensors: result = tasks.check_available_reports( MagicMock(), _date, mock_reports_status_names[licensor]) necessary_activities = { 'load_common_table': True} assert len(result['available_reports']) == len(config.common_reports) assert result['necessary_activities'] == necessary_activities def test_check_available_reports_if_fa_report_not_available( mock_spotify_task_status, mock_reports_status_names_without_download_only, mock_get_overall_status): """Test check_available_reports. Check the situation when common reports are already INGESTED, fact_analytics report is not available, but aggregated_streams is ready to ingest. """ def side_effect_status(feed_name, date, task_id): """Return False for fact analytics report.""" report = feed_name.split('_', 2)[2] if report == config.fact_analytics_report: return False return True def side_effect_overall_status(feed_name, date): """Return STATUS_INGESTED for common reports.""" report = feed_name.split('_', 2)[2] if report not in config.common_reports: return False return garcon_feed_status.STATUS_INGESTED mock_spotify_task_status.is_completed_task.side_effect = ( side_effect_status) mock_get_overall_status.side_effect = side_effect_overall_status expected_result = copy.deepcopy( mock_reports_status_names_without_download_only) for licensor in config.spotify_api_licensors: del expected_result[licensor][config.fact_analytics_report] result = tasks.check_available_reports( MagicMock(), _date, mock_reports_status_names_without_download_only[licensor]) necessary_activities = { 'load_staging_raw': True, 'load_common_table': True} assert (len(result['available_reports']) == len(config.reports) - len(config.download_only_reports) - 1) assert result['available_reports'] == expected_result[licensor] assert result['necessary_activities'] == necessary_activities def test_set_status_ingested_if_all_reports_are_ingested( mock_set_overall_status, mock_get_overall_status, mock_reports_status_names): """Test set_status_to_ingested if reports are ingested.""" def side_effect_status(feed_name, date): """Set all needed overall statuses.""" report = feed_name.split('_', 2)[2] if report in config.download_only_reports: return garcon_feed_status.STATUS_DOWNLOADED return garcon_feed_status.STATUS_POPULATED_RAW_TABLE mock_get_overall_status.side_effect = side_effect_status for licensor in config.spotify_api_licensors: tasks.set_status_ingested( MagicMock(), _date, mock_reports_status_names[licensor]) for report_name in config.reports: mock_get_overall_status.assert_any_call( 'spotify_{}_{}'.format(licensor, report_name), _date) if report_name != config.fact_analytics_report: mock_set_overall_status.assert_any_call(mock.ANY, _date, 'spotify_{}_{}'.format( licensor, report_name), 'INGESTED') def test_set_status_ingested_if_not_all_reports_are_ingested( mock_set_overall_status, mock_get_overall_status, mock_reports_status_names): """Test set_status_to_ingested if not all reports are done.""" def side_effect_status(feed_name, date): """Set overall statuses not avalible.""" report = feed_name.split('_', 2)[2] if report != config.fact_analytics_report: return garcon_feed_status.STATUS_NOT_AVAILABLE else: return garcon_feed_status.STATUS_INGESTED mock_get_overall_status.side_effect = side_effect_status for licensor in config.spotify_api_licensors: tasks.set_status_ingested( MagicMock(), _date, mock_reports_status_names[licensor]) for report_name in config.reports: mock_get_overall_status.assert_any_call( 'spotify_{}_{}'.format(licensor, report_name), _date) mock_set_overall_status.assert_not_called() @freezegun.freeze_time('2025-10-18') def test_set_overall_status_ingested_if_all_reports_are_ingested( mock_set_overall_status, mock_get_overall_status, mock_spotify_task_status): """Test set_overall_status_to_ingested.""" mock_get_overall_status.return_value = garcon_feed_status.STATUS_INGESTED for licensor in config.spotify_api_licensors: tasks.set_overall_status_ingested(MagicMock(), _date, licensor) for report_name in config.reports: if report_name != 'users': mock_get_overall_status.assert_any_call( 'spotify_{}_{}'.format(licensor, report_name), _date) mock_set_overall_status.assert_any_call( activity=mock.ANY, date=_date, feed_name='spotify_{}'.format(licensor), status=garcon_feed_status.STATUS_INGESTED) for report in config.common_reports: mock_spotify_task_status.mark_completed_task.assert_any_call( f'spotify_{licensor}_{report}', _date, 'staging_raw_table_tasks') def test_set_overall_status_ingested_if_not_all_reports_are_ingested( mock_set_overall_status, mock_get_overall_status, mock_spotify_task_status): """Test set_overall_status_to_ingested.""" mock_get_overall_status.return_value = \ garcon_feed_status.STATUS_NOT_AVAILABLE for licensor in config.spotify_api_licensors: response = tasks.set_overall_status_ingested( MagicMock(), _date, licensor) assert response == {'stop': True} mock_set_overall_status.assert_not_called() mock_spotify_task_status.mark_completed_task.assert_not_called() def test_check_report_if_report_is_common(mock_get_overall_status): """Test function _check_report return True if report is common.""" response = tasks.check_report( report_name=(set(config.reports) - set(config.common_reports)).pop(), report_feed_name='test_status', date=_date) assert response is True def test_check_report_if_report_is_ingested(mock_get_overall_status): """Test function _check_report return False if report is ingested.""" mock_get_overall_status.return_value = garcon_feed_status.STATUS_INGESTED response = tasks.check_report( report_name=(set(config.reports) - set(config.common_reports)).pop(), report_feed_name='test_status', date=_date) assert response is False def test_check_date_in_retention_period(): """Test check_date activity.""" # check before date context_date = str(date_module.today()) assert tasks.check_date( MagicMock(), context_date, 'theorchard', False ) == {} assert tasks.check_date( MagicMock(), context_date, 'sme', False ) == {} def test_check_date_over_retention_period(): """Test check_date activity.""" # check before date date_over_retention = str(date_module.today() - timedelta(days=31)) with pytest.raises(ValueError): tasks.check_date( MagicMock(), date_over_retention, 'theorchard', False ) with pytest.raises(ValueError): tasks.check_date( MagicMock(), date_over_retention, 'sme', False ) assert tasks.check_date( MagicMock(), date_over_retention, 'sme', True ) == {} @patch('feed_ingestion.flows.spotify.tasks.remove_files_from_path') @patch('feed_ingestion.flows.spotify.tasks.copy_s3_key') def test_grab_drop_files_from_s3_non_streaming( mock_copy_s3_key, mock_remove_files_from_path, mock_task_status, mock_set_overall_status): """Test grab_drop_files_from_s3 with report with a single file.""" tasks.grab_drop_files_from_s3( MagicMock(), 'spotify_sme_users', _date, 'users', 'archive_path/', 'drop_path/') mock_copy_s3_key.assert_called_with( 's3://dev-feed-drop/drop_path/users_20171116.gz', 's3://dev-cucumbers/archive_path/users_20171116.gz') assert mock_remove_files_from_path.called mock_set_overall_status.assert_called_with( mock.ANY, _date, 'spotify_sme_users', garcon_feed_status.STATUS_DOWNLOADED) def test_prepare_urls_to_s3(): spotify_api = MagicMock() spotify_api._resource_url = 'https://blah-blah' result = tasks._prepare_urls_to_s3( date='2023-11-01', report_name='users', s3_path='s3://bucket/key/', spotify_api=spotify_api ) assert result == { 'https://blah-blah/partitions': 's3://bucket/key/' } def test_prepare_urls_to_s3_use_countries(): spotify_api = MagicMock() spotify_api.get_available_countries_for_url.return_value = ['US', 'UK'] spotify_api._resource_url = 'https://blah-blah' result = tasks._prepare_urls_to_s3( date='2023-11-01', report_name='streams', s3_path='s3://bucket/key/', spotify_api=spotify_api ) assert result == { 'https://blah-blah/UK/partitions': 's3://bucket/key/UK/', 'https://blah-blah/US/partitions': 's3://bucket/key/US/' } def test_prepare_urls_to_s3_use_countries_empty_response_should_raise(): spotify_api = MagicMock() spotify_api.get_available_countries_for_url.return_value = [] spotify_api._resource_url = 'https://blah-blah' with pytest.raises(LookupError): tasks._prepare_urls_to_s3( date='2023-11-01', report_name='streams', s3_path='s3://bucket/key/', spotify_api=spotify_api ) @freezegun.freeze_time('2025-10-18') @patch.object(tasks, '_prepare_urls_to_s3') @patch.object(tasks, '_download_urls_from_api_to_s3') @patch.object(tasks, 'remove_files_from_path') @patch.object(tasks, 'SpotifyAPI') def test_grab_drop_files_partitioned( mock_SpotifyAPI, mock_remove_files_from_path, mock_download_urls_from_api_to_s3, mock_prepare_urls_to_s3, mock_task_status, mock_set_overall_status, ): """Test grab_drop_files_partitioned.""" activity_mock = MagicMock() result = tasks.grab_drop_files_partitioned( activity=activity_mock, feed_name='spotify_sme_tracks', date='2023-11-01', report_name='aggregated_streams', archive_path='spotify/tracks/', licensor='sme' ) assert result == mock_download_urls_from_api_to_s3.return_value assert mock_prepare_urls_to_s3.call_args_list == [ call('2023-11-01', 'aggregated_streams', 's3://dev-cucumbers/spotify/tracks/', mock_SpotifyAPI.return_value) ] assert mock_download_urls_from_api_to_s3.call_args_list == [ call(mock_SpotifyAPI.return_value, mock_prepare_urls_to_s3.return_value) ] assert mock_set_overall_status.call_args_list == [ call(activity=mock.ANY, date='2023-11-01', feed_name='spotify_sme_tracks', status='DOWNLOADED') ] assert mock_remove_files_from_path.call_args_list == [ call(activity_mock, 's3://dev-cucumbers/spotify/tracks/', False) ] @freezegun.freeze_time('2025-10-18') @patch.object(tasks, '_prepare_urls_to_s3') @patch.object(tasks, '_download_urls_from_api_to_s3') @patch.object(tasks, 'remove_files_from_path') @patch.object(tasks, 'SpotifyAPI') def test_grab_drop_files_partitioned_stop_on_supported_exception( mock_SpotifyAPI, mock_remove_files_from_path, mock_download_urls_from_api_to_s3, mock_prepare_urls_to_s3, mock_task_status, mock_set_overall_status ): """Test grab_drop_files_partitioned.""" activity_mock = MagicMock() mock_download_urls_from_api_to_s3.side_effect = ( LookupError('sample exception')) result = tasks.grab_drop_files_partitioned( activity=activity_mock, feed_name='spotify_sme_tracks', date='2023-11-01', report_name='aggregated_streams', archive_path='spotify/tracks/', licensor='sme' ) assert result == {'message': 'sample exception', 'stop': True} assert mock_prepare_urls_to_s3.call_args_list == [ call('2023-11-01', 'aggregated_streams', 's3://dev-cucumbers/spotify/tracks/', mock_SpotifyAPI.return_value) ] assert mock_download_urls_from_api_to_s3.call_args_list == [ call(mock_SpotifyAPI.return_value, mock_prepare_urls_to_s3.return_value) ] assert mock_set_overall_status.call_args_list == [ call(activity=mock.ANY, date='2023-11-01', feed_name='spotify_sme_tracks', status='NOT_INGESTED') ] assert mock_remove_files_from_path.call_args_list == [ call(activity_mock, 's3://dev-cucumbers/spotify/tracks/', False) ] @patch.object(tasks, '_prepare_urls_to_s3') @patch.object(tasks, '_download_urls_from_api_to_s3') @patch.object(tasks, 'remove_files_from_path') @patch.object(tasks, 'SpotifyAPI') @pytest.mark.parametrize('exception_class', [ValueError, ZeroDivisionError]) def test_grab_drop_files_partitioned_fail_on_not_supported_exception( mock_SpotifyAPI, mock_remove_files_from_path, mock_download_urls_from_api_to_s3, mock_prepare_urls_to_s3, mock_task_status, mock_set_overall_status, exception_class ): """Test grab_drop_files_partitioned.""" activity_mock = MagicMock() mock_download_urls_from_api_to_s3.side_effect = ( exception_class('sample exception')) with pytest.raises(exception_class): tasks.grab_drop_files_partitioned( activity=activity_mock, feed_name='spotify_sme_tracks', date='2023-11-01', report_name='aggregated_streams', archive_path='spotify/tracks/', licensor='sme' ) def test_create_download_tasks_for_partitions(): """Test create_download_tasks_for_partitions.""" partitions = [ { 'uri': 'https://provider-api.spotify.com/.../AB', 'description': 'AB', }, { 'uri': 'https://provider-api.spotify.com/.../AC', 'description': 'AC', }, ] s3_location = 's3://bucket/key/' result = tasks._create_download_tasks_for_partitions( partitions=partitions, s3_location=s3_location, ) assert result == [ DownloadTask( source_url='https://provider-api.spotify.com/.../AB', destination_url='s3://bucket/key/AB'), DownloadTask( source_url='https://provider-api.spotify.com/.../AC', destination_url='s3://bucket/key/AC') ] @patch.object(tasks, 'smart_downloader') @patch.object(tasks, 'uuid') def test_download_urls_from_api_to_s3( uuid_mock, downloader_mock, ): """Test download_urls_from_api_to_s3.""" downloader_mock.await_downloads_completion.return_value = { 'DONE': 45 } url_to_s3 = { 'https://api1': 's3://bucket/AA', 'https://api2': 's3://bucket/AB', } uuid_mock.uuid4.return_value = 'uuid-value' spotify_api_mock = MagicMock() result = tasks._download_urls_from_api_to_s3( spotify_api=spotify_api_mock, url_to_s3=url_to_s3 ) assert result == { 'jobId': 'swf-spotify-uuid-value', 'n_tasks': downloader_mock.send_download_requests.return_value } assert (downloader_mock.download_request_batched_generator .call_args_list) == [ call(tasks=ANY, batch_size=20), ] assert downloader_mock.send_download_requests.call_args_list == [ call( download_requests=( downloader_mock.download_request_batched_generator .return_value), job_id='swf-spotify-uuid-value', table_name='dev_smart_downloader_tasks', ttl_timeout_seconds=604800 ) ] assert downloader_mock.await_downloads_completion.call_args_list == [ call( job_id='swf-spotify-uuid-value', expected_number_of_items=(downloader_mock. send_download_requests.return_value), table_name='dev_smart_downloader_tasks', timeout_seconds=900, ) ] @patch.object(tasks, 'smart_downloader') @patch.object(tasks, 'uuid') def test_download_urls_from_api_to_s3_await_timed_out( uuid_mock, downloader_mock, ): """Test download_urls_from_api_to_s3.""" downloader_mock.await_downloads_completion.side_effect = ( TimeoutError('timed out message')) url_to_s3 = { 'https://api1': 's3://bucket/AA', 'https://api2': 's3://bucket/AB', } uuid_mock.uuid4.return_value = 'uuid-value' spotify_api_mock = MagicMock() with pytest.raises(RuntimeError) as exception_info: tasks._download_urls_from_api_to_s3( spotify_api=spotify_api_mock, url_to_s3=url_to_s3 ) assert str(exception_info.value) == ( 'Download failed: timed out message. ' 'Lookup DynamoDB "dev_smart_downloader_tasks" ' 'for jobId="swf-spotify-uuid-value"') @patch.object(tasks, 'uuid') @patch.object(tasks.smart_downloader, 'boto3') def test_download_urls_from_api_to_s3_404_not_found( boto3_mock, uuid_mock, ): """Test download_urls_from_api_to_s3. This case when spotify_api return empty content for countries. It means no data available """ url_to_s3 = { 'https://api1': 's3://bucket/AA', 'https://api2': 's3://bucket/AB', } uuid_mock.uuid4.return_value = 'uuid-value' spotify_api_mock = MagicMock() spotify_api_mock.get_partitions_for_url.side_effect = ( HTTPError('404 Not Found')) with pytest.raises(LookupError) as exception_info: tasks._download_urls_from_api_to_s3( spotify_api=spotify_api_mock, url_to_s3=url_to_s3 ) assert str(exception_info.value) == ( 'Data not available for https://api1: 404 Not Found') @patch.object(tasks, 'smart_downloader') @patch.object(tasks, 'uuid') def test_download_urls_from_api_to_s3_task_errors( uuid_mock, downloader_mock, ): """Test download_urls_from_api_to_s3.""" downloader_mock.await_downloads_completion.return_value = { 'DONE': 42, 'ERROR': 3, } url_to_s3 = { 'https://api1': 's3://bucket/AA', 'https://api2': 's3://bucket/AB', } uuid_mock.uuid4.return_value = 'uuid-value' spotify_api_mock = MagicMock() with pytest.raises(RuntimeError) as exception_info: tasks._download_urls_from_api_to_s3( spotify_api=spotify_api_mock, url_to_s3=url_to_s3 ) assert str(exception_info.value) == ( "Download failed: Final states: {'DONE': 42, 'ERROR': 3}. " 'Lookup DynamoDB "dev_smart_downloader_tasks" ' 'for jobId="swf-spotify-uuid-value"') @patch('feed_ingestion.flows.spotify.tasks.get_list_of_files_and_directories') @patch('feed_ingestion.flows.spotify.tasks.remove_files_from_path') @patch('feed_ingestion.flows.spotify.tasks.copy_s3_key') def test_grab_drop_files_from_s3_streaming( mock_copy_s3_key, mock_remove_files_from_path, mock_get_list_of_files_and_directories, mock_task_status, mock_set_overall_status): """Test grab_drop_files_from_s3 with report with several files.""" mock_get_list_of_files_and_directories.return_value = [ 'drop_path/streams_20171116_AT.gz', 'drop_path/streams_20171116_US.gz', ] tasks.grab_drop_files_from_s3( MagicMock(), 'spotify_sme_streams', _date, 'streams', 'archive_path/', 'drop_path/') mock_copy_s3_key.assert_has_calls([ call( 's3://dev-feed-drop/drop_path/streams_20171116_AT.gz', 's3://dev-cucumbers/archive_path/streams_20171116_AT.gz'), call( 's3://dev-feed-drop/drop_path/streams_20171116_US.gz', 's3://dev-cucumbers/archive_path/streams_20171116_US.gz')]) assert mock_remove_files_from_path.called mock_set_overall_status.assert_called_with( mock.ANY, _date, 'spotify_sme_streams', garcon_feed_status.STATUS_DOWNLOADED) @patch('feed_ingestion.flows.spotify.tasks.get_list_of_files_and_directories') @patch('feed_ingestion.flows.spotify.tasks.remove_files_from_path') @patch('feed_ingestion.flows.spotify.tasks.copy_s3_key') def test_grab_drop_files_from_s3_streaming_without_files( mock_copy_s3_key, mock_remove_files_from_path, mock_get_list_of_files_and_directories, mock_task_status, mock_set_overall_status): """Test grab_drop_files_from_s3 with report with single file.""" mock_get_list_of_files_and_directories.return_value = [] response = tasks.grab_drop_files_from_s3( MagicMock(), 'spotify_sme_streams', _date, 'streams', 'archive_path/', 'drop_path/') mock_copy_s3_key.assert_not_called() mock_set_overall_status.assert_not_called() assert response == {'stop': True} def test_jenkins_config(): assert config.jenkins_config['feeds_required_for_jenkins_build'] == [ 'spotify_theorchard_streams', 'spotify_theorchard_aggregated_streams', 'spotify_sme_streams', 'spotify_sme_aggregated_streams']