"""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 from feed_sender.flows.proper_changed_releases import proper 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._release_ids = None self.proper.total_releases = 0 @patch( 'feed_sender.flows.proper_changed_releases.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_changed_releases.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 self.proper.total_releases == 0 assert not self.proper._release_ids @patch('feed_sender.flows.proper_changed_releases.proper.mysql') def test_changed_products_query(self, mock_mysql): """Check if correct changed products SQL is used.""" mock_results = [ {'product_id': 1, 'column': 'test1'}, {'product_id': 2, 'column': 'test2'}] mock_mysql.execute_query = MagicMock(return_value=mock_results) actual = self.proper.changed_products_query() assert mock_mysql.execute_query.called is True assert actual == mock_results @patch('feed_sender.flows.proper_changed_releases.proper.mysql') def test_changed_product_data_query(self, mock_mysql): """Check if correct changed products SQL is used.""" mock_results = [ {'product_id': 1, 'column': 'test1'}, {'product_id': 2, 'column': 'test2'}] self.proper.total_releases = 2 self.proper._release_ids = [1, 2] mock_mysql.execute_query = MagicMock(return_value=mock_results) actual = self.proper._changed_product_data_query() assert mock_mysql.execute_query.called is True assert actual == mock_results @patch('feed_sender.flows.proper_changed_releases.proper.mysql') def test_changed_product_data_query_no_release_ids(self, mock_mysql): """Check if correct changed products SQL is used.""" mock_results = {} self.proper.total_releases = 0 mock_mysql.execute_query = MagicMock(return_value=mock_results) actual = self.proper._changed_product_data_query() assert not mock_mysql.execute_query.called assert actual == mock_results def test_squash_records_all_fields_changed(self): """Test if changes to the same product id are squashed.""" # Each change is in one row input_rows = [{ 'product_id': 'prod_1', 'field_name': 'wholesale_price', 'new_sale_start_date': None, 'new_price': '19.99', 'new_deletion_status': None}, {'product_id': 'prod_1', 'field_name': 'sale_start_date', 'new_sale_start_date': '2000-08-01', 'new_price': None, 'new_deletion_status': None}, {'product_id': 'prod_1', 'field_name': 'deletions', 'new_sale_start_date': None, 'new_price': None, 'new_deletion_status': 'Y'}, {'product_id': 'prod_1', 'field_name': 'artwork', 'artworkpath': 'https://artwork.com'}] expected_rows = [{ 'product_id': 'prod_1', 'field_name': 'wholesale_price', 'new_sale_start_date': '2000-08-01', 'new_price': '19.99', 'new_deletion_status': 'Y', 'artworkpath': 'https://artwork.com'}] # test if two lists have the same elements regardless of order self.assertCountEqual( expected_rows, self.proper.squash_records(input_rows)) def test_squash_records_one_field_changed(self): """Test if a changed to a row is returned.""" input_rows = [{ 'product_id': 'prod_2', 'field_name': 'wholesale_price', 'new_sale_start_date': None, 'new_price': '10.00', 'new_deletion_status': None}] expected_rows = [{ 'product_id': 'prod_2', 'field_name': 'wholesale_price', 'new_sale_start_date': None, 'new_deletion_status': None, 'new_price': '10.00'}] # test if two lists have the same elements regardless of order self.assertCountEqual( expected_rows, self.proper.squash_records(input_rows)) def test_squash_records_orig_equal_to_changed(self): """Test that an changed value is returned. (Even if changed value is equal to the orig value). """ input_rows = [{ 'product_id': 'prod_3', 'field_name': 'wholesale_price', 'new_sale_start_date': None, 'new_deletion_status': None, 'new_price': '10.00'}, {'product_id': 'prod_3', 'field_name': 'wholesale_price', 'new_sale_start_date': None, 'new_deletion_status': None, 'new_price': '122.00'}, {'product_id': 'prod_3', 'field_name': 'wholesale_price', 'new_sale_start_date': None, 'new_deletion_status': None, 'new_price': '10.00'}] expected_rows = [{ 'product_id': 'prod_3', 'field_name': 'wholesale_price', 'new_sale_start_date': None, 'new_price': '10.00', 'new_deletion_status': None}] # test if two lists have the same elements regardless of order self.assertCountEqual( expected_rows, self.proper.squash_records(input_rows)) def test_squash_records_new_value_is_empty(self): """Test that an changed value is returned. (Even if changed value is empty. """ input_rows = [{ 'product_id': 'prod_4', 'field_name': 'wholesale_price', 'new_sale_start_date': None, 'new_deletion_status': None, 'new_price': '10.00'}, {'product_id': 'prod_4', 'field_name': 'wholesale_price', 'new_sale_start_date': None, 'new_deletion_status': None, 'new_price': None}, {'product_id': 'prod_4', 'field_name': 'deletions', 'new_sale_start_date': None, 'new_deletion_status': 'Y', 'new_price': None}] expected_rows = [{ 'product_id': 'prod_4', 'field_name': 'wholesale_price', 'new_sale_start_date': None, 'new_price': None, 'new_deletion_status': 'Y'}] # test if two lists have the same elements regardless of order self.assertCountEqual( expected_rows, self.proper.squash_records(input_rows)) @patch('feed_sender.flows.proper_changed_releases.proper.smart_open') def test_save_release_ids_to_s3(self, mock_open): """Test if _release_ids are written to file in S3.""" # 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. sample_row = [{ 'product_id': 123, 'product_code': 'prod123', 'new_sale_start_date': '2000-07-01', 'new_price': '19.99', 'new_deletion_status': 'Y' }] self.proper.save_release_ids_to_s3(sample_row, 's3://abc', 'file.csv') release_id_to_write = '123\n' mock_file_handler.write.assert_any_call(release_id_to_write) def test_write_header_row(self): """Test that we can write out a comma-delimited string to an IO stream. Assert we can write out Proper column names. """ output = io.StringIO() self.proper._write_header_row(output) assert output.getvalue() == ( 'CatalogueNumber,Format,Artist,Title,Barcode,DateReleased,Price1,' 'Price2,Price3,Price4,Price5,LocalPrice,LabelCode,Weight,' 'Charfield1,Charfield2,Charfield3,Supplier,Charfield4,' 'DeletionType,decimalField1,PrimaryGenre,SecondaryGenre,' 'Embargo,ArtworkPath\r\n') @patch('feed_sender.flows.proper_changed_releases.proper.smart_open') def test_convert_releases_to_csv(self, mock_open): """Test if retrieved row is written out to file.""" sample_row = [{ 'product_code': 'prod123', 'new_sale_start_date': '2000-07-01', 'new_price': '19.99', 'new_deletion_status': 'Y' }] transformed_row = ( '"prod123","","","","","2000-07-01","19.99","","","","","","","",' '"","","","","","D3","","","","",""\r\n') # mock smart_open and file handles. mock_file_handler = MagicMock() setattr(mock_file_handler, 'write', MagicMock(return_value=None)) smart_open_obj = MagicMock() setattr(smart_open_obj, '__enter__', MagicMock( return_value=mock_file_handler)) setattr(smart_open_obj, '__exit__', MagicMock(return_value=None)) mock_open.smart_open = MagicMock(return_value=smart_open_obj) self.proper.convert_releases_to_csv(sample_row, 's3://abc', 'file.csv') mock_file_handler.write.assert_any_call(transformed_row) @patch('feed_sender.flows.proper_changed_releases.proper.smart_open') def testconvert_releases_to_csv_alphanumeric_price_code_(self, mock_open): """Test if retrieved row is written out to file.""" sample_row = [{ 'product_code': 'prod123', 'new_sale_start_date': '2000-07-01', 'new_price': 'X1', 'new_deletion_status': 'Y' }] transformed_row = ( '"prod123","","","","","2000-07-01","X1","","","","","","","",' '"","","","","","D3","","","","",""\r\n') # mock smart_open and file handles. mock_file_handler = MagicMock() setattr(mock_file_handler, 'write', MagicMock(return_value=None)) smart_open_obj = MagicMock() setattr(smart_open_obj, '__enter__', MagicMock( return_value=mock_file_handler)) setattr(smart_open_obj, '__exit__', MagicMock(return_value=None)) mock_open.smart_open = MagicMock(return_value=smart_open_obj) self.proper.convert_releases_to_csv(sample_row, 's3://abc', 'file.csv') mock_file_handler.write.assert_any_call(transformed_row) @freeze_time('2012-06-30 15:30:00') @patch('feed_sender.flows.proper_changed_releases.proper.mysql') @patch('feed_sender.flows.proper_changed_releases.proper.s3') def test_update_delivery_history_success(self, mock_s3, mock_mysql): """Test that 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', 'CHANGE', '2012-06-30T15:30:00')," # noqa "(2, 'PROPER', 'CHANGE', '2012-06-30T15:30:00')") # noqa update_change_table_sql = (""" UPDATE test_ar.product_physical_change_history SET delivered = 'Y' WHERE date_changed < '{cutoff}' AND ((store_id = '696' AND field_name = 'wholesale_price') OR store_id IS NULL) AND product_id IN (1,2)""").format( 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_changed_releases.proper.mysql') @patch('feed_sender.flows.proper_changed_releases.proper.s3') def test_update_delivery_history_rollback_feed_delivery_update( self, mock_s3, mock_mysql): """Test 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_changed_releases.proper.mysql') @patch('feed_sender.flows.proper_changed_releases.proper.s3') def test_update_delivery_history_rollback_change_history_update( self, mock_s3, mock_mysql): """Test 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))