"""Tests for asset ingestor main script.""" # import tempfile # from types import SimpleNamespace # from ripper_feeder import get_audio_assets, get_cover_assets, \ # update_log, create_target_paths, transfer_audio, transfer_covers, \ # preprocess_track_list import os import tempfile from types import SimpleNamespace from unittest import TestCase from unittest.mock import patch from kgb import SpyAgency from constants import paths as asset_paths import pytest from logic import ripper_feeder_logic as logic from util import file_utils from integration_scripts import file_utils as orch_file_utils # actions = dict() # release_count = 1 # results = list() # s3_phono_relid = '12345' # table_name = 'table_name' # s3_resource = 's3_resource' # bucket_name = 'bucket' # key = 'key' # table_name = 'table_name' # backup_key = 'backup_key' # source_key = 'source_key' # target_key = 'target_key' # deletion_key = 'deletion_key' # Local fixture @pytest.fixture def test_print_log_info(mocker): """Create fixture for logger.""" return mocker.patch('integration_scripts.logger.info') @pytest.fixture def test_print_log_warning(mocker): """Create fixture for logger.""" return mocker.patch('integration_scripts.logger.warning') @pytest.fixture def test_print_log_error(mocker): """Create fixture for logger.""" return mocker.patch('integration_scripts.logger.error') @pytest.fixture def test_get_s3_file_key(mocker): """Create fixture for get_s3_file_key.""" return mocker.patch( 'logic.ripper_feeder_logic.get_s3_file_key') # noinspection DuplicatedCode @pytest.mark.usefixtures('test_class_filter_list') @pytest.mark.usefixtures('test_class_audio_results') @pytest.mark.usefixtures('test_class_list_items') # @pytest.mark.usefixtures('test_audio_asset_list_class') class RipperFeederSpyTests(SpyAgency, TestCase): """Class for testing the ripper_feeder.""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.actions = dict() self.release_count = 0 self.results = list() self.foreign_relid = '12345' self.table_name = 'table_name' # Working ---------------------------------------------------------------- @patch('logic.ripper_feeder_logic.config') @patch('logic.ripper_feeder_logic.s3_util.filter_backoff') @patch('logic.ripper_feeder_logic.mark_tracks_skipped', return_value=([], 0)) def test_preprocess_track_list_skipped_S3(self, mock_skipped, mock_backoff, mock_config): """Test the pre-processing of the track list with skipped files from an S3 source.""" db_track_count_dict = {'12345': 5} mock_config.UNDO_ACTION = False mock_backoff.return_value = [] self.spy_on(file_utils.get_s3_file_key) act, process_list, release_track_count, rel_count, release_path, \ do_continue = logic.preprocess_track_list( self.actions, self.release_count, db_track_count_dict, self.test_class_audio_results, self.foreign_relid, self.table_name) assert file_utils.get_s3_file_key.calls[0].return_value == \ asset_paths.source_audio + '/' + self.foreign_relid + '/' + \ self.foreign_relid + '_' assert len(file_utils.get_s3_file_key.calls) == 1 assert mock_skipped.call_count == 1 assert mock_backoff.call_count == 1 assert do_continue is True # Working ---------------------------------------------------------------- @patch('logic.ripper_feeder_logic.config') @patch('logic.ripper_feeder_logic.mark_tracks_skipped', return_value=([], 0)) @patch('logic.ripper_feeder_logic.glob.glob') @patch('logic.ripper_feeder_logic.get_files_at_path') def test_preprocess_track_list_skipped_local(self, mock_get_file, mock_glob, mock_skipped, mock_config): """Test the pre-processing of the track list.""" db_track_count_dict = {'12345': 5} mock_config.UNDO_ACTION = False mock_glob.return_value = ['12345', ] mock_get_file.return_value = \ asset_paths.source_audio + '/' + self.foreign_relid + '/' + \ self.foreign_relid self.spy_on(orch_file_utils.get_files_at_path) act, process_list, release_track_count, rel_count, release_path, \ do_continue = logic.preprocess_track_list( self.actions, self.release_count, db_track_count_dict, self.test_class_audio_results, self.foreign_relid, self.table_name, use_local=True) assert release_path == self.foreign_relid assert mock_skipped.call_count == 1 assert do_continue is True assert process_list == list() # Working ---------------------------------------------------------------- @patch('logic.ripper_feeder_logic.config') @patch('logic.ripper_feeder_logic.s3_util.filter_backoff') @patch('logic.ripper_feeder_logic.mark_tracks_mismatched', return_value=([], 0)) def test_preprocess_track_list_mismatched_S3(self, mock_mismatch, mock_backoff, mock_config): """Test the pre-processing of the track list with mismatched tracks for an S3 source.""" db_track_count_list = {'12345': 5} mock_config.IGNORE_FILE_COUNT_MISMATCH = False mock_backoff.return_value = self.test_class_filter_list(num=4) self.spy_on(file_utils.get_s3_file_key) act, process_list, release_track_count, rel_count, release_path, \ do_continue = logic.preprocess_track_list( self.actions, self.release_count, db_track_count_list, self.test_class_audio_results, self.foreign_relid, self.table_name) expected_path = asset_paths.source_audio + '/' + \ self.foreign_relid + '/' + \ self.foreign_relid + '_' returned_path = file_utils.get_s3_file_key.calls[0].return_value assert returned_path == expected_path assert len(file_utils.get_s3_file_key.calls) == 1 assert mock_mismatch.call_count == 1 assert mock_backoff.call_count == 1 assert do_continue is True assert process_list == list() # Working ---------------------------------------------------------------- @patch('logic.ripper_feeder_logic.config') @patch('logic.ripper_feeder_logic.mark_tracks_mismatched', return_value=([], 0)) @patch('logic.ripper_feeder_logic.glob.glob') @patch('logic.ripper_feeder_logic.get_files_at_path') def test_preprocess_track_list_mismatched_local(self, mock_get_file, mock_glob, mock_mismatch, mock_config): """Test the pre-processing of the track list for a local file source.""" db_track_count_list = {'12345': 5} mock_config.IGNORE_FILE_COUNT_MISMATCH = False mock_glob.return_value = ['12345', ] mock_get_file.return_value = asset_paths.source_audio + '/' + \ self.foreign_relid + '/' + self.foreign_relid self.spy_on(orch_file_utils.get_files_at_path) act, process_list, release_track_count, rel_count, release_path, \ do_continue = logic.preprocess_track_list( self.actions, self.release_count, db_track_count_list, self.test_class_audio_results, self.foreign_relid, self.table_name, use_local=True) assert release_path == self.foreign_relid # assert len(file_utils.get_s3_file_key.calls) == 1 assert mock_mismatch.call_count == 1 assert do_continue is True assert process_list == list() # Working ---------------------------------------------------------------- @patch('logic.ripper_feeder_logic.s3_util.filter_backoff') @patch('logic.ripper_feeder_logic.mark_tracks_original') def test_preprocess_track_list_original_skip_S3(self, mock_original, mock_backoff): """Test the pre-processing of the track list.""" db_track_count_dict = {'12345': 5} mock_original.return_value = ( [], 0, False, len(self.test_class_audio_results), []) mock_backoff.return_value = self.test_class_filter_list(num=5) self.spy_on(file_utils.get_s3_file_key) act, process_list, release_track_count, rel_count, release_path, \ do_continue = logic.preprocess_track_list( self.actions, self.release_count, db_track_count_dict, self.test_class_audio_results, self.foreign_relid, self.table_name) assert file_utils.get_s3_file_key.calls[0].return_value == \ asset_paths.source_audio + '/' + self.foreign_relid + '/' + \ self.foreign_relid + '_' assert len(file_utils.get_s3_file_key.calls) == 2 assert mock_original.call_count == 1 assert mock_backoff.call_count == 2 assert do_continue is True assert process_list == list() # Working ---------------------------------------------------------------- @patch('logic.ripper_feeder_logic.config') @patch('logic.ripper_feeder_logic.mark_tracks_original') @patch('logic.ripper_feeder_logic.glob.glob') @patch('logic.ripper_feeder_logic.get_files_at_path') def test_preprocess_track_list_original_skip_local(self, mock_get_file, mock_glob, mock_original, mock_config): """Test the pre-processing of the track list.""" db_track_count_dict = {'12345': 5} mock_config.SOURCE_ROOT_FOLDER_PATH = '' mock_glob.return_value = ['12345', ] mock_original.return_value = ( [], 0, False, len(self.test_class_audio_results), []) mock_get_file.return_value = self.test_class_list_items(num=5) self.spy_on(orch_file_utils.get_files_at_path) act, process_list, release_track_count, rel_count, release_path, \ do_continue = logic.preprocess_track_list( self.actions, self.release_count, db_track_count_dict, self.test_class_audio_results, self.foreign_relid, self.table_name, use_local=True) assert release_path == self.foreign_relid # assert file_utils.get_s3_file_key.calls[0].return_value == \ # asset_paths.source_audio + '/' + self.s3_phono_relid + '/' + \ # self.s3_phono_relid + '_' # assert len(file_utils.get_s3_file_key.calls) == 2 assert mock_original.call_count == 1 # assert mock_backoff.call_count == 2 assert do_continue is True assert process_list == list() # Working ---------------------------------------------------------------- @patch('logic.ripper_feeder_logic.config') @patch('logic.ripper_feeder_logic.mark_tracks_original') @patch('logic.ripper_feeder_logic.glob.glob') @patch('logic.ripper_feeder_logic.get_files_at_path') def test_preprocess_track_list_original_no_skip_all_local(self, mock_get_file, mock_glob, mock_original, mock_config): """Test the pre-processing of the track list.""" db_track_count_dict = {'12345': 5} mock_config.SOURCE_ROOT_FOLDER_PATH = '' mock_glob.return_value = ['12345', ] mock_original.return_value = ( [], 0, True, 0, self.test_class_audio_results) mock_get_file.side_effect = [ self.test_class_list_items(num=5), self.test_class_list_items(num=5) ] self.spy_on(orch_file_utils.get_files_at_path) act, process_list, release_track_count, rel_count, release_path, \ do_continue = logic.preprocess_track_list( self.actions, self.release_count, db_track_count_dict, self.test_class_audio_results, self.foreign_relid, self.table_name, use_local=True) assert release_path == self.foreign_relid assert mock_original.call_count == 1 assert do_continue is False assert process_list == self.test_class_audio_results # Working ---------------------------------------------------------------- @patch('logic.ripper_feeder_logic.s3_util.filter_backoff') @patch('logic.ripper_feeder_logic.mark_tracks_original', return_value=([], 0, True, 0, ['test'])) def test_preprocess_track_list_original_no_skip_all_S3(self, mock_original, mock_backoff): """Test the pre-processing of the track list.""" db_track_count_dict = {'12345': 5} mock_backoff.return_value = self.test_class_filter_list(num=5) self.spy_on(file_utils.get_s3_file_key) act, process_list, release_track_count, rel_count, release_path, \ do_continue = logic.preprocess_track_list( self.actions, self.release_count, db_track_count_dict, self.test_class_audio_results, self.foreign_relid, self.table_name) assert file_utils.get_s3_file_key.calls[0].return_value == \ asset_paths.source_audio + '/' + self.foreign_relid + '/' + \ self.foreign_relid + '_' assert len(file_utils.get_s3_file_key.calls) == 2 assert mock_original.call_count == 1 assert mock_backoff.call_count == 2 assert do_continue is False assert process_list == ['test'] # Working -------------------------------------------------------------------- def test_mark_tracks_skipped_originals(test_actions_empty, test_audio_results, test_list_5_items, test_empty_list, test_print_log_info): """Test marking tracks as originals.""" release_count = 1 table_name = 'table_name' actions, release_track_count, not_skipped, skipped_count, \ update_tracks = logic.mark_tracks_original( release_count, table_name, test_actions_empty, test_audio_results, test_list_5_items, test_empty_list) assert not_skipped == 0 assert skipped_count == 5 assert release_track_count == len(test_audio_results) assert test_print_log_info.call_count == 5 assert len(update_tracks) == 0 for file_name, action in actions.items(): assert action == 'skipped' # Working -------------------------------------------------------------------- def test_mark_tracks_skipped(test_actions_empty, test_audio_results, test_print_log_warning): """Test marking tracks as skipped.""" release_count = 5 table_name = 'table_name' actions, release_track_count = logic.mark_tracks_skipped( release_count, table_name, test_actions_empty, test_audio_results) assert release_count == 5 assert release_track_count == 5 assert test_print_log_warning.call_count == 5 for file_name, action in actions.items(): assert action == 'skipped' # Working -------------------------------------------------------------------- def test_mark_tracks_originals(test_actions_empty, test_audio_results, test_print_log_info, test_list_5_items, test_empty_list): """Test marking tracks as originals.""" release_count = 1 table_name = 'table_name' actions, release_track_count, not_skipped, skipped_count, \ update_tracks = logic.mark_tracks_original( release_count, table_name, test_actions_empty, test_audio_results, test_empty_list, test_list_5_items) assert not_skipped == 5 assert skipped_count == 0 assert test_print_log_info.call_count == 0 assert len(update_tracks) == 5 assert release_track_count == len(test_audio_results) for file_name, action in actions.items(): assert action == 'original' # Working -------------------------------------------------------------------- def test_mark_tracks_new_originals(test_actions_empty, test_audio_results, test_list_5_items, test_print_log_info): """Test marking tracks as originals.""" release_count = 1 table_name = 'table_name' actions, release_track_count, not_skipped, skipped_count, \ update_tracks = logic.mark_tracks_original( release_count, table_name, test_actions_empty, test_audio_results, test_list_5_items, test_list_5_items) assert not_skipped == 5 assert skipped_count == 0 assert release_track_count == len(test_audio_results) assert test_print_log_info.call_count == 0 assert len(update_tracks) == 5 for file_name, action in actions.items(): assert action == 'new original' # Working -------------------------------------------------------------------- def test_mark_tracks_missing_originals(test_actions_empty, test_audio_results, test_empty_list, test_print_log_info): """Test marking tracks as originals.""" release_count = 1 table_name = 'table_name' actions, release_track_count, not_skipped, skipped_count, \ update_tracks = logic.mark_tracks_original( release_count, table_name, test_actions_empty, test_audio_results, test_empty_list, test_empty_list) assert not_skipped == 0 assert skipped_count == 5 assert release_track_count == 5 assert release_count == 1 assert test_print_log_info.call_count == 0 assert len(update_tracks) == 0 for file_name, action in actions.items(): assert action == 'file missing' # Working -------------------------------------------------------------------- def test_backup_donefile_s3(test_print_log_info, test_sentry, test_copy_from_backoff): """Test backing up a donefile object.""" release_count = 5 table_name = 'table_name' backup_key = 'backup_key' source_key = 'source_key' logic.backup_donefile( table_name, release_count, backup_key, source_key) assert test_copy_from_backoff.call_count == 1 assert test_print_log_info.call_count == 0 assert test_sentry.call_count == 0 # def test_backup_donefile_local(test_print_log_info, # test_sentry, # test_copy_from_backoff): # """Test backing up a donefile object.""" # release_count = 5 # table_name = 'table_name' # backup_key = 'backup_key' # source_key = 'source_key' # # logic.backup_donefile( # table_name, release_count, backup_key, source_key) # # assert test_copy_from_backoff.call_count == 1 # assert test_print_log_info.call_count == 0 # assert test_sentry.call_count == 0 # Working -------------------------------------------------------------------- def test_backup_donefile_exception_s3(test_print_log_error, test_sentry, test_copy_from_backoff_exception): """Test restoring a backed up object failure.""" release_count = 5 table_name = 'table_name' backup_key = 'backup_key' source_key = 'source_key' logic.backup_donefile( table_name, release_count, backup_key, source_key) assert test_copy_from_backoff_exception.call_count == 1 assert test_print_log_error.call_count == 1 assert test_sentry.captureMessage.call_count == 1 # def test_backup_donefile_exception_local(test_print_log_error, # test_sentry, # test_copy_from_backoff_exception): # """Test restoring a backed up object failure.""" # release_count = 5 # # table_name = 'table_name' # backup_key = 'backup_key' # source_key = 'source_key' # # logic.backup_donefile( # table_name, release_count, backup_key, source_key) # # assert test_copy_from_backoff_exception.call_count == 1 # assert test_print_log_error.call_count == 1 # assert test_sentry.captureMessage.call_count == 1 # Working -------------------------------------------------------------------- def test_backup_keys_s3( test_sentry, test_print_log_info, test_copy_from_backoff): """Test backing up an S3 object by key - Covers.""" delete_list = [] backup_key = 'backup_key' source_key = 'source_key' action, do_continue, err, delete_list = logic.backup_keys( delete_list, backup_key, source_key) assert test_copy_from_backoff.call_count == 1 assert action == 'backed up' assert do_continue is False assert test_sentry.call_count == 0 assert delete_list[0] == 'source_key' assert test_print_log_info.call_count == 0 assert err is None # def test_backup_keys_local( # test_sentry, test_print_log_info, test_copy_from_backoff): # """Test backing up an S3 object by key - Covers.""" # delete_list = [] # # backup_key = 'backup_key' # source_key = 'source_key' # # action, do_continue, err, delete_list = logic.backup_keys( # delete_list, backup_key, source_key) # # assert test_copy_from_backoff.call_count == 1 # assert action == 'backed up' # assert do_continue is False # assert test_sentry.call_count == 0 # assert delete_list[0] == 'source_key' # assert test_print_log_info.call_count == 0 # assert err is None # Working -------------------------------------------------------------------- def test_backup_keys_exception_s3(test_copy_from_backoff_exception): """Test backing up an S3 object by key failure - Covers.""" delete_list = [] backup_key = 'backup_key' source_key = 'source_key' action, do_continue, err, delete_list = logic.backup_keys( delete_list, backup_key, source_key) assert test_copy_from_backoff_exception.call_count == 1 assert action == 'backup failed' assert do_continue is True assert delete_list == [] assert err is not None # def test_backup_keys_exception_local(test_copy_from_backoff_exception): # """Test backing up an S3 object by key failure - Covers.""" # delete_list = [] # # backup_key = 'backup_key' # source_key = 'source_key' # # action, do_continue, err, delete_list = logic.backup_keys( # delete_list, backup_key, source_key) # # assert test_copy_from_backoff_exception.call_count == 1 # assert action == 'backup failed' # assert do_continue is True # assert delete_list == [] # assert err is not None # Working -------------------------------------------------------------------- def test_backup_originals_s3(test_copy_from_backoff): """Test backing up an S3 object by key - Audio.""" delete_list = [] backup_key = 'backup_key' source_key = 'source_key' file_source_key = source_key action, do_continue, err, delete_list = logic.backup_originals( delete_list, backup_key, file_source_key) assert action == 'backed up' assert do_continue is False assert test_copy_from_backoff.call_count == 1 assert err is None # def test_backup_originals_local(test_copy_from_backoff): # """Test backing up an S3 object by key - Audio.""" # delete_list = [] # # backup_key = 'backup_key' # source_key = 'source_key' # # file_source_key = source_key # # action, do_continue, err, delete_list = logic.backup_originals( # delete_list, backup_key, file_source_key) # # assert action == 'backed up' # assert do_continue is False # assert test_copy_from_backoff.call_count == 1 # assert err is None # Working -------------------------------------------------------------------- def test_backup_originals_exception_s3(test_copy_from_backoff_exception): """Test backing up an S3 object by key - Audio.""" delete_list = [] backup_key = 'backup_key' source_key = 'source_key' file_source_key = source_key action, do_continue, err, delete_list = logic.backup_originals( delete_list, backup_key, file_source_key) assert action == 'backup failed' assert do_continue is True assert test_copy_from_backoff_exception.call_count == 1 assert err is not None # def test_backup_originals_exception_local(test_copy_from_backoff_exception): # """Test backing up an S3 object by key - Audio.""" # # delete_list = [] # # backup_key = 'backup_key' # source_key = 'source_key' # # file_source_key = source_key # # action, do_continue, err, delete_list = logic.backup_originals( # delete_list, backup_key, file_source_key) # # assert action == 'backup failed' # assert do_continue is True # assert test_copy_from_backoff_exception.call_count == 1 # assert err is not None # Working -------------------------------------------------------------------- def test_check_cover_key_exists_s3(test_head_object_backoff): """Test if a cover key is present in the S3 bucket.""" key = 'key' action, do_continue, err = logic.check_cover_key_exists( key) assert do_continue is False assert action == 'exists' assert test_head_object_backoff.call_count == 1 assert err is None # def test_check_cover_key_exists_local(test_head_object_backoff): # """Test if a cover key is present in the S3 bucket.""" # key = 'key' # # action, do_continue, err = logic.check_cover_key_exists( # key) # # assert do_continue is False # assert action == 'exists' # assert test_head_object_backoff.call_count == 1 # assert err is None # Working -------------------------------------------------------------------- def test_check_cover_key_exists_exception_s3( test_head_object_backoff_exception): """Test if a cover key is present in the S3 bucket.""" key = 'key' action, do_continue, err = logic.check_cover_key_exists( key) assert do_continue is True assert action == 'missing' assert test_head_object_backoff_exception.call_count == 1 assert err is not None # def test_check_cover_key_exists_exception_local( # test_head_object_backoff_exception): # """Test if a cover key is present in the S3 bucket.""" # key = 'key' # # action, do_continue, err = logic.check_cover_key_exists( # key) # # assert do_continue is True # assert action == 'missing' # assert test_head_object_backoff_exception.call_count == 1 # assert err is not None # Working -------------------------------------------------------------------- def test_mark_tracks_mismatched( test_actions_empty, test_audio_results, test_print_log_warning): """Test marking tracks as mismatched.""" release_count = 5 table_name = 'table_name' actions, release_track_count = \ logic.mark_tracks_mismatched(release_count, table_name, test_actions_empty, test_audio_results) assert release_count == 5 assert release_track_count == 5 assert test_print_log_warning.call_count == 5 for file_name, action in actions.items(): assert action == 'file count mismatch' # Working -------------------------------------------------------------------- def test_restore_backup_s3(test_copy_from_backoff): """Test backing up a donefile object.""" delete_list = [] backup_key = 'backup_key' target_key = 'target_key' action, do_continue, err, delete_list = logic.restore_backup( delete_list, backup_key, target_key) assert action == 'restored' assert do_continue is False assert test_copy_from_backoff.call_count == 1 assert len(delete_list) > 0 assert err is None # def test_restore_backup_local(test_copy_from_backoff): # """Test backing up a donefile object.""" # delete_list = [] # # backup_key = 'backup_key' # target_key = 'target_key' # # action, do_continue, err, delete_list = logic.restore_backup( # delete_list, backup_key, target_key) # # assert action == 'restored' # assert do_continue is False # assert test_copy_from_backoff.call_count == 1 # assert len(delete_list) > 0 # assert err is None # Working -------------------------------------------------------------------- def test_restore_backup_exception_s3(test_copy_from_backoff_exception): """Test backing up a donefile object.""" delete_list = [] backup_key = 'backup_key' target_key = 'target_key' action, do_continue, err, delete_list = logic.restore_backup( delete_list, backup_key, target_key) assert action == 'restore failed' assert do_continue is True assert len(delete_list) == 0 assert test_copy_from_backoff_exception.call_count == 1 assert err is not None # def test_restore_backup_exception_local(test_copy_from_backoff_exception): # """Test backing up a donefile object.""" # delete_list = [] # # backup_key = 'backup_key' # target_key = 'target_key' # # action, do_continue, err, delete_list = logic.restore_backup( # delete_list, backup_key, target_key) # # assert action == 'restore failed' # assert do_continue is True # assert len(delete_list) == 0 # assert test_copy_from_backoff_exception.call_count == 1 # assert err is not None # Working -------------------------------------------------------------------- def test_download_file_s3(test_download_fileobj_backoff): """Test the downloading of a file object by key - Audio.""" key = 'key' target_path = tempfile.TemporaryDirectory().name action, do_continue, err = logic.download_file(key, target_path) assert test_download_fileobj_backoff.call_count == 1 assert action == 'downloaded' assert do_continue is False assert err is None # def test_download_file_local(test_download_fileobj_backoff): # """Test the downloading of a file object by key - Audio.""" # # key = 'key' # # target_path = tempfile.TemporaryDirectory().name # # action, do_continue, err = logic.download_file(key, target_path) # # assert test_download_fileobj_backoff.call_count == 1 # assert action == 'downloaded' # assert do_continue is False # assert err is None # Working -------------------------------------------------------------------- def test_download_file_exception_s3(test_download_fileobj_backoff_exception): """Test the downloading of a file object by key - Audio.""" key = 'key' target_path = tempfile.TemporaryDirectory().name action, do_continue, err = logic.download_file(key, target_path) assert test_download_fileobj_backoff_exception.call_count == 1 assert action == 'failed' assert do_continue is True assert err is not None # def test_download_file_exception_local( # test_download_fileobj_backoff_exception): # """Test the downloading of a file object by key - Audio.""" # key = 'key' # target_path = tempfile.TemporaryDirectory().name # # action, do_continue, err = logic.download_file(key, target_path) # # assert test_download_fileobj_backoff_exception.call_count == 1 # assert action == 'failed' # assert do_continue is True # assert err is not None # Working -------------------------------------------------------------------- def test_restore_donefile_s3(test_print_log_info, test_sentry, test_copy_from_backoff): """Restore a donefile from the backup directory.""" release_count = 1 key = 'key' table_name = 'table_name' backup_key = 'backup_key' logic.restore_donefile( table_name, release_count, backup_key, key) assert test_copy_from_backoff.call_count == 1 assert test_print_log_info.call_count == 0 assert test_sentry.call_count == 0 # def test_restore_donefile_local(test_print_log_info, test_sentry, # test_copy_from_backoff): # """Restore a donefile from the backup directory.""" # release_count = 1 # # key = 'key' # table_name = 'table_name' # backup_key = 'backup_key' # # logic.restore_donefile( # table_name, release_count, backup_key, key) # # assert test_copy_from_backoff.call_count == 1 # assert test_print_log_info.call_count == 0 # assert test_sentry.call_count == 0 # Working -------------------------------------------------------------------- def test_restore_donefile_exception_s3( test_print_log_error, test_sentry, test_copy_from_backoff_exception): """Restore a donefile from the backup directory.""" release_count = 1 key = 'key' table_name = 'table_name' backup_key = 'backup_key' logic.restore_donefile( table_name, release_count, backup_key, key) assert test_copy_from_backoff_exception.call_count == 1 assert test_print_log_error.call_count == 1 assert test_sentry.captureMessage.call_count == 1 # def test_restore_donefile_exception_local( # test_print_log_error, test_sentry, test_copy_from_backoff_exception): # """Restore a donefile from the backup directory.""" # release_count = 1 # # key = 'key' # table_name = 'table_name' # backup_key = 'backup_key' # # logic.restore_donefile( # table_name, release_count, backup_key, key) # # assert test_copy_from_backoff_exception.call_count == 1 # assert test_print_log_error.call_count == 1 # assert test_sentry.captureMessage.call_count == 1 # Working -------------------------------------------------------------------- def test_download_cover_exception_s3(test_download_fileobj_backoff): """Test the downloading of a file object by key - Cover.""" temp_to_delete_list = [] tmp_target_path = '/tmp/path' key = 'key' action, do_continue, err, delete_list = logic.download_cover( key, temp_to_delete_list, tmp_target_path) assert test_download_fileobj_backoff.call_count == 1 assert action == 'downloaded' assert do_continue is False assert err is None assert len(delete_list) == 1 assert tmp_target_path in delete_list # def test_download_cover_exception_local(test_download_fileobj_backoff): # """Test the downloading of a file object by key - Cover.""" # temp_to_delete_list = [] # tmp_target_path = '/tmp/path' # # key = 'key' # # action, do_continue, err, delete_list = logic.download_cover( # key, temp_to_delete_list, tmp_target_path) # # assert test_download_fileobj_backoff.call_count == 1 # assert action == 'downloaded' # assert do_continue is False # assert err is None # assert len(delete_list) == 1 # assert tmp_target_path in delete_list # Working -------------------------------------------------------------------- def test_download_cover_s3(test_download_fileobj_backoff_exception): """Test the downloading of a file object by key - Cover.""" temp_to_delete_list = [] tmp_target_path = '/tmp/path' key = 'key' action, do_continue, err, delete_list = logic.download_cover( key, temp_to_delete_list, tmp_target_path) assert test_download_fileobj_backoff_exception.call_count == 1 assert action == 'failed' assert do_continue is True assert err is not None assert len(delete_list) == 0 # def test_download_cover_local(test_download_fileobj_backoff_exception): # """Test the downloading of a file object by key - Cover.""" # temp_to_delete_list = [] # tmp_target_path = '/tmp/path' # # key = 'key' # # action, do_continue, err, delete_list = logic.download_cover( # key, temp_to_delete_list, tmp_target_path) # # assert test_download_fileobj_backoff_exception.call_count == 1 # assert action == 'failed' # assert do_continue is True # assert err is not None # assert len(delete_list) == 0 # Working -------------------------------------------------------------------- def test_mark_cover_original_s3(mocker): """Test marking cover as original.""" key = 'key' backup_key = 'backup_key' empty = [] value = [SimpleNamespace(key='value')] responses = { 'no_backup_orig': [empty, value], 'backup_orig': [value, value], 'backup_no_orig': [value, empty], 'no_backup_no_orig': [empty, empty] } mock_filter_1 = mocker.patch( 'integration_scripts.s3_backoff_utils.filter_backoff', side_effect=responses['no_backup_orig']) action_1, continue_1, found_key_1 = logic.mark_cover_original( key, backup_key) # assert mock_filter_1.call_count == 1 mock_filter_2 = mocker.patch( 'integration_scripts.s3_backoff_utils.filter_backoff', side_effect=responses['backup_orig']) action_2, continue_2, found_key_2 = logic.mark_cover_original( key, backup_key) # assert mock_filter_2.call_count == 2 mock_filter_3 = mocker.patch( 'integration_scripts.s3_backoff_utils.filter_backoff', side_effect=responses['backup_no_orig']) action_3, continue_3, found_key_3 = logic.mark_cover_original( key, backup_key) # assert mock_filter_3.call_count == 2 mock_filter_4 = mocker.patch( 'integration_scripts.s3_backoff_utils.filter_backoff', side_effect=responses['no_backup_no_orig']) action_4, continue_4, found_key_4 = logic.mark_cover_original( key, backup_key) # assert mock_filter_3.call_count == 1 assert action_1 == 'processing' assert action_2 == 'processing' assert action_3 == 'skipped' assert action_4 == 'processing' assert continue_1 is False assert continue_2 is False assert continue_3 is True assert continue_4 is False assert found_key_1 is None assert found_key_2 is None assert found_key_3 == 'value' assert found_key_4 is None assert mock_filter_1.call_count == 1 assert mock_filter_2.call_count == 2 assert mock_filter_3.call_count == 2 assert mock_filter_4.call_count == 1 # def test_mark_cover_original_local(mocker): # """Test marking cover as original.""" # key = 'key' # backup_key = 'backup_key' # # empty = [] # value = [SimpleNamespace(key='value')] # # responses = { # 'no_backup_orig': [empty, value], # 'backup_orig': [value, value], # 'backup_no_orig': [value, empty], # 'no_backup_no_orig': [empty, empty] # } # # mock_filter_1 = mocker.patch( # 'integration_scripts.s3_backoff_utils.filter_backoff', # side_effect=responses['no_backup_orig']) # # action_1, continue_1, found_key_1 = logic.mark_cover_original( # key, backup_key) # # # assert mock_filter_1.call_count == 1 # # mock_filter_2 = mocker.patch( # 'integration_scripts.s3_backoff_utils.filter_backoff', # side_effect=responses['backup_orig']) # # action_2, continue_2, found_key_2 = logic.mark_cover_original( # key, backup_key) # # # assert mock_filter_2.call_count == 2 # # mock_filter_3 = mocker.patch( # 'integration_scripts.s3_backoff_utils.filter_backoff', # side_effect=responses['backup_no_orig']) # # action_3, continue_3, found_key_3 = logic.mark_cover_original( # key, backup_key) # # # assert mock_filter_3.call_count == 2 # # mock_filter_4 = mocker.patch( # 'integration_scripts.s3_backoff_utils.filter_backoff', # side_effect=responses['no_backup_no_orig']) # # action_4, continue_4, found_key_4 = logic.mark_cover_original( # key, backup_key) # # # assert mock_filter_3.call_count == 1 # # assert action_1 == 'checking' # assert action_2 == 'processing' # assert action_3 == 'skipped' # assert action_4 == 'checking' # # assert continue_1 is False # assert continue_2 is False # assert continue_3 is True # assert continue_4 is False # # assert found_key_1 is None # assert found_key_2 is None # assert found_key_3 == 'value' # assert found_key_4 is None # # assert mock_filter_1.call_count == 1 # assert mock_filter_2.call_count == 2 # assert mock_filter_3.call_count == 2 # assert mock_filter_4.call_count == 1 # Working -------------------------------------------------------------------- def test_delete_marked_files_s3(test_delete_backoff): """Test deletion of files marked for deletion.""" actions = {} release_count = 1 delete_list = ['1/1', '2/2', '3/3'] table_name = 'table_name' actions = logic.delete_marked_files( actions, release_count, delete_list, table_name) assert test_delete_backoff.call_count == len(delete_list) assert len(actions) == len(delete_list) for file, status in actions.items(): assert status == 'completed' # def test_delete_marked_files_local(test_delete_backoff): # """Test deletion of files marked for deletion.""" # actions = {} # release_count = 1 # delete_list = ['1/1', '2/2', '3/3'] # # table_name = 'table_name' # # actions = logic.delete_marked_files( # actions, release_count, delete_list, table_name) # # assert test_delete_backoff.call_count == len(delete_list) # assert len(actions) == len(delete_list) # for file, status in actions.items(): # assert status == 'completed' # Working -------------------------------------------------------------------- def test_delete_marked_files_exception_s3(test_delete_backoff_exception): """Test deletion of files marked for deletion.""" actions = {} release_count = 1 delete_list = ['1/1', '2/2', '3/3'] table_name = 'table_name' actions = logic.delete_marked_files( actions, release_count, delete_list, table_name) assert test_delete_backoff_exception.call_count == len(delete_list) assert len(actions) == 0 # def test_delete_marked_files_exception_local(test_delete_backoff_exception): # """Test deletion of files marked for deletion.""" # actions = {} # release_count = 1 # delete_list = ['1/1', '2/2', '3/3'] # # table_name = 'table_name' # # actions = logic.delete_marked_files( # actions, release_count, delete_list, table_name) # # assert test_delete_backoff_exception.call_count == len(delete_list) # assert len(actions) == 0 # Working -------------------------------------------------------------------- def test_delete_donefile_s3( test_delete_backoff, test_print_log_info, test_sentry): """Test the deletion of donefiles.""" release_count = 1 key = 'key' table_name = 'table_name' logic.delete_donefile(release_count, key, table_name) assert test_delete_backoff.call_count == 1 assert test_print_log_info.call_count == 0 assert test_sentry.captureMessage.call_count == 0 # def test_delete_donefile_local( # test_delete_backoff, test_print_log, test_sentry): # """Test the deletion of donefiles.""" # release_count = 1 # # key = 'key' # table_name = 'table_name' # # logic.delete_donefile(release_count, key, table_name) # # assert test_delete_backoff.call_count == 1 # assert test_print_log.call_count == 0 # assert test_sentry.call_count == 0 # Working -------------------------------------------------------------------- def test_delete_donefile_exception_s3( test_delete_backoff_exception, test_print_log_error, test_sentry): """Test the deletion of donefiles.""" release_count = 1 key = 'key' table_name = 'table_name' logic.delete_donefile(release_count, key, table_name) assert test_delete_backoff_exception.call_count == 1 assert test_print_log_error.call_count == 1 assert test_sentry.captureMessage.call_count == 1 # def test_delete_donefile_exception_local( # test_delete_backoff_exception, test_print_log, test_sentry): # """Test the deletion of donefiles.""" # release_count = 1 # # key = 'key' # table_name = 'table_name' # # logic.delete_donefile(release_count, key, table_name) # # assert test_delete_backoff_exception.call_count == 1 # assert test_print_log.call_count == 1 # assert test_sentry.call_count == 1 # Working -------------------------------------------------------------------- def test_delete_covers_s3( mocker, test_delete_backoff, test_print_log_info, test_sentry): """Test the deletion of cover objects on S3 by key.""" release_count = 1 delete_list = ['1.wav', '2.wav', '3.wav'] temp_list = ['a', 'b', 'c'] actions = dict() key = 'key' table_name = 'table_name' for item in delete_list: actions[os.path.splitext(item)[0]] = '' mock_remove = mocker.patch('os.remove') actions = logic.delete_covers(actions, release_count, delete_list, key, table_name, temp_list) assert test_delete_backoff.call_count == len(delete_list) for slot, action in actions.items(): assert action == 'completed' assert mock_remove.call_count == len(temp_list) assert test_print_log_info.call_count == 0 assert test_sentry.captureMessage.call_count == 0 # def test_delete_covers_local( # mocker, test_delete_backoff, test_print_log, test_sentry): # """Test the deletion of cover objects on S3 by key.""" # release_count = 1 # delete_list = ['1.wav', '2.wav', '3.wav'] # temp_list = ['a', 'b', 'c'] # actions = dict() # # key = 'key' # table_name = 'table_name' # # for item in delete_list: # actions[os.path.splitext(item)[0]] = '' # # mock_remove = mocker.patch('os.remove') # # actions = logic.delete_covers(actions, # release_count, # delete_list, key, table_name, # temp_list) # # assert test_delete_backoff.call_count == len(delete_list) # for slot, action in actions.items(): # assert action == 'completed' # assert mock_remove.call_count == len(temp_list) # assert test_print_log.call_count == 0 # assert test_sentry.call_count == 0 # Working -------------------------------------------------------------------- def test_delete_covers_exception_s3( mocker, test_delete_backoff_exception, test_print_log_info, test_sentry): """Test the deletion of cover objects on S3 by key.""" release_count = 1 delete_list = ['1.wav', '2.wav', '3.wav'] temp_list = ['a', 'b', 'c'] actions = dict() key = 'key' table_name = 'table_name' for item in delete_list: actions[os.path.splitext(item)[0]] = '' mock_remove = mocker.patch('os.remove', side_effect=Exception()) actions = logic.delete_covers(actions, release_count, delete_list, key, table_name, temp_list) assert test_delete_backoff_exception.call_count == len(delete_list) for slot, action in actions.items(): assert action == 'begin delete' assert mock_remove.call_count == len(temp_list) assert test_print_log_info.call_count == len(delete_list) assert test_sentry.captureMessage.call_count == \ len(delete_list) + len(temp_list) # def test_delete_covers_exception_local( # mocker, test_delete_backoff_exception, test_print_log, # test_sentry): # """Test the deletion of cover objects on S3 by key.""" # release_count = 1 # delete_list = ['1.wav', '2.wav', '3.wav'] # temp_list = ['a', 'b', 'c'] # actions = dict() # # key = 'key' # table_name = 'table_name' # # for item in delete_list: # actions[os.path.splitext(item)[0]] = '' # # mock_remove = mocker.patch('os.remove', side_effect=Exception()) # # actions = logic.delete_covers(actions, # release_count, # delete_list, key, table_name, # temp_list) # # assert test_delete_backoff_exception.call_count == len(delete_list) # for slot, action in actions.items(): # assert action == 'begin delete' # assert mock_remove.call_count == len(temp_list) # assert test_print_log.call_count == len(delete_list) # assert test_sentry.call_count == len(delete_list) + len(temp_list) # Working -------------------------------------------------------------------- def test_restore_cover_backup_s3(test_copy_from_backoff, test_delete_backoff): """Test backing up a donefile object.""" key = 'key' backup_key = 'backup_key' action, do_continue, msg, err = logic.restore_cover_backup(backup_key, key) assert action == 'restored' assert do_continue is False assert test_copy_from_backoff.call_count == 1 assert test_delete_backoff.call_count == 1 assert msg == '' assert err is None # def test_restore_cover_backup_local( # test_copy_from_backoff, test_delete_backoff): # """Test backing up a donefile object.""" # # key = 'key' # backup_key = 'backup_key' # # action, do_continue, msg, err = logic.restore_cover_backup( # backup_key, key) # # assert action == 'restored' # assert do_continue is False # assert test_copy_from_backoff.call_count == 1 # assert test_delete_backoff.call_count == 1 # assert msg == '' # assert err is None # Working -------------------------------------------------------------------- def test_restore_cover_backup_exception_s3( test_copy_from_backoff_exception, test_delete_backoff): """Test backing up a donefile object.""" key = 'key' backup_key = 'backup_key' action, do_continue, msg, err = logic.restore_cover_backup(backup_key, key) assert action == 'failed' assert do_continue is True assert test_copy_from_backoff_exception.call_count == 1 assert test_delete_backoff.call_count == 0 assert 'S3 copy failed' in msg assert err is not None # def test_restore_cover_backup_exception_local( # test_copy_from_backoff_exception, test_delete_backoff): # """Test backing up a donefile object.""" # # key = 'key' # backup_key = 'backup_key' # # action, do_continue, msg, err = logic.restore_cover_backup( # backup_key, key) # # assert action == 'failed' # assert do_continue is True # assert test_copy_from_backoff_exception.call_count == 1 # assert test_delete_backoff.call_count == 0 # assert 'S3 copy failed' in msg # assert err is not None # Working -------------------------------------------------------------------- def test_restore_cover_backup_delete_exception_s3( test_copy_from_backoff, test_delete_backoff_exception): """Test backing up a donefile object.""" key = 'key' backup_key = 'backup_key' action, do_continue, msg, err = logic.restore_cover_backup(backup_key, key) assert action == 'failed' assert do_continue is True assert test_copy_from_backoff.call_count == 1 assert test_delete_backoff_exception.call_count == 1 assert 'S3 delete failed' in msg assert err is not None # def test_restore_cover_backup_delete_exception_local( # test_copy_from_backoff, test_delete_backoff_exception): # """Test backing up a donefile object.""" # # key = 'key' # backup_key = 'backup_key' # # action, do_continue, msg, err = logic.restore_cover_backup(backup_key, # key) # # assert action == 'failed' # assert do_continue is True # assert test_copy_from_backoff.call_count == 1 # assert test_delete_backoff_exception.call_count == 1 # assert 'S3 delete failed' in msg # assert err is not None # @patch('ripper_feeder.config') # @patch('ripper_feeder_logic.check_terminal_conditions', # return_value=(False, '')) # @patch('ripper_feeder_logic.start_log') # @patch('ripper_feeder.preprocess_track_list') # @patch('ripper_feeder.update_log') # @patch('ripper_feeder.get_audio_assets') # # @patch('ripper_feeder.transform_audio_assets', # # return_value=({}, 5)) # @patch('ripper_feeder.print_log') # def test_transfer_audio_tablenum(self, mock_print_log, # mock_get_audio, mock_update_log, # mock_preprocess, mock_start_log, # mock_check_term_conditions,mock_config): # """Test the transfer of audio files from s3 to the ripper.""" # args = SimpleNamespace(tablenum='1') # tmp_dir = tempfile.TemporaryDirectory() # log_file_path = tmp_dir.name # date = '01/01/3089' # skipped_all_files = True # # mock_get_audio.return_value = self.test_audio_asset_list_class # mock_preprocess.return_value = self.test_preprocess # # transfer_audio(args, log_file_path, date, skipped_all_files) # # mock_start_log.assert_called_with(args.tablenum, date, log_file_path) # # def test_get_audio_assets(self, ): # """Test getting audio assets from snowflake.""" # # get_audio_assets(table_name=None) # # def test_get_cover_assets(): # """Test getting cover assets from snowflake.""" # get_cover_assets(table_name=None) # # # def test_update_log(): # """Test updating the log.""" # update_log(log_file, action_list, t_name, source_relid, # orchard_upc, asset_type, asset_count_num) # # # def test_create_target_paths(): # """Test the creation of paths for file writing.""" # create_target_paths(orch_file_name, orch_upc, copy_local=False) # # # def test_transfer_covers(): # """Test the transfer of cover files from S3 to the ripper.""" # transfer_covers(log_file_path, skipped_all_files)