"""Test for the logic of vendor.""" from typing import Any from unittest.mock import MagicMock, patch from uuid import UUID, uuid4 import pytest from owsresponse import response from python_pdp_sdk import ( ResourceWithAttributes, ) from pythonfeatures import pythonfeatures from account.constants import connectors, constants from account.constants.features import FEATURES, FOUNDATION_ENABLED_FEATURES, SME_ENABLED_FEATURES from account.logic import vendor from account.models import ( neo4j_identity, neo4j_vendor, vend_contact, vendor as vendor_model, ) from tests.unit.helpers import vendor as vendor_test_helper # Feature constants for testing - mirrors constants/features.py ALL_FEATURES = [ 1, 2, 3, 4, 5, 6, 7, 9, 10, 11, 12, 13, 14, 15, 16, 18, 19, 20, 21, 22, 23, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, ] # Enabled features for each brand # AUDIO_RELEASE_BUILDER, PROJECT_MODE, DYI AWAL_ENABLED_FEATURES = [1, 13, 32] # AUDIO_RELEASE_BUILDER, MUSIC_VIDEO_RELEASE_BUILDER, MARKETPLACE, PROJECT_MODE, # ACCOUNTING, ANALYTICS, SHOW_GROSS_ACCOUNTING ALTAFONTE_ENABLED_FEATURES = [ 1, 2, 5, 6, 9, 13, 23, ] # AUDIO_RELEASE_BUILDER, MUSIC_VIDEO_RELEASE_BUILDER, MARKETPLACE, # ACCOUNTING, ANALYTICS, PROJECT_MODE THEORCHARD_ENABLED_FEATURES = [ 1, 2, 5, 6, 9, 13, ] # Restricted features (ALL_FEATURES - enabled_features) ALTAFONTE = [f for f in ALL_FEATURES if f not in ALTAFONTE_ENABLED_FEATURES] AWAL = [f for f in ALL_FEATURES if f not in AWAL_ENABLED_FEATURES] ORCHARD_BRANDS = [f for f in ALL_FEATURES if f not in THEORCHARD_ENABLED_FEATURES] SME_BRANDS = [f for f in ALL_FEATURES if FEATURES(f) not in SME_ENABLED_FEATURES] FOUNDATION = [f for f in ALL_FEATURES if FEATURES(f) not in FOUNDATION_ENABLED_FEATURES] # 3rd Party Payee Accounts: only ACCOUNTING (6) enabled THIRD_PARTY_PAYEE_ENABLED_FEATURES = [6] THIRD_PARTY_PAYEE = [f for f in ALL_FEATURES if f not in THIRD_PARTY_PAYEE_ENABLED_FEATURES] @pytest.fixture def fixture_valid_v2_vendor(): """Fixture response for a vendor.""" return response.Response( { 'vendor_id': 1234, 'is_distributor': 'N', 'owner': 'awal', 'name': 'Valid Vendor', 'company': 'Test Company', 'country_id': 1, 'status': 'signed', 'label_identifier': None, 'contact_email': 'contact@mail.com', 'migrated_to_abacus': False, 'assigned_to': 2903, 'quarterback_label_manager': None, 'company_brand': 'awal', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'country': 1, 'account_name': 'Blacktop 2', 'label_summary': 'Test Label', 'priority': 3, 'primary_genre': 20, 'transfer_pricing_country': 'orchard_ny', 'product_manager_id': 1, } ) @pytest.fixture def fixture_valid_v2_vendor_empty_update(): """Fixture response for a vendor.""" return response.Response( { 'vendor_id': 1234, 'label_summary': '', 'website': '', 'est_releases': 0, 'est_tracks': 0, 'is_distributor': 'N', 'owner': 'awal', 'name': 'Hudson & Sons', 'company': 'Test Company', 'country_id': 1, 'status': 'signed', 'label_identifier': None, 'contact_email': 'contact@mail.com', 'migrated_to_abacus': False, 'assigned_to': 2903, 'quarterback_label_manager': None, 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'country': 1, 'company_brand': 'awal', 'account_name': 'Blacktop 2', 'priority': 3, 'primary_genre': 20, 'transfer_pricing_country': 'orchard_ny', 'product_manager_id': 1, 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', } ) @pytest.fixture def fixture_valid_vendor_no_company(): """Fixture response for a vendor with empty company field.""" return response.Response( { 'vendor_id': 1, 'is_distributor': 'Y', 'name': 'Funky Vendor', 'country_id': 1, 'owner': 'odd', 'status': 'signed', 'label_identifier': None, 'contact_email': 'contact@mail.com', 'migrated_to_abacus': False, 'assigned_to': 2903, 'quarterback_label_manager': None, 'company_brand': 'awal', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'country': 1, 'account_name': 'Blacktop 2', 'label_summary': 'Test Label', 'priority': 3, 'primary_genre': 20, 'transfer_pricing_country': 'orchard_ny', 'product_manager_id': 1, } ) @pytest.fixture def fixture_valid_vendor_no_name(): """Fixture response for a vendor with empty name field.""" return response.Response( { 'vendor_id': 1, 'is_distributor': 'Y', 'name': None, 'country_id': 1, 'owner': 'odd', 'status': 'signed', 'label_identifier': None, 'contact_email': 'contact@mail.com', 'migrated_to_abacus': False, 'assigned_to': 2903, 'quarterback_label_manager': None, 'company_brand': 'awal', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'country': 1, 'account_name': 'Blacktop 2', 'label_summary': 'Test Label', 'priority': 3, 'primary_genre': 20, 'transfer_pricing_country': 'orchard_ny', } ) @pytest.fixture def fixture_distributor_vendor(): """Fixture response for a vendor that is a distributor.""" return response.Response({'vendor_id': 1, 'is_distributor': 'Y'}) @pytest.fixture def fixture_neo4j_fields(): """Fixture response for neo4j vendor info.""" return response.Response( { 'country': 'USA', 'source': 'gda.KafkaTopic', 'company_brand': 'awal', 'parent_company': 'theorchard', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'service_tier_name': 'diy-tier-1', 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', } ) @pytest.fixture def fixture_not_found_response(): """Fixture for a not found response.""" return response.create_not_found_response() @pytest.mark.parametrize( ('owner', 'brand', 'expected_features'), [ pytest.param( '3rd Party Payee Acco', 'awal', THIRD_PARTY_PAYEE, id='3rd party payee with awal' ), pytest.param( '3rd Party Payee Acco', 'theorchard', THIRD_PARTY_PAYEE, id='3rd party payee with theorchard', ), pytest.param('AWAL-Core', 'awal', AWAL, id='AWAL-Core owner with awal brand'), pytest.param('TTO', 'theorchard', ORCHARD_BRANDS, id='TTO owner with theorchard brand'), pytest.param('Sony Music', 'sme', SME_BRANDS, id='Sony Music owner with sme brand'), ], ) def test_get_restricted_features_for_vendor(owner, brand, expected_features): """Test that get_restricted_features_for_vendor returns correct features based on owner.""" result = vendor.get_restricted_features_for_vendor(owner, brand) assert result == expected_features def test_is_distributor_true(monkeypatch, fixture_distributor_vendor): """Test check for distributor where vendor is a distributor.""" monkeypatch.setattr( vendor_model, 'get_distributor', MagicMock(return_value=fixture_distributor_vendor), ) result = vendor.is_distributor(1) assert result assert result is not fixture_distributor_vendor def test_is_distributor_false(monkeypatch, fixture_not_found_response): """Test check for distributor where vendor is not a distributor.""" monkeypatch.setattr( vendor_model, 'get_distributor', MagicMock(return_value=fixture_not_found_response), ) result = vendor.is_distributor(1) assert not result assert result.status == 204 def test_is_vendor(monkeypatch, fixture_distributor_vendor): """Test is_vendor when model returns a vendor obj.""" monkeypatch.setattr( vendor_model, 'get_vendor', MagicMock(return_value=fixture_distributor_vendor) ) result = vendor.is_vendor(1) assert result def test_is_vendor_invalid_id(monkeypatch, fixture_not_found_response): """Test is_vendor when model returns a not found response.""" monkeypatch.setattr( vendor_model, 'get_vendor', MagicMock(return_value=fixture_not_found_response) ) result = vendor.is_vendor(100) assert not result assert result.status == 404 def test_get_vendor_closers_true(monkeypatch): """Test check for closers where vendor exists.""" vendor_uuids = [ '87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538', 'e429b770-9db6-4f98-b48d-7d9b8ced7e6c', ] model_vendors_response = [ {'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'closers': [1]}, {'uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'closers': [2]}, ] expected_result = [ {'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'closers': [1]}, {'uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'closers': [2]}, None, ] monkeypatch.setattr( vendor_model, 'get_vendor_closers', MagicMock(return_value=model_vendors_response) ) result = vendor.get_vendor_closers(vendor_uuids) assert result == expected_result vendor_model.get_vendor_closers.assert_called_once_with(vendor_uuids) def test_get_vendors_first_statement_period_true(monkeypatch): """Test check for first_statement_period where vendor exists.""" vendor_uuids = [ '87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538', 'e429b770-9db6-4f98-b48d-7d9b8ced7e6c', ] model_vendors_response = [ {'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'first_statement_period': '1'}, {'uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'first_statement_period': '1'}, ] expected_result = { 'vendors': [ {'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'first_statement_period': '1'}, {'uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'first_statement_period': '1'}, None, ] } monkeypatch.setattr( vendor_model, 'get_vendors_first_statement_period', MagicMock(return_value=model_vendors_response), ) result = vendor.get_vendors_first_statement_period(vendor_uuids) assert result == expected_result vendor_model.get_vendors_first_statement_period.assert_called_once_with(vendor_uuids) @patch('account.logic.vendor.g', spec=['log', 'request_context']) @pytest.mark.parametrize( ('fixture_vendor', 'expected_vendor'), [ ( vendor_test_helper.fixture_valid_vendor_no_name(), { 'vendor_id': 1, 'is_distributor': 'Y', 'account_name': 'Funky Vendor', 'contact_name': None, 'country_id': 1, 'owner': 'odd', 'vendor_name': 'Foo Productions', 'company_brand': 'test_awal', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'service_tier_name': 'managed-basic', 'product_manager': 'foo bar', 'sc_usa': 'test1', 'sc_ca': 'test2', 'recurring_payment_threshold': None, 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', 'assigned_reviewer': 123, }, ), ( vendor_test_helper.fixture_valid_vendor_no_company(), { 'vendor_id': 1, 'is_distributor': 'Y', 'contact_name': 'Funky Vendor', 'account_name': None, 'country_id': 1, 'owner': 'odd', 'vendor_name': 'Foo Productions', 'company_brand': '', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'service_tier_name': 'managed-basic', 'product_manager': 'foo bar', 'sc_usa': 'test1', 'sc_ca': 'test2', 'recurring_payment_threshold': None, 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', 'assigned_reviewer': 123, }, ), ], ) def test_get_vendor_label_info(mock_g, monkeypatch, fixture_vendor, expected_vendor): """Test get vendor success.""" neo4j_response = { 'vendor_name': 'Foo Productions', 'company_brand': 'awal', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'service_tier_name': 'managed-basic', 'sc_usa': 'test1', 'sc_ca': 'test2', 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', } monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response(message='disabled')), ) monkeypatch.setattr( vendor_model, 'get_vendor_label_info', MagicMock(return_value=fixture_vendor) ) monkeypatch.setattr( vendor_model, 'get_recurring_payment_threshold', MagicMock(return_value=response.create_not_found_response()), ) monkeypatch.setattr( vendor_model, 'get_product_manager_id', MagicMock(return_value=response.Response({'product_manager': 'foo bar'})), ) monkeypatch.setattr( vendor_model, 'get_soundscan_code', MagicMock(return_value=response.Response({'soundscan_code': 'test1'})), ) monkeypatch.setattr( vendor_model, 'get_soundscan_code', MagicMock(return_value=response.Response({'soundscan_code': 'test2'})), ) monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response('enabled')), ) monkeypatch.setattr( neo4j_vendor, 'get_neo4j_vendor_info', MagicMock(return_value=response.Response(neo4j_response)), ) result = vendor.get_vendor_label_info(1) assert result assert result.message == expected_vendor assert result.message['vendor_uuid'] neo4j_vendor.get_neo4j_vendor_info.assert_called_once_with(1) @patch('account.logic.vendor.g', spec=['log', 'request_context']) @patch('account.logic.vendor.get_session') @pytest.mark.parametrize( ('fixture_vendor', 'expected_vendor'), [ ( vendor_test_helper.fixture_valid_vendor_no_name(), { 'vendor_id': 1, 'is_distributor': 'Y', 'account_name': 'Funky Vendor', 'contact_name': None, 'country_id': 1, 'owner': 'odd', 'vendor_name': 'Foo Productions', 'company_brand': 'test_awal', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'service_tier_name': 'managed-basic', 'product_manager': 'foo bar', 'sc_usa': 'test1', 'sc_ca': 'test2', 'recurring_payment_threshold': None, 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', 'assigned_reviewer': 123, }, ), ( vendor_test_helper.fixture_valid_vendor_no_company(), { 'vendor_id': 1, 'is_distributor': 'Y', 'contact_name': 'Funky Vendor', 'account_name': None, 'country_id': 1, 'owner': 'odd', 'vendor_name': 'Foo Productions', 'company_brand': '', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'service_tier_name': 'managed-basic', 'product_manager': 'foo bar', 'sc_usa': 'test1', 'sc_ca': 'test2', 'recurring_payment_threshold': None, 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', 'assigned_reviewer': 123, }, ), ], ) def test_get_vendor_label_info_with_neo_tx( neo4j_session, mock_g, monkeypatch, fixture_vendor, expected_vendor ): """Test get vendor success.""" neo4j_response = { 'vendor_name': 'Foo Productions', 'company_brand': 'awal', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'service_tier_name': 'managed-basic', 'sc_usa': 'test1', 'sc_ca': 'test2', 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', } monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response(message='disabled')), ) monkeypatch.setattr( vendor_model, 'get_vendor_label_info', MagicMock(return_value=fixture_vendor) ) monkeypatch.setattr( vendor_model, 'get_recurring_payment_threshold', MagicMock(return_value=response.create_not_found_response()), ) monkeypatch.setattr( vendor_model, 'get_product_manager_id', MagicMock(return_value=response.Response({'product_manager': 'foo bar'})), ) monkeypatch.setattr( vendor_model, 'get_soundscan_code', MagicMock(return_value=response.Response({'soundscan_code': 'test1'})), ) monkeypatch.setattr( vendor_model, 'get_soundscan_code', MagicMock(return_value=response.Response({'soundscan_code': 'test2'})), ) monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response('enabled')), ) monkeypatch.setattr( neo4j_vendor, 'get_neo4j_vendor_info', MagicMock(return_value=response.Response(neo4j_response)), ) mock_session = MagicMock() mock_session.execute_read.return_value = response.Response(neo4j_response) neo4j_session.return_value = mock_session result = vendor.get_vendor_label_info(1, neo_tx=True) assert result assert result.message == expected_vendor assert result.message['vendor_uuid'] neo4j_vendor.get_neo4j_vendor_info.assert_not_called() def test_get_vendors(monkeypatch): """Test get vendors success in correct order.""" init_vendor_uuids = ['four', 'three', 'two', 'one'] vendor_uuids = ['three', 'two', 'one'] model_vendors_response = [ {'vendor_uuid': 'one', 'status': 'signed'}, {'vendor_uuid': 'two', 'status': 'deletion'}, {'vendor_uuid': 'three', 'status': 'pending'}, ] expected_result = { 'vendors': [ None, {'uuid': 'three', 'status': 'pending'}, {'uuid': 'two', 'status': 'deletion'}, {'uuid': 'one', 'status': 'signed'}, ] } monkeypatch.setattr( vendor_model, 'get_vendors', MagicMock(return_value=model_vendors_response), ) result = vendor.get_vendors(vendor_uuids, init_vendor_uuids) assert result == expected_result vendor_model.get_vendors.assert_called_once_with(['three', 'two', 'one']) def test_get_vendor_names(monkeypatch): """Test get vendor names success in correct order.""" vendor_uuids = ['three', 'two', 'one'] model_vendors_response = [ {'vendor_uuid': 'one', 'name': 'Vendor One'}, {'vendor_uuid': 'two', 'name': 'Vendor Two'}, {'vendor_uuid': 'three', 'name': 'Vendor Three'}, ] expected_result = { 'vendors': [ {'uuid': 'three', 'name': 'Vendor Three'}, {'uuid': 'two', 'name': 'Vendor Two'}, {'uuid': 'one', 'name': 'Vendor One'}, ] } monkeypatch.setattr( vendor_model, 'get_vendor_names', MagicMock(return_value=model_vendors_response), ) result = vendor.get_vendor_names(vendor_uuids) assert result == expected_result vendor_model.get_vendor_names.assert_called_once_with(['three', 'two', 'one']) @patch('account.logic.vendor.g', spec=['log', 'request_context']) def test_get_vendor_label_info_with_ff_update_company_brand_art_relations(mock_g, monkeypatch): company_brand_art_relations = 'awal_art_relations' monkeypatch.setattr( vendor_model, 'get_vendor_label_info', MagicMock( return_value=response.Response( { 'vendor_id': 1, 'name': 'test', 'company_brand': company_brand_art_relations, 'company': 'Funky Vendor', 'vendor_uuid': None, } ) ), ) monkeypatch.setattr( vendor_model, 'get_recurring_payment_threshold', MagicMock(return_value=response.create_not_found_response()), ) monkeypatch.setattr( vendor_model, 'get_product_manager_id', MagicMock(return_value=response.Response({'product_manager': 'foo bar'})), ) monkeypatch.setattr( vendor_model, 'get_soundscan_code', MagicMock(return_value=response.Response({'soundscan_code': 'test1'})), ) monkeypatch.setattr( vendor_model, 'get_soundscan_code', MagicMock(return_value=response.Response({'soundscan_code': 'test2'})), ) monkeypatch.setattr( pythonfeatures, 'get_single_feature_by_attributes', MagicMock(return_value=response.Response('enabled')), ) monkeypatch.setattr( neo4j_vendor, 'get_neo4j_vendor_info', MagicMock( return_value=response.Response( { 'vendor_name': 'Foo Productions', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'service_tier_name': 'managed-basic', 'sc_usa': 'test1', 'sc_ca': 'test2', 'vendor_uuid': None, } ) ), ) result = vendor.get_vendor_label_info(1) assert result.message.get('company_brand') == company_brand_art_relations @patch('account.logic.vendor.g', spec=['log', 'request_context']) def test_get_vendor_not_found(mock_g, monkeypatch, fixture_not_found_response): """Test get vendor success.""" monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response('disabled')), ) monkeypatch.setattr( vendor_model, 'get_vendor_label_info', MagicMock(return_value=fixture_not_found_response), ) result = vendor.get_vendor_label_info(1) assert not result assert result.status == 404 @patch('account.logic.vendor.g', spec=['log', 'request_context']) @pytest.mark.parametrize( ('input_data', 'vendor_update_details', 'sc_map_exists'), [ ( # input_data payload contains only name, not contact_name { 'name': 'Foo Productions', 'company_brand': 'knr', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'owner': 'AWAL Core Accounts', 'assigned_to_id': '2903', 'assigned_reviewer_id': '2901', 'priority': 3, 'primary_genre_id': '20', 'quarterback_label_manager_id': '1', 'sc_usa': 'test_1', 'sc_ca': 'test_2', }, { 'name': 'Foo Productions', 'company': 'Foo Productions', 'owner': 'AWAL Core Accounts', 'assigned_to': '2903', 'assigned_reviewer': '2901', 'priority': 3, 'primary_genre': '20', 'quarterback_label_manager': '1', 'vendor_id': 1234, }, True, ), ( # input_data payload contains both name and contact_name { 'name': 'Foo Productions', 'contact_name': 'TEST CONTACT NAME', 'company_brand': 'knr', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'owner': 'AWAL Core Accounts', 'assigned_to_id': '2903', 'assigned_reviewer_id': '2901', 'priority': 3, 'primary_genre_id': '20', 'quarterback_label_manager_id': '1', 'sc_usa': 'test_1', 'sc_ca': 'test_2', }, { 'name': 'Foo Productions', 'company': 'Foo Productions', 'owner': 'AWAL Core Accounts', 'assigned_to': '2903', 'assigned_reviewer': '2901', 'priority': 3, 'primary_genre': '20', 'quarterback_label_manager': '1', 'vendor_id': 1234, }, False, ), ( # input_data payload contains only contact_name, not name { 'contact_name': 'TEST CONTACT NAME', 'company_brand': 'knr', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'owner': 'AWAL Core Accounts', 'assigned_to_id': '2903', 'assigned_reviewer_id': '2901', 'priority': 3, 'primary_genre_id': '20', 'quarterback_label_manager_id': '1', 'sc_usa': 'test_1', 'sc_ca': 'test_2', }, { 'name': 'TEST CONTACT NAME', 'owner': 'AWAL Core Accounts', 'assigned_to': '2903', 'assigned_reviewer': '2901', 'priority': 3, 'primary_genre': '20', 'quarterback_label_manager': '1', 'vendor_id': 1234, }, True, ), ( # soundscan code does not exist in map { 'name': 'Foo Productions', 'contact_name': 'TEST CONTACT NAME', 'company_brand': 'knr', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'owner': 'AWAL Core Accounts', 'assigned_to_id': '2903', 'assigned_reviewer_id': '2901', 'priority': 3, 'primary_genre_id': '20', 'quarterback_label_manager_id': '1', 'sc_usa': 'test_1', 'sc_ca': 'test_2', }, { 'name': 'Foo Productions', 'company': 'Foo Productions', 'owner': 'AWAL Core Accounts', 'assigned_to': '2903', 'assigned_reviewer': '2901', 'priority': 3, 'primary_genre': '20', 'quarterback_label_manager': '1', 'vendor_id': 1234, }, False, ), ( { 'name': 'Foo Productions', 'company_brand': 'awal', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'owner': 'AWAL Core Accounts', 'assigned_to_id': '2903', 'assigned_reviewer_id': '2901', 'priority': 3, 'primary_genre_id': '20', 'quarterback_label_manager_id': '1', 'estimated_total_tracks': 15, 'sc_usa': 'test_1', 'sc_ca': 'test_2', }, { 'name': 'Foo Productions', 'company': 'Foo Productions', 'owner': 'AWAL Core Accounts', 'assigned_to': '2903', 'assigned_reviewer': '2901', 'priority': 3, 'primary_genre': '20', 'quarterback_label_manager': '1', 'est_total_tracks': 15, 'vendor_id': 1234, }, True, ), ( { 'name': 'Foo Productions', 'company_brand': 'awal', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'owner': 'AWAL Core Accounts', 'assigned_to_id': '2903', 'assigned_reviewer_id': '2901', 'priority': 3, 'primary_genre_id': '20', 'quarterback_label_manager_id': '1', 'sc_usa': 'test_1', 'sc_ca': 'test_2', }, { 'name': 'Foo Productions', 'company': 'Foo Productions', 'owner': 'AWAL Core Accounts', 'assigned_to': '2903', 'assigned_reviewer': '2901', 'priority': 3, 'primary_genre': '20', 'quarterback_label_manager': '1', 'vendor_id': 1234, }, False, ), ( { 'name': 'Foo Productions', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'owner': 'AWAL Core Accounts', 'assigned_to_id': '2903', 'assigned_reviewer_id': '2901', 'priority': 3, 'primary_genre_id': '20', 'quarterback_label_manager_id': '1', 'sc_usa': 'test_1', 'sc_ca': 'test_2', }, { 'name': 'Foo Productions', 'company': 'Foo Productions', 'owner': 'AWAL Core Accounts', 'assigned_to': '2903', 'assigned_reviewer': '2901', 'priority': 3, 'primary_genre': '20', 'quarterback_label_manager': '1', 'vendor_id': 1234, }, False, ), ( { 'vendor_id': 77160, 'name': 'Hudson & Sons', 'company': 'Hudson & Sons', 'priority': 3, 'website': 'hudson.son.com', 'label_identifier': 'Client Services', 'owner': 'AWAL-Core', 'is_owned': 'No', 'is_distributor': 'N', 'assigned_to': 'AWAL Community Support', 'quarterback_label_manager': 2717, 'date_created': '2023-06-13', 'last_update': '2023-06-15 14:01:36', 'support_contact_email': 'addison@supportsons.com', 'contact_email': 'ahanson@thefruitcom', 'sap_vendor_id': None, 'est_releases': 1, 'est_tracks': 1, 'primary_genre': 11, 'genre': 'World Music', 'country': 'USA', 'country_id': 1, 'status': 'signed', 'region': 'Central America & Carrib', 'projected_first_year_revenue': '0.000000', 'transfer_pricing_country': None, 'allow_ALW_release_builder': 'Y', 'migrated_to_abacus': 0, 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'source': None, 'vendor_uuid': 'e924f4ad-7f68-4450-907c-2561b3f6faab', 'company_brand': 'awal', 'service_tier_name': 'diy-tier-1', 'sc_usa': 'test_1', 'sc_ca': 'test_2', }, { 'vendor_id': 1234, 'name': 'Hudson & Sons', 'company': 'Hudson & Sons', 'priority': 3, 'website': 'hudson.son.com', 'label_identifier': 'Client Services', 'owner': 'AWAL-Core', 'is_owned': 'No', 'support_contact_email': 'addison@supportsons.com', 'contact_email': 'ahanson@thefruitcom', 'country_id': 1, 'region': 'Central America & Carrib', }, False, ), ], ) def test_update_vendor( mock_g, monkeypatch, input_data, vendor_update_details, sc_map_exists, fixture_valid_v2_vendor, fixture_neo4j_fields, ): """Test v2 update vendor success.""" expected = fixture_valid_v2_vendor.message expected_neo4j = fixture_neo4j_fields.message expected.update(expected_neo4j) if input_data.get('company_brand'): expected['company_brand'] = input_data['company_brand'] monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response(message='disabled')), ) if input_data.get('service_tier_uuid'): expected['service_tier_uuid'] = input_data['service_tier_uuid'] monkeypatch.setattr(vendor_model, 'update_vendor', MagicMock()) if sc_map_exists: monkeypatch.setattr(vendor_model, 'sc_map_exists', MagicMock(return_value=0)) else: monkeypatch.setattr(vendor_model, 'sc_map_exists', MagicMock(return_value=1)) monkeypatch.setattr(vendor_model, 'map_soundscan_code', MagicMock()) monkeypatch.setattr(vendor_model, 'update_soundscan_code', MagicMock()) monkeypatch.setattr( vendor_model, 'get_vendor_label_info', MagicMock(return_value=fixture_valid_v2_vendor), ) monkeypatch.setattr(vendor_model, 'get_soundscan_code', MagicMock()) monkeypatch.setattr(vendor_model, 'update_company_brand_art_relations', MagicMock()) monkeypatch.setattr(vendor_model, 'get_recurring_payment_threshold', MagicMock()) monkeypatch.setattr(vendor_model, 'get_product_manager_id', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'update_service_tier', MagicMock()) monkeypatch.setattr( neo4j_vendor, 'get_neo4j_vendor_info', MagicMock(return_value=fixture_neo4j_fields), ) monkeypatch.setattr(vendor, 'publish_account_event', MagicMock()) result = vendor.update_vendor(1234, input_data) assert result assert result.message == expected assert result.message['vendor_uuid'] input_data.update({'vendor_id': 1234, 'vendor_uuid': expected.get('vendor_uuid')}) vendor.publish_account_event.assert_called_once_with(1234, input_data, 'update') vendor_model.update_vendor.assert_called_once_with(1234, vendor_update_details) assert vendor_model.sc_map_exists.call_count == 2 if sc_map_exists: vendor_model.map_soundscan_code.assert_called_with(1234, input_data['sc_ca'], 2) assert vendor_model.map_soundscan_code.call_count == 2 vendor_model.update_soundscan_code.assert_not_called else: vendor_model.update_soundscan_code.assert_called_with(1234, input_data['sc_ca'], 2) assert vendor_model.update_soundscan_code.call_count == 2 vendor_model.map_soundscan_code.assert_not_called @patch('account.logic.vendor.g', spec=['log', 'request_context']) @pytest.mark.parametrize( 'data', [ ( { 'label_summary': '', 'estimated_total_tracks': 0, 'estimated_total_products': 0, 'website': '', 'name': 'Hudson & Sons', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', } ) ], ) def test_update_vendor_set_empty_fields( mock_g, monkeypatch, data, fixture_valid_v2_vendor_empty_update, fixture_neo4j_fields, ): """Test v2 update vendor can set certain fields to '' .""" expected = fixture_valid_v2_vendor_empty_update.message monkeypatch.setattr(vendor_model, 'update_vendor', MagicMock()) monkeypatch.setattr(vendor_model, 'update_soundscan_code', MagicMock()) monkeypatch.setattr( vendor_model, 'get_vendor_label_info', MagicMock(return_value=fixture_valid_v2_vendor_empty_update), ) monkeypatch.setattr(vendor_model, 'get_soundscan_code', MagicMock()) monkeypatch.setattr(vendor_model, 'get_recurring_payment_threshold', MagicMock()) monkeypatch.setattr(vendor_model, 'get_product_manager_id', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'update_vendor', MagicMock()) monkeypatch.setattr(vendor_model, 'update_company_brand_art_relations', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'update_service_tier', MagicMock()) monkeypatch.setattr( neo4j_vendor, 'get_neo4j_vendor_info', MagicMock(return_value=fixture_neo4j_fields), ) monkeypatch.setattr(vendor, 'publish_account_event', MagicMock()) result = vendor.update_vendor(1234, data) assert result assert result.message == expected assert vendor.publish_account_event.called @patch('account.logic.vendor.g', spec=['log', 'request_context']) @pytest.mark.parametrize('data', [({'product_manager_id': None}), ({'product_manager_id': 23456})]) def test_update_vendor_product_manager_id( mock_g, monkeypatch, data, fixture_valid_v2_vendor_empty_update, fixture_neo4j_fields, ): """Test v2 update vendor can set certain fields to '' .""" expected = fixture_valid_v2_vendor_empty_update.message monkeypatch.setattr(vendor_model, 'update_vendor', MagicMock()) if data.get('product_manager_id'): monkeypatch.setattr(vendor_model, 'update_product_manager_id', MagicMock()) else: monkeypatch.setattr(vendor_model, 'delete_product_manager_id', MagicMock()) monkeypatch.setattr(vendor_model, 'update_soundscan_code', MagicMock()) monkeypatch.setattr( vendor_model, 'get_vendor_label_info', MagicMock(return_value=fixture_valid_v2_vendor_empty_update), ) monkeypatch.setattr(vendor_model, 'get_soundscan_code', MagicMock()) monkeypatch.setattr(vendor_model, 'get_recurring_payment_threshold', MagicMock()) monkeypatch.setattr(vendor_model, 'get_product_manager_id', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'update_vendor', MagicMock()) monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response(message='disabled')), ) monkeypatch.setattr(neo4j_vendor, 'update_service_tier', MagicMock()) monkeypatch.setattr( neo4j_vendor, 'get_neo4j_vendor_info', MagicMock(return_value=fixture_neo4j_fields), ) monkeypatch.setattr(vendor, 'publish_account_event', MagicMock()) result = vendor.update_vendor(1234, data) assert result assert result.message == expected assert vendor.publish_account_event.called @patch('account.logic.vendor.g', spec=['log', 'request_context']) @pytest.mark.parametrize( 'ar_side_effect, neo4j_exception, expected', [ # no DB connection ( response.create_fatal_response('SQLAlchemyError'), Exception('Failed to update Vendor 1234 in Neo4j.'), {'code': 'internal_error', 'message': 'SQLAlchemyError'}, ), # AR issue only ( response.create_fatal_response('SQLAlchemyError'), None, {'code': 'internal_error', 'message': 'SQLAlchemyError'}, ), # Neo issue only ( MagicMock(), Exception('Failed to update Vendor 1234 in Neo4j.'), { 'code': 'internal_error', 'message': 'Unable to update vendor: Failed to update Vendor 1234 in Neo4j.', }, ), ], ) def test_update_vendor_exceptions( mock_g, monkeypatch, ar_side_effect, neo4j_exception, expected, app_context ): """Test v2 update vendor bubbles up db exceptions.""" data = { 'name': 'Foo Productions', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'owner': 'AWAL Core Accounts', 'assigned_to_id': '2903', 'assigned_reviewer_id': '2901', 'priority': 3, 'primary_genre_id': '20', 'quarterback_label_manager_id': '1', } monkeypatch.setattr(vendor_model, 'update_vendor', MagicMock(return_value=ar_side_effect)) monkeypatch.setattr( vendor_model, 'update_company_brand_art_relations', MagicMock(side_effect=ar_side_effect), ) result = vendor.update_vendor(1234, data) assert not result assert result.status == 500 assert result.errors == expected def test_update_vendor_not_found(monkeypatch, fixture_not_found_response): """Test v2 update vendor not found.""" data = { 'name': 'Foo Productions', 'company_brand': 'awal', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'assigned_to_id': '12', 'assigned_reviewer_id': '11', 'primary_genre_id': '1', 'quarterback_label_manager_id': '1', } monkeypatch.setattr( vendor_model, 'update_vendor', MagicMock(return_value=fixture_not_found_response), ) response = vendor.update_vendor(1234, data) assert not response assert response.status == 404 def test_update_vendor_closers_success(monkeypatch): """Test successful update of vendor closers publishes an account event.""" vendor_uuid = str(uuid4()) vendor_id = 4567 closers = [1, 2] last_modified_by = 1 expected = {'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id, 'closers': closers} mock_update_vendor_closers = MagicMock(return_value=expected) monkeypatch.setattr(vendor_model, 'upsert_vendor_closers', mock_update_vendor_closers) mock_publish = MagicMock() monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) result = vendor.update_vendor_closers(vendor_uuid, closers, last_modified_by) assert result == expected mock_update_vendor_closers.assert_called_once_with(vendor_uuid, closers, last_modified_by) mock_publish.assert_called_once_with( vendor_id=vendor_id, response_payload=expected, event_type=connectors.OperationType.UPDATE, ) def test_update_vendor_closers_exception(monkeypatch): """VendorUpdateException from the model propagates and no event is published.""" from account.utils.exception import VendorUpdateException vendor_uuid = str(uuid4()) closers = [1, 2] last_modified_by = 1 mock_update_vendor_closers = MagicMock( side_effect=VendorUpdateException(code='invalid_request_error', message='nope', status=500) ) monkeypatch.setattr(vendor_model, 'upsert_vendor_closers', mock_update_vendor_closers) mock_publish = MagicMock() monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) with pytest.raises(VendorUpdateException) as exc_info: vendor.update_vendor_closers(vendor_uuid, closers, last_modified_by) assert exc_info.value.status == 500 mock_update_vendor_closers.assert_called_once_with(vendor_uuid, closers, last_modified_by) mock_publish.assert_not_called() def test_update_vendor_notes_success(monkeypatch): """Test successful update of vendor notes with sanitisation.""" vendor_uuid = str(uuid4()) vendor_id = 4567 raw_notes = "Hello" cleaned_notes = 'Hello' mock_update_vendor_notes = MagicMock( return_value={ 'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id, 'relationship_notes': cleaned_notes, } ) monkeypatch.setattr(vendor_model, 'update_vendor_relationship_notes', mock_update_vendor_notes) resp = vendor.update_vendor_notes(vendor_uuid, raw_notes) assert resp.message == { 'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id, 'relationship_notes': cleaned_notes, } mock_update_vendor_notes.assert_called_once_with(vendor_uuid, cleaned_notes) @patch('account.logic.vendor.g', spec=['log', 'request_context']) def test_update_vendor_notes_exception(mock_g, monkeypatch, app_context): """Test exception handling in update of vendor notes.""" vendor_uuid = str(uuid4()) raw_notes = "Hello" cleaned_notes = 'Hello' mock_update_vendor_notes = MagicMock(side_effect=Exception('Error')) monkeypatch.setattr(vendor_model, 'update_vendor_relationship_notes', mock_update_vendor_notes) result = vendor.update_vendor_notes(vendor_uuid, raw_notes) assert result.status == 500 assert result.errors['message'] == 'Unable to update vendor: Error' mock_update_vendor_notes.assert_called_once_with(vendor_uuid, cleaned_notes) def test_update_vendor_first_statement_period_success(monkeypatch): """Test successful update of vendor first statement period.""" vendor_uuid = str(uuid4()) vendor_id = 4567 first_statement_period = 1 mock_update_vendor_first_statement_period = MagicMock( return_value={ 'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id, 'first_statement_period': first_statement_period, } ) monkeypatch.setattr( vendor, 'validate_first_statement_period', MagicMock(return_value=first_statement_period) ) monkeypatch.setattr( vendor_model, 'update_vendor_first_statement_period', mock_update_vendor_first_statement_period, ) resp = vendor.update_vendor_first_statement_period(vendor_uuid, first_statement_period) assert resp.message == { 'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id, 'first_statement_period': first_statement_period, } mock_update_vendor_first_statement_period.assert_called_once_with( vendor_uuid, first_statement_period ) @patch('account.logic.vendor.g', spec=['log', 'request_context']) def test_update_vendor_first_statement_period_exception(mock_g, monkeypatch, app_context): """Test exception handling in update of vendor first statement period.""" vendor_uuid = str(uuid4()) first_statement_period = 1 expected_error_response = { 'code': 'internal_error', 'message': 'Unable to update vendor: Error', } mock_update_vendor_first_statement_period = MagicMock(side_effect=Exception('Error')) monkeypatch.setattr( vendor, 'validate_first_statement_period', MagicMock(return_value=first_statement_period) ) monkeypatch.setattr( vendor_model, 'update_vendor_first_statement_period', mock_update_vendor_first_statement_period, ) result = vendor.update_vendor_first_statement_period(vendor_uuid, first_statement_period) assert result.status == 500 assert result.errors == expected_error_response mock_update_vendor_first_statement_period.assert_called_once_with( vendor_uuid, first_statement_period ) @patch('account.models.ows_royalties.get_statement_period') @pytest.mark.parametrize( 'input, mock_period, expected_exception', [ pytest.param( {'first_statement_period': 1}, { 'statement_period_id': 1, 'statement_period_name': 'Test 2050', 'statement_period_status': 'open', 'statement_year': 1111, 'statement_month': 5, 'closed_date': None, 'exchange_rates_delivered': False, }, None, ), pytest.param( {'first_statement_period': 450}, None, Exception('No period found for period id : 450') ), pytest.param( {'first_statement_period': 11}, { 'statement_period_id': 11, 'statement_period_name': 'Test 2030', 'statement_period_status': 'closed', 'statement_year': 2222, 'statement_month': 5, 'closed_date': None, 'exchange_rates_delivered': True, }, Exception('First statement period must not be a future date. Received : 5/2222'), ), ], ) def test_validate_first_statement_period( mock_validate_vendor_first_statement_period, input, mock_period, expected_exception ): """Test validate_first_statement_period function.""" mock_validate_vendor_first_statement_period.return_value = mock_period if expected_exception: with pytest.raises(Exception) as exc_info: vendor.validate_first_statement_period(input['first_statement_period']) assert str(exc_info.value) == str(expected_exception) else: result = vendor.validate_first_statement_period(input['first_statement_period']) assert result == input['first_statement_period'] def test_update_vendor_external_identifier_1_success(monkeypatch): """external_identifier_1 update forwards details, returns typed dict, publishes event.""" vendor_uuid = str(uuid4()) vendor_id = 4567 last_modified_by = 123 external_identifier_1 = 'ext_id_123' mock_update_vendor = MagicMock( return_value={'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id} ) mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_by_uuid', mock_update_vendor) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) result = vendor.update_vendor_external_identifier_1( vendor_uuid, last_modified_by, external_identifier_1 ) assert result == {'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id} mock_update_vendor.assert_called_once_with( vendor_uuid, { 'external_identifier_1': external_identifier_1, 'last_modified_by': last_modified_by, }, ) mock_publish.assert_called_once_with( vendor_id=vendor_id, response_payload={ 'external_identifier_1': external_identifier_1, 'last_modified_by': last_modified_by, 'vendor_id': vendor_id, 'vendor_uuid': vendor_uuid, }, event_type='update', ) def test_update_vendor_external_identifier_1_not_found_vendor(monkeypatch): """VendorUpdateException from the model propagates and no event is published.""" from account.utils.exception import VendorUpdateException vendor_uuid = str(uuid4()) mock_update_vendor = MagicMock( side_effect=VendorUpdateException(code='invalid_input', message='nope', status=404) ) mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_by_uuid', mock_update_vendor) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) with pytest.raises(VendorUpdateException) as exc_info: vendor.update_vendor_external_identifier_1(vendor_uuid, 123, 'ext_id_123') assert exc_info.value.status == 404 mock_publish.assert_not_called() def test_update_vendor_country_id_success(monkeypatch): """country_id update forwards details, returns typed dict, publishes event.""" vendor_uuid = str(uuid4()) vendor_id = 4567 last_modified_by = 123 country_id = 42 mock_update_vendor = MagicMock( return_value={'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id} ) mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_by_uuid', mock_update_vendor) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) result = vendor.update_vendor_country_id(vendor_uuid, last_modified_by, country_id) assert result == {'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id} mock_update_vendor.assert_called_once_with( vendor_uuid, { 'country_id': country_id, 'last_modified_by': last_modified_by, }, ) mock_publish.assert_called_once_with( vendor_id=vendor_id, response_payload={ 'country_id': country_id, 'last_modified_by': last_modified_by, 'vendor_id': vendor_id, 'vendor_uuid': vendor_uuid, }, event_type='update', ) def test_update_vendor_country_id_not_found_vendor(monkeypatch): """VendorUpdateException from the model propagates and no event is published.""" from account.utils.exception import VendorUpdateException vendor_uuid = str(uuid4()) mock_update_vendor = MagicMock( side_effect=VendorUpdateException(code='invalid_input', message='nope', status=404) ) mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_by_uuid', mock_update_vendor) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) with pytest.raises(VendorUpdateException) as exc_info: vendor.update_vendor_country_id(vendor_uuid, 123, 42) assert exc_info.value.status == 404 mock_publish.assert_not_called() def test_delete_vendor_success(monkeypatch): """delete_vendor publishes update event when status transitioned to deletion.""" vendor_uuid = str(uuid4()) vendor_id = 4567 last_modified_by = 123 model_result = { 'vendor_id': vendor_id, 'vendor_uuid': vendor_uuid, 'status': 'deletion', } mock_delete = MagicMock(return_value=model_result) mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'delete_vendor_by_uuid', mock_delete) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) result = vendor.delete_vendor(vendor_uuid, last_modified_by) assert result == model_result mock_delete.assert_called_once_with(vendor_uuid, last_modified_by) mock_publish.assert_called_once_with( vendor_id=vendor_id, response_payload=model_result, event_type='update', ) def test_delete_vendor_already_deleted(monkeypatch): """delete_vendor does not publish when vendor was already in deletion status.""" vendor_uuid = str(uuid4()) model_result = { 'vendor_id': 4567, 'vendor_uuid': vendor_uuid, 'status': None, } mock_delete = MagicMock(return_value=model_result) mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'delete_vendor_by_uuid', mock_delete) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) result = vendor.delete_vendor(vendor_uuid, 123) assert result == model_result mock_publish.assert_not_called() def test_delete_vendor_not_found(monkeypatch): """delete_vendor does not publish when vendor not found.""" vendor_uuid = str(uuid4()) model_result = { 'vendor_id': None, 'vendor_uuid': vendor_uuid, 'status': None, } mock_delete = MagicMock(return_value=model_result) mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'delete_vendor_by_uuid', mock_delete) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) result = vendor.delete_vendor(vendor_uuid, 123) assert result == model_result mock_publish.assert_not_called() def test_update_vendor_info_writes_supplied_fields_and_publishes(monkeypatch): """update_vendor_info forwards only the supplied fields plus last_modified_by.""" vendor_uuid = str(uuid4()) vendor_id = 99 last_modified_by = 7 mock_update_vendor = MagicMock( return_value={'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id} ) mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_by_uuid', mock_update_vendor) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) vendor.update_vendor_info( vendor_uuid, last_modified_by, name='New Name', support_contact_email=None, newsletter='N', ) mock_update_vendor.assert_called_once_with( vendor_uuid, { 'name': 'New Name', 'support_contact_email': None, 'newsletter': 'N', 'last_modified_by': last_modified_by, }, ) mock_publish.assert_called_once_with( vendor_id=vendor_id, response_payload={ 'name': 'New Name', 'support_contact_email': None, 'newsletter': 'N', 'last_modified_by': last_modified_by, 'vendor_id': vendor_id, 'vendor_uuid': vendor_uuid, }, event_type='update', ) def test_update_vendor_service_tier_writes_mysql_and_publishes(monkeypatch): """update_vendor_service_tier writes art_relations and publishes UPDATE event. Neo4j is updated downstream by the existing sync — not from this code path. """ vendor_uuid = str(uuid4()) vendor_id = 42 last_modified_by = 7 service_tier_uuid = '5f2bd4fc-df94-4f35-97d3-ef23f8573279' mock_model_update = MagicMock(return_value={'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id}) mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_service_tier_by_uuid', mock_model_update) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) result = vendor.update_vendor_service_tier(vendor_uuid, last_modified_by, service_tier_uuid) assert result == {'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id} mock_model_update.assert_called_once_with(vendor_uuid, service_tier_uuid, last_modified_by) mock_publish.assert_called_once_with( vendor_id=vendor_id, response_payload={ 'service_tier_uuid': service_tier_uuid, 'last_modified_by': last_modified_by, 'vendor_id': vendor_id, 'vendor_uuid': vendor_uuid, }, event_type='update', ) def test_update_vendor_internal_staff_success(monkeypatch): """All staff fields provided are forwarded plus last_modified_by and event published.""" vendor_uuid = str(uuid4()) vendor_id = 1 last_modified_by = 123 mock_update_vendor = MagicMock( return_value={'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id} ) mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_by_uuid', mock_update_vendor) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) result = vendor.update_vendor_internal_staff( vendor_uuid, last_modified_by, assigned_to=10, assigned_reviewer=20, quarterback_label_manager=30, wel_email_sender=40, ) assert result == {'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id} mock_update_vendor.assert_called_once_with( vendor_uuid, { 'assigned_to': 10, 'assigned_reviewer': 20, 'quarterback_label_manager': 30, 'wel_email_sender': 40, 'last_modified_by': last_modified_by, }, ) mock_publish.assert_called_once_with( vendor_id=vendor_id, response_payload={ 'assigned_to': 10, 'assigned_reviewer': 20, 'quarterback_label_manager': 30, 'wel_email_sender': 40, 'last_modified_by': last_modified_by, 'vendor_id': vendor_id, 'vendor_uuid': vendor_uuid, }, event_type='update', ) def test_update_vendor_internal_staff_partial(monkeypatch): """None fields are dropped from the update payload and event payload.""" vendor_uuid = str(uuid4()) vendor_id = 1 last_modified_by = 123 mock_update_vendor = MagicMock( return_value={'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id} ) mock_publish = MagicMock() mock_update_pm = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_by_uuid', mock_update_vendor) monkeypatch.setattr(vendor_model, 'update_product_manager_id', mock_update_pm) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) vendor.update_vendor_internal_staff( vendor_uuid, last_modified_by, assigned_to=10, ) mock_update_vendor.assert_called_once_with( vendor_uuid, { 'assigned_to': 10, 'last_modified_by': last_modified_by, }, ) mock_update_pm.assert_not_called() mock_publish.assert_called_once_with( vendor_id=vendor_id, response_payload={ 'assigned_to': 10, 'last_modified_by': last_modified_by, 'vendor_id': vendor_id, 'vendor_uuid': vendor_uuid, }, event_type='update', ) def test_update_vendor_internal_staff_with_product_manager(monkeypatch): """product_manager triggers the mapping upsert and lands in the event payload.""" vendor_uuid = str(uuid4()) vendor_id = 1 last_modified_by = 123 pm_id = 999 mock_update_vendor = MagicMock( return_value={'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id} ) mock_update_pm = MagicMock() mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_by_uuid', mock_update_vendor) monkeypatch.setattr(vendor_model, 'update_product_manager_id', mock_update_pm) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) vendor.update_vendor_internal_staff( vendor_uuid, last_modified_by, assigned_to=10, product_manager=pm_id, ) mock_update_vendor.assert_called_once_with( vendor_uuid, {'assigned_to': 10, 'last_modified_by': last_modified_by}, ) mock_update_pm.assert_called_once_with(vendor_id, pm_id) mock_publish.assert_called_once_with( vendor_id=vendor_id, response_payload={ 'assigned_to': 10, 'last_modified_by': last_modified_by, 'vendor_id': vendor_id, 'vendor_uuid': vendor_uuid, 'product_manager': pm_id, }, event_type='update', ) def test_update_vendor_internal_staff_explicit_null_clears_column(monkeypatch): """Explicit None for a nullable FK is forwarded to the model write.""" vendor_uuid = str(uuid4()) vendor_id = 1 last_modified_by = 123 mock_update_vendor = MagicMock( return_value={'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id} ) mock_update_pm = MagicMock() mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_by_uuid', mock_update_vendor) monkeypatch.setattr(vendor_model, 'update_product_manager_id', mock_update_pm) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) vendor.update_vendor_internal_staff( vendor_uuid, last_modified_by, quarterback_label_manager=None, ) mock_update_vendor.assert_called_once_with( vendor_uuid, {'quarterback_label_manager': None, 'last_modified_by': last_modified_by}, ) mock_update_pm.assert_not_called() mock_publish.assert_called_once_with( vendor_id=vendor_id, response_payload={ 'quarterback_label_manager': None, 'last_modified_by': last_modified_by, 'vendor_id': vendor_id, 'vendor_uuid': vendor_uuid, }, event_type='update', ) def test_update_vendor_internal_staff_explicit_null_product_manager_skips_upsert(monkeypatch): """product_manager=None slipping past the schema skips both upsert and event field.""" vendor_uuid = str(uuid4()) vendor_id = 1 last_modified_by = 123 mock_update_vendor = MagicMock( return_value={'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id} ) mock_update_pm = MagicMock() mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_by_uuid', mock_update_vendor) monkeypatch.setattr(vendor_model, 'update_product_manager_id', mock_update_pm) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) vendor.update_vendor_internal_staff( vendor_uuid, last_modified_by, assigned_to=10, product_manager=None, ) mock_update_pm.assert_not_called() event_payload = mock_publish.call_args.kwargs['response_payload'] assert 'product_manager' not in event_payload @pytest.mark.parametrize( 'staff_kwargs,expected_extra_payload', [ pytest.param( {'assigned_to': None}, {'assigned_to': None}, id='single-null-clears-one-field', ), pytest.param( { 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'wel_email_sender': None, }, { 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'wel_email_sender': None, }, id='all-nullables-cleared-at-once', ), pytest.param( {'assigned_to': 10, 'quarterback_label_manager': None}, {'assigned_to': 10, 'quarterback_label_manager': None}, id='mixed-set-and-clear', ), pytest.param( {'assigned_to': 10, 'product_manager': None}, {'assigned_to': 10}, id='product-manager-null-not-in-payload', ), pytest.param( {'assigned_to': None, 'product_manager': 99}, {'assigned_to': None, 'product_manager': 99}, id='clear-nullable-and-set-product-manager', ), ], ) def test_update_vendor_internal_staff_publish_event_payload( monkeypatch, staff_kwargs, expected_extra_payload ): """publish_account_event receives exactly the keys the caller sent (minus PM=None).""" vendor_uuid = str(uuid4()) vendor_id = 1 last_modified_by = 123 mock_update_vendor = MagicMock( return_value={'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id} ) mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_by_uuid', mock_update_vendor) monkeypatch.setattr(vendor_model, 'update_product_manager_id', MagicMock()) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) vendor.update_vendor_internal_staff(vendor_uuid, last_modified_by, **staff_kwargs) mock_publish.assert_called_once_with( vendor_id=vendor_id, response_payload={ **expected_extra_payload, 'last_modified_by': last_modified_by, 'vendor_id': vendor_id, 'vendor_uuid': vendor_uuid, }, event_type='update', ) def test_update_vendor_internal_staff_unknown_kwarg_raises(monkeypatch): """Unknown staff field kwargs raise so typos don't silently no-op.""" mock_update_vendor = MagicMock() mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_by_uuid', mock_update_vendor) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) with pytest.raises(ValueError, match='Unknown staff field'): vendor.update_vendor_internal_staff( str(uuid4()), 123, assigned_to=10, typo_field=999, ) mock_update_vendor.assert_not_called() mock_publish.assert_not_called() def test_update_vendor_internal_staff_not_found_does_not_publish(monkeypatch): """VendorUpdateException from the model propagates and no event is published.""" from account.utils.exception import VendorUpdateException vendor_uuid = str(uuid4()) mock_update_vendor = MagicMock( side_effect=VendorUpdateException(code='invalid_input', message='nope', status=404) ) mock_publish = MagicMock() monkeypatch.setattr(vendor_model, 'update_vendor_by_uuid', mock_update_vendor) monkeypatch.setattr(vendor, 'publish_account_event', mock_publish) with pytest.raises(VendorUpdateException) as exc_info: vendor.update_vendor_internal_staff( vendor_uuid, 123, assigned_to=10, ) assert exc_info.value.status == 404 mock_publish.assert_not_called() @patch('account.logic.vendor.g', spec=['log', 'request_context']) @pytest.mark.parametrize( 'input, ar_result, neo4j_get_vendor', [ # No vendor_id passed, so it will always create new. ( { 'vendor_name': 'Foo Productions', 'email': 'abc@xyz.com', 'owner': 'SME US Latin', 'label_identifier': 'Frontline', }, { 'create': 'successful', 'vendor_id': 1234, 'country': None, 'source': None, }, response.create_not_found_response(), ), ( { 'vendor_name': 'Foo Productions', 'email': 'abc@xyz.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', }, { 'create': 'successful', 'vendor_id': 1234, 'country': None, 'source': None, }, response.create_not_found_response(), ), ( { 'vendor_name': 'Foo Productions', 'email': 'abc@xyz.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', }, { 'create': 'successful', 'vendor_id': 1234, 'country': None, 'source': None, 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', }, response.create_not_found_response(), ), ( { 'vendor_id': 1234, 'vendor_name': 'Foo Productions', 'email': 'abc@xyz.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', }, { 'get': 'successful', 'vendor_id': 1234, 'country': None, 'source': None, }, response.Response({'foo': 'bar'}), ), ( { 'vendor_id': 1234, 'vendor_name': 'Foo Productions', 'email': 'abc@xyz.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', }, { 'get': 'successful', 'vendor_id': 1234, 'country': None, 'source': None, 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', }, response.Response({'foo': 'bar'}), ), ], ) def test_create_or_update_vendor( mock_g, input, ar_result, neo4j_get_vendor, monkeypatch, app_context ): """Test create_or_update_vendor.""" expected = { 'name': input['vendor_name'], 'company': input['vendor_name'], 'contact_email': input['email'], 'owner': input['owner'], 'label_identifier': input['label_identifier'], 'status': None, 'migrated_to_abacus': False, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'external_identifier_1': None, 'is_distributor': None, } monkeypatch.setattr( vendor_model, 'create_vendor_with_optional_service_tier', MagicMock(return_value=response.Response({'create': 'successful', 'vendor_id': 1234})), ) monkeypatch.setattr( vendor_model, 'update_vendor', MagicMock(return_value=response.Response('successful update!!')), ) monkeypatch.setattr( vendor_model, 'get_vendor', MagicMock(return_value=response.Response({'get': 'successful', 'vendor_id': 1234})), ) monkeypatch.setattr(neo4j_vendor, 'get_vendor', MagicMock(return_value=neo4j_get_vendor)) monkeypatch.setattr( neo4j_vendor, 'create_or_update_vendor', MagicMock(return_value=response.Response('success!!')), ) monkeypatch.setattr(neo4j_vendor, 'add_company_brand', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'add_service_tier', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'update_company_brand', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'update_service_tier', MagicMock()) monkeypatch.setattr(vendor, 'publish_account_event', MagicMock()) monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response('enabled')), ) result = vendor.create_or_update_vendor(input) assert result assert result.message == ar_result assert vendor.publish_account_event.called if input.get('vendor_id'): vendor_model.update_vendor.assert_called_once_with(input.get('vendor_id'), expected) assert vendor_model.get_vendor.call_count == 1 assert vendor_model.update_vendor.call_count == 1 assert vendor_model.create_vendor_with_optional_service_tier.call_count == 0 if input.get('company_brand'): assert neo4j_vendor.update_company_brand.call_count == 1 else: assert neo4j_vendor.update_company_brand.call_count == 0 if input.get('service_tier_uuid'): assert neo4j_vendor.update_service_tier.call_count == 1 else: assert neo4j_vendor.update_service_tier.call_count == 0 else: vendor_model.create_vendor_with_optional_service_tier.assert_called_once_with( expected, service_tier_uuid=input.get('service_tier_uuid') ) assert vendor_model.update_vendor.call_count == 0 assert vendor_model.create_vendor_with_optional_service_tier.call_count == 1 if input.get('company_brand'): assert neo4j_vendor.add_company_brand.call_count == 1 else: assert neo4j_vendor.add_company_brand.call_count == 0 if input.get('service_tier_uuid'): assert neo4j_vendor.add_service_tier.call_count == 1 else: assert neo4j_vendor.add_service_tier.call_count == 0 @patch('account.logic.vendor.g', spec=['log', 'request_context']) def test_update_company_brand_art_relations_for_create_vendor_via_SF(mock_g, monkeypatch): """Test create vendor for gda_account""" data = { 'vendor_name': 'Foo Productions', 'email': 'abc@xyz.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', 'company_brand': 'awal', } monkeypatch.setattr( vendor_model, 'create_vendor_with_optional_service_tier', MagicMock(return_value=response.Response({'create': 'successful', 'vendor_id': 1234})), ) monkeypatch.setattr( vendor_model, 'get_vendor', MagicMock(return_value=response.Response({'get': 'successful', 'vendor_id': 1234})), ) monkeypatch.setattr( neo4j_vendor, 'get_vendor', MagicMock(return_value=response.Response({'foo': 'bar'})), ) monkeypatch.setattr( neo4j_vendor, 'create_or_update_vendor', MagicMock(return_value=response.Response('success!!')), ) monkeypatch.setattr(neo4j_vendor, 'add_company_brand', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'add_service_tier', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'update_company_brand', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'update_service_tier', MagicMock()) monkeypatch.setattr(vendor, 'publish_account_event', MagicMock()) monkeypatch.setattr( vendor_model, 'get_company_brand_id_by_name', MagicMock(return_value=response.Response({'company_brand_id': 1})), ) result = vendor.create_or_update_vendor(data) assert result # get company brand id from art_relations assert vendor_model.get_company_brand_id_by_name.called vendor_model.create_vendor_with_optional_service_tier.assert_called_once_with( { 'name': 'Foo Productions', 'company': 'Foo Productions', 'contact_email': 'abc@xyz.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', 'status': None, 'migrated_to_abacus': False, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'external_identifier_1': None, 'is_distributor': None, 'company_brand_id': 1, # set company brand id from art_relations }, service_tier_uuid=None, ) @patch('account.logic.vendor.g', spec=['log', 'request_context']) def test_update_company_brand_art_relations_for_update_vendor_via_SF(mock_g, monkeypatch): """Test update vendor for gda_account""" data = { 'vendor_id': 1234, 'vendor_name': 'Foo Productions', 'email': 'abc@xyz.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', 'company_brand': 'awal', } monkeypatch.setattr( vendor_model, 'update_vendor', MagicMock(return_value=response.Response('successful update!!')), ) monkeypatch.setattr( vendor_model, 'get_vendor', MagicMock(return_value=response.Response({'get': 'successful', 'vendor_id': 1234})), ) monkeypatch.setattr( neo4j_vendor, 'get_vendor', MagicMock(return_value=response.Response({'foo': 'bar'})), ) monkeypatch.setattr( neo4j_vendor, 'create_or_update_vendor', MagicMock(return_value=response.Response('success!!')), ) monkeypatch.setattr(neo4j_vendor, 'add_company_brand', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'add_service_tier', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'update_company_brand', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'update_service_tier', MagicMock()) monkeypatch.setattr(vendor, 'publish_account_event', MagicMock()) monkeypatch.setattr( vendor_model, 'get_company_brand_id_by_name', MagicMock(return_value=response.Response({'company_brand_id': 1})), ) result = vendor.create_or_update_vendor(data) assert result # get company brand id from art_relations assert vendor_model.get_company_brand_id_by_name.called vendor_model.update_vendor.assert_called_once_with( 1234, { 'name': 'Foo Productions', 'company': 'Foo Productions', 'contact_email': 'abc@xyz.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', 'status': None, 'migrated_to_abacus': False, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'external_identifier_1': None, 'is_distributor': None, 'company_brand_id': 1, # set company brand id from art_relations }, ) @patch('account.logic.vendor.g', spec=['log', 'request_context']) def test_create_or_update_vendor_not_found(mock_g, monkeypatch, app_context): """Test create_or_update update not found.""" input = { 'vendor_id': 1234, 'vendor_name': 'Foo Productions', 'email': 'abc@xyz.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', 'company_brand': 'awal', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', } monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response(message='disabled')), ) monkeypatch.setattr( vendor_model, 'get_company_brand_id_by_name', MagicMock(return_value=response.Response({'company_brand_id': 1})), ) monkeypatch.setattr( vendor_model, 'update_vendor', MagicMock( return_value=response.create_not_found_response( 'Vendor update unsuccessful: No vendor found for vendor_id 1234.' ) ), ) result = vendor.create_or_update_vendor(input) assert result.status == 404 assert ( result.errors.get('message') == 'Vendor update unsuccessful: No vendor found for vendor_id 1234.' ) # noqa @patch('account.logic.vendor.g') @patch('account.models.neo4j_vendor.create_or_update_vendor', side_effect=Exception()) def test_create_exception_handling(_, g_mock, app_context): """Test _create_neo4j_vendor exception handling.""" data = { 'vendor_name': 'Foo Productions', 'email': 'abc@xyz.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', 'company_brand': 'awal', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', } ar_result = {'foo': 'bar'} result = vendor._create_neo4j_vendor(data, ar_result) assert result.status == 500 assert result.errors.get('message') == 'Could not create new vendor in neo4j.' g_mock.ows.log.error.assert_called_with('Could not create new vendor in neo4j.') @patch('account.models.neo4j_vendor.create_or_update_vendor', side_effect=Exception()) def test_update_exception_handling(mock_neo4j_vendor): """Test _update_neo4j_vendor exception handling.""" data = { 'vendor_id': 1234, 'vendor_name': 'Foo Productions', 'email': 'abc@xyz.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', 'company_brand': 'awal', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', } ar_result = {'foo': 'bar', 'vendor_id': 1234} result = vendor._update_neo4j_vendor(data, ar_result) assert result.status == 500 assert result.errors.get('message') == 'Could not update vendor 1234 in neo4j.' @patch('account.logic.vendor.g', spec=['log', 'request_context']) @pytest.mark.parametrize( 'input', [ ( { # iso alpha 3 'vendor_name': 'Wong Foo Productions', 'email': 'abc@xyz.com', 'owner': 'SME US Latin', 'label_identifier': 'Frontline', 'country': 'USA', } ), ( { # no country sent 'vendor_name': 'Wong Foo Productions', 'email': 'abc@xyz.com', 'owner': 'SME US Latin', 'label_identifier': 'Frontline', } ), ( { # invalid country code 'vendor_name': 'Wong Foo Productions', 'email': 'abc@xyz.com', 'owner': 'SME US Latin', 'label_identifier': 'Frontline', 'country': 'USB', } ), ], ) def test_create_vendor_country_validation(mock_g, input, monkeypatch, app_context): """Test country code validation when creating a vendor.""" ar_result_obj = { 'vendor_id': 1, 'is_distributor': 'N', 'owner': 'SME US Latin', 'name': 'Wong Foo Productions', 'company': 'Wong Foo Productions', 'country_id': None, 'status': 'signed', 'label_identifier': 'Frontline', 'contact_email': 'abc@xyz.com', 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', } ar_response = response.Response(ar_result_obj) monkeypatch.setattr( neo4j_vendor, 'get_vendor', MagicMock(return_value=response.create_not_found_response()), ) monkeypatch.setattr( vendor_model, 'create_vendor_with_optional_service_tier', MagicMock(return_value=ar_response), ) monkeypatch.setattr( neo4j_vendor, 'create_or_update_vendor', MagicMock(return_value=response.Response('success!!')), ) monkeypatch.setattr(vendor, 'validate_country_code', MagicMock()) monkeypatch.setattr(vendor, 'publish_account_event', MagicMock()) monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response('enabled')), ) result = vendor.create_or_update_vendor(input) if 'country' in input: ar_result_obj['country'] = input['country'] assert vendor.validate_country_code.call_count == 1 else: ar_result_obj['country'] = None if result: assert result assert result.message == ar_result_obj else: assert result.status == 422 assert result.errors.message == 'Country code not recognized: USB' assert vendor.publish_account_event.called @patch('account.logic.vendor.g', spec=['log', 'request_context']) def test_create_vendor_with_closer_id(mock_g, monkeypatch, app_context): """Test creating a vendor with closer_id sets the closer correctly.""" input_data = { 'vendor_name': 'Test Label', 'email': 'test@example.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', 'closer_id': 2912, } ar_result_obj = { 'vendor_id': 1234, 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', } ar_response = response.Response(ar_result_obj) monkeypatch.setattr( vendor_model, 'create_vendor_with_optional_service_tier', MagicMock(return_value=ar_response), ) mock_update_closers = MagicMock(return_value=response.Response({'success': True})) monkeypatch.setattr(vendor, 'update_vendor_closers', mock_update_closers) monkeypatch.setattr( neo4j_vendor, 'get_vendor', MagicMock(return_value=response.create_not_found_response()), ) monkeypatch.setattr( neo4j_vendor, 'create_or_update_vendor', MagicMock(return_value=response.Response('success!!')), ) monkeypatch.setattr(neo4j_vendor, 'add_service_tier', MagicMock()) monkeypatch.setattr(vendor, 'publish_account_event', MagicMock()) result = vendor.create_or_update_vendor(input_data) assert result assert result.message['vendor_id'] == 1234 mock_update_closers.assert_called_once_with( 'd6455851-3f86-48d7-b102-8d006eb92655', [2912], ) @patch('account.logic.vendor.g', spec=['log', 'request_context']) def test_create_vendor_without_closer_id(mock_g, monkeypatch, app_context): """Test creating a vendor without closer_id does not call update_vendor_closers.""" input_data = { 'vendor_name': 'Test Label', 'email': 'test@example.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', } ar_result_obj = { 'vendor_id': 1234, 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', } ar_response = response.Response(ar_result_obj) monkeypatch.setattr( vendor_model, 'create_vendor_with_optional_service_tier', MagicMock(return_value=ar_response), ) mock_update_closers = MagicMock(return_value=response.Response({'success': True})) monkeypatch.setattr(vendor, 'update_vendor_closers', mock_update_closers) monkeypatch.setattr( neo4j_vendor, 'get_vendor', MagicMock(return_value=response.create_not_found_response()), ) monkeypatch.setattr( neo4j_vendor, 'create_or_update_vendor', MagicMock(return_value=response.Response('success!!')), ) monkeypatch.setattr(neo4j_vendor, 'add_service_tier', MagicMock()) monkeypatch.setattr(vendor, 'publish_account_event', MagicMock()) result = vendor.create_or_update_vendor(input_data) assert result assert result.message['vendor_id'] == 1234 @patch('account.logic.vendor.g', spec=['log', 'request_context']) def test_update_vendor_with_closer_id_ignored(mock_g, monkeypatch, app_context): """Test updating a vendor with closer_id does not call update_vendor_closers.""" input_data = { 'vendor_id': 1234, # This makes it an update 'vendor_name': 'Test Label', 'email': 'test@example.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', 'closer_id': 2912, # Should be ignored for updates } ar_result_obj = { 'vendor_id': 1234, 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', } monkeypatch.setattr( vendor_model, 'update_vendor', MagicMock(return_value=response.Response('successful update!!')), ) monkeypatch.setattr( vendor_model, 'get_vendor', MagicMock(return_value=response.Response(ar_result_obj)), ) mock_update_closers = MagicMock(return_value=response.Response({'success': True})) monkeypatch.setattr(vendor, 'update_vendor_closers', mock_update_closers) monkeypatch.setattr( neo4j_vendor, 'get_vendor', MagicMock(return_value=response.Response({'foo': 'bar'})), ) monkeypatch.setattr( neo4j_vendor, 'create_or_update_vendor', MagicMock(return_value=response.Response('success!!')), ) monkeypatch.setattr(neo4j_vendor, 'update_service_tier', MagicMock()) monkeypatch.setattr(vendor, 'publish_account_event', MagicMock()) result = vendor.create_or_update_vendor(input_data) assert result @patch('account.models.vendor.create_vendor_with_optional_service_tier') @patch('account.models.feature.bulk_add_restricted_features_for_vendor') @patch('account.models.vendor.get_company_brand_id_by_name') def test_v2_create_vendor_brand_not_found( get_brand_id_mock, bulk_add_restricted_features_mock, create_vendor_mock, minimum_vendor_attrs ): get_brand_id_mock.return_value = response.create_not_found_response(message=':(') result = vendor.create_vendor(minimum_vendor_attrs) assert result.status == 404 create_vendor_mock.assert_not_called() bulk_add_restricted_features_mock.assert_not_called() @patch('account.models.feature.bulk_add_restricted_features_for_vendor') @patch('account.models.vendor.get_company_brand_id_by_name') def test_v2_create_vendor_sql_params( get_brand_id_mock, bulk_add_restricted_features_mock, monkeypatch, minimum_vendor_attrs, ): get_brand_id_mock.return_value = response.Response(message={'company_brand_id': 2}) bulk_add_restricted_features_mock.return_value = response.Response( message={'vendor_id': 1234, 'feature_id': 1} ) monkeypatch.setattr( vendor_model, 'create_vendor_with_optional_service_tier', MagicMock(return_value=response.Response(status=400)), ) vendor.create_vendor({**minimum_vendor_attrs, 'user_id': 111}) get_brand_id_mock.assert_called_with(company_brand_name=minimum_vendor_attrs['company_brand']) vendor_model.create_vendor_with_optional_service_tier.assert_called_with( details={ 'name': minimum_vendor_attrs['name'], 'owner': minimum_vendor_attrs['owner'], 'company_brand_id': 2, 'migrated_to_abacus': True, 'last_modified_by': 111, 'is_distributor': 'N', 'label_identifier': None, 'country_id': None, 'primary_genre': None, 'label_summary': None, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'wel_email_sender': None, }, service_tier_uuid=minimum_vendor_attrs['service_tier_uuid'], ) @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.models.feature.bulk_add_restricted_features_for_vendor') @patch('account.logic.vendor.publish_account_event') @pytest.mark.parametrize( 'brand_name, brand_id, migrated_to_abacus', [ ('altafonte', 10, True), ('awal', 5, True), ('knr', 1, True), ('theorchard', 2, True), ('sme', 4, True), ('hrs', 3, True), ('drm', 8, True), ('ab', 7, True), ('msk', 6, True), ('ma', 9, True), ], ) def test_v2_create_vendor_migrated_to_abacus_logic( _, bulk_add_restricted_features_mock, mock_vendor_model, mock_update_closers, brand_name, brand_id, migrated_to_abacus, minimum_vendor_attrs, ): """Test create vendor, for awal and knr brands should set migrated to abacus.""" mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( message={'company_brand_id': brand_id} ) bulk_add_restricted_features_mock.return_value = response.Response() mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( { 'create': 'successful', 'vendor_id': 1234, 'vendor_uuid': 'test-vendor-uuid', } ) mock_update_closers.return_value = response.Response({'success': True}) vendor.create_vendor({**minimum_vendor_attrs, 'company_brand': brand_name, 'user_id': 111}) if brand_name == 'awal': mock_update_closers.assert_called_once_with('test-vendor-uuid', [2912], 111) else: mock_update_closers.assert_not_called() mock_vendor_model.get_company_brand_id_by_name.assert_called_with(company_brand_name=brand_name) mock_vendor_model.create_vendor_with_optional_service_tier.assert_called_with( details={ 'name': minimum_vendor_attrs['name'], 'owner': minimum_vendor_attrs['owner'], 'company_brand_id': brand_id, 'migrated_to_abacus': migrated_to_abacus, 'last_modified_by': 111, 'is_distributor': 'N', 'label_identifier': None, 'country_id': None, 'primary_genre': None, 'label_summary': None, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'wel_email_sender': None, }, service_tier_uuid=minimum_vendor_attrs['service_tier_uuid'], ) @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.models.feature.bulk_add_restricted_features_for_vendor') @patch('account.logic.vendor.publish_account_event') @pytest.mark.parametrize( ('brand_name', 'brand_id', 'migrated_to_abacus', 'restricted_features'), [ pytest.param('ab', 7, True, ORCHARD_BRANDS, id='ab'), pytest.param('altafonte', 10, True, ALTAFONTE, id='altafonte'), pytest.param('awal', 5, True, AWAL, id='awal'), pytest.param('drm', 8, True, ORCHARD_BRANDS, id='drm'), pytest.param('hrs', 3, True, ORCHARD_BRANDS, id='hrs'), pytest.param('knr', 1, True, ALL_FEATURES, id='knr'), pytest.param('ma', 9, True, ORCHARD_BRANDS, id='ma'), pytest.param('msk', 6, True, ORCHARD_BRANDS, id='msk'), pytest.param('sme', 4, True, SME_BRANDS, id='sme'), pytest.param('theorchard', 2, True, ORCHARD_BRANDS, id='theorchard'), pytest.param('foundation', 11, True, FOUNDATION, id='foundation'), ], ) def test_v2_create_vendor_feature_control_logic( _, bulk_add_restricted_features_mock, mock_vendor_model, mock_update_closers, brand_name, brand_id, migrated_to_abacus, restricted_features, minimum_vendor_attrs, ): """Test create vendor, for awal and knr brands should persist correct feature restrictions.""" mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( message={'company_brand_id': brand_id} ) bulk_add_restricted_features_mock.return_value = response.Response( message={'vendor_id': 1234, 'feature_id': 1} ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( { 'create': 'successful', 'vendor_id': 1234, 'vendor_uuid': 'test-vendor-uuid', } ) mock_update_closers.return_value = response.Response({'success': True}) vendor.create_vendor({**minimum_vendor_attrs, 'company_brand': brand_name, 'user_id': 111}) if brand_name == 'awal': mock_update_closers.assert_called_once_with('test-vendor-uuid', [2912], 111) else: mock_update_closers.assert_not_called() mock_vendor_model.get_company_brand_id_by_name.assert_called_with(company_brand_name=brand_name) mock_vendor_model.create_vendor_with_optional_service_tier.assert_called_with( details={ 'name': minimum_vendor_attrs['name'], 'owner': minimum_vendor_attrs['owner'], 'company_brand_id': brand_id, 'migrated_to_abacus': migrated_to_abacus, 'last_modified_by': 111, 'is_distributor': 'N', 'label_identifier': None, 'country_id': None, 'primary_genre': None, 'label_summary': None, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'wel_email_sender': None, }, service_tier_uuid=minimum_vendor_attrs['service_tier_uuid'], ) bulk_add_restricted_features_mock.assert_called_with( vendor_id=1234, feature_ids=restricted_features ) @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.models.feature.bulk_add_restricted_features_for_vendor') @patch('account.logic.vendor.publish_account_event') @pytest.mark.parametrize( ('owner', 'brand_name', 'brand_id'), [ pytest.param('3rd Party Payee Acco', 'awal', 5, id='3rd party payee with awal brand'), pytest.param( '3rd Party Payee Acco', 'theorchard', 2, id='3rd party payee with theorchard brand' ), ], ) def test_v2_create_vendor_3rd_party_payee_feature_control( _, bulk_add_restricted_features_mock, mock_vendor_model, mock_update_closers, owner, brand_name, brand_id, minimum_vendor_attrs, ): """Test that 3rd Party Payee accounts get only accounting features.""" mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( message={'company_brand_id': brand_id} ) bulk_add_restricted_features_mock.return_value = response.Response( message={'vendor_id': 1234, 'feature_id': 1} ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( { 'vendor_id': 1234, 'vendor_uuid': 'test-vendor-uuid', } ) mock_update_closers.return_value = response.Response() vendor.create_vendor( {**minimum_vendor_attrs, 'owner': owner, 'company_brand': brand_name, 'user_id': 111} ) bulk_add_restricted_features_mock.assert_called_with( vendor_id=1234, feature_ids=THIRD_PARTY_PAYEE ) @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.logic.vendor.publish_account_event') @patch('account.models.vendor.get_company_brand_id_by_name') def test_v2_create_vendor_sql_error( _, publish_mock, mock_vendor_model, mock_update_closers, minimum_vendor_attrs, ): mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( status=400 ) mock_update_closers.return_value = response.Response({'success': True}) result = vendor.create_vendor(minimum_vendor_attrs) assert result.status == 400 publish_mock.assert_not_called() mock_update_closers.assert_not_called() @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.logic.vendor.publish_account_event') @patch('account.models.feature.bulk_add_restricted_features_for_vendor') @patch('account.models.ows_users.get_user_profiles_by_identity_id') def test_v2_create_vendor_no_errors( get_user_profiles_by_identity_id_mock, _, publish_mock, mock_vendor_model, mock_update_closers, minimum_vendor_attrs, ): expected_attrs = { **minimum_vendor_attrs, 'company_brand': 'theorchard', 'vendor_id': 9, 'last_modified_by': 111, } get_user_profiles_by_identity_id_mock.return_value = response.Response( message={'items': [{'profile_type': 'OrchAdminProfile', 'profile_id': 111}]} ) response_obj = response.Response(message=expected_attrs) mock_update_closers.return_value = response.Response({'success': True}) mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( {'company_brand_id': 2} ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response_obj result = vendor.create_vendor({**minimum_vendor_attrs, 'company_brand': 'theorchard'}) mock_update_closers.assert_not_called() assert result == response_obj publish_mock.assert_called_with( vendor_id=9, # Includes payment_currency field response_payload=expected_attrs, event_type='create', ) @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.logic.vendor.publish_account_event') @patch('account.logic.vendor.feature_model') def test_v2_create_vendor_with_is_distributor_is_true( mock_feature_model: MagicMock, publish_mock: MagicMock, mock_vendor_model: MagicMock, mock_update_closers: MagicMock, minimum_vendor_attrs: dict[str, str], ) -> None: """Assert is_distributor can be set with 'Y' when creating the vendor.""" extended_attrs = { **minimum_vendor_attrs, 'company_brand': 'theorchard', 'is_distributor': True, 'user_id': 111, } mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( { 'company_brand_id': 2, } ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( {'create': 'successful', 'vendor_id': 1234, 'vendor_uuid': 'test-uuid'} ) mock_update_closers.return_value = response.Response({'success': True}) vendor.create_vendor(extended_attrs) mock_update_closers.assert_not_called() mock_vendor_model.create_vendor_with_optional_service_tier.assert_called_once_with( details={ 'name': minimum_vendor_attrs['name'], 'owner': minimum_vendor_attrs['owner'], 'company_brand_id': 2, 'migrated_to_abacus': True, 'last_modified_by': 111, 'is_distributor': 'Y', 'label_identifier': 'D3', 'country_id': None, 'primary_genre': None, 'label_summary': None, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'wel_email_sender': None, }, service_tier_uuid=minimum_vendor_attrs['service_tier_uuid'], ) mock_feature_model.bulk_add_restricted_features_for_vendor.assert_called_once() publish_mock.assert_called_once() @pytest.mark.parametrize( 'is_distributor_value', [ (False), ('Y'), ('N'), ('T'), ('F'), ('true'), ('false'), ], ) @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.logic.vendor.publish_account_event') @patch('account.logic.vendor.feature_model') def test_v2_create_vendor_with_is_distributor_is_false( mock_feature_model: MagicMock, publish_mock: MagicMock, mock_vendor_model: MagicMock, mock_update_closers: MagicMock, is_distributor_value: str, minimum_vendor_attrs: dict[str, str], ) -> None: """Assert various is_distributor values that will translate to 'N' when creating the vendor.""" extended_attrs = { **minimum_vendor_attrs, 'company_brand': 'theorchard', 'is_distributor': is_distributor_value, 'user_id': 111, } mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( { 'company_brand_id': 2, } ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( {'create': 'successful', 'vendor_id': 1234, 'vendor_uuid': 'test-uuid'} ) mock_update_closers.return_value = response.Response({'success': True}) vendor.create_vendor(extended_attrs) mock_update_closers.assert_not_called() mock_vendor_model.create_vendor_with_optional_service_tier.assert_called_once_with( details={ 'name': minimum_vendor_attrs['name'], 'owner': minimum_vendor_attrs['owner'], 'company_brand_id': 2, 'migrated_to_abacus': True, 'last_modified_by': 111, 'is_distributor': 'N', 'label_identifier': None, 'country_id': None, 'primary_genre': None, 'label_summary': None, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'wel_email_sender': None, }, service_tier_uuid=minimum_vendor_attrs['service_tier_uuid'], ) mock_feature_model.bulk_add_restricted_features_for_vendor.assert_called_once() publish_mock.assert_called_once() @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.logic.vendor.publish_account_event') @patch('account.logic.vendor.feature_model') @patch('account.logic.vendor.country_model') def test_v2_create_vendor_with_country( mock_country_model: MagicMock, mock_feature_model: MagicMock, publish_mock: MagicMock, mock_vendor_model: MagicMock, mock_update_closers: MagicMock, minimum_vendor_attrs: dict, ) -> None: """Assert country code is resolved to country_id and stored.""" mock_country_model.get_country_id_by_code.return_value = 6 mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( {'company_brand_id': 2} ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( {'create': 'successful', 'vendor_id': 1234, 'vendor_uuid': 'test-uuid'} ) mock_update_closers.return_value = response.Response({'success': True}) vendor.create_vendor( {**minimum_vendor_attrs, 'company_brand': 'theorchard', 'country': 'FRA', 'user_id': 111} ) mock_update_closers.assert_not_called() mock_country_model.get_country_id_by_code.assert_called_once_with('FRA') mock_vendor_model.create_vendor_with_optional_service_tier.assert_called_once_with( details={ 'name': minimum_vendor_attrs['name'], 'owner': minimum_vendor_attrs['owner'], 'company_brand_id': 2, 'migrated_to_abacus': True, 'last_modified_by': 111, 'is_distributor': 'N', 'label_identifier': None, 'country_id': 6, 'primary_genre': None, 'label_summary': None, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'wel_email_sender': None, }, service_tier_uuid=minimum_vendor_attrs['service_tier_uuid'], ) @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.logic.vendor.publish_account_event') @patch('account.logic.vendor.feature_model') @patch('account.logic.vendor.country_model') def test_v2_create_vendor_with_unknown_country_stores_null( mock_country_model: MagicMock, mock_feature_model: MagicMock, publish_mock: MagicMock, mock_vendor_model: MagicMock, mock_update_closers: MagicMock, minimum_vendor_attrs: dict, ) -> None: """Assert unknown country code stores NULL (no error raised).""" mock_country_model.get_country_id_by_code.return_value = None mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( {'company_brand_id': 2} ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( {'create': 'successful', 'vendor_id': 1234, 'vendor_uuid': 'test-uuid'} ) mock_update_closers.return_value = response.Response({'success': True}) vendor.create_vendor( {**minimum_vendor_attrs, 'company_brand': 'theorchard', 'country': 'ZZZ', 'user_id': 111} ) mock_update_closers.assert_not_called() mock_vendor_model.create_vendor_with_optional_service_tier.assert_called_once_with( details={ 'name': minimum_vendor_attrs['name'], 'owner': minimum_vendor_attrs['owner'], 'company_brand_id': 2, 'migrated_to_abacus': True, 'last_modified_by': 111, 'is_distributor': 'N', 'label_identifier': None, 'country_id': None, 'primary_genre': None, 'label_summary': None, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'wel_email_sender': None, }, service_tier_uuid=minimum_vendor_attrs['service_tier_uuid'], ) @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.logic.vendor.publish_account_event') @patch('account.logic.vendor.feature_model') def test_v2_create_vendor_with_primary_genre( mock_feature_model: MagicMock, publish_mock: MagicMock, mock_vendor_model: MagicMock, mock_update_closers: MagicMock, minimum_vendor_attrs: dict, ) -> None: """Assert primary_genre is passed through directly.""" mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( {'company_brand_id': 2} ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( {'create': 'successful', 'vendor_id': 1234, 'vendor_uuid': 'test-uuid'} ) mock_update_closers.return_value = response.Response({'success': True}) vendor.create_vendor( {**minimum_vendor_attrs, 'company_brand': 'theorchard', 'primary_genre': 42, 'user_id': 111} ) mock_update_closers.assert_not_called() mock_vendor_model.create_vendor_with_optional_service_tier.assert_called_once_with( details={ 'name': minimum_vendor_attrs['name'], 'owner': minimum_vendor_attrs['owner'], 'company_brand_id': 2, 'migrated_to_abacus': True, 'last_modified_by': 111, 'is_distributor': 'N', 'label_identifier': None, 'country_id': None, 'primary_genre': 42, 'label_summary': None, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'wel_email_sender': None, }, service_tier_uuid=minimum_vendor_attrs['service_tier_uuid'], ) @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.logic.vendor.publish_account_event') @patch('account.logic.vendor.feature_model') def test_v2_create_vendor_with_staff_fields( mock_feature_model: MagicMock, publish_mock: MagicMock, mock_vendor_model: MagicMock, mock_update_closers: MagicMock, minimum_vendor_attrs: dict, ) -> None: """Assert staff people fields are passed through to the model.""" mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( {'company_brand_id': 2} ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( {'create': 'successful', 'vendor_id': 1234, 'vendor_uuid': 'test-uuid'} ) mock_update_closers.return_value = response.Response({'success': True}) vendor.create_vendor( { **minimum_vendor_attrs, 'company_brand': 'theorchard', 'user_id': 111, 'assigned_to': 10, 'assigned_reviewer': 20, 'quarterback_label_manager': 30, 'wel_email_sender': 40, } ) mock_update_closers.assert_not_called() mock_vendor_model.create_vendor_with_optional_service_tier.assert_called_once_with( details={ 'name': minimum_vendor_attrs['name'], 'owner': minimum_vendor_attrs['owner'], 'company_brand_id': 2, 'migrated_to_abacus': True, 'last_modified_by': 111, 'is_distributor': 'N', 'label_identifier': None, 'country_id': None, 'primary_genre': None, 'label_summary': None, 'assigned_to': 10, 'assigned_reviewer': 20, 'quarterback_label_manager': 30, 'wel_email_sender': 40, }, service_tier_uuid=minimum_vendor_attrs['service_tier_uuid'], ) @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.logic.vendor.publish_account_event') @patch('account.logic.vendor.feature_model') def test_v2_create_vendor_with_product_manager_calls_setter( mock_feature_model: MagicMock, publish_mock: MagicMock, mock_vendor_model: MagicMock, mock_update_closers: MagicMock, minimum_vendor_attrs: dict, ) -> None: """Assert update_product_manager_id called when product_manager provided.""" mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( {'company_brand_id': 2} ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( {'create': 'successful', 'vendor_id': 1234, 'vendor_uuid': 'test-uuid'} ) mock_update_closers.return_value = response.Response({'success': True}) vendor.create_vendor( { **minimum_vendor_attrs, 'company_brand': 'theorchard', 'product_manager': 55, 'user_id': 111, } ) mock_update_closers.assert_not_called() mock_vendor_model.update_product_manager_id.assert_called_once_with(1234, 55) publish_mock.assert_called_once() publish_payload = publish_mock.call_args.kwargs['response_payload'] assert publish_payload['product_manager'] == 55 @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.logic.vendor.publish_account_event') @patch('account.logic.vendor.feature_model') def test_v2_create_vendor_without_product_manager_skips_setter( mock_feature_model: MagicMock, publish_mock: MagicMock, mock_vendor_model: MagicMock, mock_update_closers: MagicMock, minimum_vendor_attrs: dict, ) -> None: """Assert update_product_manager_id NOT called when product_manager omitted/None.""" mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( {'company_brand_id': 2} ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( {'create': 'successful', 'vendor_id': 1234, 'vendor_uuid': 'test-uuid'} ) mock_update_closers.return_value = response.Response({'success': True}) vendor.create_vendor( { **minimum_vendor_attrs, 'company_brand': 'theorchard', 'product_manager': None, 'user_id': 111, } ) mock_update_closers.assert_not_called() mock_vendor_model.update_product_manager_id.assert_not_called() publish_payload = publish_mock.call_args.kwargs['response_payload'] assert 'product_manager' not in publish_payload @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.logic.vendor.publish_account_event') @patch('account.logic.vendor.feature_model') def test_v2_create_awal_vendor_defaults_closer_to_2912( mock_feature_model: MagicMock, publish_mock: MagicMock, mock_vendor_model: MagicMock, mock_update_closers: MagicMock, minimum_vendor_attrs: dict, ) -> None: """Assert AWAL vendor creation sets default closer ID 2912.""" mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( {'company_brand_id': 5} ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( {'create': 'successful', 'vendor_id': 1234, 'vendor_uuid': 'awal-test-uuid'} ) mock_update_closers.return_value = response.Response({'success': True}) vendor.create_vendor({**minimum_vendor_attrs, 'company_brand': 'awal', 'user_id': 111}) mock_update_closers.assert_called_once_with('awal-test-uuid', [2912], 111) @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.logic.vendor.publish_account_event') @patch('account.logic.vendor.feature_model') def test_v2_create_vendor_with_closers_sets_closers( mock_feature_model: MagicMock, publish_mock: MagicMock, mock_vendor_model: MagicMock, mock_update_closers: MagicMock, minimum_vendor_attrs: dict, ) -> None: """Assert closers from the payload are applied within the create.""" mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( {'company_brand_id': 2} ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( {'create': 'successful', 'vendor_id': 1234, 'vendor_uuid': 'test-uuid'} ) mock_update_closers.return_value = response.Response({'success': True}) vendor.create_vendor( { **minimum_vendor_attrs, 'company_brand': 'foundation', 'closers': [3868], 'user_id': 111, } ) mock_update_closers.assert_called_once_with('test-uuid', [3868], 111) @patch('account.logic.vendor.update_vendor_closers') @patch('account.logic.vendor.vendor_model') @patch('account.logic.vendor.publish_account_event') @patch('account.logic.vendor.feature_model') def test_v2_create_awal_vendor_payload_closers_override_default( mock_feature_model: MagicMock, publish_mock: MagicMock, mock_vendor_model: MagicMock, mock_update_closers: MagicMock, minimum_vendor_attrs: dict, ) -> None: """Assert explicit closers in the payload override the AWAL default.""" mock_vendor_model.get_company_brand_id_by_name.return_value = response.Response( {'company_brand_id': 5} ) mock_vendor_model.create_vendor_with_optional_service_tier.return_value = response.Response( {'create': 'successful', 'vendor_id': 1234, 'vendor_uuid': 'awal-test-uuid'} ) mock_update_closers.return_value = response.Response({'success': True}) vendor.create_vendor( {**minimum_vendor_attrs, 'company_brand': 'awal', 'closers': [777], 'user_id': 111} ) mock_update_closers.assert_called_once_with('awal-test-uuid', [777], 111) def test_get_vendor_document(monkeypatch, fixture_vendor_document): """Test get vendor document by id.""" monkeypatch.setattr( vendor_model, 'get_vendor_document_by_id', MagicMock(return_value=response.Response(fixture_vendor_document)), ) result = vendor.get_vendor_document(2) assert result.message == fixture_vendor_document vendor_model.get_vendor_document_by_id.assert_called_once_with(2, False) def test_get_vendor_document_with_tenant_uuids(monkeypatch, fixture_vendor_document): """Test get vendor document by id.""" monkeypatch.setattr( vendor_model, 'get_vendor_document_by_id', MagicMock(return_value=response.Response(fixture_vendor_document)), ) result = vendor.get_vendor_document(2, True) assert result.message == fixture_vendor_document vendor_model.get_vendor_document_by_id.assert_called_once_with(2, True) def test_get_vendor_document_for_no_results(monkeypatch): """Test get vendor document by id for no results.""" vendor_id = 2 not_found_response = response.create_not_found_response( 'Vendor : {} not found.'.format(vendor_id) ) monkeypatch.setattr( vendor_model, 'get_vendor_document_by_id', MagicMock(return_value=not_found_response), ) result = vendor.get_vendor_document(vendor_id) assert result.message == not_found_response.message assert result.status == 404 def test_update_is_distributor_in_vendor(monkeypatch): """Test update_is_distributor_in_vendor.""" vendor_id = 2 monkeypatch.setattr( vendor_model, 'update_is_distributor_in_vendor', MagicMock(return_value=response.Response(constants.VENDOR_UPDATED)), ) result = vendor.update_is_distributor_in_vendor(vendor_id) assert result assert result.message == constants.VENDOR_UPDATED def test_get_company_brand(monkeypatch): """Test get vendor company brand with valid vendor_id.""" monkeypatch.setattr( neo4j_vendor, 'get_company_brand', MagicMock(return_value=response.Response('awal')), ) result = vendor.get_vendor_company_brand(1) assert result assert result.status == 200 def test_get_company_brand_with_invalid_vendor_id(monkeypatch): """Test get vendor company brand with valid vendor_id.""" result = vendor.get_vendor_company_brand(None) assert result.errors.get('message') == 'Invalid vendor id' assert result.status == 400 @patch('account.logic.vendor.get_session') def test_get_company_brand_tx(neo4j_session, monkeypatch): """Test get vendor company brand with valid vendor_id, using a Managed Transaction""" mock_response = MagicMock() mock_response.status = 200 mock_response.message = 'awal' monkeypatch.setattr( neo4j_vendor, 'get_company_brand_tx', MagicMock(return_value=mock_response), ) # mock session and execute_read method mock_session = MagicMock() mock_session.execute_read.return_value = mock_response neo4j_session.return_value = mock_session result = vendor.get_vendor_company_brand_tx(1) assert result assert result.status == 200 assert result.message == 'awal' @patch('account.logic.vendor.get_session') def test_get_company_brand_tx_with_invalid_vendor_id(neo4j_session, monkeypatch): """Test get vendor company brand with invalid vendor_id, using a Managed Transaction.""" result = vendor.get_vendor_company_brand_tx(None) assert result.errors.get('message') == 'Invalid vendor id' assert result.status == 400 def test_get_service_tiers(monkeypatch): """Test get service tiers.""" expected = [ {'uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'displayName': 'Basic'}, { 'uuid': '1dd92c83-25a3-4034-9a3d-f7c3f434f4ce', 'displayName': 'Premium Services', }, ] monkeypatch.setattr( neo4j_vendor, 'get_service_tiers', MagicMock(return_value=response.Response(expected)), ) result = vendor.get_service_tiers() assert result assert result.status == 200 assert result.message == expected @pytest.fixture def expected_lookup_vendors_by_uuids_response() -> list[Any]: return [ {'vendor_id': '*', 'uuid': '053a1a75-acc5-4cd8-9206-a194335d2afa'}, {'vendor_id': 35976, 'uuid': 'bad45218-68d8-42b6-b4ee-730ef9e16309'}, ] @patch('account.logic.vendor.format_for_dataloader') @patch('account.logic.vendor.vendor_model') def test_lookup_vendors_by_uuids( mock_vendor_model: MagicMock, mock_format_for_dataloader: MagicMock, expected_lookup_vendors_by_uuids_response: list[Any], ): """Test lookup vendors using uuids.""" mock_vendor_model.lookup_vendors_by_uuids.return_value = response.Response( expected_lookup_vendors_by_uuids_response ) uuids = [ UUID('053a1a75-acc5-4cd8-9206-a194335d2afa'), UUID('bad45218-68d8-42b6-b4ee-730ef9e16309'), ] mock_format_for_dataloader.return_value = expected_lookup_vendors_by_uuids_response result = vendor.lookup_vendors_by_uuids(uuids) assert result assert result.status == 200 assert result.message == {'vendors': expected_lookup_vendors_by_uuids_response} mock_format_for_dataloader.assert_called_once_with( expected_lookup_vendors_by_uuids_response, [str(uuid) for uuid in uuids], 'uuid', ) @patch('account.logic.vendor.vendor_model') def test_lookup_vendors_by_uuids_non_200( mock_vendor_model: MagicMock, ): """Test lookup vendors using uuids when non-200.""" expected = [] mock_vendor_model.lookup_vendors_by_uuids.return_value = response.Response(expected, status=400) result = vendor.lookup_vendors_by_uuids(['some', 'uuids']) assert not result assert result.status == 400 assert result.message == expected @patch('account.logic.vendor.format_for_dataloader') @patch('account.logic.vendor.vendor_model') def test_lookup_vendors_by_uuids_uses_fetch_flags( mock_vendor_model: MagicMock, mock_format_for_dataloader: MagicMock, ) -> None: """Test lookup vendors using uuids and hierarchy.""" uuids = [ UUID('053a1a75-acc5-4cd8-9206-a194335d2afa'), UUID('bad45218-68d8-42b6-b4ee-730ef9e16309'), ] uuids_str = [str(uuid) for uuid in uuids] expected = [] mock_vendor_model.lookup_vendors_by_uuids.return_value = response.Response(expected) mock_format_for_dataloader.return_value = expected result = vendor.lookup_vendors_by_uuids( uuids, [constants.FETCH_TENANT_HIERARCHY, constants.FETCH_IS_DISTRIBUTOR] ) assert result assert result.status == 200 assert result.message == {'vendors': expected} mock_vendor_model.lookup_vendors_by_uuids.assert_called_once_with( uuids_str, [constants.FETCH_TENANT_HIERARCHY, constants.FETCH_IS_DISTRIBUTOR], ) mock_format_for_dataloader.assert_called_once_with(expected, uuids_str, 'uuid') @patch('account.logic.vendor.format_for_dataloader') @patch('account.logic.vendor.vendor_model') def test_lookup_vendors_by_vendor_ids( mock_vendor_model: MagicMock, mock_format_for_dataloader: MagicMock, expected_lookup_vendors_by_uuids_response: list[Any], ) -> None: """Test lookup vendors using vendor_ids.""" mock_vendor_model.lookup_vendors_by_vendor_ids.return_value = response.Response( expected_lookup_vendors_by_uuids_response ) mock_format_for_dataloader.return_value = expected_lookup_vendors_by_uuids_response missing_vendor_id = 32 vendor_ids = ['*', missing_vendor_id, 35976] result = vendor.lookup_vendors_by_vendor_ids(vendor_ids) assert result assert result.status == 200 assert result.message == {'vendors': expected_lookup_vendors_by_uuids_response} mock_vendor_model.lookup_vendors_by_vendor_ids.assert_called() mock_format_for_dataloader.assert_called_once_with( expected_lookup_vendors_by_uuids_response, vendor_ids, 'vendor_id', ) @patch('account.logic.vendor.format_for_dataloader') @patch('account.logic.vendor.vendor_model') def test_lookup_vendors_by_vendor_ids_uses_fetch_flags( mock_vendor_model: MagicMock, mock_format_for_dataloader: MagicMock, ) -> None: """Test lookup vendors using uuids and hierarchy.""" vendor_ids = [7123, 8869] expected = [] mock_vendor_model.lookup_vendors_by_vendor_ids.return_value = response.Response(expected) mock_format_for_dataloader.return_value = expected result = vendor.lookup_vendors_by_vendor_ids( vendor_ids, [constants.LOOKUP_FETCH_FLAGS.TENANT_HIERARCHY] ) assert result assert result.status == 200 assert result.message == {'vendors': expected} mock_vendor_model.lookup_vendors_by_vendor_ids.assert_called_once_with( vendor_ids, [constants.LOOKUP_FETCH_FLAGS.TENANT_HIERARCHY], ) mock_format_for_dataloader.assert_called_once_with(expected, vendor_ids, 'vendor_id') @patch('account.logic.vendor.vendor_model') def test_lookup_vendors_by_vendor_ids_non_200( mock_vendor_model: MagicMock, monkeypatch, ) -> None: """Test lookup vendors using vendor_ids when non-200.""" expected = 'some error' mock_vendor_model.lookup_vendors_by_vendor_ids.return_value = response.Response( expected, status=400 ) result = vendor.lookup_vendors_by_vendor_ids(['some', 'vendor_ids']) assert not result assert result.status == 400 assert result.message == expected mock_vendor_model.lookup_vendors_by_vendor_ids.assert_called() @pytest.mark.parametrize( ('fixture_input_data', 'expected_response'), [ ( ('test_external_identifier_1', 'test_owner'), response.Response({'items': [{'vendor_id': 1}]}), ), ( ('test_external_identifier_2', 'test_owner'), response.Response({'items': []}), ), ( ('test_external_identifier_2', 'test_owner'), response.Response({'items': [{'vendor_id': 1}, {'vendor_id': 2}]}), ), ], ) def test_get_vendor_by_external_identifier(monkeypatch, fixture_input_data, expected_response): """Test get_vendor_by_external_identifier.""" monkeypatch.setattr( vendor_model, 'get_vendors_by_external_identifier', MagicMock(return_value=expected_response), ) result = vendor.get_vendors_by_external_identifier(*fixture_input_data) assert result == expected_response @pytest.mark.parametrize( 'ar_side_effect, expected', [ # no DB connection ( response.create_fatal_response('SQLAlchemyError'), {'code': 'internal_error', 'message': 'SQLAlchemyError'}, ), # AR issue only ( response.create_fatal_response('SQLAlchemyError'), {'code': 'internal_error', 'message': 'SQLAlchemyError'}, ), ], ) def test_update_soundscan_code_exceptions(monkeypatch, ar_side_effect, expected, app_context): """Test update soundscan_code exceptions.""" monkeypatch.setattr(vendor_model, 'update_vendor', MagicMock(return_value=ar_side_effect)) response = vendor.update_soundscan_code(1, 'testing', 1) assert not response assert response.status == 500 assert response.errors == expected @patch('account.logic.vendor.g', spec=['log', 'request_context']) @pytest.mark.parametrize( 'data', [ ( { 'vendor_id': 1234, 'vendor_name': 'Foo Productions', 'email': 'abc@xyz.com', 'owner': 'AWAL-Core', 'label_identifier': 'Frontline', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', } ) ], ) def test_update_company_brand_art_relations( mock_g, monkeypatch, data, fixture_valid_v2_vendor_empty_update, fixture_neo4j_fields, ): """Test update_company_brand_AR""" expected = fixture_valid_v2_vendor_empty_update.message monkeypatch.setattr(vendor_model, 'update_vendor', MagicMock()) monkeypatch.setattr(vendor_model, 'update_soundscan_code', MagicMock()) monkeypatch.setattr( vendor_model, 'get_vendor_label_info', MagicMock(return_value=fixture_valid_v2_vendor_empty_update), ) monkeypatch.setattr(vendor_model, 'get_soundscan_code', MagicMock()) monkeypatch.setattr(vendor_model, 'get_recurring_payment_threshold', MagicMock()) monkeypatch.setattr(vendor_model, 'get_product_manager_id', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'update_vendor', MagicMock()) monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response(message='enabled')), ) monkeypatch.setattr(vendor_model, 'update_company_brand_art_relations', MagicMock()) monkeypatch.setattr(neo4j_vendor, 'update_service_tier', MagicMock()) monkeypatch.setattr( neo4j_vendor, 'get_neo4j_vendor_info', MagicMock(return_value=fixture_neo4j_fields), ) monkeypatch.setattr(vendor, 'publish_account_event', MagicMock()) result = vendor.update_vendor(1234, data) assert result assert result.message == expected assert vendor_model.update_company_brand_art_relations.called def test_get_company_brand_id_not_found(monkeypatch, fixture_not_found_response): """Test get company_brand_id for non-existing company_brand.""" monkeypatch.setattr( vendor_model, '_get_company_brand_id', MagicMock(return_value=fixture_not_found_response), ) result = vendor_model._get_company_brand_id(1, 'test') assert not result assert result.status == 404 def test_get_master_contact_success(monkeypatch): """Test get_master_contact for valid vendor_id.""" monkeypatch.setattr( vend_contact.VendContact, 'get_master_contact', MagicMock( return_value=[ vend_contact.VendContact( vend_contact_id=1, vendor_id=2, auth0_user_id='1234abcd', ) ] ), ) monkeypatch.setattr( neo4j_identity, 'get_identity_by_auth0_user_id', MagicMock(return_value='test-user-identity-id'), ) result = vendor.get_master_contact(2) assert result.message.get('identity_id') == 'test-user-identity-id' assert neo4j_identity.get_identity_by_auth0_user_id.called_once_with('1234abcd') def test_get_master_contact_none_found(monkeypatch): """Test get_master_contact for valid vendor_id.""" monkeypatch.setattr(vend_contact.VendContact, 'get_master_contact', MagicMock(return_value=[])) result = vendor.get_master_contact(2) assert result.errors['message'] == 'Master contact not found.' def test_get_master_contact_more_than_one_found(monkeypatch): """Test get_master_contact returns more than 1 vend_contact.""" monkeypatch.setattr( vend_contact.VendContact, 'get_master_contact', MagicMock( return_value=[ vend_contact.VendContact( vend_contact_id=1, vendor_id=2, auth0_user_id='1234abcd', ), vend_contact.VendContact( vend_contact_id=2, vendor_id=2, auth0_user_id='5678efgh', ), ] ), ) result = vendor.get_master_contact(2) assert result.errors['message'] == 'Multiple master contacts found for vendor.' @patch('account.logic.vendor.neo4j_identity.get_identity_by_auth0_user_id') @patch('account.logic.vendor.vend_contact.VendContact.get_master_contact') def test_get_master_contact_no_identity_id( mock_get_master_contact: MagicMock, mock_get_identity: MagicMock, ) -> None: """Test get_master_contact when a vend_contact has no associated identity_id.""" vc = vend_contact.VendContact(vend_contact_id=1, vendor_id=2, auth0_user_id=None) mock_get_master_contact.return_value = [vc] mock_get_identity.return_value = None result = vendor.get_master_contact(2) assert result.status == 404 assert result.errors['message'] == 'Identity not found for auth0_user_id: None.' @pytest.fixture def mock_lookup_vendors_by_uuids_with_brand_response() -> dict[str, list[dict[str, Any]]]: return { 'vendors': [ { 'vendor_id': 15063, 'uuid': 'da3ad171-1d38-414a-b280-6a619bfa397f', 'company_brand_uuid': 'd25a4cd1-e820-45f2-be5c-56edcfeb8298', }, { 'vendor_id': 35976, 'uuid': 'bad45218-68d8-42b6-b4ee-730ef9e16309', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', }, ] } @pytest.mark.parametrize( 'auth_response,expected_result', [ ([True, False], ['da3ad171-1d38-414a-b280-6a619bfa397f']), ( [True, True], ['da3ad171-1d38-414a-b280-6a619bfa397f', 'bad45218-68d8-42b6-b4ee-730ef9e16309'], ), ([False, False], []), ], ) @patch('account.logic.vendor.lookup_vendors_by_uuids') @patch('account.logic.vendor.authorization_backend') def test_authorize_vendors_pdp( mock_authorization_backend: MagicMock, mock_lookup_vendors_by_uuids: MagicMock, mock_lookup_vendors_by_uuids_with_brand_response: dict[str, list[dict[str, Any]]], auth_response: list[bool], expected_result: list[str], ) -> None: """Test authorize_vendors_pdp.""" vendor_uuids = ['da3ad171-1d38-414a-b280-6a619bfa397f', 'bad45218-68d8-42b6-b4ee-730ef9e16309'] mock_lookup_vendors_by_uuids.return_value = response.Response( mock_lookup_vendors_by_uuids_with_brand_response ) mock_authorization_backend.is_authorized_many.return_value = auth_response result = vendor.get_pp_accessible_vendors(vendor_uuids) assert result == expected_result mock_lookup_vendors_by_uuids.assert_called_once_with( vendor_uuids, fetch_flags=[constants.FETCH_TENANT_HIERARCHY] ) mock_authorization_backend.is_authorized_many.assert_called_once_with( action='view_account_info', resource_type='account', resources_with_attributes=[ ResourceWithAttributes( resource_id='da3ad171-1d38-414a-b280-6a619bfa397f', attributes={ 'tenant': { 'tenant_type': 'account', 'tenant_uuid': 'da3ad171-1d38-414a-b280-6a619bfa397f', 'tenant_hierarchy': ['d25a4cd1-e820-45f2-be5c-56edcfeb8298'], } }, ), ResourceWithAttributes( resource_id='bad45218-68d8-42b6-b4ee-730ef9e16309', attributes={ 'tenant': { 'tenant_type': 'account', 'tenant_uuid': 'bad45218-68d8-42b6-b4ee-730ef9e16309', 'tenant_hierarchy': ['31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588'], } }, ), ], ) @patch('account.logic.vendor.lookup_vendors_by_uuids') @patch('account.logic.vendor.authorization_backend') def test_authorize_vendors_pdp_raises_value_error_when_auth_response_length_mismatch( mock_authorization_backend: MagicMock, mock_lookup_vendors_by_uuids: MagicMock, mock_lookup_vendors_by_uuids_with_brand_response: dict[str, list[dict[str, Any]]], ) -> None: vendor_uuids = ['da3ad171-1d38-414a-b280-6a619bfa397f', 'bad45218-68d8-42b6-b4ee-730ef9e16309'] mock_lookup_vendors_by_uuids.return_value = response.Response( mock_lookup_vendors_by_uuids_with_brand_response ) # Return only one auth response for two vendors mock_authorization_backend.is_authorized_many.return_value = [True] with pytest.raises(ValueError) as exc_info: vendor.get_pp_accessible_vendors(vendor_uuids) assert str(exc_info.value) == 'Authorization response contained 1 results for 2 vendors.' mock_lookup_vendors_by_uuids.assert_called_once_with( vendor_uuids, fetch_flags=[constants.FETCH_TENANT_HIERARCHY] ) @patch('account.models.vendor.get_vendor_company_brands') def test_get_vendor_company_brands_success(mock_model_get_vendor_company_brands: MagicMock) -> None: """Test get_vendor_company_brands with successful model call.""" vendor_uuids = ['uuid1', 'uuid2'] mock_model_data = [ { 'vendor_id': 1, 'uuid': 'uuid1', 'company_brand_id': 10, 'company_brand_uuid': 'brand_uuid1', }, { 'vendor_id': 2, 'uuid': 'uuid2', 'company_brand_id': 20, 'company_brand_uuid': 'brand_uuid2', }, ] expected_formatted = { 'vendors': [ { 'vendor_id': 1, 'uuid': 'uuid1', 'company_brand_id': 10, 'company_brand_uuid': 'brand_uuid1', }, { 'vendor_id': 2, 'uuid': 'uuid2', 'company_brand_id': 20, 'company_brand_uuid': 'brand_uuid2', }, ] } mock_model_get_vendor_company_brands.return_value = mock_model_data result = vendor.get_vendor_company_brands(vendor_uuids) assert result == expected_formatted mock_model_get_vendor_company_brands.assert_called_once_with(vendor_uuids) @patch('account.models.vendor.get_vendor_company_brands') @patch('account.logic.vendor.format_for_dataloader') def test_get_vendor_company_brands_empty_list( mock_format_for_dataloader: MagicMock, mock_model_get_vendor_company_brands: MagicMock, ) -> None: """Test get_vendor_company_brands with empty vendor list.""" vendor_uuids = [] mock_model_data = [] expected_formatted = {'vendors': []} mock_model_get_vendor_company_brands.return_value = mock_model_data mock_format_for_dataloader.return_value = [] result = vendor.get_vendor_company_brands(vendor_uuids) assert result == expected_formatted mock_model_get_vendor_company_brands.assert_called_once_with(vendor_uuids) mock_format_for_dataloader.assert_called_once_with(mock_model_data, vendor_uuids, 'uuid') @patch('account.models.vendor.get_vendor_company_brands') @patch('account.logic.vendor.format_for_dataloader') def test_get_vendor_company_brands_no_results( mock_format_for_dataloader: MagicMock, mock_model_get_vendor_company_brands: MagicMock, ) -> None: """Test get_vendor_company_brands when no vendors found.""" vendor_uuids = ['nonexistent_uuid'] mock_model_data = [] expected_formatted = {'vendors': [None]} mock_model_get_vendor_company_brands.return_value = mock_model_data mock_format_for_dataloader.return_value = [None] result = vendor.get_vendor_company_brands(vendor_uuids) assert result == expected_formatted mock_model_get_vendor_company_brands.assert_called_once_with(vendor_uuids) mock_format_for_dataloader.assert_called_once_with(mock_model_data, vendor_uuids, 'uuid') @patch('account.models.vendor.get_vendor_company_brands', side_effect=Exception('Database error')) def test_get_vendor_company_brands_model_exception(mock_model_get_vendor_company_brands): """Test get_vendor_company_brands when model raises exception.""" vendor_uuids = ['uuid1'] with pytest.raises(Exception) as exc_info: vendor.get_vendor_company_brands(vendor_uuids) assert str(exc_info.value) == 'Database error' mock_model_get_vendor_company_brands.assert_called_once_with(vendor_uuids) @patch('account.models.vendor.get_vendor_service_tier') def test_get_vendor_service_tier_success(mock_model_get_vendor_service_tier: MagicMock) -> None: """Test get_vendor_service_tier with successful model call.""" vendor_uuids = ['uuid1', 'uuid2'] mock_model_data = [ { 'vendor_id': 1, 'uuid': 'uuid1', 'service_tier_uuid': 'st_uuid_1', 'service_tier_name': 'premium-services', 'service_tier_display_name': 'Premium Services', }, { 'vendor_id': 2, 'uuid': 'uuid2', 'service_tier_uuid': 'st_uuid_2', 'service_tier_name': 'tier-1', 'service_tier_display_name': 'Tier 1', }, ] expected_formatted = { 'vendors': [ { 'vendor_id': 1, 'uuid': 'uuid1', 'service_tier_uuid': 'st_uuid_1', 'service_tier_name': 'premium-services', 'service_tier_display_name': 'Premium Services', }, { 'vendor_id': 2, 'uuid': 'uuid2', 'service_tier_uuid': 'st_uuid_2', 'service_tier_name': 'tier-1', 'service_tier_display_name': 'Tier 1', }, ] } mock_model_get_vendor_service_tier.return_value = mock_model_data result = vendor.get_vendor_service_tier(vendor_uuids) assert result == expected_formatted mock_model_get_vendor_service_tier.assert_called_once_with(vendor_uuids) @patch('account.models.vendor.get_vendor_service_tier') @patch('account.logic.vendor.format_for_dataloader') def test_get_vendor_service_tier_empty_list( mock_format_for_dataloader: MagicMock, mock_model_get_vendor_service_tier: MagicMock, ) -> None: """Test get_vendor_service_tier with empty vendor list.""" vendor_uuids = [] mock_model_data = [] expected_formatted = {'vendors': []} mock_model_get_vendor_service_tier.return_value = mock_model_data mock_format_for_dataloader.return_value = [] result = vendor.get_vendor_service_tier(vendor_uuids) assert result == expected_formatted mock_model_get_vendor_service_tier.assert_called_once_with(vendor_uuids) mock_format_for_dataloader.assert_called_once_with(mock_model_data, vendor_uuids, 'uuid') @patch('account.models.vendor.get_vendor_service_tier') @patch('account.logic.vendor.format_for_dataloader') def test_get_vendor_service_tier_no_results( mock_format_for_dataloader: MagicMock, mock_model_get_vendor_service_tier: MagicMock, ) -> None: """Test get_vendor_service_tier when no vendors found.""" vendor_uuids = ['nonexistent_uuid'] mock_model_data = [] expected_formatted = {'vendors': [None]} mock_model_get_vendor_service_tier.return_value = mock_model_data mock_format_for_dataloader.return_value = [None] result = vendor.get_vendor_service_tier(vendor_uuids) assert result == expected_formatted mock_model_get_vendor_service_tier.assert_called_once_with(vendor_uuids) mock_format_for_dataloader.assert_called_once_with(mock_model_data, vendor_uuids, 'uuid') @patch('account.models.vendor.get_vendor_service_tier', side_effect=Exception('Database error')) def test_get_vendor_service_tier_model_exception(mock_model_get_vendor_service_tier): """Test get_vendor_service_tier when model raises exception.""" vendor_uuids = ['uuid1'] with pytest.raises(Exception) as exc_info: vendor.get_vendor_service_tier(vendor_uuids) assert str(exc_info.value) == 'Database error' mock_model_get_vendor_service_tier.assert_called_once_with(vendor_uuids) def test_get_vendors_relationship_notes_true(monkeypatch): """Test check for relationship_notes where vendor exists.""" vendor_uuids = [ '11111111-1111-1111-1111-111111111111', '22222222-2222-2222-2222-222222222222', '33333333-3333-3333-3333-333333333333', ] model_notes_response = [ { 'vendor_id': 1, 'vendor_uuid': '11111111-1111-1111-1111-111111111111', 'relationship_notes': 'Note A', }, { 'vendor_id': 2, 'vendor_uuid': '22222222-2222-2222-2222-222222222222', 'relationship_notes': 'Note B', }, ] expected_result = { 'vendors': [ { 'vendor_id': 1, 'vendor_uuid': '11111111-1111-1111-1111-111111111111', 'relationship_notes': 'Note A', }, { 'vendor_id': 2, 'vendor_uuid': '22222222-2222-2222-2222-222222222222', 'relationship_notes': 'Note B', }, None, ] } monkeypatch.setattr( vendor_model, 'get_relationship_notes', MagicMock(return_value=model_notes_response), ) result = vendor.get_vendors_relationship_notes(vendor_uuids) assert result == expected_result def test_publish_account_event_serializes_dates(monkeypatch): """Test publish_account_event serializes datetime values to ISO strings in the payload.""" import datetime mock_produce = MagicMock() monkeypatch.setattr(vendor.kafka_producer, 'produce', mock_produce) date_created = datetime.datetime(2024, 3, 10, 9, 30, 0) event = {'vendor_id': 42, 'date_created': date_created, 'name': 'Test Label'} vendor.publish_account_event(42, event, 'create') mock_produce.assert_called_once() _, _, payload, _ = mock_produce.call_args.args assert payload['payload']['date_created'] == '2024-03-10T09:30:00' assert payload['payload']['vendor_id'] == 42 assert payload['payload']['name'] == 'Test Label'