"""Tests for Handlers.""" import json from unittest.mock import MagicMock from unittest.mock import patch from flask.ctx import AppContext from pytest_mock import MockerFixture import application from oto import response from owsrequest import access, flask_request from owsrequest.constants import headers as ows_headers import pytest from product import auth, handlers from product.constants import error from product.constants import header from product.logic import hfa from product.logic import localization as localization_logic from product.logic import product as product_logic from product.logic import profile as profile_logic from product.logic import upc as upc_logic from product.models import release_artist from product.models import release_asset_version from product.models import release_phonetic_translations from product.models import release_subgenre @patch('product.handlers.g', spec=['log']) def test_exception_handler(mock_g): """Verify exception_Handler returns 500 status code and json payload.""" message = ( 'The server encountered an internal error ' 'and was unable to complete your request.') mock_error = MagicMock() server_response = handlers.exception_handler(mock_error) mock_g.log.exception.assert_called_with(mock_error) # assert status code is 500 assert server_response.status_code == 500 # assert json payload response_message = json.loads(server_response.data.decode()) assert response_message['message'] == message assert response_message['code'] == response.error.ERROR_CODE_INTERNAL_ERROR def test_handler_vendor_product_ownership_with_logic_success( test_vendor, test_product, mocker): """Test HEAD /vendor/{vendor_id}/product/{product_id}. Verify handler success with logic layer success. """ mocker.patch.object( product_logic, 'check_product_ownership', return_value=response.Response(status=200)) with application.app.test_request_context(): ownership_response = handlers.check_product_ownership( header.GRASS_ACCOUNT_TYPE_VENDOR, test_vendor, test_product) assert ownership_response.status_code == 200 def test_handler_subaccount_product_ownership_with_logic_success( test_subaccount, test_product, mocker): """Test HEAD /subaccount/{subaccount_id}/product/{product_id}. Verify handler success with logic layer success. """ mocker.patch.object( product_logic, 'check_product_ownership', return_value=response.Response(status=200)) with application.app.test_request_context(): ownership_response = handlers.check_product_ownership( header.GRASS_ACCOUNT_TYPE_SUBACCOUNT, test_subaccount, test_product) assert ownership_response.status_code == 200 def test_handler_ownership_fails_with_invalid_account_type( test_subaccount, test_product): """Test ownership fails with invalid account type.""" with application.app.test_request_context(): ownership_response = handlers.check_product_ownership( 'monkeys', test_subaccount, test_product) assert ownership_response.status_code == 400 def test_handler_ownership_product_not_found( test_subaccount, test_product, mocker): """Test ownership fails if product not found.""" mocker.patch.object( product_logic, 'check_product_ownership', return_value=response.Response(status=404)) with application.app.test_request_context(): ownership_response = handlers.check_product_ownership( header.GRASS_ACCOUNT_TYPE_SUBACCOUNT, test_subaccount, test_product) assert ownership_response.status_code == 404 def test_handler_ownership_forbidden_for_invalid_account( test_subaccount, test_product, mocker): """Test ownership fails if vendor or subaccount is not the owner.""" mocker.patch.object( product_logic, 'check_product_ownership', return_value=response.Response(status=403)) with application.app.test_request_context(): ownership_response = handlers.check_product_ownership( header.GRASS_ACCOUNT_TYPE_SUBACCOUNT, test_subaccount, test_product) assert ownership_response.status_code == 403 @pytest.mark.parametrize( "access_check_enabled, account_id, account_type, deletion_status, status", ( (False, None, None, 200, 200), (False, None, None, 400, 400), (True, "1", None, None, 400), (True, None, "vendor", None, 400), (True, "1", "vendor", 200, 200) ) ) def test_handler_delete_product_with_grass_headers_only( access_check_enabled: bool, account_id: str | None, account_type: str | None, deletion_status: int, status: int, test_product: int, mocker: MockerFixture, context: AppContext ): """Test delete product flow without profile headers checks.""" features_mock = mocker.patch.object( handlers.feature_control_util, 'is_feature_enabled', return_value=access_check_enabled ) delete_mock = mocker.patch.object( product_logic, 'delete_product', return_value=response.Response(status=deletion_status) ) context.g.request_context.profile_type = None context.g.request_context.profile_id = None headers = {} for k, v in ( (ows_headers.GRASS_ACCOUNT_ID, account_id), (ows_headers.GRASS_ACCOUNT_TYPE, account_type) ): if v is not None: headers[k] = v with application.app.test_request_context(headers=headers): product_delete_response = handlers.delete_product_by_product_id( test_product) assert product_delete_response.status_code == status features_mock.assert_called_once() if deletion_status is not None: delete_mock.assert_called_once_with(test_product, account_type, account_id) else: delete_mock.assert_not_called() @pytest.mark.parametrize( "grass_headers_check_status, profile_headers_check_status, " "profile_check_status, deletion_status, status", ( (200, 200, 200, 200, 200), # success (400, None, None, None, 400), # incomplete grass headers (403, None, None, None, 403), # invalid grass headers (200, 400, None, None, 400), # incomplete profile headers (200, 403, None, None, 403), # invalid profile headers (200, 200, 403, None, 403), # profile has no access to the product (200, 200, 200, 500, 500), # error on deletion ) ) def test_handler_delete_product_with_profile_access( grass_headers_check_status: int, profile_headers_check_status: int | None, profile_check_status: int | None, deletion_status: int | None, status: int, test_product: int, mocker: MockerFixture, context: AppContext ): """Test delete product flow with grass and profile headers checks.""" features_mock = mocker.patch.object( handlers.feature_control_util, 'is_feature_enabled', return_value=True ) grass_headers_check_mock = mocker.patch.object( access, 'verify_grass_headers', return_value=response.Response(status=grass_headers_check_status) ) profile_headers_check_mock = mocker.patch.object( auth, 'verify_profile_headers', return_value=response.Response(status=profile_headers_check_status) ) profile_check_mock = mocker.patch.object( profile_logic, 'check_profile_access_to_product', return_value=response.Response(status=profile_check_status) ) delete_mock = mocker.patch.object( product_logic, 'delete_product', return_value=response.Response(status=deletion_status) ) profile_id, profile_type = "1", "LabelProfile" context.g.request_context.profile_type = profile_type context.g.request_context.profile_id = profile_id account_id, account_type = "1", "vendor" with application.app.test_request_context(headers={ ows_headers.GRASS_ACCOUNT_ID: account_id, ows_headers.GRASS_ACCOUNT_TYPE: account_type, }): product_delete_response = handlers.delete_product_by_product_id( test_product) assert product_delete_response.status_code == status features_mock.assert_called_once() grass_headers_check_mock.assert_called_once_with( account_type, account_id, required=True ) if profile_headers_check_status is not None: profile_headers_check_mock.assert_called_once_with( profile_type, profile_id, allowed_types=("ContentProfile", "LabelProfile", "OrchAdminProfile") ) else: profile_headers_check_mock.assert_not_called() if profile_check_status is not None: profile_check_mock.assert_called_once_with(1, profile_type, test_product) else: profile_check_mock.assert_not_called() if deletion_status is not None: delete_mock.assert_called_once_with(test_product, account_type, account_id) else: delete_mock.assert_not_called() def test_delete_product_localizations( test_product, fixture_language, mocker): """Test delete product localizations when it succeeds.""" mocker.patch.object( localization_logic, 'delete', return_value=response.Response( message='product was successfully deleted')) with application.app.test_request_context(): actual_response = handlers.delete_product_localizations( test_product, fixture_language) assert actual_response.status_code == 200 def test_delete_product_localizations_legacy( test_product, fixture_language, mocker): """Test delete product localizations when it succeeds for legacy.""" mocker.patch.object( localization_logic, 'delete', return_value=response.Response( message='Successfully deleted release and artist localizations.')) with application.app.test_request_context(query_string={'legacy': 1}): actual_response = handlers.delete_product_localizations( test_product, fixture_language) assert actual_response.status_code == 200 localization_logic.delete.assert_called_with( test_product, fixture_language, True) def test_delete_non_existing_product_localizations( test_product, fixture_language, mocker): """Test delete product localizations for non-existing product.""" mocker.patch.object( localization_logic, 'delete', return_value=response.create_not_found_response()) with application.app.test_request_context(): actual_response = handlers.delete_product_localizations( test_product, fixture_language) assert actual_response.status_code == 404 def test_get_product_localization(test_product, mocker): """Test get product localization succeed.""" mocker.patch.object( localization_logic, 'get_localization_by_product_id', return_value=response.Response(message={'product_name': 'new name'})) with application.app.test_request_context(): handler_response = handlers.get_product_localization( test_product) assert handler_response.status_code == 200 def test_get_itunes_language_by_id(mocker): """Test get itunes language succeed.""" mocker.patch.object( localization_logic, 'get_itunes_language_by_id', return_value=response.Response(message={ 'language_id': 1, 'language': 'Test Language', 'language_code': 'abc' })) mock_language_id = 1 with application.app.test_request_context(): handler_response = handlers.get_itunes_language_by_id( mock_language_id) assert handler_response.status_code == 200 def test_get_all_itunes_languages(fixture_all_itunes_languages, mocker): """Test get all itunes_languages succeed.""" mocker.patch.object( localization_logic, 'get_all_itunes_languages', return_value=response.Response(message=fixture_all_itunes_languages)) with application.app.test_request_context(): handler_response = handlers.get_all_itunes_languages() assert handler_response.status_code == 200 def test_get_public_all_itunes_languages(fixture_all_itunes_languages, mocker): """Test get all itunes_languages succeed.""" mocker.patch.object( localization_logic, 'get_all_itunes_languages', return_value=response.Response(message=fixture_all_itunes_languages)) with application.app.test_request_context(): handler_response = handlers.get_public_all_itunes_languages() assert handler_response.headers['Cache-Control'] == 'max-age=86400' assert handler_response.status_code == 200 def test_get_product_localization_not_found(test_product, mocker): """Test get product localization with not found error.""" mocker.patch.object( localization_logic, 'get_localization_by_product_id', return_value=response.create_not_found_response()) with application.app.test_request_context(): handler_response = handlers.get_product_localization(test_product) assert handler_response.status_code == 404 def test_get_product_localization_auth_error(test_product, mocker): """Test get product localization with authentication failures.""" auth_error = response.create_error_response( code=error.ERROR_CODE_AUTHORIZATION, message=error.ERROR_MESSAGE_FORBIDDEN_USER, status=403) mocker.patch.object( localization_logic, 'get_localization_by_product_id', return_value=auth_error) with application.app.test_request_context(): handler_response = handlers.get_product_localization(test_product) assert handler_response.status_code == 403 def test_create_product_localization(test_product, fixture_language, mocker): """Test create product localizations with success.""" message = {'product_name': 'abc', 'language_id': 12} mocker.patch.object( localization_logic, 'create', return_value=response.Response(message=message)) with application.app.test_request_context( data=json.dumps(dict(product_name='test name')), content_type='application/json'): create_response = handlers.create_product_localizations( test_product, fixture_language) assert create_response.status_code == 200 with application.app.test_request_context( data=json.dumps(dict(no_name='test name')), content_type='application/json'): create_response = handlers.create_product_localizations( test_product, fixture_language) assert create_response.status_code == 400 def test_create_product_localization_legacy( test_product, fixture_language, mocker): """Test create legacy product localizations with success.""" message = { 'product_name': 'abc', 'language_id': fixture_language, 'performer': {'3427519': 'perform 4', '4044560': 'perform 4'}} mocker.patch.object( localization_logic, 'create', return_value=response.Response(message=message)) with application.app.test_request_context( data=json.dumps(message), query_string={'legacy': 1}, content_type='application/json'): create_response = handlers.create_product_localizations( test_product, fixture_language) assert create_response.status_code == 200 localization_logic.create.assert_called_once_with( test_product, fixture_language, message, True) def test_create_product_localization_legacy_error( test_product, fixture_language, mocker): """Test create legacy product localizations when required param missing.""" with application.app.test_request_context( data=json.dumps(dict(no_name='test name')), query_string={'legacy': 1}, content_type='application/json'): mocker.patch.object(localization_logic, 'create') create_response = handlers.create_product_localizations( test_product, fixture_language) assert create_response.status_code == 400 assert not localization_logic.create.called def test_check_display_upc_availability_for_vendor_success( test_product, test_vendor, mocker): """Test successful response when requesting availability of display_upc.""" mocker.patch.object( product_logic, 'check_display_upc_availability', return_value=response.Response(status=200)) with application.app.test_request_context('?context_type=physical'): handler_response = handlers.check_display_upc_availability_for_vendor( test_vendor, test_product) assert handler_response.status_code == 200 def test_check_display_upc_availability_for_vendor_query_param_failure( test_product, test_vendor, mocker): """Test successful response when requesting availability of display_upc.""" mocker.patch.object( product_logic, 'check_display_upc_availability', return_value=response.Response(status=200)) with application.app.test_request_context(): handler_response = handlers.check_display_upc_availability_for_vendor( test_vendor, test_product) assert handler_response.status_code == 400 error_message = json.loads(handler_response.data.decode('utf8')).get( 'message') assert error_message == error.ERROR_MESSAGE_CONTEXT_TYPE_REQUIRED def test_check_display_upc_availability_for_vendor_query_param_wrong( test_product, test_vendor, mocker): """Test successful response when requesting availability of display_upc.""" mocker.patch.object( product_logic, 'check_display_upc_availability', return_value=response.Response(status=200)) with application.app.test_request_context('?context_type=watermelon'): handler_response = handlers.check_display_upc_availability_for_vendor( test_vendor, test_product) assert handler_response.status_code == 400 error_message = json.loads(handler_response.data.decode('utf8')).get( 'message') assert error_message == error.ERROR_MESSAGE_CONTEXT_TYPE_REQUIRED def test_check_display_upc_availability_for_vendor_failure( test_product, test_vendor, mocker): """Test failure response when requesting availability of display_upc.""" mocker.patch.object( product_logic, 'check_display_upc_availability', return_value=response.Response(status=403)) with application.app.test_request_context('?context_type=physical'): handler_response = handlers.check_display_upc_availability_for_vendor( test_vendor, test_product) assert handler_response.status_code == 403 def test_check_display_upc_availability_for_vendor_auth_failure( test_product, test_vendor, mocker): """Test failure response when requesting availability of display_upc.""" mocker.patch.object( access, 'verify_grass_access', return_value=response.Response(status=403)) with application.app.test_request_context('?context_type=physical'): handler_response = handlers.check_display_upc_availability_for_vendor( test_vendor, test_product) assert handler_response.status_code == 403 def test_get_vendor_product_by_display_upc_success( test_valid_product, mocker): """Test successful response when requesting upc of vendor's display_upc.""" mocker.patch.object( product_logic, 'get_product_by_display_upc', return_value=response.Response( status=200, message=test_valid_product['upc'])) request_context = '?context_type={}'.format( test_valid_product['context_type']) with application.app.test_request_context(request_context): handler_response = handlers.get_upc_from_vendor_display_upc( test_valid_product['vendor_id'], test_valid_product['display_upc']) assert handler_response.status_code == 200 def test_get_vendor_product_by_display_upc_success_deleted_product( test_deleted_product, mocker): """Test successful response of deleted product when requesting upc of vendor's display_upc.""" mocker.patch.object( product_logic, 'get_product_by_display_upc', return_value=response.Response( status=200, message=test_deleted_product['upc'])) request_context = '?context_type={}&show_deletions=True'.format( test_deleted_product['context_type']) with application.app.test_request_context(request_context): handler_response = handlers.get_upc_from_vendor_display_upc( test_deleted_product['vendor_id'], test_deleted_product['display_upc']) assert handler_response.status_code == 200 def test_get_vendor_product_by_display_upc_not_found( test_valid_product, mocker): """Test response when no upc is found.""" mocker.patch.object( product_logic, 'get_product_by_display_upc', return_value=response.Response(status=404)) request_context = '?context_type={}'.format( test_valid_product['context_type']) with application.app.test_request_context(request_context): handler_response = handlers.get_upc_from_vendor_display_upc( test_valid_product['vendor_id'], test_valid_product['display_upc']) assert handler_response.status_code == 404 def test_get_vendor_product_by_display_upc_invalid_context_type( test_valid_product, mocker, invalid_context_type): """Test failure with invalid context_type.""" request_context = '?context_type={}'.format(invalid_context_type) with application.app.test_request_context(request_context): handler_response = handlers.get_upc_from_vendor_display_upc( test_valid_product['vendor_id'], test_valid_product['display_upc']) assert handler_response.status_code == 400 def test_get_vendor_product_by_display_upc_auth_failure( mocker, test_valid_product): """Test failure with invalid creds.""" mocker.patch.object( access, 'verify_grass_access', return_value=response.Response(status=403)) request_context = '?context_type={}'.format( test_valid_product['context_type']) with application.app.test_request_context(request_context): handler_response = handlers.check_display_upc_availability_for_vendor( test_valid_product['vendor_id'], test_valid_product['display_upc']) assert handler_response.status_code == 403 def test_check_display_upc_availability_for_vendor_db_failure( test_product, test_vendor, mocker): """Test db failure response when requesting availability of display_upc.""" mocker.patch.object( product_logic, 'check_display_upc_availability', return_value=response.Response(status=500)) with application.app.test_request_context('?context_type=physical'): handler_response = handlers.check_display_upc_availability_for_vendor( test_vendor, test_product) assert handler_response.status_code == 500 def test_check_display_upc_availability_for_subaccount_success( test_product, test_subaccount, mocker): """Test successful response when requesting availability of display_upc.""" mocker.patch.object( product_logic, 'check_display_upc_availability', return_value=response.Response(status=200)) with application.app.test_request_context('?context_type=physical'): handler_response = \ handlers.check_display_upc_availability_for_subaccount( test_subaccount, test_product) assert handler_response.status_code == 200 def test_check_display_upc_availability_for_subaccount_query_param_failure( test_product, test_subaccount, mocker): """Test successful response when requesting availability of display_upc.""" mocker.patch.object( product_logic, 'check_display_upc_availability', return_value=response.Response(status=200)) with application.app.test_request_context(): handler_response = \ handlers.check_display_upc_availability_for_subaccount( test_subaccount, test_product) assert handler_response.status_code == 400 error_message = json.loads(handler_response.data.decode('utf8')).get( 'message') assert error_message == error.ERROR_MESSAGE_CONTEXT_TYPE_REQUIRED def test_check_display_upc_availability_for_subaccount_query_param_wrong( test_product, test_subaccount, mocker): """Test successful response when requesting availability of display_upc.""" mocker.patch.object( product_logic, 'check_display_upc_availability', return_value=response.Response(status=200)) with application.app.test_request_context('?context_type=watermelon'): handler_response = \ handlers.check_display_upc_availability_for_subaccount( test_subaccount, test_product) assert handler_response.status_code == 400 error_message = json.loads(handler_response.data.decode('utf8')).get( 'message') assert error_message == error.ERROR_MESSAGE_CONTEXT_TYPE_REQUIRED def test_check_display_upc_availability_for_subaccount_failure( test_product, test_subaccount, mocker): """Test failure response when requesting availability of display_upc.""" mocker.patch.object( product_logic, 'check_display_upc_availability', return_value=response.Response(status=403)) with application.app.test_request_context('?context_type=physical'): handler_response = \ handlers.check_display_upc_availability_for_subaccount( test_subaccount, test_product) assert handler_response.status_code == 403 def test_check_display_upc_availability_for_subaccount_auth_failure( test_product, test_subaccount, mocker): """Test failure response when requesting availability of display_upc.""" mocker.patch.object( access, 'verify_grass_access', return_value=response.Response(status=403)) with application.app.test_request_context('?context_type=physical'): handler_response = \ handlers.check_display_upc_availability_for_subaccount( test_subaccount, test_product) assert handler_response.status_code == 403 def test_check_display_upc_availability_for_subaccount_db_failure( test_product, test_subaccount, mocker): """Test db failure response when requesting availability of display_upc.""" mocker.patch.object( product_logic, 'check_display_upc_availability', return_value=response.Response(status=500)) with application.app.test_request_context('?context_type=physical'): handler_response = \ handlers.check_display_upc_availability_for_subaccount( test_subaccount, test_product) assert handler_response.status_code == 500 def test_is_orchard_upc(mocker, test_upc): """Test is_orchard_upc.""" expected_response = response.Response( status=12321, message='applesauce bananas') mocker.patch.object( upc_logic, 'is_orchard_upc', return_value=expected_response) handler_response = handlers.is_orchard_upc(test_upc) upc_logic.is_orchard_upc.assert_called_once_with(test_upc) assert handler_response.status_code == 12321 assert handler_response.data.decode() == 'applesauce bananas' def test_delete_track_localization_with_tuids(mocker): """Test delete_track_localization_with_tuids success.""" mocker.patch.object( localization_logic, 'delete_track_localization', return_value=response.Response(message='2 track localizations deleted') ) with application.app.test_request_context(): delete_response = handlers.delete_track_localization_with_tuids( [100, 200]) assert delete_response localization_logic.delete_track_localization.assert_called_once_with( [100, 200]) def test_delete_track_localization_with_language(mocker): """Test delete_track_localization_with_language success.""" mocker.patch.object( localization_logic, 'delete_track_localization', return_value=response.Response(message='2 track localizations deleted') ) with application.app.test_request_context(): delete_response = handlers.delete_track_localization_with_language( 100, 1) assert delete_response localization_logic.delete_track_localization.assert_called_once_with( [100], 1) def test_get_track_localization(mocker): """Test get_track_localization when it succeeds.""" expected_response = {'pagination': { 'total_records': 1, 'type': 'none' }, 'items': [{ 'tuid': 21236647, 'version': 'trans 66', 'artists': { '67486761': 'artist 55' }, 'language_id': 12, 'track_name': 'trans 66' }]} mocker.patch.object( localization_logic, 'get_track_localization', return_value=response.Response(message=expected_response)) with application.app.test_request_context(): handler_response = handlers.get_track_localization([100, 200]) assert handler_response.status_code == 200 localization_logic.get_track_localization.assert_called_with( [100, 200]) def test_get_track_localization_post_success(mocker): """Test get_track_localization with POST when it succeeds.""" expected_response = {'pagination': { 'total_records': 1, 'type': 'none' }, 'items': [{ 'tuid': 21236647, 'version': 'trans 66', 'artists': { '67486761': 'artist 55' }, 'language_id': 12, 'track_name': 'trans 66' }]} data = {'tuids': [100, 200]} mocker.patch.object( localization_logic, 'get_track_localization', return_value=response.Response(message=expected_response)) with application.app.test_client() as c: handler_response = c.post( '/localization/track', data=json.dumps(data), content_type='application/json') assert handler_response.status_code == 200 localization_logic.get_track_localization.assert_called_with( [100, 200]) def test_get_track_localization_post_failure(mocker): """Test get_track_localization with POST when it fails.""" data = {'tuids': []} with application.app.test_client() as c: handler_response = c.post( '/localization/track', data=json.dumps(data), content_type='application/json') assert handler_response.status_code == 400 actual_result = json.loads(handler_response.data.decode()) assert 'tuids' in actual_result['message'] def test_update_track_localization(mocker): """Test successful response when update_track_localization.""" data = { 'track_name': 'abc', 'version': 'ver', 'language_id': 1, 'artists': {'3427519': 'perform 4', '4044560': 'perform 4'}} mocker.patch.object( localization_logic, 'update_track_localization', return_value=response.Response(message='success')) with application.app.test_request_context( data=json.dumps(data), content_type='application/json'): handler_response = handlers.update_track_localization( tuid=123, language_id=1) assert handler_response.status_code == 200 localization_logic.update_track_localization.assert_called_once_with( 123, 1, data) def test_update_multiple_tracks_localization(mocker): """Test successful response when update_multiple_tracks_localization.""" track1 = { 'track_name': 'abc', 'version': 'ver', 'language_id': 1, 'tuid': 100} data = {'items': [track1]} mocker.patch.object( localization_logic, 'update_multiple_localizations', return_value=response.Response(message='success')) with application.app.test_request_context( data=json.dumps(data), content_type='application/json'): handler_response = handlers.update_multiple_tracks_localization() assert handler_response.status_code == 200 localization_logic.update_multiple_localizations.\ assert_called_once_with(data) def test_update_multiple_tracks_invalid_data(mocker): """Test spec validator check before update_multiple_tracks_localization.""" data = {'dummy': 'data'} mocker.patch.object( localization_logic, 'update_multiple_localizations') expected_error = { 'code': 'bad_request', 'message': {'items': "\'items\' is a required property"}} with application.app.test_request_context( data=json.dumps(data), content_type='application/json'): handler_response = handlers.update_multiple_tracks_localization() actual_result = json.loads(handler_response.data.decode()) assert handler_response.status_code == 400 assert actual_result == expected_error assert localization_logic.update_multiple_localizations.call_count == 0 def test_update_multiple_tracks_missing_keys(mocker): """Test spec validator check before update_multiple_tracks_localization.""" track1 = { 'track_x': 'abc', 'version_x': 'ver', 'language_x': 1, 'tuid_x': 100} data = {'items': [track1]} mocker.patch.object( localization_logic, 'update_multiple_localizations') expected_error = { 'code': 'bad_request', 'message': { 'language_id': "'language_id' is a required property", 'track_name': "'track_name' is a required property", 'tuid': "'tuid' is a required property", 'version': "'version' is a required property" } } with application.app.test_request_context( data=json.dumps(data), content_type='application/json'): handler_response = handlers.update_multiple_tracks_localization() actual_result = json.loads(handler_response.data.decode()) assert handler_response.status_code == 400 assert actual_result == expected_error assert localization_logic.update_multiple_localizations.call_count == 0 def test_get_upcs_by_product_ids_success(mocker): """Test get upcs by valid product ids.""" expected_response = { 'items': [{ 'product_id': 2078459, 'upc': 19177321792 }, { 'product_id': 2078461, 'upc': 191773455477 } ] } mocker.patch.object( product_logic, 'get_upcs_by_product_ids', return_value=response.Response(message=expected_response)) with application.app.test_request_context('?product_ids=2078459,2078461'): handler_response = handlers.get_upcs_by_product_ids() assert handler_response.status_code == 200 product_logic.get_upcs_by_product_ids.assert_called_with( '2078459,2078461') def test_get_upcs_by_product_ids_not_found(mocker): """Test get upcs by invalid product ids.""" expected_response = response.create_not_found_response( message='All product ids are invalid') mocker.patch.object( product_logic, 'get_upcs_by_product_ids', return_value=expected_response) with application.app.test_request_context( '?product_ids=123456781222,1234567871'): handler_response = handlers.get_upcs_by_product_ids() assert handler_response.status_code == 404 product_logic.get_upcs_by_product_ids.assert_called_with( '123456781222,1234567871') def test_get_products_by_upcs_success( mocker, test_products_by_upcs): """Test get products by valid upcs.""" message = {'items': test_products_by_upcs} mocker.patch.object( product_logic, 'get_products_by_upcs', return_value=response.Response(message=message)) with application.app.test_request_context( data=json.dumps({'upcs': ['123456781222', '123456000000']}), content_type='application/json'): result = handlers.get_products_by_upcs() assert result.status_code == 200 assert product_logic.get_products_by_upcs.called def test_get_products_by_upcs_with_sorting_success( mocker, test_products_by_upcs): """Test get products by valid upcs with sort parameters.""" message = {'items': test_products_by_upcs} mocker.patch.object( product_logic, 'get_products_by_upcs', return_value=response.Response(message=message)) request_params = { 'upcs': ['123456781222', '123456000000'], 'sort_param': 'release_id', 'sort_direction': 'desc' } with application.app.test_request_context( data=json.dumps(request_params), content_type='application/json'): result = handlers.get_products_by_upcs() assert result.status_code == 200 product_logic.get_products_by_upcs.assert_called_with(request_params) def test_get_products_by_upcs_not_found(mocker): """Test get upcs by invalid product ids.""" expected_response = response.create_not_found_response( message='All upcs are invalid') mocker.patch.object( product_logic, 'get_products_by_upcs', return_value=expected_response) with application.app.test_request_context( data=json.dumps({'upcs': ['123', '456']}), content_type='application/json'): handler_response = handlers.get_products_by_upcs() assert handler_response.status_code == 404 product_logic.get_products_by_upcs.assert_called_with( {'upcs': ['123', '456']}) @pytest.mark.parametrize('with_company_brand,with_tenant_uuids', [ # with only with_company_brand (False, None), (True, None), (None, None), # with only with_tenant_uuids (None, False), (None, True), # mix of both (False, False), (True, False), (False, True), (True, True), ]) def test_get_product_document_success(with_company_brand, with_tenant_uuids, mocker): """Test error when passing grass headers.""" product_id = 123 get_product_document_mocked = mocker.patch.object( product_logic, 'get_product_document', return_value=response.Response()) params = {} if with_company_brand: params['with_company_brand'] = json.dumps(with_company_brand) if with_tenant_uuids: params['with_tenant_uuids'] = json.dumps(with_tenant_uuids) with application.app.test_request_context( query_string=params ): res = handlers.get_product_document(product_id) assert res.status_code == 200 get_product_document_mocked.assert_called_once_with( product_id, with_company_brand or False, with_tenant_uuids or False) def test_get_products_documents_success(mocker): """Test success for get_products_documents_success.""" product_ids = '123, 456' mocker.patch.object( product_logic, 'get_products_documents', return_value=response.Response()) data = {'product_ids': product_ids} with application.app.test_request_context( data=json.dumps(data), content_type='application/json'): product_post_response = handlers.get_products_documents() assert product_post_response.status_code == 200 def test_handler_delete_release_subgenre_success(test_product, mocker): """Test delete product success.""" mocker.patch.object( release_subgenre, 'delete', return_value=response.Response(status=200) ) with application.app.test_request_context(): product_delete_response = handlers.delete_release_subgenre( test_product) assert product_delete_response.status_code == 200 def test_handler_get_release_subgenre(test_product, mocker): """Test get product success.""" mocker.patch.object( release_subgenre, 'get', return_value=response.Response(status=200) ) with application.app.test_request_context(): product_get_response = handlers.get_release_subgenre( test_product) assert product_get_response.status_code == 200 def test_handler_create_release_subgenre(test_product, mocker): """Test post product success.""" mocker.patch.object( release_subgenre, 'create', return_value=response.Response(status=200) ) with application.app.test_request_context( data=json.dumps({'product_name': 'test name'}), content_type='application/json'): product_post_response = handlers.create_release_subgenre( test_product) assert product_post_response.status_code == 200 def test_handler_update_release_subgenre(test_product, mocker): """Test update product success.""" mocker.patch.object( release_subgenre, 'update', return_value=response.Response(status=200) ) with application.app.test_request_context( data=json.dumps({}), content_type='application/json'): product_post_response = handlers.update_release_subgenre( test_product) assert product_post_response.status_code == 200 def test_handler_delete_release_artist_success(test_product, mocker): """Test delete product success.""" mocker.patch.object( release_artist, 'delete', return_value=response.Response(status=200) ) with application.app.test_request_context(): product_delete_response = handlers.delete_release_artist( test_product) assert product_delete_response.status_code == 200 def test_handler_get_release_artist(test_product, mocker): """Test get product success.""" mocker.patch.object( release_artist, 'get', return_value=response.Response(status=200) ) with application.app.test_request_context(): product_get_response = handlers.get_release_artist( test_product) assert product_get_response.status_code == 200 def test_handler_create_release_artist(test_product, mocker): """Test post product success.""" mocker.patch.object( release_artist, 'create', return_value=response.Response(status=200) ) with application.app.test_request_context( data=json.dumps({'product_name': 'test name'}), content_type='application/json'): product_post_response = handlers.create_release_artist( test_product) assert product_post_response.status_code == 200 def test_handler_update_release_artist(test_product, mocker): """Test update product success.""" mocker.patch.object( release_artist, 'update', return_value=response.Response(status=200) ) with application.app.test_request_context( data=json.dumps({}), content_type='application/json'): product_post_response = handlers.update_release_artist( test_product) assert product_post_response.status_code == 200 @pytest.mark.parametrize( ('account_id', 'account_type'), [(1234, 'vendor'), (5, 'subaccount')]) def test_check_products_ownership_by_upcs_success( account_id, account_type, mocker): """Test GET /vendor/{vendor_id}/upcs. Verify handler success with logic layer success. """ mocker.patch.object( product_logic, 'check_products_ownership_by_upcs', return_value=response.Response()) with application.app.test_request_context( data=json.dumps({'upcs': ['123456781222', '123456000000']}), content_type='application/json'): ownership_response = handlers.check_products_ownership_by_upcs( account_type, account_id) assert ownership_response.status_code == 200 @pytest.mark.parametrize( ('account_id', 'account_type'), [(1234, 'vendor'), (5, 'subaccount')]) def test_check_products_ownership_by_upcs_not_found( account_id, account_type, mocker): """Test ownership fails if products are not found.""" mocker.patch.object( product_logic, 'check_products_ownership_by_upcs', return_value=response.create_not_found_response()) with application.app.test_request_context( data=json.dumps({'upcs': ['123456781222', '123456000000']}), content_type='application/json'): ownership_response = handlers.check_products_ownership_by_upcs( account_type, account_id) assert ownership_response.status_code == 404 def test_check_products_ownership_by_upcs_error(mocker): """Test product ownership without providing a vendor or subaccount.""" mocker.patch.object( product_logic, 'check_products_ownership_by_upcs', return_value=response.create_not_found_response()) with application.app.test_request_context( data=json.dumps({'upcs': ['123456781222', '123456000000']}), content_type='application/json'): ownership_response = handlers.check_products_ownership_by_upcs( header.GRASS_ACCOUNT_TYPE_VENDOR, 'None') assert ownership_response.status_code == 404 def test_get_product_by_upc_with_oa_headers(mocker, test_valid_product): """Test get_product_by_upc handles OA headers.""" mocker.patch.object(flask_request, 'verify_grass_access', wraps=flask_request.verify_grass_access) mocker.patch.object(product_logic, 'get_product_by_upc', return_value=response.Response( message=test_valid_product)) with application.app.test_request_context(headers={ header.ORCHARD_USER_ID: 'oa:111', }): product_response = handlers.get_product_by_upc( '194491190514') flask_request.verify_grass_access.assert_called_once assert product_response.status_code == 200 def test_get_product_by_product_id_with_oa_headers(mocker, test_valid_product): """Test get_product_by_product_id handles OA headers.""" mocker.patch.object(flask_request, 'verify_grass_access', wraps=flask_request.verify_grass_access) mocker.patch.object(product_logic, 'get_product_by_product_id', return_value=response.Response( message=test_valid_product)) with application.app.test_request_context(headers={ header.ORCHARD_USER_ID: 'oa:111', }): product_response = handlers.get_product_by_product_id( '10000') flask_request.verify_grass_access.assert_called_once assert product_response.status_code == 200 def test_get_product_by_product_id_with_tenant_uuids(mocker, test_valid_product): """Test get_product_by_product_id handles OA headers.""" mocker.patch.object( flask_request, 'verify_grass_access', wraps=flask_request.verify_grass_access) mocker.patch.object( product_logic,'get_product_by_product_id', return_value=response.Response(message=test_valid_product)) with application.app.test_request_context('/product/10000?with_tenant_uuids=1'): product_response = handlers.get_product_by_product_id(10000) flask_request.verify_grass_access.assert_called_once product_logic.get_product_by_product_id.assert_called_once_with( 10000, None, None, True) assert product_response.status_code == 200 def test_get_product_document_with_oa_headers(mocker, test_valid_product): """Test get_product_document handles OA headers.""" mocker.patch.object(product_logic, 'get_product_document', return_value=response.Response( message=test_valid_product)) with application.app.test_request_context(headers={ header.ORCHARD_USER_ID: 'oa:111', }): product_response = handlers.get_product_document( '10000') assert product_response.status_code == 200 def test_get_products_by_isrc_with_oa_headers(mocker, test_products_by_upcs): """Test get_products_by_isrc handles OA headers.""" mocker.patch.object(product_logic, 'get_products_by_isrc', return_value=response.Response( message=test_products_by_upcs)) with application.app.test_request_context(headers={ header.ORCHARD_USER_ID: 'oa:111', }): product_response = handlers.get_products_by_isrc( 'QMEU31824811') assert product_response.status_code == 200 def test_get_products_by_isrc_for_vendor(mocker, test_products_by_upcs): """Test get_products_by_isrc rejects non-OA headers.""" mocker.patch.object(product_logic, 'get_products_by_isrc', return_value=response.Response( message=test_products_by_upcs)) with application.app.test_request_context(headers={ header.GRASS_ACCOUNT_TYPE: header.GRASS_ACCOUNT_TYPE_VENDOR, header.GRASS_ACCOUNT_ID: 1}): product_response = handlers.get_products_by_isrc( 'QMEU31824811') assert product_response.status_code == 403 def test_handler_get_phonetic_translations(mocker): """Test get product success.""" mocker.patch.object( release_phonetic_translations, 'get', return_value=response.Response(status=200)) with application.app.test_request_context(): get_response = handlers.get_phonetic_translations(1, None) assert get_response.status_code == 200 def test_handler_get_phonetic_translations_error(mocker): """Test get phonetic translations error.""" mocker.patch.object( release_phonetic_translations, 'get', return_value=response.Response(status=400)) with application.app.test_request_context(): get_response = handlers.get_phonetic_translations(None, None) assert get_response.status_code == 400 def test_handler_create_phonetic_translations(mocker): """Test create phonetic translations with success.""" message = {'test': 'test'} mocker.patch.object( release_phonetic_translations, 'create', return_value=response.Response(message=message)) with application.app.test_request_context( data=json.dumps([ {'field_name': 'test', 'phonetic_translation': 'test'} ]), content_type='application/json'): create_response = handlers.create_phonetic_translations( message) assert create_response.status_code == 200 def test_handler_update_phonetic_translations(mocker): """Test update phonetic translations with success.""" message = {'test': 'test'} mocker.patch.object( release_phonetic_translations, 'update', return_value=response.Response(message=message)) with application.app.test_request_context( data=json.dumps([ {'field_name': 'test', 'phonetic_translation': 'test'} ]), content_type='application/json'): update_response = handlers.update_phonetic_translations(1) assert update_response.status_code == 200 def test_handler_delete_phonetic_translations(mocker): """Test delete phonetic translations with success.""" message = {'test': 'test'} mocker.patch.object( release_phonetic_translations, 'delete', return_value=response.Response(message='sucessfully deleted rows')) with application.app.test_request_context( data=json.dumps(message), content_type='application/json'): delete_response = handlers.delete_phonetic_translations(1) assert delete_response.status_code == 200 def test_get_product_by_release_artist_id_with_oa_headers( mocker, test_valid_product): """Test get_product_by_release_artist_id handles OA headers.""" mocker.patch.object(product_logic, 'get_product_by_release_artist_id', return_value=response.Response( message=test_valid_product)) with application.app.test_request_context(headers={ header.ORCHARD_USER_ID: 'oa:111' }): product_response = handlers.get_products_by_release_artist_id( 12345) assert product_response.status_code == 200 def test_get_product_by_release_artist_id_with_vendor_headers( mocker, test_valid_product): """Test get_product_by_release_artist_id handles OA headers.""" mocker.patch.object(product_logic, 'get_product_by_release_artist_id', return_value=response.Response( message=test_valid_product)) with application.app.test_request_context(headers={ header.GRASS_ACCOUNT_TYPE: header.GRASS_ACCOUNT_TYPE_VENDOR, header.GRASS_ACCOUNT_ID: 1}): product_response = handlers.get_products_by_release_artist_id( 12345) assert product_response.status_code == 500 MARK_USED_ALL_TYPES = [ (json.dumps({'mark_used': True})), (json.dumps({'mark_used': False})), (json.dumps({'mark_used': None})), (json.dumps({})) ] @pytest.mark.parametrize('mark_used', MARK_USED_ALL_TYPES) def test_handler_provision_upc_happy_path(mocker, mark_used): """Test provision upc with success.""" mocker.patch.object( upc_logic, 'retrieve_upc', return_value=response.Response(message='{"upc": "1234"}')) with application.app.test_request_context( content_type='application/json', data=mark_used): provisioner_response = handlers.provision_upc() assert provisioner_response.status_code == 200 assert provisioner_response.data == b'{"upc": "1234"}' RESPONSE_STATUS = [ (response.create_not_found_response(), 404), (response.create_fatal_response(), 500) ] @pytest.mark.parametrize('mark_used', MARK_USED_ALL_TYPES) @pytest.mark.parametrize('return_value, status_code', RESPONSE_STATUS) def test_handler_provision_upc_returns_error( mocker, mark_used, return_value, status_code): """Test provision upc returns 404 or 500.""" mocker.patch.object( upc_logic, 'retrieve_upc', return_value=return_value ) with application.app.test_request_context( content_type='application/json', data=mark_used): provisioner_response = handlers.provision_upc() assert provisioner_response.status_code == status_code def test_get_asset_version(mocker): """Test get asset version.""" message = {'release_id': '1234', 'api_version': 'v2'} mocker.patch.object( release_asset_version, 'get', return_value=response.Response(message=message)) with application.app.test_request_context(): handler_response = handlers.get_asset_version(1234) assert handler_response.status_code == 200 def test_update_asset_version(mocker): """Test update asset version.""" product_id = 1234 api_version = 'api_version' expected_response = {'release_id': product_id, 'api_version': api_version} mocker.patch.object( release_asset_version, 'update_api_version', return_value=response.Response(message=expected_response)) with application.app.test_client() as c: handler_response = c.put( 'product/{product_id}/asset-version'.format(product_id=product_id), data=json.dumps({'api_version': api_version}), content_type='application/json') assert handler_response.status_code == 200 release_asset_version.update_api_version.assert_called_with( product_id, api_version) actual_result = json.loads(handler_response.data.decode()) assert actual_result == expected_response def test_update_asset_version_product_empty_api_version(mocker): """Test update asset version.""" product_id = 1234 api_version = '' expected_response = {'release_id': product_id, 'api_version': api_version} mocker.patch.object( release_asset_version, 'update_api_version', return_value=response.Response(message=expected_response)) with application.app.test_client() as c: handler_response = c.put( 'product/{product_id}/asset-version'.format(product_id=product_id), data=json.dumps({'api_version': api_version}), content_type='application/json') assert handler_response.status_code == 400 def test_get_hfa_eligible_tracks_success(mocker): """Test get_hfa_pending_tracks success.""" expected_response = {'mock': 'mockhfadata'} mocker.patch.object( hfa, 'get_hfa_tracks_for_processing', return_value=response.Response(message=expected_response) ) handler_response = handlers.get_hfa_eligible_tracks() assert handler_response.status_code == 200 def test_get_product_id_by_upc_dataloader_success(mocker): """Test get_product_id_by_upc_dataloader returns products for valid upcs.""" test_data = {'upcs': ['123456789012', '987654321098']} expected_response = {'products': [ {'upc': '123456789012', 'product_id': 1}, {'upc': '987654321098', 'product_id': 2} ]} mocker.patch.object( product_logic, 'get_product_id_by_upc_dataloaded', return_value=response.Response(message=expected_response) ) with application.app.test_request_context( data=json.dumps(test_data), content_type='application/json' ): handler_response = handlers.get_product_id_by_upc_dataloader() assert handler_response.status_code == 200 actual = json.loads(handler_response.data.decode()) assert actual == expected_response def test_get_product_id_by_upc_dataloader_empty_payload(mocker): """Test get_product_id_by_upc_dataloader with empty payload.""" expected_response = {'products': []} mocker.patch.object( product_logic, 'get_product_id_by_upc_dataloaded', return_value=response.Response(message=expected_response) ) with application.app.test_request_context( data=json.dumps({}), content_type='application/json' ): handler_response = handlers.get_product_id_by_upc_dataloader() assert handler_response.status_code == 200 actual = json.loads(handler_response.data.decode()) assert actual == expected_response def test_get_product_id_by_upc_dataloader_invalid_json(mocker): """Test get_product_id_by_upc_dataloader handles invalid JSON gracefully.""" expected_response = {'products': []} mocker.patch.object( product_logic, 'get_product_id_by_upc_dataloaded', return_value=response.Response(message=expected_response) ) with application.app.test_request_context( data="not a json", content_type='application/json' ): handler_response = handlers.get_product_id_by_upc_dataloader() assert handler_response.status_code == 200 actual = json.loads(handler_response.data.decode()) assert actual == expected_response @pytest.mark.parametrize( 'data,expected_status_code,expected_response', [ pytest.param( None, 400, {'code': 'bad_request', 'message': 'No product ids in request body'}, id='An null request is a bad request', ), pytest.param( {}, 400, {'code': 'bad_request', 'message': 'No product ids in request body'}, id='An empty request is a bad request', ), pytest.param( {'product_ids': []}, 400, {'code': 'bad_request', 'message': 'No product ids in request body'}, id='A list of empty product ids is a bad request', ), pytest.param( {'product_ids': [12, 'not', 'ints', False]}, 400, {'code': 'bad_request', 'message': 'All product ids must be integers'}, id='All the product ids need to be ints', ), pytest.param( {'product_ids': [1, 2, 3, 77]}, 200, {'product_ids': [None, None, None, None]}, id='Uses product_logic and flaskify\'s its response', ), ] ) @patch('product.handlers.product_logic.lookup_product_ownership') @patch('product.handlers.g', spec=['request_context']) def test_lookup_product_ownership( mock_g: MagicMock, mock_lookup_product_ownership: MagicMock, data: dict | None, expected_status_code: int, expected_response: dict, ) -> None: """Test lookup_product_ownership.""" mock_g.request_context.jwt_identity_id = 'this is an identity uuid' mock_lookup_product_ownership.return_value = response.Response(message={ 'product_ids': [None, None, None, None] }) with application.app.test_request_context( data=json.dumps(data), content_type='application/json' ): actual_response = handlers.lookup_product_ownership() assert actual_response.status_code == expected_status_code, actual_response.data assert json.loads(actual_response.data.decode()) == expected_response, actual_response.data if expected_status_code == 200 and data: mock_lookup_product_ownership.assert_called_once_with(data.get('product_ids')) else: mock_lookup_product_ownership.assert_not_called() @patch('product.handlers.g', spec=['request_context']) def test_lookup_product_ownership_no_jwt( mock_g: MagicMock, ) -> None: """Test lookup_product_ownership rejects when the JWT has no identity id.""" mock_g.request_context.jwt_identity_id = None with application.app.test_request_context( data=json.dumps({'product_ids': [1, 2, 3, 77]}), content_type='application/json' ): response = handlers.lookup_product_ownership() assert response.status_code == 401, response.data assert json.loads(response.data.decode()) == \ {"code": "authorization_error", "message": "Invalid Authorization"}