"""Test for Product model.""" from unittest.mock import MagicMock from unittest.mock import patch import pytest import sentry_sdk from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm.session import Session from product.constants import error from product.models import product from product.models.sql import product_document from product.models.sql import products_by_isrc from product.models.sql import products_documents from product.models.sql.product_document import PRODUCT_TENANT_UUIDS from tests.testutils import db non_deleted_product = { 'product_id': 1, 'upc': 888831283041, 'display_upc': '888831283041', 'subaccount_id': 1, 'distribution_format_id': 2, 'product_type_id': 1, 'project_id': 22, 'vendor_id': 1, 'context_type': 'physical', 'release_date': '2017-09-09', 'status': 'label_processing', 'product_name': 'product test 1', 'deletions': 'N', 'not_for_distribution': 'N' } deleted_product = { 'context_type': 'physical', 'deletions': 'Y', 'display_upc': '888831283041', 'distribution_format_id': 2, 'not_for_distribution': 'N', 'product_id': 4, 'product_name': None, 'product_type_id': 1, 'project_id': 22, 'release_date': None, 'status': 'label_processing', 'subaccount_id': 1, 'upc': 888831283041, 'vendor_id': 1 } @db.test_schema def test_get_product_by_id(): """Test a product for the specified product_id is returned.""" expected_response = { 'product_id': 1, 'upc': 888831283041, 'display_upc': '888831283041', 'subaccount_id': 1, 'distribution_format_id': 2, 'product_type_id': 1, 'project_id': 22, 'vendor_id': 1, 'context_type': 'physical', 'release_date': '2017-09-09', 'status': 'label_processing', 'product_name': 'product test 1', 'deletions': 'N', 'not_for_distribution': 'N' } db.insert_vw_product() response = product.get_product_by_id(1) assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_product_not_found(): """Test a 404 is returned when a product id is not found.""" response = product.get_product_by_id(9999) assert response.status == 404 @db.test_schema def test_get_product_by_upc_none_found(test_upc): """Test that 404 is returned when no product has the given UPC.""" result = product.get_product_by_upc(test_upc) assert result.status == 404 assert result.message is None @db.test_schema def test_get_product_by_upc_found(): """Test that 200 with payload is returned when a product is found.""" fixture_upc = 888831283041 expected_response = { 'product_id': 1, 'upc': 888831283041, 'display_upc': '888831283041', 'subaccount_id': 1, 'distribution_format_id': 2, 'product_type_id': 1, 'project_id': 22, 'vendor_id': 1, 'context_type': 'physical', 'release_date': '2017-09-09', 'status': 'label_processing', 'product_name': 'product test 1', 'deletions': 'N', 'not_for_distribution': 'N' } db.insert_vw_product() result = product.get_product_by_upc(fixture_upc) assert result.status == 200 assert result.message == expected_response @db.test_schema def test_get_product_by_upc_found_when_duplicate_upc_found(): """Test that 200 with payload is returned when a product is found.""" fixture_upc = 888831283041 db.insert_vw_product() db.insert_duplicate_vw_product() result = product.get_product_by_upc(fixture_upc) assert result.status == 500 @db.test_schema def test_get_first_product_by_product_code_for_vendor(): """Return product matching product_code for a vendor.""" expected_response = { 'product_id': 89, 'upc': 888831283044, 'display_upc': '888831283044', 'subaccount_id': 5, 'distribution_format_id': 2, 'product_type_id': 1, 'project_id': 33, 'vendor_id': 8, 'context_type': 'digital', 'release_date': '2017-09-09', 'status': 'label_processing', 'product_name': 'product test 89', 'deletions': 'N', 'not_for_distribution': 'N' } db.insert_vw_product() response = product.get_first_product_by_product_code_for_account( 'prod-code-something', 'vendor', 8) assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_first_product_by_product_code_for_vendor_multiple(): """Return first product matching product_code for a vendor.""" expected_response = { 'product_id': 91, 'upc': 888831283046, 'display_upc': '888831283046', 'subaccount_id': 5, 'distribution_format_id': 2, 'product_type_id': 1, 'project_id': 33, 'vendor_id': 8, 'context_type': 'digital', 'release_date': '2017-09-09', 'status': 'label_processing', 'product_name': 'product test 91', 'deletions': 'N', 'not_for_distribution': 'N' } db.insert_vw_product() response = product.get_first_product_by_product_code_for_account( 'prod-code-something-twice', 'vendor', 8) assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_first_product_by_product_code_for_subaccount(): """Return product matching product_code for a subaccount.""" expected_response = { 'product_id': 90, 'upc': 888831283045, 'display_upc': '888831283045', 'subaccount_id': 8, 'distribution_format_id': 2, 'product_type_id': 1, 'project_id': 33, 'vendor_id': 5, 'context_type': 'digital', 'release_date': '2017-09-17', 'status': 'label_processing', 'product_name': 'product test 90', 'deletions': 'N', 'not_for_distribution': 'N' } db.insert_vw_product() response = product.get_first_product_by_product_code_for_account( 'prod-code-something', 'subaccount', 8) assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_first_product_by_product_code_not_found(): """Return 404 when product code not found.""" db.insert_vw_product() response = product.get_first_product_by_product_code_for_account( 'not-found-code', 'subaccount', 8) assert response.status == 404 assert response.message is None @db.test_schema def test_get_products_by_product_code_for_vendor(): db.insert_vw_product() db.insert_vendor() db.insert_subaccount() response = product.get_products_by_product_code_for_account( ['prod-code', 'code-cool', 'prod-code-2'], 'vendor', '51be2aab-33c4-4822-b983-a7b6a4fb33f1' ) assert response.status == 200 assert response.message == ['prod-code', 'prod-code-2'] response = product.get_products_by_product_code_for_account( ['doesnt-exist', 'code-cool'], 'vendor', '51be2aab-33c4-4822-b983-a7b6a4fb33f1' ) assert response.status == 200 assert response.message == [] @db.test_schema def test_get_products_by_product_code_for_subaccount(): db.insert_vw_product() db.insert_vendor() db.insert_subaccount() response = product.get_products_by_product_code_for_account( ['code-cool', 'prod-code-something', 'prod-code-something-twice'], 'subaccount', '97218691-610b-441a-9822-45a1e11a1b49', ) assert response.status == 200 assert response.message == ['prod-code-something', 'prod-code-something-twice'] response = product.get_products_by_product_code_for_account( ['doesnt-exist', 'code-cool'], 'subaccount', '97218691-610b-441a-9822-45a1e11a1b49' ) assert response.status == 200 assert response.message == [] @db.test_schema def test_get_product_id_by_product_code_for_vendor(): db.insert_vw_product() db.insert_vendor() db.insert_subaccount() # success response = product.get_product_id_by_product_code_for_account( 'prod-code-something-twice', 'vendor', '7851e7af-3b16-4648-a69e-aaf33596a8c6' ) assert response.status == 200 assert response.message == 91 # not found response = product.get_product_id_by_product_code_for_account( 'doesnt-exist', 'vendor', '51be2aab-33c4-4822-b983-a7b6a4fb33f1' ) assert response.status == 404 # multiple records found response = product.get_product_id_by_product_code_for_account( 'prod-code-something-twice', 'vendor', 'f372bfe8-c07d-4389-9ff1-3b250dac7c5a' ) assert response.status == 400 assert response.errors == { "code": error.ERROR_CODE_BAD_REQUEST, "message": error.ERROR_MESSAGE_MULTIPLE_PRODUCTS_FOUND } @db.test_schema def test_get_product_id_by_product_code_for_subaccount(): db.insert_vw_product() db.insert_vendor() db.insert_subaccount() # success response = product.get_product_id_by_product_code_for_account( 'prod-code-something', 'subaccount', '488c8115-224a-421d-a0c2-6aeb0aab18a6', ) assert response.status == 200 assert response.message == 90 # not found response = product.get_product_id_by_product_code_for_account( 'prod-code-3', 'subaccount', '97218691-610b-441a-9822-45a1e11a1b49' ) assert response.status == 404 @db.test_schema def test_get_products_by_vendor_id_default_status(test_vendor): """Return products matching vendor_id and status of label_processing.""" expected_response = {'items': [{ 'distribution_format_id': 2, 'upc': 888831283048, 'display_upc': '888831283048', 'product_name': 'product test 93', 'release_date': '2017-09-17', 'product_id': 93, 'status': 'label_processing', 'vendor_id': 1234, 'project_id': 33, 'subaccount_id': 0, 'context_type': 'physical', 'product_type_id': 1, 'deletions': 'N', 'not_for_distribution': 'N' }, { 'distribution_format_id': 2, 'upc': 888831283049, 'display_upc': '888831283049', 'product_name': 'product test 94', 'release_date': '2017-09-17', 'product_id': 94, 'status': 'label_processing', 'vendor_id': 1234, 'project_id': 33, 'subaccount_id': 0, 'context_type': 'physical', 'product_type_id': 1, 'deletions': 'Y', 'not_for_distribution': 'N' }, { 'context_type': 'physical', 'deletions': 'Y', 'distribution_format_id': 2, 'product_id': 2078461, 'product_name': 'product test 2078461', 'product_type_id': 1, 'project_id': 33, 'release_date': '2017-09-17', 'status': 'label_processing', 'subaccount_id': 1, 'upc': 191773455477, 'display_upc': '191773455477', 'vendor_id': 1234, 'not_for_distribution': 'N' }, { 'distribution_format_id': 2, 'upc': 888831283123, 'display_upc': '888831283123', 'product_name': 'product test 10', 'release_date': None, 'product_id': 10, 'status': 'label_processing', 'vendor_id': 1234, 'project_id': 22, 'subaccount_id': 0, 'context_type': 'physical', 'product_type_id': 1, 'deletions': 'Y', 'not_for_distribution': 'N' }, { 'context_type': 'physical', 'deletions': 'Y', 'distribution_format_id': 2, 'product_id': 2078459, 'product_name': 'product test 2078459', 'product_type_id': 1, 'project_id': 22, 'release_date': None, 'status': 'label_processing', 'subaccount_id': 1, 'upc': 191773455453, 'display_upc': '191773455453', 'vendor_id': 1234, 'not_for_distribution': 'N' }], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 5} } db.insert_vw_product() response = product.get_products_by_account_id(test_vendor) assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_products_by_vendor_id_specified_status(test_vendor): """Return products matching vendor_id and specified status (in_content).""" expected_response = {'items': [{ 'product_id': 92, 'upc': 888831283047, 'display_upc': '888831283047', 'subaccount_id': 5, 'distribution_format_id': 2, 'product_type_id': 1, 'project_id': 33, 'vendor_id': 1234, 'context_type': 'digital', 'release_date': '2017-09-17', 'status': 'in_content', 'product_name': 'product test 92', 'deletions': 'N', 'not_for_distribution': 'N' }], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 1} } db.insert_vw_product() response = product.get_products_by_account_id( test_vendor, status='in_content') assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_products_by_vendor_id_all_status(test_vendor): """Return products matching vendor_id and specified status (in_content).""" expected_response = {'items': [ { 'context_type': 'digital', 'deletions': 'N', 'distribution_format_id': 2, 'product_id': 92, 'product_name': 'product test 92', 'product_type_id': 1, 'project_id': 33, 'release_date': '2017-09-17', 'status': 'in_content', 'subaccount_id': 5, 'upc': 888831283047, 'display_upc': '888831283047', 'vendor_id': 1234, 'not_for_distribution': 'N' }, { 'context_type': 'physical', 'deletions': 'N', 'distribution_format_id': 2, 'product_id': 93, 'product_name': 'product test 93', 'product_type_id': 1, 'project_id': 33, 'release_date': '2017-09-17', 'status': 'label_processing', 'subaccount_id': 0, 'upc': 888831283048, 'display_upc': '888831283048', 'vendor_id': 1234, 'not_for_distribution': 'N' }, { 'context_type': 'physical', 'deletions': 'Y', 'distribution_format_id': 2, 'product_id': 94, 'product_name': 'product test 94', 'product_type_id': 1, 'project_id': 33, 'release_date': '2017-09-17', 'status': 'label_processing', 'subaccount_id': 0, 'upc': 888831283049, 'display_upc': '888831283049', 'vendor_id': 1234, 'not_for_distribution': 'N' }, { 'context_type': 'physical', 'deletions': 'Y', 'distribution_format_id': 2, 'product_id': 2078461, 'product_name': 'product test 2078461', 'product_type_id': 1, 'project_id': 33, 'release_date': '2017-09-17', 'status': 'label_processing', 'subaccount_id': 1, 'upc': 191773455477, 'display_upc': '191773455477', 'vendor_id': 1234, 'not_for_distribution': 'N' }, { 'context_type': 'physical', 'deletions': 'Y', 'distribution_format_id': 2, 'product_id': 10, 'product_name': 'product test 10', 'product_type_id': 1, 'project_id': 22, 'release_date': None, 'status': 'label_processing', 'subaccount_id': 0, 'upc': 888831283123, 'display_upc': '888831283123', 'vendor_id': 1234, 'not_for_distribution': 'N' }, { 'context_type': 'physical', 'deletions': 'Y', 'distribution_format_id': 2, 'product_id': 2078459, 'product_name': 'product test 2078459', 'product_type_id': 1, 'project_id': 22, 'release_date': None, 'status': 'label_processing', 'subaccount_id': 1, 'upc': 191773455453, 'display_upc': '191773455453', 'vendor_id': 1234, 'not_for_distribution': 'N' }], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 6} } db.insert_vw_product() response = product.get_products_by_account_id( test_vendor, status='all') assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_products_by_vendor_id_page_offset(): """Return products matching vendor_id for specific page offset.""" db.insert_vw_product() full_response = product.get_products_by_account_id(1) response = product.get_products_by_account_id( 1, page_offset=1) all_products = full_response.message.get('items') offset_products = response.message.get('items') assert len(all_products) == 3 assert len(offset_products) == 2 assert all_products[1]['product_id'] == offset_products[0]['product_id'] @db.test_schema def test_get_products_by_vendor_id_page_limit(): """Return products matching vendor_id for specific page limit.""" db.insert_vw_product() full_response = product.get_products_by_account_id(1) limit = 1 response = product.get_products_by_account_id( 1, page_limit=limit) all_products = full_response.message.get('items') limited_products = response.message.get('items') assert len(all_products) == 3 assert len(limited_products) == limit @db.test_schema def test_get_products_by_vendor_id_invalid_status(): """Test error is returned for invalid product status filter.""" db.insert_vw_product() response = product.get_products_by_account_id( 1, status='random') assert response.status == 400 assert response.errors.get('message') == error.ERROR_MESSAGE_INVALID_STATUS @db.test_schema def test_get_products_by_vendor_id_no_results(): """Test error is returned if no products are found for given vendor.""" db.insert_vw_product() response = product.get_products_by_account_id( 99999) products = response.message.get('items') assert response.status == 200 assert len(products) == 0 @db.test_schema def test_get_products_by_subaccount_id_with_status(test_subaccount): """Return products matching subaccount_id.""" expected_response = {'items': [{ 'product_id': 92, 'upc': 888831283047, 'display_upc': '888831283047', 'subaccount_id': 5, 'distribution_format_id': 2, 'product_type_id': 1, 'project_id': 33, 'vendor_id': 1234, 'context_type': 'digital', 'release_date': '2017-09-17', 'status': 'in_content', 'product_name': 'product test 92', 'deletions': 'N', 'not_for_distribution': 'N' }], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 1} } db.insert_vw_product() response = product.get_products_by_account_id( subaccount_id=test_subaccount, status='in_content') assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_products_by_subaccount_id_all_status(test_subaccount): """Return products matching subaccount_id.""" expected_response = {'items': [ { 'product_id': 92, 'upc': 888831283047, 'display_upc': '888831283047', 'subaccount_id': 5, 'distribution_format_id': 2, 'product_type_id': 1, 'project_id': 33, 'vendor_id': 1234, 'context_type': 'digital', 'release_date': '2017-09-17', 'status': 'in_content', 'product_name': 'product test 92', 'deletions': 'N', 'not_for_distribution': 'N' }, { 'context_type': 'digital', 'deletions': 'N', 'distribution_format_id': 2, 'product_id': 89, 'product_name': 'product test 89', 'product_type_id': 1, 'project_id': 33, 'release_date': '2017-09-09', 'status': 'label_processing', 'subaccount_id': 5, 'upc': 888831283044, 'display_upc': '888831283044', 'vendor_id': 8, 'not_for_distribution': 'N' }, { 'context_type': 'digital', 'deletions': 'N', 'distribution_format_id': 2, 'product_id': 91, 'product_name': 'product test 91', 'product_type_id': 1, 'project_id': 33, 'release_date': '2017-09-09', 'status': 'label_processing', 'subaccount_id': 5, 'upc': 888831283046, 'display_upc': '888831283046', 'vendor_id': 8, 'not_for_distribution': 'N' } ], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 3} } db.insert_vw_product() response = product.get_products_by_account_id( subaccount_id=test_subaccount, status='all') assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_products_by_vendor_id_with_deletions(test_vendor): """Return products matching vendor_id and deletions of N.""" expected_response = {'items': [{ 'distribution_format_id': 2, 'upc': 888831283048, 'display_upc': '888831283048', 'product_name': 'product test 93', 'release_date': '2017-09-17', 'product_id': 93, 'status': 'label_processing', 'vendor_id': 1234, 'project_id': 33, 'subaccount_id': 0, 'context_type': 'physical', 'product_type_id': 1, 'deletions': 'N', 'not_for_distribution': 'N' }], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 1} } db.insert_vw_product() response = product.get_products_by_account_id(test_vendor, deletions='N') assert response.status == 200 assert response.message == expected_response @db.test_schema @pytest.mark.parametrize( ('account_type', 'account_id', 'display_upc', 'context_type', 'expected_message'), [ ('vendor', 1, 888831283040, 'physical', False), ('subaccount', 1, 888831283040, 'physical', False), ('vendor', 1, 888831283041, 'physical', True), ('vendor', 1, '0888831283041', 'physical', True), ('vendor', 55, 1234567654321, 'digital', True), ('vendor', 55, 1234567654321, None, True), ('vendor', 55, 888831283040, None, False) ]) def test_is_display_upc_used_by_account( account_type, account_id, display_upc, context_type, expected_message): """Test is_display_upc_used_by_account.""" db.insert_vw_product() response = product.is_display_upc_used_by_account( account_type, account_id, display_upc, context_type) assert response.status == 200 assert response.message == expected_message @patch('product.connectors.mysql._db_session') def test_is_display_upc_used_by_account_db_failure(db_session, mocker): """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 result = product.is_display_upc_used_by_account('', 101010, '', '') assert result.status == 500 assert sentry_sdk.capture_exception.called @db.test_schema def test_get_upcs_by_product_ids(): """Test fetching upcs for all the product ids.""" expected_response = {'items': [ {'product_id': 2078459, 'upc': 191773455453}, {'product_id': 2078461, 'upc': 191773455477}]} db.insert_vw_product() product_ids = [2078459, 2078461] response = product.get_upcs_by_product_ids(product_ids) assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_upcs_by_product_ids_for_not_found(): """Test upcs not found for product ids.""" product_ids = [100] response = product.get_upcs_by_product_ids(product_ids) assert response.status == 404 @patch('product.connectors.mysql._db_session') def test_get_upcs_by_product_ids_db_failure( db_session, mocker): """Test fatal db error for fetching upcs for product ids.""" mocker.patch.object( sentry_sdk, 'capture_exception', return_value='', autospec=True) session = MagicMock() session.query = MagicMock(side_effect=SQLAlchemyError()) db_session.return_value = session product_ids = [100] result = product.get_upcs_by_product_ids(product_ids) assert result.status == 500 @db.test_schema def test_get_upcs_by_product_ids_empty(): """Test upcs not found for empty product ids.""" product_ids = [] response = product.get_upcs_by_product_ids(product_ids) assert response.status == 404 @db.test_schema def test_get_products_by_upcs(test_products_by_upcs): """Test get products from upcs.""" expected_response = test_products_by_upcs upcs = [888831283041, 888831283042] db.insert_product_type() db.create_vw_product_table_format_and_artist_name() db.insert_vw_product(query=db.INSERT_PRODUCT_FORMAT_ARTIST_NAME) expected_response[0]['format'] = 'Full Length' expected_response[0]['artist_name'] = 'Artist 3' expected_response[1]['format'] = 'EP' expected_response[1]['artist_name'] = 'Dino Felipe' response = product.get_products_by_upcs(upcs) assert response.status == 200 assert response.message == expected_response @pytest.mark.parametrize('sort_direction', [('asc'), ('desc')]) @db.test_schema def test_get_products_by_upcs_with_sorting( monkeypatch, test_products_by_upcs, sort_direction): """Test get products from upcs with sorting parameters.""" expected_response = test_products_by_upcs upcs = [888831283041, 888831283042] db.insert_product_type() db.create_vw_product_table_format_and_artist_name() db.insert_vw_product(query=db.INSERT_PRODUCT_FORMAT_ARTIST_NAME) expected_response[0]['format'] = 'Full Length' expected_response[0]['artist_name'] = 'Artist 3' expected_response[1]['format'] = 'EP' expected_response[1]['artist_name'] = 'Dino Felipe' sort_param = 'release_name' if sort_direction == 'desc': expected_response.reverse() response = product.get_products_by_upcs(upcs, sort_param, sort_direction) assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_products_by_upcs_not_found(): """Test get products by upcs when not found.""" upcs = [99999] result = product.get_products_by_upcs(upcs) assert result.status == 404 assert result.message is None @db.test_schema def test_map_product_for_bulk(): """Test map product for bulk route.""" expected_response = { 'product_id': 1, 'distribution_format_id': 2, 'project_id': 1, 'product_type_id': 1, 'upc': 888831283041, 'status': 'in_content', 'subaccount_id': 1, 'vendor_id': 1, 'context_type': 'physical', 'release_date': '2017-09-09', 'product_type': 'Music', 'product_name': 'product test 1', 'deletions': 'N', 'display_upc': '888831283041', 'not_for_distribution': 'N' } row = [ ('release_id', 1), ('upc', 888831283041), ('subaccount_id', 1), ('vendor_id', 1), ('context_type', 'physical'), ('release_name', 'product test 1'), ('release_date', '2017-09-09'), ('product_type', 'Music'), ('deletions', 'N'), ('release_status', 'in_content'), ('product_type_id', 1), ('project_id', 1), ('distribution_format_id', 2), ('display_upc', '888831283041'), ('not_for_distribution', 'N') ] expected_response['format'] = 'Full Length' expected_response['artist_name'] = 'Artist 3' row.extend([('format', 'Full Length'), ('artist_name', 'Artist 3')]) result = product.map_product_for_bulk(row) assert result == expected_response @db.test_schema def test_get_product_by_display_upc_not_found(test_invalid_product): """Test get product by display_upc when not found.""" display_upc = test_invalid_product['display_upc'] vendor_id = test_invalid_product['vendor_id'] context_type = test_invalid_product['context_type'] response = product.get_product_by_display_upc( vendor_id, display_upc, context_type) assert response.status == 200 assert not response.message @db.test_schema def test_get_product_by_display_upc_found(test_valid_product): """Test get product by display_upc and context_type when found.""" expected_response = [test_valid_product] db.insert_vw_product() display_upc = test_valid_product['display_upc'] vendor_id = test_valid_product['vendor_id'] context_type = test_valid_product['context_type'] response = product.get_product_by_display_upc( vendor_id, display_upc, context_type) assert response.status == 200 assert response.message == expected_response @pytest.mark.parametrize( "show_deletions, records", [ pytest.param( False, [non_deleted_product], id="show deletions is false, so only filter non-deleted products." ), pytest.param( True, [non_deleted_product, deleted_product], id="show deletions is True, so show both deleted and non-deleted products." ) ] ) @db.test_schema def test_get_product_by_display_upc_duplicate_but_deleted( show_deletions: bool, records, test_valid_product, test_deleted_product): """Test get product by display_upc when duplicate but deleted.""" expected_response = records db.insert_vw_product() db.insert_deleted_duplicate_vw_product() display_upc = test_valid_product['display_upc'] vendor_id = test_valid_product['vendor_id'] context_type = test_valid_product['context_type'] response = product.get_product_by_display_upc( vendor_id, display_upc, context_type, show_deletions) assert response.message == expected_response assert response.status == 200 @patch('product.connectors.mysql._db_session') def test_get_product_by_display_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.get_product_by_display_upc( vendor_id, display_upc, context_type) assert result.status == 500 assert sentry_sdk.capture_exception.called @patch('product.connectors.mysql._db_session') def test_get_products_by_upcs_db_failure( db_session, mocker): """Test fatal db error for fetching products by upcs.""" mocker.patch.object( sentry_sdk, 'capture_exception', return_value='', autospec=True) session = MagicMock() session.execute = MagicMock(side_effect=SQLAlchemyError()) db_session.return_value = session upcs = [888831283041] result = product.get_products_by_upcs(upcs) assert result.status == 500 assert sentry_sdk.capture_exception.called @pytest.mark.parametrize('with_company_brand,with_tenant_uuids,expected_sql', [ # with only with_company_brand (False, None, product_document.PRODUCT_DOCUMENT), (True, None, product_document.PRODUCT_DOCUMENT_WITH_COMPANY_BRAND), (None, None, product_document.PRODUCT_DOCUMENT), # with only with_tenant_uuids (None, False, product_document.PRODUCT_DOCUMENT), (None, True, product_document.PRODUCT_DOCUMENT_WITH_TENANT_UUIDS), # mix of both (False, False, product_document.PRODUCT_DOCUMENT), (True, False, product_document.PRODUCT_DOCUMENT_WITH_COMPANY_BRAND), (False, True, product_document.PRODUCT_DOCUMENT_WITH_TENANT_UUIDS), (True, True, product_document.PRODUCT_DOCUMENT_WITH_COMPANY_BRAND), ]) def test_get_product_document_by_id( with_company_brand, with_tenant_uuids, expected_sql, monkeypatch): """Test get product document by id with product video type included.""" test_release_id = 100 result = { 'release_id': test_release_id, 'release_date': '2005-04-26', 'track_list': 'Rockabilly Fever\nFolsom Prison Blues', 'isrc_list': None, 'project_id': 2124515, 'track_type': 'music\nmusic', 'type_of_video': 'Other', 'preorder_date': '2005-04-20', 'ingestion_completed': '2022-02-24', 'distribution_date': '2022-04-22', 'priority': 3 } if with_tenant_uuids: result['spatial_upc'] = 154558 result['spatial_isrc_list'] = 'one\two' result_message = MagicMock(fetchone=lambda: result) execute_mock = MagicMock(return_value=result_message) monkeypatch.setattr(Session, 'execute', value=execute_mock) params = {} if with_company_brand: params['with_company_brand'] = with_company_brand if with_tenant_uuids: params['with_tenant_uuids'] = with_tenant_uuids response = product.get_product_document_by_id( test_release_id, **params, ) expected_params = { 'release_id': test_release_id } execute_mock.assert_called_with( (expected_sql), expected_params, ) assert execute_mock.call_count == 2 assert response.status == 200 def test_get_product_document_by_id_no_results(monkeypatch): """Test get product document by id.""" result = {'release_id': None} result_message = MagicMock(fetchone=lambda: result) monkeypatch.setattr(Session, 'execute', value=MagicMock( return_value=result_message)) response = product.get_product_document_by_id(100) assert response.status == 404 def test_get_tenant_uuids(test_valid_product, monkeypatch): """Test get_tenant_uuids when found.""" test_release_id = 100 result = { "product_id": 2983455, "vendorUUID": "df45d0a9-2c55-4925-95e0-8324f37f1283", "subaccountUUID": None, "companyBrandUUID": "d25a4cd1-e820-45f2-be5c-56edcfeb8298", "parentCompanyUUID": "955a1bbd-b623-4ea1-ab5f-8d6620c442fb" } result_message = MagicMock(fetchone=lambda: result) execute_mock = MagicMock(return_value=result_message) monkeypatch.setattr(Session, 'execute', value=execute_mock) response = product.get_tenant_uuids(test_release_id) execute_mock.assert_called_with( PRODUCT_TENANT_UUIDS, {'release_id': test_release_id}, ) assert response.status == 200 @patch('product.connectors.mysql._db_session') def test_get_product_document_by_id_failure( db_session, mocker): """Test get product document by id.""" db_session.return_value.execute = MagicMock( side_effect=SQLAlchemyError()) response = product.get_product_document_by_id(100) assert response.status == 500 def test_get_products_documents_by_ids_success(monkeypatch): """Test get products documents by ids.""" test_release_ids = '100,101' result = [ { 'release_id': 100, 'release_date': '2005-04-26', 'track_list': 'Rockabilly Fever\nFolsom Prison Blues', 'isrc_list': None, 'project_id': 21245150, 'track_type': 'music\nmusic', 'type_of_video': 'Other', 'preorder_date': '2005-04-20', 'ingestion_completed': '2022-02-24', 'distribution_date': '2022-04-22', 'priority': 3 }, { 'release_id': 101, 'release_date': '2005-04-26', 'track_list': 'Rockabilly Fever\nFolsom Prison Blues 1', 'isrc_list': None, 'project_id': 21245151, 'track_type': 'music\nmusic', 'type_of_video': 'Other', 'preorder_date': '2005-04-20', 'ingestion_completed': '2022-02-24', 'distribution_date': '2022-04-22', 'priority': 3 } ] result_message = MagicMock(fetchall=lambda: result) execute_mock = MagicMock(return_value=result_message) monkeypatch.setattr(Session, 'execute', value=execute_mock) response = product.get_products_documents_by_ids(test_release_ids) expected_params = { 'release_ids': [100, 101] } execute_mock.assert_called_with( products_documents.PRODUCTS_DOCUMENTS, expected_params) assert response.status == 200 assert response.message == [{'document': item} for item in result] def test_get_products_documents_by_ids_no_results(monkeypatch): """Test get products documents by ids no results.""" result = [] result_message = MagicMock(fetchall=lambda: result) monkeypatch.setattr(Session, 'execute', value=MagicMock( return_value=result_message)) response = product.get_products_documents_by_ids('100,101') assert response.status == 200 @pytest.fixture def expected_date_filtered_product(): """Fixture for product, filtered by date.""" return { 'context_type': 'digital', 'deletions': 'N', 'distribution_format_id': 2, 'product_id': 99, 'product_name': 'product test 92', 'product_type_id': 1, 'project_id': 33, 'release_date': '2018-12-01', 'status': 'in_content', 'subaccount_id': 5, 'upc': 888831283047, 'display_upc': '888831283047', 'vendor_id': 1234, 'not_for_distribution': 'N' } @db.test_schema def test_get_products_by_vendor_id_start_date( test_vendor, expected_date_filtered_product): """Return products matching vendor_id and release_date.""" expected_response = { 'items': [ expected_date_filtered_product, ], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 1} } db.insert_vw_product() db.insert_vw_product(db.INSERT_DATE_FILTER_PRODUCT) response = product.get_products_by_account_id( test_vendor, status='all', start_date='2018-12-01') assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_products_by_vendor_id_start_end_date( test_vendor, expected_date_filtered_product): """Return products matching vendor_id and release_date.""" expected_response = { 'items': [ expected_date_filtered_product, ], 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 1} } db.insert_vw_product() db.insert_vw_product(db.INSERT_DATE_FILTER_PRODUCT) response = product.get_products_by_account_id( test_vendor, status='all', start_date='2018-12-01', end_date='2018-12-31' ) assert response.status == 200 assert response.message == expected_response @pytest.fixture def expected_date_sorted_product(): """Fixture for product, sorted by date.""" return [ { 'context_type': 'digital', 'deletions': 'N', 'distribution_format_id': 2, 'product_id': 99, 'product_name': 'product test 92', 'product_type_id': 1, 'project_id': 33, 'release_date': '2018-12-01', 'status': 'in_content', 'subaccount_id': 5, 'upc': 888831283047, 'display_upc': '888831283047', 'vendor_id': 1234, 'not_for_distribution': 'N' }, { 'context_type': 'digital', 'deletions': 'N', 'distribution_format_id': 2, 'product_id': 100, 'product_name': 'product test 100', 'product_type_id': 1, 'project_id': 33, 'release_date': '2018-12-02', 'status': 'in_content', 'subaccount_id': 5, 'upc': 888831283048, 'display_upc': '888831283048', 'vendor_id': 1234, 'not_for_distribution': 'N' }, ] @pytest.mark.parametrize( 'sort_order', [ ('asc'), ('desc'), (None) ] ) @db.test_schema def test_get_products_by_vendor_id_sort_dir( test_vendor, expected_date_sorted_product, sort_order): """Return products matching vendor_id and with sort.""" reverse = sort_order in ('desc', None) expected_date_sorted_product.sort( key=lambda item: item['release_date'], reverse=reverse) expected_response = { 'items': expected_date_sorted_product, 'pagination': { 'type': 'standard', 'offset': 0, 'limit': 50, 'total_records': 2} } db.insert_vw_product(db.INSERT_DATE_SORT_PRODUCT) response = product.get_products_by_account_id( test_vendor, status='all', sort_order=sort_order) assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_products_by_vendorid_and_upcs_found(): """Test products returned for the vendor and upc(s).""" db.insert_vw_product() upcs = ['888831283041'] vendor_id = 1 expected_response = [{ 'product_id': 1, 'distribution_format_id': 2, 'project_id': 22, 'product_type_id': 1, 'upc': 888831283041, 'display_upc': '888831283041', 'context_type': 'physical', 'vendor_id': 1, 'subaccount_id': 1, 'release_date': '2017-09-09', 'status': 'label_processing', 'product_name': 'product test 1', 'deletions': 'N', 'not_for_distribution': 'N'}] response = product.get_products_by_accountid_and_upcs( upcs, vendor_id, subaccount_id=None) assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_products_by_vendorid_and_upcs_not_found(): """Test products returned for the vendor and upc(s).""" db.insert_vw_product() upcs = ['888831283044'] vendor_id = 1 expected_response = [] response = product.get_products_by_accountid_and_upcs( upcs, vendor_id, subaccount_id=None) assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_products_by_subaccountid_and_upcs_found(): """Test products returned for the subaccount and upc(s).""" db.insert_vw_product() upcs = ['1234567654321'] subaccount_id = 55 vendor_id = None expected_response = [{ 'product_id': 32123, 'distribution_format_id': 1, 'project_id': 22, 'product_type_id': 1, 'upc': 1234567654321, 'display_upc': '1234567654321', 'context_type': 'digital', 'vendor_id': 55, 'subaccount_id': 55, 'release_date': '2017-09-09', 'status': 'label_processing', 'product_name': 'product test XIV', 'deletions': 'N', 'not_for_distribution': 'N'}] response = product.get_products_by_accountid_and_upcs( upcs, vendor_id, subaccount_id) assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_products_by_subaccountid_and_upcs_not_found(): """Test products returned for the subaccount and upc(s).""" db.insert_vw_product() upcs = ['888831283043'] vendor_id = None subaccount_id = 5 expected_response = [] response = product.get_products_by_accountid_and_upcs( upcs, vendor_id, subaccount_id) assert response.status == 200 assert response.message == expected_response def test_get_products_by_isrc(monkeypatch, test_valid_product): """Test get products by isrc.""" test_isrc = 'GBAYE1234567' row = [ ('release_id', 1), ('upc', 888831283041) ] rows = [row] result_message = MagicMock(fetchall=lambda: rows) execute_mock = MagicMock(return_value=result_message) monkeypatch.setattr(Session, 'execute', value=execute_mock) response = product.get_products_by_isrc(test_isrc) expected_params = { 'isrc': test_isrc } execute_mock.assert_called_with( products_by_isrc.PRODUCTS_BY_ISRC, expected_params) assert response.status == 200 def test_get_products_by_isrc_none_found(monkeypatch, test_valid_product): """Test get products by isrc.""" test_isrc = 'GBAYE1234567' rows = [] result_message = MagicMock(fetchall=lambda: rows) execute_mock = MagicMock(return_value=result_message) monkeypatch.setattr(Session, 'execute', value=execute_mock) response = product.get_products_by_isrc(test_isrc) expected_params = { 'isrc': test_isrc } execute_mock.assert_called_with( products_by_isrc.PRODUCTS_BY_ISRC, expected_params) assert response.status == 404 @db.test_schema def test_get_product_by_release_artist_id(): """Test get products by release_artist_id.""" db.insert_vw_product() db.insert_release_artist() result = product.get_product_by_release_artist_id(1) assert result.message == { 'context_type': 'digital', 'deletions': 'N', 'distribution_format_id': 1, 'product_id': 2, 'product_name': 'product test 2', 'product_type_id': 1, 'project_id': 33, 'release_date': '2017-09-17', 'status': 'label_processing', 'subaccount_id': 0, 'upc': 888831283042, 'display_upc': '888831283042', 'vendor_id': 1, 'not_for_distribution': 'N' } @db.test_schema def test_get_product_by_release_artist_id_not_found(): """Test get products by release_artist_id.""" db.insert_vw_product() db.insert_release_artist() result = product.get_product_by_release_artist_id(1001) assert result.status == 404 @db.test_schema def test_filter_upcs_with_existing_products(): """Test filter upcs with existing products.""" db.insert_vw_product() vw_product_upc = 888831283041 missing_upc = 195497264421 result = product.filter_upcs_with_existing_products([vw_product_upc, missing_upc]) assert result.status == 200 assert result.message == [vw_product_upc] @db.test_schema def test_get_product_id_by_upc_dataloaded_found(): """Test get_product_id_by_upc_dataloaded returns product ids for valid upcs.""" db.insert_vw_product() upcs = [888831283041, 888831283042] expected_response = [ {'upc': 888831283041, 'product_id': 1}, {'upc': 888831283042, 'product_id': 2} ] response = product.get_product_id_by_upc_dataloaded(upcs) assert response.status == 200 assert response.message == expected_response @db.test_schema def test_get_product_id_by_upc_dataloaded_not_found(): """Test get_product_id_by_upc_dataloaded returns 200 and None for product_id if no products found. """ upcs = [999999999999] response = product.get_product_id_by_upc_dataloaded(upcs) assert response.status == 200 assert response.message == [{'upc': 999999999999, 'product_id': None}] @patch('product.connectors.mysql._db_session') def test_get_product_id_by_upc_dataloaded_db_failure(db_session, mocker): """Test fatal db error for get_product_id_by_upc_dataloaded.""" mocker.patch.object( sentry_sdk, 'capture_exception', return_value='', autospec=True) session = MagicMock() session.execute = MagicMock(side_effect=SQLAlchemyError()) db_session.return_value = session upcs = [888831283041] result = product.get_product_id_by_upc_dataloaded(upcs) assert result.status == 500 assert sentry_sdk.capture_exception.called @db.test_schema @pytest.mark.parametrize( 'product_ids, expected_status, expected_errors, expected_message', [ pytest.param( [], 200, None, [], id='empty list of product ids returns an empty list of ownership objects', ), pytest.param( [10, 9020, 2, 93, 1], 200, None, [ {'product_id': 10, 'tenant_id': 1234, 'tenant_type': 'account'}, None, {'product_id': 2, 'tenant_id': 1, 'tenant_type': 'account'}, {'product_id': 93, 'tenant_id': 1234, 'tenant_type': 'account'}, {'product_id': 1, 'tenant_id': 1, 'tenant_type': 'subaccount'}, ], id='dataloader response is returned', ), pytest.param( [40890428049, 1094803843], 200, None, [None, None], id='List of None when product_ids are not found', ) ]) def test_get_product_ownership_by_product_id_dataloaded( product_ids: list[int], expected_status: int, expected_errors: dict | None, expected_message: list | None, ) -> None: """Test get_product_ownership_by_product_id_dataloaded.""" db.insert_vw_product() response = product.get_product_ownership_by_product_id_dataloaded(product_ids) assert response.status == expected_status, response.__dict__ assert response.errors == expected_errors, response.__dict__ assert response.message == expected_message def test_product_document_joins_additional_tables_filtering_atmos_and_soft_deleted() -> None: # The product-document SQL is mock-tested (no additional-recording tables in # the SQLite test DB), so guard the contract statically: both joins select # the atmos edition and exclude soft-deleted rows via deleted_at IS NULL in # the ON clause (not WHERE -- that would drop releases whose tracks have no # live spatial row). product_additional_upc soft-deletes, so unlike the old # release_spatial it must be filtered too, or deleted UPCs would resurface. normalized = " ".join( product_document.PRODUCT_DOCUMENT_WITH_TENANT_UUIDS.split() ) assert ( "LEFT JOIN track_additional_isrc tsp ON tsp.track_id = t.id " "AND tsp.type = 'atmos' AND tsp.deleted_at IS NULL" in normalized ) assert ( "LEFT JOIN product_additional_upc rsp ON rsp.product_id = r.release_id " "AND rsp.type = 'atmos' AND rsp.deleted_at IS NULL" in normalized )