"""Tests for vendor handlers (vendor.py).""" import json from unittest.mock import MagicMock, patch from uuid import UUID import pytest from flask.app import AppContext from flask.testing import FlaskClient from owsrequest import flask_request from owsresponse import response from account.constants import error from account.logic import feature, sme, subaccount, supplier, vendor from account.models import neo4j_vendor from account.utils import pagination from tests.unit import db_operations from .conftest import fast_patch @pytest.mark.parametrize('url', [('/distributor'), ('/{0}/distributor'.format('any'))]) def test_is_distributor(monkeypatch, fixture_client, url, fixture_ok_response): """Test route that checks if vendor is a distributor succeeds.""" fast_patch( monkeypatch, { flask_request: dict(verify_grass_access=fixture_ok_response), vendor: dict(is_distributor=fixture_ok_response), }, ) result = fixture_client.head(url) assert result.status_code == 200 assert result.headers.get('Correlation-Id') assert not result.data def test_is_distributor_fails_for_subaccount( monkeypatch, fixture_client, fixture_ok_response, fixture_grass_subaccount_account ): """Test route that checks if subaccount user is a distributor fails.""" fast_patch( monkeypatch, { flask_request: dict( get_grass_headers=fixture_grass_subaccount_account, verify_grass_access=fixture_ok_response, ) }, ) result = fixture_client.head('/distributor') assert result.status_code == 204 assert result.headers.get('Correlation-Id') assert not result.data @pytest.mark.parametrize( 'url, method', [ ('/subaccounts', 'get'), ('/{0}/subaccounts'.format('any'), 'get'), ('/subaccount/{0}'.format('any'), 'get'), ('/{0}/subaccount/{1}'.format('any', 'thing'), 'head'), ('/distributor', 'head'), ('/{0}/distributor'.format('any'), 'head'), ('/sony/vendor/{0}'.format('any'), 'head'), ('/sony/subaccount/{0}'.format('any'), 'head'), ], ) def test_validation_failure( monkeypatch, fixture_client, url, method, fixture_grass_account, fixture_pagination, fixture_error_response, ): """Test routes where validation fails.""" fast_patch( monkeypatch, { pagination: dict(get_pagination=fixture_pagination), flask_request: dict( get_grass_headers=fixture_grass_account, verify_grass_access=fixture_error_response, ), }, ) test_method = getattr(fixture_client, method) result = test_method(url) assert result.status_code != 200 assert result.headers.get('Correlation-Id') @pytest.mark.parametrize( 'url', [('/sony/vendor/{0}'.format('any')), ('/sony/subaccount/{0}'.format('any'))] ) def test_is_sme_success( monkeypatch, fixture_client, url, fixture_ok_response, fixture_error_response ): """Test routes that check if account is SME succeeds.""" fast_patch( monkeypatch, { flask_request: dict(verify_grass_access=fixture_ok_response), sme: dict(is_sme_vendor=fixture_ok_response, is_sme_subaccount=fixture_ok_response), }, ) result = fixture_client.head(url) assert result.status_code == 200 assert result.headers.get('Correlation-Id') assert not result.data @pytest.mark.parametrize( 'url', [('/sony/vendor/{0}'.format('any')), ('/sony/subaccount/{0}'.format('any'))] ) def test_is_sme_failure( monkeypatch, fixture_client, url, fixture_ok_response, fixture_error_not_found_response, ): """Test routes where SME validation fails.""" fast_patch( monkeypatch, { flask_request: dict(verify_grass_access=fixture_ok_response), sme: dict( is_sme_vendor=fixture_error_not_found_response, is_sme_subaccount=fixture_error_not_found_response, ), }, ) result = fixture_client.head(url) assert result.status_code == 404 assert not result.data def test_is_valid_vendor(monkeypatch, fixture_client, fixture_ok_response): """Test route that checks if vendor_id is a valid vendor.""" monkeypatch.setattr(vendor, 'is_vendor', MagicMock(return_value=fixture_ok_response)) result = fixture_client.head('/vendor/1') assert result.status_code == 200 def test_is_valid_vendor_invalid_id(monkeypatch, fixture_client, fixture_error_not_found_response): """Test route that checks if vendor_id is invalid.""" monkeypatch.setattr( vendor, 'is_vendor', MagicMock(return_value=fixture_error_not_found_response) ) result = fixture_client.head('/vendor/100') assert result.status_code == 404 @pytest.mark.parametrize('url', [('/vendor/{}/suppliers'.format('1'))]) def test_get_suppliers(monkeypatch, fixture_client, url): """Test routes that supplier selections.""" success_payload = [{'vendor_id': 1, 'store_id': 1}, {'vendor_id': 1, 'store_id': 2}] monkeypatch.setattr( supplier, 'get_suppliers', MagicMock(return_value=response.Response(success_payload)), ) result = fixture_client.get(url) assert result.status_code == 200 assert json.loads(result.get_data(as_text=True)) == success_payload @pytest.mark.parametrize('url', [('/vendor/{}/suppliers'.format('5'))]) def test_get_suppliers_not_found( monkeypatch, fixture_client, url, fixture_error_not_found_response ): """Test routes that returns supplier selections.""" monkeypatch.setattr( supplier, 'get_suppliers', MagicMock(return_value=fixture_error_not_found_response), ) result = fixture_client.get(url) assert result.status_code == 404 @pytest.mark.parametrize('url', [('/vendor/{}/suppliers'.format(1))]) def test_set_suppliers(monkeypatch, fixture_client, url): """Test updating suppliers list.""" payload = {'store_ids': [1, 2, 3]} success_response = {'inserted_rows': 3} supplier_mock = MagicMock(return_value=response.Response(message=success_response, status=201)) monkeypatch.setattr(supplier, 'set_suppliers', supplier_mock) result = fixture_client.put(url, data=json.dumps(payload)) supplier_mock.assert_called_with(1, payload.get('store_ids')) assert result.status_code == 201 assert json.loads(result.get_data(as_text=True)) == success_response @pytest.mark.parametrize('url', [('/vendor/{}/suppliers'.format(1))]) def test_set_suppliers_with_null(monkeypatch, fixture_client, url): """Test updating suppliers list with none for store_ids value.""" payload = {'store_ids': None} success_response = {'inserted_rows': 0} supplier_mock = MagicMock(return_value=response.Response(message=success_response, status=201)) monkeypatch.setattr(supplier, 'set_suppliers', supplier_mock) result = fixture_client.put(url, data=json.dumps(payload)) supplier_mock.assert_called_with(1, []) assert result.status_code == 201 assert json.loads(result.get_data(as_text=True)) == success_response @pytest.mark.parametrize('url', [('/vendor/{}/suppliers'.format(5))]) def test_set_suppliers_empty(monkeypatch, fixture_client, url): """Test updating suppliers list with no store ids.""" payload = [] success_response = {'inserted_rows': 0} supplier_mock = MagicMock(return_value=response.Response(message=success_response, status=201)) monkeypatch.setattr(supplier, 'set_suppliers', supplier_mock) result = fixture_client.put(url, data=json.dumps(payload)) supplier_mock.assert_called_with(5, payload) assert result.status_code == 201 assert json.loads(result.get_data(as_text=True)) == success_response def test_get_vendor_closers(monkeypatch, fixture_client): """Test routes that returns vendor closers.""" expected = [ {'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'closers': [1, 2]}, {'uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'closers': [3, 4]}, ] mock_get_vendor_closers = MagicMock(return_value=expected) monkeypatch.setattr( vendor, 'get_vendor_closers', mock_get_vendor_closers, ) result = fixture_client.post( '/vendors/closers/dataloader', json={ 'vendor_uuids': [ '87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538', ] }, ) assert result.status_code == 200 assert json.loads(result.data) == expected mock_get_vendor_closers.assert_called_once_with( [UUID('87682992-bff2-40ff-aa75-18eb6214679e'), UUID('6097ad5a-2186-4dbe-8286-82ce933d3538')] ) @pytest.mark.parametrize( 'input_data, expected_status_code, expected_error_message', [ pytest.param( {'vendor_uuids': []}, 400, {'vendor_uuids': ['Shorter than minimum length 1.']}, id='Empty vendor_uuids', ), pytest.param( {'other_field': [1, 2]}, 400, { 'vendor_uuids': ['Missing data for required field.'], 'other_field': ['Unknown field.'], }, id='Missing vendor_uuids field', ), pytest.param( {'vendor_uuids': ['1', '2']}, 400, {'vendor_uuids': {'0': ['Not a valid UUID.'], '1': ['Not a valid UUID.']}}, id='Invalid vendor_uuids (e.g., not uuid)', ), pytest.param( {'vendor_uuids': '1,2'}, 400, {'vendor_uuids': ['Not a valid list.']}, id='Non-list vendor_uuids', ), pytest.param( [], 400, {'vendor_uuids': ['Missing data for required field.']}, id='Empty list' ), ], ) def test_get_vendor_closers_invalid_input( monkeypatch, fixture_client, input_data, expected_status_code, expected_error_message ): """Test routes vendor closers for invalid input data""" expected = { 'code': 'input_validation_error', 'message': expected_error_message, } mock_get_vendor_closers = MagicMock() monkeypatch.setattr( vendor, 'get_vendor_closers', mock_get_vendor_closers, ) result = fixture_client.post( '/vendors/closers/dataloader', json=input_data, ) assert result.status_code == expected_status_code assert json.loads(result.data) == expected assert mock_get_vendor_closers.call_count == 0 def test_get_vendors_first_statement_period(monkeypatch, fixture_client): """Test get_vendors_first_statement_period handler""" expected = [ {'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'first_statement_period': 1}, {'uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'first_statement_period': 1}, ] mock_get_vendor_first_statement_periods = MagicMock(return_value=expected) monkeypatch.setattr( vendor, 'get_vendors_first_statement_period', mock_get_vendor_first_statement_periods, ) monkeypatch.setattr( flask_request, 'verify_rules_access_standalone', value=MagicMock(return_value=True), ) result = fixture_client.post( '/vendors/first_statement_period/dataloader', json={ 'vendor_uuids': [ '87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538', ] }, ) assert flask_request.verify_rules_access_standalone.called assert result.status_code == 200 assert json.loads(result.data) == expected mock_get_vendor_first_statement_periods.assert_called_once_with( [UUID('87682992-bff2-40ff-aa75-18eb6214679e'), UUID('6097ad5a-2186-4dbe-8286-82ce933d3538')] ) @pytest.mark.parametrize( 'input_data, expected_status_code, expected_error_message', [ pytest.param( {'vendor_uuids': []}, 400, {'vendor_uuids': ['Shorter than minimum length 1.']}, id='Empty vendor_uuids', ), pytest.param( {'other_field': [1, 2]}, 400, { 'vendor_uuids': ['Missing data for required field.'], 'other_field': ['Unknown field.'], }, id='Missing vendor_uuids field', ), pytest.param( {'vendor_uuids': ['1', '2']}, 400, {'vendor_uuids': {'0': ['Not a valid UUID.'], '1': ['Not a valid UUID.']}}, id='Invalid vendor_uuids (e.g., not uuid)', ), pytest.param( {'vendor_uuids': '1,2'}, 400, {'vendor_uuids': ['Not a valid list.']}, id='Non-list vendor_uuids', ), pytest.param( [], 400, {'vendor_uuids': ['Missing data for required field.']}, id='Empty list' ), ], ) def test_get_vendors_first_statement_period_invalid_input( monkeypatch, fixture_client, input_data, expected_status_code, expected_error_message ): """Test get_vendors_first_statement_period for invalid input data""" expected = { 'code': 'input_validation_error', 'message': expected_error_message, } mock_get_vendor_first_statement_period = MagicMock() monkeypatch.setattr( vendor, 'get_vendors_first_statement_period', mock_get_vendor_first_statement_period, ) result = fixture_client.post( '/vendors/first_statement_period/dataloader', json=input_data, ) assert result.status_code == expected_status_code assert json.loads(result.data) == expected assert mock_get_vendor_first_statement_period.call_count == 0 @pytest.mark.parametrize( ('feature_enabled_responses'), [ { 'use_managed_tx': response.Response(message='enabled'), }, { 'use_managed_tx': response.Response(message='disabled'), }, { 'use_managed_tx': response.Response(message='disabled'), }, { 'use_managed_tx': response.Response(message='enabled'), }, ], ) @patch('account.handlers.vendor.g') @patch('connector_neo4j.Neo4jSession') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') @patch('account.handlers.vendor.pythonfeatures.get_single_feature') def test_get_vendor( mock_feature, mock_session, neo4j_enter, neo4j_exit, mock_g, feature_enabled_responses, monkeypatch, fixture_client, fixture_valid_vendor, app_context, ): """Test get vendor basic information.""" monkeypatch.setattr( vendor, 'get_vendor_label_info', MagicMock(return_value=fixture_valid_vendor) ) # Set side effect to match order of calls to get_single_feature mock_feature.side_effect = [ feature_enabled_responses['use_managed_tx'], ] mock_profile_has_access = MagicMock(return_value=True) mock_profile_has_access_tx = MagicMock(return_value=True) monkeypatch.setattr(vendor, 'profile_has_access', mock_profile_has_access) monkeypatch.setattr(vendor, 'profile_has_access_tx', mock_profile_has_access_tx) result = fixture_client.get('/vendor/1') vendor_payload = { 'vendor_id': 1, 'is_distributor': 'Y', 'name': 'Funky Vendor', 'country_id': 1, 'owner': 'odd', } assert result.status_code == 200 assert json.loads(result.get_data(as_text=True)) == vendor_payload if feature_enabled_responses['use_managed_tx'].message == 'enabled': mock_profile_has_access_tx.assert_called_once() mock_profile_has_access.assert_not_called() elif feature_enabled_responses['use_managed_tx'].message == 'disabled': mock_profile_has_access.assert_called_once() mock_profile_has_access_tx.assert_not_called() if feature_enabled_responses['use_managed_tx'].message == 'enabled': vendor.get_vendor_label_info.assert_called_once_with('1', neo_tx=True) else: vendor.get_vendor_label_info.assert_called_once_with('1') @pytest.mark.parametrize( ('feature_enabled_responses'), [ { 'use_managed_tx': response.Response(message='enabled'), }, { 'use_managed_tx': response.Response(message='disabled'), }, ], ) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.AccountByIdResourceGetter') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') @patch('account.handlers.vendor.pythonfeatures.get_single_feature') def test_get_vendor_with_pp_auth_check( mock_feature, _, __, mock_resource_getter, mock_authorization_backend, feature_enabled_responses, monkeypatch, fixture_client, fixture_valid_vendor, ): """Test get vendor basic information with pp auth check.""" fast_patch( monkeypatch, { flask_request: dict(verify_rules_access_standalone=False), vendor: dict(get_vendor_label_info=fixture_valid_vendor), }, ) mock_authorization_backend.is_authorized.return_value = True # Set side effect to match order of calls to get_single_feature mock_feature.side_effect = [ feature_enabled_responses['use_managed_tx'], ] result = fixture_client.get('/vendor/1') vendor_payload = { 'vendor_id': 1, 'is_distributor': 'Y', 'name': 'Funky Vendor', 'country_id': 1, 'owner': 'odd', } assert result.status_code == 200 assert json.loads(result.get_data(as_text=True)) == vendor_payload mock_authorization_backend.is_authorized.assert_called_once_with( action='view', resource_id='1', resource_type='account', resource_getter=mock_resource_getter(1), ) if feature_enabled_responses['use_managed_tx'].message == 'enabled': vendor.get_vendor_label_info.assert_called_once_with('1', neo_tx=True) else: vendor.get_vendor_label_info.assert_called_once_with('1') @patch('connector_neo4j.Neo4jSession') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_vendor_subaccount( mock_session, neo4j_enter, neo4j_exit, monkeypatch, fixture_client, fixture_grass_subaccount_account, fixture_valid_vendor, fixture_subaccount, ): """Test getting vendor info by one of its subaccounts.""" fast_patch( monkeypatch, { flask_request: dict(get_grass_headers=fixture_grass_subaccount_account), subaccount: dict(is_subaccount_for_vendor=fixture_subaccount), vendor: dict(get_vendor_label_info=fixture_valid_vendor), }, ) result = fixture_client.get('/vendor/1') vendor_payload = { 'vendor_id': 1, 'is_distributor': 'Y', 'name': 'Funky Vendor', 'country_id': 1, 'owner': 'odd', } assert result.status_code == 200 assert json.loads(result.get_data(as_text=True)) == vendor_payload @patch('connector_neo4j.Neo4jSession') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_vendor_not_found( mock_session, neo4j_enter, neo4j_exit, monkeypatch, fixture_client, fixture_error_not_found_response, ): """Test get vendor basic information.""" monkeypatch.setattr( vendor, 'get_vendor_label_info', MagicMock(return_value=fixture_error_not_found_response), ) result = fixture_client.get('/vendor/1') assert result.status_code == 404 @patch('connector_neo4j.Neo4jSession') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_vendor_access_denied( mock_session, neo4j_enter, neo4j_exit, monkeypatch, fixture_client, fixture_grass_vendor_account, ): """Test get vendor info access denied for another vendor.""" fast_patch( monkeypatch, {flask_request: dict(get_grass_headers=fixture_grass_vendor_account)}, ) result = fixture_client.get('/vendor/222') assert result.status_code == 403 @patch('connector_neo4j.Neo4jSession') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_vendor_with_oa( mock_session, neo4j_enter, neo4j_exit, monkeypatch, fixture_client, fixture_valid_vendor, fixture_oa_user_header, ): """Test get vendor with oa grass header bypasses ownership check""" monkeypatch.setattr( vendor, 'get_vendor_label_info', MagicMock(return_value=fixture_valid_vendor) ) result = fixture_client.get('/vendor/1') vendor_payload = { 'vendor_id': 1, 'is_distributor': 'Y', 'name': 'Funky Vendor', 'country_id': 1, 'owner': 'odd', } fixture_client.get('/vendor/1', headers=fixture_oa_user_header) assert result.status_code == 200 assert json.loads(result.data) == vendor_payload def test_update_vendor_closers(fixture_oa_user_header, monkeypatch, fixture_client): """Test update vendor closers.""" vendor_uuid = '87682992-bff2-40ff-aa75-18eb6214679e' expected = {'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'vendor_id': 1, 'closers': [1, 2]} mock_update_vendor_closers = MagicMock(return_value=expected) monkeypatch.setattr( vendor, 'update_vendor_closers', mock_update_vendor_closers, ) result = fixture_client.put( f'/vendor/{vendor_uuid}/closers', json={'closers': [1, 2]}, headers=fixture_oa_user_header ) assert result.status_code == 200 assert json.loads(result.data) == expected mock_update_vendor_closers.assert_called_once_with(vendor_uuid, [1, 2], 100) def test_update_vendor_notes(monkeypatch, fixture_client): """Test update vendor notes successfully.""" vendor_uuid = '87682992-bff2-40ff-aa75-18eb6214679e' expected = { 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'vendor_id': 1, 'relationship_notes': 'These are some vendor notes.', } monkeypatch.setattr( flask_request, 'verify_rules_access_standalone', value=MagicMock(return_value=True), ) mock_update_vendor_notes = MagicMock(return_value=response.Response(expected)) monkeypatch.setattr( vendor, 'update_vendor_notes', mock_update_vendor_notes, ) result = fixture_client.patch( f'/vendor/{vendor_uuid}/relationship_notes', json={'relationship_notes': 'These are some vendor notes.'}, ) assert flask_request.verify_rules_access_standalone.called assert result.status_code == 200 assert json.loads(result.data) == expected mock_update_vendor_notes.assert_called_once_with(vendor_uuid, 'These are some vendor notes.') def test_get_vendor_document(monkeypatch, fixture_client, fixture_vendor_document): """Test route that gets a specific vendor document.""" vendor_id = fixture_vendor_document['label_id'] fast_patch( monkeypatch, {vendor: dict(get_vendor_document=response.Response(fixture_vendor_document))}, ) result = fixture_client.get('/vendor/{0}/document'.format(vendor_id)) vendor.get_vendor_document.assert_called_with(vendor_id, False) assert result.status_code == 200 assert json.loads(result.data.decode('utf-8')) == fixture_vendor_document def test_get_vendor_document_with_tenant_uuids( monkeypatch, fixture_client, fixture_vendor_document ): """Test route that gets a specific vendor document.""" vendor_id = fixture_vendor_document['label_id'] fast_patch( monkeypatch, {vendor: dict(get_vendor_document=response.Response(fixture_vendor_document))}, ) result = fixture_client.get('/vendor/{0}/document?with_tenant_uuids=1'.format(vendor_id)) vendor.get_vendor_document.assert_called_with(vendor_id, True) assert result.status_code == 200 assert json.loads(result.data.decode('utf-8')) == fixture_vendor_document def test_get_vendor_document_for_rejected_grass_access( monkeypatch, fixture_client, fixture_vendor_document ): """Test rejected GRASS access when fetching vendor document.""" from account.constants import header error_response = ( 'Direct access through ows-grass is blocked. ' 'Only non-ows-grass microservice-to-microservice' ' requests are allowed.' ) vendor_id = fixture_vendor_document['label_id'] result = fixture_client.get( '/vendor/{0}/document'.format(vendor_id), headers={ header.GRASS_ACCOUNT_TYPE: 'vendor', header.GRASS_ACCOUNT_ID: vendor_id, }, ) assert result.status_code == 400 assert result.data.decode('utf-8') == error_response @patch('connector_neo4j.Neo4jSession') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_vendor_service_details( mock_session, neo4j_enter, neo4j_exit, monkeypatch, db_fixture, headers, fixture_client, ): """Test vendor/{vendor_id}/service-details.""" mock_service_tier = {'uuid': '123', 'name': 'service'} fast_patch( monkeypatch, {vendor: dict(get_vendor_service_details=response.Response(mock_service_tier))}, ) result = fixture_client.get('/vendor/123/service-details', headers=headers) assert result.status_code == 200 assert json.loads(result.get_data(as_text=True)) == mock_service_tier def test_get_vendor_assigned_to(fixture_client): """Test route that gets a specific vendor's assigned to user data.""" db_operations.seed_vendor_assigned_to_data() result = fixture_client.get('/vendor/24/assigned-to') assert result.status_code == 200 assert json.loads(result.data.decode('utf-8')) == { 'id': 1, 'f_name': 'foo', 'l_name': 'bar', } def test_get_vendor_secondary_internal_contact(fixture_client): """Test route that gets a vendor's secondary internal contact user data.""" db_operations.seed_vendor_secondary_internal_contact_data() result = fixture_client.get('/vendor/14/secondary-internal-contact') assert result.status_code == 404 result = fixture_client.get('/vendor/13/secondary-internal-contact') assert result.status_code == 200 assert json.loads(result.data.decode('utf-8')) == { 'id': 4, 'f_name': 'baz', 'l_name': 'qux', } def test_set_vendor_distributor(monkeypatch, fixture_client, fixture_vendor): """Set is_distributor field from N to Y.""" monkeypatch.setattr( vendor, 'update_is_distributor_in_vendor', MagicMock(return_value=response.Response(fixture_vendor)), ) result = fixture_client.patch('/vendor/1/distributor') assert result.status_code == 200 def test_get_vendors_by_external_identifier(fixture_client): """Test route that gets a specific vendor's by external_identifier_1.""" db_operations.seed_vendor_table() result = fixture_client.get('/vendors/by-external-identifier-1/non-existing?owner=TEST2') assert result.status_code == 200 assert json.loads(result.data.decode('utf-8')) == {'items': []} result = fixture_client.get('/vendors/by-external-identifier-1/test_external_id?owner=TEST2') assert result.status_code == 200 assert json.loads(result.data.decode('utf-8')) == { 'items': [{'vendor_id': 2, 'vendor_uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538'}] } result = fixture_client.get('/vendors/by-external-identifier-1/test_external_id?owner=TEST') assert result.status_code == 200 assert json.loads(result.data.decode('utf-8')) == {'items': []} @pytest.mark.parametrize('url', [('/vendor/{}/features'.format('1'))]) def test_get_enabled_features_for_vendor(monkeypatch, fixture_client, url): """Test get enabled features route.""" success_payload = { 'items': [ {'feature_id': 1, 'feature_name': 'One'}, {'feature_id': 2, 'feature_name': 'Two'}, ] } monkeypatch.setattr( feature, 'get_enabled_features_for_vendor', MagicMock(return_value=response.Response(success_payload)), ) result = fixture_client.get(url) assert result.status_code == 200 assert json.loads(result.get_data(as_text=True)) == success_payload @pytest.mark.parametrize('url', [('/vendor/{}/features'.format('1'))]) def test_get_enabled_features_for_vendor_invalid_headers(monkeypatch, fixture_client, url): """Test get enabled features route when called with invalid headers.""" monkeypatch.setattr(flask_request, 'get_grass_headers', MagicMock(return_value=('vendor', 2))) result = fixture_client.get(url) assert result.status_code == 403 @pytest.mark.parametrize('url', [('/vendor/{}/restricted_features/add'.format('1'))]) def test_bulk_add_restricted_features_for_vendor(monkeypatch, fixture_client, url): """Test bulk add restricted features for a vendor.""" data = {'feature_ids': [1, 2]} success_payload = [ {'vendor_restricted_features_id': 1, 'vendor_id': 1, 'feature_id': 1}, {'vendor_restricted_features_id': 2, 'vendor_id': 1, 'feature_id': 2}, ] monkeypatch.setattr( feature, 'bulk_add_restricted_features_for_vendor', MagicMock(return_value=response.Response(success_payload)), ) result = fixture_client.post(url, json=data) assert result.status_code == 200 assert json.loads(result.get_data(as_text=True)) == success_payload @pytest.mark.parametrize('url', [('/vendor/{}/restricted_features/add'.format('1'))]) def test_bulk_add_restricted_features_for_vendor_empty_input(monkeypatch, fixture_client, url): """Test bulk add restricted features for a vendor with empty input.""" monkeypatch.setattr(feature, 'bulk_add_restricted_features_for_vendor', MagicMock()) result = fixture_client.post(url, json={}) feature.bulk_add_restricted_features_for_vendor.assert_not_called() assert result.status_code == 400 assert json.loads(result.get_data())['code'] == error.ERROR_CODE_VALIDATION_ERROR @pytest.mark.parametrize('url', [('/vendor/{}/restricted_features/add'.format('1'))]) def test_bulk_add_restricted_features_for_vendor_invalid_input(monkeypatch, fixture_client, url): """Test bulk add restricted features for a vendor with invalid input.""" data = {'feature_ids': [1, 2, 'test']} monkeypatch.setattr(feature, 'bulk_add_restricted_features_for_vendor', MagicMock()) result = fixture_client.post(url, json=data) feature.bulk_add_restricted_features_for_vendor.assert_not_called() assert result.status_code == 400 assert json.loads(result.get_data())['code'] == error.ERROR_CODE_VALIDATION_ERROR @pytest.mark.parametrize('url', [('/vendor/{}/restricted_features/remove'.format('1'))]) def test_bulk_remove_restricted_features_for_vendor(monkeypatch, fixture_client, url): """Test bulk remove restricted features for a vendor.""" data = {'feature_ids': [1, 2]} success_payload = [ {'vendor_restricted_features_id': 1, 'vendor_id': 1, 'feature_id': 1}, {'vendor_restricted_features_id': 2, 'vendor_id': 1, 'feature_id': 2}, ] monkeypatch.setattr( feature, 'bulk_remove_restricted_features_for_vendor', MagicMock(return_value=response.Response(success_payload)), ) result = fixture_client.post(url, json=data) feature.bulk_remove_restricted_features_for_vendor.assert_called() assert result.status_code == 200 assert json.loads(result.get_data(as_text=True)) == success_payload @pytest.mark.parametrize('url', [('/vendor/{}/restricted_features/remove'.format('1'))]) def test_bulk_remove_restricted_features_for_vendor_empty_input(monkeypatch, fixture_client, url): """Test bulk remove restricted features for a vendor with empty input.""" monkeypatch.setattr(feature, 'bulk_remove_restricted_features_for_vendor', MagicMock()) result = fixture_client.post(url, json={}) feature.bulk_remove_restricted_features_for_vendor.assert_not_called() assert result.status_code == 400 assert json.loads(result.get_data())['code'] == error.ERROR_CODE_VALIDATION_ERROR @pytest.mark.parametrize('url', [('/vendor/{}/restricted_features/remove'.format('1'))]) def test_bulk_remove_restricted_features_for_vendor_invalid_input(monkeypatch, fixture_client, url): """Test bulk remove restricted features for a vendor with invalid input.""" data = {'feature_ids': [1, 2, 'test']} monkeypatch.setattr(feature, 'bulk_remove_restricted_features_for_vendor', MagicMock()) result = fixture_client.post(url, json=data) feature.bulk_remove_restricted_features_for_vendor.assert_not_called() assert result.status_code == 400 assert json.loads(result.get_data())['code'] == error.ERROR_CODE_VALIDATION_ERROR def test_get_vendors_names_by_vendor_uuids( monkeypatch, fixture_client, app_context, ) -> None: """Test get vendors names by vendor_uuids.""" expected = { 'vendors': [ { 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'vendor_id': 1, 'name': 'Vendor 1', }, { 'uuid': '6097ad5a-2186-4dbe-8286-82ce933d3538', 'vendor_id': 2, 'name': 'Vendor 2', }, ] } mock_get_vendor_names = MagicMock(return_value=expected) monkeypatch.setattr(vendor, 'get_vendor_names', mock_get_vendor_names) result = fixture_client.post( '/vendors/names/dataloader', json=[ '87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538', ], ) mock_get_vendor_names.assert_called_once_with( [ '87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538', ] ) assert json.loads(result.data) == expected def test_get_vendors_names_by_vendor_uuids_exception( monkeypatch, fixture_client, app_context, ) -> None: """Test get vendor names by vendor_uuids handles exception.""" def mock_get_vendor_names(*args, **kwargs): raise Exception('Mocked exception in get_vendor_names') monkeypatch.setattr(vendor, 'get_vendor_names', mock_get_vendor_names) result = fixture_client.post( '/vendors/names/dataloader', json=[ '87682992-bff2-40ff-aa75-18eb6214679e', '6097ad5a-2186-4dbe-8286-82ce933d3538', ], ) assert json.loads(result.data) == { 'code': 'internal_error', 'message': ['Mocked exception in get_vendor_names'], } @pytest.mark.parametrize( ('headers', 'data', 'status', 'session_count'), [ # no headers - not allowed ( {}, { 'vendor_name': 'Blacktop 2', 'email': 'abc@xyz.com', 'owner': 'SME US Latin', 'label_identifier': 'Frontline', 'source': 'event.gdaApproval', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', }, 403, 0, # because header validation is before logic ), # grass headers - not allowed ( { 'account_id': 2, 'account_type': 'vendor', 'account_name': 'FatCat Records', }, { 'vendor_name': 'Blacktop 2', 'email': 'abc@xyz.com', 'owner': 'SME US Latin', 'label_identifier': 'Frontline', 'source': 'event.gdaApproval', 'service_tier_uuid': '7410e51c-90da-4092-99da-b5489f364fa8', 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', }, 403, 0, # because header validation is before logic ), # profile headers, data validation failed. ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', }, { 'vendor_name': 'Blacktop 2', 'owner': 'SME US Latin', 'label_identifier': 'Frontline', 'country': 'US', 'service_tier_uuid': '9363881e-1120-4495-ae09-657e13609e11', 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', }, 400, 0, # because post body validation returns before entering the fn. ), ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', }, { 'vendor_name': 'Blacktop 2', 'email': 'abc@xyz.com', 'company_brand': 'dummy', }, 400, 0, ), ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', }, { 'vendor_name': 'Blacktop 2', 'email': 'abc@xyz.com', 'company_brand': 'awal', 'source': 'event.gdaApproval', 'service_tier_uuid': '1dd92c83-25a3-4034-9a3d-f7c3f434f4ce', 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', }, 200, 1, ), ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', }, { 'vendor_name': 'Blacktop 2', 'email': 'abc@xyz.com', 'company_brand': 'awal', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'source': 'event.gdaApproval', 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', }, 200, 1, ), # profile headers, data validation success. ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', }, { 'vendor_name': 'Blacktop 2', 'email': 'abc@xyz.com', 'owner': 'SME US Latin', 'label_identifier': 'Frontline', 'source': 'event.gdaApproval', 'service_tier_uuid': '2304f272-7a12-40fc-ba48-959fff435223', 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', }, 200, 1, ), # profile headers, POST request body with new column migrated_to_abacus. ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', }, { 'vendor_name': 'Blacktop 2', 'email': 'abc@xyz.com', 'owner': 'SME US Latin', 'label_identifier': 'Frontline', 'migrated_to_abacus': True, 'source': 'event.gdaApproval', 'service_tier_uuid': '1ed7aac0-ceb6-4c09-9166-afda8f349316', 'vendor_uuid': 'd6455851-3f86-48d7-b102-8d006eb92655', }, 200, 1, ), ], ) @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_create_update_vendor( neo4j_exit, neo4j_enter, headers, data, status, session_count, monkeypatch, fixture_client, ): """Test create_update_vendor""" monkeypatch.setattr( vendor, 'create_or_update_vendor', MagicMock(return_value=response.Response('success')), ) result = fixture_client.patch('/vendor', json=data, headers=headers) assert result.status_code == status assert neo4j_enter.call_count == session_count @pytest.mark.parametrize( ('headers', 'data', 'status'), [ ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', 'Orchard-User-Id': 'oa:1111', }, { 'assigned_to_id': '1', 'assigned_reviewer_id': '2', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', 'contact_email': 'test2903@test.com', 'contact_name': 'Blacktop 2', 'country_id': 1, 'is_owned': 'No', 'label_identifier': 'Frontline', 'name': 'Blacktop 2', 'owner': 'odd', 'primary_genre_id': '2', 'priority': 3, 'product_manager_id': '1', 'quarterback_label_manager_id': '12', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'show_release_builder': 'N', 'support_contact_email': 'test_support2903@test.com', }, 200, ), ], ) @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_update_vendor_with_user_id_header( neo4j_exit, neo4j_enter, headers, data, status, monkeypatch, fixture_client, fixture_valid_update_vendor_with_last_updated_by, ): """Test v2 update_vendor.""" monkeypatch.setattr( vendor, 'update_vendor', MagicMock(return_value=fixture_valid_update_vendor_with_last_updated_by), ) result = fixture_client.patch('/vendor/1234', json=data, headers=headers) assert result.status_code == status @pytest.mark.parametrize( 'input_data, expected_status_code, expected_error_message', [ pytest.param( {'other_field': 'These are some vendor notes.'}, 400, { 'relationship_notes': ['Missing data for required field.'], 'other_field': ['Unknown field.'], }, id='Missing notes field', ), pytest.param( {'relationship_notes': 12345}, 400, {'relationship_notes': ['Not a valid string.']}, id='Invalid notes (e.g., not string)', ), pytest.param( [], 400, {'relationship_notes': ['Missing data for required field.']}, id='Empty list', ), ], ) def test_update_vendor_notes_invalid_input( monkeypatch, fixture_client, input_data, expected_status_code, expected_error_message ): """Test routes vendor notes for invalid input data""" vendor_uuid = '87682992-bff2-40ff-aa75-18eb6214679e' expected = { 'code': 'input_validation_error', 'message': expected_error_message, } mock_update_vendor_notes = MagicMock() monkeypatch.setattr( vendor, 'update_vendor_notes', mock_update_vendor_notes, ) result = fixture_client.patch( f'/vendor/{vendor_uuid}/relationship_notes', json=input_data, ) assert result.status_code == expected_status_code assert json.loads(result.data) == expected assert mock_update_vendor_notes.call_count == 0 @pytest.mark.parametrize( 'headers, expected_status_code, expected_response', [ pytest.param( { 'Orchard-Requestor-Service': 'graphql-account-test', 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', }, 403, {'code': 'authorization_error', 'message': 'Forbidden'}, id='Invalid LabelProfile headers', ), pytest.param( { 'Orchard-Requestor-Service': 'graphql-account-test', 'Orchard-Profile-Type': 'OrchAdminProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', 'Orchard-User-Id': 'test:1111', }, 401, { 'code': 'authorization_error', 'message': 'Unauthorized user with user_id = test:1111', }, id='Invalid orchard-user-id', ), ], ) def test_update_vendor_notes_invalid_headers( monkeypatch, fixture_client, headers, expected_status_code, expected_response ): """Test update vendor notes for invalid headers.""" vendor_uuid = '87682992-bff2-40ff-aa75-18eb6214679e' mock_update_vendor_notes = MagicMock() monkeypatch.setattr( vendor, 'update_vendor_notes', mock_update_vendor_notes, ) result = fixture_client.patch( f'/vendor/{vendor_uuid}/relationship_notes', json={'relationship_notes': 'These are some vendor notes.'}, headers=headers, ) assert result.status_code == expected_status_code assert json.loads(result.data) == expected_response assert mock_update_vendor_notes.call_count == 0 def test_update_vendor_first_statement_period(monkeypatch, fixture_client): """Test update vendor first statement period.""" vendor_uuid = '87682992-bff2-40ff-aa75-18eb6214679e' expected = { 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'vendor_id': 1, 'first_statement_period': '1', } monkeypatch.setattr( flask_request, 'verify_rules_access_standalone', value=MagicMock(return_value=True), ) mock_update_vendor_first_statement_period = MagicMock(return_value=response.Response(expected)) monkeypatch.setattr( vendor, 'update_vendor_first_statement_period', mock_update_vendor_first_statement_period, ) result = fixture_client.patch( f'/vendor/{vendor_uuid}/first_statement_period', json={'first_statement_period': 1} ) assert flask_request.verify_rules_access_standalone.called assert result.status_code == 200 assert json.loads(result.data) == expected mock_update_vendor_first_statement_period.assert_called_once_with(vendor_uuid, 1) @pytest.mark.parametrize( 'input_data, expected_status_code, expected_error_message', [ pytest.param( {'other_field': 1}, 400, { 'first_statement_period': ['Missing data for required field.'], 'other_field': ['Unknown field.'], }, id='Missing first_statement_period field', ), pytest.param( {'first_statement_period': 'invalid'}, 400, {'first_statement_period': ['Not a valid integer.']}, id='Invalid first_statement_period (e.g., not integer)', ), pytest.param( [], 400, {'first_statement_period': ['Missing data for required field.']}, id='Empty list', ), ], ) def test_update_vendor_first_statement_period_invalid_input( monkeypatch, fixture_client, input_data, expected_status_code, expected_error_message ): """Test routes vendor first statement period for invalid input data""" vendor_uuid = '87682992-bff2-40ff-aa75-18eb6214679e' expected = { 'code': 'input_validation_error', 'message': expected_error_message, } mock_update_vendor_first_statement_period = MagicMock() monkeypatch.setattr( vendor, 'update_vendor_first_statement_period', mock_update_vendor_first_statement_period, ) result = fixture_client.patch( f'/vendor/{vendor_uuid}/first_statement_period', json=input_data, ) assert result.status_code == expected_status_code assert json.loads(result.data) == expected assert mock_update_vendor_first_statement_period.call_count == 0 @pytest.mark.parametrize( 'input_data, expected_status_code, expected_error_message', [ pytest.param( {'closers': []}, 400, {'closers': ['Shorter than minimum length 1.']}, id='Empty closers', ), pytest.param( {'other_field': [1, 2]}, 400, { 'closers': ['Missing data for required field.'], 'other_field': ['Unknown field.'], }, id='Missing closers field', ), pytest.param( {'closers': ['invalid1', 'invalid2']}, 400, {'closers': {'0': ['Not a valid integer.'], '1': ['Not a valid integer.']}}, id='Invalid closers (e.g., not integer)', ), pytest.param( {'closers': '1,2'}, 400, {'closers': ['Not a valid list.']}, id='Non-list closers', ), pytest.param([], 400, {'closers': ['Missing data for required field.']}, id='Empty list'), ], ) def test_update_vendor_closers_invalid_input( monkeypatch, fixture_client, input_data, expected_status_code, expected_error_message ): """Test routes vendor closers for invalid input data""" vendor_uuid = '87682992-bff2-40ff-aa75-18eb6214679e' expected = { 'code': 'input_validation_error', 'message': expected_error_message, } mock_update_vendor_closers = MagicMock() monkeypatch.setattr( vendor, 'update_vendor_closers', mock_update_vendor_closers, ) result = fixture_client.put( f'/vendor/{vendor_uuid}/closers', json=input_data, ) assert result.status_code == expected_status_code assert json.loads(result.data) == expected assert mock_update_vendor_closers.call_count == 0 @pytest.mark.parametrize( 'headers, expected_status_code, expected_response', [ pytest.param( { 'Orchard-Requestor-Service': 'graphql-account-test', 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', }, 403, {'code': 'authorization_error', 'message': 'Forbidden'}, id='Invalid LabelProfile headers', ), pytest.param( { 'Orchard-Requestor-Service': 'graphql-account-test', 'Orchard-Profile-Type': 'OrchAdminProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', 'Orchard-User-Id': 'test:1111', }, 401, { 'code': 'authorization_error', 'message': 'Unauthorized user with user_id = test:1111', }, id='Invalid orchard-user-id', ), ], ) def test_update_vendor_closers_invalid_headers( monkeypatch, fixture_client, headers, expected_status_code, expected_response ): """Test routes vendor closers for invalid headers""" vendor_uuid = '87682992-bff2-40ff-aa75-18eb6214679e' mock_update_vendor_closers = MagicMock() monkeypatch.setattr( vendor, 'update_vendor_closers', mock_update_vendor_closers, ) result = fixture_client.put( f'/vendor/{vendor_uuid}/closers', json={'closers': [1, 2]}, headers=headers ) assert result.status_code == expected_status_code assert json.loads(result.data) == expected_response assert mock_update_vendor_closers.call_count == 0 @pytest.mark.parametrize( 'headers, expected_status_code, expected_response', [ pytest.param( { 'Orchard-Requestor-Service': 'graphql-account-test', 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', }, 403, {'code': 'authorization_error', 'message': 'Forbidden'}, id='Invalid LabelProfile headers', ), pytest.param( { 'Orchard-Requestor-Service': 'graphql-account-test', 'Orchard-Profile-Type': 'OrchAdminProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', 'Orchard-User-Id': 'test:1111', }, 401, { 'code': 'authorization_error', 'message': 'Unauthorized user with user_id = test:1111', }, id='Invalid orchard-user-id', ), ], ) def test_update_vendor_first_statement_period_invalid_headers( monkeypatch, fixture_client, headers, expected_status_code, expected_response ): """Test routes vendor first statement period for invalid headers""" vendor_uuid = '87682992-bff2-40ff-aa75-18eb6214679e' mock_update_vendor_first_statement_period = MagicMock() monkeypatch.setattr( vendor, 'update_vendor_first_statement_period', mock_update_vendor_first_statement_period, ) result = fixture_client.patch( f'/vendor/{vendor_uuid}/first_statement_period', json={'first_statement_period': 1}, headers=headers, ) assert result.status_code == expected_status_code assert json.loads(result.data) == expected_response assert mock_update_vendor_first_statement_period.call_count == 0 @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') def test_update_vendor_closers_invalid_forbidden( mock_identity_logic, mock_g, monkeypatch, fixture_client, app_context ): """Test routes vendor closers for invalid jwt identity id""" vendor_uuid = '87682992-bff2-40ff-aa75-18eb6214679e' mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = None mock_update_vendor_closers = MagicMock() monkeypatch.setattr( vendor, 'update_vendor_closers', mock_update_vendor_closers, ) expected_response = {'code': 'authorization_error', 'message': 'Forbidden'} result = fixture_client.put( f'/vendor/{vendor_uuid}/closers', json={'closers': [1, 2]}, ) assert result.status_code == 401 assert json.loads(result.data) == expected_response assert mock_update_vendor_closers.call_count == 0 @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_update_vendor_external_identifier_1( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """Test v2 update_vendor_external_identifier_1.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 1394 mock_vendor_logic.update_vendor_external_identifier_1.return_value = {'okie': 'dokie'} result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}/external-identifier-1', json={ 'external_identifier_1': 'sme parent rep owner identifier', }, ) assert result.status_code == 200, result.text assert json.loads(result.get_data()) == {'okie': 'dokie'}, result.text mock_authorization_backend.is_authorized.assert_called_once_with( action='update_external_identifier_1', resource_id=vendor_uuid, resource_type='account', resource_getter=mock_resource_getter(vendor_uuid), ) mock_identity_logic.get_oa_user_id.assert_called_once_with('some_identity_uuid') mock_vendor_logic.update_vendor_external_identifier_1.assert_called_once_with( vendor_uuid, 1394, 'sme parent rep owner identifier', ) @pytest.mark.parametrize( 'is_authorized,oa_user_id', [ pytest.param(False, 82, id='PDP did not authorize'), pytest.param(True, None, id='No OA User Id'), ], ) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') def test_update_vendor_external_identifier_1_forbidden( mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, is_authorized: bool, oa_user_id: int | None, fixture_client: MagicMock, app_context, ) -> None: """Test v2 update_vendor_external_identifier forbids.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = is_authorized mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = oa_user_id result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}/external-identifier-1', json={ 'external_identifier_1': 'sme parent rep owner identifier', }, ) assert result.status_code == 403, result.text mock_vendor_logic.update_vendor_external_identifier_1.assert_not_called() mock_authorization_backend.is_authorized.assert_called_once() if is_authorized: mock_identity_logic.get_oa_user_id.assert_called_once() else: mock_identity_logic.get_oa_user_id.assert_not_called() @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_update_vendor_country_id( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """Test v2 update_vendor_country_id.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 9876 mock_vendor_logic.update_vendor_country_id.return_value = {'okie': 'dokie'} result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}/country_id', json={ 'country_id': 1, }, ) assert result.status_code == 200, result.text assert json.loads(result.get_data()) == {'okie': 'dokie'}, result.text mock_authorization_backend.is_authorized.assert_called_once_with( action='update:country_id', resource_id=vendor_uuid, resource_type='account', resource_getter=mock_resource_getter(vendor_uuid), ) mock_identity_logic.get_oa_user_id.assert_called_once_with('some_identity_uuid') mock_vendor_logic.update_vendor_country_id.assert_called_once_with( vendor_uuid, 9876, 1, ) @pytest.mark.parametrize( 'is_authorized,oa_user_id', [ pytest.param(False, 82, id='PDP did not authorize'), pytest.param(True, None, id='No OA User Id'), ], ) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') def test_update_vendor_country_id_forbidden( mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, is_authorized: bool, oa_user_id: int | None, fixture_client: MagicMock, app_context, ) -> None: """Test v2 update_vendor_country_id forbids.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = is_authorized mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = oa_user_id result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}/country_id', json={ 'country_id': 5, }, ) assert result.status_code == 403, result.text mock_vendor_logic.update_vendor_external_identifier_1.assert_not_called() mock_authorization_backend.is_authorized.assert_called_once() if is_authorized: mock_identity_logic.get_oa_user_id.assert_called_once() else: mock_identity_logic.get_oa_user_id.assert_not_called() @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_update_vendor_external_identifier_1_not_found( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """VendorUpdateException from logic is translated to the matching error response.""" from account.utils.exception import VendorUpdateException vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 1394 mock_vendor_logic.update_vendor_external_identifier_1.side_effect = VendorUpdateException( code=error.ERROR_CODE_INVALID_INPUT, message='Vendor not found', status=404, ) result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}/external-identifier-1', json={'external_identifier_1': 'ext_id'}, ) assert result.status_code == 404, result.text @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_update_vendor_country_id_not_found( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """VendorUpdateException from logic is translated to the matching error response.""" from account.utils.exception import VendorUpdateException vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 9876 mock_vendor_logic.update_vendor_country_id.side_effect = VendorUpdateException( code=error.ERROR_CODE_INVALID_INPUT, message='Vendor not found', status=404, ) result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}/country_id', json={'country_id': 1}, ) assert result.status_code == 404, result.text SERVICE_TIER_UUID = '5f2bd4fc-df94-4f35-97d3-ef23f8573279' @patch('account.utils.api_utils.g', spec=['request_context']) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_update_vendor_service_tier( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, mock_g_api: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """Test v2 update_vendor_service_tier.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_g_api.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 9876 mock_vendor_logic.update_vendor_service_tier.return_value = {'okie': 'dokie'} result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}/service_tier', json={'service_tier_uuid': SERVICE_TIER_UUID}, ) assert result.status_code == 200, result.text assert json.loads(result.get_data()) == {'okie': 'dokie'}, result.text mock_authorization_backend.is_authorized.assert_called_once_with( action='update:service_tier', resource_id=vendor_uuid, resource_type='account', resource_getter=mock_resource_getter(vendor_uuid), ) mock_identity_logic.get_oa_user_id.assert_called_once_with('some_identity_uuid') mock_vendor_logic.update_vendor_service_tier.assert_called_once_with( vendor_uuid, 9876, SERVICE_TIER_UUID, ) @pytest.mark.parametrize( 'is_authorized,oa_user_id', [ pytest.param(False, 82, id='PDP did not authorize'), pytest.param(True, None, id='No OA User Id'), ], ) @patch('account.utils.api_utils.g', spec=['request_context']) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') def test_update_vendor_service_tier_forbidden( mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, mock_g_api: MagicMock, is_authorized: bool, oa_user_id: int | None, fixture_client: MagicMock, app_context, ) -> None: """Test v2 update_vendor_service_tier forbids.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = is_authorized mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_g_api.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = oa_user_id result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}/service_tier', json={'service_tier_uuid': SERVICE_TIER_UUID}, ) assert result.status_code == 403, result.text mock_vendor_logic.update_vendor_service_tier.assert_not_called() mock_authorization_backend.is_authorized.assert_called_once() if is_authorized: mock_identity_logic.get_oa_user_id.assert_called_once() else: mock_identity_logic.get_oa_user_id.assert_not_called() @patch('account.utils.api_utils.g', spec=['request_context']) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_update_vendor_service_tier_not_found( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, mock_g_api: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """VendorUpdateException from logic is translated to the matching error response.""" from account.utils.exception import VendorUpdateException vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_g_api.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 9876 mock_vendor_logic.update_vendor_service_tier.side_effect = VendorUpdateException( code=error.ERROR_CODE_INVALID_INPUT, message='Vendor not found', status=404, ) result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}/service_tier', json={'service_tier_uuid': SERVICE_TIER_UUID}, ) assert result.status_code == 404, result.text @patch('account.utils.api_utils.g', spec=['request_context']) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_update_vendor_info( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, mock_g_api: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """Test v2 update_vendor_info happy path.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_g_api.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 9876 mock_vendor_logic.update_vendor_info.return_value = {'okie': 'dokie'} result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}', json={ 'name': 'New Name', 'support_contact_email': None, 'newsletter': 'N', }, ) assert result.status_code == 200, result.text assert json.loads(result.get_data()) == {'okie': 'dokie'}, result.text mock_authorization_backend.is_authorized.assert_called_once_with( action='update:info', resource_id=vendor_uuid, resource_type='account', resource_getter=mock_resource_getter(vendor_uuid), ) mock_identity_logic.get_oa_user_id.assert_called_once_with('some_identity_uuid') mock_vendor_logic.update_vendor_info.assert_called_once_with( vendor_uuid, 9876, name='New Name', support_contact_email=None, newsletter='N', ) @pytest.mark.parametrize( 'is_authorized,oa_user_id', [ pytest.param(False, 82, id='PDP did not authorize'), pytest.param(True, None, id='No OA User Id'), ], ) @patch('account.utils.api_utils.g', spec=['request_context']) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') def test_update_vendor_info_forbidden( mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, mock_g_api: MagicMock, is_authorized: bool, oa_user_id: int | None, fixture_client: MagicMock, app_context, ) -> None: """Test v2 update_vendor_info forbids.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = is_authorized mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_g_api.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = oa_user_id result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}', json={'name': 'New Name'}, ) assert result.status_code == 403, result.text mock_vendor_logic.update_vendor_info.assert_not_called() mock_authorization_backend.is_authorized.assert_called_once() @patch('account.utils.api_utils.g', spec=['request_context']) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_update_vendor_info_not_found( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, mock_g_api: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """VendorUpdateException from logic is translated to the matching error response.""" from account.utils.exception import VendorUpdateException vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_g_api.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 9876 mock_vendor_logic.update_vendor_info.side_effect = VendorUpdateException( code=error.ERROR_CODE_INVALID_INPUT, message='Vendor not found', status=404, ) result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}', json={'name': 'X'}, ) assert result.status_code == 404, result.text @patch('account.utils.api_utils.g', spec=['request_context']) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_update_vendor_internal_staff( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, mock_g_api: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """Test v2 update_vendor_internal_staff.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_g_api.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 1394 mock_vendor_logic.update_vendor_internal_staff.return_value = {'okie': 'dokie'} result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}/internal-staff', json={ 'assigned_to': 10, 'assigned_reviewer': 20, 'quarterback_label_manager': 30, 'wel_email_sender': 40, 'product_manager': 50, }, ) assert result.status_code == 200, result.text assert json.loads(result.get_data()) == {'okie': 'dokie'}, result.text mock_authorization_backend.is_authorized.assert_called_once_with( action='update:internal_staff', resource_id=vendor_uuid, resource_type='account', resource_getter=mock_resource_getter(vendor_uuid), ) mock_identity_logic.get_oa_user_id.assert_called_once_with('some_identity_uuid') mock_vendor_logic.update_vendor_internal_staff.assert_called_once_with( vendor_uuid, 1394, assigned_to=10, assigned_reviewer=20, quarterback_label_manager=30, wel_email_sender=40, product_manager=50, ) @patch('account.utils.api_utils.g', spec=['request_context']) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_update_vendor_internal_staff_partial( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, mock_g_api: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """Omitted staff fields are not forwarded to the logic layer.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_g_api.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 1394 mock_vendor_logic.update_vendor_internal_staff.return_value = {'ok': True} result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}/internal-staff', json={'assigned_to': 10, 'quarterback_label_manager': None}, ) assert result.status_code == 200, result.text mock_vendor_logic.update_vendor_internal_staff.assert_called_once_with( vendor_uuid, 1394, assigned_to=10, quarterback_label_manager=None, ) @patch('account.utils.api_utils.g', spec=['request_context']) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_update_vendor_internal_staff_not_found( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, mock_g_api: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """VendorUpdateException from logic is translated to the matching error response.""" from account.utils.exception import VendorUpdateException vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_g_api.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 1394 mock_vendor_logic.update_vendor_internal_staff.side_effect = VendorUpdateException( code=error.ERROR_CODE_INVALID_INPUT, message='Vendor not found', status=404, ) result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}/internal-staff', json={'assigned_to': 10}, ) assert result.status_code == 404, result.text @pytest.mark.parametrize( 'is_authorized,oa_user_id', [ pytest.param(False, 82, id='PDP did not authorize'), pytest.param(True, None, id='No OA User Id'), ], ) @patch('account.utils.api_utils.g', spec=['request_context']) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') def test_update_vendor_internal_staff_forbidden( mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, mock_g_api: MagicMock, is_authorized: bool, oa_user_id: int | None, fixture_client: MagicMock, app_context, ) -> None: """Test v2 update_vendor_internal_staff forbids.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = is_authorized mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_g_api.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = oa_user_id result = fixture_client.patch( f'/v2/vendor/{vendor_uuid}/internal-staff', json={'assigned_to': 10}, ) assert result.status_code == 403, result.text mock_vendor_logic.update_vendor_internal_staff.assert_not_called() mock_authorization_backend.is_authorized.assert_called_once() if is_authorized: mock_identity_logic.get_oa_user_id.assert_called_once() else: mock_identity_logic.get_oa_user_id.assert_not_called() @pytest.mark.parametrize( ('headers', 'data', 'status', 'session_count'), [ # no headers - not allowed ( {}, { 'assigned_to_id': '1', 'assigned_reviewer_id': '2', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', 'contact_email': 'test2903@test.com', 'contact_name': 'Blacktop 2', 'country_id': 1, 'is_owned': 'No', 'label_identifier': 'Frontline', 'name': 'Blacktop 2', 'owner': 'odd', 'primary_genre_id': '2', 'priority': 3, 'product_manager_id': '1', 'quarterback_label_manager_id': '12', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'show_release_builder': 'N', 'support_contact_email': 'test_support2903@test.com', }, 403, 0, # because header validation is before logic ), # grass headers - not allowed ( { 'account_id': 2, 'account_type': 'vendor', 'account_name': 'FatCat Records', }, { 'assigned_to_id': '1', 'assigned_reviewer_id': '2', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', 'contact_email': 'test2903@test.com', 'contact_name': 'Blacktop 2', 'country_id': 1, 'is_owned': 'Yes', 'label_identifier': 'Frontline', 'name': 'Blacktop 2', 'owner': 'odd', 'primary_genre_id': '2', 'priority': 3, 'product_manager_id': '1', 'quarterback_label_manager_id': '12', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'show_release_builder': 'N', 'support_contact_email': 'test_support2903@test.com', }, 403, 0, # because header validation is before logic ), # profile headers, data validation failed. ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', }, { 'contact_name': 'Blacktop 2', 'account_name': 'Blacktop 2', 'owner': 'SME US Latin', 'label_identifiers': 'Frontline', 'country': 'US', 'service_tier_uuid': '9363881e-1120-4495-ae09-657e13609e11', }, 400, 0, # because post body validation returns before entering the fn. ), ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', }, { 'contact_name': 'Blacktop 2', 'account_name': 'Blacktop 2', 'company_brand_uuid': 'd25a4cd1-e820-45f2-be5c-56edcfeb8298', 'service_tier_uuid': 'diy-tier-1', }, 400, 0, ), # profile headers, data validation success. ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', }, { 'assigned_to_id': '1', 'assigned_reviewer_id': '2', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', 'contact_email': 'test2903@test.com', 'contact_name': 'Blacktop 2', 'country_id': 1, 'is_owned': 'No', 'label_identifier': 'Frontline', 'name': 'Blacktop 2', 'owner': 'odd', 'primary_genre_id': '2', 'priority': 3, 'product_manager_id': '1', 'quarterback_label_manager_id': '12', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'show_release_builder': 'N', 'support_contact_email': 'test_support2903@test.com', }, 200, 1, ), # Orchard_User_Id header. ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', 'Orchard-User-Id': 'oa:1111', }, { 'assigned_to_id': '1', 'assigned_reviewer_id': '2', 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', 'contact_email': 'test2903@test.com', 'contact_name': 'Blacktop 2', 'country_id': 1, 'is_owned': 'No', 'label_identifier': 'Frontline', 'name': 'Blacktop 2', 'owner': 'odd', 'primary_genre_id': '2', 'priority': 3, 'product_manager_id': '1', 'quarterback_label_manager_id': '12', 'service_tier_uuid': '5f2bd4fc-df94-4f35-97d3-ef23f8573279', 'show_release_builder': 'N', 'support_contact_email': 'test_support2903@test.com', }, 200, 1, ), # Orchard_User_Id header with wrong values. ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', 'Orchard-User-Id': 'test:1111', }, {}, 401, 1, ), ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', 'Orchard-User-Id': 'alw:1234', }, {}, 401, 1, ), ( { 'Orchard-Profile-Type': 'LabelProfile', 'Orchard-Profile-Id': 123, 'Orchard-Identity-Id': 'uuid', 'Orchard-Identity-UUID': 'uuid', 'Orchard-User-Id': 4321, }, {}, 401, 1, ), ], ) @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_update_vendor( neo4j_exit, neo4j_enter, headers, data, status, session_count, monkeypatch, fixture_client, fixture_valid_update_vendor, ): """Test v2 update_vendor.""" monkeypatch.setattr( vendor, 'update_vendor', MagicMock(return_value=fixture_valid_update_vendor) ) result = fixture_client.patch('/vendor/1234', json=data, headers=headers) assert result.status_code == status assert neo4j_enter.call_count == session_count @patch('account.handlers.vendor.g') @patch('account.logic.identity.get_oa_user_id') @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.CompanyBrandByNameResourceGetter') @patch('account.handlers.vendor.vendor') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_v2_create_vendor( _, __, mock_vendor_logic, mock_resource_getter, mock_authorization_backend, mock_get_oa_user_id, mock_g, fixture_client, minimum_vendor_attrs, app_context, ): mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'identityId' mock_get_oa_user_id.return_value = 111 mock_vendor_logic.create_vendor.return_value = response.Response() result = fixture_client.post('/v2/vendors', json=minimum_vendor_attrs) assert result.status_code == 200, result.text mock_vendor_logic.create_vendor.assert_called_with( { **minimum_vendor_attrs, 'user_id': 111, 'identity_id': 'identityId', } ) mock_authorization_backend.is_authorized.assert_called_once_with( action='create', resource_id=0, resource_type='account', resource_getter=mock_resource_getter('theorchard'), ) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.CompanyBrandByNameResourceGetter') @patch('account.handlers.vendor.vendor') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_v2_create_vendor_not_authorized( _, __, mock_vendor_logic, mock_resource_getter, mock_authorization_backend, fixture_client, minimum_vendor_attrs, ): mock_authorization_backend.is_authorized.return_value = False mock_vendor_logic.create_vendor.return_value = response.Response() result = fixture_client.post('/v2/vendors', json=minimum_vendor_attrs) assert result.status_code == 403 mock_vendor_logic.create_vendor.assert_not_called() mock_authorization_backend.is_authorized.assert_called_once_with( action='create', resource_id=0, resource_type='account', resource_getter=mock_resource_getter('theorchard'), ) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_v2_delete_vendor( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """Test DELETE /v2/vendors/.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 1394 mock_vendor_logic.delete_vendor.return_value = { 'vendor_id': 42, 'vendor_uuid': vendor_uuid, 'status': 'deletion', } result = fixture_client.delete(f'/v2/vendors/{vendor_uuid}') assert result.status_code == 200, result.text assert json.loads(result.get_data()) == {'vendor_uuid': vendor_uuid} mock_authorization_backend.is_authorized.assert_called_once_with( action='delete', resource_id=vendor_uuid, resource_type='account', resource_getter=mock_resource_getter(vendor_uuid), ) mock_identity_logic.get_oa_user_id.assert_called_once_with('some_identity_uuid') mock_vendor_logic.delete_vendor.assert_called_once_with(vendor_uuid, 1394) @pytest.mark.parametrize( 'is_authorized,oa_user_id', [ pytest.param(False, 82, id='PDP did not authorize'), pytest.param(True, None, id='No OA User Id'), ], ) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') def test_v2_delete_vendor_forbidden( mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, is_authorized: bool, oa_user_id: int | None, fixture_client: MagicMock, app_context, ) -> None: """Test DELETE /v2/vendors/ forbids.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = is_authorized mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = oa_user_id result = fixture_client.delete(f'/v2/vendors/{vendor_uuid}') assert result.status_code == 403, result.text mock_vendor_logic.delete_vendor.assert_not_called() mock_authorization_backend.is_authorized.assert_called_once() if is_authorized: mock_identity_logic.get_oa_user_id.assert_called_once() else: mock_identity_logic.get_oa_user_id.assert_not_called() @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_v2_delete_vendor_idempotent_already_deleted( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """Already-deleted vendors still return 200 with the UUID (idempotent).""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 1394 mock_vendor_logic.delete_vendor.return_value = { 'vendor_id': 42, 'vendor_uuid': vendor_uuid, 'status': None, } result = fixture_client.delete(f'/v2/vendors/{vendor_uuid}') assert result.status_code == 200, result.text assert json.loads(result.get_data()) == {'vendor_uuid': vendor_uuid} @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.g') @patch('account.handlers.vendor.identity_logic') @patch('account.handlers.vendor.vendor') @patch('account.handlers.vendor.AccountByUuidResourceGetter') def test_v2_delete_vendor_fatal( mock_resource_getter: MagicMock, mock_vendor_logic: MagicMock, mock_identity_logic: MagicMock, mock_g: MagicMock, mock_authorization_backend: MagicMock, fixture_client: MagicMock, app_context, ) -> None: """Test DELETE /v2/vendors/ returns 500 when delete raises.""" vendor_uuid = '436ba196-8de7-4011-8cae-9f714df5133e' mock_authorization_backend.is_authorized.return_value = True mock_g.request_context.jwt_identity_id = 'some_identity_uuid' mock_identity_logic.get_oa_user_id.return_value = 1394 mock_vendor_logic.delete_vendor.side_effect = Exception('boom') result = fixture_client.delete(f'/v2/vendors/{vendor_uuid}') assert result.status_code == 500, result.text @patch('account.handlers.vendor.g') @patch('connector_neo4j.Neo4jSession') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_vendors_by_vendor_uuids( neo4j_exit, neo4j_enter, mock_session, mock_g, monkeypatch, fixture_client, app_context, ): """Test get vendors by vendor_uuids.""" mock_g.request_context.jwt_identity_id = 'identityUuid' mock_g.request_context.identity_id = 'identityUuid' mock_g.request_context.profile_id = 'profileId' mock_g.request_context.profile_type = 'profileType' headers = { 'Orchard-Identity-Id': 'identityUuid', 'Orchard-Profile-Id': 'profileId', 'Orchard-Profile-Type': 'profileType', } expected = {'vendors': [None, {'uuid': '84e09dd0-9732-4538-926f-e2010caaa113'}, None]} mock_accessible_vendors = MagicMock(return_value=['84e09dd0-9732-4538-926f-e2010caaa113']) monkeypatch.setattr(neo4j_vendor, 'accessible_vendors', mock_accessible_vendors) mock_get_vendors = MagicMock(return_value=expected) monkeypatch.setattr(vendor, 'get_vendors', mock_get_vendors) result = fixture_client.post( '/vendors/dataloader', headers=headers, json=[ 'not_existing_vendor', '84e09dd0-9732-4538-926f-e2010caaa113', 'vendor_without_access', ], ) mock_accessible_vendors.assert_called_once_with( profile_id='profileId', profile_type='profileType', vendor_uuids=[ 'not_existing_vendor', '84e09dd0-9732-4538-926f-e2010caaa113', 'vendor_without_access', ], identity_id='identityUuid', ) mock_get_vendors.assert_called_with( accessible_vendor_uuids=[ '84e09dd0-9732-4538-926f-e2010caaa113', ], vendor_uuids=[ 'not_existing_vendor', '84e09dd0-9732-4538-926f-e2010caaa113', 'vendor_without_access', ], ) assert json.loads(result.data) == expected @patch('account.handlers.vendor.g') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_vendors_by_vendor_uuids_exception_get_vendors( neo4j_exit, neo4j_enter, mock_g, monkeypatch, fixture_client, app_context, ): """Test get vendors by vendor_uuids handles exception.""" mock_accessible_vendors = MagicMock(return_value=['84e09dd0-9732-4538-926f-e2010caaa113']) monkeypatch.setattr(neo4j_vendor, 'accessible_vendors', mock_accessible_vendors) def mock_get_vendors(*args, **kwargs): raise Exception('Mocked exception in get_vendors') monkeypatch.setattr(vendor, 'get_vendors', mock_get_vendors) result = fixture_client.post( '/vendors/dataloader', json=[ '84e09dd0-9732-4538-926f-e2010caaa113', ], ) assert json.loads(result.data) == { 'code': 'internal_error', 'message': ['Mocked exception in get_vendors'], } @patch('account.handlers.vendor.g') @patch('connector_neo4j.Neo4jSession') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_vendors_by_vendor_uuids_exception_accessible_vendors( neo4j_exit, neo4j_enter, mock_session, mock_g, monkeypatch, fixture_client, app_context, ): """Test get vendors by vendor_uuids handles exception.""" def mock_accessible_vendors(*args, **kwargs): raise Exception('Mocked exception in accessible_vendors') monkeypatch.setattr(neo4j_vendor, 'accessible_vendors', mock_accessible_vendors) expected = {'vendors': [None, {'uuid': '84e09dd0-9732-4538-926f-e2010caaa113'}, None]} mock_get_vendors = MagicMock(return_value=expected) monkeypatch.setattr(vendor, 'get_vendors', mock_get_vendors) result = fixture_client.post( '/vendors/dataloader', json=[ '84e09dd0-9732-4538-926f-e2010caaa113', ], ) assert json.loads(result.data) == { 'code': 'internal_error', 'message': ['Mocked exception in accessible_vendors'], } @patch('account.handlers.vendor.g') @patch('connector_neo4j.Neo4jSession') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_vendors_by_vendor_uuids_pp_auth_check( neo4j_exit, neo4j_enter, mock_session, mock_g, monkeypatch, fixture_client, app_context, ): """Test get vendors by vendor_uuids.""" monkeypatch.setattr( flask_request, 'verify_rules_access_standalone', MagicMock(return_value=False) ) expected = {'vendors': [None, {'uuid': '84e09dd0-9732-4538-926f-e2010caaa113'}, None]} mock_authorize_vendors_pp = MagicMock(return_value=['84e09dd0-9732-4538-926f-e2010caaa113']) monkeypatch.setattr(vendor, 'get_pp_accessible_vendors', mock_authorize_vendors_pp) mock_get_vendors = MagicMock(return_value=expected) monkeypatch.setattr(vendor, 'get_vendors', mock_get_vendors) result = fixture_client.post( '/vendors/dataloader', json=[ 'not_existing_vendor', '84e09dd0-9732-4538-926f-e2010caaa113', 'vendor_without_access', ], ) mock_authorize_vendors_pp.assert_called_once_with( vendor_uuids=[ 'not_existing_vendor', '84e09dd0-9732-4538-926f-e2010caaa113', 'vendor_without_access', ] ) mock_get_vendors.assert_called_with( accessible_vendor_uuids=[ '84e09dd0-9732-4538-926f-e2010caaa113', ], vendor_uuids=[ 'not_existing_vendor', '84e09dd0-9732-4538-926f-e2010caaa113', 'vendor_without_access', ], ) assert json.loads(result.data) == expected @patch('account.handlers.vendor.g') @patch('connector_neo4j.Neo4jSession') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') @patch('account.handlers.vendor.pythonfeatures.get_single_feature') def test_get_company_brand( mock_feature, neo4j_exit, neo4j_enter, mock_session, mock_g, feature_enabled_response, monkeypatch, fixture_client, app_context, ): expected_company_brand = {'name': 'Company Brand'} expected_response = response.Response(expected_company_brand) mock_feature.return_value = feature_enabled_response mock_get_vendor_company_brand_tx = MagicMock(return_value=expected_response) mock_get_vendor_company_brand = MagicMock(return_value=expected_response) monkeypatch.setattr(vendor, 'get_vendor_company_brand', mock_get_vendor_company_brand) monkeypatch.setattr(vendor, 'get_vendor_company_brand_tx', mock_get_vendor_company_brand_tx) result = fixture_client.get('/vendor/brand/12345') assert result.status_code == 200 assert json.loads(result.get_data(as_text=True)) == expected_company_brand if feature_enabled_response.message == 'enabled': mock_get_vendor_company_brand_tx.assert_called_once_with('12345') mock_get_vendor_company_brand.assert_not_called() if feature_enabled_response.message == 'disabled': mock_get_vendor_company_brand.assert_called_once_with('12345') mock_get_vendor_company_brand_tx.assert_not_called() @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.AccountByIdResourceGetter') @patch('account.handlers.vendor.vendor') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_master_contact( _, __, mock_vendor_logic, mock_resource_getter, mock_authorization_backend, fixture_client, ): mock_authorization_backend.is_authorized.return_value = True mock_vendor_logic.get_master_contact.return_value = response.Response( {'identity_id': 'test-identity-id'} ) result = fixture_client.get('/vendor/12345/master-contact') assert result.status_code == 200 mock_vendor_logic.get_master_contact.assert_called_once_with(12345) mock_authorization_backend.is_authorized.assert_called_once_with( action='view_account_info', resource_id=12345, resource_type='account', resource_getter=mock_resource_getter(12345), ) @patch('account.handlers.vendor.authorization_backend') @patch('account.handlers.vendor.AccountByIdResourceGetter') @patch('account.handlers.vendor.vendor') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_master_contact_unauthorized( _, __, mock_vendor_logic, mock_resource_getter, mock_authorization_backend, fixture_client, ): mock_authorization_backend.is_authorized.return_value = False result = fixture_client.get('/vendor/12345/master-contact') assert result.status_code == 403 mock_vendor_logic.get_master_contact.assert_not_called() mock_authorization_backend.is_authorized.assert_called_once_with( action='view_account_info', resource_id=12345, resource_type='account', resource_getter=mock_resource_getter(12345), ) @patch('account.logic.vendor.get_vendor_company_brands') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_vendor_company_brands_success_with_skip_flag( neo4j_exit, neo4j_enter, mock_logic_get_vendor_company_brands: MagicMock, fixture_client: FlaskClient, ) -> None: """Test POST /vendors/company_brands/dataloader without a permissions check.""" vendor_uuids = ['0662568b-f291-43c5-91b3-7bf639ffe568', '263357da-3e46-49a7-827b-f2505a95c218'] expected_response = { 'vendors': [ { 'vendor_id': 1, 'vendor_uuid': '0662568b-f291-43c5-91b3-7bf639ffe568', 'company_brand_id': 10, }, { 'vendor_id': 2, 'vendor_uuid': '263357da-3e46-49a7-827b-f2505a95c218', 'company_brand_id': 20, }, ] } mock_logic_get_vendor_company_brands.return_value = expected_response result = fixture_client.post( '/vendors/company_brands/dataloader', data=json.dumps({'vendor_uuids': vendor_uuids, 'skip_access_check': True}), content_type='application/json', ) assert result.status_code == 200 assert json.loads(result.data) == expected_response vendor_uuid_objects = [UUID(v) for v in vendor_uuids] mock_logic_get_vendor_company_brands.assert_called_once_with(vendor_uuid_objects) @patch('account.logic.vendor.get_vendor_company_brands') @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_vendor_company_brands_empty_list( neo4j_exit, neo4j_enter, mock_logic_get_vendor_company_brands: MagicMock, fixture_client: FlaskClient, ) -> None: """Test POST /vendors/company_brands/dataloader with empty list.""" payload = {'vendor_uuids': [], 'skip_access_check': True} expected_response = {'vendors': []} mock_logic_get_vendor_company_brands.return_value = expected_response result = fixture_client.post( '/vendors/company_brands/dataloader', data=json.dumps(payload), content_type='application/json', ) assert result.status_code == 200 assert json.loads(result.data) == expected_response mock_logic_get_vendor_company_brands.assert_called_once_with([]) @patch('account.logic.vendor.get_vendor_company_brands') def test_get_vendor_company_brands_invalid_json( mock_logic_get_vendor_company_brands: MagicMock, fixture_client: FlaskClient ) -> None: """Test POST /vendors/company_brands/dataloader with invalid JSON.""" result = fixture_client.post( '/vendors/company_brands/dataloader', data='invalid json', content_type='application/json' ) assert result.status_code == 400 mock_logic_get_vendor_company_brands.assert_not_called() @patch('account.logic.vendor.get_vendor_company_brands') def test_get_vendor_company_brands_non_list_data( mock_logic_get_vendor_company_brands: MagicMock, fixture_client: FlaskClient ) -> None: """Test POST /vendors/company_brands/dataloader with non-list data.""" result = fixture_client.post( '/vendors/company_brands/dataloader', data=json.dumps({'vendor_uuids': 'a list'}), content_type='application/json', ) assert result.status_code == 400 mock_logic_get_vendor_company_brands.assert_not_called() @patch('account.logic.vendor.get_vendor_company_brands') def test_get_vendor_company_brands_list_with_non_string( mock_logic_get_vendor_company_brands: MagicMock, fixture_client: FlaskClient ) -> None: """Test POST /vendors/company_brands/dataloader with list containing non-strings.""" result = fixture_client.post( '/vendors/company_brands/dataloader', data=json.dumps({'vendor_uuids': ['uuid1', 123, 'uuid2']}), content_type='application/json', ) assert result.status_code == 400 mock_logic_get_vendor_company_brands.assert_not_called() @patch('account.logic.vendor.get_vendor_company_brands', side_effect=Exception('Database error')) @patch('connector_neo4j.Neo4jSession.__enter__') @patch('connector_neo4j.Neo4jSession.__exit__') def test_get_vendor_company_brands_logic_exception( neo4j_exit, neo4j_enter, mock_logic_get_vendor_company_brands, fixture_client ): """Test POST /vendors/company_brands/dataloader when logic raises exception.""" vendor_uuids = ['0662568b-f291-43c5-91b3-7bf639ffe568'] result = fixture_client.post( '/vendors/company_brands/dataloader', data=json.dumps({'vendor_uuids': vendor_uuids, 'skip_access_check': True}), content_type='application/json', ) assert result.status_code == 500 vendor_uuid_objects = [UUID(v) for v in vendor_uuids] mock_logic_get_vendor_company_brands.assert_called_once_with(vendor_uuid_objects) @patch('account.logic.vendor.get_vendor_service_tier') def test_get_vendor_service_tier_success_with_skip_flag( mock_logic_get_vendor_service_tier: MagicMock, fixture_client: FlaskClient ) -> None: """Test POST /vendors/service_tier/dataloader without a permissions check.""" vendor_uuids = ['d8361153-faaf-11ef-8476-0ef8c77b7565', 'c304de34-0006-11f0-8476-0ef8c77b7565'] expected_response = { 'vendors': [ { 'vendor_id': 1, 'vendor_uuid': 'd8361153-faaf-11ef-8476-0ef8c77b7565', 'service_tier_uuid': '1dd92c83-25a3-4034-9a3d-f7c3f434f4ce', 'service_tier_name': 'premium-services', 'service_tier_display_name': 'Premium Services', }, { 'vendor_id': 2, 'vendor_uuid': 'c304de34-0006-11f0-8476-0ef8c77b7565', 'service_tier_uuid': '1dd92c83-25a3-4034-9a3d-f7c3f434f4ce', 'service_tier_name': 'premium-services', 'service_tier_display_name': 'Premium Services', }, ] } mock_logic_get_vendor_service_tier.return_value = expected_response result = fixture_client.post( '/vendors/service_tier/dataloader', data=json.dumps({'vendor_uuids': vendor_uuids, 'skip_access_check': True}), content_type='application/json', ) assert result.status_code == 200 assert json.loads(result.data) == expected_response mock_logic_get_vendor_service_tier.assert_called_once_with([UUID(v) for v in vendor_uuids]) @pytest.mark.parametrize( ('has_access', 'expected_status'), [ (True, 200), (False, 403), ], ) @patch('account.logic.vendor.get_vendor_service_tier') @patch('account.models.ows_permissions.ows_client.get') @patch('account.models.ows_permissions.g') def test_get_vendor_service_tier_success_without_skip_flag( mock_g: MagicMock, mock_ows_client: MagicMock, mock_logic_get_vendor_service_tier: MagicMock, has_access: bool, expected_status: int, fixture_client: FlaskClient, app_context: AppContext, ) -> None: """Test POST /vendors/service_tier/dataloader with a permissions check.""" mock_g.request_context.profile_id = '123' mock_g.request_context.identity_id = 'test-identity-id' mock_g.request_context.profile_type = 'SettingsProfile' mock_ows_client.return_value = MagicMock( status_code=200, json=lambda: {'has_access': has_access} ) vendor_uuids = ['d8361153-faaf-11ef-8476-0ef8c77b7565', 'c304de34-0006-11f0-8476-0ef8c77b7565'] expected_response = { 'vendors': [ { 'vendor_id': 1, 'vendor_uuid': 'd8361153-faaf-11ef-8476-0ef8c77b7565', 'service_tier_uuid': '1dd92c83-25a3-4034-9a3d-f7c3f434f4ce', 'service_tier_name': 'premium-services', 'service_tier_display_name': 'Premium Services', }, { 'vendor_id': 2, 'vendor_uuid': 'c304de34-0006-11f0-8476-0ef8c77b7565', 'service_tier_uuid': '1dd92c83-25a3-4034-9a3d-f7c3f434f4ce', 'service_tier_name': 'premium-services', 'service_tier_display_name': 'Premium Services', }, ] } mock_logic_get_vendor_service_tier.return_value = expected_response result = fixture_client.post( '/vendors/service_tier/dataloader', data=json.dumps({'vendor_uuids': vendor_uuids, 'skip_access_check': False}), content_type='application/json', ) assert result.status_code == expected_status mock_ows_client.assert_called_once_with( 'ows-permissions', '/v2/profile/self/all-label-access', headers={ 'Orchard-Identity-Id': 'test-identity-id', 'Orchard-Profile-Id': '123', 'Orchard-Profile-Type': 'SettingsProfile', }, ) if has_access: assert json.loads(result.data) == expected_response mock_logic_get_vendor_service_tier.assert_called_once_with([UUID(v) for v in vendor_uuids]) else: mock_logic_get_vendor_service_tier.assert_not_called() @patch('account.logic.vendor.get_vendor_service_tier') def test_get_vendor_service_tier_empty_list( mock_logic_get_vendor_service_tier: MagicMock, fixture_client: FlaskClient ) -> None: """Test POST /vendors/service_tier/dataloader with empty list.""" payload = {'vendor_uuids': [], 'skip_access_check': True} expected_response = {'vendors': []} mock_logic_get_vendor_service_tier.return_value = expected_response result = fixture_client.post( '/vendors/service_tier/dataloader', data=json.dumps(payload), content_type='application/json', ) assert result.status_code == 200 assert json.loads(result.data) == expected_response mock_logic_get_vendor_service_tier.assert_called_once_with([]) @patch('account.logic.vendor.get_vendor_service_tier') def test_get_vendor_service_tier_invalid_json( mock_logic_get_vendor_service_tier: MagicMock, fixture_client: FlaskClient ) -> None: """Test POST /vendors/service_tier/dataloader with invalid JSON.""" result = fixture_client.post( '/vendors/service_tier/dataloader', data='invalid json', content_type='application/json' ) assert result.status_code == 400 mock_logic_get_vendor_service_tier.assert_not_called() @patch('account.logic.vendor.get_vendor_service_tier') def test_get_vendor_service_tier_non_list_data( mock_logic_get_vendor_service_tier: MagicMock, fixture_client: FlaskClient ) -> None: """Test POST /vendors/service_tier/dataloader with non-list data.""" result = fixture_client.post( '/vendors/service_tier/dataloader', data=json.dumps({'vendor_uuids': 'a list'}), content_type='application/json', ) assert result.status_code == 400 mock_logic_get_vendor_service_tier.assert_not_called() @patch('account.logic.vendor.get_vendor_service_tier') def test_get_vendor_service_tier_list_with_non_string( mock_logic_get_vendor_service_tier: MagicMock, fixture_client: FlaskClient ) -> None: """Test POST /vendors/service_tier/dataloader with list containing non-strings.""" result = fixture_client.post( '/vendors/service_tier/dataloader', data=json.dumps({'vendor_uuids': ['uuid1', 123, 'uuid2']}), content_type='application/json', ) assert result.status_code == 400 mock_logic_get_vendor_service_tier.assert_not_called() @patch('account.logic.vendor.get_vendor_service_tier', side_effect=Exception('Database error')) def test_get_vendor_service_tier_logic_exception( mock_logic_get_vendor_service_tier, fixture_client ): """Test POST /vendors/service_tier/dataloader when logic raises exception.""" vendor_uuids = ['d8361153-faaf-11ef-8476-0ef8c77b7565'] result = fixture_client.post( '/vendors/service_tier/dataloader', data=json.dumps({'vendor_uuids': vendor_uuids, 'skip_access_check': True}), content_type='application/json', ) assert result.status_code == 500 mock_logic_get_vendor_service_tier.assert_called_once_with([UUID(v) for v in vendor_uuids]) def test_get_vendors_relationship_notes_success(monkeypatch, fixture_client): """Test get_vendors_relationship_notes_handler with valid input""" expected = { 'vendors': [ { 'vendor_uuid': '5ea89d97-c524-44bf-91bb-245e40d2949f', 'relationship_notes': 'test', 'vendor_id': 12345, }, { 'vendor_uuid': '0640d861-7fbe-4034-aebb-c6083edb850e', 'relationship_notes': None, 'vendor_id': 67890, }, ] } mock_logic = MagicMock(return_value=expected) monkeypatch.setattr( vendor, 'get_vendors_relationship_notes', mock_logic, ) monkeypatch.setattr( flask_request, 'verify_rules_access_standalone', value=MagicMock(return_value=True), ) result = fixture_client.post( '/vendors/relationship_notes/dataloader', json={ 'vendor_uuids': [ '5ea89d97-c524-44bf-91bb-245e40d2949f', '0640d861-7fbe-4034-aebb-c6083edb850e', ] }, ) assert flask_request.verify_rules_access_standalone.called assert result.status_code == 200 assert json.loads(result.data) == expected mock_logic.assert_called_once_with( [ UUID('5ea89d97-c524-44bf-91bb-245e40d2949f'), UUID('0640d861-7fbe-4034-aebb-c6083edb850e'), ] ) @pytest.mark.parametrize( 'input_data, expected_status_code, expected_error_message', [ pytest.param( {'vendor_uuids': []}, 400, {'vendor_uuids': ['Shorter than minimum length 1.']}, id='Empty vendor_uuids', ), pytest.param( {'other_field': [1, 2]}, 400, { 'vendor_uuids': ['Missing data for required field.'], 'other_field': ['Unknown field.'], }, id='Missing vendor_uuids field', ), pytest.param( {'vendor_uuids': ['1', '2']}, 400, {'vendor_uuids': {'0': ['Not a valid UUID.'], '1': ['Not a valid UUID.']}}, id='Invalid UUIDs', ), pytest.param( {'vendor_uuids': '1,2'}, 400, {'vendor_uuids': ['Not a valid list.']}, id='Non-list vendor_uuids', ), pytest.param( [], 400, {'vendor_uuids': ['Missing data for required field.']}, id='Empty JSON body', ), ], ) def test_get_vendors_relationship_notes_invalid_input( monkeypatch, fixture_client, input_data, expected_status_code, expected_error_message ): """Test invalid input validation for relationship notes handler""" expected = { 'code': 'input_validation_error', 'message': expected_error_message, } mock_logic = MagicMock() monkeypatch.setattr( vendor, 'get_vendors_relationship_notes', mock_logic, ) result = fixture_client.post( '/vendors/relationship_notes/dataloader', json=input_data, ) assert result.status_code == expected_status_code assert json.loads(result.data) == expected assert mock_logic.call_count == 0 def test_get_vendors_relationship_notes_forbidden(monkeypatch, fixture_client): """Test handler returns 403 when access rule fails""" monkeypatch.setattr( flask_request, 'verify_rules_access_standalone', value=MagicMock(return_value=False), ) result = fixture_client.post( '/vendors/relationship_notes/dataloader', json={'vendor_uuids': ['5ea89d97-c524-44bf-91bb-245e40d2949f']}, ) assert result.status_code == 403 assert json.loads(result.data) == { 'code': 'authorization_error', 'message': 'Forbidden', } def test_get_vendors_relationship_notes_internal_error(monkeypatch, fixture_client): """Test handler returns 500 when logic layer throws an exception""" monkeypatch.setattr( flask_request, 'verify_rules_access_standalone', value=MagicMock(return_value=True), ) mock_logic = MagicMock(side_effect=Exception('DB failure')) monkeypatch.setattr( vendor, 'get_vendors_relationship_notes', mock_logic, ) result = fixture_client.post( '/vendors/relationship_notes/dataloader', json={'vendor_uuids': ['5ea89d97-c524-44bf-91bb-245e40d2949f']}, ) assert result.status_code == 500 assert json.loads(result.data) == { 'code': 'internal_error', 'message': 'DB failure', }