import decimal from operator import itemgetter import time from unittest.mock import call from unittest.mock import MagicMock from unittest.mock import Mock from unittest.mock import patch import uuid from botocore.exceptions import ClientError from flexmock import flexmock from freezegun import freeze_time from moto import mock_aws import boto3 from oto import response import pytest from sqlalchemy.exc import SQLAlchemyError from masters_registry import config from masters_registry.connectors import mysql from masters_registry.connectors import sentry from masters_registry.constant import db_const from masters_registry.constant import field_const from masters_registry.constant import opcode_const from masters_registry.logic import masters_registry from masters_registry.models import ownership from tests.helpers import patches from masters_registry.connectors.dynamodb import client as outside_dynamodb_client dynamodb_resource = boto3.resource("dynamodb") def test_get_ownership(monkeypatch): """Test getting an ownership information. """ item = dict() isrc = int(time.time()) get_item = MagicMock( spec=ownership.active_table.get_item, return_value={'Item': item}) monkeypatch.setattr(ownership.active_table, 'get_item', get_item) ownership_information = ownership.get_ownership(isrc) get_item.assert_called_with(Key={field_const.ISRC: isrc}) assert ownership_information.message is item def test_get_ownership_null_isrc(): """Testing case when input isrc == None """ result = ownership.get_ownership(None) assert result.message == {} def test_get_inexistant_ownership(monkeypatch): """Test getting ownership information that does not exist. """ get_item = MagicMock( spec=ownership.active_table.get_item, return_value={}) monkeypatch.setattr(ownership.active_table, 'get_item', get_item) isrc = int(time.time()) ownership_information = ownership.get_ownership(isrc) get_item.assert_called_with(Key={field_const.ISRC: isrc}) assert not ownership_information.message assert isinstance(ownership_information.message, dict) @pytest.mark.parametrize('value', [None, 0, '', [], {}]) def test_getting_tracks_without_tuids(monkeypatch, value): """Test getting tracks without tuids. """ response = ownership.get_tracks(value) assert response assert not response.message assert isinstance(response.message, dict) def test_create_record(monkeypatch): """Assert that create_audit_op returns expected audit item """ actual_time = time.time() monkeypatch.setattr( time, 'time', MagicMock(spec=time.time, return_value=actual_time)) opcode = opcode_const.ADD isrc = 'ZW1010600008' tuid = 123 territory = 'FR' correlation_id = str(uuid.uuid1()) user = '456' expected_audit_item = { field_const.ISRC_TERRITORY: '{}_{}'.format(isrc, territory), 'correlation_id': correlation_id, field_const.TUID: tuid, field_const.USER: user, field_const.OPCODE: opcode, field_const.TIMESTAMP: decimal.Decimal(str(actual_time * 1000)) } actual_audit_item = ownership.create_record( opcode, isrc, territory, correlation_id, user, tuid) assert time.time.called assert expected_audit_item == actual_audit_item def test_get_isrcs_exception(): """Test get_isrcs function for valid exception. """ sentry.sentry_client = flexmock() sentry.sentry_client.should_receive('captureException').once() (flexmock(mysql) .should_receive('mr_session_scope') .and_raise(SQLAlchemyError)) result = ownership.get_isrcs([ '100075236325', '123456789120', '123456789121']) assert result.status == 500 @pytest.fixture def get_isrcs_valid_response(): return response.Response(message={ '123456789120': [ { 'track_name': 'Track 1', 'isrc': 'USA560319122', 'tuid': 140030, 'deletions': 'N', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': '123456789120', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': ( db_const.DIGITAL_DISTRIBUTION_FORMAT), }, { 'track_name': 'Track 2', 'isrc': 'USA560319123', 'tuid': 140031, 'deletions': 'N', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': '123456789120', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': ( db_const.DIGITAL_DISTRIBUTION_FORMAT), }, { 'track_name': 'Track 3', 'isrc': 'USA560319124', 'tuid': 140032, 'deletions': 'N', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': '123456789120', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': ( db_const.DIGITAL_DISTRIBUTION_FORMAT), }, { 'track_name': 'Track 4', 'isrc': 'USA560319125', 'tuid': 140033, 'deletions': 'N', 'product_type_id': db_const.MUSIC_PRODUCT_TYPE_ID, 'upc': '123456789120', 'release_status': 'in_content', 'not_for_distribution': 'N', 'track_type': 'music', 'distribution_context_type': ( db_const.DIGITAL_DISTRIBUTION_FORMAT), } ] }) def test_get_isrcs_success(get_isrcs_valid_response, get_single_isrc_from_db): """Test get_isrcs function for success response. Args: get_isrcs_valid_response: mock of valid response of function. get_isrcs_from_db: mock of valid response of db query. """ session_scope = flexmock( execute=lambda query, data: get_single_isrc_from_db ) (flexmock(mysql) .should_receive('mr_session_scope') .and_return(session_scope)) result = ownership.get_isrcs(['123456789120']) assert result.message == get_isrcs_valid_response.message def test_get_isrcs_with_leading_zero_upc( get_isrcs_valid_response, get_single_isrc_from_db): """Test get_isrcs function with leading zero UPC Args: get_isrcs_valid_response: mock of valid response of function. get_isrcs_from_db: mock of valid response of db query. """ session_scope = flexmock( execute=lambda query, data: get_single_isrc_from_db ) (flexmock(mysql) .should_receive('mr_session_scope') .and_return(session_scope)) result = ownership.get_isrcs(['0123456789120']) assert result.message == get_isrcs_valid_response.message def test_get_existing_isrcs_in_active_table(monkeypatch): """Test getting existing isrcs from active table """ isrcs = [str((time.time())) for i in range(150)] batch_get_item_kwargs = { 'RequestItems': { config.DYNAMODB_MASTERS_ACTIVE: { 'Keys': [{'isrc': isrc} for isrc in isrcs[100:150]], 'ConsistentRead': True } } } batch_response = [ { 'isrc': isrc, field_const.LOCKED_TERRITORIES: {}, field_const.TERRITORIES: {} } for isrc in isrcs[:75] ] batch_get_item = MagicMock( spec=ownership.dynamodb.client.batch_get_item, side_effect=[ { 'Responses': { config.DYNAMODB_MASTERS_ACTIVE: batch_response, }, 'UnprocessedKeys': { config.DYNAMODB_MASTERS_ACTIVE: { 'Keys': isrcs[50:75] } } }, { 'Responses': { config.DYNAMODB_MASTERS_ACTIVE: batch_response, }, 'UnprocessedKeys': {} }, { 'Responses': { config.DYNAMODB_MASTERS_ACTIVE: batch_response, }, 'UnprocessedKeys': {} }, ] ) monkeypatch.setattr( ownership.dynamodb.client, 'batch_get_item', batch_get_item) existing_isrcs = ownership.get_existing_isrcs_in_active_table(isrcs) batch_get_item.assert_called_with(**batch_get_item_kwargs) assert batch_get_item.call_count == 3 assert batch_response * 3 == existing_isrcs def test_get_existing_isrcs_in_tracks_success(): """Test getting existing isrcs from art_relations.track table """ isrcs = ['1', '2', '3'] session_scope = flexmock( execute=lambda query, data: [[isrc] for isrc in isrcs] ) (flexmock(mysql) .should_receive('mr_session_scope') .and_return(session_scope)) result = ownership.get_existing_isrcs_in_tracks(isrcs) assert result == set(isrcs) def test_get_existing_isrcs_in_tracks_exception(): """Test get_existing_isrcs_in_tracks function for valid exception. """ isrcs = ['1', '2', '3'] (flexmock(mysql) .should_receive('mr_session_scope') .and_raise(SQLAlchemyError)) result = ownership.get_existing_isrcs_in_tracks(isrcs) assert result.status == 500 @patch( 'masters_registry.models.ownership.new_audit_table.batch_writer', autospec=True) def test_update_multiple_ownerships(table): """Test update_multiple_ownerships function """ batch_writer = MagicMock() batch_writer.__enter__.return_value = batch_writer batch_writer.put_item = MagicMock() table.return_value = batch_writer items = [1, 2] ownership.update_multiple_ownerships(items) batch_writer.put_item.assert_called_with(Item=items[-1]) assert batch_writer.put_item.call_count == len(items) @patch( 'masters_registry.models.ownership.active_record_remove_attributes', autospec=True) @patch( 'masters_registry.models.ownership.active_record_add_attributes', autospec=True) def test_update_lock_active_record(add_attr_mock, remove_attr_mock): """Test update_lock_active_record function """ territories = { 'to_claim': ['QA'], 'to_lock': ['PL'], 'to_unlock': ['US'], 'to_remove': ['CA'] } isrc = 'US1' reason = 'Lock reason' tuid = 11111 ownership.update_active_record( isrc, territories, lock_reason=reason, tuid=tuid) add_calls = [ call( isrc, territories['to_claim'], 'territories', {field_const.TUID: tuid} ), call( isrc, territories['to_lock'], 'locked_territories', {field_const.REASON: reason} ) ] add_attr_mock.assert_has_calls(add_calls) remove_calls = [ call( isrc, territories['to_unlock'], 'locked_territories', ), call( isrc, territories['to_remove'], 'territories', ) ] remove_attr_mock.assert_has_calls(remove_calls) assert add_attr_mock.call_count == 2 assert remove_attr_mock.call_count == 2 @patch( 'masters_registry.connectors.dynamodb.batch_update_item_fields', autospec=True) def test_active_record_add_attributes(batch_update_item_fields): """Test active_record_add_attributes function """ batch_update_item_fields.return_value = False isrc = 'US1' territories = ['PL', 'CA'] field = 'locked_territories' value = {field_const.LOCK_REASON: 'Lock reason'} expressions = [ 'locked_territories.#t0=:tr0', 'locked_territories.#t1=:tr1' ] attribute_names = {'#t0': 'PL', '#t1': 'CA'} attribute_values = {':tr0': value, ':tr1': value} args = [ ownership.active_table, {field_const.ISRC: isrc}, 'SET', expressions, attribute_names, attribute_values, ] ownership.active_record_add_attributes(isrc, territories, field, value) batch_update_item_fields.assert_called_with(*args, force=True) assert batch_update_item_fields.call_count == len(territories) + 1 @patch( 'masters_registry.connectors.dynamodb.batch_update_item_fields', autospec=True) def test_active_record_remove_attributes(batch_update_item_fields): """Test active_record_remove_attributes function """ batch_update_item_fields.return_value = False isrc = 'US1' territories = ['PL', 'CA'] field = 'locked_territories' expressions = ['locked_territories.#t0', 'locked_territories.#t1'] attribute_names = {'#t0': 'PL', '#t1': 'CA'} args = [ ownership.active_table, {field_const.ISRC: isrc}, 'REMOVE', expressions, attribute_names, ] ownership.active_record_remove_attributes(isrc, territories, field) batch_update_item_fields.assert_called_with( *args, attribute_values=None, force=True) assert batch_update_item_fields.call_count == len(territories) + 1 def test_create_audit_record_with_reson(): """Test create_audit_record with reason passed """ opcode = 'ADD' isrc = 'US1' territories = ['CA', 'QA'] correlation_id = '19' user = '1919' reason = 'Lock reason' record = ownership.create_audit_record( opcode, isrc, territories, correlation_id, user, reason=reason) assert record[field_const.ISRC] == isrc assert record[field_const.USER] == user assert record[field_const.OPCODE] == opcode assert record[field_const.TERRITORIES] == territories assert record[field_const.CORRELATION_ID_FIELD] == correlation_id assert record[field_const.LOCK_REASON] == reason assert record.get(field_const.TUID) is None def test_create_audit_record_with_tuid(): """Test create_audit_record with tuid passed """ opcode = 'ADD' isrc = 'US1' tuid = 222 territories = {'CA': tuid, 'QA': tuid} correlation_id = '19' user = '1919' record = ownership.create_audit_record( opcode, isrc, territories, correlation_id, user, tuid=tuid) assert record[field_const.ISRC] == isrc assert record[field_const.USER] == user assert record[field_const.OPCODE] == opcode assert record[field_const.TERRITORIES] == territories assert record[field_const.CORRELATION_ID_FIELD] == correlation_id assert record.get(field_const.LOCK_REASON) is None def test_create_audit_record_with_conflict(created_conflict): """Test create_audit_record with conflict.""" opcode = 'ADD' isrc = 'USA1323123123' tuid = 222 territories = {'CA': tuid, 'QA': tuid} correlation_id = str(uuid.uuid4()) user = '1919' conflict = created_conflict source = field_const.MANUAL_EDIT_ISRC record = ownership.create_audit_record( opcode, isrc, territories, correlation_id, user, source=source, tuid=tuid, conflict=conflict) assert record[field_const.ISRC] == isrc assert record[field_const.USER] == user assert record[field_const.SOURCE] == source assert record[field_const.OPCODE] == opcode assert record[field_const.TERRITORIES] == territories assert record[field_const.CORRELATION_ID_FIELD] == correlation_id assert record.get(field_const.LOCK_REASON) is None assert record[field_const.CONFLICT] == created_conflict @patch( 'masters_registry.utils.get_timestamp_for_dynamo', new=lambda: patches.TEST_TIME_AS_TIMESTAMP) @mock_aws def test_create_active_record(setup_masters_active_table): """Test create_active_record function """ 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) user = '123' isrc = 'US1' tuid = 222 territories = ['QA', 'AD'] ownership.create_active_record(isrc, tuid, territories, user) ownership_info = ownership.get_ownership(isrc).message expected_ownership_info = { 'territories': { 'AD': [{'tuid': decimal.Decimal(tuid)}], 'QA': [{'tuid': decimal.Decimal(tuid)}] }, 'isrc': 'US1', 'locked_territories': {}, 'timestamp': patches.TEST_TIME_AS_TIMESTAMP } assert ownership_info == expected_ownership_info @patch( 'masters_registry.utils.get_timestamp_for_dynamo', new=lambda: patches.TEST_TIME_AS_TIMESTAMP) @freeze_time(patches.TEST_TIME_AS_STR) @mock_aws def test_create_ownership_internal( setup_masters_active_table, setup_new_audit_table): """Test create_ownership function """ 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) setup_masters_active_table(dynamodb_resource.meta.client) user = '123' isrc = 'US1' tuid = 222 territories = ['QA', 'AD'] correlation_id = 'test' ownership.create_ownership(isrc, territories, tuid, correlation_id, user) ownership_info = ownership.get_ownership(isrc).message expected_ownership_info = { 'territories': { 'AD': [{'tuid': decimal.Decimal(tuid)}], 'QA': [{'tuid': decimal.Decimal(tuid)}] }, 'isrc': 'US1', 'locked_territories': {}, 'timestamp': patches.TEST_TIME_AS_TIMESTAMP } assert ownership_info == expected_ownership_info audit_record = ownership.get_ownership_audit(isrc).message[0] expected_audit_record = { 'opcode': opcode_const.ADD, 'territories': { 'AD': decimal.Decimal(tuid), 'QA': decimal.Decimal(tuid) }, 'isrc': isrc, 'user': user, 'correlation_id': correlation_id, 'timestamp': decimal.Decimal('1512140400000.0') } assert audit_record == expected_audit_record @patch( 'masters_registry.models.ownership.active_table.put_item') @patch( 'masters_registry.utils.get_timestamp_for_dynamo', new=lambda: patches.TEST_TIME_AS_TIMESTAMP) def test_create_initial_record(put_item_mock): """Test create_initial_record function """ isrc = 'US1' territories = {'US': [{field_const.TUID: '123'}]} kwargs = { 'Item': { field_const.ISRC: isrc, field_const.TERRITORIES: territories, field_const.LOCKED_TERRITORIES: {}, field_const.TIMESTAMP: patches.TEST_TIME_AS_TIMESTAMP }, 'ConditionExpression': 'attribute_not_exists({})'.format( field_const.ISRC) } ownership.create_initial_record(isrc, territories) put_item_mock.assert_called_once_with(**kwargs) def test_get_ownership_audit(monkeypatch): isrc = 'TESTISRC' correlation_id = str(uuid.uuid1()) record1 = { field_const.ISRC: isrc, field_const.TIMESTAMP: decimal.Decimal(str(time.time() * 1000)), field_const.USER: 2233, field_const.OPCODE: 'LOCK', field_const.TERRITORIES: ['US', 'CA'], field_const.CORRELATION_ID_FIELD: correlation_id } record2 = { field_const.ISRC: isrc, field_const.TIMESTAMP: decimal.Decimal(str(time.time() * 1000)), field_const.USER: 2233, field_const.OPCODE: 'UNLOCK', field_const.TERRITORIES: ['QA'], field_const.CORRELATION_ID_FIELD: correlation_id } get_paginator = Mock() get_paginator.return_value.paginate.return_value = [ {'Items': [record1]}, {'Items': [record2]}, ] monkeypatch.setattr( ownership.dynamodb.client, 'get_paginator', get_paginator) ownership_audit = ownership.get_ownership_audit(isrc) assert ownership_audit.message == [record1, record2] @patch( 'masters_registry.models.ownership.get_ownership_audit', new=patches.ownership_get_ownership_audit_patch) @patch( 'masters_registry.models.users.get_orchard_user_names', new=patches.users_get_orchard_user_names_patch) def test_get_lock_history_for_ownership(): """Test get_lock_history_for_ownership function """ ownership_info = patches.ownership_get_ownership_patch('').message masters_registry._build_tuid_territories_map(ownership_info) masters_registry._group_locked_territories_by_reason(ownership_info) ownership.get_lock_history_for_ownership(ownership_info) ownership_info[field_const.LOCKED_TERRITORIES] = sorted( ownership_info[field_const.LOCKED_TERRITORIES], key=itemgetter(field_const.REASON)) assert 'created' in ownership_info[field_const.LOCKED_TERRITORIES][0] assert 'updated' in ownership_info[field_const.LOCKED_TERRITORIES][0] assert 'created' in ownership_info[field_const.LOCKED_TERRITORIES][1] assert 'updated' in ownership_info[field_const.LOCKED_TERRITORIES][1] assert (ownership_info[field_const.LOCKED_TERRITORIES][0][ 'created']['user'] == 'Alex Ivanov') assert (ownership_info[field_const.LOCKED_TERRITORIES][0][ 'created']['datetime'] == '2016-11-16T06:37:45.286801') assert ownership_info[field_const.LOCKED_TERRITORIES][0]['updated'] == {} assert (ownership_info[field_const.LOCKED_TERRITORIES][1][ 'updated']['user'] == 'Alex Ivanov') assert (ownership_info[field_const.LOCKED_TERRITORIES][1][ 'updated']['datetime'] == '2016-11-16T06:38:20.286801') @patch( 'masters_registry.models.ownership.get_ownership_audit', new=patches.ownership_get_ownership_audit_patch) @patch( 'masters_registry.models.users.get_orchard_user_names', new=patches.users_get_orchard_user_names_patch) def test_get_lock_history_for_single_reason(): """Test get_lock_history_for_single_reason function """ isrc = 'US4R30920517' reason = 'Owned by Warner' history = ownership.get_lock_history_for_single_reason( reason, isrc).message assert (history['created']['user'] == 'Alex Ivanov') assert (history['created']['datetime'] == '2016-11-16T06:37:10.286801') assert (history['updated']['user'] == 'Alex Ivanov') assert (history['updated']['datetime'] == '2016-11-16T06:38:20.286801') def test_get_upc_by_tuid(): """Assert get_upc_by_tuid returns valid response""" tuid = 333333333 upc = 1313131313 result_proxy = flexmock(fetchone=lambda: {field_const.UPC: upc}) session_scope = flexmock(execute=lambda query, data: result_proxy) (flexmock(mysql) .should_receive('mr_session_scope') .and_return(session_scope)) result = ownership.get_upc_by_tuid(tuid) assert result assert result.message[field_const.UPC] == upc def test_get_upc_by_tuid_not_found(): """Assert get_upc_by_tuid returns not found response""" tuid = 333333333 result_proxy = flexmock(fetchone=lambda: {}) session_scope = flexmock( execute=lambda query, data: result_proxy ) (flexmock(mysql) .should_receive('mr_session_scope') .and_return(session_scope)) result = ownership.get_upc_by_tuid(tuid) assert not result assert result.status == 404 def test_get_upc_by_tuid_failed(): """Assert get_upc_by_tuid handles SQLAlchemy exceptions""" tuid = 333333333 (flexmock(mysql) .should_receive('mr_session_scope') .and_raise(SQLAlchemyError)) result = ownership.get_upc_by_tuid(tuid) assert not result assert result.status == 500 def test_update_active_record_no_lock_reason(): """Expect error because territories_to_lock requires lock_reason.""" territories = {'to_lock': ['AF']} with pytest.raises(ValueError): ownership.update_active_record( 'test_isrc', territories) def test_update_active_record_no_tuid(): """Expect error because territories_to_claim requires tuid.""" territories = {'to_claim': ['AF']} with pytest.raises(ValueError): ownership.update_active_record( 'test_isrc', territories) @patch( 'masters_registry.models.ownership.active_table.put_item', new=Mock(side_effect=ClientError({'Error': {}}, 'test operation'))) def test_create_initial_record_client_error(): """Expect to handle boto ClientError silently.""" ownership.create_initial_record('test_isrc', {})