"""Test handler.""" from unittest.mock import call, MagicMock, patch from constants import queries from ddex_ingester_common.constants.ddex_providers import \ SME_ANALYTICS_PROVIDER from ddex_ingester_common.lambda_exceptions import SetProductException from ddex_ingester_common.release_correction.release_correction_constants import ( # noqa RELEASE_CORRECTION_PRODUCT_FIELDS) from ddex_ingester_common.release_correction.release_correction_diffs import ( ReleaseCorrectionDiffDetail) from ddex_ingester_common.schemas.s3_schema import S3Schema from ddex_ingester_common.schemas.state_machine_schema import ( StateMachineSchema) import index from lambdacommon.graphql import graphql from marshmallow.utils import get_value import pytest import release_correction import utils @patch('index.graphql_gateway.execute') @patch('index.config.catalog_ingestion_session') def test_create_product_payload( mock_catalog_ingestion_session, mock_graphql_execute, context, s3_ddex ): """Test that create product calls graphql with correct payload.""" product_id = 123 create_product_payload = [ 'productName', 'productCode', 'productHighlights', 'projectId', 'accountId', 'subaccountId', 'upc', 'metaLanguage', 'deliveredVersion', 'participations', 'pLine', 'cLine', 'genreId', 'subgenreId', 'specialInstructions', 'productLocalizations', 'format', 'imprint', 'notForDistribution', 'version', 'vendorReleaseIdentifier', 'manufacturerUpc', ] set_product_pricing_payload = { 'data': { 'productId': product_id, 'orchardPricingTier': 15, 'pricingFamily': 'MUSIC_ALBUM' } } set_track_pricing_payload = { 'data': { 'productId': product_id, 'orchardPricingTier': 25, 'pricingFamily': 'MUSIC_TRACK' } } mock_graphql_execute.reset_mock() mock_graphql_execute.return_value = { 'data': {'createProduct': {'productId': product_id}} } context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) index.create_product(context, s3_ddex_data) mock_graphql_execute.assert_called() call_dict = mock_graphql_execute.mock_calls[0][1][1]['data'] call_keys = list(call_dict) assert create_product_payload == call_keys call_dict = mock_graphql_execute.mock_calls[1][1][1] assert set_product_pricing_payload == call_dict call_dict = mock_graphql_execute.mock_calls[2][1][1] assert set_track_pricing_payload == call_dict mock_catalog_ingestion_session.save.assert_called() mock_catalog_ingestion_session.add.assert_called() add_call_object = mock_catalog_ingestion_session.add.call_args[0][0] assert add_call_object.action == 'insert' assert add_call_object.result == 'Success' @patch('index.graphql_gateway.execute') @patch('index.config.catalog_ingestion_session') @patch('index.update_display_upc') @patch('index.store_placeholder_upc') def test_create_product_with_placeholder_upc( store_placeholder_upc, mock_update_display_upc, mock_catalog_ingestion_session, mock_graphql_execute, context, s3_ddex ): """Test that create product updates display_upc.""" product_id = 123 display_upc = '7777777' upc = '999' mock_graphql_execute.reset_mock() mock_graphql_execute.return_value = { 'data': {'createProduct': {'productId': product_id, 'upc': upc}} } context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) context.ddex_provider = SME_ANALYTICS_PROVIDER context.product.display_upc = display_upc context.product.upc = None index.create_product(context, s3_ddex_data) mock_graphql_execute.assert_called() mock_update_display_upc.assert_called_with(product_id, display_upc) store_placeholder_upc.assert_called_with(context, display_upc, upc) @patch('index.graphql_gateway.execute') def test_create_product_payload_fails_on_format( mock_graphql_execute, context, s3_ddex ): """Test that create product calls graphql with correct payload.""" mock_graphql_execute.reset_mock() context = StateMachineSchema().load(context) context.product.release_type = 'Broken' s3_ddex_data = S3Schema().load(s3_ddex) with pytest.raises(ValueError): index.create_product(context, s3_ddex_data) @patch('index.graphql_gateway.execute') @patch('index.config.catalog_ingestion_session') def test_update_product_payload( mock_catalog_ingestion_session, mock_graphql_execute, context, s3_ddex, get_product_response ): """Test that update product calls graphql with correct payload.""" payload = [ 'productName', 'productCode', 'productId', 'metaLanguage', 'deliveredVersion', 'participations', 'pLine', 'genreId', 'subgenreId', 'specialInstructions', 'productLocalizations', 'format', 'imprint', 'notForDistribution', 'vendorReleaseIdentifier', 'manufacturerUpc', 'cLine', ] mock_graphql_execute.reset_mock() context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] index.update_product(context, s3_ddex_data) call_dict = mock_graphql_execute.mock_calls[0][1][1]['data'] call_keys = list(call_dict) mock_graphql_execute.assert_called_once() assert payload == call_keys mock_catalog_ingestion_session.save.assert_called() mock_catalog_ingestion_session.add.assert_called() add_call_object = mock_catalog_ingestion_session.add.call_args[0][0] assert add_call_object.action == 'update' assert add_call_object.result == 'Success' @patch('index.load_rc_json') @patch('index.check_for_product') @patch('index.load_ddex_json') def test_create_product_if_none_exists( mock_load_ddex_json, mock_check_for_product, mock_rc_json, context, s3_ddex, get_product_no_result, create_product_success, s3_release_corrections): """Test that a new product will be created if none found.""" mock_rc_json.return_value = s3_release_corrections graphql_result = create_product_success['data']['createProduct'] no_product = get_product_no_result['data']['productByUpc'] index.create_product = MagicMock() index.create_product.return_value = graphql_result mock_check_for_product.return_value = no_product mock_load_ddex_json.return_value = s3_ddex index.handler(context, None) index.create_product.assert_called_once() @patch('index.load_rc_json') @patch('index.check_for_product') @patch('index.load_ddex_json') def test_create_product_if_duplicate_product_code( mock_load_ddex_json, mock_check_for_product, mock_rc_json, context, s3_ddex, get_product_no_result, s3_release_corrections): """Test that create_product will be called again if duplicate product code found.""" mock_rc_json.return_value = s3_release_corrections graphql_result = [{'extensions': {'response': {'body': {'message': {'product_code': {'error_code': 'product_code_used'}}}}}}] # noqa no_product = get_product_no_result['data']['productByUpc'] index.create_product = MagicMock() index.create_product.return_value = graphql_result mock_check_for_product.return_value = no_product mock_load_ddex_json.return_value = s3_ddex index.handler(context, None) index.create_product.call_count == 2 @patch('index.update_product', wraps=index.update_product) @patch('release_correction.write_rc_json') @patch('index.load_rc_json') @patch('index.graphql_gateway.execute') @patch('index.load_ddex_json') def test_update_product_if_found( mock_load_ddex_json, mock_graphql_execute, mock_rc_json, mock_write_rc_json, mock_update_product, context, s3_ddex, get_product_response, update_product_success, s3_release_corrections): """Test that an existing product will be updated if found.""" mock_rc_json.return_value = s3_release_corrections graphql_result = update_product_success['data']['updateProduct'] index.update_product.return_value = graphql_result mock_load_ddex_json.return_value = s3_ddex mock_graphql_execute.return_value = get_product_response response = index.handler(context, None) mock_update_product.assert_called_once() assert len(mock_update_product.call_args) == 2 assert response.get('product').get('product_id') @patch('utils.is_product_role', return_value=True) def test_get_participations( mock_is_product_role, context, s3_ddex ): """Test that participations are returned correctly.""" context = StateMachineSchema().load(context) s3_context = S3Schema().load(s3_ddex) expected_artists = [ { 'labelParticipantUuid': 'AAA', 'role': 'PRIMARY_ARTIST', 'artist_name': 'Digital Farm Animals', 'localizations': [{ 'name': 'Russian Farm Animals', 'languageId': 35 }] }, { 'labelParticipantUuid': 'AAA', 'role': 'PRODUCER', 'artist_name': 'Digital Farm Animals', 'localizations': [{ 'name': 'Russian Farm Animals', 'languageId': 35 }] }, { 'labelParticipantUuid': 'BBB', 'role': 'PRIMARY_ARTIST', 'artist_name': 'Cash Cash', 'localizations': [{ 'name': 'Russian Cash Cash', 'languageId': 35 }] }, { 'labelParticipantUuid': 'CCC', 'role': 'FEATURED_ARTIST', 'artist_name': 'Nelly', 'localizations': [{ 'name': 'Russian Nelly', 'languageId': 35 }] } ] artists = utils.get_participations( s3_context.product.display_artists, s3_context.label_participants, [], context, True ) assert artists == expected_artists @patch('utils.is_product_role', return_value=False) def test_get_participations_no_results( mock_is_product_role, context, invalid_display_artists ): """Test that participations are not returned for invalid values.""" context = StateMachineSchema().load(context) s3_context = S3Schema().load(invalid_display_artists) expected_artists = [] artists = utils.get_participations( s3_context.product.display_artists, [], [], context ) assert artists == expected_artists def test_get_participations_no_display_artists( context, invalid_display_artists ): """Test that participations are not returned for invalid values.""" expected_artists = [] artists = utils.get_participations([], [], [], []) assert artists == expected_artists def test_get_title_localizations(s3_ddex): """Test get_title_localizations.""" s3_data = S3Schema().load(s3_ddex) expected_result = [ { 'languageId': 15, 'productName': 'French Title', 'deliveredVersion': 'Original' }, { 'languageId': 35, 'productName': 'Russian Title', 'deliveredVersion': 'Remaster' } ] result = utils.get_title_localizations(s3_data.product) assert result == expected_result def test_get_title_localizations_null_localized_titles(s3_ddex): """Test get_title_localizations with null localized titles.""" s3_data = S3Schema().load(s3_ddex) s3_data.product.localized_titles = None result = utils.get_title_localizations(s3_data.product) assert result == [] def test_get_title_localizations_all_null(s3_ddex): """Test get_title_localizations null localized titles and artists.""" s3_data = S3Schema().load(s3_ddex) s3_data.product.localized_titles = None s3_data.product.display_artists = None result = utils.get_title_localizations(s3_data.product) assert result == [] def test_is_product_role_true(s3_ddex): """Test is_product_role.""" parsed_ddex = S3Schema().load(s3_ddex) parsed_ddex.product.genres[0].genre = 'CLASSICAL-00' genres = parsed_ddex.product.genres result = utils.is_product_role('Composer', genres) assert result def test_role_is_product_role_false(s3_ddex): """Test is_product_role.""" parsed_ddex = S3Schema().load(s3_ddex) genres = parsed_ddex.product.genres result = utils.is_product_role('Composer', genres) assert not result def test_role_is_product_role_unknown(s3_ddex): """Test is_product_role.""" parsed_ddex = S3Schema().load(s3_ddex) genres = parsed_ddex.product.genres result = utils.is_product_role('unknown', genres) assert not result def test_role_is_product_role_skip(s3_ddex): """Test is_product_role.""" result = utils.is_product_role('MusicDirector', [], skip_genre_check=True) assert result @patch('release_correction.write_rc_json') @patch('index.load_rc_json') @patch('index.graphql_gateway.execute') @patch('index.load_ddex_json') def test_diff_for_release_corrections_raises_exception( mock_load_ddex_json, mock_graphql_execute, mock_rc_json, mock_write_rc, context, s3_ddex, get_product_response, s3_release_corrections): """Test that diff_for_release_corrections raises a ValueError.""" state_machine_context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] get_product_response['productCode'] = 'breaks' mock_rc_json.return_value = s3_release_corrections mock_load_ddex_json.return_value = s3_ddex mock_graphql_execute.return_value = get_product_response with pytest.raises(release_correction.ReleaseCorrectionUpdateException): release_correction.diff_for_release_corrections( state_machine_context, s3_ddex_data, get_product_response, s3_release_corrections) mock_write_rc.assert_called_once() @patch('release_correction.write_rc_json') @patch('index.load_rc_json') @patch('index.graphql_gateway.execute') @patch('index.load_ddex_json') def test_diff_for_release_corrections_no_diff( mock_load_ddex_json, mock_graphql_execute, mock_rc_json, mock_write_rc, context, s3_ddex, get_product_response, s3_release_corrections): """Test that diff_for_release_corrections will write an RC detail.""" state_machine_context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] mock_rc_json.return_value = s3_release_corrections mock_load_ddex_json.return_value = s3_ddex mock_graphql_execute.return_value = get_product_response release_correction.diff_for_release_corrections( state_machine_context, s3_ddex_data, get_product_response, s3_release_corrections) mock_write_rc.assert_not_called() @patch('release_correction.write_rc_json') @patch('index.load_rc_json') @patch('index.graphql_gateway.execute') @patch('index.load_ddex_json') def test_diff_for_release_corrections_creates_diff( mock_load_ddex_json, mock_graphql_execute, mock_rc_json, mock_write_rc, context, s3_ddex, get_product_response, s3_release_corrections): """Test that diff_for_release_corrections will write an RC detail.""" state_machine_context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] get_product_response['cLine'] = 'Old cLine' get_product_response['deliveredVersion'] = 'old version' get_product_response['imprint'] = 'old imprint' get_product_response['subgenreId'] = 300 mock_rc_json.return_value = s3_release_corrections mock_load_ddex_json.return_value = s3_ddex mock_graphql_execute.return_value = get_product_response event_details = { 'bucket': state_machine_context.bucket, 'key': state_machine_context.key } expected_changes = {'changes': [ ReleaseCorrectionDiffDetail( field_name='deliveredVersion', old='old version', new='product version', db_table_name='releases', db_field_name='delivered_version', email_customer=True, accept_update=True), ReleaseCorrectionDiffDetail( field_name='subgenreId', old=300, new=[621], db_table_name='releases', db_field_name='release_subgenre', email_customer=True, accept_update=True), ReleaseCorrectionDiffDetail( field_name='cLine', old='Old cLine', new='Product Cline 2020', db_table_name='releases', db_field_name='c_line', email_customer=False, accept_update=True), ReleaseCorrectionDiffDetail( field_name='imprint', old='old imprint', new='product imprint', db_table_name='releases', db_field_name='label', email_customer=False, accept_update=True)]} release_correction.diff_for_release_corrections( state_machine_context, s3_ddex_data, get_product_response, s3_release_corrections) mock_write_rc.assert_called_with(event_details, expected_changes) def test_create_release_correction_detail(): """Test that create_release_correction_detail returns DiffDetail.""" field_name = 'format' old_value = 'old val' new_value = 'new val' rc_product_field = RELEASE_CORRECTION_PRODUCT_FIELDS[field_name] expected_return = ReleaseCorrectionDiffDetail( field_name, None, old_value, new_value, rc_product_field.db_table_name, rc_product_field.db_field_name, rc_product_field.email_customer, rc_product_field.accept_update) response = release_correction.create_release_correction_detail( field_name, old_value, new_value) assert response == expected_return @patch('release_correction.write_rc_json') @patch('index.load_rc_json') @patch('index.check_for_product') @patch('index.load_ddex_json') def test_update_product_not_called_for_swbdummy( mock_load_ddex_json, mock_graphql_execute, mock_rc_json, mock_write_rc_json, context, s3_ddex, get_product_response, update_product_success, s3_release_corrections): """Check update_product not called for NFD=N for in_content product.""" mock_rc_json.return_value = s3_release_corrections index.update_product = MagicMock() graphql_result = update_product_success['data']['updateProduct'] index.update_product.return_value = graphql_result mock_load_ddex_json.return_value = s3_ddex mock_graphql_execute.return_value =\ get_product_response['data']['productByUpc'] context['product']['not_for_distribution'] = 'N' index.handler(context, None) index.update_product.assert_not_called() @patch('release_correction.write_rc_json') @patch('index.load_rc_json') @patch('index.check_for_product') @patch('index.load_ddex_json') def test_update_product_called_for_swbdummy( mock_load_ddex_json, mock_graphql_execute, mock_rc_json, mock_write_rc_json, context, s3_ddex, get_product_response, update_product_success, s3_release_corrections): """Check update_product called for NFD!=N for in_content product.""" mock_rc_json.return_value = s3_release_corrections index.update_product = MagicMock() graphql_result = update_product_success['data']['updateProduct'] index.update_product.return_value = graphql_result mock_load_ddex_json.return_value = s3_ddex get_product_response = get_product_response['data']['productByUpc'] get_product_response['status'] = 'in_progress' mock_graphql_execute.return_value = get_product_response index.handler(context, None) index.update_product.assert_called() @patch('index.update_product_nfd') @patch('release_correction.write_rc_json') @patch('index.load_rc_json') @patch('index.check_for_product') @patch('index.load_ddex_json') def test_update_nfd_called_for_in_content( mock_load_ddex_json, mock_graphql_execute, mock_rc_json, mock_write_rc_json, mock_update_nfd, context, s3_ddex, get_product_response, update_product_success, s3_release_corrections): """Check update_product not called for NFD=N for in_content product.""" mock_rc_json.return_value = s3_release_corrections mock_load_ddex_json.return_value = s3_ddex index.update_product = MagicMock() context['product']['not_for_distribution'] = 'N' graphql_result = get_product_response['data']['productByUpc'] graphql_result['notForDistribution'] = 'SwitchboardDummy' mock_graphql_execute.return_value = graphql_result index.handler(context, None) index.update_product.assert_not_called() mock_update_nfd.assert_called_once() @patch('utils.is_product_role', return_value=True) def test_compare_list_of_dicts_same_data( mock_is_product_role, context, get_product_response): """Test compare_list_of_dicts compares correctly.""" context = StateMachineSchema().load(context) s3_participations = [ { 'artist_name': 'Digital Farm Animals', 'role': 'performer', }, { 'artist_name': 'Digital Farm Animals', 'role': 'producer', }, { 'artist_name': 'Cash Cash', 'role': 'performer', }, { 'artist_name': 'Nelly', 'role': 'featuring', } ] gql_participations = get_value( get_product_response, 'data.productByUpc.labelParticipations', {}) formatted_gql = release_correction.format_graphql_participations( gql_participations) diff = release_correction.compare_list_of_dicts( s3_participations, formatted_gql ) assert not diff @patch('utils.is_product_role', return_value=True) def test_compare_list_of_dicts_diff_data( mock_is_product_role, context, get_product_response): """Test compare_list_of_dicts compares correctly.""" context = StateMachineSchema().load(context) s3_participations = [ { 'artist_name': 'Digital Farm Animals', 'role': 'performer', }, { 'artist_name': 'Digital Farm Animals', 'role': 'remixer', }, { 'artist_name': 'Cash Cash', 'role': 'performer', }, { 'artist_name': 'Nelly', 'role': 'featuring', } ] expected = [ { 'artist_name': 'Digital Farm Animals', 'role': 'remixer' }, { 'role': 'producer', 'artist_name': 'Digital Farm Animals' } ] gql_participations = get_value( get_product_response, 'data.productByUpc.labelParticipations', {}) formatted_gql = release_correction.format_graphql_participations( gql_participations) diff = release_correction.compare_list_of_dicts( s3_participations, formatted_gql ) assert diff == expected @patch('release_correction.compare_list_of_dicts') @patch('release_correction.write_rc_json') @patch('index.load_rc_json') @patch('index.graphql_gateway.execute') @patch('index.load_ddex_json') def test_diff_for_release_corrections_for_artists( mock_load_ddex_json, mock_graphql_execute, mock_rc_json, mock_write_rc, mock_compare_dicts, context, s3_ddex, get_product_response, s3_release_corrections): """Test that diff_for_release_corrections will write artist RC detail.""" state_machine_context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] mock_rc_json.return_value = s3_release_corrections mock_compare_dicts.return_value = [ { 'artist_name': 'Digital Farm Animals', 'role': 'remixer' }, { 'role': 'producer', 'artist_name': 'Digital Farm Animals', } ] mock_load_ddex_json.return_value = s3_ddex mock_graphql_execute.return_value = get_product_response release_correction.diff_for_release_corrections( state_machine_context, s3_ddex_data, get_product_response, s3_release_corrections) event_details = { 'bucket': state_machine_context.bucket, 'key': state_machine_context.key } expected_changes = {'changes': [ ReleaseCorrectionDiffDetail( field_name='performer', old=None, new=[{'role': 'performer', 'artist_name': 'Digital Farm Animals'}], db_table_name='releases', db_field_name='performer', email_customer=True, accept_update=True), ReleaseCorrectionDiffDetail( field_name='producer', old=None, new=[{'role': 'producer', 'artist_name': 'Digital Farm Animals'}], db_table_name='releases', db_field_name='producer', email_customer=True, accept_update=True), ReleaseCorrectionDiffDetail( field_name='performer', old=None, new=[{'role': 'performer', 'artist_name': 'Cash Cash'}], db_table_name='releases', db_field_name='performer', email_customer=True, accept_update=True), ReleaseCorrectionDiffDetail( field_name='featuring', old=None, new=[{'role': 'featuring', 'artist_name': 'Nelly'}], db_table_name='releases', db_field_name='featuring', email_customer=True, accept_update=True), ReleaseCorrectionDiffDetail( field_name='remixer', old=None, new=[None], db_table_name='releases', db_field_name='remixer', email_customer=True, accept_update=True) ]} mock_write_rc.assert_called_with(event_details, expected_changes) @patch('release_correction.write_rc_json') def test_unsupported_updates_product_code( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test that check_for_unsupported_updates with different product codes.""" state_machine_context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] get_product_response['productCode'] = 'AAAAAAAAAAAA' with pytest.raises(release_correction.ReleaseCorrectionUpdateException): release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) @patch('release_correction.write_rc_json') def test_unsupported_updates_isrc( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test that check_for_unsupported_updates finds different track ISRCs.""" state_machine_context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] get_product_response['tracks'][0]['isrc'] = 'AAAAAAAAAAAA' with pytest.raises(release_correction.ReleaseCorrectionUpdateException): release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) @patch('release_correction.write_rc_json') def test_unsupported_updates_isrc_no_diff( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test that check_for_unsupported_updates finds different track ISRCs.""" state_machine_context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] get_product_response['tracks'][0]['isrc'] = 'QZCDB2000001' release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) @patch('release_correction.write_rc_json') def test_unsupported_updates_isrc_null( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test that check_for_unsupported_updates finds different track ISRCs.""" state_machine_context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) s3_ddex_data.tracks = None get_product_response = get_product_response['data']['productByUpc'] get_product_response['tracks'] = [] release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) @patch('release_correction.write_rc_json') def test_unsupported_updates_track_number( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test check_for_unsupported_updates with different track number.""" state_machine_context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] get_product_response['tracks'][0]['trackNumber'] = 2 with pytest.raises(release_correction.ReleaseCorrectionUpdateException): release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) @patch('release_correction.write_rc_json') def test_unsupported_updates_track_ownership_rights( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test check_for_unsupported_updates with diff track ownership rights.""" state_machine_context = StateMachineSchema().load(context) state_machine_context.product.not_for_distribution = 'N' s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] get_product_response['tracks'][0]['ownershipRights'] = 'exclusive_licensee' with pytest.raises(release_correction.ReleaseCorrectionUpdateException): release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) @patch('release_correction.write_rc_json') def test_unsupported_updates_track_recording_country_code( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test check_for_unsupported_updates with diff recording country code.""" state_machine_context = StateMachineSchema().load(context) state_machine_context.product.not_for_distribution = 'N' s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] get_product_response['tracks'][0]['recordingCountryId'] = 2 with pytest.raises(release_correction.ReleaseCorrectionUpdateException): release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) @patch('release_correction.write_rc_json') def test_unsupported_updates_track_copyright_owner_country( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test check_for_unsupported_updates with diff copyright owner country.""" state_machine_context = StateMachineSchema().load(context) state_machine_context.product.not_for_distribution = 'N' s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] get_product_response['tracks'][0]['originalRightsHolderCountryId'] = 2 with pytest.raises(release_correction.ReleaseCorrectionUpdateException): release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) @patch('release_correction.write_rc_json') def test_unsupported_updates_track_publisher( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test check_for_unsupported_updates with different track publisher.""" state_machine_context = StateMachineSchema().load(context) state_machine_context.product.not_for_distribution = 'N' s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] get_product_response['tracks'][0]['publishing']['publisherNames'].\ append('Publisher 3') release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) assert state_machine_context.warnings[0].get('field') == 'publisherNames' @patch('release_correction.write_rc_json') def test_unsupported_updates_track_us_publishing_obligation( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test check_for_unsupported_updates with dif US Publishing Obligation.""" state_machine_context = StateMachineSchema().load(context) state_machine_context.product.not_for_distribution = 'N' s3_ddex_data = S3Schema().load(s3_ddex) get_product_response = get_product_response['data']['productByUpc'] get_product_response['tracks'][0]['publishing'][ 'usPublishingObligation'] = 'COMPOSITION' with pytest.raises(release_correction.ReleaseCorrectionUpdateException): release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) @patch('release_correction.write_rc_json') def test_unsupported_updates_title_localization( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test check_for_unsupported_updates with diff product localizations.""" state_machine_context = StateMachineSchema().load(context) state_machine_context.product.not_for_distribution = 'N' s3_ddex_data = S3Schema().load(s3_ddex) s3_ddex_data.product.localized_titles[0].language_code =\ 'en' get_product_response = get_product_response['data']['productByUpc'] with pytest.raises(release_correction.ReleaseCorrectionUpdateException): release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) @patch('release_correction.write_rc_json') def test_unsupported_updates_artist_localization( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test check_for_unsupported_updates with diff artist localizations.""" state_machine_context = StateMachineSchema().load(context) state_machine_context.product.not_for_distribution = 'N' s3_ddex_data = S3Schema().load(s3_ddex) s3_ddex_data.product.display_artists[0].localized_names[0].language_code =\ 'en' get_product_response = get_product_response['data']['productByUpc'] with pytest.raises(release_correction.ReleaseCorrectionUpdateException): release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) @patch('release_correction.write_rc_json') def test_unsupported_updates_track_title_localization( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test check_for_unsupported_updates with diff track localizations.""" state_machine_context = StateMachineSchema().load(context) state_machine_context.product.not_for_distribution = 'N' s3_ddex_data = S3Schema().load(s3_ddex) s3_ddex_data.tracks[0].localized_titles[0].language_code =\ 'en' get_product_response = get_product_response['data']['productByUpc'] with pytest.raises(release_correction.ReleaseCorrectionUpdateException): release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) @patch('release_correction.write_rc_json') def test_unsupported_updates_track_artist_localization( mock_write_rc_json, context, s3_ddex, s3_release_corrections, get_product_response): """Test check_for_unsupported_updates with track artist localizations.""" state_machine_context = StateMachineSchema().load(context) state_machine_context.product.not_for_distribution = 'N' s3_ddex_data = S3Schema().load(s3_ddex) s3_ddex_data.tracks[0].display_artists[0].localized_names[0].\ language_code = 'en' get_product_response = get_product_response['data']['productByUpc'] with pytest.raises(release_correction.ReleaseCorrectionUpdateException): release_correction.check_for_unsupported_updates( state_machine_context, s3_ddex_data, get_product_response, [], s3_release_corrections, {}) @patch('index.graphql_gateway.execute') def test_update_product_nfd( mock_graphql_execute, context): """Test update_product_nfd method.""" state_machine_context = StateMachineSchema().load(context) state_machine_context.product.not_for_distribution = 'SwitchboardDummy' payload = { 'data': { 'productId': 3141281, 'notForDistribution': 'SwitchboardDummy' } } release_correction.update_product_nfd(state_machine_context) mock_graphql_execute.assert_called_with( queries.UPDATE_PRODUCT, payload) def test_is_track_role_finds_classical(s3_ddex): """Test is_track_role method finds classical role.""" s3_ddex_data = S3Schema().load(s3_ddex) s3_ddex_data.product.genres[0].genre = 'CLASSICAL-00' role = 'Composer' genre = s3_ddex_data.product.genres assert utils.is_track_role(role, genre) def test_sanitize_special_instructions(): """Test sanitize_special_instructions strips values.""" special_instructions = 'Special instructions ../' expected = 'Special instructions' sanitized = index.sanitize_special_instructions(special_instructions) assert sanitized == expected @patch('index.update_product') @patch('index.sleep') @patch('index.get_message_error_code', return_value='product_code_used') def test_handle_product_code_collision( mock_get_message_error_code, mock_sleep, mock_update_product, context, s3_ddex): """Test handle_product_code_collision.""" context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) mock_update_product.side_effect = [ graphql.GraphQLError([{'error': 'message'}]), None ] index.handle_product_code_collision(context, s3_ddex_data) mock_update_product_calls = mock_update_product.call_args_list assert mock_update_product_calls == [ call(context, s3_ddex_data), call(context, s3_ddex_data), ] @patch('index.update_product') @patch('index.sleep') @patch('index.get_message_error_code', return_value='product_code_used') def test_handle_product_code_collision_not_found( mock_get_message_error_code, mock_sleep, mock_update_product, context, s3_ddex): """Test handle_product_code_collision when a product code is not found.""" context = StateMachineSchema().load(context) s3_ddex_data = S3Schema().load(s3_ddex) side_effect_list = [ graphql.GraphQLError([{'error': 'message'}]) for _ in range(300) ] mock_update_product.side_effect = side_effect_list with pytest.raises(SetProductException): index.handle_product_code_collision(context, s3_ddex_data) def test_get_message_error_code(): """Test get_message_error_code returns the error code.""" mock_error = MagicMock() error_code = 'product_code_used' mock_error.get_response_body.return_value = { 'message': { 'product_code': { 'error_code': error_code } } } result = index.get_message_error_code(mock_error) assert result == error_code def test_get_message_error_code_different_format(): """Test get_message_error_code finds the error for a diff type of error.""" mock_error = MagicMock() error_code = 'error message' mock_error.get_response_body.return_value = { 'message': error_code } result = index.get_message_error_code(mock_error) assert result == error_code