"""Test set_video_metadata handler.""" from unittest.mock import call, patch import pytest from ddex_ingester_common.models.s3.body import Body as S3Context from ddex_ingester_common.models.state_machine.body import \ Body as StateMachineContext 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 constants import queries from constants.video import DEFAULT_VIDEO_CHANNEL, RC_JSON_NAME @patch('index.logger') @patch('index.load_ddex_json') @patch('index.update_video_metadata') @patch('index.check_for_product') def test_handler( mock_check_for_product, mock_update_video_metadata, mock_load_ddex_json, mock_logger, context_data): """Test handler.""" mock_load_ddex_json.return_value = context_data index.handler(context_data, None) mock_update_video_metadata.assert_called() args, _ = mock_update_video_metadata.call_args assert len(args) == 2 assert isinstance(args[0], StateMachineContext) assert isinstance(args[1], S3Context) @patch('index.logger') @patch('index.load_ddex_json') @patch('index.check_for_video_updates') @patch('index.check_for_product') def test_handler_in_content_path( mock_check_for_product, mock_check_for_video_updates, mock_load_ddex_json, mock_logger, context_data): """Test handler.""" context_data['product']['status'] = 'in_content' mock_load_ddex_json.return_value = context_data index.handler(context_data, None) mock_check_for_video_updates.assert_called() args, _ = mock_check_for_video_updates.call_args assert len(args) == 3 assert isinstance(args[0], StateMachineContext) assert isinstance(args[1], S3Context) @patch('index.write_rc_json') @patch('index.create_blank_rc_json') @patch('index.handle_channel_selection', return_value=[{}, {}]) @patch('index.get_special_instructions', return_value='special instructions') @patch('index.graphql_gateway.execute') @patch('index.check_for_video_updates', wraps=index.check_for_video_updates) def test_check_for_video_updates( mock_check_for_video_updates, mock_graphql_execute, mock_get_special_instructions, mock_handle_channel_selection, mock_create_json, mock_write_rc, context_data, s3_context, s3_release_corrections): """Test check_for_video_updates.""" graphql_result = { 'specialInstructions': 'test instructions', 'imprint': 'test imprint', 'productCode': 'test product code', 'notForDistribution': 'test nfd', 'subgenreId': 'test subgenre', 'genreId': 'test genre', 'productId': 'test product id', 'vendorId': 'test vendor id', 'subaccountId': 'test subaccount id', 'deliveredVersion': 'test delivered version', 'version': 'test video version', 'videoProduct': { 'typeOfVideo': 'test type of video', 'pLine': { 'year': 2021, 'holder': 'pLine Holder' }, 'cLine': { 'year': 2022, 'holder': 'cLine Holder' }, 'lyrics': 'Here are my cool lyrics', 'languageOfVideoContent': 'FRE', 'languageOfVideoTitle': 'FRE', 'description': 'Here is my cool description', 'parentalAdvisory': 'NO', 'keywords': None, } } context_data['product']['status'] = 'in_content' s3_context['product']['status'] = 'in_content' context = StateMachineSchema().load(context_data) s3_context = S3Schema().load(s3_context) index.check_for_video_updates(context, s3_context, graphql_result) event_details = {'bucket': context.bucket, 'key': context.key} expected_s3_output = {'changes': [ReleaseCorrectionDiffDetail(field_name='productId', isrc=None, old='test product id', new=3382207, db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='vendorId', isrc=None, old='test vendor id', new=3453, db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='videoTitle', isrc=None, old=None, new='Before Your Love', db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='deliveredVersion', isrc=None, old='test video version', new='VIDEO', db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='genreId', isrc=None, old='test genre', new=1, db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='subgenreId', isrc=None, old='test subgenre', new=2, db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='notForDistribution', isrc=None, old='test nfd', new='SwitchboardDummy', db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='productCode', isrc=None, old='test product code', new='G010000025264A', db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='specialInstructions', isrc=None, old='test instructions', new='special instructions', db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='lyrics', isrc=None, old='Here are my cool lyrics', new='Some lyrics', db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='languageOfVideoContent', isrc=None, old='FRE', new='ENG', db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='languageOfVideoTitle', isrc=None, old='FRE', new='ENG', db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='description', isrc=None, old='Here is my cool description', new='Some video description', db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='keywords', isrc=None, old=set(), new={'keyword1', 'keyword2'}, db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='typeOfVideo', isrc=None, old='test type of video', new='Official Music Video', db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='plineYear', isrc=None, old=2021, new=2018, db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='plineCopyrightHolder', isrc=None, old='pLine Holder', new='SG Productions', db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='clineYear', isrc=None, old=2022, new=2018, db_table_name='', db_field_name='', email_customer=False, accept_update=False), ReleaseCorrectionDiffDetail(field_name='clineCopyrightHolder', isrc=None, old='cLine Holder', new='SG Productions', db_table_name='', db_field_name='', email_customer=False, accept_update=False)]} # noqa mock_write_rc.assert_called_with( event_details, expected_s3_output, RC_JSON_NAME) @patch('index.graphql_gateway.execute') @patch('index.format_primary_artists') @patch('index.handle_channel_selection', return_value=('orchardmusic', False)) @patch('index.get_special_instructions', return_value='special instructions') def test_update_video_metadata( mock_get_special_instructions, mock_handle_channel_selection, mock_format_primary_artists, mock_graphql_execute, context_data): """Test update_video_metadata.""" context = StateMachineSchema().load(context_data) s3_context = S3Schema().load(context_data) product_id = context_data['product']['product_id'] mock_response = { 'data': { 'saveVideoSingleProduct': { 'productId': product_id, } } } mock_graphql_execute.return_value = mock_response mock_primary_artists = [{ 'artistName': 'Kelly Clarkson', 'artistType': 'featuring', 'artistId': '1A', 'vendorId': 3453, 'subaccountId': 0 }] mock_format_primary_artists.return_value = mock_primary_artists expected_payload = { 'data': { 'update': { 'productId': str(product_id), 'accountId': context_data['product']['vendor_id'], 'subaccountId': context_data['product']['subaccount_id'], 'videoTitle': context_data['video']['video_name'], 'version': context_data['video']['video_version'], 'typeOfVideo': 'Official Music Video', 'languageOfVideoTitle': context_data['video']['title_language'], 'languageOfVideoContent': context_data['video']['content_language'], 'imprint': context_data['video']['imprint'], 'description': context_data['video']['description'], 'lyrics': context_data['video']['lyrics'], 'parentalAdvisory': context_data['video']['parental_advisory'], 'specialInstructions': 'special instructions', 'associatedTrackId': context_data['video']['associated_track_tuid'], 'genreId': 1, 'subgenreId': 2, 'plineYear': context_data['video']['p_line']['year'], 'plineCopyrightHolder': context_data['video']['p_line']['line'], 'clineYear': context_data['video']['c_line']['year'], 'clineCopyrightHolder': context_data['video']['c_line']['line'], 'productCode': context_data['video']['product_code'], 'isrc': context_data['video']['isrc'], 'primaryArtists': mock_primary_artists, 'keywords': context_data['video']['keywords'], 'notForDistribution': ( context_data['product']['not_for_distribution']), 'vendorReleaseIdentifier': None, 'manufacturerUpc': '111115264505', 'channelSelection': 'orchardmusic', 'vendorCatalogNumber': context_data['product']['catalog_number'], # noqa } } } return_value = index.update_video_metadata(context, s3_context) mock_graphql_execute.assert_called_with( queries.SAVE_VIDEO_SINGLE_PRODUCT, expected_payload ) assert return_value == mock_response['data']['saveVideoSingleProduct'] @patch('index.graphql_gateway.execute') @patch('index.format_primary_artists') @patch('index.handle_channel_selection', return_value=('orchardmusic', False)) @patch('index.get_special_instructions', return_value='special instructions') def test_update_video_metadata_for_awal( mock_get_special_instructions, mock_handle_channel_selection, mock_format_primary_artists, mock_graphql_execute, context_data): """Test update_video_metadata.""" context = StateMachineSchema().load(context_data) context.ddex_provider = 'AWAL' s3_context = S3Schema().load(context_data) product_id = context_data['product']['product_id'] mock_response = { 'data': { 'saveVideoSingleProduct': { 'productId': product_id, } } } mock_graphql_execute.return_value = mock_response mock_primary_artists = [{ 'artistName': 'Kelly Clarkson', 'artistType': 'featuring', 'artistId': '1A', 'vendorId': 3453, 'subaccountId': 0 }] mock_format_primary_artists.return_value = mock_primary_artists expected_payload = { 'data': { 'update': { 'productId': str(product_id), 'accountId': context_data['product']['vendor_id'], 'subaccountId': context_data['product']['subaccount_id'], 'videoTitle': context_data['video']['video_name'], 'version': context_data['video']['video_version'], 'typeOfVideo': 'Official Music Video', 'languageOfVideoTitle': context_data['video']['title_language'], 'languageOfVideoContent': context_data['video']['content_language'], 'imprint': context_data['video']['imprint'], 'description': context_data['video']['description'], 'lyrics': context_data['video']['lyrics'], 'parentalAdvisory': context_data['video']['parental_advisory'], 'specialInstructions': 'special instructions', 'associatedTrackId': context_data['video']['associated_track_tuid'], 'genreId': 1, 'subgenreId': 2, 'plineYear': context_data['video']['p_line']['year'], 'plineCopyrightHolder': context_data['video']['p_line']['line'], 'clineYear': context_data['video']['c_line']['year'], 'clineCopyrightHolder': context_data['video']['c_line']['line'], 'productCode': s3_context.video.product_code, 'isrc': context_data['video']['isrc'], 'primaryArtists': mock_primary_artists, 'keywords': context_data['video']['keywords'], 'notForDistribution': ( context_data['product']['not_for_distribution']), 'vendorReleaseIdentifier': None, 'manufacturerUpc': '111115264505', 'channelSelection': 'orchardmusic', 'vendorCatalogNumber': s3_context.product.catalog_number, } } } return_value = index.update_video_metadata(context, s3_context) mock_graphql_execute.assert_called_with( queries.SAVE_VIDEO_SINGLE_PRODUCT, expected_payload ) assert return_value == mock_response['data']['saveVideoSingleProduct'] def test_format_primary_artists(context_data, s3_context): """Test format_primary_artists.""" context = StateMachineSchema().load(context_data) s3_context = S3Schema().load(s3_context) expected_result = [ { 'artistName': 'Kelly Clarkson', 'artistType': 'featuring', 'artistId': 1, 'vendorId': 3453, 'subaccountId': 0 }, { 'artistName': 'Artist', 'artistType': 'director', 'artistId': 2, 'vendorId': 3453, 'subaccountId': 0 }, { 'artistName': 'Artist', 'artistType': 'videographer', 'artistId': 2, 'vendorId': 3453, 'subaccountId': 0 }, { 'artistName': 'John', 'artistType': 'editor', 'artistId': 3, 'vendorId': 3453, 'subaccountId': 0 } ] result = index.format_primary_artists(context, s3_context) assert result == expected_result @patch('index.graphql_gateway.execute') def test_handle_channel_selection( mock_graphql_execute, save_video_single_product, available_channels_by_product_id): """Test handle channel selection.""" product_id = 12345 video_channel = 'TestVEVO' mock_graphql_execute.side_effect = [ available_channels_by_product_id, save_video_single_product ] selected_channel, valid = index.handle_channel_selection( product_id, None, video_channel) expected_calls = [ call(queries.GET_VIDEO_CHANNELS, { 'productId': product_id }) ] mock_graphql_execute.assert_has_calls( expected_calls ) assert selected_channel == video_channel assert valid @patch('index.graphql_gateway.execute') def test_handle_channel_selection_defaults_if_none( mock_graphql_execute, save_video_single_product, available_channels_by_product_id): """Test handle channel selection defaults if None.""" product_id = 12345 video_channel = None # Should default the video channel mock_graphql_execute.side_effect = [ available_channels_by_product_id, save_video_single_product ] selected_channel, valid = index.handle_channel_selection( product_id, None, video_channel) expected_calls = [ call(queries.GET_VIDEO_CHANNELS, { 'productId': product_id }) ] mock_graphql_execute.assert_has_calls( expected_calls ) assert selected_channel == DEFAULT_VIDEO_CHANNEL assert not valid @patch('index.graphql_gateway.execute') def test_handle_channel_selection_defaults_if_not_supported( mock_graphql_execute, save_video_single_product, available_channels_by_product_id): """Test handle channel selection defaults if unsupported.""" product_id = 12345 video_channel = 'UnsupportedVEVO' # Should default the video channel mock_graphql_execute.side_effect = [ available_channels_by_product_id, save_video_single_product, '' ] selected_channel, valid = index.handle_channel_selection( product_id, None, video_channel) expected_calls = [ call(queries.GET_VIDEO_CHANNELS, { 'productId': product_id }) ] mock_graphql_execute.assert_has_calls( expected_calls ) assert selected_channel == DEFAULT_VIDEO_CHANNEL assert not valid @patch('index.graphql_gateway.execute') def test_handle_channel_selection_if_matching_channel_id( mock_graphql_execute, save_video_single_product, available_channels_by_product_id): """Test handle channel selection.""" product_id = 12345 # Note: It is unknown whether or not DDEX is delivered with channel ids video_channel = 'UCCu6XdOryMB1I0VLK9M5kHw' # A channel ID was provided mock_graphql_execute.side_effect = [ available_channels_by_product_id, save_video_single_product ] selected_channel, valid = index.handle_channel_selection( product_id, None, video_channel) expected_calls = [ call(queries.GET_VIDEO_CHANNELS, { 'productId': product_id }) ] mock_graphql_execute.assert_has_calls( expected_calls ) assert selected_channel == video_channel assert valid @patch('index.graphql_gateway.execute') def test_handle_channel_selection_if_matching_channel_id_with_invalid_name( mock_graphql_execute, save_video_single_product, available_channels_by_product_id): """Test handle channel selection.""" product_id = 12345 # Note: It is unknown whether or not DDEX is delivered with channel ids # Channel ID was provided and a name is associated to this in the Orchard video_channel = 'ACCu6XdOryMB1I0VLK9M5kHw' mock_graphql_execute.side_effect = [ available_channels_by_product_id, save_video_single_product ] selected_channel, valid = index.handle_channel_selection( product_id, None, video_channel) expected_calls = [ call(queries.GET_VIDEO_CHANNELS, { 'productId': product_id }), ] mock_graphql_execute.assert_has_calls( expected_calls ) assert selected_channel == video_channel assert valid @patch('index.graphql_gateway.execute') def test_handle_channel_selection_existing_valid_channel( mock_graphql_execute, save_video_single_product, available_channels_by_product_id): """Test handle channel selection with an existing valid channel.""" product_id = 12345 current_channel = 'ValidChannel' selected_channel, valid = index.handle_channel_selection( product_id, current_channel, None) assert selected_channel is None assert not valid @pytest.mark.parametrize('test_input,expected', [ ('VEVO MESSAGE. VEVO Channel Name: AudreyVEVO was received from SME but not available', # noqa 'VEVO MESSAGE. VEVO Channel Name: TestVEVO was received from SME but not available'), # noqa ('blah blah VEVO Channel Name: AudreyVEVO was received from SME but not available', # noqa 'blah blah VEVO Channel Name: TestVEVO was received from SME but not available'), # noqa ('VEVO Channel Name: AudreyVEVO was received from SME but not available', 'VEVO Channel Name: TestVEVO was received from SME but not available'), ('VEVO Channel Name: AudreyVEVO was received from SME but not available blah blah', # noqa 'VEVO Channel Name: TestVEVO was received from SME but not available blah blah'), # noqa ('Lots of other text but no Channel message.', 'Lots of other text but no Channel message.\nVEVO Channel Name: TestVEVO was received from SME but not available'), # noqa ('', 'VEVO Channel Name: TestVEVO was received from SME but not available') ]) @patch('index.graphql_gateway.execute') def test_get_special_instructions( mock_graphql_execute, test_input, expected, context_data ): """Test regex for update special instructions.""" context = StateMachineSchema().load(context_data) context.product.special_instructions = None channel_name = 'TestVEVO' upc = '886445897142' mock_graphql_execute.return_value = { 'data': { 'productByUpc': { 'specialInstructions': test_input } } } result = index.get_special_instructions( upc, channel_name, False, context.product.special_instructions, 'SME' ) assert result == expected mock_graphql_execute.assert_called_with( queries.GET_PRODUCT_BY_UPC, {'upc': upc} ) @pytest.mark.parametrize('test_input,expected', [ ('Switchboard instructions\nVEVO MESSAGE. VEVO Channel Name: TestVEVO was received from SME but not available', # noqa 'Switchboard instructions\nVEVO MESSAGE. VEVO Channel Name: TestVEVO was received from SME but not available'), # noqa ('', 'Switchboard instructions\nVEVO Channel Name: TestVEVO was received from SME but not available') # noqa ]) @patch('index.graphql_gateway.execute') def test_get_special_instructions_with_switchboard_instructions( mock_graphql_execute, test_input, expected, context_data ): """Test logic for get_special_instructions when switchboard instructions exist.""" # noqa context = StateMachineSchema().load(context_data) context.product.special_instructions = 'Switchboard instructions' channel_name = 'TestVEVO' upc = '886445897142' mock_graphql_execute.return_value = { 'data': { 'productByUpc': { 'specialInstructions': test_input } } } result = index.get_special_instructions( upc, channel_name, False, context.product.special_instructions, 'SME' ) assert result == expected