"""Unit tests for the get_grps_data Lambda function.""" from unittest.mock import ANY, MagicMock, patch from common.models.state_machine.participant import Participant from common.models.state_machine.product import Product from src import index from src.constants.grps_queries import GET_PRODUCT_DATA_BY_UPC, \ GET_PRODUCT_TRACKS_BY_UPC, GET_PRODUCT_TRACKS_PARTICIPANTS_BY_UPC_ISRC, \ GET_PROJECT_BY_PRODUCT_UPC, GET_PROJECT_PARTICIPANT_BY_REC_PROJECT_ID, \ GET_VENDOR_AND_SUBACCOUNT_BY_PARENT_AND_REP_OWNER_KEY from src.index import get_project, get_tracks, get_vendor_and_subaccount @patch('src.index.get_project') @patch('src.index.get_product') @patch('src.index.get_tracks') @patch('src.index.prep_track_sequence_and_volume_number') @patch('src.index.prep_track_name_length') @patch('src.index.prep_forbidden_sequence') def test_handler(prep_forbidden_sequence_mock, prep_track_name_length_mock, prep_track_sequence_and_volume_number_mock, get_tracks_mock, get_product_mock, get_project_mock): """Test handler function.""" test_upc = '100000' test_prod_no = 'XYZ' get_product_mock.return_value = (MagicMock(), 'AUDIO') index.handler( {'upc': test_upc, 'grps_ingestion_id': 1, 'prod_no': test_prod_no}, None) get_tracks_mock.assert_called_with(test_upc, test_prod_no, ANY) get_product_mock.assert_called_with(test_upc, test_prod_no) get_project_mock.assert_called_with(test_upc, ANY, test_prod_no) assert prep_forbidden_sequence_mock.call_count == 1 assert prep_track_name_length_mock.call_count == 1 assert prep_track_sequence_and_volume_number_mock.call_count == 1 @patch('src.index.execute_snowflake_query') def test_get_tracks(execute_snowflake_query_mock): """Test the get tracks getting participant for each track.""" def side_effect(query, params): if query == GET_PRODUCT_TRACKS_BY_UPC: return [MagicMock(), MagicMock(), MagicMock()] elif query == GET_PRODUCT_TRACKS_PARTICIPANTS_BY_UPC_ISRC: return MagicMock() return [] execute_snowflake_query_mock.side_effect = side_effect test_upc = '100000' test_prod_no = 'XYZ' get_tracks(test_upc, test_prod_no, 'Single') matching_calls = [c for c in execute_snowflake_query_mock.call_args_list if c[0][0] == GET_PRODUCT_TRACKS_PARTICIPANTS_BY_UPC_ISRC] assert len(matching_calls) == 3 @patch('src.index.execute_snowflake_query') @patch('src.index.get_vendor_and_subaccount') @patch('src.index.get_rep_owner_data') @patch('src.index.get_label_name') def test_get_product_raise_exception(get_label_name_mock, get_rep_owner_data_mock, get_vendor_and_subaccount_mock, execute_snowflake_query_mock): """Test get product raise exception when no product found.""" def side_effect(query, params): if query == GET_PRODUCT_DATA_BY_UPC: return [MagicMock()] return [] execute_snowflake_query_mock.side_effect = side_effect get_rep_owner_data_mock.return_value = (20, 30, 'name') get_vendor_and_subaccount_mock.return_value = (1, 2, True) get_label_name_mock.return_value = 'random_label_name' test_upc = '100000' test_prod_no = 'XYZ' try: index.get_product(test_upc, test_prod_no) assert False, 'Expected RepOwnerDoNotIngestException to be raised' except index.RepOwnerDoNotIngestException as e: assert str( e) == ('Provided rep owner (20, 30) is marked as do not ingest ' 'in mapping table; Cannot proceed with ingestion.') @patch('src.index.execute_snowflake_query') @patch('src.index.get_vendor_and_subaccount') @patch('src.index.get_rep_owner_data') @patch('src.index.get_label_name') def test_get_product_for_palm_tree(get_label_name_mock, get_rep_owner_data_mock, get_vendor_and_subaccount_mock, execute_snowflake_query_mock): """Test get product for palm tree records.""" def side_effect(query, params): if query == GET_PRODUCT_DATA_BY_UPC: return [MagicMock()] return [] execute_snowflake_query_mock.side_effect = side_effect get_rep_owner_data_mock.return_value = (20, 30, 'name') get_vendor_and_subaccount_mock.return_value = (1, 2, True) get_label_name_mock.return_value = 'Palm Tree Records 123' test_upc = '100000' test_prod_no = 'XYZ' result, _ = index.get_product(test_upc, test_prod_no) assert result.imprint == 'Palm Tree Records 123' @patch('src.index.execute_snowflake_query') def test_get_project(execute_snowflake_query_mock): """Test get_project function.""" def side_effect(query, params): if query == GET_PROJECT_BY_PRODUCT_UPC: return [{'REC_PROJECT_TITLE': 'test_title', 'REC_PROJECT_ID': 777, 'REC_PROJECT_NUMBER': 888, }] elif query == GET_PROJECT_PARTICIPANT_BY_REC_PROJECT_ID: return [{'PARTICIP_FULL_NAME': 'artist_name', 'PARTICIP_SPOTIFY_URI': 'uri_1', 'APPLE_ARTIST_ID': 'uri_2', }] execute_snowflake_query_mock.side_effect = side_effect upc = '123' test_prod_no = 'XYZ' artist = Participant(name='test_artist_name') product = Product(upc=upc, product_name='test_product_name', display_artists=[artist]) result = get_project(upc, product, test_prod_no) assert result.name == 'test_title' assert result.project_code == 888 @patch('src.index.execute_snowflake_query') def test_get_project_data_is_created_if_missing_is_grps( execute_snowflake_query_mock): """Test that project data is created if missing in GRPS.""" def side_effect(query, params): if query == GET_PROJECT_BY_PRODUCT_UPC: return [] return [] execute_snowflake_query_mock.side_effect = side_effect upc = '123' test_prod_no = 'XYZ' artist = Participant(name='test_artist_name') product = Product(upc=upc, product_name='test_product_name', display_artists=[artist]) result = get_project(upc, product, test_prod_no) assert result.name == 'test_product_name' assert result.project_code == upc @patch('src.index.execute_snowflake_query') def test_get_project_fail_for_two_projects( execute_snowflake_query_mock): """Test that get_project raises exception if multiple projects found.""" def side_effect(query, params): return [{'name': 1}, {'name': 2}] execute_snowflake_query_mock.side_effect = side_effect upc = '123' test_prod_no = 'XYZ' artist = Participant(name='test_artist_name') product = Product(upc=upc, product_name='test_product_name', display_artists=[artist]) try: get_project(upc, product, test_prod_no) assert False, 'Expected RepOwnerDoNotIngestException to be raised' except index.GetGrpsDataException as e: assert str(e) == ( 'Expected exactly one project for UPC 123, but got 2') @patch('src.index.execute_snowflake_query') @patch('src.index.get_vendor_and_subaccount') @patch('src.index.get_rep_owner_data') @patch('src.index.get_label_name') @patch('src.index.product_code_is_used_by_different_upc') def test_get_product_with_product_code_used_by_other_upc( product_code_is_used_by_different_upc_mock, get_label_name_mock, get_rep_owner_data_mock, get_vendor_and_subaccount_mock, execute_snowflake_query_mock): """Test _GRPS suffix added to catalog_number if used by another release.""" def side_effect(query, params): if query == GET_PRODUCT_DATA_BY_UPC: return [{'UPC': 'upc', 'PRODUCT_NAME': 'product_name', 'RELEASE_DATE': 'RELEASE_DATE', 'RELEASE_TYPE': 'RELEASE_TYPE', 'SALE_START_DATE': 'SALE_START_DATE', 'PROD_NO': 'FOO_BAR_1', 'PRODUCT_TYPE': 'VIDEO', 'GRID_NO': 'GRID_NO'}] return [] # catalog_number = catalog_number, # sale_start_date = grps_product['SALE_START_DATE'], execute_snowflake_query_mock.side_effect = side_effect get_rep_owner_data_mock.return_value = (20, 30, 'name') get_vendor_and_subaccount_mock.return_value = (1, 2, False) get_label_name_mock.return_value = 'random_label_name' product_code_is_used_by_different_upc_mock.return_value = True test_upc = '100000' test_prod_no = 'FOO_BAR_1' result, _ = index.get_product(test_upc, test_prod_no) assert result.catalog_number == 'FOO_BAR_1_GRPS' @patch('src.index.execute_snowflake_query') def test_get_vendor_and_subaccount(execute_snowflake_query_mock): """Test get_vendor_and_subaccount function.""" def side_effect(query, params): if query == GET_VENDOR_AND_SUBACCOUNT_BY_PARENT_AND_REP_OWNER_KEY: return [{'VENDOR_ID': '1', 'SUBACCOUNT_ID': '2', 'DO_NOT_INGEST': 'N', }] execute_snowflake_query_mock.side_effect = side_effect result = get_vendor_and_subaccount('parent_rep_owner_key', 'rep_owner_key') assert result == ('1', '2', 'N') @patch('src.index.execute_snowflake_query') def test_get_vendor_and_subaccount_with_ambiguous_result( execute_snowflake_query_mock): """Test get_vendor_and_subaccount function.""" def side_effect(query, params): if query == GET_VENDOR_AND_SUBACCOUNT_BY_PARENT_AND_REP_OWNER_KEY: return [{'VENDOR_ID': '1', 'SUBACCOUNT_ID': '2', 'DO_NOT_INGEST': 'N', }, {'VENDOR_ID': '11', 'SUBACCOUNT_ID': '12', 'DO_NOT_INGEST': 'N', } ] execute_snowflake_query_mock.side_effect = side_effect try: get_vendor_and_subaccount('parent_rep_owner_key', 'rep_owner_key') assert False, 'Expected AmbiguousVendorOrSubaccountMapping' except index.AmbiguousVendorOrSubaccountMapping as e: assert str(e) == ( 'Ambiguous vendor/subaccount mapping parent_rep_owner_key ' 'parent_rep_owner_key and rep_owner_key rep_owner_key') @patch('src.index.execute_snowflake_query') def test_get_vendor_and_subaccount_not_found( execute_snowflake_query_mock): """Test get_vendor_and_subaccount function.""" def side_effect(query, params): if query == GET_VENDOR_AND_SUBACCOUNT_BY_PARENT_AND_REP_OWNER_KEY: return [] execute_snowflake_query_mock.side_effect = side_effect result = get_vendor_and_subaccount('parent_rep_owner_key', 'rep_owner_key') assert result == (None, None, None)