"""Unit tests for masters_registry module.""" from decimal import Decimal from unittest.mock import ANY from unittest.mock import MagicMock, Mock from unittest.mock import patch import uuid from moto import mock_aws import boto3 from oto import response import pytest from masters_registry.constant import api_const from masters_registry.constant import error from masters_registry.constant import field_const from masters_registry.constant import opcode_const from masters_registry.logic import masters_registry from masters_registry.logic import ownership as ownership_logic from masters_registry.models import ownership from masters_registry.models import ows_territories from masters_registry.models import yt_ownership from masters_registry.tasks import bulk from tests import test_utils from tests.helpers import patches from masters_registry.connectors.dynamodb import client as outside_dynamodb_client dynamodb_resource = boto3.resource("dynamodb") @patch( 'masters_registry.models.ows_territories.get_complement_territories', new=patches.ownership_get_complement_patch) def test_get_unclaimed_territories(): ownership_info = { field_const.TERRITORIES: { 'KZ': {field_const.TUID: 111111}, 'US': {field_const.TUID: 111111}, 'AR': {field_const.TUID: 23223}}, field_const.LOCKED_TERRITORIES: { 'FR': {field_const.REASON: 'owned by WB'}, 'DE': {field_const.REASON: 'owned by Universal'} } } unclaimed_territories = \ masters_registry._get_unclaimed_territories( ownership_info) assert set(('AF', 'AL', 'AX')) == set( unclaimed_territories.message) @patch( 'masters_registry.models.ows_territories.get_complement_territories', new=patches.ownership_get_complement_patch) @patch( 'masters_registry.models.ows_territories.get_territories', new=patches.ows_territories_get_territories_patch) def test_get_unclaimed_territories_empty(): ownership_info = { field_const.TERRITORIES: {}, field_const.LOCKED_TERRITORIES: {} } unclaimed_territories = \ masters_registry._get_unclaimed_territories( ownership_info) assert set(('AF', 'AL', 'AS', 'AX', 'DZ', 'IT', 'PL', 'GB')) == set( unclaimed_territories.message) def test_extract_tuids(): ownership_info = { field_const.TERRITORIES: { 'KZ': {field_const.TUID: 111111}, 'US': {field_const.TUID: 111111}, 'AR': {field_const.TUID: 23223}}} tuids = masters_registry._extract_tuids(ownership_info) assert set([111111, 23223]) == tuids def test_extract_tuids_empty_territories(): ownership_info = {field_const.TERRITORIES: {}} tuids = masters_registry._extract_tuids(ownership_info) assert set() == tuids def test_extract_tuids_internal_conflict(): ownership_info = { field_const.TERRITORIES: { 'KZ': [ {field_const.TUID: 111111}, {field_const.TUID: 222222} ], 'US': {field_const.TUID: 111111}, 'AR': {field_const.TUID: 23223}}} tuids = masters_registry._extract_tuids(ownership_info) assert set([111111, 23223, 222222]) == tuids def test_build_tuid_territories_map(): ownership_info = { field_const.TERRITORIES: { 'KZ': [ {field_const.TUID: 111111}, {field_const.TUID: 222222} ], 'US': {field_const.TUID: 111111}, 'AR': {field_const.TUID: 23223}}} terr_map = masters_registry._build_tuid_territories_map(ownership_info) assert set(['US', 'KZ']) == set(terr_map[111111]) assert set(['AR']) == set(terr_map[23223]) assert set(['KZ']) == set(terr_map[222222]) assert set([111111, 23223, 222222]) == set(terr_map.keys()) def test_build_tuid_territories_map_internal_conflict(): ownership_info = { field_const.TERRITORIES: { 'KZ': {field_const.TUID: 111111}, 'US': {field_const.TUID: 111111}, 'AR': {field_const.TUID: 23223}}} terr_map = masters_registry._build_tuid_territories_map(ownership_info) assert set(['US', 'KZ']) == set(terr_map[111111]) assert set(['AR']) == set(terr_map[23223]) assert set([111111, 23223]) == set(terr_map.keys()) def test_build_tuid_territories_map_empty_territories(): ownership_info = {field_const.TERRITORIES: {}} terr_map = masters_registry._build_tuid_territories_map(ownership_info) assert {} == terr_map def test_build_tuid_territories_map_missing_territory(): ownership_info = {} terr_map = masters_registry._build_tuid_territories_map(ownership_info) assert {} == terr_map def test_extend_ownership_with_track(): ownership_info = {} tuid_territories = { 111111: ['US', 'KZ'], 23223: ['AR']} track_info = { 111111: {field_const.VENDOR_ID: 47}, 23223: {field_const.VENDOR_ID: 4372}} masters_registry._extend_ownership_with_track( ownership_info, track_info, tuid_territories) assert field_const.TRACKS in ownership_info assert 2 == len(ownership_info[field_const.TRACKS]) @patch( 'masters_registry.models.ownership.get_tracks', new=patches.ownership_get_tracks_patch) @patch( 'masters_registry.models.ownership.get_ownership', new=patches.ownership_get_ownership_patch) @patch( 'masters_registry.models.ows_territories.get_complement_territories', new=patches.ownership_get_complement_patch) @mock_aws def test_get_ownership(monkeypatch, feature_engine, setup_new_audit_table): from moto.core import patch_client, patch_resource patch_client(outside_dynamodb_client) patch_resource(dynamodb_resource) setup_new_audit_table(dynamodb_resource.meta.client) get_lock_history = Mock(return_value=True) monkeypatch.setattr( ownership, 'get_lock_history_for_ownership', get_lock_history) test_isrc = 'US4R30920517' ownership_response = masters_registry.get_ownership(test_isrc) assert 200 == ownership_response.status message = ownership_response.message assert field_const.ISRC in message assert field_const.LOCKED_TERRITORIES in message assert field_const.TRACKS in message assert field_const.UNCLAIMED_TERRITORIES in message assert 3 == len(message[field_const.TRACKS]) for item in message[field_const.LOCKED_TERRITORIES]: territories = set(item[field_const.TERRITORIES]) if item[field_const.REASON] == 'Owned by Warner': assert set(['BE', 'NL']) == territories elif item[field_const.REASON] == 'Conflict with Universal': assert set(['RU', 'IT']) == territories @pytest.mark.parametrize('territories,expected_result', [ (['WW'], True), (['US', 'CA'], False) ]) def test_is_worldwide_flag(territories, expected_result): assert masters_registry._is_worldwide_flag(territories) == expected_result def test_group_locked_territories_by_reason(): ownership_info = { field_const.LOCKED_TERRITORIES: { 'IT': {field_const.REASON: 'Conflict with Universal'}, 'BE': {field_const.REASON: 'Owned by Warner'}, 'NL': {field_const.REASON: 'Owned by Warner'}, 'RU': {field_const.REASON: 'Conflict with Universal'}}} masters_registry._group_locked_territories_by_reason(ownership_info) locked_territories = ownership_info[field_const.LOCKED_TERRITORIES] assert 2 == len(locked_territories) @patch( 'masters_registry.models.ownership.get_tracks', new=patches.ownership_get_tracks_patch) def test_is_valid_isrc_tuid_success(): assert masters_registry._is_valid_isrc_tuid('ABC123456', 111111) @patch( 'masters_registry.models.ownership.get_tracks', new=patches.ownership_get_tracks_patch) def test_is_valid_isrc_tuid_fail(): assert not masters_registry._is_valid_isrc_tuid('BADISRC', 111111) @patch( 'masters_registry.models.ownership.get_tracks', new=patches.ownership_get_tracks_patch) def test_is_valid_isrc_tuid_lowercase_success(): assert masters_registry._is_valid_isrc_tuid('abc123456', 111111) @patch( 'masters_registry.models.ows_territories.get_territories', new=patches.ows_territories_get_territories_patch) def test_is_valid_territories_success(): mr_response = masters_registry._is_valid_territories(['AF', 'AX']) assert mr_response @patch.object( ows_territories, 'get_territories', new=Mock(return_value=patches.ows_territories_get_territories_patch(''))) def test_is_valid_territories_correlation_id(): """Verify that correlation_id is being passed to get_territories.""" test_territories = ['AF', 'AX'] test_correlation_id = 'test' masters_registry._is_valid_territories( test_territories, correlation_id=test_correlation_id) ows_territories.get_territories.assert_called_with( field_const.ISO_3166_1_2016, correlation_id=test_correlation_id) @pytest.fixture() def valid_territories(): valid_response = patches.ows_territories_get_territories_patch('') result = [ territory[field_const.TERRITORY_CODE_A2] for territory in valid_response.message['items']] return result @patch( 'masters_registry.models.ows_territories.get_territories', new=patches.ows_territories_get_territories_patch) def test_is_valid_territories_worldwide(valid_territories): mr_response = masters_registry._is_valid_territories(['WW']) assert mr_response assert sorted(mr_response.message) == sorted(valid_territories) @patch( 'masters_registry.models.ows_territories.get_territories', new=patches.ows_territories_get_territories_patch) def test_is_valid_territories_fail(): mr_response = masters_registry._is_valid_territories(['QQ']) assert not mr_response @pytest.fixture def mock_carve_in_territories(): return [ 'AF', 'AX', 'AL', 'DE', 'KZ', 'CH', 'CN', 'JP', 'CA', 'US', 'AS', 'BR', 'AR', 'FR', 'ES', 'IT', 'RU', 'BE', 'NL'] @patch( 'masters_registry.models.ows_carveouts.get_territory_carveout_for_upc', new=patches.ows_carveouts_get_territory_carveout_for_upc) @patch( 'masters_registry.models.ows_territories.convert_territories', new=patches.ows_territories_get_territories_patch) @patch( 'masters_registry.models.ows_territories.get_complement_territories', new=patches.ownership_get_complement_patch) def test_get_carved_in_territories_success(): correlation_id = str(uuid.uuid4()) test_upc = 232323 response = masters_registry._get_carved_in_territories( test_upc, correlation_id) assert response assert response.message == ['AF', 'AX', 'AL'] @patch( 'masters_registry.models.ows_carveouts.get_territory_carveout_for_upc', new=patches.ows_carveouts_get_error_response) def test_get_carved_in_territories_carveout_fail(): correlation_id = str(uuid.uuid4()) test_upc = 232323 response = masters_registry._get_carved_in_territories( test_upc, correlation_id) assert not response assert response.errors['code'] == error.OWS_CARVEOUTS_ERROR @patch( 'masters_registry.models.ows_carveouts.get_territory_carveout_for_upc', new=patches.ows_carveouts_get_territory_carveout_for_upc) @patch( 'masters_registry.models.ows_territories.convert_territories', new=patches.ows_territories_get_error_response) def test_get_carved_in_territories_convert_fail(): correlation_id = str(uuid.uuid4()) test_upc = 232323 response = masters_registry._get_carved_in_territories( test_upc, correlation_id) assert not response assert response.errors['code'] == error.OWS_TERRITORIES_ERROR @patch( 'masters_registry.models.ows_carveouts.get_territory_carveout_for_upc', new=patches.ows_carveouts_get_territory_carveout_for_upc) @patch( 'masters_registry.models.ows_territories.convert_territories', new=patches.ows_territories_get_territories_patch) @patch( 'masters_registry.models.ows_territories.get_complement_territories', new=patches.ows_territories_get_error_response) def test_get_carved_in_territories_complement_fail(): correlation_id = str(uuid.uuid4()) test_upc = 232323 response = masters_registry._get_carved_in_territories( test_upc, correlation_id) assert not response assert response.errors['code'] == error.OWS_TERRITORIES_ERROR @patch('masters_registry.models.ownership.get_ownership', new=patches.ownership_get_ownership_not_found_patch) @patch('masters_registry.models.ownership.create_ownership', new=patches.create_ownership_patch) @patch( 'masters_registry.models.ownership.get_upc_by_tuid', new=patches.ownership_get_upc_by_tuid_patch) @patch( 'masters_registry.models.ows_carveouts.get_dms_carveout_for_upc', new=patches.ows_carveouts_get_dms_carveout_for_upc_not_carved_out_patch) def test_process_isrc_info_does_not_exist_in_registry_ff_off( mock_valid_isrc_info, mock_carve_in_territories, feature_engine): """Test successful import of ISRC that does not exist in registry """ result = masters_registry._process_isrc_info( mock_valid_isrc_info, mock_carve_in_territories, str(uuid.uuid1()), '123') assert result == { field_const.ISRC: '131131131', field_const.SUCCESS: True, field_const.WARNING: False, field_const.RESOLVED_CONFLICT: False, field_const.ERROR_MESSAGE: '', field_const.LOCKED_TERRITORIES: None, field_const.SUBSTORE_CARVEOUTS: [('453', set())], field_const.CLAIMED_BY_ANOTHER_OWNER: [], field_const.CLAIMED_BY_THE_SAME_LABEL: [], } @patch('masters_registry.models.ownership.get_ownership', new=patches.ownership_get_ownership_not_found_patch) @patch('masters_registry.models.ownership.create_ownership', new=patches.create_ownership_patch) @patch( 'masters_registry.models.ownership.get_upc_by_tuid', new=patches.ownership_get_upc_by_tuid_patch) @patch( 'masters_registry.models.ows_carveouts.get_dms_carveout_for_upc', new=patches.ows_carveouts_get_dms_carveout_for_upc_store_carved_out_patch) @pytest.mark.parametrize( 'carveouts_flag, expected_status, expected_error', [(1, True, 'upc_youtube_carved_out')]) def test_process_isrc_info_does_not_exist_in_registry_dms_carveout_ff_off( mock_valid_isrc_info, mock_carve_in_territories, carveouts_flag, expected_status, expected_error, feature_engine): """Test successful import of ISRC that does not exist in registry UPC has DMS carveouts """ result = masters_registry._process_isrc_info( mock_valid_isrc_info, mock_carve_in_territories, str(uuid.uuid1()), '123') assert result == { field_const.ISRC: '131131131', field_const.SUCCESS: expected_status, field_const.WARNING: False, field_const.RESOLVED_CONFLICT: False, field_const.ERROR_MESSAGE: expected_error, field_const.LOCKED_TERRITORIES: None, field_const.SUBSTORE_CARVEOUTS: [], field_const.CLAIMED_BY_ANOTHER_OWNER: [], field_const.CLAIMED_BY_THE_SAME_LABEL: [], } @patch('masters_registry.models.ownership.get_ownership', new=patches.ownership_get_ownership_not_found_patch) @patch('masters_registry.models.ownership.create_ownership', new=patches.create_ownership_patch) @patch( 'masters_registry.models.ownership.get_upc_by_tuid', new=patches.ownership_get_upc_by_tuid_patch) @patch( 'masters_registry.models.ows_carveouts.get_dms_carveout_for_upc', new=patches.ows_carveouts_get_dms_carveout_for_upc_store_carved_out_patch) @pytest.mark.parametrize( 'carveouts_flag, expected_status, expected_error', [(1, True, 'upc_youtube_carved_out')]) def test_process_isrc_info_does_not_exist_in_registry_dms_carveout( mock_valid_isrc_info, mock_carve_in_territories, carveouts_flag, expected_status, expected_error, feature_engine): """Test successful import of ISRC that does not exist in registry UPC has DMS carveouts """ result = masters_registry._process_isrc_info( mock_valid_isrc_info, mock_carve_in_territories, str(uuid.uuid1()), '123') assert result == { field_const.ISRC: '131131131', field_const.SUCCESS: expected_status, field_const.WARNING: False, field_const.RESOLVED_CONFLICT: False, field_const.ERROR_MESSAGE: expected_error, field_const.LOCKED_TERRITORIES: None, field_const.SUBSTORE_CARVEOUTS: [], field_const.CLAIMED_BY_ANOTHER_OWNER: [], field_const.CLAIMED_BY_THE_SAME_LABEL: [], } @patch( 'masters_registry.models.ownership.get_ownership', new=patches.ownership_get_ownership_not_found_patch) @patch( 'masters_registry.models.ownership.create_ownership', new=patches.create_ownership_patch) @patch( 'masters_registry.models.ownership.get_upc_by_tuid', new=patches.ownership_get_upc_by_tuid_patch) @patch( 'masters_registry.models.ows_carveouts.get_dms_carveout_for_upc', new=patches.ows_carveouts_get_dms_carveout_for_upc_substore_patch) @patch( 'masters_registry.models.ows_territories.convert_territories', new=patches.ows_territories_get_territories_patch) @pytest.mark.parametrize( 'carveouts_flag, expected_status, substore_carveouts', [(1, False, [('453', {'AL', 'GB', 'AX', 'IT', 'PL', 'DZ', 'AF', 'AS'})])]) def test_process_isrc_info_does_not_exist_in_registry_substore_carveout( mock_valid_isrc_info, mock_carve_in_territories, carveouts_flag, expected_status, substore_carveouts, feature_engine): """Test successful import of ISRC that does not exist in registry UPC has substore carveouts """ result = masters_registry._process_isrc_info( mock_valid_isrc_info, mock_carve_in_territories, str(uuid.uuid1()), '123') assert result == { field_const.ISRC: '131131131', field_const.SUCCESS: expected_status, field_const.WARNING: False, field_const.RESOLVED_CONFLICT: False, field_const.ERROR_MESSAGE: '', field_const.LOCKED_TERRITORIES: None, field_const.SUBSTORE_CARVEOUTS: substore_carveouts, field_const.CLAIMED_BY_ANOTHER_OWNER: [], field_const.CLAIMED_BY_THE_SAME_LABEL: [], } @patch('masters_registry.models.ownership.get_ownership', new=patches.ownership_get_ownership_not_found_patch) @patch('masters_registry.models.ownership.create_ownership', new=patches.create_ownership_patch) def test_process_isrc_info_does_not_exist_in_registry( mock_valid_isrc_info, mock_carve_in_territories, feature_engine): """Test successful import of ISRC that does not exist in registry """ result = masters_registry._process_isrc_info( mock_valid_isrc_info, mock_carve_in_territories, str(uuid.uuid1()), '123') assert result == { field_const.ISRC: '131131131', field_const.SUCCESS: True, field_const.WARNING: False, field_const.RESOLVED_CONFLICT: False, field_const.ERROR_MESSAGE: '', field_const.LOCKED_TERRITORIES: None, field_const.SUBSTORE_CARVEOUTS: [], field_const.CLAIMED_BY_ANOTHER_OWNER: [], field_const.CLAIMED_BY_THE_SAME_LABEL: [], } @patch('masters_registry.models.ownership.get_ownership', new=patches.ownership_get_ownership_not_found_patch) @patch('masters_registry.models.ownership.create_ownership', new=patches.ownership_update_ownership_failure_patch) def test_process_isrc_info_does_not_exist_in_registry_import_fail( mock_valid_isrc_info, mock_carve_in_territories, feature_engine): """Test failed import of ISRC that does not exist in registry """ result = masters_registry._process_isrc_info( mock_valid_isrc_info, mock_carve_in_territories, str(uuid.uuid1()), '123') assert result == { field_const.ISRC: '131131131', field_const.SUCCESS: False, field_const.WARNING: False, field_const.RESOLVED_CONFLICT: False, field_const.ERROR_MESSAGE: '', field_const.LOCKED_TERRITORIES: None, field_const.SUBSTORE_CARVEOUTS: [], field_const.CLAIMED_BY_ANOTHER_OWNER: [], field_const.CLAIMED_BY_THE_SAME_LABEL: [], } @patch( 'masters_registry.models.ows_carveouts.get_territory_carveout_for_upc', new=patches.ows_carveouts_get_territory_carveout_for_upc) @patch( 'masters_registry.models.ows_territories.convert_territories', new=patches.ows_territories_get_territories_patch) @patch( 'masters_registry.models.ows_territories.get_complement_territories', new=patches.ownership_get_complement_patch) @patch('masters_registry.models.ownership.get_ownership', new=patches.ownership_get_ownership_not_found_patch) @patch('masters_registry.models.ownership.create_ownership', new=patches.create_ownership_patch) def test_process_upc(mock_valid_isrc_info, feature_engine): """Test successful import of UPC into registry """ result = masters_registry._process_upc( 'UPC22332233', [mock_valid_isrc_info], str(uuid.uuid1()), '123') assert result assert result.message['error_count'] == 0 assert result.message['success_count'] == 1 assert result.message['status_report'][0]['success'] @patch( 'masters_registry.models.ows_carveouts.get_territory_carveout_for_upc', new=patches.ows_carveouts_get_error_response) def test_process_upc_carvein_territories_error(mock_valid_isrc_info): """Test failed import of UPC into registry """ result = masters_registry._process_upc( 'UPC22332233', [mock_valid_isrc_info], str(uuid.uuid1()), '123') assert not result assert result.message['error_count'] == 1 assert result.message['success_count'] == 0 @patch( 'masters_registry.models.ownership.get_tracks', new=patches.ownership_get_tracks_patch) def test_calculate_differences(feature_engine): """Test calculate differences between data in registry and submitted data """ ownership_patch = patches.ownership_get_ownership_patch('').message tuid = 111111 user = '123' locked = ['BE', 'NL'] existing = ['AF', 'AS', 'AR'] missing = ['DZ', 'AX', 'AR'] removed = [ 'BR', 'CA', 'CH', 'CN', 'DE', 'ES', 'FR', 'JP', 'KZ', 'US'] territories = set(locked + existing + missing) difference = masters_registry._calculate_difference( ownership_patch, territories, tuid, user) assert sorted(difference[field_const.EXISTING]) == sorted(existing) assert sorted(difference[field_const.LOCKED]) == sorted(locked) assert sorted(difference[field_const.MISSING]) == sorted(missing) assert sorted(difference[field_const.REMOVED]) == sorted(removed) @patch( 'masters_registry.models.ownership.get_tracks', new=patches.ownership_get_tracks_patch) def test_calculate_differences_includes_claimed_by_another_owner( feature_engine): """Test territories owned by another owner are included in result.""" ownership_response = ( patches.ownership_get_ownership_claimed_by_another_owner_patch()) ownership_patch = ownership_response.message claimed_by_another_owner = [('UA', [3333])] ownership_patch[field_const.TERRITORIES]['UA'] = {field_const.TUID: 3333} tuid = 111111 user = '123' difference = masters_registry._calculate_difference( ownership_patch, ownership_patch[field_const.TERRITORIES].keys(), tuid, user) assert (difference.get(field_const.CLAIMED_BY_ANOTHER_OWNER) == claimed_by_another_owner) @patch( 'masters_registry.models.ownership.get_tracks', new=patches.ownership_get_tracks_patch) def test_calculate_differences_includes_claimed_by_the_same_label( feature_engine): """Test if territories claimed by the same label are included in result.""" ownership_response = ( patches.ownership_get_ownership_patch('')) ownership_patch = ownership_response.message tuid = 11 user = '123' difference = masters_registry._calculate_difference( ownership_patch, ['CA', 'IT', 'UK', 'US'], tuid, user) assert (set(difference.get(field_const.CLAIMED_BY_THE_SAME_LABEL)) == {'CA', 'US', }) @patch( 'masters_registry.models.ownership.get_tracks', new=patches.ownership_get_tracks_patch) @patch('masters_registry.models.yt_ownership.send_message', new=MagicMock()) @patch( 'masters_registry.models.ownership.get_upc_by_tuid', new=patches.ownership_get_upc_by_tuid_patch) @patch( 'masters_registry.models.ows_carveouts.get_dms_carveout_for_upc', new=patches.ows_carveouts_get_dms_carveout_for_upc_not_carved_out_patch) @mock_aws def test_process_isrc_info_internal_conflict( mock_valid_isrc_info, setup_masters_active_table, seed_masters_active_table, setup_new_audit_table): """Test territories claimed by another owner.""" from moto.core import patch_client, patch_resource patch_client(outside_dynamodb_client) patch_resource(dynamodb_resource) tuid = 321 correlation_id = str(uuid.uuid1()) isrc = '131131131' user = '123' ownership_info = { field_const.ISRC: isrc, field_const.LOCKED_TERRITORIES: {}, field_const.TERRITORIES: { 'QA': [{field_const.TUID: tuid}], 'CA': {field_const.TUID: tuid}, } } setup_masters_active_table(dynamodb_resource.meta.client) setup_new_audit_table(dynamodb_resource.meta.client) seed_masters_active_table(dynamodb_resource.meta.client, item=ownership_info) ignore_fields = ['timestamp', 'updated_timestamp'] isrc_info = dict(mock_valid_isrc_info, tuid=12345) result = masters_registry._process_isrc_info( isrc_info, ['QA', 'CA'], correlation_id, user) assert result == { field_const.ISRC: mock_valid_isrc_info[field_const.ISRC], field_const.SUCCESS: True, field_const.WARNING: True, field_const.RESOLVED_CONFLICT: False, field_const.ERROR_MESSAGE: '', field_const.LOCKED_TERRITORIES: None, field_const.CLAIMED_BY_ANOTHER_OWNER: [('QA', [tuid]), ('CA', [tuid])], field_const.SUBSTORE_CARVEOUTS: [], field_const.CLAIMED_BY_THE_SAME_LABEL: [], } expected_ownership_record = { 'isrc': isrc, 'locked_territories': {}, 'territories': { 'QA': [{'tuid': Decimal('321')}, {'tuid': Decimal('12345')}], 'CA': [{'tuid': Decimal('321')}, {'tuid': Decimal('12345')}], } } ownership_record = ownership.get_ownership(isrc).message assert test_utils.equal_dicts( expected_ownership_record, ownership_record, ignore_fields) expected_audit_records = [ { 'opcode': opcode_const.ADD, 'territories': {'QA': Decimal('12345'), 'CA': Decimal('12345')}, 'user': user, 'isrc': isrc, 'correlation_id': correlation_id, }, { 'opcode': opcode_const.CONFLICT_CREATED, 'territories': {'QA': Decimal('12345'), 'CA': Decimal('12345')}, 'source': 'BULK_UPDATE_UPC', 'conflict': { 'status': 'CONFLICT_CREATED', 'conflicting_tuid': [Decimal('321'), Decimal('12345')] }, 'user': user, 'isrc': isrc, 'correlation_id': correlation_id, } ] audit_records = ownership.get_ownership_audit(isrc).message for i, record in enumerate(audit_records): assert test_utils.equal_dicts( expected_audit_records[i], record, ignore_fields) @patch( 'masters_registry.models.ows_carveouts.get_territory_carveout_for_upc', new=patches.ows_carveouts_get_territory_carveout_for_upc) @patch( 'masters_registry.models.ows_territories.convert_territories', new=patches.ows_territories_get_territories_patch) @patch( 'masters_registry.models.ows_territories.get_complement_territories', new=patches.ownership_get_complement_patch) @patch('masters_registry.models.ownership.get_ownership', new=patches.ownership_get_ownership_not_found_patch) @patch('masters_registry.models.ownership.create_ownership', new=patches.create_ownership_patch) @patch( 'masters_registry.models.users.get_orchard_user_names', new=Mock(return_value=False)) def test_bulk_import_upcs_success( feature_engine, expected_response, bulk_status_db_fixture, monkeypatch): """Test successful import bulk_import_upcs """ upcs = ['123456789120', '123456789121'] correlation_id = str(uuid.uuid1()) user = '1234' task_id = 1 bulk_import_upcs = Mock() monkeypatch.setattr(bulk.bulk_import_upcs, 'delay', bulk_import_upcs) patches.mock_session_scope(expected_response) masters_registry.bulk_import_upcs(upcs, correlation_id, user) bulk_import_upcs.assert_called_once_with( upcs, task_id, correlation_id, user ) @patch( 'masters_registry.models.ows_territories.get_territories', new=patches.ows_territories_get_territories_patch) def test_add_tuid_for_territory_invalid_territories(): """Test add_tuid_for_territory with invalid territories codes """ tuid = 123 isrc = 'US1' territories = ['AF', 'XX'] correlation_id = str(uuid.uuid1()) user = '1234' result = masters_registry.add_tuid_for_territory( tuid, isrc, territories, correlation_id, user) assert result.errors['code'] == error.INVALID_TERRITORIES @patch( 'masters_registry.models.ownership.get_track', new=patches.ownership_get_track) @patch( 'masters_registry.models.ows_territories.get_territories', new=patches.ows_territories_get_territories_patch) def test_add_tuid_for_territory_not_existing_tuid(): """Test add_tuid_for_territory with not existing tuid """ tuid = 999 isrc = 'US1' territories = ['AF', 'AX'] correlation_id = str(uuid.uuid1()) user = '1234' result = masters_registry.add_tuid_for_territory( tuid, isrc, territories, correlation_id, user) assert result.errors['code'] == error.INVALID_TUID @patch( 'masters_registry.models.ownership.get_track', new=patches.ownership_get_track) @patch( 'masters_registry.models.ows_territories.get_territories', new=patches.ows_territories_get_territories_patch) def test_add_tuid_for_territory_mismatch_tuid(): """Test add_tuid_for_territory with tuid that does not matches ISRC """ tuid = 222 isrc = 'US2' territories = ['AF', 'AX'] correlation_id = str(uuid.uuid1()) user = '1234' result = masters_registry.add_tuid_for_territory( tuid, isrc, territories, correlation_id, user) assert result.errors['code'] == error.INVALID_TUID_ISRC @patch( 'masters_registry.models.ows_territories.convert_territories', new=patches.ows_territories_convert_territories_patch) @patch( 'masters_registry.models.ownership.get_ownership', new=patches.ownership_get_ownership_patch) @patch( 'masters_registry.models.ownership.get_track', new=patches.ownership_get_track) @patch( 'masters_registry.models.ows_territories.get_territories', new=patches.ows_territories_get_territories_patch) def test_add_tuid_for_territory_claimed_tuid(): """Test add_tuid_for_territory with tuid that is already claimed """ tuid = 111111 isrc = 'ABC123456' territories = ['AF', 'AX'] correlation_id = str(uuid.uuid1()) user = '1234' result = masters_registry.add_tuid_for_territory( tuid, isrc, territories, correlation_id, user) assert result.errors['code'] == error.TUID_CLAIMED @patch( 'masters_registry.models.ows_territories.convert_territories', new=patches.ows_territories_convert_territories_patch) @patch( 'masters_registry.models.ownership.get_ownership', new=patches.ownership_get_ownership_patch) @patch( 'masters_registry.models.ownership.get_track', new=patches.ownership_get_track) @patch( 'masters_registry.models.ows_territories.get_territories', new=patches.ows_territories_get_territories_patch) def test_add_tuid_for_territory_locked(): """Test add_tuid_for_territory with claimed territories """ tuid = 222 isrc = 'US1' territories = ['IT'] correlation_id = str(uuid.uuid1()) user = '1234' result = masters_registry.add_tuid_for_territory( tuid, isrc, territories, correlation_id, user) assert result.errors['code'] == error.INVALID_TERRITORIES assert result.errors['message'] == 'Territory IT is already locked.' @patch( 'masters_registry.models.ows_territories.convert_territories', new=patches.ows_territories_convert_territories_patch) @patch( 'masters_registry.models.ownership.get_ownership', new=patches.ownership_get_ownership_patch) @patch( 'masters_registry.models.ownership.get_track', new=patches.ownership_get_track) @patch( 'masters_registry.models.ows_territories.get_territories', new=patches.ows_territories_get_territories_patch) @patch( 'masters_registry.models.ownership.get_tracks', new=patches.ownership_get_tracks_patch) def test_add_tuid_for_territory_claimed_by_the_same_label(): """Test add_tuid_for_territory with territories that are claimed by the same label """ tuid = 11 isrc = 'ZW1010600008' territories = ['AF', 'AS', 'DZ'] correlation_id = str(uuid.uuid1()) user = '1234' result = masters_registry.add_tuid_for_territory( tuid, isrc, territories, correlation_id, user) assert 400 == result.status errors = result.errors assert error.TERRITORIES_CLAIMED_BY_THE_SAME_LABEL == errors['code'] message = errors['message'] assert set(message['territories']) == set(['AF', 'AS']) @patch( 'masters_registry.models.ownership.get_ownership', new=patches.ownership_get_ownership_patch) @patch( 'masters_registry.models.ownership.get_track', new=patches.ownership_get_track) @patch( 'masters_registry.models.ows_territories.get_territories', new=patches.ows_territories_get_territories_patch) @patch( 'masters_registry.models.ownership.get_upc_by_tuid', new=patches.ownership_get_upc_by_tuid_patch) @patch( 'masters_registry.models.ows_carveouts.get_dms_carveout_for_upc', new=patches.ows_carveouts_get_dms_carveout_for_upc_not_carved_out_patch) @patch( 'masters_registry.models.ownership.get_tracks', new=patches.ownership_get_tracks_patch) def test_add_tuid_for_territory_valid_data_internal_conflict(monkeypatch): """Test add_tuid_for_territory with valid data """ tuid = 222 isrc = 'US1' territories = ['AF'] correlation_id = str(uuid.uuid1()) user = '1234' update_ownership = Mock() monkeypatch.setattr( ownership_logic, 'update_ownership', update_ownership) send_message = Mock() monkeypatch.setattr(yt_ownership, 'send_message', send_message) ownership_info = patches.ownership_get_ownership_patch('').message claimed = list(ownership_info[field_const.TERRITORIES].keys()) claimed.extend(territories) track_info = patches.ownership_get_track(tuid).message expected_track = { field_const.TUID: tuid, field_const.VENDOR_NAME: track_info[field_const.VENDOR_NAME], field_const.VENDOR_ID: track_info[field_const.VENDOR_ID], field_const.ARTIST_NAME: track_info[field_const.ARTIST_NAME], field_const.TRACK_NAME: track_info[field_const.TRACK_NAME], field_const.UPC: track_info[field_const.UPC], field_const.RELEASE_NAME: track_info[field_const.RELEASE_NAME], field_const.TERRITORIES: territories } result = masters_registry.add_tuid_for_territory( tuid, isrc, territories, correlation_id, user) send_message_args = send_message.call_args[0] assert result.message == expected_track update_ownership.assert_called_once_with( isrc, tuid, territories, ownership_info[field_const.TERRITORIES], correlation_id, user) send_message.assert_called_once_with(isrc, ANY, correlation_id) assert set(send_message_args[1]) == set(claimed) def test_generate_processing_result_success(): """Assert ISRC processing result is correct when all results are successful """ result_one = response.Response() result_two = response.Response() calculated_result = masters_registry._generate_processing_result([ result_one, result_two]) assert calculated_result def test_generate_processing_result_all_failed(): """Assert ISRC processing result is correct when all results failed """ result_one = response.Response(message='Failure one', status=400) result_two = response.Response(message='Failure two', status=500) calculated_result = masters_registry._generate_processing_result([ result_one, result_two]) assert not calculated_result assert calculated_result.message == 'Failure one;Failure two;' def test_generate_processing_result_some_failed(): """Assert ISRC processing result is correct when some results failed """ result_one = response.Response() result_two = response.Response(message='Failure two', status=500) calculated_result = masters_registry._generate_processing_result([ result_one, result_two]) assert not calculated_result assert calculated_result.message == 'Failure two;' @patch.object( masters_registry, '_get_unique_tuid_from_ownership_data', new=Mock(return_value=[1])) @patch.object( masters_registry, '_pick_territories_by_tuid', new=Mock(return_value=[1])) @patch.object( masters_registry.dms_carveout, 'tuid_check_dms_carveout', new=Mock(return_value=response.create_fatal_response('test'))) @patch.object(masters_registry.celery_log, 'error', new=Mock()) def test_update_youtube_with_carveout_failed_to_get_dms_carveout(): """Should get error when failed to get DMS carveout.""" isrc_territories = {'AF': [{'tuid': 1}]} result = masters_registry._update_youtube_with_carveout( isrc='QA123', tuid=1, territories=isrc_territories, correlation_id='test correlation') assert result.status == 400 assert result.message == error.FAILED_TO_GET_DMS_CARVEOUTS get_unique_tuid_from_ownership_data = ( masters_registry._get_unique_tuid_from_ownership_data) get_unique_tuid_from_ownership_data.assert_called_once_with( isrc_territories) assert masters_registry.celery_log.error.call_count == 1 @patch('masters_registry.models.ownership.get_isrcs', new=patches.ownership_get_ownership_not_found_patch) def test_bulk_import_upc_failed_validation(): """Should return validation error""" result = masters_registry.bulk_import_upcs( upcs=['1234'], correlation_id='test', user='oa:123') expected_error = [{ 'upc': '1234', 'status_report': [], 'success_count': 0, 'error_count': 1, 'error_message': 'UPC does not exist' }] assert isinstance(result, response.Response) assert result.status == 400 assert result.message == expected_error @mock_aws @patch.object( masters_registry.dms_carveout, 'tuid_check_dms_carveout', new=Mock(return_value=response.create_fatal_response('test'))) def test_process_isrc_info_carveouts_failed( feature_engine, setup_masters_active_table, setup_new_audit_table, valid_correlation_id): """Should only add ISRC to registry. Failed carveouts check should be ignored. """ from moto.core import patch_client, patch_resource patch_client(outside_dynamodb_client) patch_resource(dynamodb_resource) setup_masters_active_table(dynamodb_resource.meta.client) setup_new_audit_table(dynamodb_resource.meta.client) processing_response = masters_registry._process_isrc_info( {'isrc': 'QA123', 'tuid': 123}, ['AF'], valid_correlation_id, 'oa:123') expected_response = { 'warning': False, 'claimed_by_the_same_label': [], 'locked_territories': None, 'substore_carveouts': [], 'resolved_conflict': False, 'error_message': '', 'success': True, 'claimed_by_another_owner': [], 'isrc': 'QA123' } assert processing_response == expected_response created_ownership = ownership.get_ownership('QA123').message expected_entry = { 'isrc': 'QA123', 'territories': {'AF': [{'tuid': Decimal('123')}]}, 'timestamp': ANY, 'locked_territories': {} } assert created_ownership == expected_entry def carveouts_info_response(): """Result of check for carveouts when there are no carveouts.""" carveouts_info = { 'allowed_territories': ['AF'], 'substore_carveouts': (api_const.YOUTUBE_DMS, set()) } return response.Response(carveouts_info) @mock_aws @patch.object( masters_registry.dms_carveout, 'tuid_check_dms_carveout', new=Mock(return_value=carveouts_info_response())) def test_process_isrc_info_carveouts_exist( feature_engine, setup_masters_active_table, setup_new_audit_table, valid_correlation_id): """Should add ISRC to registry and check for carveouts.""" from moto.core import patch_client, patch_resource patch_client(outside_dynamodb_client) patch_resource(dynamodb_resource) setup_masters_active_table(dynamodb_resource.meta.client) setup_new_audit_table(dynamodb_resource.meta.client) result = masters_registry._process_isrc_info( {'isrc': 'QA123', 'tuid': 123}, ['AF'], valid_correlation_id, 'oa:123') expected = { 'warning': False, 'claimed_by_the_same_label': [], 'locked_territories': None, 'resolved_conflict': False, 'substore_carveouts': [(api_const.YOUTUBE_DMS, set())], 'error_message': '', 'success': True, 'claimed_by_another_owner': [], 'isrc': 'QA123' } assert result == expected created_ownership = ownership.get_ownership('QA123').message expected_entry = { 'isrc': 'QA123', 'territories': {'AF': [{'tuid': Decimal('123')}]}, 'timestamp': ANY, 'locked_territories': {} } assert created_ownership == expected_entry def carveouts_info_response_with_store_carveouts(): """Result of check for carveouts when there are store carveouts.""" carveouts_info = { 'allowed_territories': [], 'substore_carveouts': () } return response.Response(carveouts_info) @mock_aws @patch.object( masters_registry.dms_carveout, 'tuid_check_dms_carveout', new=Mock(return_value=carveouts_info_response_with_store_carveouts())) def test_process_isrc_info_store_carveouts( feature_engine, setup_masters_active_table, setup_new_audit_table, valid_correlation_id): """Should add ISRC to registry and return info about store carveouts.""" from moto.core import patch_client, patch_resource patch_client(outside_dynamodb_client) patch_resource(dynamodb_resource) setup_masters_active_table(dynamodb_resource.meta.client) setup_new_audit_table(dynamodb_resource.meta.client) result = masters_registry._process_isrc_info( {'isrc': 'QA123', 'tuid': 123}, ['AF'], valid_correlation_id, 'oa:123') expected = { 'warning': False, 'claimed_by_the_same_label': [], 'locked_territories': None, 'resolved_conflict': False, 'substore_carveouts': [], 'error_message': 'upc_youtube_carved_out', 'success': True, 'claimed_by_another_owner': [], 'isrc': 'QA123' } assert result == expected created_ownership = ownership.get_ownership('QA123').message expected_entry = { 'isrc': 'QA123', 'territories': {'AF': [{'tuid': Decimal('123')}]}, 'timestamp': ANY, 'locked_territories': {} } assert created_ownership == expected_entry def carveouts_info_response_with_substore_carveouts(): """Result of check for carveouts when there are substore carveouts.""" carveouts_info = { 'allowed_territories': ['AF'], 'substore_carveouts': (api_const.YOUTUBE_DMS, {'AF'}) } return response.Response(carveouts_info) @mock_aws @patch.object( masters_registry.dms_carveout, 'tuid_check_dms_carveout', new=Mock(return_value=carveouts_info_response_with_substore_carveouts())) def test_process_isrc_info_substore_carveouts( setup_masters_active_table, setup_new_audit_table, valid_correlation_id): """Should add ISRC to registry and return info about substore carveouts.""" from moto.core import patch_client, patch_resource patch_client(outside_dynamodb_client) patch_resource(dynamodb_resource) setup_masters_active_table(dynamodb_resource.meta.client) setup_new_audit_table(dynamodb_resource.meta.client) result = masters_registry._process_isrc_info( {'isrc': 'QA123', 'tuid': 123}, ['AF'], valid_correlation_id, 'oa:123') expected = { 'warning': False, 'claimed_by_the_same_label': [], 'locked_territories': None, 'resolved_conflict': False, 'substore_carveouts': [(api_const.YOUTUBE_DMS, {'AF'})], 'error_message': '', 'success': False, 'claimed_by_another_owner': [], 'isrc': 'QA123' } assert result == expected created_ownership = ownership.get_ownership('QA123').message expected_entry = { 'isrc': 'QA123', 'territories': {'AF': [{'tuid': Decimal('123')}]}, 'timestamp': ANY, 'locked_territories': {} } assert created_ownership == expected_entry def test_get_processing_response_template(): """Should return correct template.""" template = masters_registry._get_processing_response_template('QA123') expected_template = { 'success': False, 'warning': False, 'substore_carveouts': [], 'resolved_conflict': False, 'error_message': '', 'claimed_by_the_same_label': [], 'claimed_by_another_owner': [], 'locked_territories': None, 'isrc': 'QA123' } assert template == expected_template @mock_aws @patch.object( masters_registry, '_calculate_difference', new=Mock(return_value={'locked': []})) @patch.object( masters_registry, '_update_youtube_with_carveout', new=Mock(return_value=response.Response())) def test_process_isrc_info_with_update_no_differences_in_ownership( feature_engine, setup_masters_active_table, seed_masters_active_table, setup_new_audit_table, valid_correlation_id): """Should update youtube and return processing response.""" from moto.core import patch_client, patch_resource patch_client(outside_dynamodb_client) patch_resource(dynamodb_resource) setup_masters_active_table(dynamodb_resource.meta.client) seed_masters_active_table(dynamodb_resource.meta.client) setup_new_audit_table(dynamodb_resource.meta.client) masters_registry._update_youtube_with_carveout.reset_mock() result = masters_registry._process_isrc_info( {'isrc': 'QA123', 'tuid': 1}, ['AF'], valid_correlation_id, 'oa:123') expected = { 'warning': False, 'success': True, 'claimed_by_another_owner': [], 'isrc': 'QA123', 'claimed_by_the_same_label': [], 'resolved_conflict': False, 'locked_territories': None, 'substore_carveouts': [], 'error_message': '' } assert result == expected masters_registry._update_youtube_with_carveout.assert_called_once_with( 'QA123', 1, {'AF': [{'tuid': Decimal('1')}]}, valid_correlation_id) @mock_aws @patch.object( masters_registry, '_calculate_difference', new=Mock(return_value={'locked': ['AR']})) @patch.object( masters_registry, '_update_youtube_with_carveout', new=Mock(return_value=response.Response())) def test_process_isrc_info_with_update_locked_territories( feature_engine, setup_masters_active_table, seed_masters_active_table, setup_new_audit_table, valid_correlation_id): """Should update youtube and return processing response. Response should not be success because there is locked territory. """ from moto.core import patch_client, patch_resource patch_client(outside_dynamodb_client) patch_resource(dynamodb_resource) setup_masters_active_table(dynamodb_resource.meta.client) seed_masters_active_table(dynamodb_resource.meta.client) setup_new_audit_table(dynamodb_resource.meta.client) masters_registry._update_youtube_with_carveout.reset_mock() result = masters_registry._process_isrc_info( {'isrc': 'QA123', 'tuid': 1}, ['AF'], valid_correlation_id, 'oa:123') expected = { 'warning': False, 'success': True, 'claimed_by_another_owner': [], 'isrc': 'QA123', 'claimed_by_the_same_label': [], 'resolved_conflict': False, 'locked_territories': ['AR'], 'substore_carveouts': [], 'error_message': '' } assert result == expected masters_registry._update_youtube_with_carveout.assert_called_once_with( 'QA123', 1, {'AF': [{'tuid': Decimal('1')}]}, valid_correlation_id) @mock_aws @patch.object( masters_registry, '_calculate_difference', new=Mock(return_value={'missing': ['AR'], 'removed': [], 'locked': []})) @patch.object( masters_registry, '_update_youtube_with_carveout', new=Mock(return_value=response.Response())) def test_process_isrc_info_with_update_missing_territories( setup_masters_active_table, seed_masters_active_table, setup_new_audit_table, valid_correlation_id): """Should add missing territory and update youtube.""" from moto.core import patch_client, patch_resource patch_client(outside_dynamodb_client) patch_resource(dynamodb_resource) setup_masters_active_table(dynamodb_resource.meta.client) seed_masters_active_table(dynamodb_resource.meta.client) setup_new_audit_table(dynamodb_resource.meta.client) masters_registry._update_youtube_with_carveout.reset_mock() result = masters_registry._process_isrc_info( {'isrc': 'QA123', 'tuid': 1}, ['AF', 'AR'], valid_correlation_id, 'oa:123') expected = { 'warning': False, 'success': True, 'claimed_by_another_owner': [], 'isrc': 'QA123', 'claimed_by_the_same_label': [], 'resolved_conflict': False, 'locked_territories': None, 'substore_carveouts': [], 'error_message': '' } assert result == expected masters_registry._update_youtube_with_carveout.assert_called_once_with( 'QA123', 1, {'AF': [{'tuid': Decimal('1')}], 'AR': [{'tuid': Decimal('1')}]}, valid_correlation_id) updated_ownership = ownership.get_ownership('QA123').message expected_entry = { 'isrc': 'QA123', 'territories': {'AF': [{'tuid': 1}], 'AR': [{'tuid': 1}]}, 'timestamp': ANY, 'updated_timestamp': ANY, 'locked_territories': {} } assert updated_ownership == expected_entry @mock_aws @patch.object( masters_registry, '_calculate_difference', new=Mock(return_value={'missing': [], 'removed': ['AR'], 'locked': []})) @patch.object( masters_registry, '_update_youtube_with_carveout', new=Mock(return_value=response.Response())) def test_process_isrc_info_with_update_remove_territories( setup_masters_active_table, seed_masters_active_table, setup_new_audit_table, valid_correlation_id): """Should remove not carved in territory and update youtube.""" from moto.core import patch_client, patch_resource patch_client(outside_dynamodb_client) patch_resource(dynamodb_resource) setup_masters_active_table(dynamodb_resource.meta.client) seed_masters_active_table(dynamodb_resource.meta.client) setup_new_audit_table(dynamodb_resource.meta.client) masters_registry._update_youtube_with_carveout.reset_mock() result = masters_registry._process_isrc_info( {'isrc': 'QA123', 'tuid': 1}, ['AF', 'AR'], valid_correlation_id, 'oa:123') expected = { 'warning': False, 'success': True, 'claimed_by_another_owner': [], 'isrc': 'QA123', 'claimed_by_the_same_label': [], 'resolved_conflict': False, 'locked_territories': None, 'substore_carveouts': [], 'error_message': '' } assert result == expected masters_registry._update_youtube_with_carveout.assert_called_once_with( 'QA123', 1, {'AF': [{'tuid': Decimal('1')}]}, valid_correlation_id) updated_ownership = ownership.get_ownership('QA123').message expected_entry = { 'isrc': 'QA123', 'territories': {'AF': [{'tuid': 1}]}, 'timestamp': ANY, 'updated_timestamp': ANY, 'locked_territories': {} } assert updated_ownership == expected_entry @mock_aws @patch.object( masters_registry, '_calculate_difference', new=Mock( return_value={'locked': [], 'claimed_by_another_owner': [('AF', 2)]})) @patch.object( masters_registry, '_update_youtube_with_carveout', new=Mock(return_value=response.Response())) def test_process_isrc_info_with_update_claimed_by_another_owner( setup_masters_active_table, seed_masters_active_table, setup_new_audit_table, valid_correlation_id): """Should update youtube and return processing response. Expect to account for territories claimed by another owner. """ from moto.core import patch_client, patch_resource patch_client(outside_dynamodb_client) patch_resource(dynamodb_resource) setup_masters_active_table(dynamodb_resource.meta.client) seed_masters_active_table(dynamodb_resource.meta.client) setup_new_audit_table(dynamodb_resource.meta.client) masters_registry._update_youtube_with_carveout.reset_mock() result = masters_registry._process_isrc_info( {'isrc': 'QA123', 'tuid': 1}, ['AF'], valid_correlation_id, 'oa:123') expected = { 'warning': True, 'success': True, 'claimed_by_another_owner': [('AF', 2)], 'isrc': 'QA123', 'claimed_by_the_same_label': [], 'resolved_conflict': False, 'locked_territories': None, 'substore_carveouts': [], 'error_message': '' } assert result == expected masters_registry._update_youtube_with_carveout.assert_called_once_with( 'QA123', 1, {'AF': [{'tuid': Decimal('1')}]}, valid_correlation_id) @mock_aws @patch.object( masters_registry, '_calculate_difference', new=Mock( return_value={'locked': [], 'claimed_by_the_same_label': ['AF']})) @patch.object( masters_registry, '_update_youtube_with_carveout', new=Mock(return_value=response.Response())) def test_process_isrc_info_claimed_by_same_label( setup_masters_active_table, seed_masters_active_table, setup_new_audit_table, valid_correlation_id): """Should update youtube and return processing response. Expect to account for territories claimed by the same label. """ from moto.core import patch_client, patch_resource patch_client(outside_dynamodb_client) patch_resource(dynamodb_resource) setup_masters_active_table(dynamodb_resource.meta.client) seed_masters_active_table(dynamodb_resource.meta.client) setup_new_audit_table(dynamodb_resource.meta.client) masters_registry._update_youtube_with_carveout.reset_mock() result = masters_registry._process_isrc_info( {'isrc': 'QA123', 'tuid': 1}, ['AF'], valid_correlation_id, 'oa:123') expected = { 'warning': False, 'success': False, 'claimed_by_another_owner': [], 'isrc': 'QA123', 'claimed_by_the_same_label': ['AF'], 'resolved_conflict': False, 'locked_territories': None, 'substore_carveouts': [], 'error_message': '' } assert result == expected masters_registry._update_youtube_with_carveout.assert_called_once_with( 'QA123', 1, {'AF': [{'tuid': Decimal('1')}]}, valid_correlation_id)