"""Functional tests for ReferenceSigningEntity.""" from datetime import datetime, timezone from unittest.mock import patch import pytest from abacus_common_logic.constants.constants import SYSTEM_TIMEZONE from freezegun import freeze_time from abacus_contract import models from abacus_contract.tests.utils.factories import ( ReferenceSapProfitCenterFactory, ReferenceSigningEntityFactory, SigningEntitySapProfitCenterFactory, ) from core.config import Config def test_get_reference_signing_entity_by_id(fixture_client): """Test to get ReferenceSigningEntity by ID.""" reference_signing_entity = ReferenceSigningEntityFactory.create(company_code='4914') reference_signing_entity_id = reference_signing_entity.reference_signing_entity_id res = fixture_client.get(f'/reference-signing-entity/{reference_signing_entity_id}') assert res.status_code == 200 cache_control = res.headers.get('Cache-Control', '') assert 'max-age=300' in cache_control assert 'public' in cache_control assert res.json == { 'reference_signing_entity_id': reference_signing_entity.reference_signing_entity_id, 'reference_payment_entity_id': reference_signing_entity.reference_payment_entity_id, 'reference_sap_profit_center_id': reference_signing_entity.reference_sap_profit_center_id, 'company_code': reference_signing_entity.company_code, 'legal_name': reference_signing_entity.legal_name, 'vat_number': reference_signing_entity.vat_number, 'company_registration_number': reference_signing_entity.company_registration_number, 'tax_entity_company_code': reference_signing_entity.tax_entity_company_code, 'address': reference_signing_entity.address, } def test_get_reference_signing_entity_list(fixture_client): """Test to get a list of reference_signing_entity.""" company_codes = ['4915', '4916', '4917'] [ ReferenceSigningEntityFactory.create(company_code=company_code) for company_code in ['4915', '4916', '4917'] ] res = fixture_client.get('/reference-signing-entities/') assert res.status_code == 200 items = res.json.get('items') assert len(company_codes) == res.json.get('total_count') assert items[0]['company_code'] == company_codes[0] assert items[1]['company_code'] == company_codes[1] assert items[2]['company_code'] == company_codes[2] def test_get_reference_signing_entity_by_id_returns_404_when_deleted(fixture_client): """Test that a soft-deleted ReferenceSigningEntity returns 404.""" reference_signing_entity = ReferenceSigningEntityFactory.create( deleted_at=datetime(2025, 1, 1), deleted_by='test_user', ) reference_signing_entity_id = reference_signing_entity.reference_signing_entity_id res = fixture_client.get( f'/reference-signing-entity/{reference_signing_entity_id}/' ) assert res.status_code == 404 def test_get_reference_signing_entity_list_excludes_deleted(fixture_client): """Test that soft-deleted ReferenceSigningEntities are excluded from the list.""" ReferenceSigningEntityFactory.create(company_code='4915') ReferenceSigningEntityFactory.create(company_code='4916') ReferenceSigningEntityFactory.create( company_code='4917', deleted_at=datetime(2025, 1, 1), deleted_by='test_user', ) res = fixture_client.get('/reference-signing-entities/') assert res.status_code == 200 assert res.json.get('total_count') == 2 returned_codes = [item['company_code'] for item in res.json.get('items')] assert '4915' in returned_codes assert '4916' in returned_codes assert '4917' not in returned_codes # --- GET /reference-signing-entity//reference-sap-profit-centers/ --- @pytest.mark.db('mysql') def test_get_profit_centers_for_signing_entity_returns_only_authorized_live( fixture_client, ): """Dropdown endpoint surfaces only PCs with live junction rows for the SE.""" se = ReferenceSigningEntityFactory.create(with_junction_row=False) authorized_pc = ReferenceSapProfitCenterFactory.create(display_name='Authorized') deleted_pc = ReferenceSapProfitCenterFactory.create(display_name='Was authorized') _unrelated_pc = ReferenceSapProfitCenterFactory.create(display_name='Unrelated') SigningEntitySapProfitCenterFactory.create( reference_signing_entity=se, reference_sap_profit_center=authorized_pc ) soft_deleted_junction = SigningEntitySapProfitCenterFactory.create( reference_signing_entity=se, reference_sap_profit_center=deleted_pc ) models.SigningEntitySapProfitCenter.delete_by_id_or_error( soft_deleted_junction.signing_entity_sap_profit_center_id, soft_delete=True ) res = fixture_client.get( f'/reference-signing-entity/{se.reference_signing_entity_id}' '/reference-sap-profit-centers/' ) assert res.status_code == 200, res.json assert res.json['total_count'] == 1 [item] = res.json['items'] assert item['sap_profit_center']['reference_sap_profit_center_id'] == ( authorized_pc.reference_sap_profit_center_id ) assert item['sap_profit_center']['display_name'] == 'Authorized' @pytest.mark.db('mysql') def test_get_profit_centers_for_signing_entity_search_filters_display_and_profit_center( fixture_client, ): """?search_term matches against display_name or profit_center.""" se = ReferenceSigningEntityFactory.create(with_junction_row=False) pc_display = ReferenceSapProfitCenterFactory.create( profit_center='AAA001', display_name='Foundation Media' ) pc_code = ReferenceSapProfitCenterFactory.create( profit_center='SANTA42', display_name='Other label' ) pc_none = ReferenceSapProfitCenterFactory.create( profit_center='ZZZ999', display_name='Different' ) for pc in (pc_display, pc_code, pc_none): SigningEntitySapProfitCenterFactory.create( reference_signing_entity=se, reference_sap_profit_center=pc ) res_display = fixture_client.get( f'/reference-signing-entity/{se.reference_signing_entity_id}' '/reference-sap-profit-centers/?search_term=foundation' ) assert { item['sap_profit_center']['reference_sap_profit_center_id'] for item in res_display.json['items'] } == {pc_display.reference_sap_profit_center_id} res_code = fixture_client.get( f'/reference-signing-entity/{se.reference_signing_entity_id}' '/reference-sap-profit-centers/?search_term=santa' ) assert { item['sap_profit_center']['reference_sap_profit_center_id'] for item in res_code.json['items'] } == {pc_code.reference_sap_profit_center_id} @pytest.mark.db('mysql') def test_get_profit_centers_for_signing_entity_pagination(fixture_client): """?limit + ?offset paginate; total_count reflects full set.""" se = ReferenceSigningEntityFactory.create(with_junction_row=False) for n in range(5): pc = ReferenceSapProfitCenterFactory.create(display_name=f'PC {n}') SigningEntitySapProfitCenterFactory.create( reference_signing_entity=se, reference_sap_profit_center=pc ) res = fixture_client.get( f'/reference-signing-entity/{se.reference_signing_entity_id}' '/reference-sap-profit-centers/?limit=2&offset=0' ) assert res.status_code == 200, res.json assert res.json['total_count'] == 5 assert len(res.json['items']) == 2 @pytest.mark.db('mysql') def test_get_profit_centers_for_signing_entity_empty(fixture_client): """SE with no live mappings -> 200 with empty items.""" se = ReferenceSigningEntityFactory.create(with_junction_row=False) res = fixture_client.get( f'/reference-signing-entity/{se.reference_signing_entity_id}' '/reference-sap-profit-centers/' ) assert res.status_code == 200, res.json assert res.json == {'items': [], 'total_count': 0} @pytest.mark.db('mysql') def test_get_profit_centers_for_signing_entity_unknown_id_returns_404(fixture_client): """Unknown SE id -> 404.""" res = fixture_client.get( '/reference-signing-entity/99999999/reference-sap-profit-centers/' ) assert res.status_code == 404, res.data @pytest.mark.db('mysql') def test_get_profit_centers_for_signing_entity_non_numeric_limit_returns_400( fixture_client, ): """Non-numeric ?limit= surfaces as a client error (not a 500).""" se = ReferenceSigningEntityFactory.create(with_junction_row=False) res = fixture_client.get( f'/reference-signing-entity/{se.reference_signing_entity_id}' '/reference-sap-profit-centers/?limit=abc' ) assert res.status_code < 500, res.data assert res.status_code >= 400, res.data @pytest.mark.db('mysql') def test_get_profit_centers_for_signing_entity_non_numeric_offset_returns_400( fixture_client, ): """Non-numeric ?offset= surfaces as a client error (not a 500).""" se = ReferenceSigningEntityFactory.create(with_junction_row=False) res = fixture_client.get( f'/reference-signing-entity/{se.reference_signing_entity_id}' '/reference-sap-profit-centers/?offset=xyz' ) assert res.status_code < 500, res.data assert res.status_code >= 400, res.data @freeze_time(datetime(2026, 6, 9, 0, 0, 0, tzinfo=SYSTEM_TIMEZONE)) @pytest.mark.db('mysql') def test_get_profit_centers_dataloader_returns_active_mappings( fixture_client, ): """Test that endpoint returns all active, live SAP profit center mappings for the given signing entity IDs.""" # noqa: E501 mock_signing_entities = [ ReferenceSigningEntityFactory.create(with_junction_row=False), ReferenceSigningEntityFactory.create(with_junction_row=False), ReferenceSigningEntityFactory.create(with_junction_row=False), ] mock_profit_centers = [ ReferenceSapProfitCenterFactory.create( profit_center='AAA001', display_name='Foundation Media' ), ReferenceSapProfitCenterFactory.create( profit_center='SANTA42', display_name='Other label' ), ReferenceSapProfitCenterFactory.create( profit_center='USD998', display_name='BlastBeat' ), ] mock_mappings = [ SigningEntitySapProfitCenterFactory.create( reference_signing_entity=mock_signing_entities[0], reference_sap_profit_center=pc, created_at=datetime(2026, 6, 9, 0, 0, 0, tzinfo=timezone.utc), ) for pc in mock_profit_centers[0:2] ] mock_mappings.append( SigningEntitySapProfitCenterFactory.create( reference_signing_entity=mock_signing_entities[1], reference_sap_profit_center=mock_profit_centers[2], ) ) signing_entity_ids = [ mock_signing_entity.reference_signing_entity_id for mock_signing_entity in mock_signing_entities ] result = fixture_client.post( '/reference-signing-entity/reference-sap-profit-centers/dataloader', json=signing_entity_ids + signing_entity_ids, ) assert result.status_code == 200 assert result.json == [ { 'data': [ { 'sap_profit_center': { 'business_group': 'ORC', 'company_code': '4912', 'display_name': 'Foundation Media', 'profit_center': 'AAA001', 'reference_sap_profit_center_id': mock_profit_centers[ 0 ].reference_sap_profit_center_id, }, 'created_at': '2026-06-09T00:00:00.000000', 'reference_signing_entity_id': mock_signing_entities[ 0 ].reference_signing_entity_id, 'signing_entity_sap_profit_center_id': mock_mappings[ 0 ].signing_entity_sap_profit_center_id, }, { 'sap_profit_center': { 'business_group': 'ORC', 'company_code': '4912', 'display_name': 'Other label', 'profit_center': 'SANTA42', 'reference_sap_profit_center_id': mock_profit_centers[ 1 ].reference_sap_profit_center_id, }, 'created_at': '2026-06-09T00:00:00.000000', 'reference_signing_entity_id': mock_signing_entities[ 0 ].reference_signing_entity_id, 'signing_entity_sap_profit_center_id': mock_mappings[ 1 ].signing_entity_sap_profit_center_id, }, ], }, { 'data': [ { 'sap_profit_center': { 'business_group': 'ORC', 'company_code': '4912', 'display_name': 'BlastBeat', 'profit_center': 'USD998', 'reference_sap_profit_center_id': mock_profit_centers[ 2 ].reference_sap_profit_center_id, }, 'created_at': '2026-06-09T00:00:00.000000', 'reference_signing_entity_id': mock_signing_entities[ 1 ].reference_signing_entity_id, 'signing_entity_sap_profit_center_id': mock_mappings[ 2 ].signing_entity_sap_profit_center_id, }, ], }, { 'data': None, }, ] @pytest.mark.db('mysql') def test_get_profit_centers_dataloader_returns_empty_when_unmapped( fixture_client, ): """Test that an empty list is returned when the provided signing entity IDs have no active profit center mappings.""" # noqa: E501 mock_signing_entities = [ ReferenceSigningEntityFactory.create(with_junction_row=False), ReferenceSigningEntityFactory.create(with_junction_row=False), ReferenceSigningEntityFactory.create(with_junction_row=False), ] mock_profit_centers = [ ReferenceSapProfitCenterFactory.create( profit_center='AAA001', display_name='Foundation Media', ), ReferenceSapProfitCenterFactory.create( profit_center='SANTA42', display_name='Other label', ), ] SigningEntitySapProfitCenterFactory.create( reference_signing_entity=mock_signing_entities[0], reference_sap_profit_center=mock_profit_centers[0], deleted_by='Test User', ) SigningEntitySapProfitCenterFactory.create( reference_signing_entity=mock_signing_entities[1], reference_sap_profit_center=mock_profit_centers[1], deleted_by='Test User', ) signing_entity_ids = [ mock_signing_entity.reference_signing_entity_id for mock_signing_entity in mock_signing_entities ] result = fixture_client.post( '/reference-signing-entity/reference-sap-profit-centers/dataloader', json=signing_entity_ids, ) assert result.status_code == 200 assert result.json == [ {'data': None}, {'data': None}, {'data': None}, ] @patch( 'abacus_contract.blueprints.reference_signing_entity.flask_request.verify_rules_access_standalone' ) def test_get_profit_centers_by_signing_entity_dataloader_forbidden( mock_verify_rules, fixture_client ): """Test that 403 is returned when the user is not authorized to get a list of profit centers.""" mock_verify_rules.return_value = False signing_entity_ids = [1, 2] result = fixture_client.post( '/reference-signing-entity/reference-sap-profit-centers/dataloader', json=signing_entity_ids, ) assert result.status_code == 401 assert result.json['message'] == 'Unauthorized' def test_get_profit_centers_dataloader_invalid_request_data(fixture_client): """Test that endpoint returns a 400 Bad Request when passed an invalid request body payload.""" signing_entity_ids = [1, 2, 'test'] result = fixture_client.post( '/reference-signing-entity/reference-sap-profit-centers/dataloader', json=signing_entity_ids, ) assert result.status_code == 400 assert result.json['message'] == 'ReferenceSigningEntity IDs must be integers' @patch( 'abacus_contract.blueprints.reference_signing_entity.flask_request.verify_rules_access_standalone' ) def test_get_signing_entities_dataloader_forbidden(mock_verify_rules, fixture_client): """Test that 403 is returned when the user is not authorized to get a list of signing entities.""" mock_verify_rules.return_value = False signing_entity_ids = [1, 2] result = fixture_client.post( '/reference-signing-entities/dataloader', json=signing_entity_ids, ) assert result.status_code == 401 assert result.json['message'] == 'Unauthorized' def test_get_signing_entities_dataloader_empty_list_data(fixture_client): """Test that endpoint returns a 400 Bad Request when passed an empty list in payload.""" signing_entity_ids = [] result = fixture_client.post( '/reference-signing-entities/dataloader', json=signing_entity_ids, ) assert result.status_code == 400 assert result.json['message'] == 'ReferenceSigningEntity IDs must be integers' @pytest.mark.db('mysql') def test_get_signing_entities_dataloader_for_non_matching_ids( fixture_client, ): """Test that an empty list is returned when the provided non-existent IDs returns.""" signing_entity_ids = [99999, 1999999] result = fixture_client.post( '/reference-signing-entities/dataloader', json=signing_entity_ids, ) assert result.status_code == 200 assert result.json == [ {'data': None}, {'data': None}, ] @pytest.mark.db('mysql') def test_get_signing_entities_dataloader( fixture_client, ): """Test that endpoint returns signing entities for the given signing entity IDs.""" mock_signing_entities = [ ReferenceSigningEntityFactory.create( with_junction_row=False, company_code='0020' ), ReferenceSigningEntityFactory.create( with_junction_row=False, company_code='0021' ), ReferenceSigningEntityFactory.create( with_junction_row=False, company_code='0022' ), ] signing_entity_ids = [ mock_signing_entity.reference_signing_entity_id for mock_signing_entity in mock_signing_entities ] result = fixture_client.post( '/reference-signing-entities/dataloader', json=signing_entity_ids + signing_entity_ids, ) assert result.status_code == 200 assert result.json == [ { 'data': { 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'reference_payment_entity_id': 1, 'company_code': '0020', 'legal_name': 'AWAL Digital Limited', 'vat_number': 'GB 423 4787 86', 'company_registration_number': '04430703', 'tax_entity_company_code': None, 'address': '2 Canal Reach, London, N1C 4DB', } }, { 'data': { 'reference_signing_entity_id': 2, 'reference_sap_profit_center_id': 2, 'reference_payment_entity_id': 2, 'company_code': '0021', 'legal_name': 'AWAL Digital Limited', 'vat_number': 'GB 423 4787 86', 'company_registration_number': '04430703', 'tax_entity_company_code': None, 'address': '2 Canal Reach, London, N1C 4DB', } }, { 'data': { 'reference_signing_entity_id': 3, 'reference_sap_profit_center_id': 3, 'reference_payment_entity_id': 3, 'company_code': '0022', 'legal_name': 'AWAL Digital Limited', 'vat_number': 'GB 423 4787 86', 'company_registration_number': '04430703', 'tax_entity_company_code': None, 'address': '2 Canal Reach, London, N1C 4DB', } }, ] def test_get_signing_entities_dataloader_excludes_soft_deleted(fixture_client): """A soft-deleted signing entity is not returned by the batched dataloader.""" live = ReferenceSigningEntityFactory.create( with_junction_row=False, company_code='0030' ) deleted = ReferenceSigningEntityFactory.create( with_junction_row=False, company_code='0031', deleted_at=datetime(2026, 1, 1, tzinfo=timezone.utc), deleted_by='someone', ) result = fixture_client.post( '/reference-signing-entities/dataloader', json=[ live.reference_signing_entity_id, deleted.reference_signing_entity_id, ], ) assert result.status_code == 200 # Live id -> its data; soft-deleted id -> null (excluded, not resurrected). assert ( result.json[0]['data']['reference_signing_entity_id'] == live.reference_signing_entity_id ) assert result.json[1]['data'] is None def test_get_signing_entities_dataloader_over_limit_returns_400(fixture_client): """More than OWS_BATCH_LIMIT ids are rejected before touching the DB.""" result = fixture_client.post( '/reference-signing-entities/dataloader', json=list(range(Config.OWS_BATCH_LIMIT + 1)), ) assert result.status_code == 400 def test_get_signing_entities_dataloader_null_id_returns_400(fixture_client): """A null id raises TypeError (not ValueError); must still be 400, not 500.""" result = fixture_client.post('/reference-signing-entities/dataloader', json=[None]) assert result.status_code == 400 @pytest.mark.db('mysql') def test_get_reference_signing_entities(fixture_client): """Test to get a list of signing entities.""" ReferenceSigningEntityFactory.create( company_code='0001', legal_name='ScreenMedia', with_junction_row=False ) ReferenceSigningEntityFactory.create( company_code='0002', legal_name='JIVE Records', with_junction_row=False ) ReferenceSigningEntityFactory.create( company_code='0003', legal_name='Other label', with_junction_row=False ) ReferenceSigningEntityFactory.create( company_code='0004', legal_name='RCA Records', with_junction_row=False ) result = fixture_client.get( '/reference-signing-entities/?limit=4&offset=0&sort_by=company_code&sort_order=desc' ) assert result.status_code == 200 assert result.json['total_count'] == 4 assert result.json['items'] == [ { 'reference_signing_entity_id': 4, 'reference_sap_profit_center_id': 4, 'reference_payment_entity_id': 4, 'company_code': '0004', 'legal_name': 'RCA Records', 'vat_number': 'GB 423 4787 86', 'company_registration_number': '04430703', 'tax_entity_company_code': None, 'address': '2 Canal Reach, London, N1C 4DB', }, { 'reference_signing_entity_id': 3, 'reference_sap_profit_center_id': 3, 'reference_payment_entity_id': 3, 'company_code': '0003', 'legal_name': 'Other label', 'vat_number': 'GB 423 4787 86', 'company_registration_number': '04430703', 'tax_entity_company_code': None, 'address': '2 Canal Reach, London, N1C 4DB', }, { 'reference_signing_entity_id': 2, 'reference_sap_profit_center_id': 2, 'reference_payment_entity_id': 2, 'company_code': '0002', 'legal_name': 'JIVE Records', 'vat_number': 'GB 423 4787 86', 'company_registration_number': '04430703', 'tax_entity_company_code': None, 'address': '2 Canal Reach, London, N1C 4DB', }, { 'reference_signing_entity_id': 1, 'reference_sap_profit_center_id': 1, 'reference_payment_entity_id': 1, 'company_code': '0001', 'legal_name': 'ScreenMedia', 'vat_number': 'GB 423 4787 86', 'company_registration_number': '04430703', 'tax_entity_company_code': None, 'address': '2 Canal Reach, London, N1C 4DB', }, ]