"""Test for a vendor model.""" from typing import Any from unittest.mock import ANY, MagicMock, patch from uuid import uuid4 import pytest from owsresponse import response from pythonfeatures import pythonfeatures from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm.session import Session from account.connectors import mysql from account.constants import constants from account.models import vendor from account.models.sql.vendor_document import ( VENDOR_DOCUMENT_SQL, VENDOR_DOCUMENT_WITH_TENANT_UUIDS_SQL, ) from account.utils.exception import VendorUpdateException from tests.unit import db_operations from tests.unit.conftest import get_session_mock @pytest.fixture def db_fixture(): """Set up the vendor/label info tables.""" db_operations.create_tables() db_operations.seed_vendor_table() db_operations.seed_vendor_closers_table() db_operations.seed_genre_table() db_operations.seed_region_table() db_operations.seed_country_table() db_operations.seed_parent_company_table() db_operations.seed_company_brand_table() db_operations.seed_soundscan_codes_table() db_operations.seed_vendor_assigned_to_data() db_operations.seed_service_tier_table() db_operations.seed_vendor_service_tier_table() def test_get_distributor_success(db_fixture): """Test succeeds when vendor is a distributor.""" response = vendor.get_distributor(1) assert response def test_get_distributor_fails_for_non_existent(db_fixture): """Test fails when vendor does not exist.""" response = vendor.get_distributor(100) assert not response def test_get_distributor_fails_for_non_distributor(db_fixture): """Test fails when vendor is not a distributor.""" response = vendor.get_distributor(2) assert not response def test_get_owner_success(db_fixture): """Test succeeds when owner information is retrieved.""" response = vendor.get_owner(1) assert response assert response.message == {'owner': 'TEST1'} def test_get_owner_fails_for_not_exist(db_fixture): """Test fails when vendor_id does not return a record.""" response = vendor.get_owner(100) assert response.status == 404 assert not response def test_get_vendor(db_fixture): """Test get_vendor to return a record.""" response = vendor.get_vendor(2) expected_result = { 'country_id': 1, 'owner': 'TEST2', 'name': 'Test', 'company': None, 'vendor_id': 2, 'is_distributor': 'N', 'status': 'signed', 'label_identifier': None, 'contact_email': None, 'migrated_to_abacus': False, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'last_modified_by': None, } assert response assert response.message == expected_result @patch('account.logic.vendor.g', spec=['log', 'request_context']) def test_get_vendor_label_info(mock_g, db_fixture, app_context): """Test get_vendor_label_info to return a record.""" response = vendor.get_vendor_label_info(1) expected_result = { 'vendor_id': 1, 'is_distributor': 'Y', 'owner': 'TEST1', 'country': 'France', 'country_code': 'FR', 'status': 'signed', 'transfer_pricing_country': None, 'website': None, 'est_tracks': None, 'company': None, 'priority': None, 'support_contact_email': None, 'name': 'Test', 'projected_first_year_revenue': None, 'est_releases': 0, 'label_summary': None, 'is_owned': None, 'last_update': None, 'primary_genre': 12, 'assigned_to': None, 'region': 'Canada', 'wel_email_sender': None, 'wel_email_send_date': None, 'allow_ALW_release_builder': None, 'contact_email': None, 'country_id': 6, 'sap_vendor_id': None, 'quarterback_label_manager': None, 'assigned_reviewer': None, 'oa_admin_first': None, 'oa_admin_last': None, 'migrated_to_abacus': 0, 'label_identifier': None, 'date_created': None, 'genre': 'Classical', 'vendor_uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'company_brand': 'test_theorchard', } assert response assert response.message == expected_result def test_get_vendors_valid_uuids(db_fixture, app_context): """Test get_vendors to return a records.""" response = vendor.get_vendors( ['87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538'] ) assert len(response) == 2 assert response[0] == { 'company_brand_id': 1, 'is_distributor': 'Y', 'name': 'Test', 'owner': 'TEST1', 'status': 'signed', 'vendor_id': 1, 'vendor_uuid': '87682992-bff2-40ff-aa75-18eb6214679e', } assert response[1] == { 'company_brand_id': 2, 'is_distributor': 'N', 'name': 'Test', 'owner': 'TEST2', 'status': 'signed', 'vendor_id': 2, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', } def test_get_vendors_empty_list(db_fixture, app_context): """Test get_vendors to return empty list""" response = vendor.get_vendors([]) assert len(response) == 0 def test_get_vendors_invalid_uuids(db_fixture, app_context): """Test get_vendors to return for invalid uuid""" response = vendor.get_vendors(['invalid_uuid']) assert len(response) == 0 def test_get_vendors_valid_and_not_valid_uuids(db_fixture, app_context): """Test get_vendors to return a records for valid uuids.""" response = vendor.get_vendors(['87682992-bff2-40ff-aa75-18eb6214679e', 'invalid_uuid']) assert len(response) == 1 assert response[0] == { 'company_brand_id': 1, 'is_distributor': 'Y', 'name': 'Test', 'owner': 'TEST1', 'status': 'signed', 'vendor_id': 1, 'vendor_uuid': '87682992-bff2-40ff-aa75-18eb6214679e', } def test_get_vendors_names_valid_uuids(db_fixture, app_context): """Test get_vendor_names to return valid records.""" response = vendor.get_vendor_names( ['87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538'] ) assert len(response) == 2 assert response[0] == { 'name': 'Test', 'vendor_id': 1, 'vendor_uuid': '87682992-bff2-40ff-aa75-18eb6214679e', } assert response[1] == { 'name': 'Test', 'vendor_id': 2, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', } def test_get_vendor_names_empty_list(db_fixture, app_context): """Test get_vendor_names to return empty list""" response = vendor.get_vendor_names([]) assert len(response) == 0 def test_get_vendor_names_invalid_uuids(db_fixture, app_context): """Test get_vendor_names to return for invalid uuid""" response = vendor.get_vendor_names(['invalid_uuid']) assert len(response) == 0 def test_get_vendor_names_valid_and_not_valid_uuids(db_fixture, app_context): """Test get_vendor_names to return a records for valid uuids.""" response = vendor.get_vendor_names(['87682992-bff2-40ff-aa75-18eb6214679e', 'invalid_uuid']) assert len(response) == 1 assert response[0] == { 'name': 'Test', 'vendor_id': 1, 'vendor_uuid': '87682992-bff2-40ff-aa75-18eb6214679e', } @pytest.mark.parametrize( 'vendor_uuids, expected_len, expected_result', [ pytest.param( ['87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538'], 2, [ {'uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'closers': [2]}, {'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'closers': [2, 3]}, ], id='Valid vendor uuids', ), pytest.param([], 0, [], id='Empty list of vendor IDs'), pytest.param(['invalid_vendor_uuids'], 0, [], id='Invalid vendor ID'), pytest.param( ['87682992-bff2-40ff-aa75-18eb6214679e', 'invalid_vendor_uuids'], 1, [{'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'closers': [2, 3]}], id='Mixed valid and invalid vendor uuids', ), pytest.param( ['42d022c2-50e4-4f5c-af30-f10dca6b8a50'], 0, [], id='UUID in vendor table but no closers', ), ], ) def test_get_vendor_closers(vendor_uuids, expected_len, expected_result, db_fixture, app_context): result = vendor.get_vendor_closers(vendor_uuids) assert len(result) == expected_len assert result == expected_result @pytest.mark.parametrize( 'vendor_uuids, expected_len, expected_result', [ pytest.param( ['87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538'], 2, [ {'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'first_statement_period': '1'}, {'uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'first_statement_period': '229'}, ], id='Valid vendor uuids', ), pytest.param([], 0, [], id='Empty list of vendor IDs'), pytest.param(['invalid_vendor_uuids'], 0, [], id='Invalid vendor ID'), pytest.param( ['87682992-bff2-40ff-aa75-18eb6214679e', 'invalid_vendor_uuids'], 1, [{'first_statement_period': '1', 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e'}], id='Mixed valid and invalid vendor uuids', ), ], ) def test_get_vendors_first_statement_period( vendor_uuids, expected_len, expected_result, db_fixture, app_context ): result = vendor.get_vendors_first_statement_period(vendor_uuids) assert len(result) == expected_len assert result == expected_result def test_get_soundscan_code(db_fixture, app_context): """Test get_soundscan_code to return a record.""" response = vendor.get_soundscan_code(5, 1).message.get('soundscan_codes_soundscan_code') expected_result = 't1' assert response assert response == expected_result def test_update_product_manager_id(): """Test update_product_manager_id. Note: this test avoids the requirement for a live/seeded mysql instance and is a pattern we can follow to prevent failures related to the test database and unrelated to the code under test. """ session_mock = get_session_mock() with patch.object(mysql, 'session_scope', return_value=session_mock): result = vendor.update_product_manager_id(1, 123) assert result assert result.message == 'product manager 123 assigned to vendor 1.' args = session_mock.__enter__().execute.call_args assert ( args[0][0].text == 'INSERT INTO product_manager_mapping_vendor( product_manager_id, vendor_id, updated_at) VALUES (:pm_id, :vendor_id, current_timestamp) ON DUPLICATE KEY UPDATE product_manager_id = :pm_id, updated_at = current_timestamp;' # noqa: E501 ) # noqa def test_delete_product_manager(): """Test delete_product_manager_id from vendor.""" session_mock = get_session_mock() with patch.object(mysql, 'session_scope', return_value=session_mock): result = vendor.delete_product_manager_id(123) assert result args = session_mock.__enter__().execute.call_args assert ( args[0][0].text == '\nDELETE FROM product_manager_mapping_vendor\nWHERE vendor_id=:vendor_id\n' ) # noqa def test_get_product_manager(): """Test get_product_manager_id.""" session_mock = get_session_mock() with patch.object(mysql, 'session_scope', return_value=session_mock): result = vendor.get_product_manager_id(123) assert result args = session_mock.__enter__().execute.call_args assert ( args[0][0] == '\nSELECT oa.id AS product_manager\nFROM orchadmin_users oa\nJOIN product_manager_mapping_vendor pm\n ON pm.product_manager_id = oa.id\nWHERE pm.vendor_id=:vendor_id\n' # noqa: E501 ) # noqa @pytest.mark.parametrize( ('payload', 'expected'), [ ( { 'vendor_id': 2, 'label_identifier': 'Catalog', 'name': 'TEST2 NEW', 'company': 'TEST2 NEW', 'last_modified_by': 1111, }, { 'country_id': 1, 'owner': 'TEST2', 'name': 'TEST2 NEW', 'company': 'TEST2 NEW', 'vendor_id': 2, 'is_distributor': 'N', 'status': 'signed', 'label_identifier': 'Catalog', 'contact_email': None, 'migrated_to_abacus': False, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'last_modified_by': 1111, }, ), ( { 'label_identifier': 'Catalog', 'name': 'TEST2 NEW', 'company': 'TEST2 NEW', 'migrated_to_abacus': True, 'assigned_to': 1234, 'last_modified_by': 1111, }, { 'country_id': 1, 'owner': 'TEST2', 'name': 'TEST2 NEW', 'company': 'TEST2 NEW', 'vendor_id': 2, 'is_distributor': 'N', 'status': 'signed', 'label_identifier': 'Catalog', 'contact_email': None, 'migrated_to_abacus': True, 'assigned_to': 1234, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'last_modified_by': 1111, }, ), ], ) def test_update_vendor(db_fixture, payload, expected): """Test update_vendor.""" response = vendor.update_vendor(2, payload) expected_result = expected assert response assert response.message == { 'vendor_id': 2, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', } response = vendor.get_vendor(2) assert response.message == expected_result def test_update_vendor_will_update_external_identifier_1(db_fixture): """Test update_vendor.""" external_identifier_1 = 'i am a parent rep owner' payload = { 'vendor_id': 2, 'external_identifier_1': external_identifier_1, 'last_modified_by': 1211, } expected = { 'country_id': 1, 'owner': 'TEST2', 'name': 'Test', 'company': None, 'vendor_id': 2, 'is_distributor': 'N', 'status': 'signed', 'label_identifier': None, 'contact_email': None, 'migrated_to_abacus': False, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'last_modified_by': 1211, } response = vendor.update_vendor(2, payload) expected_result = expected assert response assert response.message == { 'vendor_id': 2, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', } response = vendor.get_vendor(2) assert response.message == expected_result, 'Only the last_modified_by is updated' vendors_response = vendor.get_vendors_by_external_identifier(external_identifier_1, 'TEST2') assert vendors_response assert len(vendors_response.message['items']) == 1 assert vendors_response.message['items'][0] == { 'vendor_id': 2, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', } @patch('account.connectors.mysql._account_session', side_effect=SQLAlchemyError()) def test_update_vendor_failure(session_scope, monkeypatch): """Test get update_vendor when failed.""" monkeypatch.setattr(session_scope, 'commit', value=MagicMock(return_value=False)) response = vendor.update_vendor(2, {'name': 'Vendor Update'}) assert response.status == 500 @pytest.mark.parametrize( ('vendor_id', 'vendor_uuid', 'closers'), [ pytest.param( 1, '87682992-bff2-40ff-aa75-18eb6214679e', [1, 2], id='Updating closers (add new, remove old)', ), pytest.param(2, '6097ad5a-2186-4dbe-8286-82ce933d3538', [], id='Removing existing closers'), pytest.param(3, 'e429b770-9db6-4f98-b48d-7d9b8ced7e6c', [3], id='No change in closers'), pytest.param(4, '42d022c2-50e4-4f5c-af30-f10dca6b8a50', [2], id='Adding new closers'), ], ) def test_upsert_vendor_closers(db_fixture, vendor_id, vendor_uuid, closers): """Test upsert_vendor_closers under various conditions.""" last_modified_by = 1 result = vendor.upsert_vendor_closers(vendor_uuid, closers, last_modified_by) assert result['vendor_id'] == vendor_id assert result['vendor_uuid'] == vendor_uuid assert set(result['closers']) == set(closers) def test_upsert_vendor_closers_vendor_not_found(db_fixture): """Test upsert_vendor_closers invalid vendor.""" with pytest.raises(VendorUpdateException) as exc_info: vendor.upsert_vendor_closers('invalid_uuid', [2, 3], 1) assert exc_info.value.status == 404 assert exc_info.value.message == 'Vendor with UUID invalid_uuid not found' @patch('account.connectors.mysql._account_session', side_effect=SQLAlchemyError()) @patch('account.models.vendor.g') def test_upsert_vendor_closers_failure(session_scope, monkeypatch, app_context): """Test upsert_vendor_closers failure scenario.""" monkeypatch.setattr(session_scope, 'commit', value=MagicMock(return_value=False)) with pytest.raises(VendorUpdateException) as exc_info: vendor.upsert_vendor_closers('42d022c2-50e4-4f5c-af30-f10dca6b8a50', [2, 3], 1) assert exc_info.value.status == 500 @patch('account.models.vendor.update_vendor_relationship_notes') def test_update_vendor_relationship_notes_success(mock_update): """Test update_vendor_relationship_notes model.""" vendor_uuid = str(uuid4()) vendor_id = 1234 relationship_notes = 'These are some relationship notes.' expected_response = response.Response( { 'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id, 'relationship_notes': relationship_notes, } ) mock_update.return_value = expected_response resp = vendor.update_vendor_relationship_notes(vendor_uuid, relationship_notes) assert resp.message == expected_response.message mock_update.assert_called_once_with(vendor_uuid, relationship_notes) def test_update_vendor_relationship_notes_vendor_not_found(db_fixture): """Test update_vendor_relationship_notes invalid vendor.""" with pytest.raises(VendorUpdateException) as exc_info: vendor.update_vendor_relationship_notes('invalid_uuid', 'Some notes.') assert exc_info.value.message == ( 'Vendor update unsuccessful: No vendor found for vendor_uuid invalid_uuid.' ) @patch( 'account.connectors.mysql._account_session', side_effect=SQLAlchemyError('DB connection failed') ) @patch('account.models.vendor.g') def test_update_vendor_relationship_notes_failure(session_scope, monkeypatch, app_context): """Test update_vendor_relationship_notes failure scenario.""" monkeypatch.setattr(session_scope, 'commit', value=MagicMock(return_value=False)) with pytest.raises(Exception) as exc_info: vendor.update_vendor_relationship_notes( '42d022c2-50e4-4f5c-af30-f10dca6b8a50', 'Some notes.' ) assert 'DB connection failed' in str(exc_info.value) @patch('account.models.vendor.update_vendor_first_statement_period') def test_update_vendor_first_statement_period_success(mock_update): vendor_uuid = str(uuid4()) vendor_id = 4567 first_statement_period = 1 expected_response = response.Response( { 'vendor_uuid': vendor_uuid, 'vendor_id': vendor_id, 'first_statement_period': str(first_statement_period), } ) mock_update.return_value = expected_response resp = vendor.update_vendor_first_statement_period(vendor_uuid, first_statement_period) assert resp.message == expected_response.message mock_update.assert_called_once_with(vendor_uuid, first_statement_period) def test_update_vendor_first_statement_period_vendor_not_found(db_fixture): """Test update_vendor_first_statement_period invalid vendor.""" with pytest.raises(Exception) as exc_info: vendor.update_vendor_first_statement_period('invalid_uuid', 1) assert ( str(exc_info.value) == 'Vendor update unsuccessful: No vendor found for vendor_uuid invalid_uuid.' ) @patch( 'account.connectors.mysql._account_session', side_effect=SQLAlchemyError('DB connection failed') ) @patch('account.models.vendor.g') def test_update_vendor_first_statement_period_failure(session_scope, monkeypatch, app_context): """Test update_vendor_first_statement_period failure scenario.""" monkeypatch.setattr(session_scope, 'commit', value=MagicMock(return_value=False)) with pytest.raises(Exception) as exc_info: vendor.update_vendor_first_statement_period('42d022c2-50e4-4f5c-af30-f10dca6b8a50', 1) assert 'DB connection failed' in str(exc_info.value) def test_update_vendor_service_tier_by_uuid_success(db_fixture, app_context): """Happy path: replaces the vendor's service tier and returns vendor ids.""" vendor_uuid = '87682992-bff2-40ff-aa75-18eb6214679e' new_tier_uuid = '2ed7aac0-ceb6-4c09-9166-afda8f349316' result = vendor.update_vendor_service_tier_by_uuid( vendor_uuid, new_tier_uuid, last_modified_by=1 ) assert result['vendor_uuid'] == vendor_uuid with mysql.session_scope(read_only=True) as session: updated = ( session.query(vendor.Vendor).filter(vendor.Vendor.vendor_uuid == vendor_uuid).one() ) assert updated.service_tier.uuid == new_tier_uuid def test_update_vendor_service_tier_by_uuid_invalid_tier(db_fixture, app_context): """Unknown service_tier_uuid raises VendorUpdateException(400).""" with pytest.raises(VendorUpdateException) as exc_info: vendor.update_vendor_service_tier_by_uuid( '87682992-bff2-40ff-aa75-18eb6214679e', '00000000-0000-0000-0000-000000000000', 1, ) assert exc_info.value.status == 400 @patch( 'account.connectors.mysql._account_session', side_effect=SQLAlchemyError('DB connection failed') ) @patch('account.models.vendor.g') def test_update_vendor_service_tier_by_uuid_db_error(session_scope, monkeypatch, app_context): """SQLAlchemyError is wrapped as VendorUpdateException(500).""" monkeypatch.setattr(session_scope, 'commit', value=MagicMock(return_value=False)) with pytest.raises(VendorUpdateException) as exc_info: vendor.update_vendor_service_tier_by_uuid( '87682992-bff2-40ff-aa75-18eb6214679e', '2ed7aac0-ceb6-4c09-9166-afda8f349316', 1, ) assert exc_info.value.status == 500 def test_get_vendor_invalid_id(db_fixture): """Test get_vendor fail with invalid vendor.""" response = vendor.get_vendor(100) assert not response assert response.status == 404 def test_lookup_vendors_and_company_brands_by_uuids_valid_uuids(db_fixture, app_context): """Test lookup_vendors_and_company_brands_by_uuids with valid UUIDs.""" vendor_uuids = ['87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538'] with mysql.session_scope(read_only=True) as session: result = vendor.lookup_vendors_and_company_brands_by_uuids(vendor_uuids, session) assert len(result) == 2 # Check first vendor vendor_1 = result['87682992-bff2-40ff-aa75-18eb6214679e'] assert vendor_1[0]['vendor_id'] == 1 assert vendor_1[0]['uuid'] == '87682992-bff2-40ff-aa75-18eb6214679e' assert vendor_1[0]['migrated_to_abacus'] is False assert vendor_1[0]['owner'] == 'TEST1' assert vendor_1[1]['id'] == 1 assert vendor_1[1]['name'] == 'test_theorchard' # Check second vendor vendor_2 = result['6097ad5a-2186-4dbe-8286-82ce933d3538'] assert vendor_2[0]['vendor_id'] == 2 assert vendor_2[0]['uuid'] == '6097ad5a-2186-4dbe-8286-82ce933d3538' assert vendor_2[0]['migrated_to_abacus'] is False assert vendor_2[0]['owner'] == 'TEST2' assert vendor_2[1]['id'] == 2 assert vendor_2[1]['name'] == 'test_awal' def test_lookup_vendors_and_company_brands_by_uuids_empty_list( db_fixture, app_context: MagicMock ) -> None: """Test lookup_vendors_and_company_brands_by_uuids with empty list.""" with mysql.session_scope(read_only=True) as session: result = vendor.lookup_vendors_and_company_brands_by_uuids([], session) assert result == {} def test_lookup_vendors_and_company_brands_by_uuids_invalid_uuids( db_fixture, app_context: MagicMock ) -> None: """Test lookup_vendors_and_company_brands_by_uuids with invalid UUIDs.""" vendor_uuids = ['invalid-uuid-1', 'invalid-uuid-2'] with mysql.session_scope(read_only=True) as session: result = vendor.lookup_vendors_and_company_brands_by_uuids(vendor_uuids, session) assert result == {} def test_lookup_vendors_and_company_brands_by_uuids_mixed_uuids( db_fixture, app_context: MagicMock ) -> None: """Test lookup_vendors_and_company_brands_by_uuids with mix of valid and invalid UUIDs.""" vendor_uuids = ['87682992-bff2-40ff-aa75-18eb6214679e', 'invalid-uuid'] with mysql.session_scope(read_only=True) as session: result = vendor.lookup_vendors_and_company_brands_by_uuids(vendor_uuids, session) assert len(result) == 1 assert '87682992-bff2-40ff-aa75-18eb6214679e' in result assert 'invalid-uuid' not in result vendor_1 = result['87682992-bff2-40ff-aa75-18eb6214679e'] assert vendor_1[0]['vendor_id'] == 1 assert vendor_1[0]['uuid'] == '87682992-bff2-40ff-aa75-18eb6214679e' assert vendor_1[0]['owner'] == 'TEST1' @pytest.mark.parametrize( 'vendor_uuids, expected_length, expected_keys', [ pytest.param( [], 0, [], id='empty_list', ), pytest.param( ['12345678-1234-4567-8901-123456789012', '87654321-4321-4567-8901-210987654321'], 0, [], id='not_found_uuids', ), pytest.param( ['87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538'], 2, ['87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538'], id='valid_uuids', ), pytest.param( ['87682992-bff2-40ff-aa75-18eb6214679e', '12345678-1234-4567-8901-123456789012'], 1, ['87682992-bff2-40ff-aa75-18eb6214679e'], id='mixed_uuids', ), ], ) def test_lookup_vendors_and_company_brands_by_uuids( db_fixture, app_context: MagicMock, vendor_uuids: list[str], expected_length: int, expected_keys: list[str], ) -> None: """Test lookup_vendors_and_company_brands_by_uuids.""" with mysql.session_scope(read_only=True) as session: result = vendor.lookup_vendors_and_company_brands_by_uuids(vendor_uuids, session) assert len(result) == expected_length for key in expected_keys: assert key in result # Validate structure for valid entries if '87682992-bff2-40ff-aa75-18eb6214679e' in result: vendor_1 = result['87682992-bff2-40ff-aa75-18eb6214679e'] assert vendor_1[0]['vendor_id'] == 1 assert vendor_1[0]['uuid'] == '87682992-bff2-40ff-aa75-18eb6214679e' assert vendor_1[0]['migrated_to_abacus'] is False assert vendor_1[1]['id'] == 1 assert vendor_1[1]['name'] == 'test_theorchard' if '6097ad5a-2186-4dbe-8286-82ce933d3538' in result: vendor_2 = result['6097ad5a-2186-4dbe-8286-82ce933d3538'] assert vendor_2[0]['vendor_id'] == 2 assert vendor_2[0]['uuid'] == '6097ad5a-2186-4dbe-8286-82ce933d3538' assert vendor_2[0]['migrated_to_abacus'] is False assert vendor_2[1]['id'] == 2 assert vendor_2[1]['name'] == 'test_awal' @pytest.mark.parametrize( ('payload', 'expected'), [ ( { 'name': 'foo', 'owner': 'bar', 'contact_email': 'abc@xyz.com', 'status': 'signed', }, { 'company': None, 'company_brand_id': None, 'contact_email': 'abc@xyz.com', 'country_id': None, 'first_statement_period': 1, 'relationship_notes': None, 'is_distributor': 'N', 'label_identifier': None, 'name': 'foo', 'owner': 'bar', 'status': 'signed', 'vendor_id': 25, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'migrated_to_abacus': False, 'primary_genre': None, 'priority': None, 'region': None, 'show_release_builder': None, 'support_contact_email': None, 'transfer_pricing_country': None, 'is_owned': 'No', 'vendor_uuid': None, 'last_modified_by': None, 'service_tier': None, 'wel_email_sender': None, }, ), ( { 'name': 'foo', 'owner': 'bar', 'contact_email': 'abc@xyz.com', 'status': 'signed', 'migrated_to_abacus': True, 'assigned_to': 1234, }, { 'company': None, 'company_brand_id': None, 'contact_email': 'abc@xyz.com', 'country_id': None, 'first_statement_period': 1, 'relationship_notes': None, 'is_distributor': 'N', 'label_identifier': None, 'name': 'foo', 'owner': 'bar', 'status': 'signed', 'vendor_id': 25, 'assigned_to': 1234, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'migrated_to_abacus': True, 'primary_genre': None, 'priority': None, 'region': None, 'show_release_builder': None, 'support_contact_email': None, 'transfer_pricing_country': None, 'is_owned': 'No', 'vendor_uuid': None, 'last_modified_by': None, 'service_tier': None, 'wel_email_sender': None, }, ), ], ) def test_create_vendor_without_service_tier_uuid(db_fixture, payload, expected): """Test create_vendor_with_optional_service_tier when no service tier arg is given.""" response = vendor.create_vendor_with_optional_service_tier(payload) assert response assert response.message == expected @patch('account.models.vendor.g') @patch('account.models.service_tier.ServiceTier.get_by_uuid', return_value=None) def test_create_vendor_with_invalid_service_tier_uuid( get_service_tier_mock, g_mock, db_fixture, app_context ): """Test create_vendor_with_optional_service_tier when an invalid service tier arg is given.""" attrs = {'name': '🐕', 'owner': '🦮', 'company_brand_id': 2} service_tier_uuid = 'not-a-good-uuid' response = vendor.create_vendor_with_optional_service_tier(attrs, service_tier_uuid) assert response.status == 400 get_service_tier_mock.assert_called_with(uuid=service_tier_uuid, session=ANY) g_mock.log.error.assert_called_with(f'Invalid service tier uuid {service_tier_uuid} given') def test_create_vendor_with_valid_service_tier_uuid(db_fixture, app_context): """Test create_vendor_with_optional_service_tier when a service tier is created.""" attrs = {'name': '🐕', 'owner': '🦮', 'company_brand_id': 2} service_tier_uuid = '1ed7aac0-ceb6-4c09-9166-afda8f349316' result = vendor.create_vendor_with_optional_service_tier(attrs, service_tier_uuid) assert result.message['service_tier']['uuid'] == service_tier_uuid # Not a pure unit test, but wanted to test successful passing of the sqlalchemy session with mysql.session_scope(read_only=True) as session: created_vendor = session.query(vendor.Vendor).get(result.message['vendor_id']) assert created_vendor.service_tier.uuid == service_tier_uuid def test_get_vendor_document_by_id(monkeypatch, fixture_vendor_document): """Test get vendor document by id.""" result_message = MagicMock(fetchone=lambda: fixture_vendor_document) monkeypatch.setattr(Session, 'execute', value=MagicMock(return_value=result_message)) response = vendor.get_vendor_document_by_id(2) assert response.status == 200 assert response.message == fixture_vendor_document Session.execute.assert_called_once_with(VENDOR_DOCUMENT_SQL, {'vendor_id': 2}) def test_get_vendor_document_by_id_with_tenant(monkeypatch, fixture_vendor_document): """Test get vendor document by id.""" result_message = MagicMock(fetchone=lambda: fixture_vendor_document) monkeypatch.setattr(Session, 'execute', value=MagicMock(return_value=result_message)) response = vendor.get_vendor_document_by_id(2, True) assert response.status == 200 assert response.message == fixture_vendor_document Session.execute.assert_called_once_with(VENDOR_DOCUMENT_WITH_TENANT_UUIDS_SQL, {'vendor_id': 2}) def test_get_vendor_document_by_id_for_no_results(monkeypatch): """Test get vendor document by id for no results.""" result = None result_message = MagicMock(fetchone=lambda: result) monkeypatch.setattr(Session, 'execute', value=MagicMock(return_value=result_message)) response = vendor.get_vendor_document_by_id(2) assert response.status == 404 @patch('account.connectors.mysql._account_session', side_effect=SQLAlchemyError()) @patch('account.models.vendor.g') def test_get_vendor_document_by_id_failure(session_scope, monkeypatch, app_context): """Test get vendor document by id when failed.""" monkeypatch.setattr(session_scope, 'execute', value=MagicMock(return_value=False)) response = vendor.get_vendor_document_by_id(2) assert response.status == 500 def test_get_all_vendor_currency_codes(monkeypatch, fixture_vendor_currency_codes): """Test get all vendor currency codes.""" result_message = MagicMock(fetchall=lambda: fixture_vendor_currency_codes) monkeypatch.setattr(Session, 'execute', value=MagicMock(return_value=result_message)) response = vendor.get_all_vendor_currency_codes() assert response.status == 200 assert response.message == { 'items': [ {'vendor_id': 123, 'currency_code': 'USD'}, {'vendor_id': 456, 'currency_code': 'EUR'}, ] } @patch('account.connectors.mysql._account_session', side_effect=SQLAlchemyError()) @patch('account.models.vendor.g') def test_get_all_vendor_currency_codes_failure(session_scope, monkeypatch, app_context): """Test get all vendor currency codes when failed.""" monkeypatch.setattr(session_scope, 'execute', value=MagicMock(return_value=False)) response = vendor.get_all_vendor_currency_codes() assert response.status == 500 def test_update_is_distributor_in_vendor(db_fixture): """Test update_vendor_is_distributor.""" vendor_id = 2 response = vendor.update_is_distributor_in_vendor(vendor_id) expected_result = { 'country_id': 1, 'owner': 'TEST2', 'name': 'Test', 'company': None, 'vendor_id': 2, 'is_distributor': 'Y', 'status': 'signed', 'label_identifier': 'D3', 'contact_email': None, 'migrated_to_abacus': False, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'last_modified_by': None, } assert response assert response.message == constants.VENDOR_UPDATED response = vendor.get_vendor(vendor_id) assert response.message == expected_result def test_update_is_distributor_in_vendor_already_distributor(db_fixture): """Test update_vendor_is_distributor.""" vendor_id = 1 response = vendor.update_is_distributor_in_vendor(vendor_id) expected_result = { 'country_id': 6, 'owner': 'TEST1', 'name': 'Test', 'company': None, 'vendor_id': 1, 'is_distributor': 'Y', 'status': 'signed', 'label_identifier': None, 'contact_email': None, 'migrated_to_abacus': False, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'vendor_uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'last_modified_by': None, } assert response assert response.message == constants.VENDOR_ALREADY_DISTRIBUTOR response = vendor.get_vendor(vendor_id) assert response.message == expected_result def test_update_is_distributor_in_vendor_no_vendor_found(monkeypatch): vendor_id = 100 not_found_response = response.create_not_found_response( 'No vendor found for vendor_id {}.'.format(vendor_id) ) monkeypatch.setattr( vendor, 'update_is_distributor_in_vendor', MagicMock(return_value=not_found_response), ) result = vendor.update_is_distributor_in_vendor(vendor_id) assert result.message == not_found_response.message assert result.status == 404 def test_get_vendor_by_external_identifier_not_exist(db_fixture): """Test get_vendor_by_external_identifier to return a empty list.""" result = vendor.get_vendors_by_external_identifier('non_existing', 'TEST1') expected_result = {'items': []} assert result assert result.message == expected_result def test_get_vendor_by_external_identifier(db_fixture): """Test get_vendor to return a list with vendor id.""" result = vendor.get_vendors_by_external_identifier('test_external_id', 'TEST2') expected_result = { 'items': [{'vendor_id': 2, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538'}] } assert result assert result.message == expected_result @pytest.mark.parametrize( 'fetch_flags', [ ([]), (['IGNORED']), ], ) @pytest.mark.parametrize( 'uuids, expected, description', [ ([], [], 'empty list returns empty list'), ( [ '87682992-bff2-40ff-aa75-18eb6214679e', '42d022c2-50e4-4f5c-af30-f10dca6b8a50', '9fe492d1-c9a0-4e62-b3ab-4d6413f36adb', ], [ { 'vendor_id': 1, 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', }, { 'vendor_id': 4, 'uuid': '42d022c2-50e4-4f5c-af30-f10dca6b8a50', }, ], 'only two items are returned; 9fe492d1-c9a0-4e62-b3ab-4d6413f36adb is not in db', ), ], ) @patch('account.models.vendor.g', spec=['request_context']) def test_lookup_vendors_by_uuids( mock_g, monkeypatch, db_fixture, fetch_flags: list[str], uuids: list[str], expected: list[dict[str, Any]], description: str, ) -> None: """Test lookup_vendors_by_uuids.""" mock_get_single_feature = MagicMock(return_value=response.Response(message='enabled')) monkeypatch.setattr( pythonfeatures, 'get_single_feature', mock_get_single_feature, ) result = vendor.lookup_vendors_by_uuids(uuids, fetch_flags) assert result.message == expected, description @pytest.mark.parametrize( 'fetch_flags', [ ([constants.FETCH_TENANT_HIERARCHY]), ([constants.FETCH_TENANT_HIERARCHY, 'IGNORED']), ], ) @pytest.mark.parametrize( 'uuids, expected, description', [ ([], [], 'empty list returns empty list'), ( [ '87682992-bff2-40ff-aa75-18eb6214679e', '42d022c2-50e4-4f5c-af30-f10dca6b8a50', '9fe492d1-c9a0-4e62-b3ab-4d6413f36adb', ], [ { 'vendor_id': 1, 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'company_brand_uuid': 'd25a4cd1-e820-45f2-be5c-56edcfeb8298', 'parent_company_uuid': '955a1bbd-b623-4ea1-ab5f-8d6620c442fb', }, { 'vendor_id': 4, 'uuid': '42d022c2-50e4-4f5c-af30-f10dca6b8a50', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', 'parent_company_uuid': '955a1bbd-b623-4ea1-ab5f-8d6620c442fb', }, ], 'only two items are returned; 9fe492d1-c9a0-4e62-b3ab-4d6413f36adb is not in db', ), ], ) @patch('account.models.vendor.g', spec=['request_context']) def test_lookup_vendors_and_hierarchy_by_uuids( mock_g, monkeypatch, db_fixture, fetch_flags: list[str], uuids: list[str], expected: list[dict[str, Any]], description: str, ) -> None: """Test lookup_vendors_by_uuids with tenant_hierarchy.""" mock_get_single_feature = MagicMock(return_value=response.Response(message='enabled')) monkeypatch.setattr( pythonfeatures, 'get_single_feature', mock_get_single_feature, ) result = vendor.lookup_vendors_by_uuids(uuids, fetch_flags) assert result.message == expected, description @pytest.mark.parametrize( 'fetch_flags', [ ([constants.FETCH_TENANT_HIERARCHY, constants.FETCH_IS_DISTRIBUTOR]), ([constants.FETCH_TENANT_HIERARCHY, constants.FETCH_IS_DISTRIBUTOR, 'IGNORED']), ], ) @pytest.mark.parametrize( 'uuids, expected, description', [ ([], [], 'empty list returns empty list'), ( [ '87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538', '9fe492d1-c9a0-4e62-b3ab-4d6413f36adb', ], [ { 'vendor_id': 1, 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'company_brand_uuid': 'd25a4cd1-e820-45f2-be5c-56edcfeb8298', 'parent_company_uuid': '955a1bbd-b623-4ea1-ab5f-8d6620c442fb', 'is_distributor': True, }, { 'vendor_id': 2, 'uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', 'parent_company_uuid': '955a1bbd-b623-4ea1-ab5f-8d6620c442fb', 'is_distributor': False, }, ], 'only two items are returned; 9fe492d1-c9a0-4e62-b3ab-4d6413f36adb is not in db', ), ], ) @patch('account.models.vendor.g', spec=['request_context']) def test_lookup_vendors_and_hierarchy_and_is_distributor_by_uuids( mock_g, db_fixture, fetch_flags: list[str], uuids: list[str], expected: list[dict[str, Any]], description: str, ) -> None: """Test lookup_vendors_by_uuids with tenant_hierarchy.""" result = vendor.lookup_vendors_by_uuids(uuids, fetch_flags) assert result.message == expected, description @pytest.mark.parametrize('fetch_flags', [([]), (['IGNORED'])]) @pytest.mark.parametrize( 'vendor_ids, expected, description', [ ([], [], 'empty list returns empty list'), ( [ 1, 4, 39, ], [ { 'vendor_id': 1, 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', }, { 'vendor_id': 4, 'uuid': '42d022c2-50e4-4f5c-af30-f10dca6b8a50', }, ], 'only two items are returned; 39 is not in db', ), ], ) @patch('account.models.vendor.g', spec=['request_context']) def test_lookup_vendors_by_vendor_ids( mock_g, db_fixture, fetch_flags: list[str], vendor_ids: list[str], expected: list[dict[str, Any]], description: str, ) -> None: """Test lookup_vendors_by_vendor_ids.""" result = vendor.lookup_vendors_by_vendor_ids(vendor_ids, fetch_flags) assert result.message == expected, description @pytest.mark.parametrize( 'fetch_flags', [ ([constants.FETCH_TENANT_HIERARCHY]), ([constants.FETCH_TENANT_HIERARCHY, 'IGNORED']), ], ) @pytest.mark.parametrize( 'vendor_ids, expected, description', [ ([], [], 'empty list returns empty list'), ( [1, 4, 80], [ { 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'company_brand_uuid': 'd25a4cd1-e820-45f2-be5c-56edcfeb8298', 'vendor_id': 1, 'parent_company_uuid': '955a1bbd-b623-4ea1-ab5f-8d6620c442fb', }, { 'uuid': '42d022c2-50e4-4f5c-af30-f10dca6b8a50', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', 'vendor_id': 4, 'parent_company_uuid': '955a1bbd-b623-4ea1-ab5f-8d6620c442fb', }, ], 'returns two items, vendor_id 80 is not found.', ), ], ) @patch('account.models.vendor.g', spec=['request_context']) def test_lookup_vendors_and_hierarchy_by_ids( mock_g, monkeypatch, db_fixture, fetch_flags: list[str], vendor_ids: list[str], expected: list[dict[str, Any]], description: str, ) -> None: """Test lookup_vendors_by_vendor_ids with tenant_hierarchy.""" monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response(message='enabled')), ) result = vendor.lookup_vendors_by_vendor_ids(vendor_ids, fetch_flags) assert result.message == expected, description @pytest.mark.parametrize( 'fetch_flags', [ ([constants.FETCH_TENANT_HIERARCHY, constants.FETCH_IS_DISTRIBUTOR]), ([constants.FETCH_TENANT_HIERARCHY, constants.FETCH_IS_DISTRIBUTOR, 'IGNORED']), ], ) @pytest.mark.parametrize( 'vendor_ids, expected, description', [ ([], [], 'empty list returns empty list'), ( [1, 2, 80], [ { 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'company_brand_uuid': 'd25a4cd1-e820-45f2-be5c-56edcfeb8298', 'parent_company_uuid': '955a1bbd-b623-4ea1-ab5f-8d6620c442fb', 'vendor_id': 1, 'is_distributor': True, }, { 'uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', 'parent_company_uuid': '955a1bbd-b623-4ea1-ab5f-8d6620c442fb', 'vendor_id': 2, 'is_distributor': False, }, ], 'returns two items, vendor_id 80 is not found.', ), ], ) @patch('account.models.vendor.g', spec=['request_context']) def test_lookup_vendors_and_hierarchy_and_is_distributor_by_ids( mock_g, db_fixture, fetch_flags: list[str], vendor_ids: list[str], expected: list[dict[str, Any]], description: str, ) -> None: """Test lookup_vendors_by_vendor_ids with tenant_hierarchy.""" result = vendor.lookup_vendors_by_vendor_ids(vendor_ids, fetch_flags) assert result.message == expected, description @pytest.mark.parametrize( 'uuid, expected', [ pytest.param( '87682992-bff2-40ff-aa75-18eb6214679e', ( { 'vendor_id': 1, 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'migrated_to_abacus': False, 'owner': 'TEST1', }, {'id': 1, 'name': 'test_theorchard'}, ), id='Valid UUID', ), pytest.param('not-a-uuid', None, id='Invalid UUID'), ], ) def test_lookup_vendor_and_company_brand_by_uuid( db_fixture, uuid: str, expected: dict[str, Any] ) -> None: """Test lookup_vendor_and_company_brand_by_uuid.""" with mysql.session_scope(read_only=True) as session: result = vendor.lookup_vendor_and_company_brand_by_uuid(uuid, session) assert result == expected @pytest.mark.parametrize( ('payload', 'expected'), [ ( { 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'label_identifier': 'Catalog', 'name': 'TEST2 NEW', 'company': 'TEST2 NEW', 'last_modified_by': 1111, }, { 'country_id': 1, 'owner': 'TEST2', 'name': 'TEST2 NEW', 'company': 'TEST2 NEW', 'vendor_id': 2, 'is_distributor': 'N', 'status': 'signed', 'label_identifier': 'Catalog', 'contact_email': None, 'migrated_to_abacus': False, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'last_modified_by': 1111, }, ), ( { 'label_identifier': 'Catalog', 'name': 'TEST2 NEW', 'company': 'TEST2 NEW', 'migrated_to_abacus': True, 'assigned_to': 1234, 'last_modified_by': 1111, }, { 'country_id': 1, 'owner': 'TEST2', 'name': 'TEST2 NEW', 'company': 'TEST2 NEW', 'vendor_id': 2, 'is_distributor': 'N', 'status': 'signed', 'label_identifier': 'Catalog', 'contact_email': None, 'migrated_to_abacus': True, 'assigned_to': 1234, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'last_modified_by': 1111, }, ), ], ) def test_update_vendor_by_uuid_v2_success(db_fixture, payload, expected) -> None: """update_vendor_by_uuid persists details and returns a plain dict on success.""" vendor_uuid = '6097ad5a-2186-4dbe-8286-82ce933d3538' result = vendor.update_vendor_by_uuid(vendor_uuid, payload) assert result == {'vendor_id': 2, 'vendor_uuid': vendor_uuid} persisted = vendor.get_vendor(2) assert persisted.message == expected def test_update_vendor_by_uuid_v2_will_update_external_identifier_1(db_fixture) -> None: """update_vendor_by_uuid persists external_identifier_1.""" external_identifier_1 = 'i am a parent rep owner' vendor_uuid = '6097ad5a-2186-4dbe-8286-82ce933d3538' payload = { 'vendor_id': 2, 'external_identifier_1': external_identifier_1, 'last_modified_by': 1211, } expected = { 'country_id': 1, 'owner': 'TEST2', 'name': 'Test', 'company': None, 'vendor_id': 2, 'is_distributor': 'N', 'status': 'signed', 'label_identifier': None, 'contact_email': None, 'migrated_to_abacus': False, 'assigned_to': None, 'assigned_reviewer': None, 'quarterback_label_manager': None, 'vendor_uuid': vendor_uuid, 'last_modified_by': 1211, } result = vendor.update_vendor_by_uuid(vendor_uuid, payload) assert result == {'vendor_id': 2, 'vendor_uuid': vendor_uuid} persisted = vendor.get_vendor(2) assert persisted.message == expected, 'Only the last_modified_by is updated' vendors_response = vendor.get_vendors_by_external_identifier(external_identifier_1, 'TEST2') assert vendors_response assert len(vendors_response.message['items']) == 1 assert vendors_response.message['items'][0] == { 'vendor_id': 2, 'vendor_uuid': vendor_uuid, } def test_update_vendor_by_uuid_v2_not_found(db_fixture) -> None: """update_vendor_by_uuid raises 404 VendorUpdateException when no row matches.""" with pytest.raises(VendorUpdateException) as exc_info: vendor.update_vendor_by_uuid( '8f9c46ca-0579-4efd-b914-65b930a68f49', {'name': 'nope'}, ) assert exc_info.value.status == 404 @patch('account.connectors.mysql._account_session', side_effect=SQLAlchemyError()) def test_update_vendor_by_uuid_v2_failure(session_scope, monkeypatch) -> None: """update_vendor_by_uuid raises 500 VendorUpdateException on DB error.""" monkeypatch.setattr(session_scope, 'commit', value=MagicMock(return_value=False)) with pytest.raises(VendorUpdateException) as exc_info: vendor.update_vendor_by_uuid( '6097ad5a-2186-4dbe-8286-82ce933d3538', {'name': 'Vendor Update'}, ) assert exc_info.value.status == 500 def test_delete_vendor_by_uuid_active(db_fixture) -> None: """Deleting an active vendor sets status='deletion' and last_modified_by.""" vendor_uuid = '6097ad5a-2186-4dbe-8286-82ce933d3538' last_modified_by = 7777 result = vendor.delete_vendor_by_uuid(vendor_uuid, last_modified_by) assert result == { 'vendor_id': 2, 'vendor_uuid': vendor_uuid, 'status': 'deletion', 'last_modified_by': last_modified_by, } refreshed = vendor.get_vendor(2).message assert refreshed['status'] == 'deletion' assert refreshed['last_modified_by'] == last_modified_by def test_delete_vendor_by_uuid_already_deleted(db_fixture) -> None: """Deleting a vendor already in 'deletion' status is idempotent and returns status=None.""" vendor_uuid = '6097ad5a-2186-4dbe-8286-82ce933d3538' vendor.update_vendor_by_uuid(vendor_uuid, {'status': 'deletion', 'last_modified_by': 1}) result = vendor.delete_vendor_by_uuid(vendor_uuid, 9999) assert result == { 'vendor_id': 2, 'vendor_uuid': vendor_uuid, 'status': None, 'last_modified_by': 9999, } refreshed = vendor.get_vendor(2).message assert refreshed['last_modified_by'] == 1, 'last_modified_by must not change on no-op delete' def test_delete_vendor_by_uuid_not_found(db_fixture) -> None: """Deleting a non-existent vendor UUID returns vendor_id=None without raising.""" vendor_uuid = '00000000-0000-0000-0000-000000000000' result = vendor.delete_vendor_by_uuid(vendor_uuid, 9999) assert result == { 'vendor_id': None, 'vendor_uuid': vendor_uuid, 'status': None, 'last_modified_by': 9999, } def test_get_vendor_company_brands_valid_uuids(db_fixture, app_context): """Test get_vendor_company_brands with valid UUIDs.""" vendor_uuids = ['87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538'] result = vendor.get_vendor_company_brands(vendor_uuids) assert len(result) == 2 assert result[0] == { 'vendor_id': 1, 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'company_brand_id': 1, 'company_brand_uuid': 'd25a4cd1-e820-45f2-be5c-56edcfeb8298', } assert result[1] == { 'vendor_id': 2, 'uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'company_brand_id': 2, 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', } def test_get_vendor_company_brands_empty_list(db_fixture, app_context): """Test get_vendor_company_brands with empty list.""" result = vendor.get_vendor_company_brands([]) assert len(result) == 0 def test_get_vendor_company_brands_invalid_uuids(db_fixture, app_context): """Test get_vendor_company_brands with invalid UUIDs.""" result = vendor.get_vendor_company_brands(['invalid_uuid']) assert len(result) == 0 def test_get_vendor_company_brands_mixed_valid_invalid_uuids(db_fixture, app_context): """Test get_vendor_company_brands with mix of valid and invalid UUIDs.""" vendor_uuids = ['87682992-bff2-40ff-aa75-18eb6214679e', 'invalid_uuid'] result = vendor.get_vendor_company_brands(vendor_uuids) assert len(result) == 1 assert result[0] == { 'vendor_id': 1, 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'company_brand_id': 1, 'company_brand_uuid': 'd25a4cd1-e820-45f2-be5c-56edcfeb8298', } @patch('account.models.vendor.g') @patch('account.models.vendor.mysql.session_scope') def test_get_vendor_company_brands_database_error(mock_session_scope, mock_g, app_context): """Test get_vendor_company_brands when database error occurs.""" vendor_uuids = ['87682992-bff2-40ff-aa75-18eb6214679e'] # Make the context manager raise SQLAlchemyError on enter mock_session_scope.side_effect = SQLAlchemyError('Database error') with pytest.raises(Exception) as exc_info: vendor.get_vendor_company_brands(vendor_uuids) assert 'Database error' in str(exc_info.value) mock_g.log.exception.assert_called_once_with(('Database error',)) def test_get_vendor_service_tier_valid_uuids(db_fixture, app_context): """Test get_vendor_service_tier with valid UUIDs.""" vendor_uuids = ['87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538'] result = vendor.get_vendor_service_tier(vendor_uuids) assert len(result) == 2 assert result[0] == { 'vendor_id': 1, 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'service_tier_uuid': '1ed7aac0-ceb6-4c09-9166-afda8f349316', 'service_tier_name': 'tier-1', 'service_tier_display_name': 'Tier 1', } assert result[1] == { 'vendor_id': 2, 'uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'service_tier_uuid': '2ed7aac0-ceb6-4c09-9166-afda8f349316', 'service_tier_name': 'tier-2', 'service_tier_display_name': 'Tier 2', } def test_get_vendor_service_tier_empty_list(db_fixture, app_context): """Test get_vendor_service_tier with empty list.""" result = vendor.get_vendor_service_tier([]) assert len(result) == 0 def test_get_vendor_service_tier_invalid_uuids(db_fixture, app_context): """Test get_vendor_service_tier with invalid UUIDs.""" result = vendor.get_vendor_service_tier(['invalid_uuid']) assert len(result) == 0 def test_get_vendor_service_tier_mixed_valid_invalid_uuids(db_fixture, app_context): """Test get_vendor_service_tier with mix of valid and invalid UUIDs.""" vendor_uuids = ['87682992-bff2-40ff-aa75-18eb6214679e', 'invalid_uuid'] result = vendor.get_vendor_service_tier(vendor_uuids) assert len(result) == 1 assert result[0] == { 'vendor_id': 1, 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'service_tier_uuid': '1ed7aac0-ceb6-4c09-9166-afda8f349316', 'service_tier_name': 'tier-1', 'service_tier_display_name': 'Tier 1', } @patch('account.models.vendor.g') @patch('account.models.vendor.mysql.session_scope') def test_get_vendor_service_tier_database_error(mock_session_scope, mock_g, app_context): """Test get_vendor_service_tier when database error occurs.""" vendor_uuids = ['87682992-bff2-40ff-aa75-18eb6214679e'] # Make the context manager raise SQLAlchemyError on enter mock_session_scope.side_effect = SQLAlchemyError('Database error') with pytest.raises(Exception) as exc_info: vendor.get_vendor_service_tier(vendor_uuids) assert 'Database error' in str(exc_info.value) mock_g.log.exception.assert_called_once_with(('Database error',)) @pytest.mark.parametrize( 'vendor_uuids, expected_len, expected_result', [ pytest.param( [ '87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538', ], 2, [ { 'vendor_id': 1, 'vendor_uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'relationship_notes': 'Note A', }, { 'vendor_id': 2, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'relationship_notes': 'Note B', }, ], id='valid vendor uuids', ), pytest.param([], 0, [], id='empty uuid list'), pytest.param(['invalid_uuid'], 0, [], id='invalid uuid'), pytest.param( [ '87682992-bff2-40ff-aa75-18eb6214679e', 'invalid_uuid', ], 1, [ { 'vendor_id': 1, 'vendor_uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'relationship_notes': 'Note A', } ], id='mixed valid + invalid uuid', ), ], ) def test_get_relationship_notes( vendor_uuids, expected_len, expected_result, db_fixture, app_context ): """ Test fetching relationship notes for vendors using UUIDs. """ result = vendor.get_relationship_notes(vendor_uuids) assert len(result) == expected_len assert result == expected_result