"""Tests for product logic.""" import random import uuid from unittest.mock import call from unittest.mock import MagicMock from unittest.mock import patch from oto import response from owsrequest import access import pytest from pythonfeatures import pythonfeatures from pythonfeatures.constants import split as split_constants import sentry_sdk from sqlalchemy.exc import SQLAlchemyError from product.constants import error from product.constants import product as product_constants from product.logic import product as product_logic from product.models import ows_account from product.models import ows_carveouts from product.models import ows_carveouts_python from product.models import ows_marketing from product.models import ows_product_digital from product.models import ows_product_physical from product.models import ows_product_review from product.models import ows_timed_release from product.models import ows_video from product.models import product as product_model from product.models import release_artist from product.models import release_subgenre from product.models import upc as upc_model from product.utils import feature_control_util test_message_param = { 'context_type': 'physical', 'deletions': 'N', 'display_upc': '888831283041', 'distribution_format_id': 2, 'not_for_distribution': 'N', 'product_id': 1, 'product_name': 'product test 1', 'product_type_id': 1, 'project_id': 22, 'release_date': '2017-09-09', 'status': 'label_processing', 'subaccount_id': 1, 'upc': 888831283041, 'vendor_id': 1 } def mock_product_found_by_id(mocker, payload='override if needed'): """Mock a 200 response from the Product model.""" product_response = response.Response(status=200, message=payload) mocker.patch.object( product_model, 'get_product_by_id', return_value=product_response) def mock_product_not_found_by_id(mocker): """Mock a 404 response from the Product model.""" product_response = response.Response(status=404) mocker.patch.object( product_model, 'get_product_by_id', return_value=product_response) def mock_product_found_by_upc(mocker, payload='override if needed'): """Mock a 200 response from the Product model.""" product_response = response.Response(status=200, message=payload) mocker.patch.object( product_model, 'get_product_by_upc', return_value=product_response) def mock_product_not_found_by_upc(mocker): """Mock a 404 response from the Product model.""" product_response = response.Response(status=404) mocker.patch.object( product_model, 'get_product_by_upc', return_value=product_response) def test_product_ownership_vendor_and_subaccount_success( mocker, test_product, test_vendor, test_subaccount): """Test ownership with model layer success. Test ownership when both vendor_id and subaccount_id are provided. """ product_payload = { 'vendor_id': test_vendor, 'subaccount_id': test_subaccount, 'product_id': test_product } mock_product_found_by_id(mocker, payload=product_payload) result = product_logic.check_product_ownership( test_product, vendor_id=test_vendor, subaccount_id=test_subaccount) assert result.status == 200 def test_product_ownership_vendor_success( mocker, test_product, test_vendor): """Test vendor level ownership with success. Test ownership when vendor_id only is provided. """ product_payload = { 'vendor_id': test_vendor, 'product_id': test_product } mock_product_found_by_id(mocker, payload=product_payload) result = product_logic.check_product_ownership( test_product, vendor_id=test_vendor) assert result.status == 200 def test_product_ownership_subaccount_success( mocker, test_product, test_subaccount): """Test subaccount level ownership with success. Test ownership when subaccount_id only is provided. """ product_payload = { 'subaccount_id': test_subaccount, 'product_id': test_product } mock_product_found_by_id(mocker, payload=product_payload) result = product_logic.check_product_ownership( test_product, subaccount_id=test_subaccount) assert result.status == 200 def test_ownership_fails_when_product_not_found( mocker, test_product): """Test ownership when product is not found.""" mock_product_not_found_by_id(mocker) result = product_logic.check_product_ownership(test_product) assert result.status == 404 def test_ownership_fails_when_vendor_is_not_owner( mocker, test_product, test_vendor): """Test product ownership when vendor is forbidden.""" product_payload = { 'vendor_id': '444444', 'product_id': test_product } mock_product_found_by_id(mocker, payload=product_payload) result = product_logic.check_product_ownership( test_product, vendor_id=test_vendor) assert result.status == 403 def test_ownership_fails_when_subaccount_is_not_owner( mocker, test_product, test_subaccount): """Test product ownership when subaccount is forbidden.""" product_payload = { 'subaccount_id': '55555', 'product_id': test_product } mock_product_found_by_id(mocker, payload=product_payload) result = product_logic.check_product_ownership( test_product, subaccount_id=test_subaccount) assert result.status == 403 def test_ownership_fails_when_vendor_and_subaccount_missing( mocker, test_product): """Test product ownership when subaccount is forbidden.""" mock_product_found_by_id(mocker) result = product_logic.check_product_ownership( test_product, vendor_id=None, subaccount_id=None) assert result.status == 400 def test_get_product_by_upc_vendor_and_subaccount_success( mocker, test_product, test_vendor, test_subaccount): """Test get product with model layer success. Test get product when both vendor_id and subaccount_id are provided. """ product_payload = { 'vendor_id': test_vendor, 'subaccount_id': test_subaccount, 'product_id': test_product } mock_product_found_by_upc(mocker, payload=product_payload) result = product_logic.get_product_by_upc( test_product, vendor_id=test_vendor, subaccount_id=test_subaccount) assert result.status == 200 def test_get_product_by_upc_fails_when_product_not_found( mocker, test_product): """Test ownership when product is not found.""" mock_product_not_found_by_upc(mocker) result = product_logic.get_product_by_upc(test_product) assert result.status == 404 def test_get_product_by_upc_succeeds_when_vendor_and_subaccount_missing( mocker, test_product): """Test get product by UPC succeeds when ownership info is not provided. This will be the case in microservice to microservice communication. """ mock_product_found_by_upc(mocker) result = product_logic.get_product_by_upc( test_product, vendor_id=None, subaccount_id=None) assert result.status == 200 def test_get_product_by_upc_fails_when_vendor_is_not_owner( mocker, test_product, test_vendor): """Test product ownership when vendor is forbidden.""" product_payload = { 'vendor_id': '444444', 'product_id': test_product } mock_product_found_by_upc(mocker, payload=product_payload) result = product_logic.get_product_by_upc( test_product, vendor_id=test_vendor) assert result.status == 403 def test_get_product_by_upc_fails_when_subaccount_is_not_owner( mocker, test_product, test_subaccount): """Test product ownership when subaccount is forbidden.""" product_payload = { 'subaccount_id': '55555', 'product_id': test_product } mock_product_found_by_upc(mocker, payload=product_payload) result = product_logic.get_product_by_upc( test_product, subaccount_id=test_subaccount) assert result.status == 403 def test_is_vendor_product_owner(): """Test if a vendor owns a product.""" test_product = {'vendor_id': 1234} assert product_logic.is_vendor_product_owner(test_product, 1234) assert product_logic.is_vendor_product_owner(test_product, '1234') assert not product_logic.is_vendor_product_owner(test_product, 2345) def test_is_subaccount_product_owner(): """Test if a subaccount owns a product.""" test_product = {'subaccount_id': 1234, 'product_id': 1} assert product_logic.is_subaccount_product_owner(test_product, 1234) assert product_logic.is_subaccount_product_owner(test_product, '1234') assert not product_logic.is_subaccount_product_owner(test_product, 2345) def test_get_product_by_product_id_vendor_and_subaccount_success( mocker, test_product, test_vendor, test_subaccount): """Test get product by product_id with model layer success. Test get product when both vendor_id and subaccount_id are provided. """ product_payload = { 'vendor_id': test_vendor, 'subaccount_id': test_subaccount, 'product_id': test_product } mock_product_found_by_id(mocker, payload=product_payload) result = product_logic.get_product_by_product_id( test_product, vendor_id=test_vendor, subaccount_id=test_subaccount) assert result.status == 200 def test_get_product_by_product_id_fails_when_product_not_found( mocker, test_product): """Test getting product by product_id when product is not found.""" mock_product_not_found_by_id(mocker) result = product_logic.get_product_by_product_id(test_product) assert result.status == 404 def test_get_product_by_id_succeeds_without_vendor_and_subaccount( mocker, test_product): """Test get product by id when no ownership information is provided. This is essential for microservice to microservice calls. """ mock_product_found_by_id(mocker) result = product_logic.get_product_by_product_id(test_product) assert result.status == 200 def test_get_product_by_id_succeeds_with_tenant_uuids( mocker, test_product): """Test get product by id with_tenant_uuids True.""" product_response = response.Response(status=200, message={'product_id': test_product}) mocker.patch.object( product_model, 'get_product_by_id', return_value=product_response) expected = { "product_id": test_product, "vendorUUID": "df45d0a9-2c55-4925-95e0-8324f37f1283", "subaccountUUID": None, "companyBrandUUID": "d25a4cd1-e820-45f2-be5c-56edcfeb8298", "parentCompanyUUID": "955a1bbd-b623-4ea1-ab5f-8d6620c442fb" } tenant_response = response.Response(status=200, message=expected) mocker.patch.object( product_model, 'get_tenant_uuids', return_value=tenant_response) result = product_logic.get_product_by_product_id(test_product, None, None, True) assert result.status == 200 assert result.message == expected def test_get_product_by_id_fails_when_vendor_is_not_owner( mocker, test_product, test_vendor): """Test get product by id when vendor is forbidden.""" product_payload = { 'vendor_id': '444444', 'product_id': test_product } mock_product_found_by_id(mocker, payload=product_payload) result = product_logic.get_product_by_product_id( test_product, vendor_id=test_vendor) assert result.status == 403 def test_get_product_by_id_fails_when_subaccount_is_not_owner( mocker, test_product, test_subaccount): """Test get product by id when subaccount is forbidden.""" product_payload = { 'subaccount_id': '55555', 'product_id': test_product } mock_product_found_by_id(mocker, payload=product_payload) result = product_logic.get_product_by_product_id( test_product, subaccount_id=test_subaccount) assert result.status == 403 def test_product_code_found_for_account_found(mocker): """Return 200 when product code found for account.""" _mock_get_product_by_product_code_for_account(mocker, 200) result = product_logic.product_code_found_for_account( 'some-code', 'vendor', 123, 0) assert result.status == 200 assert result.message is None def test_product_code_found_for_account_not_found(mocker): """Return 404 when product code found for account.""" _mock_get_product_by_product_code_for_account(mocker, 404) result = product_logic.product_code_found_for_account( 'some-code', 'vendor', 123, 0) assert result.status == 404 assert result.message is None def test_product_code_found_for_account_found_excluded(mocker): """Return 200 when product code found for account.""" _mock_get_product_by_product_code_for_account(mocker, 200) result = product_logic.product_code_found_for_account( 'some-code', 'vendor', 123, 12) assert result.status == 404 assert result.message is None def _mock_get_product_by_product_code_for_account(mocker, status_code): """Mock an empty response with status code from the product model. Args: mocker (obj): the mocker object. status_code (int): the expected status code. Returns: None """ product_response = response.Response( status=status_code, message={'some': 'thing', 'product_id': 12}) mocker.patch.object( product_model, 'get_first_product_by_product_code_for_account', return_value=product_response) def test_product_codes_available_for_account(mocker): """Return list of product codes not found for account.""" product_response = response.Response(message=['ABC']) mocker.patch.object( product_model, 'get_products_by_product_code_for_account', return_value=product_response) result = product_logic.product_codes_available_for_account( ['ABC', 'DEF', 'GHI'], 'vendor', uuid.uuid4()) assert set(result.message['available_product_codes']) == {'DEF', 'GHI'} result = product_logic.product_codes_available_for_account( ['ABC'], 'vendor', uuid.uuid4()) assert result.message == {'available_product_codes': []} def test_product_by_product_code(mocker): """Return product_id for product code.""" mocker.patch.object( product_model, 'get_product_id_by_product_code_for_account', return_value=response.Response(message=11)) result = product_logic.get_product_id_by_product_code( 'ABC', 'vendor', uuid.uuid4()) assert result.message['product_id'] == 11 def test_delete_product_success(mocker): """Test delete product success.""" product_id = 1234 account_type = 'vendor' account_id = 12 mocker.patch.object( product_model, 'get_product_by_id', return_value=response.Response( message={ 'product_id': product_id, 'upc': 123, 'vendor_id': account_id, 'context_type': 'physical'}), autospec=True) mocker.patch.object( access, 'verify_grass_access', return_value=response.Response() ) mocker.patch.object( ows_carveouts, 'delete_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_carveouts_python, 'delete_release_dms_master_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_carveouts_python, 'delete_release_dms_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_marketing, 'delete_marketing_highlights_by_release' ) mocker.patch.object( ows_timed_release, 'delete_timed_release', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_product_physical, 'delete_physical_product', return_value=response.Response() ) result = product_logic.delete_product(product_id, account_type, account_id) assert result def test_delete_product_failed(mocker): """Test delete product failed.""" product_id = 1234 account_type = 'vendor' account_id = 12 mocker.patch.object( product_model, 'get_product_by_id', return_value=response.Response( message={ 'product_id': product_id, 'upc': 123, 'vendor_id': account_id, 'context_type': 'digital'}), autospec=True) mocker.patch.object( access, 'verify_grass_access', return_value=response.Response() ) mocker.patch.object( ows_carveouts, 'delete_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_marketing, 'delete_marketing_highlights_by_release' ) mocker.patch.object( ows_timed_release, 'delete_timed_release', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_product_physical, 'delete_physical_product', return_value=response.Response() ) result = product_logic.delete_product(product_id, account_type, account_id) assert not result def test_delete_product_ownership_failed(mocker): """Test delete product success.""" product_id = 1234 account_type = 'vendor' account_id = 12 mocker.patch.object( product_model, 'get_product_by_id', return_value=response.Response( message={ 'product_id': product_id, 'upc': 123, 'vendor_id': 13, 'context_type': 'physical'}), autospec=True) mocker.patch.object( access, 'verify_grass_access', return_value=response.Response() ) mocker.patch.object( ows_carveouts, 'delete_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_marketing, 'delete_marketing_highlights_by_release' ) mocker.patch.object( ows_timed_release, 'delete_timed_release', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_product_physical, 'delete_physical_product', return_value=response.Response() ) result = product_logic.delete_product(product_id, account_type, account_id) assert not result def test_delete_product_no_grass_success(mocker): """Test delete product success.""" product_id = 1234 mocker.patch.object( product_model, 'get_product_by_id', return_value=response.Response( message={ 'product_id': product_id, 'upc': 123, 'context_type': 'physical'}), autospec=True) mocker.patch.object( access, 'verify_grass_access', return_value=response.Response() ) mocker.patch.object( ows_carveouts, 'delete_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_carveouts_python, 'delete_release_dms_master_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_carveouts_python, 'delete_release_dms_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_marketing, 'delete_marketing_highlights_by_release' ) mocker.patch.object( ows_timed_release, 'delete_timed_release', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_product_physical, 'delete_physical_product', return_value=response.Response() ) result = product_logic.delete_product(product_id) assert result def test_delete_product_video(mocker): """Test delete product for a video success.""" product_id = 1234 account_type = 'vendor' account_id = 12 upc = 123 video_format_id = product_constants.VIDEO_DISTRIBUTION_FORMAT_ID mocker.patch.object( product_model, 'get_product_by_id', return_value=response.Response( message={ 'product_id': product_id, 'upc': upc, 'vendor_id': account_id, 'context_type': 'digital', 'distribution_format_id': video_format_id }), autospec=True) mocker.patch.object( access, 'verify_grass_access', return_value=response.Response() ) mocker.patch.object( ows_carveouts, 'delete_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_carveouts_python, 'delete_release_dms_master_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_carveouts_python, 'delete_release_dms_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_video, 'delete_product', return_value=response.Response() ) mocker.patch.object( release_artist, 'delete', return_value=response.Response() ) mocker.patch.object( release_subgenre, 'delete', return_value=response.Response() ) product_logic.delete_product(product_id, account_type, account_id) ows_video.delete_product.assert_called_with(product_id) ows_carveouts_python.delete_release_dms_carveout.assert_called_with(product_id) ows_carveouts_python.delete_release_dms_master_carveout.assert_called_with(product_id) ows_carveouts.delete_carveout.assert_called_with(upc) release_subgenre.delete.assert_called_with(product_id) release_artist.delete.assert_called_with(product_id) @pytest.mark.parametrize( ( 'is_display_upc_used_by_account_calls', 'is_display_upc_used_by_account_responses', 'get_upcs_by_values_calls', 'get_upcs_by_values_response', 'is_feature_control_available', 'expected_response' ), [ ( [ call('account_type', 'account_id', 'display_upc', 'context_type') ], [response.create_fatal_response()], [], 'NOT RUN', False, response.create_fatal_response() ), ( [ call('account_type', 'account_id', 'display_upc', 'context_type') ], [response.Response(message=True)], [], 'NOT RUN', False, product_logic.upc_unavailable_response ), ( [ call('account_type', 'account_id', 'display_upc', 'context_type') ], [response.Response(message=False)], [call(['display_upc'])], response.create_fatal_response(), False, response.create_fatal_response() ), ( [ call('account_type', 'account_id', 'display_upc', 'context_type') ], [response.Response(message=False)], [call(['display_upc'])], response.Response(status=404), False, response.Response() ), ( [ call('account_type', 'account_id', 'display_upc', 'context_type') ], [response.Response(message=False)], [call(['display_upc'])], response.Response(message=[{'status': 'unused'}]), False, product_logic.upc_unavailable_response ), ( [ call('account_type', 'account_id', 'display_upc', 'context_type'), call('account_type', 'account_id', 'display_upc') ], [ response.Response(message=False), response.create_fatal_response() ], [call(['display_upc'])], response.Response(message=[{'status': 'used'}]), False, response.create_fatal_response() ), ( [ call('account_type', 'account_id', 'display_upc', 'context_type'), call('account_type', 'account_id', 'display_upc') ], [ response.Response(message=False), response.Response(message=False) ], [call(['display_upc'])], response.Response(message=[{'status': 'used'}]), False, product_logic.upc_unavailable_response ), ( [ call('account_type', 'account_id', 'display_upc', 'context_type'), call('account_type', 'account_id', 'display_upc') ], [ response.Response(message=False), response.Response(message=True) ], [call(['display_upc'])], response.Response(message=[{'status': 'used'}]), False, response.Response() ), ( [ call('account_type', 'account_id', 'display_upc', 'context_type'), call('account_type', 'account_id', 'display_upc') ], [ response.Response(message=False), response.Response(message=True) ], [call(['display_upc'])], response.Response(message=[{'status': 'used'}]), True, response.Response() ) ]) def test_check_display_upc_availability( mocker, is_display_upc_used_by_account_calls, is_display_upc_used_by_account_responses, get_upcs_by_values_calls, get_upcs_by_values_response, is_feature_control_available, expected_response): """Test check_display_upc_availability.""" mocker.patch.object( product_model, 'is_display_upc_used_by_account', side_effect=is_display_upc_used_by_account_responses) mocker.patch.object( upc_model, 'get_upcs_by_values', return_value=get_upcs_by_values_response) mocker.patch.object( feature_control_util, 'check_feature_control_for_subaccount', return_value=is_feature_control_available) upc_availability_response = product_logic.check_display_upc_availability( 'account_type', 'account_id', 'display_upc', 'context_type') product_model.is_display_upc_used_by_account.assert_has_calls( is_display_upc_used_by_account_calls) upc_model.get_upcs_by_values.assert_has_calls(get_upcs_by_values_calls) assert upc_availability_response.status == expected_response.status assert upc_availability_response.message == expected_response.message assert upc_availability_response.errors == expected_response.errors def test_get_products_by_vendor_id_default_status_success( mocker, test_valid_product, test_vendor, test_subaccount): """Test get list of products with model layer success. Test get list of products for a given vendor without specifiying status. """ expected_product_response = { 'items': [test_valid_product], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 1 } } mocker.patch.object( product_model, 'get_products_by_account_id', return_value=response.Response(message=expected_product_response), autospec=True) result = product_logic.get_products_for_account( vendor_id=test_vendor, subaccount_id=test_subaccount) product_model.get_products_by_account_id.assert_called_with( test_vendor, test_subaccount, status=None, page_offset=None, page_limit=None, start_date=None, end_date=None, sort_order=None, deletions=None, sort_by=None) assert result.status == 200 assert result.message == expected_product_response def test_get_products_by_vendor_id_specified_status_success( mocker, test_valid_product, test_vendor, test_subaccount): """Test get list of products with model layer success. Test get list of products for a given vendor while specifing the status. """ # Change status of test product to in_content assert test_valid_product['status'] == 'label_processing' test_valid_product['status'] = 'in_content' expected_product_response = { 'items': [test_valid_product], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 1 } } mocker.patch.object( product_model, 'get_products_by_account_id', return_value=response.Response(message=expected_product_response), autospec=True) result = product_logic.get_products_for_account( vendor_id=test_vendor, subaccount_id=test_subaccount, status='in_content') product_model.get_products_by_account_id.assert_called_with( test_vendor, test_subaccount, status='in_content', page_offset=None, page_limit=None, start_date=None, end_date=None, sort_order=None, deletions=None, sort_by=None) assert result.status == 200 assert result.message == expected_product_response assert result.message['items'][0]['status'] == 'in_content' def test_get_products_by_vendor_id_specified_deletions_success( mocker, test_valid_product, test_vendor, test_subaccount): """Test get list of products with model layer success. Test get list of products for a given vendor while specifing deletions. """ # Change status of test product to in_content assert test_valid_product['deletions'] == 'N' test_valid_product['deletions'] = 'Y' expected_product_response = { 'items': [test_valid_product], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 1 } } mocker.patch.object( product_model, 'get_products_by_account_id', return_value=response.Response(message=expected_product_response), autospec=True) result = product_logic.get_products_for_account( vendor_id=test_vendor, subaccount_id=test_subaccount, deletions='Y') product_model.get_products_by_account_id.assert_called_with( test_vendor, test_subaccount, status=None, page_offset=None, page_limit=None, start_date=None, end_date=None, sort_order=None, deletions='Y', sort_by=None) assert result.status == 200 assert result.message == expected_product_response assert result.message['items'][0]['deletions'] == 'Y' def test_get_products_by_vendor_id_fails_without_account_id(mocker): """Test get list of products without providing a vendor or subaccount. This should fail the request. """ result = product_logic.get_products_for_account( vendor_id=None, subaccount_id=None) assert result.status == 400 def test_valid_get_upcs_by_product_id(mocker): """Test get upcs for all product ids.""" expected_response = { 'items': [{ 'product_id': 2078459, 'upc': 19177321792 }] } mocker.patch.object( product_model, 'get_upcs_by_product_ids', return_value=response.Response(message=expected_response), autospec=True) product = '2078459' result = product_logic.get_upcs_by_product_ids(product) assert result.status == 200 assert result.message == expected_response def test_valid_get_upcs_by_product_ids(mocker): """Test get upcs for all product ids.""" expected_response = { 'items': [{ 'product_id': 2078459, 'upc': 19177321792 }, { 'product_id': 2078461, 'upc': 191773455477 }]} mocker.patch.object( product_model, 'get_upcs_by_product_ids', return_value=response.Response(message=expected_response), autospec=True) product = '2078459,2078461' result = product_logic.get_upcs_by_product_ids(product) assert result.status == 200 assert result.message == expected_response def test_get_upcs_by_few_invalid_product_ids(mocker): """Test get upcs for few invalid product ids.""" expected_data = { 'items': [{ 'product_id': 2078459, 'upc': 19177321792 }] } expected_response = { 'code': error.ERROR_CODE_NOT_FOUND, 'message': [{2348888888: { 'code': error.ERROR_CODE_NOT_FOUND, 'message': error.ERROR_MESSAGE_INVALID_PRODUCT_ID}}]} mocker.patch.object( product_model, 'get_upcs_by_product_ids', return_value=response.Response(message=expected_data), autospec=True) product = '2078459,2348888888' result = product_logic.get_upcs_by_product_ids(product) assert result.status == 404 assert result.errors == expected_response def test_all_invalid_get_upcs_by_product_ids(mocker): """Test get upcs for all invalid product ids.""" mocker.patch.object( product_model, 'get_upcs_by_product_ids', return_value=response.create_not_found_response( message=error.ERROR_MESSAGE_ALL_PRODUCTS_FAILED), autospec=True) product = '123456781222,1234567871' result = product_logic.get_upcs_by_product_ids(product) assert result.status == 404 def test_product_max_limit_get_upcs_by_product_ids(mocker): """Test get upcs for product ids max limit reached.""" product = random.sample(range(2078459, 2078560), 101) product = map(str, product) product_id_str = ','.join(product) result = product_logic.get_upcs_by_product_ids(product_id_str) assert result.status == 400 def test_get_upc_by_empty_product_id(mocker): """Test get upcs for empty product id.""" product = '' result = product_logic.get_upcs_by_product_ids(product) assert result.status == 404 def test_get_upc_by_none_product_id(mocker): """Test get upcs for None product id.""" product = None result = product_logic.get_upcs_by_product_ids(product) assert result.status == 404 def test_get_upcs_by_special_chars_product_ids(mocker): """Test get upcs for product ids special chars.""" product = '£20191, abc, efg123' result = product_logic.get_upcs_by_product_ids(product) assert result.status == 400 def test_get_products_by_upcs(mocker, test_products_payload): """Test get products from upcs.""" expected_response = response.Response( message={'items': test_products_payload}) mocker.patch.object( product_model, 'get_products_by_upcs', return_value=response.Response( message=test_products_payload), autospec=True) data = {'upcs': ['123456781222', '123456000000']} result = product_logic.get_products_by_upcs(data) assert result.status == expected_response.status assert result.message == expected_response.message @pytest.mark.parametrize('sort_direction', [('asc'), ('desc')]) def test_get_products_by_upcs_with_sorting( mocker, test_products_payload, sort_direction): """Test get products from upcs with sorting parameters.""" expected_response = response.Response( message={'items': test_products_payload}) if sort_direction == 'desc': test_products_payload.reverse() expected_response = response.Response( message={'items': test_products_payload}) mocker.patch.object( product_model, 'get_products_by_upcs', return_value=response.Response(message=test_products_payload), autospec=True) data = { 'upcs': ['123456781222', '123456000000'], 'sort_param': 'release_name', 'sort_direction': sort_direction } result = product_logic.get_products_by_upcs(data) assert result.status == expected_response.status assert result.message == expected_response.message def test_get_products_by_upcs_with_zero_as_upc(mocker): """Test get products for upc: 0. Zero should be handled as a proper upc.""" expected_response = response.Response(message={ 'items': [ { 'product_id': 123, 'upc': 0, 'display_upc': '0', 'product_type': 'Film', 'product_name': 'Title 1', 'context_type': 'null', 'vendor_id': 123, 'subaccount_id': 123, 'release_date': '01-01-2017' } ] }, status=200) products_payload = [ { 'product_id': 123, 'upc': 0, 'display_upc': '0', 'product_type': 'Film', 'product_name': 'Title 1', 'context_type': 'null', 'vendor_id': 123, 'subaccount_id': 123, 'release_date': '01-01-2017' } ] mocker.patch.object( product_model, 'get_products_by_upcs', return_value=response.Response( message=products_payload), autospec=True) data = {'upcs': ['0']} result = product_logic.get_products_by_upcs(data) assert result.status == expected_response.status assert result.message == expected_response.message def test_get_products_by_upcs_with_not_found_product(mocker): """Test get products for all upcs with not found product.""" mocker.patch.object( product_model, 'get_products_by_upcs', return_value=response.Response(status=404), autospec=True) data = {'upcs': ['123456781222']} result = product_logic.get_products_by_upcs(data) assert result.status == 404 def test_get_product_ids_with_no_upcs(mocker): """Test get products for empty upcs.""" data = {} result = product_logic.get_products_by_upcs(data) assert result.status == 400 assert result.errors['message'] == 'Missing upcs' def test_get_product_ids_by_empty_upcs(mocker): """Test get products for empty upcs.""" data = {'upc': []} result = product_logic.get_products_by_upcs(data) assert result.status == 400 assert result.errors['message'] == 'Missing upcs' def test_get_products_by_upcs_max_limit(mocker): """Test get products for upcs with max limit reached.""" upcs_list = ['1'] * 101 data = {'upcs': upcs_list} result = product_logic.get_products_by_upcs(data) assert result.status == 400 assert result.errors['message'] == 'Only 100 upcs accepted' @pytest.mark.parametrize('with_company_brand', (False, True, None)) def test_get_product_document_with_status(with_company_brand, mocker): """Test get product document with display status.""" product_id = 123 doc = { 'product_id': product_id, 'upc': 123456781222, 'release_status': 'in_content', 'correction_status': None, 'approval_queue_status': None, 'distribution_format_id': 2, 'vendor_id': '1234', 'vendor_owner': 'odd', 'assigned_to': 187, 'assigned_reviewer': 2903 } get_product_document_by_id_mocked = mocker.patch.object( product_model, 'get_product_document_by_id', return_value=response.Response(doc)) mocker.patch.object( ows_account, 'get_vendor_company_brand', return_value=response.Response('theorchard')) mocker.patch.object( pythonfeatures, 'get_single_feature_by_attributes', return_value=response.Response(message=split_constants.FEATURE_DISABLED)) result = product_logic.get_product_document( product_id, **({'with_company_brand': with_company_brand} if with_company_brand is not None else {}), ) get_product_document_by_id_mocked.assert_called_once_with( product_id, with_company_brand or False, False ) assert result.status == 200 assert 'display_status' in result.message assert result.message['display_status'] == product_constants.COMPLETED assert 'correction_status' not in result.message assert 'approval_queue_status' not in result.message def test_get_product_document_with_status_failed(mocker): """Test get product document with incorrect display status.""" doc = { 'product_id': 123, 'upc': 123456781222, 'release_status': 'in_content', 'correction_status': 'submitted', 'approval_queue_status': None, 'distribution_format_id': 2, 'vendor_id': '1234', 'vendor_owner': 'odd', 'assigned_to': 187, 'assigned_reviewer': 2903 } mocker.patch.object( product_model, 'get_product_document_by_id', return_value=response.Response(doc)) mocker.patch.object( ows_account, 'get_vendor_company_brand', return_value=response.Response('theorchard')) mocker.patch.object( pythonfeatures, 'get_single_feature_by_attributes', return_value=response.Response(message=split_constants.FEATURE_DISABLED)) mocker.patch.object( sentry_sdk, 'capture_exception', return_value='', autospec=True) result = product_logic.get_product_document(123) assert result.status == 200 assert 'display_status' not in result.message assert 'correction_status' not in result.message assert 'approval_queue_status' not in result.message assert sentry_sdk.capture_exception.called def test_get_product_document_with_product_video(mocker): """Test get product document with display status.""" doc = { 'product_id': 123, 'upc': 123456781222, 'release_status': 'in_content', 'correction_status': 'submitted', 'approval_queue_status': None, 'type_of_video': 'Other', 'distribution_format_id': 57, 'vendor_id': '1234', 'vendor_owner': 'odd', 'assigned_to': 187, 'assigned_reviewer': 2903 } mocker.patch.object( product_model, 'get_product_document_by_id', return_value=response.Response(doc)) mocker.patch.object( ows_account, 'get_vendor_company_brand', return_value=response.Response('theorchard')) mocker.patch.object( pythonfeatures, 'get_single_feature_by_attributes', return_value=response.Response(message=split_constants.FEATURE_DISABLED)) result = product_logic.get_product_document(123) assert result.status == 200 assert 'type_of_video' in result.message assert result.message['type_of_video'] == 'Other' def test_get_product_document_error(mocker): """Test get products for upcs with max limit reached.""" mocker.patch.object( product_model, 'get_product_document_by_id', return_value=response.Response(status=404)) result = product_logic.get_product_document(100) assert result.status == 404 def test_get_product_by_display_upc( mocker, test_valid_product): """Test get product by display_upc.""" mocker.patch.object( product_model, 'get_product_by_display_upc', return_value=response.Response( message=[test_valid_product['display_upc']])) expected_response = test_valid_product['display_upc'] result = product_logic.get_product_by_display_upc( test_valid_product['vendor_id'], test_valid_product['display_upc'], test_valid_product['context_type']) assert result.status == 200 assert result.message == expected_response def test_get_product_by_display_upc_not_found( mocker, test_valid_product): """Test logic response when no product found.""" mocker.patch.object( product_model, 'get_product_by_display_upc', return_value=response.create_not_found_response() ) result = product_logic.get_product_by_display_upc( test_valid_product['vendor_id'], test_valid_product['display_upc'], test_valid_product['context_type']) assert result.status == 404 @pytest.mark.parametrize( "show_deletions, status_code, message", [ pytest.param( False, 500, None, id="show deletions is false, so trigger error." ), pytest.param( True, 200, test_message_param, id="show deletions is True, so show success error." ) ] ) def test_get_product_by_display_upc_overlap( show_deletions: bool, status_code, message, mocker, test_valid_product ): """Test logic response when multiple products not deleted found.""" mocker.patch.object( product_model, 'get_product_by_display_upc', return_value=response.Response( message=[ test_valid_product, test_valid_product])) result = product_logic.get_product_by_display_upc( test_valid_product['vendor_id'], test_valid_product['display_upc'], test_valid_product['context_type'], show_deletions ) assert result.status == status_code assert result.message == message def test_get_product_by_display_upc_overlap_with_deletion( mocker, test_valid_product, test_deleted_product): """Test logic response when multiple products found with 1 deletion.""" mocker.patch.object( product_model, 'get_product_by_display_upc', return_value=response.Response( message=[ test_valid_product, test_deleted_product])) result = product_logic.get_product_by_display_upc( test_valid_product['vendor_id'], test_valid_product['display_upc'], test_valid_product['context_type'], True ) assert result.status == 200 assert result.message == test_valid_product def test_get_product_by_display_upc_overlap_with_all_deletion( mocker, test_deleted_product): """Test logic response when multiple deleted products found.""" mocker.patch.object( product_model, 'get_product_by_display_upc', return_value=response.Response( message=[ test_deleted_product, test_deleted_product])) result = product_logic.get_product_by_display_upc( test_deleted_product['vendor_id'], test_deleted_product['display_upc'], test_deleted_product['context_type'], True ) assert result.status == 200 assert result.message == test_deleted_product @patch('product.connectors.mysql._db_session') def test_get_product_by_diplay_upc_db_failure( db_session, mocker, test_valid_product): """Test fatal db error.""" mocker.patch.object( sentry_sdk, 'capture_exception', return_value='', autospec=True) session = MagicMock() session.query = MagicMock(side_effect=SQLAlchemyError()) db_session.return_value = session display_upc = test_valid_product['display_upc'] vendor_id = test_valid_product['vendor_id'] context_type = test_valid_product['context_type'] result = product_logic.get_product_by_display_upc( vendor_id, display_upc, context_type) assert result.status == 500 assert sentry_sdk.capture_exception.called def test_get_product_by_display_upc_exception( mocker, test_valid_product): """Test display_upc response when model returns an error response.""" mocker.patch.object( product_model, 'get_product_by_display_upc', return_value=response.create_fatal_response()) result = product_logic.get_product_by_display_upc( test_valid_product['vendor_id'], test_valid_product['display_upc'], test_valid_product['context_type']) assert result.status == 500 @pytest.mark.parametrize( ('product_doc', 'expected_display_status'), [ ( { 'release_status': product_constants.LABEL_PROCESSING, 'correction_status': None, 'approval_queue_status': None }, product_constants.IN_PROGRESS ), ( { 'release_status': product_constants.ORCHARD_PROCESSING, 'correction_status': None, 'approval_queue_status': None }, product_constants.IN_PROGRESS ), ( { 'release_status': product_constants.TRANSFER_TO_CONTENT, 'correction_status': None, 'approval_queue_status': product_constants.CHECKED_IN }, product_constants.SUBMITTED ), ( { 'release_status': product_constants.IN_CONTENT, 'correction_status': product_constants.SUBMITTED, 'approval_queue_status': product_constants.CHECKED_IN }, product_constants.SUBMITTED ), ( { 'release_status': product_constants.TRANSFER_TO_CONTENT, 'correction_status': None, 'approval_queue_status': product_constants.CHECKED_OUT }, product_constants.SUBMITTED ), ( { 'release_status': product_constants.IN_CONTENT, 'correction_status': product_constants.SUBMITTED, 'approval_queue_status': product_constants.CHECKED_OUT }, product_constants.SUBMITTED ), ( { 'release_status': product_constants.TRANSFER_TO_CONTENT, 'correction_status': None, 'approval_queue_status': None }, product_constants.SUBMITTED ), ( { 'release_status': product_constants.TRANSFER_TO_CONTENT, 'correction_status': None, 'approval_queue_status': product_constants.REJECTED, }, product_constants.ACTION_REQUIRED ), ( { 'release_status': product_constants.LABEL_PROCESSING, 'correction_status': None, 'approval_queue_status': product_constants.REJECTED, }, product_constants.ACTION_REQUIRED ), ( { 'release_status': product_constants.ORCHARD_PROCESSING, 'correction_status': None, 'approval_queue_status': product_constants.REJECTED, }, product_constants.ACTION_REQUIRED ), ( { 'release_status': product_constants.IN_CONTENT, 'correction_status': product_constants.APPLIED, 'approval_queue_status': None }, product_constants.COMPLETED ), ( { 'release_status': product_constants.IN_CONTENT, 'correction_status': None, 'approval_queue_status': None }, product_constants.COMPLETED ), ( { 'release_status': product_constants.IN_CONTENT, 'correction_status': product_constants.ACTIVE, 'approval_queue_status': product_constants.CHECKED_IN }, product_constants.ERROR_CORRECTION ), ( { 'release_status': product_constants.IN_CONTENT, 'correction_status': product_constants.ACTIVE, 'approval_queue_status': product_constants.CHECKED_OUT }, product_constants.ERROR_CORRECTION ), ( { 'release_status': product_constants.IN_CONTENT, 'correction_status': product_constants.ACTIVE, 'approval_queue_status': product_constants.APPROVED }, product_constants.ERROR_CORRECTION ), ( { 'release_status': product_constants.IN_CONTENT, 'correction_status': product_constants.ACTIVE, 'approval_queue_status': None }, product_constants.ERROR_CORRECTION ), ( { 'release_status': product_constants.IN_CONTENT, 'correction_status': product_constants.ACTIVE, 'approval_queue_status': product_constants.REJECTED }, product_constants.ACTION_REQUIRED ), ] ) def test_calculate_display_status(product_doc, expected_display_status, mocker): """Test calculate valid display statuses.""" product_doc.update({ 'distribution_format_id': '1', 'vendor_id': 123, 'vendor_owner': 'odd', 'assigned_to': 187, 'assigned_reviewer': 2903 }) mocker.patch.object( ows_account, 'get_vendor_company_brand', return_value=response.Response('theorchard')) mocker.patch.object( pythonfeatures, 'get_single_feature_by_attributes', return_value=response.Response(message=split_constants.FEATURE_DISABLED)) display_status = product_logic.calculate_display_status(product_doc) assert display_status == expected_display_status def test_calculate_display_status_content_review(mocker): """Test calculate valid display statuses with content review enabled.""" product_doc = { 'release_id': 1234, 'release_status': product_constants.TRANSFER_TO_CONTENT, 'correction_status': product_constants.SUBMITTED, 'approval_queue_status': None, 'distribution_format_id': '1', 'vendor_id': 123, 'vendor_owner': 'odd', 'assigned_to': 187, 'assigned_reviewer': 2903 } mocker.patch.object( ows_account, 'get_vendor_company_brand', return_value=response.Response('theorchard')) mocker.patch.object( pythonfeatures, 'get_single_feature_by_attributes', return_value=response.Response(message=split_constants.FEATURE_ENABLED)) mocker.patch.object( ows_product_review, 'get_product_review_status', return_value=response.Response({ 'release_approval_status': product_constants.CHECKED_IN })) display_status = product_logic.calculate_display_status(product_doc) assert display_status == product_constants.SUBMITTED def test_calculate_display_status_content_review_error(mocker): """Test calculate display statuses with content review enabled.""" product_doc = { 'release_id': 1234, 'release_status': product_constants.TRANSFER_TO_CONTENT, 'correction_status': product_constants.SUBMITTED, 'approval_queue_status': None, 'distribution_format_id': '1', 'vendor_id': 123, 'assigned_to': 187, 'assigned_reviewer': 2903 } mocker.patch.object( pythonfeatures, 'get_single_feature_by_attributes', return_value=response.Response(message=split_constants.FEATURE_ENABLED)) mocker.patch.object( ows_product_review, 'get_product_review_status', return_value=response.Response({ 'release_approval_status': product_constants.CHECKED_IN })) with pytest.raises(Exception): product_logic.calculate_display_status(product_doc) @pytest.mark.parametrize( 'product_doc', [ ( { 'release_status': None, 'correction_status': None, 'approval_queue_status': None, } ), ( { 'release_status': product_constants.TRANSFER_TO_CONTENT, 'correction_status': None, 'approval_queue_status': product_constants.APPROVED } ), ( { 'release_status': product_constants.IN_CONTENT, 'correction_status': product_constants.SUBMITTED, 'approval_queue_status': None } ) ] ) def test_calculate_invalid_display_status(product_doc, mocker): """Test calculate invalid display statuses.""" product_doc.update({ 'distribution_format_id': '1', 'vendor_id': 123, 'vendor_owner': 'odd', 'assigned_to': 187, 'assigned_reviewer': 2903 }) mocker.patch.object( ows_account, 'get_vendor_company_brand', return_value=response.Response('theorchard')) mocker.patch.object( pythonfeatures, 'get_single_feature_by_attributes', return_value=response.Response(message=split_constants.FEATURE_DISABLED)) with pytest.raises(product_logic.IncorrectDisplayStatusError): product_logic.calculate_display_status(product_doc) def test_check_products_ownership_by_upcs_error(mocker): """Test product ownership without providing a vendor or subaccount.""" upcs = ['12345', '2343242'] result = product_logic.check_products_ownership_by_upcs( upcs, vendor_id=None, subaccount_id=None) assert result.status == 400 @pytest.mark.parametrize( ('vendor_id', 'subaccount_id', 'upcs', 'result', 'status'), [ (1, None, ['888831283041'], [{'upc': '888831283041'}], 200), (1, None, ['888831283044'], [], 404), (None, 5, ['888831283047'], [{'upc': '888831283047'}], 200) ]) def test_check_products_ownership_by_upcs( mocker, vendor_id, subaccount_id, upcs, result, status): """Test products ownership by upc(s) and account id.""" product_response = response.Response(message=result) mocker.patch.object( product_model, 'get_products_by_accountid_and_upcs', return_value=product_response ) res = product_logic.check_products_ownership_by_upcs( upcs, vendor_id=vendor_id, subaccount_id=subaccount_id) assert res.status == status def test_get_products_by_isrc(mocker, test_valid_product): """Test get products by isrc.""" expected_product_response = { 'items': [test_valid_product], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 1 } } mocker.patch.object( product_model, 'get_products_by_isrc', return_value=response.Response( message=expected_product_response)) result = product_logic.get_products_by_isrc('GBAYE1234567') assert result.status == 200 assert result.message == expected_product_response def test_get_product_by_release_artist_id(mocker, test_valid_product): """Test get product by release_artist_id.""" expected_product_response = test_valid_product mocker.patch.object( product_model, 'get_product_by_release_artist_id', return_value=response.Response( message=expected_product_response)) result = product_logic.get_product_by_release_artist_id(12345) assert result.status == 200 assert result.message == expected_product_response @patch('product.logic.product.release_country_dates.delete_all_by_product_id') @patch('product.logic.product.release_country_dates.insert_dates') @patch('product.logic.product.country.get_country_ids') @patch('product.logic.product.product_model.get_product_by_id') def test_replace_product_territory_dates( mock_get_product_id, mock_get_country_ids, mock_insert_dates, mock_delete_dates): """Test replace_product_territory_dates.""" mock_get_product_id.return_value = response.Response( message={'product_id': 2342, 'upc': '434523452435'}) data = [ { 'release_date': '2020-08-12', 'country_codes': ['US', 'GB'], }, { 'release_date': '2020-08-23', 'country_codes': ['BE', 'DE', 'SS'], }, { 'country_codes': ['FR', 'CA'], }, ] expected_data = [ { 'release_date': '2020-08-12', 'country_id': 1, 'release_id': 2342, 'upc': '434523452435', 'sale_start_date': None, 'vod_start_date': None, 'preorder_date': None, }, { 'release_date': '2020-08-12', 'country_id': 2, 'release_id': 2342, 'upc': '434523452435', 'sale_start_date': None, 'vod_start_date': None, 'preorder_date': None, }, { 'release_date': '2020-08-23', 'country_id': 3, 'release_id': 2342, 'upc': '434523452435', 'sale_start_date': None, 'vod_start_date': None, 'preorder_date': None, }, { 'release_date': '2020-08-23', 'country_id': 4, 'release_id': 2342, 'upc': '434523452435', 'sale_start_date': None, 'vod_start_date': None, 'preorder_date': None, }, ] mock_get_country_ids.return_value = [ { 'id': 1, 'country_code': 'US', }, { 'id': 2, 'country_code': 'GB', }, { 'id': 3, 'country_code': 'BE', }, { 'id': 4, 'country_code': 'DE', }, ] product_logic.replace_product_territory_dates(2342, data) mock_get_product_id.assert_called_with(2342) mock_delete_dates.assert_called_with(2342) mock_insert_dates.assert_called_with(expected_data) def test_delete_product_for_digital_product(mocker): """Test delete product for a digital product success.""" product_id = 1234 account_type = 'vendor' account_id = 12 upc = 123 digital_product_format_id = product_constants.MUSIC_DISTRIBUTION_FORMAT_ID mocker.patch.object( product_model, 'get_product_by_id', return_value=response.Response( message={ 'product_id': product_id, 'upc': upc, 'vendor_id': account_id, 'context_type': 'digital', 'distribution_format_id': digital_product_format_id }), autospec=True) mocker.patch.object( access, 'verify_grass_access', return_value=response.Response() ) mocker.patch.object( ows_carveouts, 'delete_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_carveouts_python, 'delete_release_dms_master_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_carveouts_python, 'delete_release_dms_carveout', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_timed_release, 'delete_timed_release', return_value=response.Response(), autospec=True ) mocker.patch.object( ows_product_digital, 'delete_digital_product', return_value=response.Response() ) product_logic.delete_product(product_id, account_type, account_id) ows_product_digital.delete_digital_product.assert_called_with( product_id, account_type, account_id) ows_carveouts_python.delete_release_dms_carveout.assert_called_with(product_id) ows_carveouts_python.delete_release_dms_master_carveout.assert_called_with(product_id) ows_carveouts.delete_carveout.assert_called_with(upc) def test_get_product_id_by_upc_dataloaded_success(mocker): """Test get_product_id_by_upc_dataloaded returns product ids for valid upcs.""" upcs = ['123456789012', '987654321098'] expected_items = [ {'upc': 123456789012, 'product_id': 1}, {'upc': 987654321098, 'product_id': 2} ] mocker.patch.object( product_model, 'get_product_id_by_upc_dataloaded', return_value=response.Response(message=expected_items) ) data = {'upcs': upcs} result = product_logic.get_product_id_by_upc_dataloaded(data) assert result.status == 200 assert result.message == {'items': expected_items} def test_get_product_id_by_upc_dataloaded_missing_upcs(mocker): """Test get_product_id_by_upc_dataloaded returns error if 'upcs' missing.""" data = {} result = product_logic.get_product_id_by_upc_dataloaded(data) assert result.status == 400 assert result.errors['message'] == error.ERROR_MESSAGE_MISSING_UPC def test_get_product_id_by_upc_dataloaded_empty_upcs(mocker): """Test get_product_id_by_upc_dataloaded returns error if upcs is empty.""" data = {'upcs': []} result = product_logic.get_product_id_by_upc_dataloaded(data) assert result.status == 400 assert result.errors['message'] == error.ERROR_MESSAGE_MISSING_UPC def test_get_product_id_by_upc_dataloaded_max_limit(mocker): """Test get_product_id_by_upc_dataloaded returns error if upcs exceeds limit.""" upcs = [str(i) for i in range(int(product_logic.config.LIMIT) + 1)] data = {'upcs': upcs} result = product_logic.get_product_id_by_upc_dataloaded(data) assert result.status == 400 assert result.errors['message'] == error.ERROR_MESSAGE_MAX_UPC def test_get_product_id_by_upc_dataloaded_no_results(mocker): """Test get_product_id_by_upc_dataloaded returns empty if no results.""" upcs = ['123456789012'] mocker.patch.object( product_model, 'get_product_id_by_upc_dataloaded', return_value=None ) data = {'upcs': upcs} result = product_logic.get_product_id_by_upc_dataloaded(data) assert result is None @pytest.mark.parametrize( 'mock_model_response, expected_response', [ pytest.param( response.Response(message=[ {'product_id': 1, 'tenant_type': 'account', 'tenant_id': 7123}, {'product_id': 2, 'tenant_type': 'subaccount', 'tenant_id': 4890}, {'product_id': 3, 'tenant_type': 'account', 'tenant_id': 6971}, ]), response.Response(message={'product_ids': [ {'product_id': 1, 'tenant_type': 'account', 'tenant_id': 7123}, {'product_id': 2, 'tenant_type': 'subaccount', 'tenant_id': 4890}, {'product_id': 3, 'tenant_type': 'account', 'tenant_id': 6971}, ]}), id='Return the ownership list in a dictionary keyed with product_ids' ), pytest.param( response.create_fatal_response(), response.Response(message={'product_ids': [None, None, None]}), id='Return dataloader-ed Nones when there has been an error fetching ownership', ), ] ) @patch('product.logic.product.product_model.get_product_ownership_by_product_id_dataloaded') def test_lookup_product_ownership( mock_get_product_ownership_by_product_id_dataloaded: MagicMock, mock_model_response: response.Response, expected_response: response.Response, ) -> None: """Test lookup_product_ownership.""" mock_get_product_ownership_by_product_id_dataloaded.return_value = mock_model_response product_ids = [1, 2, 3] actual = product_logic.lookup_product_ownership(product_ids) assert actual.status == expected_response.status assert actual.message == expected_response.message assert actual.errors == expected_response.errors mock_get_product_ownership_by_product_id_dataloaded.assert_called_once_with(product_ids)