"""Unit tests for reference_sap_profit_center logic.""" from unittest.mock import call, patch import pytest from marshmallow import ValidationError from sqlalchemy.exc import IntegrityError, SQLAlchemyError from werkzeug.exceptions import HTTPException from abacus_contract import models from abacus_contract.constants.constants import ( SAP_PROFIT_CENTER_SORT_OPTIONS, SORT_ORDER_OPTIONS, ) from abacus_contract.logic import reference_sap_profit_center as logic from abacus_contract.tests.utils.factories import ( ReferenceSapProfitCenterFactory, ReferenceSigningEntityFactory, SigningEntitySapProfitCenterFactory, ) # --- update_reference_sap_profit_center (PUT) --- def test_update_reference_sap_profit_center_sets_display_name(): """update_attributes flips display_name and the row is committed.""" pc = ReferenceSapProfitCenterFactory.create(display_name='Old name') result = logic.update_reference_sap_profit_center(pc, display_name='New name') assert result.status == 200 assert result.message['display_name'] == 'New name' refreshed = models.ReferenceSapProfitCenter.query.get( pc.reference_sap_profit_center_id ) assert refreshed.display_name == 'New name' def test_update_reference_sap_profit_center_preserves_other_fields(): """PUT display_name doesn't accidentally null out profit_center / company_code / business_group.""" pc = ReferenceSapProfitCenterFactory.create( profit_center='ABC123', company_code='4914', business_group='ORC', display_name='Original', ) result = logic.update_reference_sap_profit_center(pc, display_name='Updated') assert result.status == 200 refreshed = models.ReferenceSapProfitCenter.query.get( pc.reference_sap_profit_center_id ) assert refreshed.display_name == 'Updated' assert refreshed.profit_center == 'ABC123' assert refreshed.company_code == '4914' assert refreshed.business_group == 'ORC' # --- list_signing_entities_for_sap_profit_center (drawer endpoint) --- def test_list_signing_entities_for_sap_profit_center_returns_only_authorized_live(): """Only SEs with a live junction row for the PC are returned, with junction metadata.""" pc = ReferenceSapProfitCenterFactory.create() authorized_se = ReferenceSigningEntityFactory.create( legal_name='Authorized Inc.', with_junction_row=False ) other_se = ReferenceSigningEntityFactory.create( legal_name='Other Co.', with_junction_row=False ) _unrelated_se = ReferenceSigningEntityFactory.create( legal_name='Unrelated LLC', with_junction_row=False ) junction = SigningEntitySapProfitCenterFactory.create( reference_signing_entity=authorized_se, reference_sap_profit_center=pc ) soft_deleted_junction = SigningEntitySapProfitCenterFactory.create( reference_signing_entity=other_se, reference_sap_profit_center=pc ) models.SigningEntitySapProfitCenter.delete_by_id_or_error( soft_deleted_junction.signing_entity_sap_profit_center_id, soft_delete=True ) result = logic.list_signing_entities_for_sap_profit_center( pc.reference_sap_profit_center_id ) assert result.status == 200 assert result.message['total_count'] == 1 items = result.message['items'] assert len(items) == 1 assert ( items[0]['signing_entity_sap_profit_center_id'] == junction.signing_entity_sap_profit_center_id ) assert items[0]['signing_entity']['reference_signing_entity_id'] == ( authorized_se.reference_signing_entity_id ) assert items[0]['signing_entity']['legal_name'] == 'Authorized Inc.' def test_list_signing_entities_for_sap_profit_center_search_escapes_like_wildcards(): """search_term='%' should match nothing (escaped), not every row.""" pc = ReferenceSapProfitCenterFactory.create() se1 = ReferenceSigningEntityFactory.create( legal_name='Real Inc.', with_junction_row=False ) se2 = ReferenceSigningEntityFactory.create( legal_name='Other Co.', with_junction_row=False ) for se in (se1, se2): SigningEntitySapProfitCenterFactory.create( reference_signing_entity=se, reference_sap_profit_center=pc ) result = logic.list_signing_entities_for_sap_profit_center( pc.reference_sap_profit_center_id, search_term='%' ) assert result.message['total_count'] == 0 assert result.message['items'] == [] def test_list_signing_entities_for_sap_profit_center_q_filters_legal_name(): """?search_term filters SE legal_name (case-insensitive).""" pc = ReferenceSapProfitCenterFactory.create() match = ReferenceSigningEntityFactory.create( legal_name='Foundation Media SAS', with_junction_row=False ) no_match = ReferenceSigningEntityFactory.create( legal_name='Santa Anna LLC', with_junction_row=False ) for se in (match, no_match): SigningEntitySapProfitCenterFactory.create( reference_signing_entity=se, reference_sap_profit_center=pc ) result = logic.list_signing_entities_for_sap_profit_center( pc.reference_sap_profit_center_id, search_term='foundation' ) assert { item['signing_entity']['reference_signing_entity_id'] for item in result.message['items'] } == {match.reference_signing_entity_id} def test_list_signing_entities_for_sap_profit_center_empty_when_pc_has_no_mappings(): """PC with no live mappings -> empty items, 200 (not 404).""" pc = ReferenceSapProfitCenterFactory.create() result = logic.list_signing_entities_for_sap_profit_center( pc.reference_sap_profit_center_id ) assert result.status == 200 assert result.message['items'] == [] assert result.message['total_count'] == 0 def test_list_signing_entities_for_sap_profit_center_unknown_pc_raises_404(): """Non-existent PC id -> framework abort 404.""" with pytest.raises(HTTPException) as exc_info: logic.list_signing_entities_for_sap_profit_center(99999999) assert exc_info.value.code == 404 def test_list_signing_entities_for_sap_profit_center_pagination_limits_results(): """?limit caps the page; total_count still reflects the full match set.""" pc = ReferenceSapProfitCenterFactory.create() for n in range(5): se = ReferenceSigningEntityFactory.create( legal_name=f'SE {n}', with_junction_row=False ) SigningEntitySapProfitCenterFactory.create( reference_signing_entity=se, reference_sap_profit_center=pc ) result = logic.list_signing_entities_for_sap_profit_center( pc.reference_sap_profit_center_id, limit=2, offset=0 ) assert result.status == 200 assert result.message['total_count'] == 5 assert len(result.message['items']) == 2 def test_list_signing_entities_for_sap_profit_center_offset_past_end_returns_empty(): """?offset past the matching set -> empty items, total_count still accurate.""" pc = ReferenceSapProfitCenterFactory.create() for n in range(3): se = ReferenceSigningEntityFactory.create( legal_name=f'SE {n}', with_junction_row=False ) SigningEntitySapProfitCenterFactory.create( reference_signing_entity=se, reference_sap_profit_center=pc ) result = logic.list_signing_entities_for_sap_profit_center( pc.reference_sap_profit_center_id, limit=10, offset=100 ) assert result.status == 200 assert result.message['total_count'] == 3 assert result.message['items'] == [] def test_list_signing_entities_for_sap_profit_center_orders_by_legal_name(): """Results are ordered by SE legal_name ascending (pinned for UI stability).""" pc = ReferenceSapProfitCenterFactory.create() for name in ('Charlie LLC', 'Alpha Inc.', 'Bravo Co.'): se = ReferenceSigningEntityFactory.create( legal_name=name, with_junction_row=False ) SigningEntitySapProfitCenterFactory.create( reference_signing_entity=se, reference_sap_profit_center=pc ) result = logic.list_signing_entities_for_sap_profit_center( pc.reference_sap_profit_center_id ) names = [item['signing_entity']['legal_name'] for item in result.message['items']] assert names == ['Alpha Inc.', 'Bravo Co.', 'Charlie LLC'] def test_list_signing_entities_for_sap_profit_center_search_does_not_strip_whitespace(): """search_term is used as-is (no implicit strip), matching account.py precedent.""" pc = ReferenceSapProfitCenterFactory.create() se = ReferenceSigningEntityFactory.create( legal_name='Match', with_junction_row=False ) SigningEntitySapProfitCenterFactory.create( reference_signing_entity=se, reference_sap_profit_center=pc ) result = logic.list_signing_entities_for_sap_profit_center( pc.reference_sap_profit_center_id, search_term=' match ' ) assert result.message['items'] == [] assert result.message['total_count'] == 0 # Note re: soft-deleted parent PC. ReferenceSapProfitCenter has no `deleted_at` # column today, so a PC cannot be soft-deleted from the application. No test is # possible/needed for the soft-deleted-PC case. If PCs gain a `deleted_at` # column later, mirror the SE-side check in # `reference_signing_entity.py`'s logic and add a corresponding test. @patch('abacus_contract.logic.reference_sap_profit_center.models') def test_get_reference_sap_profit_centers(mock_models): """Test to get reference sap profit centers.""" mock_profit_center = ReferenceSapProfitCenterFactory.create() mock_models.ReferenceSapProfitCenter.get_reference_sap_profit_centers.return_value = ( [mock_profit_center], 1, ) mock_request_params = { 'limit': 10, 'offset': 0, 'sort_by': SAP_PROFIT_CENTER_SORT_OPTIONS.COMPANY_CODE, 'sort_order': SORT_ORDER_OPTIONS.DESC, } result = logic.get_reference_sap_profit_centers(mock_request_params) assert result.status == 200 assert result.message['total_count'] == 1 mock_models.ReferenceSapProfitCenter.get_reference_sap_profit_centers.assert_called_once_with( **mock_request_params ) @patch( 'abacus_contract.logic.reference_sap_profit_center.SigningEntitySapProfitCenterWithSigningEntitySchema' ) @patch('abacus_contract.logic.reference_sap_profit_center.models') @patch( 'abacus_contract.logic.reference_sap_profit_center.prepare_dataload_with_data_as_list_response' ) def test_get_signing_entities_by_profit_centers( mock_dataload_response, mock_models, mock_mapping_schema ): """Test to get signing entities by profit centers.""" mock_signing_entities = [ { 'signing_entity_sap_profit_center_id': 1, 'reference_sap_profit_center_id': 1, 'created_at': '2026-05-29T16:51:25.000000', 'signing_entity': { 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'reference_payment_entity_id': 1, 'company_code': '4914', 'legal_name': 'AWAL Digital Limited', 'vat_number': 'GB 423 4787 86', 'company_registration_number': '04430703', 'address': '2 Canal Reach, London, N1C 4DB', }, } ] mock_formatted_data = [{'data': mock_signing_entities}, {'data': None}] mock_models.ReferenceSigningEntity.get_authorized_for_sap_profit_centers.return_value = ( mock_signing_entities, 1, ) mock_dataload_response.return_value = mock_formatted_data mock_mapping_schema.return_value.dump.return_value = mock_signing_entities result = logic.get_signing_entities_by_profit_centers([1, 2]) assert result.status == 200 assert result.message == mock_formatted_data mock_models.ReferenceSigningEntity.get_authorized_for_sap_profit_centers.assert_called_with( [1, 2], 100, 0 ) mock_dataload_response.assert_called_with( [1, 2], mock_signing_entities, 'reference_sap_profit_center_id' ) @patch('abacus_contract.logic.reference_sap_profit_center.models') @patch('abacus_contract.logic.reference_sap_profit_center.prepare_dataload_response') def test_get_sap_profit_centers_by_ids(mock_dataload_response, mock_models): """Test to get SAP profit centers by ids.""" mock_profit_centers = [ { 'reference_sap_profit_center_id': 1, 'profit_center': 'PC001', 'company_code': '4914', 'business_group': 'ORC', 'display_name': 'Test', } ] mock_formatted_data = [{'data': mock_profit_centers[0]}, {'data': None}] mock_models.ReferenceSapProfitCenter.get_by_ids.return_value = mock_profit_centers mock_dataload_response.return_value = mock_formatted_data result = logic.get_sap_profit_centers_by_ids([1, 2]) assert result.status == 200 assert result.message == mock_formatted_data mock_models.ReferenceSapProfitCenter.get_by_ids.assert_called_with([1, 2]) mock_dataload_response.assert_called_with( [1, 2], mock_profit_centers, 'reference_sap_profit_center_id' ) @patch( 'abacus_contract.logic.reference_sap_profit_center.SigningEntitySapProfitCenterWithSigningEntitySchema' ) @patch('abacus_contract.logic.reference_sap_profit_center.models') @patch( 'abacus_contract.logic.reference_sap_profit_center.prepare_dataload_with_data_as_list_response' ) def test_get_signing_entities_by_duplicate_profit_centers( mock_dataload_response, mock_models, mock_mapping_schema ): """Test to get signing entities by profit centers.""" mock_signing_entities = [ { 'signing_entity_sap_profit_center_id': 1, 'reference_sap_profit_center_id': 1, 'created_at': '2026-05-29T16:51:25.000000', 'signing_entity': { 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'reference_payment_entity_id': 1, 'company_code': '4914', 'legal_name': 'AWAL Digital Limited', 'vat_number': 'GB 423 4787 86', 'company_registration_number': '04430703', 'address': '2 Canal Reach, London, N1C 4DB', }, } ] mock_formatted_data = [{'data': mock_signing_entities}, {'data': None}] mock_models.ReferenceSigningEntity.get_authorized_for_sap_profit_centers.return_value = ( mock_signing_entities, 1, ) mock_dataload_response.return_value = mock_formatted_data mock_mapping_schema.return_value.dump.return_value = mock_signing_entities result = logic.get_signing_entities_by_profit_centers([1, 2, 2, 2, 2]) assert result.status == 200 assert result.message == mock_formatted_data mock_models.ReferenceSigningEntity.get_authorized_for_sap_profit_centers.assert_called_with( [1, 2], 100, 0 ) mock_dataload_response.assert_called_with( [1, 2], mock_signing_entities, 'reference_sap_profit_center_id' ) @patch('abacus_contract.logic.reference_sap_profit_center.models') @patch('abacus_contract.logic.reference_sap_profit_center.prepare_dataload_response') def test_get_sap_profit_centers_by_duplicate_ids(mock_dataload_response, mock_models): """Test to get SAP profit centers by duplicate ids.""" mock_profit_centers = [ { 'reference_sap_profit_center_id': 1, 'profit_center': 'PC001', 'company_code': '4914', 'business_group': 'ORC', 'display_name': 'Test', } ] mock_formatted_data = [{'data': mock_profit_centers[0]}, {'data': None}] mock_models.ReferenceSapProfitCenter.get_by_ids.return_value = mock_profit_centers mock_dataload_response.return_value = mock_formatted_data result = logic.get_sap_profit_centers_by_ids([1, 2, 1, 2, 1, 2, 3]) assert result.status == 200 assert result.message == mock_formatted_data mock_models.ReferenceSapProfitCenter.get_by_ids.assert_called_with([1, 2, 3]) mock_dataload_response.assert_called_with( [1, 2, 3], mock_profit_centers, 'reference_sap_profit_center_id' ) # --- create_reference_sap_profit_center (POST) --- @patch('abacus_contract.logic.reference_sap_profit_center.db') @patch( 'abacus_contract.logic.reference_sap_profit_center._check_whether_signing_entities_exist' ) @patch('abacus_contract.logic.reference_sap_profit_center.models') def test_create_reference_sap_profit_center_without_signing_entities( mock_models, mock_validation, mock_db ): """No reference_signing_entity_ids -> PC is created, no junction rows.""" mock_profit_center = ReferenceSapProfitCenterFactory.create() mock_models.ReferenceSapProfitCenter.build.return_value = mock_profit_center mock_db.session.commit.return_value = True mock_validation.return_value = True mock_post_request_body = { 'profit_center': 'UK9999', 'company_code': '4912', 'business_group': 'ORC', 'display_name': 'New Profit Center', } result = logic.create_reference_sap_profit_center(**mock_post_request_body) assert result.status == 201 mock_validation.assert_called_once_with([]) mock_models.ReferenceSapProfitCenter.build.assert_called_once_with( **mock_post_request_body ) mock_models.SigningEntitySapProfitCenter.build.assert_not_called() mock_db.session.commit.assert_called_once() @patch('abacus_contract.logic.reference_sap_profit_center.db') @patch( 'abacus_contract.logic.reference_sap_profit_center._check_whether_signing_entities_exist' ) @patch('abacus_contract.logic.reference_sap_profit_center.models') def test_create_reference_sap_profit_center_with_empty_signing_entity_list( mock_models, mock_validation, mock_db ): """An explicit empty list behaves the same as None -> no junction rows.""" mock_profit_center = ReferenceSapProfitCenterFactory.create() mock_models.ReferenceSapProfitCenter.build.return_value = mock_profit_center mock_db.session.commit.return_value = True mock_validation.return_value = True mock_post_request_body = { 'profit_center': 'UK9999', 'company_code': '4912', 'business_group': 'ORC', 'display_name': 'New Profit Center', } result = logic.create_reference_sap_profit_center(**mock_post_request_body) assert result.status == 201 mock_validation.assert_called_once_with([]) mock_models.ReferenceSapProfitCenter.build.assert_called_once_with( **mock_post_request_body ) mock_models.SigningEntitySapProfitCenter.build.assert_not_called() mock_db.session.commit.assert_called_once() @patch('abacus_contract.logic.reference_sap_profit_center.db') @patch( 'abacus_contract.logic.reference_sap_profit_center._check_whether_signing_entities_exist' ) @patch('abacus_contract.logic.reference_sap_profit_center.models') def test_create_reference_sap_profit_center_maps_signing_entities( mock_models, mock_validation, mock_db ): """Valid reference_signing_entity_ids -> a junction row is created per id.""" mock_profit_center = ReferenceSapProfitCenterFactory.create() mock_db.session.commit.return_value = True mock_signing_sap_profit_centers = SigningEntitySapProfitCenterFactory.create_batch( 2 ) signing_entity_ids = [ profit_center.reference_signing_entity_id for profit_center in mock_signing_sap_profit_centers ] mock_models.ReferenceSapProfitCenter.build.return_value = mock_profit_center mock_validation.return_value = True mock_post_request_body = { 'profit_center': 'UK9999', 'company_code': '4912', 'business_group': 'ORC', 'display_name': 'New Profit Center', 'reference_signing_entity_ids': signing_entity_ids, } result = logic.create_reference_sap_profit_center(**mock_post_request_body) del mock_post_request_body['reference_signing_entity_ids'] assert result.status == 201 mock_validation.assert_called_once_with(signing_entity_ids) mock_models.ReferenceSapProfitCenter.build.assert_called_once_with( **mock_post_request_body ) mock_models.SigningEntitySapProfitCenter.build.assert_has_calls( [ call( reference_signing_entity_id=signing_entity_ids[0], reference_sap_profit_center_id=mock_profit_center.reference_sap_profit_center_id, ), call( reference_signing_entity_id=signing_entity_ids[1], reference_sap_profit_center_id=mock_profit_center.reference_sap_profit_center_id, ), ] ) mock_db.session.commit.assert_called_once() @patch('abacus_contract.logic.reference_sap_profit_center.db') @patch( 'abacus_contract.logic.reference_sap_profit_center._check_whether_signing_entities_exist' ) @patch('abacus_contract.logic.reference_sap_profit_center.models') def test_create_reference_sap_profit_center_dedupes_duplicate_signing_entity_ids( mock_models, mock_validation, mock_db ): """Duplicate ids in the input only produce one junction row per unique id.""" mock_profit_center = ReferenceSapProfitCenterFactory.create() mock_signing_sap_profit_centers = SigningEntitySapProfitCenterFactory.create_batch( 2 ) signing_entity_ids = [ profit_center.reference_signing_entity_id for profit_center in mock_signing_sap_profit_centers ] mock_models.ReferenceSapProfitCenter.build.return_value = mock_profit_center mock_validation.return_value = True mock_post_request_body = { 'profit_center': 'UK9999', 'company_code': '4912', 'business_group': 'ORC', 'display_name': 'New Profit Center', 'reference_signing_entity_ids': signing_entity_ids * 2, } result = logic.create_reference_sap_profit_center(**mock_post_request_body) del mock_post_request_body['reference_signing_entity_ids'] assert result.status == 201 mock_validation.assert_called_once_with(signing_entity_ids) mock_models.ReferenceSapProfitCenter.build.assert_called_once_with( **mock_post_request_body ) mock_models.SigningEntitySapProfitCenter.build.assert_has_calls( [ call( reference_signing_entity_id=signing_entity_ids[0], reference_sap_profit_center_id=mock_profit_center.reference_sap_profit_center_id, ), call( reference_signing_entity_id=signing_entity_ids[1], reference_sap_profit_center_id=mock_profit_center.reference_sap_profit_center_id, ), ] ) mock_db.session.commit.assert_called_once() @patch('abacus_contract.logic.reference_sap_profit_center.db') def test_create_reference_sap_profit_center_integrity_error_rolls_back(mock_db): """IntegrityError on commit is rolled back and re-raised (no 409 translation here).""" mock_db.session.commit.side_effect = IntegrityError( statement=None, params=None, orig=Exception('unique constraint') ) with pytest.raises(IntegrityError): logic.create_reference_sap_profit_center( profit_center='UK9993', company_code='4912', business_group='ORC', display_name='Boom', ) mock_db.session.rollback.assert_called_once() @patch('abacus_contract.logic.reference_sap_profit_center.db') def test_create_reference_sap_profit_center_sqlalchemy_error_rolls_back(mock_db): """Generic SQLAlchemyError on commit is rolled back and re-raised.""" mock_db.session.commit.side_effect = SQLAlchemyError('connection lost') with pytest.raises(SQLAlchemyError): logic.create_reference_sap_profit_center( profit_center='UK9992', company_code='4912', business_group='ORC', display_name='Boom Again', ) mock_db.session.rollback.assert_called_once() # # --- _check_whether_signing_entities_exist --- def test_check_whether_signing_entities_exist_returns_true_for_empty_input(): """No ids to check -> trivially True, no query performed.""" assert logic._check_whether_signing_entities_exist([]) is True @patch('abacus_contract.logic.reference_sap_profit_center.models') def test_check_whether_signing_entities_exist_returns_true_when_all_present( mock_models, ): """All ids exist -> True.""" se1 = ReferenceSigningEntityFactory.create(with_junction_row=False) se2 = ReferenceSigningEntityFactory.create(with_junction_row=False) mock_models.ReferenceSigningEntity.get_by_ids.return_value = [se1, se2] result = logic._check_whether_signing_entities_exist( [se1.reference_signing_entity_id, se2.reference_signing_entity_id] ) assert result is True @patch('abacus_contract.logic.reference_sap_profit_center.models') def test_check_whether_signing_entities_exist_raises_for_missing_ids(mock_models): """Any missing id raises ValidationError naming the missing ids.""" se = ReferenceSigningEntityFactory.create(with_junction_row=False) mock_models.ReferenceSigningEntity.get_by_ids.return_value = [se] with pytest.raises(ValidationError) as exc_info: logic._check_whether_signing_entities_exist( [se.reference_signing_entity_id, 888888888] ) assert '888888888' in str(exc_info.value)