"""Statement attachment model tests.""" from datetime import datetime from datetime import timedelta from fastapi import HTTPException import pytest from moneyhub.constants.constants import NumberFormat from moneyhub.constants.constants import StatementAttachmentFileType from moneyhub.constants.constants import StatementAttachmentStatus from moneyhub.constants.constants import StatementAttachmentType from moneyhub.models.statement_attachment import StatementAttachment from tests.unit.conftest import DISTRIBUTION_FEE_INVOICE from tests.unit.conftest import insert_mock_account_payment_terms from tests.unit.conftest import insert_mock_data from tests.unit.conftest import SELF_BILLING_INVOICE def _insert_mock_attachment_data( account_id, statement_period_id, subaccount_id=None, statement_period_ids=None): """Insert mock data used in several tests.""" insert_mock_data({ 'account': [ {'account_id': account_id}, {'account_id': account_id + 1}, ], 'reference_payment_entity': {'reference_payment_entity_id': 1}, 'reference_sap_profit_center': { 'reference_sap_profit_center_id': 1, 'profit_center': 'UK1234', 'company_code': '1234', 'business_group': 'ORC' }, 'reference_signing_entity': { 'reference_signing_entity_id': 1, 'reference_payment_entity_id': 1, 'reference_sap_profit_center_id': 1, 'company_code': '1234', 'tax_entity_company_code': '1234', 'legal_name': 'Company', 'vat_number': '12341234', 'company_registration_number': '12341234', 'address': None, }, 'contract': [ { 'contract_id': 10001, 'reference_signing_entity_id': 1, }, { 'contract_id': 10002, 'reference_signing_entity_id': 1, }, ], 'statement_period': [ {'statement_period_id': 123, 'statement_period_status': 'closed'}, {'statement_period_id': 124, 'statement_period_status': 'closed'}, ], 'statement_attachment': [ { 'account_id': account_id, 'subaccount_id': None, 'contract_id': 10001, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.DISTRIBUTION_FEE_INVOICE, 'statement_period_id': statement_period_id, 'statement_period_ids': statement_period_ids, 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_at': datetime.utcnow() - timedelta(hours=1), 'created_by': 'test' }, { 'account_id': account_id, 'subaccount_id': subaccount_id, 'contract_id': 10001, 'statement_attachment_status': StatementAttachmentStatus.COMPLETE, 'statement_attachment_type': StatementAttachmentType.REVENUE_DETAIL, 'statement_period_id': statement_period_id, 'statement_period_ids': None, 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_at': datetime.now(), 'created_by': 'test' }, { 'account_id': account_id, 'subaccount_id': None, 'contract_id': 10002, 'statement_attachment_status': StatementAttachmentStatus.ERROR, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'statement_period_id': statement_period_id, 'statement_period_ids': None, 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_at': datetime.now(), 'created_by': 'test' }, { 'account_id': account_id + 1, 'subaccount_id': None, 'contract_id': 10001, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.DISTRIBUTION_FEE_INVOICE, 'statement_period_id': statement_period_id, 'statement_period_ids': None, 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_at': datetime.utcnow() - timedelta(hours=1), 'created_by': 'test' }, { 'account_id': account_id + 1, 'subaccount_id': subaccount_id, 'contract_id': 10001, 'statement_attachment_status': StatementAttachmentStatus.COMPLETE, 'statement_attachment_type': StatementAttachmentType.REVENUE_DETAIL, 'statement_period_id': statement_period_id, 'statement_period_ids': None, 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_at': datetime.now(), 'created_by': 'test' }, { 'account_id': account_id + 1, 'subaccount_id': None, 'contract_id': 10001, 'statement_attachment_status': StatementAttachmentStatus.ERROR, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'statement_period_id': statement_period_id, 'statement_period_ids': None, 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_at': datetime.now(), 'created_by': 'test' }, { 'account_id': account_id + 1, 'subaccount_id': subaccount_id, 'contract_id': 10001, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.REVENUE_DETAIL, 'statement_period_id': 124, 'statement_period_ids': None, 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_at': datetime.utcnow() - timedelta(hours=1), 'created_by': 'test' }, ], }) @pytest.mark.parametrize('statement_period_id', [1, 2, 3]) def test_delete_by_id(statement_period_id): """Test deleting an attachment.""" _insert_mock_attachment_data(24601, 123) attachment = StatementAttachment.get_by_id(statement_period_id) assert attachment is not None result = StatementAttachment.delete_by_id(statement_period_id) assert result == attachment assert StatementAttachment.get_by_id(statement_period_id) is None @pytest.mark.parametrize('statement_period_id', [111, 222, 333]) def test_delete_by_id_missing(statement_period_id): """Test deleting an attachment that doesn't exist.""" _insert_mock_attachment_data(24601, 123) attachment = StatementAttachment.get_by_id(statement_period_id) assert attachment is None with pytest.raises(HTTPException) as excinfo: StatementAttachment.delete_by_id(statement_period_id) assert excinfo.value.detail == f'StatementAttachment {statement_period_id} does not exist.' def test_get_by_statement_period(): """Test getting attachments by statement period.""" statement_period_id = 123 _insert_mock_attachment_data(24601, statement_period_id) statement_attachments = StatementAttachment.get_by_statement_period(statement_period_id) assert len(statement_attachments) == 6 def test_get_by_statement_period_with_account(): """Test getting attachments by statement period and account.""" statement_period_id = 123 account_id = 24601 _insert_mock_attachment_data(account_id, statement_period_id) statement_attachments = StatementAttachment.get_by_statement_period( statement_period_id, account_id=account_id) assert len(statement_attachments) == 3 def test_get_by_statement_period_with_contract(): """Test getting attachments by statement period and contract.""" statement_period_id = 123 account_id = 24601 contract_id = 10002 _insert_mock_attachment_data(account_id, statement_period_id) statement_attachments = StatementAttachment.get_by_statement_period( statement_period_id, contract_id=contract_id) assert len(statement_attachments) == 1 def test_get_by_statement_period_with_subaccount(): """Test getting attachments by statement period and subaccount.""" statement_period_id = 123 account_id = 24601 subaccount_id = 10002 _insert_mock_attachment_data(account_id, statement_period_id, subaccount_id) statement_attachments = StatementAttachment.get_by_statement_period( statement_period_id, account_id=account_id, subaccount_id=subaccount_id) assert len(statement_attachments) == 1 def test_get_by_statement_period_with_no_account(): """Test getting attachments by statement period with no account.""" statement_period_id = 123 account_id = 24601 subaccount_id = 10002 _insert_mock_attachment_data(account_id, statement_period_id, subaccount_id) statement_attachments = StatementAttachment.get_by_statement_period( statement_period_id, subaccount_id=subaccount_id) assert len(statement_attachments) == 6 def test_get_by_statement_period_with_statuses(): """Test getting attachments by statement period and statuses.""" statement_period_id = 123 _insert_mock_attachment_data(24601, statement_period_id) statement_attachments = StatementAttachment.get_by_statement_period( statement_period_id, statuses=[ StatementAttachmentStatus.IN_PROGRESS, StatementAttachmentStatus.ERROR]) assert len(statement_attachments) == 4 def test_get_by_statement_period_with_types(): """Test getting attachments by statement period and types.""" statement_period_id = 123 _insert_mock_attachment_data(24601, statement_period_id) statement_attachments = StatementAttachment.get_by_statement_period( statement_period_id, types=[StatementAttachmentType.DISTRIBUTION_FEE_INVOICE]) assert len(statement_attachments) == 2 def test_get_by_statement_period_all_filters(): """Test getting attachments by statement period and types.""" statement_period_id = 123 account_id = 24601 _insert_mock_attachment_data(account_id, statement_period_id) statement_attachments = StatementAttachment.get_by_statement_period( statement_period_id, account_id=account_id, statuses=[ StatementAttachmentStatus.IN_PROGRESS, StatementAttachmentStatus.ERROR], types=[StatementAttachmentType.DISTRIBUTION_FEE_INVOICE]) assert len(statement_attachments) == 1 def test_get_invoices_by_payment_entity(): """Test getting statement attachment invoices by payment entity.""" account_id = 1 statement_period_id = 123 _insert_mock_attachment_data(account_id, statement_period_id) insert_mock_account_payment_terms() payment_entity_id = 1 statement_attachments = StatementAttachment.get_invoices_by_payment_entity( payment_entity_id, statement_period_id) assert len(statement_attachments) == 4 for statement in statement_attachments: assert statement.statement_attachment_type in \ [DISTRIBUTION_FEE_INVOICE, SELF_BILLING_INVOICE] def test_get_failed_invoices_by_payment_entity(): """Test getting failed statement attachment invoices by payment entity.""" account_id = 1 statement_period_id = 123 _insert_mock_attachment_data(account_id, statement_period_id) insert_mock_account_payment_terms() payment_entity_id = 1 statement_attachments = StatementAttachment.get_failed_invoices_by_payment_entity( payment_entity_id, statement_period_id) assert len(statement_attachments) == 2 for statement in statement_attachments: assert statement.statement_attachment_type in \ [DISTRIBUTION_FEE_INVOICE, SELF_BILLING_INVOICE] for statement in statement_attachments: assert statement.statement_attachment_status == StatementAttachmentStatus.ERROR def test_get_stuck_in_progress(): """Test getting attachments stuck in progress.""" _insert_mock_attachment_data(24601, 124) statement_attachments = StatementAttachment.get_stuck_in_progress() assert len(statement_attachments) == 3 def test_get_by_statement_periods(): """Test getting attachments by statement periods and account.""" statement_period_ids = [123, 124] _insert_mock_attachment_data(24601, statement_period_ids[0]) statement_attachments = StatementAttachment.get_by_account_id_and_statement_periods( 24601, None, statement_period_ids) assert len(statement_attachments) == 3 def test_get_by_statement_periods_subaccount(): """Test getting subaccount attachments by statement periods and account.""" statement_period_ids = [123, 124] subaccount_id = 10002 _insert_mock_attachment_data(24601, statement_period_ids[0], subaccount_id) statement_attachments = StatementAttachment.get_by_account_id_and_statement_periods( 24601, None, statement_period_ids) assert len(statement_attachments) == 2 def test_exists(): """Test checking if a statement attachment exists.""" statement_period_id = 123 statement_period_ids = '123,456' account_id = 24601 file_location = None _insert_mock_attachment_data( account_id, statement_period_id, statement_period_ids=statement_period_ids) result = StatementAttachment.exists( account_id, statement_period_id, file_location, statement_period_ids=statement_period_ids) assert result def test_exists_with_statement_attachment_id(): """Test checking if a statement attachment exists.""" statement_attachment_id = 1 statement_period_id = 123 statement_period_ids = '123,456' account_id = 24601 _insert_mock_attachment_data( account_id, statement_period_id, statement_period_ids=statement_period_ids) result = StatementAttachment.exists(statement_attachment_id=statement_attachment_id) assert result def test_exists_with_subaccount(): """Test checking if a subaccount statement attachment exists.""" statement_period_id = 123 account_id = 24601 subaccount_id = 10002 file_location = None _insert_mock_attachment_data(account_id, statement_period_id, subaccount_id) result = StatementAttachment.exists( account_id=account_id, statement_period_id=statement_period_id, file_location=file_location, subaccount_id=subaccount_id) assert result def test_exists_without_subaccount(): """Test checking if a statement attachment exists without a subaccount.""" statement_period_id = 123 account_id = 24601 contract_id = 10001 subaccount_id = 123 file_location = None _insert_mock_attachment_data(account_id, statement_period_id, subaccount_id) result = StatementAttachment.exists( account_id=account_id, contract_id=contract_id, statement_period_id=statement_period_id, file_location=file_location, attachment_type=StatementAttachmentType.REVENUE_DETAIL, subaccount_id=None) assert not result def test_not_exists(): """Test checking if a statement attachment does not exist.""" statement_period_id = 123 account_id = 24601 file_location = 's3://fake-file' _insert_mock_attachment_data(account_id, statement_period_id) result = StatementAttachment.exists(account_id, statement_period_id, file_location) assert not result def test_create_with_filters(): """Test creating a statement attachment with filters.""" account_id = 24601 statement_period_id = 123 filters = { 'transaction_type_ids': [1, 2, 3], 'transaction_type_group_ids': [4, 5, 6] } _insert_mock_attachment_data(account_id, statement_period_id) attachment = StatementAttachment.create( account_id=account_id, contract_id=10001, statement_period_id=statement_period_id, statement_attachment_type=StatementAttachmentType.LEGACY_REVENUE_DETAIL_FULL, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, file_type=StatementAttachmentFileType.XLS, number_format=NumberFormat.US, created_by='test', filters=filters ) StatementAttachment.commit_changes() retrieved = StatementAttachment.get_by_id(attachment.statement_attachment_id) assert retrieved.filters == filters def test_create_without_filters(): """Test creating a statement attachment without filters.""" account_id = 24601 statement_period_id = 123 _insert_mock_attachment_data(account_id, statement_period_id) attachment = StatementAttachment.create( account_id=account_id, contract_id=10001, statement_period_id=statement_period_id, statement_attachment_type=StatementAttachmentType.LEGACY_REVENUE_DETAIL_FULL, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, file_type=StatementAttachmentFileType.XLS, number_format=NumberFormat.US, created_by='test' ) StatementAttachment.commit_changes() retrieved = StatementAttachment.get_by_id(attachment.statement_attachment_id) assert retrieved.filters is None def test_get_latest_account_self_billing_invoice_number(): """Test getting the latest self-billing invoice numbers.""" insert_mock_data({ 'account': [ {'account_id': 24601}, {'account_id': 54321}, {'account_id': 99999}, ], 'reference_payment_entity': {'reference_payment_entity_id': 1}, 'reference_sap_profit_center': { 'reference_sap_profit_center_id': 1, 'profit_center': 'UK1234', 'company_code': '1234', 'business_group': 'ORC' }, 'reference_signing_entity': { 'reference_signing_entity_id': 1, 'reference_payment_entity_id': 1, 'reference_sap_profit_center_id': 1, 'company_code': '1234', 'tax_entity_company_code': '1234', 'legal_name': 'Company', 'vat_number': '12341234', 'company_registration_number': '12341234', 'address': None, }, 'contract': [ {'contract_id': 10001, 'reference_signing_entity_id': 1}, {'contract_id': 10002, 'reference_signing_entity_id': 1}, {'contract_id': 10003, 'reference_signing_entity_id': 1}, ], 'statement_period': {'statement_period_id': 123, 'statement_period_status': 'closed'}, 'statement_attachment': [ { 'account_id': 24601, 'subaccount_id': None, 'contract_id': 10001, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'statement_period_id': 123, 'statement_period_ids': None, 'file_type': StatementAttachmentFileType.PDF, 'invoice_number': '4941_2025_SB000123', 'number_format': NumberFormat.US, 'created_at': datetime.utcnow(), 'created_by': 'test' }, { 'account_id': 24601, 'subaccount_id': None, 'contract_id': 10001, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'statement_period_id': 123, 'statement_period_ids': None, 'file_type': StatementAttachmentFileType.PDF, 'invoice_number': '4941_2025_24601_001', 'number_format': NumberFormat.US, 'created_at': datetime.utcnow(), 'created_by': 'test' }, { 'account_id': 24601, 'subaccount_id': None, 'contract_id': 10001, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'statement_period_id': 123, 'statement_period_ids': None, 'file_type': StatementAttachmentFileType.PDF, 'invoice_number': '4941_2025_24601_002', 'number_format': NumberFormat.US, 'created_at': datetime.utcnow(), 'created_by': 'test' }, { 'account_id': 24601, 'subaccount_id': None, 'contract_id': 10001, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'statement_period_id': 123, 'statement_period_ids': None, 'file_type': StatementAttachmentFileType.PDF, 'invoice_number': '4941_2026_24601_001', 'number_format': NumberFormat.US, 'created_at': datetime.utcnow(), 'created_by': 'test' }, { 'account_id': 99999, 'subaccount_id': None, 'contract_id': 10002, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'statement_period_id': 123, 'statement_period_ids': None, 'file_type': StatementAttachmentFileType.PDF, 'invoice_number': '4941_2025_99999_001', 'number_format': NumberFormat.US, 'created_at': datetime.utcnow(), 'created_by': 'test' }, { 'account_id': 54321, 'subaccount_id': None, 'contract_id': 10002, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'statement_period_id': 123, 'statement_period_ids': None, 'file_type': StatementAttachmentFileType.PDF, 'invoice_number': '4941_2025_54321_001', 'number_format': NumberFormat.US, 'created_at': datetime.utcnow(), 'created_by': 'test' }, ], }) result = StatementAttachment.get_latest_account_self_billing_invoice_number( 2025, [24601, 99999]) assert [item._asdict() for item in result] == [ {'account_id': 24601, 'invoice_number': '4941_2025_24601_002'}, {'account_id': 99999, 'invoice_number': '4941_2025_99999_001'}, ]