"""Test handler.""" from unittest.mock import MagicMock, patch import pytest from ddex_ingester_common.constants.swb_deal_types import (FOR_DISTRIBUTION, SWBDUMMY) from ddex_ingester_common.lambda_exceptions import ( LookupSwitchboardDealException, LookupSwitchboardInactiveDealException) from ddex_ingester_common.schemas.s3_schema import S3Schema from ddex_ingester_common.schemas.state_machine_schema import \ StateMachineSchema from marshmallow.utils import get_value from constants.sql_queries import (SELECT_FROM_INBOUND_MAJOR_LABEL_MAPPING, SELECT_ORIGINAL_UPC) from index import (define_swb_deal_type, format_assigned_responsibility, format_not_applicable_responsibility, format_special_instructions, format_worldwide_responsibility, get_original_remap_upc, get_switchboard_data, get_vendor_and_subaccount_for_major_label, handler) @patch('index.logger') @patch('index.get_switchboard_data', wraps=get_switchboard_data) @patch('index.boto3.client') @patch('index.generate_switchboard_jwt') @patch('index.load_ddex_json') @patch('index.save_s3_context') @patch('index.config.graphql_switchboard') @patch('index.get_original_remap_upc') def test_handler( mock_original_upc_remap, mock_graphql_switchboard, mock_save_s3_context, mock_load_ddex_json, mock_generate_switchboard_jwt, mock_boto, mock_get_switchboard_data, mock_logger, context): """Test the main handler.""" expected_vendor_id = 123 expected_subaccount_id = 999 mock_graphql_switchboard.execute.return_value = { 'data': { 'getProductByUpc': { 'maintenanceOwner': { 'name': 'Legacy Recordings' }, 'labelAccount': { 'companyCode': expected_vendor_id, }, 'subAccount': { 'companyCode': expected_subaccount_id, }, 'switchboardExtensions': { 'governingDeal': { 'dealId': '157', 'status': 'ACTIVE', 'dealCoordinators': [ 'awesome-coordinator@theorchard.com'], 'dealResponsibilities': [{ 'dealResponsibilityType': { 'dealResponsibilityTypeId': 1, 'name': 'Other Name' }, 'appliesTo': 'OTHER', 'assignments': [{ 'scope': 'scope', 'system': 'SONY', 'territories': ['A', 'B'] }] }] } } } } } mock_load_ddex_json.return_value = context handler(context, None) mock_generate_switchboard_jwt.assert_called_once() mock_get_switchboard_data.assert_called_once() mock_save_s3_context.assert_called_once() generated_s3_context = mock_save_s3_context.call_args[0][1] assert generated_s3_context.product.vendor_id == expected_vendor_id assert generated_s3_context.product.subaccount_id == expected_subaccount_id @patch('index.logger') @patch('index.get_switchboard_data', wraps=get_switchboard_data) @patch('index.boto3.client') @patch('index.generate_switchboard_jwt') @patch('index.load_ddex_json') @patch('index.save_s3_context') @patch('index.config.graphql_switchboard') @patch('index.get_original_remap_upc') def test_handler_for_multiple_rep_owners( mock_original_upc_remap, mock_graphql_switchboard, mock_save_s3_context, mock_load_ddex_json, mock_generate_switchboard_jwt, mock_boto, mock_get_switchboard_data, mock_logger, context): """Test the main handler when there are multiple rep owners.""" context['product']['parent_repertoire_owner_code'] = '9999' context['product']['repertoire_owner_code'] = '9999' maint_owner_code = '1234' context['product']['maintenance_owner_code'] = maint_owner_code mock_graphql_switchboard.execute.return_value = {} mock_get_switchboard_data.return_value = None mock_load_ddex_json.return_value = context handler(context, None) mock_generate_switchboard_jwt.assert_called_once() mock_get_switchboard_data.assert_called_once() mock_save_s3_context.assert_called_once() s3_context = mock_save_s3_context.call_args[0][1] assert s3_context.product.parent_repertoire_owner_code == maint_owner_code assert s3_context.product.repertoire_owner_code == maint_owner_code @patch('index.config.graphql_switchboard') def test_get_switchboard_data(mock_graphql_switchboard, s3_ddex): """Test get_switchboard_data.""" mock_graphql_switchboard.execute.return_value = { 'data': { 'getProductByUpc': { 'maintenanceOwner': { 'name': 'Legacy Recordings' }, 'labelAccount': { 'companyCode': 123, }, 'subAccount': { 'companyCode': 999, }, 'switchboardExtensions': { 'governingDeal': { 'dealId': '157', 'status': 'ACTIVE', 'dealCoordinators': [ 'awesome-coordinator@theorchard.com'], 'dealResponsibilities': [{ 'dealResponsibilityType': { 'dealResponsibilityTypeId': 1, 'name': 'Other Name' }, 'appliesTo': 'OTHER', 'assignments': [{ 'scope': 'scope', 'system': 'SONY', 'territories': ['A', 'B'] }] }] } } } } } context = MagicMock() get_switchboard_data(context, 123, s3_ddex) assert context.maintenance_owner == 'Legacy Recordings' assert context.product.vendor_id == 123 assert context.product.subaccount_id == 999 assert context.product.special_instructions == \ 'Switchboard Deal Summary:\nOther Name: SONY - excludes A, B' assert context.product.not_for_distribution == SWBDUMMY assert context.deal_coordinators == ['awesome-coordinator@theorchard.com'] @patch('index.get_vendor_and_subaccount_for_rep_owner_code_rows') @patch('index.config.graphql_switchboard') @patch('index.is_ccm_som_livre_grps_ddex_ingestion_enabled') def test_get_switchboard_data_missing_deal( mock_is_ccm_som_livre_grps_ddex_ingestion_enabled, mock_graphql_switchboard, mock_get_vendor_and_subaccount_for_rep_owner_code_rows, s3_ddex, context): """Test get_switchboard_data without deal nor major label mapping.""" mock_is_ccm_som_livre_grps_ddex_ingestion_enabled.return_value = True mock_get_vendor_and_subaccount_for_rep_owner_code_rows.return_value = None mock_graphql_switchboard.execute.return_value = { 'data': { 'getProductByUpc': { 'maintenanceOwner': { 'name': 'Legacy Recordings' }, 'switchboardExtensions': None } } } context = StateMachineSchema().load(context) # Remove major label and rep owners from the context del s3_ddex['product']['major_label'] del s3_ddex['product']['parent_repertoire_owner_code'] del s3_ddex['product']['repertoire_owner_code'] s3_data = S3Schema().load(s3_ddex) with pytest.raises(LookupSwitchboardDealException) as exception: get_switchboard_data(context, 123, s3_data) upc = context.product.upc assert str( exception.value) == ( f'Unable to lookup vendor/subaccount for UPC: {upc}') @patch('index.config.graphql_switchboard') def test_get_switchboard_inactive_deal(mock_graphql_switchboard, s3_ddex): """Test get_switchboard_data.""" deal_status = 'DRAFT' mock_graphql_switchboard.execute.return_value = { 'data': { 'getProductByUpc': { 'maintenanceOwner': { 'name': 'Legacy Recordings' }, 'switchboardExtensions': { 'governingDeal': { 'status': deal_status, 'dealId': 280 } } } } } context = MagicMock() with pytest.raises(LookupSwitchboardInactiveDealException) as exception: get_switchboard_data(context, 123, s3_ddex) assert str( exception.value) == ( f'Switchboard deal is not Active. Deal status: {deal_status}') def test_format_special_instructions_simple(): """Test format_special_instructions in the same format as other tests.""" input_data = { 'switchboardExtensions': { 'governingDeal': { 'dealId': '157', 'dealResponsibilities': [ { 'dealResponsibilityType': { 'dealResponsibilityTypeId': 1, 'name': 'Other Name' }, 'appliesTo': 'OTHER', 'assignments': [{ 'scope': 'scope', 'system': 'SONY', 'territories': ['A', 'B'] }] }, { 'dealResponsibilityType': { 'dealResponsibilityTypeId': 1, 'name': 'Not Applicable Name' }, 'appliesTo': 'NOT_APPLICABLE', 'assignments': [{ 'scope': 'scope', 'system': 'SONY', 'territories': ['A', 'B'] }] }, { 'dealResponsibilityType': { 'dealResponsibilityTypeId': 1, 'name': 'Worldwide Name' }, 'appliesTo': 'WORLDWIDE', 'assignments': [{ 'scope': 'scope', 'system': 'SONY', 'territories': ['A', 'B'] }] } ] } } } expected_result = \ 'Switchboard Deal Summary:\n' + \ 'Not Applicable Name: NOT_APPLICABLE\n' + \ 'Other Name: SONY - excludes A, B\n' + \ 'Worldwide Name: SONY WORLDWIDE' result = format_special_instructions(input_data) assert result == expected_result def test_format_special_instructions_invalid(): """Test format_special_instructions with invalid input.""" result = format_special_instructions([]) assert not result @pytest.mark.parametrize('special_instructions_data,expected_output', [ ( [{ 'dealResponsibilityType': { 'dealResponsibilityTypeId': '2', 'name': 'Copyright Administration' }, 'appliesTo': 'WORLDWIDE', 'assignments': [ { 'scope': None, 'system': 'ORCHARD', 'territories': [ ] } ] }], ( 'Switchboard Deal Summary:\n' 'Copyright Administration: ORCHARD WORLDWIDE' ) ), ( [{ 'dealResponsibilityType': { 'dealResponsibilityTypeId': '2', 'name': 'Copyright Administration' }, 'appliesTo': 'NOT_APPLICABLE', 'assignments': [] }], ( 'Switchboard Deal Summary:\n' 'Copyright Administration: NOT_APPLICABLE' ) ), ( [{ 'dealResponsibilityType': { 'dealResponsibilityTypeId': '2', 'name': 'Copyright Administration' }, 'appliesTo': 'SPECIFIC_TERRITORIES', 'assignments': [ { 'scope': 'INCLUDED', 'system': 'ORCHARD', 'territories': [ 'BN', 'KH', 'CN', 'GE', 'HK', 'IN' ] }, { 'scope': 'INCLUDED', 'system': 'SONY', 'territories': [ 'HR', 'CY', 'CZ', 'EE', 'FO', 'FR', 'GI', 'GB', 'GR', 'GG', 'HU', ] }, ] }], ( 'Switchboard Deal Summary:\n' 'Copyright Administration: ORCHARD - includes BN, KH, CN, GE, HK, IN\n' # noqa 'Copyright Administration: SONY - includes GB, FR, HR, CY, CZ, EE, FO, GI, GR, GG and some other countries' # noqa ) ), ]) def test_format_special_instructions( special_instructions_data, expected_output): """Test special instructions formatter.""" output = format_special_instructions({ 'switchboardExtensions': { 'governingDeal': { 'dealResponsibilities': special_instructions_data } } }) assert output == expected_output def test_format_not_applicable_responsibility(): """Test not applicable responsibility formatter.""" responsibility = { 'dealResponsibilityType': { 'dealResponsibilityTypeId': '2', 'name': 'A responsibility name' }, 'appliesTo': 'NOT_APPLICABLE', 'assignments': [] } output = format_not_applicable_responsibility(responsibility) assert output == 'A responsibility name: NOT_APPLICABLE' def test_format_worldwide_responsibility(): """Test worldwide responsibility formatter.""" responsibility = { 'dealResponsibilityType': { 'dealResponsibilityTypeId': '2', 'name': 'A responsibility name' }, 'appliesTo': 'WORLDWIDE', 'assignments': [ { 'scope': None, 'system': 'ORCHARD', 'territories': [ ] } ] } output = format_worldwide_responsibility(responsibility) assert output == 'A responsibility name: ORCHARD WORLDWIDE' @pytest.mark.parametrize('assignment,expected_output', [ ( { 'scope': 'INCLUDED', 'system': 'ORCHARD', 'territories': [ 'BN', 'KH', 'CN', 'GE', 'HK', 'IN', 'GB' ] }, ( 'A responsibility name: ORCHARD' ' - includes GB, BN, KH, CN, GE, HK, IN' ) ), ( { 'scope': 'INCLUDED', 'system': 'ORCHARD', 'territories': [ 'HR', 'CY', 'CZ', 'EE', 'FO', 'FR', 'GI', 'GB', 'GR', 'GG', 'HU', ] }, ( 'A responsibility name: ORCHARD - includes ' 'GB, FR, HR, CY, CZ, EE, FO, GI, GR, GG ' 'and some other countries' ) ), ( { 'scope': 'INCLUDED', 'system': 'ORCHARD', 'territories': [ ] }, ( 'A responsibility name: ORCHARD - includes ' 'no territories' ) ), ( { 'scope': 'EXCLUDED', 'system': 'ORCHARD', 'territories': [ ] }, ( 'A responsibility name: ORCHARD - excludes ' 'no territories' ) ), ]) def test_format_assigned_responsibility(assignment, expected_output): """Test assignment responsibility formatter.""" responsibility = { 'dealResponsibilityType': { 'dealResponsibilityTypeId': '2', 'name': 'A responsibility name' }, 'appliesTo': 'SPECIFIC_TERRITORIES', } output = format_assigned_responsibility(responsibility, assignment) assert output == expected_output def test_define_swb_deal_type_nfd_n(swb_query_response): """Test define_swb_deal_type.""" responsibilities = get_value( swb_query_response, 'data.getProductByUpc.switchboardExtensions.governingDeal.dealResponsibilities', # noqa []) response = define_swb_deal_type(responsibilities) assert response == FOR_DISTRIBUTION @patch('index.logger') @patch('index.get_switchboard_data', wraps=get_switchboard_data) @patch('index.boto3.client') @patch('index.generate_switchboard_jwt') @patch('index.load_ddex_json') @patch('index.save_s3_context') @patch('index.config.graphql_switchboard') def test_handler_ignores_s3_context_for_purged_release( mock_graphql_switchboard, mock_save_s3_context, mock_load_ddex_json, mock_generate_switchboard_jwt, mock_boto, mock_get_switchboard_data, mock_logger, purged_release_context): """Test the handler ignores s3 context for purged releases.""" mock_graphql_switchboard.execute.return_value = { 'data': { 'getProductByUpc': { 'maintenanceOwner': { 'name': 'Legacy Recordings' }, 'labelAccount': { 'companyCode': 123, }, 'subAccount': { 'companyCode': 999, }, 'switchboardExtensions': { 'governingDeal': { 'dealId': '157', 'status': 'ACTIVE', 'dealCoordinators': [ 'awesome-coordinator@theorchard.com'], 'dealResponsibilities': [{ 'dealResponsibilityType': { 'dealResponsibilityTypeId': 1, 'name': 'Other Name' }, 'appliesTo': 'OTHER', 'assignments': [{ 'scope': 'scope', 'system': 'SONY', 'territories': ['A', 'B'] }] }] } } } } } handler(purged_release_context, None) mock_load_ddex_json.assert_not_called() mock_save_s3_context.assert_not_called() mock_generate_switchboard_jwt.assert_called_once() mock_get_switchboard_data.assert_called_once() @patch('ddex_ingester_common.helpers.rds.mysql_connection') def test_remap_upc_no_remap( mock_mysql_connection, context, s3_ddex): """Test get_original_remap_upc with a UPC that does not need remapping.""" context = StateMachineSchema().load(context) s3_data = S3Schema().load(s3_ddex) expected_upc = context.product.upc mock_rds_conn = MagicMock() mock_mysql_connection.return_value.__enter__.return_value = mock_rds_conn mock_cursor = MagicMock() mock_rds_conn.cursor.return_value.__enter__.return_value = mock_cursor mock_cursor.fetchall.return_value = [] get_original_remap_upc(context, s3_data) mock_cursor.execute.assert_called_with( SELECT_ORIGINAL_UPC, args=(expected_upc, None) ) mock_cursor.fetchall.assert_called_with() assert context.product.upc == expected_upc @patch('ddex_ingester_common.helpers.rds.mysql_connection') def test_remap_upc_remap( mock_mysql_connection, context, s3_ddex): """Test get_original_remap_upc with a UPC that needs remapping.""" context = StateMachineSchema().load(context) s3_data = S3Schema().load(s3_ddex) vendor_id = s3_data.product.vendor_id original_upc = '231232463123' mock_rds_conn = MagicMock() mock_mysql_connection.return_value.__enter__.return_value = mock_rds_conn mock_cursor = MagicMock() mock_rds_conn.cursor.return_value.__enter__.return_value = mock_cursor mock_cursor.fetchall.return_value = [{ 'original_upc': original_upc, 'vendor_id': vendor_id, }] returned_upc = get_original_remap_upc(context, s3_data) mock_cursor.execute.assert_called_with( SELECT_ORIGINAL_UPC, args=(context.product.upc, None) ) mock_cursor.fetchall.assert_called_with() assert returned_upc == original_upc @patch('ddex_ingester_common.helpers.rds.mysql_connection') def test_get_vendor_and_subaccount_for_major_label( mock_mysql_connection, context, s3_ddex): """Test get_vendor_and_subaccount_for_major_label logic.""" context = StateMachineSchema().load(context) s3_data = S3Schema().load(s3_ddex) returned_vendor_id = 12345 returned_subaccount_id = 54321 mock_rds_conn = MagicMock() mock_mysql_connection.return_value.__enter__.return_value = mock_rds_conn mock_cursor = MagicMock() mock_rds_conn.cursor.return_value.__enter__.return_value = mock_cursor mock_cursor.fetchall.return_value = [{ 'vendor_id': returned_vendor_id, 'subaccount_id': returned_subaccount_id, }] returned_tuple = get_vendor_and_subaccount_for_major_label( context, s3_data) mock_cursor.execute.assert_called_with( SELECT_FROM_INBOUND_MAJOR_LABEL_MAPPING, args=(s3_data.product.major_label,) ) mock_cursor.fetchall.assert_called_with() assert returned_tuple == (returned_vendor_id, returned_subaccount_id) @patch('index.logger') @patch('index.config.graphql_switchboard') @patch('index.is_ccm_som_livre_grps_ddex_ingestion_enabled') @patch('index.get_vendor_and_subaccount_for_major_label') def test_get_switchboard_data_with_major_label( mock_get_vendor_and_subaccount_for_major_label, mock_is_ccm_som_livre_grps_ddex_ingestion_enabled, mock_graphql_switchboard, mock_logger, context, s3_ddex): """Test get_switchboard_data when there's no deal but is a major label.""" mock_get_vendor_and_subaccount_for_major_label.return_value = (12345, None) mock_is_ccm_som_livre_grps_ddex_ingestion_enabled.return_value = True mock_graphql_switchboard.execute.return_value = { 'data': { 'getProductByUpc': { 'maintenanceOwner': { 'name': 'Legacy Recordings' }, 'switchboardExtensions': None } } } context = StateMachineSchema().load(context) s3_data = S3Schema().load(s3_ddex) context.major_label = 'TestCode' get_switchboard_data(context, 123, s3_data) mock_is_ccm_som_livre_grps_ddex_ingestion_enabled.assert_called_once() mock_get_vendor_and_subaccount_for_major_label.assert_called_with( context, s3_data) # Check that vendor id has been changed assert context.product.vendor_id == 12345 @patch('index.logger') @patch('index.config.graphql_switchboard') @patch('index.is_ccm_som_livre_grps_ddex_ingestion_enabled') @patch('index.get_vendor_and_subaccount_for_major_label') def test_get_switchboard_data_with_rep_owner( mock_get_vendor_and_subaccount_for_rep_owner_code_rows, mock_is_ccm_som_livre_grps_ddex_ingestion_enabled, mock_graphql_switchboard, mock_logger, context, s3_ddex): """Test get_switchboard_data when there's no deal but rep owner exists.""" mock_get_vendor_and_subaccount_for_rep_owner_code_rows.return_value = ( 12345, None) mock_is_ccm_som_livre_grps_ddex_ingestion_enabled.return_value = True mock_graphql_switchboard.execute.return_value = { 'data': { 'getProductByUpc': { 'maintenanceOwner': { 'name': 'Legacy Recordings' }, 'switchboardExtensions': None } } } context = StateMachineSchema().load(context) s3_data = S3Schema().load(s3_ddex) context.major_label = 'TestCode' get_switchboard_data(context, 123, s3_data) mock_is_ccm_som_livre_grps_ddex_ingestion_enabled.assert_called_once() mock_get_vendor_and_subaccount_for_rep_owner_code_rows.assert_called_with( context, s3_data) # Check that vendor id has been changed assert context.product.vendor_id == 12345