"""Tests for bulk import UPC/ISRC reports.""" from unittest.mock import Mock from unittest.mock import patch import boto3 from moto import mock_aws from oto import response import pytest from masters_registry.api import app from masters_registry.constant import field_const from masters_registry.logic import bulk_tasks from masters_registry.models import bulk_tasks as task_model from masters_registry.models import ows_carveouts from tests.helpers import patches s3_resource = boto3.resource("s3") @pytest.fixture def task_data_partial_fail(): """Returns task data for partially failed task.""" task_data = { 'type': 'BULK_IMPORT', 'finish_datetime': patches.TEST_TIME_AS_DATETIME, 'user_name': 'Test user', 'status': 'PARTIAL_FAIL', 'count': 2, 'user_id': '1', 'create_datetime': patches.TEST_TIME_AS_DATETIME, 'correlation_id': 'test correlation id', 'result': [ {'status_report': [ { 'vendor_id': 11, 'claimed_by_another_owner': [], 'success': True, 'claimed_by_the_same_label': [], 'error_message': "", 'warning': False, 'locked_territories': None, 'substore_carveouts': [['453', set()]], 'resolved_conflict': False, 'isrc': 'QA123', 'tuid': '1', 'upc': '1234' }], 'error_count': 0, 'success_count': 1, 'upc': '1234', 'warning_count': 0 }, {'status_report': [ { 'vendor_id': 22, 'isrc': 'QA456', 'tuid': '2', 'error_message': 'Unable to find territories carve-in', 'upc': '1234', 'success': False }], 'error_count': 1, 'success_count': 0, 'upc': '039841931709', 'warning_count': 0 } ] } return task_data @mock_aws def test_bulk_import_upc_report_upc_partial_fail( setup_bulk_status_db, seed_bulk_status_db, feature_engine, task_data_partial_fail): """Should render correct UPC report for partial fail case.""" task_status = task_model.TaskStatus(**task_data_partial_fail) seed_bulk_status_db([task_status]) with app.app_context(), patch('masters_registry.logic.bulk_tasks.s3') as mock_s3: mock_s3.check_if_file_exists.return_value = False bulk_tasks.generate_report( task_id=1, correlation_id='test', import_report_type='upc') finish_date = str(patches.TEST_TIME_AS_DATETIME).split(' ') expected_key = 'masters-registry-reports/bulk-action/' \ '{}_Bulk_Update_UPCs_UPC_1.csv'.\ format(finish_date[0]) mock_s3.check_if_file_exists.assert_called_with(expected_key) mock_s3.upload_file_object.assert_called() mock_s3.get_file_url.assert_called_with(expected_key) @pytest.fixture def task_data_claimed_by_another_owner(): """Task data when ISRC claimed by another owner.""" task_data = { 'type': 'BULK_IMPORT', 'finish_datetime': patches.TEST_TIME_AS_DATETIME, 'user_name': 'Test user', 'status': 'FAILED', 'count': 1, 'user_id': '1', 'create_datetime': patches.TEST_TIME_AS_DATETIME, 'correlation_id': 'test correlation id', 'result': [ {"status_report": [ { "claimed_by_another_owner": [["AF", [111111]]], "success": False, "claimed_by_the_same_label": [], "error_message": "", "warning": False, "locked_territories": None, "substore_carveouts": [], "resolved_conflict": False, "isrc": "QA123", "upc": "1234" }], "error_count": 0, "success_count": 1, "upc": "1234", "warning_count": 0 } ] } return task_data def ownership_get_tracks_patch(_): """Returns single track.""" message = { 111111: { field_const.VENDOR_ID: 47, field_const.SUBACCOUNT_ID: None, field_const.SUBACCOUNT_NAME: None, field_const.TUID: 111111, field_const.ISRC: 'ABC123456', field_const.RELEASE_NAME: 'Zodiac Symphony', field_const.UPC: 669910758229, field_const.TRACK_NAME: 'Taurus', field_const.ARTIST_NAME: 'Burkhard Schmidl', field_const.VENDOR_NAME: 'Amin Refaat' } } return response.Response(message) @patch( 'masters_registry.models.ownership.get_tracks', new=ownership_get_tracks_patch) def test_bulk_import_upc_report_upc_another_owner( setup_bulk_status_db, seed_bulk_status_db, task_data_claimed_by_another_owner): """Should render correct UPC report when claimed by another owner.""" task_status = task_model.TaskStatus(**task_data_claimed_by_another_owner) seed_bulk_status_db([task_status]) with app.app_context(), patch('masters_registry.logic.bulk_tasks.s3') as mock_s3: mock_s3.check_if_file_exists.return_value = False bulk_tasks.generate_report( task_id=1, correlation_id='test', import_report_type='upc') finish_date = str(patches.TEST_TIME_AS_DATETIME).split(' ') expected_key = 'masters-registry-reports/bulk-action/' \ '{}_Bulk_Update_UPCs_UPC_1.csv'.\ format(finish_date[0]) mock_s3.check_if_file_exists.assert_called_with(expected_key) mock_s3.upload_file_object.assert_called() mock_s3.get_file_url.assert_called_with(expected_key) @pytest.fixture def task_data_same_owner(): """Returns task data when ISRC is claimed by the same owner.""" task_data = { 'type': 'BULK_IMPORT', 'finish_datetime': patches.TEST_TIME_AS_DATETIME, 'user_name': 'Test user', 'status': 'FAILED', 'count': 1, 'user_id': '1', 'create_datetime': patches.TEST_TIME_AS_DATETIME, 'correlation_id': 'test correlation id', 'result': [ {"status_report": [ { "claimed_by_another_owner": [], "success": False, "claimed_by_the_same_label": ["AF"], "error_message": "", "warning": False, "locked_territories": None, "substore_carveouts": [], "resolved_conflict": False, "isrc": "QA123", "upc": "1234" }], "error_count": 0, "success_count": 1, "upc": "1234", "warning_count": 0 } ] } return task_data def test_bulk_import_upc_report_upc_same_owner( setup_bulk_status_db, seed_bulk_status_db, feature_engine, task_data_claimed_by_another_owner): """Should render correct UPC report when claimed by same owner.""" with app.app_context(), patch('masters_registry.logic.bulk_tasks.s3') as mock_s3: mock_s3.check_if_file_exists.return_value = False task_status = task_model.TaskStatus(**task_data_claimed_by_another_owner) seed_bulk_status_db([task_status]) finish_date = str(patches.TEST_TIME_AS_DATETIME).split(' ') expected_key = 'masters-registry-reports/bulk-action/' \ '{}_Bulk_Update_UPCs_UPC_1.csv'.\ format(finish_date[0]) bulk_tasks.generate_report( task_id=1, correlation_id='test', import_report_type='upc') mock_s3.check_if_file_exists.assert_called_with(expected_key) mock_s3.upload_file_object.assert_called() mock_s3.get_file_url.assert_called_with(expected_key) @patch( 'masters_registry.models.ownership.get_tracks', new=ownership_get_tracks_patch) def test_bulk_import_upc_report_isrc_same_owner_internal_conflicts_on( setup_bulk_status_db, seed_bulk_status_db, task_data_claimed_by_another_owner): """Should render correct ISRC report when claimed by same owner.""" task_status = task_model.TaskStatus(**task_data_claimed_by_another_owner) seed_bulk_status_db([task_status]) with app.app_context(), patch('masters_registry.logic.bulk_tasks.s3') as mock_s3: mock_s3.check_if_file_exists.return_value = False finish_date = str(patches.TEST_TIME_AS_DATETIME).split(' ') expected_key = 'masters-registry-reports/bulk-action/' \ '{}_Bulk_Update_UPCs_ISRC_1.csv'.\ format(finish_date[0]) bulk_tasks.generate_report( task_id=1, correlation_id='test', import_report_type='isrc') mock_s3.check_if_file_exists.assert_called() mock_s3.upload_file_object.assert_called() mock_s3.get_file_url.assert_called_with(expected_key) @pytest.fixture def task_data_locked_territories(): """Returns task data when there are locked territories.""" task_data = { 'type': 'BULK_IMPORT', 'finish_datetime': patches.TEST_TIME_AS_DATETIME, 'user_name': 'Test user', 'status': 'PARTIAL_FAIL', 'count': 1, 'user_id': '1', 'create_datetime': patches.TEST_TIME_AS_DATETIME, 'correlation_id': 'test correlation id', 'result': [ {"status_report": [ { "vendor_id": 11, "warning": False, "claimed_by_the_same_label": [], "resolved_conflict": False, "locked_territories": ["AF"], "isrc": "QA123", "tuid": '1', "success": False, "upc": "1234", "claimed_by_another_owner": [], "error_message": "", "substore_carveouts": [], }, { "vendor_id": 11, "warning": False, "claimed_by_the_same_label": [], "resolved_conflict": False, "locked_territories": None, "isrc": "QA456", "tuid": '2', "success": True, "upc": "1234", "claimed_by_another_owner": [], "error_message": "", "substore_carveouts": [["453", []]] }], "error_count": 1, "success_count": 1, "upc": "1234", "warning_count": 0 } ] } return task_data @pytest.fixture def reports_refactoring_on_task_data(): task_data = { 'type': 'BULK_IMPORT', 'finish_datetime': patches.TEST_TIME_AS_DATETIME, 'user_name': 'Test user', 'status': 'PARTIAL_FAIL', 'count': 1, 'user_id': '1', 'create_datetime': patches.TEST_TIME_AS_DATETIME, 'correlation_id': 'test correlation id', 'result': [ {"status_report": [ { "vendor_id": "11", "warning": False, "claimed_by_the_same_label": [], "resolved_conflict": False, "locked_territories": ["AF"], "isrc": "QA12", "tuid": "9", "success": False, "upc": "1234", "claimed_by_another_owner": [], "error_message": "", "substore_carveouts": [] }, { "vendor_id": "11", "warning": False, "claimed_by_the_same_label": [], "resolved_conflict": True, "removed_territories": ['QA'], "locked_territories": None, "isrc": "QA34", "tuid": "1", "success": True, "upc": "1234", "claimed_by_another_owner": [], "error_message": "", 'substore_carveouts': [], }, { "vendor_id": "11", "warning": True, "claimed_by_the_same_label": [], "resolved_conflict": False, "removed_territories": [], "locked_territories": None, "isrc": "QA56", "tuid": "2", "success": True, "upc": "1234", "claimed_by_another_owner": [["QA", [111111]]], "error_message": "", 'substore_carveouts': [], }, { "vendor_id": "11", "warning": False, "claimed_by_the_same_label": [], "resolved_conflict": False, "removed_territories": [], "locked_territories": None, "isrc": "QA78", "tuid": "3", "success": False, "upc": "1234", "claimed_by_another_owner": [], "error_message": "", 'substore_carveouts': [['453', {'QA'}]], }, ], "error_count": 1, "success_count": 1, "upc": "1234", "warning_count": 2 }, { "status_report": [ { "vendor_id": "22", "warning": False, "claimed_by_the_same_label": ['QA'], "resolved_conflict": False, "locked_territories": [], "isrc": "PT12", "tuid": "4", "success": False, "upc": "789", "claimed_by_another_owner": [], "error_message": "", "substore_carveouts": [] }, ], "error_count": 0, "success_count": 1, "upc": "56", "warning_count": 0 }, { "status_report": [ { "vendor_id": "33", "warning": False, "claimed_by_the_same_label": [], "resolved_conflict": False, "locked_territories": [], "isrc": "PT34", "tuid": "5", "success": False, "upc": "78", "claimed_by_another_owner": [], "error_message": "upc_youtube_carved_out", "substore_carveouts": [] }, ], "error_count": 1, "success_count": 0, "upc": "78", "warning_count": 0 } ] } return task_data @patch( 'masters_registry.models.ownership.get_tracks', new=ownership_get_tracks_patch) @patch.object( ows_carveouts, 'get_store_name', new=Mock( return_value=response.Response('YouTube'))) def test_bulk_import_upc_report_isrc_reports_refactoring_on( setup_bulk_status_db, seed_bulk_status_db, reports_refactoring_on_task_data): """Should render correct ISRC report.""" task_status = task_model.TaskStatus(**reports_refactoring_on_task_data) seed_bulk_status_db([task_status]) with app.app_context(), patch('masters_registry.logic.bulk_tasks.s3') as mock_s3: mock_s3.check_if_file_exists.return_value = False bulk_tasks.generate_report( task_id=1, correlation_id='test', import_report_type='isrc') finish_date = str(patches.TEST_TIME_AS_DATETIME).split(' ') expected_key = 'masters-registry-reports/bulk-action/' \ '{}_Bulk_Update_UPCs_ISRC_1.csv'.\ format(finish_date[0]) mock_s3.check_if_file_exists.assert_called() mock_s3.upload_file_object.assert_called() mock_s3.get_file_url.assert_called_with(expected_key) def test_bulk_import_upc_reports_refactoring_on( setup_bulk_status_db, seed_bulk_status_db, feature_engine, task_data_partial_fail): """Should render correct UPC report.""" task_status = task_model.TaskStatus(**task_data_partial_fail) seed_bulk_status_db([task_status]) with app.app_context(), patch('masters_registry.logic.bulk_tasks.s3') as mock_s3: mock_s3.check_if_file_exists.return_value = False bulk_tasks.generate_report( task_id=1, correlation_id='test', import_report_type='upc') finish_date = str(patches.TEST_TIME_AS_DATETIME).split(' ') expected_key = 'masters-registry-reports/bulk-action/' \ '{}_Bulk_Update_UPCs_UPC_1.csv'.\ format(finish_date[0]) mock_s3.check_if_file_exists.assert_called() mock_s3.upload_file_object.assert_called() mock_s3.get_file_url.assert_called_with(expected_key)