"""Tests for Proper class.""" import io import unittest from unittest.mock import call from unittest.mock import MagicMock from unittest.mock import patch from freezegun import freeze_time import pytest from feed_sender.flows.proper_new_releases_tracks import exceptions from feed_sender.flows.proper_new_releases_tracks import proper from feed_sender.flows.proper_new_releases_tracks.conf import settings from feed_sender.flows.proper_new_releases_tracks.conf import ( transform_settings) class TestProper(unittest.TestCase): """Testing class for the unit tests executing.""" def setUp(self): """Fixture for the Proper class.""" cutoff = '2000-06-30 09:50' self.proper = proper.Proper(cutoff=cutoff) self.proper.art_relations_db = 'test_ar' self.proper.physical_delivery_db = 'test_pd' self.proper.direct_delivery_db = 'test_dd' self.proper.store_id = 88888 self.proper._release_ids = None self.proper.total_releases = 0 @patch( 'feed_sender.flows.proper_new_releases_tracks.proper.s3.read_s3_object' ) def test_fetch_products(self, mock_get): """Test if ids are set properly from queue messages.""" mock_get.return_value = '1,11111\n2,22222\n3,33333\n' self.proper.fetch_products(MagicMock(), MagicMock()) assert self.proper._release_ids == [11111, 22222, 33333] assert self.proper.total_releases == 3 @patch( 'feed_sender.flows.proper_new_releases_tracks.proper.s3.read_s3_object' ) def test_fetch_products_when_no_jobs(self, mock_get): """Test when there's no jobs.""" mock_get.return_value = '' self.proper.fetch_products(MagicMock(), MagicMock()) assert not self.proper._release_ids assert self.proper.total_releases == 0 @patch('feed_sender.flows.proper_new_releases_tracks.proper.mysql') def test_filter_products_query(self, mock_mysql): """Check if correct filter SQL is used.""" mock_results = [ {'release_id': 1, 'last_updated': '2000-06-30 09:50', 'column': 'test1'}, {'release_id': 2, 'last_updated': '2000-06-30 10:50', 'column': 'test2'}] mock_mysql.execute_query = MagicMock(return_value=mock_results) self.proper._release_ids = [] self.proper.filter_products_query() assert self.proper._release_ids == [1, 2] @patch('feed_sender.flows.proper_new_releases_tracks.proper.mysql') def test_product_data_query(self, mock_mysql): """Check if correct filter SQL is used.""" self.proper._release_ids = [1, 2] self.proper.total_releases = 1 mock_results = [ {'release_id': 1, 'column': 'test1'}, {'release_id': 2, 'column': 'test2'}] mock_mysql.execute_query = MagicMock(return_value=mock_results) actual = self.proper._product_data_query() assert mock_mysql.execute_query.called is True assert actual == mock_results @patch('feed_sender.flows.proper_new_releases_tracks.proper.' 'Proper._product_data_query') def test_convert_releases_to_csv_wrong_mapping( self, mock_product_data_query): """Test error when number of records does not match total releases.""" sample_row = [ { 'vendor_id': 123, 'company': 'ABC' } ] mock_product_data_query.return_value = sample_row mock_genre_subgenre_mapping = { '1_4': {'subgenre': 'Sub Rock', 'genre': 'Rock'} } self.proper.total_releases = 2 with patch( 'feed_sender.flows.proper_new_releases_tracks.proper.' 'Proper.fetch_genre_mapping', return_value=mock_genre_subgenre_mapping): with pytest.raises(Exception) as err: self.proper.convert_releases_to_csv('s3://abc', 'file.csv') assert str(err.value) == ( 'Task: convert_releases_to_csv. Bad mapping occurred in ' 'the database. Number of records: 1, Total releases: 2') @patch('feed_sender.flows.proper_new_releases_tracks.proper.smart_open') def test_write_releases_to_csv(self, mock_open): """Test if transformed row is written via smart open.""" def test_function(instring, **kwargs): condition = kwargs.get('individual') if condition == 'Y': return instring test_map = [ {'proper': 'LabelCode', 'orchard': 'vendor_id'}, {'proper': 'LabelName', 'orchard': 'company'}, {'proper': 'Artist', 'orchard': 'artist_name', 'transform': test_function, 'depends': ['individual']}] sample_row = [ { 'vendor_id': 123, 'company': 'ABC', 'artist_name': 'Maria', 'individual': 'Y', } ] transformed_row = '"123","ABC","Maria"\r\n' self.proper._product_data_query = MagicMock(return_value=sample_row) # mock smart_open and file handles. smart_open_obj = MagicMock() mock_file_handler = MagicMock() mock_file_handler.write.return_value = None smart_open_obj.__enter__.return_value = mock_file_handler smart_open_obj.__exit__.return_value = None mock_open.smart_open.return_value = smart_open_obj # call function under test. self.proper._write_records_to_csv( 's3://abc', sample_row, test_map) mock_file_handler.write.assert_any_call(transformed_row) @patch('feed_sender.flows.proper_new_releases_tracks.proper.sentry') @patch('feed_sender.flows.proper_new_releases_tracks.proper.smart_open') def test_write_releases_to_csv_exception(self, mock_open, mock_sentry): """Test if transformed row is written via smart open.""" def raising(output, **kwargs): raise exceptions.TransformException('oops') self.proper._release_ids = [1, 2] self.proper.total_releases = 1 test_map = [{ 'release_id': 'release_id', 'proper': 'Org', 'orchard': 'org', 'transform': raising}] sample_row = [{'release_id': 1, 'org': 'Maria'}] # mock smart_open and file handles. smart_open_obj = MagicMock() mock_file_handler = MagicMock() mock_file_handler.write.return_value = None smart_open_obj.__enter__.return_value = mock_file_handler smart_open_obj.__exit__.return_value = None mock_open.smart_open.return_value = smart_open_obj chained = call.get_client().captureMessage('oops') call_list = chained.call_list() self.proper._write_records_to_csv( 's3://abc', sample_row, test_map) mock_file_handler.write.assert_any_call('Org\r\n') assert mock_sentry.mock_calls == call_list @patch('feed_sender.flows.proper_new_releases_tracks.proper.' 'Proper._write_records_to_csv') def test_convert_releases_to_csv_success(self, mock_write): """Test if retrieved releases to passed to the write function. Assert if row is converted into correct output format. """ sample_row = [{'code': '123'}, {'code': '456'}] self.proper.total_releases = 2 self.proper._product_data_query = MagicMock(return_value=sample_row) self.proper.convert_releases_to_csv('s3://abc', '1.a') mock_write.assert_any_call( 's3://abc/1.a', sample_row, transform_settings.NEW_RELEASES_MAP, False) def test_convert_releases_to_csv_fail(self): """Test if retrieved releases files if num records mismatch.""" sample_row = [{'code': '123'}, {'code': '456'}] self.proper.total_releases = 1 self.proper._product_data_query = MagicMock(return_value=sample_row) # call function under test. with self.assertRaises(Exception) as context: self.proper.convert_releases_to_csv('s3://abc', '1.a') self.assertTrue( 'Bad mapping' in str(context.exception)) @patch('feed_sender.flows.proper_new_releases_tracks.proper.smart_open') def test_convert_tracks_to_csv(self, mock_open): """Test if retrieved track row is converted. Assert if row is converted into correct output format. """ sample_row = [ { 'product_code': 'prod123', 'disc': 3, 'side': 1, 'track_id': 1343, 'track_name': 'Holy Swimsuit and the Bears', 'length_minutes': 4, 'length_seconds': 20, 'isrc': 89384342, 'performer': ' Anna Banana ' } ] transformed_row = ( '"prod123","3","1","1343","Holy Swimsuit and the Bears",' '"04:20","89384342","Anna Banana"\r\n') # noqa self.proper._track_data_query = MagicMock(return_value=sample_row) # mock smart_open and file handles. smart_open_obj = MagicMock() mock_file_handler = MagicMock() mock_file_handler.write.return_value = None smart_open_obj.__enter__.return_value = mock_file_handler smart_open_obj.__exit__.return_value = None mock_open.smart_open.return_value = smart_open_obj self.proper.convert_tracks_to_csv( 's3://abc', 'release_ids.txt', 'tracks.csv') mock_file_handler.write.assert_any_call(transformed_row) def test_convert(self): """Test for proper conversion for CSV formatting.""" num_input = 3 assert proper._convert(num_input) == '"3"' comma_input = 'abc , def' assert proper._convert(comma_input) == '"abc , def"' quote_input = 'abc "' assert proper._convert(quote_input) == '"abc """' empty = None assert proper._convert(empty) == '""' def test_write_releases_header_row(self): """Test writing string to an IO stream. Assert that we can write out a comma-delimited string of Proper column names to an IO stream for the New Releases feed. """ output = io.StringIO() self.proper._write_header_row(output, transform_settings.NEW_RELEASES_MAP) assert output.getvalue() == ( 'LabelCode,LabelName,Sublabel,Barcode,CatalogueNumber,Artist,' 'Title,FullArtist,FullTitle,ReleaseDate,PrimaryGenre,' 'SecondaryGenre,Format,Formid,Units,Packageid,Weight,Length,' 'Width,Height,ExplicitLyrics,Narrative,DealerPrice,BuyPrice,' 'ArtworkPath,EmbargoDate,ExportOnly,ExclusiveFor\r\n') def test_write_tracks_header_row(self): """Test writing string to an IO stream. Assert that we can write out a comma-delimited string of Proper column names to an IO stream for the New Tracks feed. """ output = io.StringIO() self.proper._write_header_row(output, transform_settings.NEW_TRACKS_MAP) assert output.getvalue() == ( 'CatalogueNumber,DiskNo,SideNo,TrackNo,TrackName,Duration,ISRC,' 'Contributor\r\n') @patch('feed_sender.flows.proper_new_releases_tracks.proper.smart_open') def test_save_release_ids_to_s3(self, mock_open): """Test if _release_ids are written to file in s3.""" self.proper._release_ids = [1, 2] self.proper.total_releases = 1 # mock smart_open and file handles. smart_open_obj = MagicMock() mock_file_handler = MagicMock() mock_file_handler.write.return_value = None smart_open_obj.__enter__.return_value = mock_file_handler smart_open_obj.__exit__.return_value = None mock_open.smart_open.return_value = smart_open_obj # call function under test. self.proper.save_release_ids_to_s3('s3://abc', 'file.csv') release_id_to_write = '1\n' mock_file_handler.write.assert_any_call(release_id_to_write) @freeze_time('2012-06-30 15:30:00') @patch('feed_sender.flows.proper_new_releases_tracks.proper.mysql') @patch('feed_sender.flows.proper_new_releases_tracks.proper.s3') def test_update_delivery_history_success(self, mock_s3, mock_mysql): """Assert successful delivery history update executes proper SQL.""" input_from_s3 = b'1\n2' mock_s3.read_s3_object = MagicMock(return_value=input_from_s3) mock_mysql.execute_write_query = MagicMock(return_value=2) # call function under test. self.proper.update_delivery_history('s3://abc', 'file.txt') update_feed_delivery_history_sql = ( "INSERT INTO test_pd.product_physical_feed_delivery_" # noqa "history(product_id, supply_chain, delivery_type, " # noqa "date_delivered) " # noqa "VALUES (1, 'PROPER', 'NEW', '2012-06-30T15:30:00')," # noqa "(2, 'PROPER', 'NEW', '2012-06-30T15:30:00')") # noqa update_change_table_sql = ( "UPDATE test_ar.product_physical_change_history " # noqa "SET delivered = 'Y' WHERE date_changed < '{cutoff}' " # noqa "AND product_id IN (1,2)").format( # noqa cutoff=self.proper.cutoff) mock_mysql.execute_write_query.assert_has_calls([ call(self.proper.db_connection, update_feed_delivery_history_sql), call(self.proper.db_connection, update_change_table_sql) ]) @patch('feed_sender.flows.proper_new_releases_tracks.proper.mysql') @patch('feed_sender.flows.proper_new_releases_tracks.proper.s3') def test_update_delivery_history_rollback_feed_delivery_update( self, mock_s3, mock_mysql): """Test the failure to update pp_feed_delivery_history table. Assert that it causes update_delivery_history to rollback all updates. """ input_from_s3 = b'1\n2' mock_s3.read_s3_object = MagicMock(return_value=input_from_s3) self.proper._update_feed_delivery_history_table = MagicMock( return_value=False) # call function under test. with self.assertRaises(Exception) as context: self.proper.update_delivery_history('s3://abc', 'file.txt') self.assertTrue( 'Failed to update delivery history.' in str(context.exception)) @patch('feed_sender.flows.proper_new_releases_tracks.proper.mysql') @patch('feed_sender.flows.proper_new_releases_tracks.proper.s3') def test_update_delivery_history_rollback_change_history_update( self, mock_s3, mock_mysql): """Test the failure to update product_physical_change_history table. Assert that it causes update_delivery_history to rollback all updates. """ input_from_s3 = b'1\n2' mock_s3.read_s3_object = MagicMock(return_value=input_from_s3) self.proper._update_delivered_flag_on_changes_table = MagicMock( return_value=False) # call function under test. with self.assertRaises(Exception) as context: self.proper.update_delivery_history('s3://abc', 'file.txt') self.assertTrue( 'Failed to update delivery history.' in str(context.exception)) @patch( 'feed_sender.flows.proper_new_releases_tracks.proper.' 'correlation_id.get_correlation_id') @patch( 'feed_sender.flows.proper_new_releases_tracks.proper.' 'Proper.fetch_genre_mapping') def test_inject_dependencies(self, mock_mapping, mock_get_corr_id): """Test _inject_dependencies.""" mock_genre_subgenre_mapping = { '1_4': {'subgenre': 'Sub Rock', 'genre': 'Rock'} } mock_mapping.return_value = mock_genre_subgenre_mapping mock_get_corr_id.return_value = 'cool_id' mock_record = {'test_field': 'bla bla bla'} actual = self.proper._inject_dependencies(mock_record) assert actual == { 'test_field': 'bla bla bla', 'carveout_service_name': 'ows-carveouts', 'genre_subgenre_mapping': { '1_4': {'genre': 'Rock', 'subgenre': 'Sub Rock'} }, 'application_env': settings.ENV, 'application_name': 'proper_new_releases_tracks', 'carveout_service_resource_path': '/carveout/{upc}/territory', 'correlation_id': 'cool_id' } @patch('feed_sender.flows.proper_new_releases_tracks.proper.mysql') @patch('feed_sender.flows.proper_new_releases_tracks.proper.s3') def test_track_data_query_success(self, mock_s3, mock_mysql): """Test if DB call for track data retrieval.""" input_from_s3 = b'1\n2' mock_s3.read_s3_object = MagicMock(return_value=input_from_s3) mock_results = [ {'isrc': 'QM4TW1612900', 'product_code': 'test_1234', 'track_id': 1, 'length_minutes': 11, 'length_seconds': 57}, {'isrc': '', 'product_code': 'test_1297', 'track_id': 1, 'length_minutes': 0, 'length_seconds': 0}] mock_mysql.execute_query = MagicMock(return_value=mock_results) # call function under test. actual = self.proper._track_data_query('s3://abc', 'file.txt') expected_sql = ( 'SELECT ' 'rv.product_code, ' 't.cd as disc, ' '1 as side, ' 't.track_id, ' 't.track_name, ' 't.length_minute as length_minutes, ' 't.length_seconds, ' 't.isrc, ' 'ta.name as performer, ' 'ta.type ' 'FROM ' ' (SELECT r.release_id, r.product_code ' ' FROM test_ar.releases r ' ' WHERE r.release_id in (1,2)) AS rv ' 'JOIN test_ar.track t ' ' ON t.release_id = rv.release_id ' 'LEFT JOIN test_ar.track_artist ta ' ' ON t.id = ta.track_id ' "WHERE ta.type = 'performer'") mock_mysql.execute_query.assert_any_call( None, expected_sql) assert actual == mock_results