"""Test statement attachment logic.""" from collections import namedtuple from datetime import datetime from unittest.mock import patch from fastapi import HTTPException import pytest from moneyhub.config import Config from moneyhub.constants.constants import ContractType from moneyhub.constants.constants import KNR_SAP_IDS 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.constants.constants import VatCategory from moneyhub.constants.error import FORBIDDEN_INVOICE_URL_ACCESS from moneyhub.constants.error import FORBIDDEN_URL_ACCESS from moneyhub.constants.error import INVALID_FILE_TYPE from moneyhub.constants.error import INVALID_STATEMENT_ATTACHMENT from moneyhub.constants.error import NO_INVOICE_FILE_LOCATION from moneyhub.constants.error import SIGNING_ENTITY_NOT_SUPPORTED from moneyhub.constants.error import STATEMENT_PERIOD_NOT_VISIBLE from moneyhub.constants.features import FEATURE_INVOICE_ACCOUNT_SEQUENCE_NUMBERS from moneyhub.logic import statement_attachment as logic from tests.unit.conftest import REVENUE_DETAIL from tests.utils.factories import ContractFactory from tests.utils.factories import LedgerAccountingRunVatFactory from tests.utils.factories import LedgerVatSummaryFactory from tests.utils.factories import ReferenceSigningEntityFactory from tests.utils.factories import StatementAttachmentFactory from tests.utils.factories import StatementPeriodFactory DISTRIBUTION_FEE_INVOICE = StatementAttachmentType.DISTRIBUTION_FEE_INVOICE SELF_BILLING_INVOICE = StatementAttachmentType.SELF_BILLING_INVOICE StatementAttachmentContract = namedtuple( 'StatementAttachmentContract', ['account_id', 'contract_id', 'statement_period_id', 'signing_entity_id', 'signing_entity_name', 'company_code', 'tax_entity_company_code'] ) AccountInvoiceNumber = namedtuple( 'AccountInvoiceNumber', ['account_id', 'invoice_number'], ) @pytest.mark.parametrize('file_location, attachment_type, display_file_name', [ ( 's3://fake_fee_invoice.xls', StatementAttachmentType.DISTRIBUTION_FEE_INVOICE, None ), ( 's3://L1_1_rps_001_Aquilo_Third_Party_Jan2026.xls', StatementAttachmentType.ROYALTY_SHARE_STATEMENT, 'rps_001_Aquilo_Third_Party_Jan2026.xls' ) ]) @patch('moneyhub.logic.statement_attachment.models.StatementAttachment') def test_get_statement_attachments( mock_model, file_location, attachment_type, display_file_name ): """Test get statement attachments by an account and statement period method.""" account_id = 3 statement_period_id = 3 contract_id = 123 upload_date = datetime(2010, 9, 8, 7, 6, 5) statement_attachment = StatementAttachmentFactory.build( account_id=account_id, contract_id=contract_id, invoice_number=None, file_location=file_location, statement_attachment_type=attachment_type, statement_attachment_status='complete', file_type='xls', statement_period_id=statement_period_id, created_at=upload_date ) mock_model.get_by_account_id_and_statement_periods.return_value = \ [statement_attachment] result = logic.get_statement_attachments_by_account_and_statement_periods( account_id, [statement_period_id], contract_id) mock_model.get_by_account_id_and_statement_periods.assert_called_with( statement_period_ids=[statement_period_id], account_id=account_id, contract_id=contract_id, subaccount_id=None ) res_dict = result[0].__dict__ res_dict.pop('_sa_instance_state') assert res_dict == { 'account_id': account_id, 'subaccount_id': None, 'contract_id': contract_id, 'file_location': file_location, 'displayed_file_name': display_file_name, 'file_type': 'xls', 'invoice_number': None, 'statement_attachment_status': 'complete', 'statement_attachment_type': attachment_type, 'filters': None, 'failure_reason': None, 'number_format': NumberFormat.US, 'statement_period_id': statement_period_id, 'statement_period_ids': None, 'created_at': upload_date, 'created_by': 'me', } @patch('moneyhub.logic.statement_attachment.models.StatementAttachment') def test_get_statement_attachments_subaccount(mock_model): """Test get subaccount statement attachments by an account and statement period method.""" account_id = 3 statement_period_id = 3 contract_id = 123 subaccount_id = 1234 statement_attachment = StatementAttachmentFactory.build( account_id=account_id, invoice_number='f2398921', statement_period_id=statement_period_id, subaccount_id=subaccount_id, ) mock_model.get_by_account_id_and_statement_periods.return_value = \ [statement_attachment] result = logic.get_statement_attachments_by_account_and_statement_periods( account_id, [statement_period_id], contract_id, subaccount_id) assert len(result) == 1 mock_model.get_by_account_id_and_statement_periods.assert_called_with( statement_period_ids=[statement_period_id], account_id=account_id, contract_id=contract_id, subaccount_id=subaccount_id ) @patch('moneyhub.logic.statement_attachment.models.StatementAttachment') @patch('moneyhub.logic.statement_attachment.profile_has_access_to_resource') def test_get_statement_attachment_by_id( mock_profile_has_access_to_resource, mock_StatementAttachment ): """Test getting an attachment by ID.""" account_id = 24601 subaccount_id = None statement_attachment_id = 1234 profile_type = 'MoneyhubProfile' profile_id = 54321 statement_attachment = StatementAttachmentFactory.build( statement_attachment_id=statement_attachment_id, account_id=account_id, subaccount_id=subaccount_id, ) mock_StatementAttachment.get_by_id_or_error.return_value = statement_attachment mock_profile_has_access_to_resource.return_value = True result = logic.get_statement_attachment_by_id( statement_attachment_id, profile_type, profile_id) assert result == statement_attachment mock_StatementAttachment.get_by_id_or_error.assert_called_with( statement_attachment_id) mock_profile_has_access_to_resource.assert_called_once_with( profile_type, profile_id, account_id, subaccount_id) @patch('moneyhub.logic.statement_attachment.models.StatementAttachment') @patch('moneyhub.logic.statement_attachment.profile_has_access_to_resource') def test_get_statement_attachment_by_id_no_access( mock_profile_has_access_to_resource, mock_StatementAttachment ): """Test getting an attachment by ID when the user doesn't have access.""" account_id = 24601 subaccount_id = None statement_attachment_id = 1234 profile_type = 'MoneyhubProfile' profile_id = 54321 statement_attachment = StatementAttachmentFactory.build( statement_attachment_id=statement_attachment_id, account_id=account_id, subaccount_id=subaccount_id, ) mock_StatementAttachment.get_by_id_or_error.return_value = statement_attachment mock_profile_has_access_to_resource.return_value = False with pytest.raises(HTTPException) as error: logic.get_statement_attachment_by_id( statement_attachment_id, profile_type, profile_id) assert error.value.status_code == 403 assert error.value.detail == FORBIDDEN_URL_ACCESS @patch('moneyhub.logic.statement_attachment.models') def test_update_statement_attachment(mock_models): """Test logic method to update statement attachment.""" account_id = 10 statement_period_id = 10 statement_attachment_status = StatementAttachmentStatus.COMPLETE statement_attachment = StatementAttachmentFactory.create( account_id=account_id, statement_period_id=statement_period_id, invoice_number='10_10_01' ) mock_models.StatementAttachment.get_by_id_or_error.return_value = \ statement_attachment put_body = {'statement_attachment_status': statement_attachment_status} response = logic.update_statement_attachment(statement_period_id, **put_body) assert response.statement_attachment_status == statement_attachment_status mock_models.StatementAttachment.commit_changes.assert_called_once() @patch('moneyhub.logic.statement_attachment.models') def test_get_latest_account_invoice_number_map(mock_models): """Test to get latest invoice sequence numbers for accounts.""" year = 2025 account_ids = [24601, 99999] mock_models.StatementAttachment.get_latest_account_self_billing_invoice_number.return_value = [ AccountInvoiceNumber(24601, '6021_2025_24601_004'), AccountInvoiceNumber(99999, '4912_2025_99999_123'), ] statement_attachments = logic.get_latest_account_invoice_number_map(year, account_ids) assert statement_attachments == { 24601: 4, 99999: 123, } @patch('moneyhub.logic.statement_attachment.models') def test_get_latest_entity_invoice_number_map( mock_models, latest_statement_invoice_fixture ): """Test to get latest statemenet attachements invoices.""" mock_models.StatementAttachment.get_latest_statement_attachments_invoices \ .return_value = latest_statement_invoice_fixture statement_attachments = logic.get_latest_entity_invoice_number_map(2021) assert statement_attachments == { '4921_distribution_fee_invoice': 2, '1901_self_billing_invoice': 1 } @pytest.mark.parametrize('year, company_code, account_id, sequence_number, expected', [ (2021, 4912, 24601, 123, '4912_2021_24601_124'), (1995, 6021, 99999, 1, '6021_1995_99999_002'), ]) def test_generate_account_invoice_number( year, company_code, account_id, sequence_number, expected): """Test generating an account invoice number.""" sequence_number_map = { account_id: sequence_number, account_id + 100: sequence_number, } result = logic.generate_account_invoice_number( year, company_code, account_id, sequence_number_map) assert result == expected assert sequence_number_map == { account_id: sequence_number + 1, account_id + 100: sequence_number, } def test_generate_account_invoice_number_missing_account(): """Test generating an account invoice number when the account doesn't exist in the map.""" account_id = 24601 company_code = 4912 sequence_number_map = { 12345: 4, 99999: 12, } result = logic.generate_account_invoice_number( 2001, company_code, account_id, sequence_number_map) assert result == '4912_2001_24601_001' assert sequence_number_map == { 12345: 4, 99999: 12, 24601: 1, } @pytest.mark.parametrize('year, company_code, attachment_type, sequence_number, expected', [ (2021, '4921', DISTRIBUTION_FEE_INVOICE, 123, '4921_2021_0000000124'), (2021, '4921', SELF_BILLING_INVOICE, 123, '4921_2021_SB00000124'), (1995, 'SAPP', DISTRIBUTION_FEE_INVOICE, 12345, 'SAPP_1995_0000012346'), (1995, 'SAPP', SELF_BILLING_INVOICE, 12345, 'SAPP_1995_SB00012346'), ]) def test_generate_entity_invoice_number( year, company_code, attachment_type, sequence_number, expected): """Test to get next invoice numbers.""" sequence_number_map = { f'{company_code}_{attachment_type.value}': sequence_number, f'other_{attachment_type.value}': sequence_number, } result = logic.generate_entity_invoice_number( year, company_code, attachment_type, sequence_number_map) assert result == expected assert sequence_number_map == { f'{company_code}_{attachment_type.value}': sequence_number + 1, f'other_{attachment_type.value}': sequence_number, } @patch('moneyhub.logic.statement_attachment.db') @patch('moneyhub.logic.statement_attachment.models') def test_bulk_create_statement_attachments(mock_models, mock_db): """Test for creating multiple statement attachments.""" params = [ { 'account_id': 1, 'statement_period_id': 1, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': DISTRIBUTION_FEE_INVOICE, 'invoice_number': '4921_2021_0000000001' }, { 'account_id': 2, 'statement_period_id': 1, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': SELF_BILLING_INVOICE, 'invoice_number': '4921_2021_SB00000001' } ] mock_models.StatementAttachment.commit_changes.return_value = True res = logic.bulk_create_statement_attachments(params) assert len(res) == len(params) assert mock_models.StatementAttachment.build.call_count == len(params) mock_models.StatementAttachment.commit_changes.assert_called_once() @patch('moneyhub.logic.statement_attachment.bulk_create_statement_attachments') @patch('moneyhub.logic.statement_attachment.get_contracts_for_account') @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.sqs') def test_create_revenue_detail_reports( mock_sqs, mock_models, mock_get_contracts_for_account, mock_bulk_create_statement_attachments, contract_fixture ): """Test creating revenue detail report type Revenue Detail.""" account_id = 111 statement_period_id = 123 correlation_id = 'abc' orchard_identity_id = 'me' new_attachments = [ StatementAttachmentFactory.build(), StatementAttachmentFactory.build() ] mock_models.ReferenceSigningEntity.get_by_contract_id.return_value = ReferenceSigningEntityFactory.build() # noqa: E501 mock_models.AccountStatementPeriods.are_statement_period_ids_visible.return_value = True mock_models.StatementAttachment.get_by_statement_period.return_value = [] # noqa: E501 mock_get_contracts_for_account.return_value = [contract_fixture[0]] mock_bulk_create_statement_attachments.return_value = new_attachments result = logic.create_revenue_detail_reports( account_id, statement_period_id, orchard_identity_id, correlation_id) assert result == new_attachments mock_bulk_create_statement_attachments.assert_called_once_with([ { 'account_id': account_id, 'subaccount_id': None, 'contract_id': contract_fixture[0]['contract_id'], 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.REVENUE_DETAIL, 'file_type': StatementAttachmentFileType.CSV, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id } ]) mock_sqs.send_message.assert_called_once_with( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, { 'account_id': account_id, 'subaccount_id': None, 'statement_period_id': statement_period_id, } ) @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.sqs') def test_create_revenue_detail_reports_with_contract( mock_sqs, mock_models, contract_fixture ): """Test creating revenue detail report type Revenue Detail.""" account_id = 111 statement_period_id = 123 contract_id = 1 correlation_id = 'abc' orchard_identity_id = 'me' expected = [ StatementAttachmentFactory.build( contract_id=contract_id, statement_attachment_type=StatementAttachmentType.REVENUE_DETAIL, file_type=StatementAttachmentFileType.CSV, number_format=NumberFormat.US ), ] mock_models.Contract.get_by_id.return_value = ContractFactory.build(contract_type=ContractType.DISTRIBUTION) # noqa: E501 mock_models.AccountStatementPeriods.are_statement_period_ids_visible.return_value = True mock_models.ReferenceSigningEntity.get_by_contract_id.return_value = ReferenceSigningEntityFactory.build() # noqa: E501 mock_models.StatementAttachment.exists.return_value = False mock_models.StatementAttachment.create.side_effect = expected result = logic.create_revenue_detail_reports( account_id, statement_period_id, orchard_identity_id, correlation_id, contract_id) assert result == expected mock_models.StatementAttachment.create.assert_called_once_with( account_id=account_id, subaccount_id=None, contract_id=contract_id, statement_period_id=statement_period_id, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, statement_attachment_type=REVENUE_DETAIL, file_type=StatementAttachmentFileType.CSV, number_format=NumberFormat.US, created_by=orchard_identity_id, ) mock_models.StatementAttachment.exists.assert_called() mock_sqs.send_message.assert_called_once_with( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, { 'account_id': account_id, 'subaccount_id': None, 'statement_period_id': statement_period_id, } ) @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.sqs') def test_create_revenue_detail_reports_with_subaccount( mock_sqs, mock_models ): """Test creating subaccount revenue detail report type Revenue Detail.""" account_id = 111 statement_period_id = 123 subaccount_id = 1234 correlation_id = 'abc' orchard_identity_id = 'me' expected = [ StatementAttachmentFactory.build( contract_id=None, subaccount_id=subaccount_id, statement_attachment_type=StatementAttachmentType.REVENUE_DETAIL, file_type=StatementAttachmentFileType.CSV, number_format=NumberFormat.US ), ] mock_models.Contract.get_by_id.return_value = ContractFactory.build(contract_type=ContractType.DISTRIBUTION) # noqa: E501 mock_models.AccountStatementPeriods.are_statement_period_ids_visible.return_value = True mock_models.ReferenceSigningEntity.get_by_contract_id.return_value = ReferenceSigningEntityFactory.build() # noqa: E501 mock_models.StatementAttachment.exists.return_value = False mock_models.StatementAttachment.create.side_effect = expected result = logic.create_revenue_detail_reports( account_id, statement_period_id, orchard_identity_id, correlation_id, None, subaccount_id) assert result == expected mock_models.StatementAttachment.create.assert_called_once_with( account_id=account_id, subaccount_id=subaccount_id, contract_id=None, statement_period_id=statement_period_id, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, statement_attachment_type=REVENUE_DETAIL, file_type=StatementAttachmentFileType.CSV, number_format=NumberFormat.US, created_by=orchard_identity_id, ) mock_models.StatementAttachment.exists.assert_called() mock_sqs.send_message.assert_called_once_with( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, { 'account_id': account_id, 'subaccount_id': subaccount_id, 'statement_period_id': statement_period_id, } ) @patch('moneyhub.logic.statement_attachment.bulk_create_statement_attachments') @patch('moneyhub.logic.statement_attachment.get_contracts_for_account') @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.sqs') def test_create_revenue_detail_reports_neighbouring_rights_perfomer( mock_sqs, mock_models, mock_get_contracts_for_account, mock_bulk_create_statement_attachments, contract_fixture): """Test creating neighbouring rights revenue report type NR performer.""" account_id = 111 statement_period_id = 123 correlation_id = 'abc' orchard_identity_id = 'me' new_attachments = [ StatementAttachmentFactory.build(), StatementAttachmentFactory.build() ] contract = contract_fixture[2] mock_models.StatementAttachment.get_by_statement_period.return_value = [] # noqa: E501 mock_models.AccountStatementPeriods.are_statement_period_ids_visible.return_value = True mock_get_contracts_for_account.return_value = [contract] mock_bulk_create_statement_attachments.return_value = new_attachments result = logic.create_revenue_detail_reports( account_id, statement_period_id, orchard_identity_id, correlation_id) assert result == new_attachments mock_bulk_create_statement_attachments.assert_called_once_with([ { 'account_id': account_id, 'subaccount_id': None, 'contract_id': contract['contract_id'], 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.NEIGHBOURING_RIGHTS_PERFORMER_REVENUE, 'file_type': StatementAttachmentFileType.CSV, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id } ]) mock_sqs.send_message.assert_called_once_with( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, { 'account_id': account_id, 'subaccount_id': None, 'statement_period_id': statement_period_id, } ) @patch('moneyhub.logic.statement_attachment.bulk_create_statement_attachments') @patch('moneyhub.logic.statement_attachment.get_contracts_for_account') @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.sqs') def test_create_revenue_detail_reports_neighbouring_rights_label( mock_sqs, mock_models, mock_get_contracts_for_account, mock_bulk_create_statement_attachments, contract_fixture): """Test creating revenue detail report type NR label.""" account_id = 111 statement_period_id = 123 correlation_id = 'abc' orchard_identity_id = 'me' new_attachments = [ StatementAttachmentFactory.build(), StatementAttachmentFactory.build() ] mock_models.ReferenceSigningEntity.get_by_contract_id.return_value = ReferenceSigningEntityFactory.build(company_code='4919') # noqa: E501 mock_models.AccountStatementPeriods.are_statement_period_ids_visible.return_value = True mock_models.StatementAttachment.get_by_statement_period.return_value = [] # noqa: E501 mock_get_contracts_for_account.return_value = [contract_fixture[1]] mock_bulk_create_statement_attachments.return_value = new_attachments result = logic.create_revenue_detail_reports( account_id, statement_period_id, orchard_identity_id, correlation_id) assert result == new_attachments mock_bulk_create_statement_attachments.assert_called_once_with([ { 'account_id': account_id, 'subaccount_id': None, 'contract_id': contract_fixture[1]['contract_id'], 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.NEIGHBOURING_RIGHTS_LABEL_REVENUE, # noqa: E501 'file_type': StatementAttachmentFileType.CSV, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id } ]) mock_sqs.send_message.assert_called_once_with( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, { 'account_id': account_id, 'subaccount_id': None, 'statement_period_id': statement_period_id, } ) @patch('moneyhub.logic.statement_attachment.models') def test_create_revenue_detail_reports_statement_period_not_visible(mock_models): """Test creating revenue detail report when period is not visible to the customer.""" mock_models.AccountStatementPeriods.are_statement_period_ids_visible.return_value = False with pytest.raises(HTTPException) as error: logic.create_revenue_detail_reports(123, 456, 'abc') assert error.value.status_code == 400 assert error.value.detail == STATEMENT_PERIOD_NOT_VISIBLE.format(statement_period_ids=[456]) mock_models.AccountStatementPeriods.are_statement_period_ids_visible \ .assert_called_once_with(123, [456]) @patch('moneyhub.logic.statement_attachment.bulk_create_statement_attachments') @patch('moneyhub.logic.statement_attachment.models') def test_create_revenue_detail_reports_attachment_exists( mock_models, mock_bulk_create_statement_attachments): """Test creating revenue detail report when the attachment exists.""" attachment = StatementAttachmentFactory.build( statement_attachment_type=StatementAttachmentType.REVENUE_DETAIL) mock_models.AccountStatementPeriods.are_statement_period_ids_visible.return_value = True mock_models.StatementAttachment.get_by_statement_period.return_value = [ attachment] result = logic.create_revenue_detail_reports(123, 456, 'abc') assert result == [attachment] mock_bulk_create_statement_attachments.assert_not_called() @patch('moneyhub.logic.statement_attachment.bulk_create_statement_attachments') @patch('moneyhub.logic.statement_attachment.datetime') @patch('moneyhub.logic.statement_attachment.db') @patch('moneyhub.logic.statement_attachment.is_feature_enabled') @patch('moneyhub.logic.statement_attachment.models') def test_create_statement_attachments( mock_models, mock_is_feature_enabled, mock_db, mock_datetime, mock_bulk_create_statement_attachments, latest_statement_invoice_fixture): """Test for creating statement attachments.""" statement_period_id = 123 orchard_identity_id = 'me' contracts = [ StatementAttachmentContract( 1, 1001, statement_period_id, 2, 'AWAL', '4921', '4921'), StatementAttachmentContract( 1, 1002, statement_period_id, 2, 'AWAL', '4921', '4921'), StatementAttachmentContract( 1, 1003, statement_period_id, 2, 'AWAL', '4921', '4921'), StatementAttachmentContract( 1, 1004, statement_period_id, 2, 'AWAL', '4921', '4921'), ] vat_entries = [ LedgerAccountingRunVatFactory.build( contract_id=1001, distribution_fee=12.34, gross_vat_rate=10.0), LedgerAccountingRunVatFactory.build( contract_id=1002, distribution_fee=12.34, gross_vat_rate=10.0), ] ledger_vat_entries = [ LedgerVatSummaryFactory.build( contract_id=1001, vat_category=VatCategory.GROSS_REVENUE), LedgerVatSummaryFactory.build( contract_id=1001, vat_category=VatCategory.DISTRIBUTION_FEE), LedgerVatSummaryFactory.build( contract_id=1003, vat_category=VatCategory.DISTRIBUTION_FEE), LedgerVatSummaryFactory.build( contract_id=1003, vat_category=VatCategory.GROSS_REVENUE), LedgerVatSummaryFactory.build( contract_id=1004, vat_category=VatCategory.CUSTOM_PAYMENT), ] mock_is_feature_enabled.return_value = False mock_models.StatementAttachment.get_latest_statement_attachments_invoices \ .return_value = latest_statement_invoice_fixture mock_models.StatementAttachment.get_latest_account_self_billing_invoice_number.return_value = [] mock_datetime.now.return_value = datetime(1999, 12, 1) mock_models.LedgerAccountContract.get_contracts_by_statement_period.return_value = contracts mock_models.LedgerAccountingRunVat.get_by_statement_period.return_value = vat_entries mock_models.LedgerVatSummary.get_by_activity_statement_period.return_value = ledger_vat_entries mock_models.StatementAttachment.get_by_statement_period.return_value = [] mock_models.StatementPeriod.get_by_id.return_value = StatementPeriodFactory.build() mock_bulk_create_statement_attachments.return_value = ['yes'] res = logic.create_statement_attachments(statement_period_id, orchard_identity_id) assert res == ['yes'] mock_bulk_create_statement_attachments.assert_called_once_with([ { 'account_id': contracts[0].account_id, 'contract_id': contracts[0].contract_id, 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.DISTRIBUTION_FEE_INVOICE, 'invoice_number': '4921_1999_0000000003', 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id }, { 'account_id': contracts[0].account_id, 'contract_id': contracts[0].contract_id, 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'invoice_number': '4921_1999_SB00000001', 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id }, { 'account_id': contracts[1].account_id, 'contract_id': contracts[1].contract_id, 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.DISTRIBUTION_FEE_INVOICE, 'invoice_number': '4921_1999_0000000004', 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id }, { 'account_id': contracts[2].account_id, 'contract_id': contracts[2].contract_id, 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.DISTRIBUTION_FEE_INVOICE, 'invoice_number': '4921_1999_0000000005', 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id }, { 'account_id': contracts[2].account_id, 'contract_id': contracts[2].contract_id, 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'invoice_number': '4921_1999_SB00000002', 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id }, { 'account_id': contracts[3].account_id, 'contract_id': contracts[3].contract_id, 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'invoice_number': '4921_1999_SB00000003', 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id }, ]) @patch('moneyhub.logic.statement_attachment.bulk_create_statement_attachments') @patch('moneyhub.logic.statement_attachment.datetime') @patch('moneyhub.logic.statement_attachment.db') @patch('moneyhub.logic.statement_attachment.is_feature_enabled') @patch('moneyhub.logic.statement_attachment.models') def test_create_statement_attachments_invoice_ff_on( mock_models, mock_is_feature_enabled, mock_db, mock_datetime, mock_bulk_create_statement_attachments, latest_statement_invoice_fixture): """Test for creating statement attachments with the invoice numbers feature flag enabled.""" statement_period_id = 123 orchard_identity_id = 'me' mock_is_feature_enabled.side_effect = lambda ff: ff == FEATURE_INVOICE_ACCOUNT_SEQUENCE_NUMBERS mock_models.StatementAttachment.get_latest_statement_attachments_invoices \ .return_value = latest_statement_invoice_fixture mock_models.StatementAttachment.get_latest_account_self_billing_invoice_number.return_value = [ AccountInvoiceNumber(24601, '4921_1999_24601_123'), ] mock_datetime.now.return_value = datetime(1999, 12, 1) mock_models.LedgerAccountContract.get_contracts_by_statement_period.return_value = [ StatementAttachmentContract( 24601, 1001, statement_period_id, 2, 'AWAL', '4921', '4921'), StatementAttachmentContract( 99999, 1002, statement_period_id, 2, 'AWAL', '6021', '6021'), ] mock_models.LedgerAccountingRunVat.get_by_statement_period.return_value = [] mock_models.LedgerVatSummary.get_by_activity_statement_period.return_value = [ LedgerVatSummaryFactory.build( contract_id=1001, vat_category=VatCategory.GROSS_REVENUE), LedgerVatSummaryFactory.build( contract_id=1002, vat_category=VatCategory.CUSTOM_PAYMENT), ] mock_models.StatementAttachment.get_by_statement_period.return_value = [] mock_models.StatementPeriod.get_by_id.return_value = StatementPeriodFactory.build() mock_bulk_create_statement_attachments.return_value = [] logic.create_statement_attachments(statement_period_id, orchard_identity_id) mock_bulk_create_statement_attachments.assert_called_once_with([ { 'account_id': 24601, 'contract_id': 1001, 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'invoice_number': '4921_1999_24601_124', 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id }, { 'account_id': 99999, 'contract_id': 1002, 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'invoice_number': '6021_1999_99999_001', 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id }, ]) @pytest.mark.parametrize('contract_id', [ 123, None ]) @patch('moneyhub.logic.statement_attachment.models') def test_create_internal_attachment(mock_models, contract_id): """Test for creating internal statement attachments with an existing account.""" statement_period_id = 1 account_id = 1 orchard_identity_id = 'me' file_location = 'https://fakefile.pdf' file_type = StatementAttachmentFileType.PDF number_format = NumberFormat.US, upload_date = datetime(2010, 9, 8, 7, 6, 5) model = StatementAttachmentFactory.build( account_id=account_id, contract_id=contract_id, statement_attachment_id=1, statement_attachment_status=StatementAttachmentStatus.COMPLETE, statement_attachment_type=StatementAttachmentType.INTERNAL_UPLOAD, statement_period_id=statement_period_id, file_location=file_location, file_type=file_type, invoice_number=None, number_format=number_format, created_at=upload_date, created_by='me', ) mock_models.Account.exists.return_value = True mock_models.StatementAttachment.exists.return_value = False mock_models.StatementAttachment.create.return_value = model res = logic.create_internal_attachment( account_id, statement_period_id, file_type, file_location, orchard_identity_id, upload_date, contract_id) assert res == model mock_models.Account.exists.assert_called() mock_models.StatementAttachment.exists.assert_called() mock_models.StatementAttachment.create.assert_called_once_with( account_id=account_id, contract_id=contract_id, statement_period_id=statement_period_id, file_type=file_type, file_location=file_location, created_by=orchard_identity_id, created_at=upload_date, statement_attachment_status=StatementAttachmentStatus.COMPLETE, statement_attachment_type=StatementAttachmentType.INTERNAL_UPLOAD ) @pytest.mark.parametrize('file_location, display_file_name, attachment_type', [ ( 'https://L0_1_fakefile.xls', None, StatementAttachmentType.INTERNAL_UPLOAD ), ( 's3://bucket/L1_1_Publishing_Statement_Jan2026.xls', None, StatementAttachmentType.PUBLISHING_DETAIL ), ( 's3://L1_1_rps_001_Aquilo_Third_Party_Jan2026.xls', 'rps_001_Aquilo_Third_Party_Jan2026.xls', StatementAttachmentType.ROYALTY_SHARE_STATEMENT ), ]) @patch('moneyhub.logic.statement_attachment.models') def test_create_internal_attachment_with_pattern_matching( mock_models, file_location, display_file_name, attachment_type ): """Test that a created internal statement file is saved with the expected attachment type.""" statement_period_id = 1 account_id = 1 contract_id = 123 orchard_identity_id = 'me' file_type = StatementAttachmentFileType.XLS upload_date = datetime(2010, 9, 8, 7, 6, 5) model = StatementAttachmentFactory.build( account_id=account_id, contract_id=contract_id, statement_attachment_id=1, statement_attachment_status=StatementAttachmentStatus.COMPLETE.value, statement_attachment_type=attachment_type, statement_period_id=statement_period_id, file_location=file_location, file_type=file_type, invoice_number=None, created_at=upload_date, created_by='me', ) mock_models.Account.exists.return_value = True mock_models.StatementAttachment.exists.return_value = False mock_models.StatementAttachment.create.return_value = model res = logic.create_internal_attachment( account_id, statement_period_id, file_type, file_location, orchard_identity_id, upload_date, contract_id) res_dict = res.__dict__ res_dict.pop('_sa_instance_state') assert res_dict == { 'account_id': account_id, 'subaccount_id': None, 'contract_id': contract_id, 'file_location': file_location, 'displayed_file_name': display_file_name, 'file_type': file_type, 'invoice_number': None, 'statement_attachment_id': 1, 'statement_attachment_status': 'complete', 'statement_attachment_type': attachment_type, 'filters': None, 'failure_reason': None, 'number_format': NumberFormat.US, 'statement_period_id': statement_period_id, 'statement_period_ids': None, 'created_at': upload_date, 'created_by': 'me', } mock_models.StatementAttachment.create.assert_called_once_with( account_id=account_id, contract_id=contract_id, statement_period_id=statement_period_id, file_type=file_type, file_location=file_location, created_by=orchard_identity_id, created_at=upload_date, statement_attachment_status=StatementAttachmentStatus.COMPLETE, statement_attachment_type=attachment_type ) @patch('moneyhub.logic.statement_attachment.models') def test_create_internal_attachment_already_existing(mock_models): """Test for creating internal statement attachments that already exists.""" statement_period_id = 1 account_id = 1 orchard_identity_id = 'me' file_location = 'https://fakefile.pdf' file_type = StatementAttachmentFileType.PDF upload_date = datetime(2010, 9, 8, 7, 6, 5) mock_models.Account.exists.return_value = True mock_models.StatementAttachment.exists.return_value = True res = logic.create_internal_attachment( account_id, statement_period_id, file_type, file_location, orchard_identity_id, upload_date) assert not res mock_models.Account.exists.assert_called() mock_models.StatementAttachment.exists.assert_called() mock_models.StatementAttachment.create.assert_not_called() @patch('moneyhub.logic.statement_attachment.models') def test_create_internal_attachment_account_does_not_exist(mock_models): """Test for creating internal statement attachments where an account does not exist.""" statement_period_id = 1 account_id = 1 orchard_identity_id = 'me' file_location = 'https://fakefile.pdf' file_type = StatementAttachmentFileType.PDF upload_date = datetime(2010, 9, 8, 7, 6, 5) mock_models.Account.exists.return_value = False res = logic.create_internal_attachment( account_id, statement_period_id, file_type, file_location, orchard_identity_id, upload_date) # noqa: E501 assert not res mock_models.Account.exists.assert_called() mock_models.StatementAttachment.exists.assert_not_called() mock_models.StatementAttachment.create.assert_not_called() @patch('moneyhub.logic.statement_attachment.bulk_create_statement_attachments') @patch('moneyhub.logic.statement_attachment.datetime') @patch('moneyhub.logic.statement_attachment.db') @patch('moneyhub.logic.statement_attachment.models') def test_create_statement_attachments_existing_attachments( mock_models, mock_db, mock_datetime, mock_bulk_create_statement_attachments, latest_statement_invoice_fixture): """Test for creating statement attachments when existing attachments exist.""" statement_period_id = 123 orchard_identity_id = 'me' contracts = [ StatementAttachmentContract( 1, 1001, statement_period_id, 2, 'AWAL', '4921', '4921'), StatementAttachmentContract( 1, 1002, statement_period_id, 2, 'AWAL', '4921', '4921'), StatementAttachmentContract( 1, 1003, statement_period_id, 2, 'AWAL', '4921', '4921'), ] vat_entries = [ LedgerAccountingRunVatFactory.build( contract_id=1001, distribution_fee=12.34, gross_vat_rate=10.0), LedgerAccountingRunVatFactory.build( contract_id=1002, distribution_fee=12.34, gross_vat_rate=10.0), ] ledger_vat_entries = [ LedgerVatSummaryFactory.build( contract_id=1001, vat_category='gross_revenue'), LedgerVatSummaryFactory.build( contract_id=1001, vat_category='distribution_fee'), LedgerVatSummaryFactory.build( contract_id=1003, vat_category='distribution_fee'), LedgerVatSummaryFactory.build( contract_id=1003, vat_category='gross_revenue'), ] existing_attachments = [ StatementAttachmentFactory.build( account_id=contracts[0].account_id, contract_id=contracts[0].contract_id, statement_period_id=statement_period_id, statement_attachment_type=StatementAttachmentType.DISTRIBUTION_FEE_INVOICE, # noqa: E501 file_type=StatementAttachmentFileType.PDF, number_format=NumberFormat.US ), StatementAttachmentFactory.build( account_id=contracts[0].account_id, contract_id=contracts[0].contract_id, statement_period_id=statement_period_id, statement_attachment_type=StatementAttachmentType.SELF_BILLING_INVOICE, file_type=StatementAttachmentFileType.PDF, number_format=NumberFormat.US ), ] mock_models.StatementAttachment.get_latest_statement_attachments_invoices \ .return_value = latest_statement_invoice_fixture mock_datetime.now.return_value = datetime(1999, 12, 1) mock_models.LedgerAccountContract.get_contracts_by_statement_period.return_value = contracts mock_models.LedgerAccountingRunVat.get_by_statement_period.return_value = vat_entries mock_models.LedgerVatSummary.get_by_activity_statement_period.return_value = ledger_vat_entries mock_models.StatementAttachment.get_by_statement_period.return_value = \ existing_attachments mock_models.StatementPeriod.get_by_id.return_value = StatementPeriodFactory.build() mock_bulk_create_statement_attachments.return_value = ['yes'] res = logic.create_statement_attachments(statement_period_id, orchard_identity_id) assert res == ['yes'] mock_bulk_create_statement_attachments.assert_called_once_with([ { 'account_id': contracts[1].account_id, 'contract_id': contracts[1].contract_id, 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.DISTRIBUTION_FEE_INVOICE, 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'invoice_number': '4921_1999_0000000003', 'created_by': orchard_identity_id }, { 'account_id': contracts[2].account_id, 'contract_id': contracts[2].contract_id, 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.DISTRIBUTION_FEE_INVOICE, 'invoice_number': '4921_1999_0000000004', 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id }, { 'account_id': contracts[2].account_id, 'contract_id': contracts[2].contract_id, 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'invoice_number': '4921_1999_SB00000001', 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id }, ]) @patch('moneyhub.logic.statement_attachment.bulk_create_statement_attachments') @patch('moneyhub.logic.statement_attachment.datetime') @patch('moneyhub.logic.statement_attachment.db') @patch('moneyhub.logic.statement_attachment.models') def test_create_statement_attachments_mixed_vat_entries( mock_models, mock_db, mock_datetime, mock_bulk_create_statement_attachments, latest_statement_invoice_fixture): """Test for creating statement attachments with a mixture of vat entries.""" statement_period_id = 123 orchard_identity_id = 'me' contracts = [ StatementAttachmentContract( 1, 1001, statement_period_id, 2, 'AWAL', '4921', '4921'), StatementAttachmentContract( 1, 1002, statement_period_id, 2, 'AWAL', '4921', '4921'), StatementAttachmentContract( 1, 1003, statement_period_id, 2, 'AWAL', '4921', '4921'), StatementAttachmentContract( 1, 1004, statement_period_id, 2, 'AWAL', '4921', '4921'), ] vat_entries = [ LedgerAccountingRunVatFactory.build( contract_id=1001, distribution_fee=12.34), LedgerAccountingRunVatFactory.build( contract_id=1002, gross_vat_rate=10.0), LedgerAccountingRunVatFactory.build( contract_id=1003, exempt_reason='Something'), ] ledger_vat_entries = [ LedgerVatSummaryFactory.build( contract_id=1001, vat_category='gross_revenue'), LedgerVatSummaryFactory.build( contract_id=1001, vat_category='distribution_fee'), ] mock_models.StatementAttachment.get_latest_statement_attachments_invoices \ .return_value = latest_statement_invoice_fixture mock_datetime.now.return_value = datetime(1999, 12, 1) mock_models.LedgerAccountContract.get_contracts_by_statement_period.return_value = contracts mock_models.LedgerAccountingRunVat.get_by_statement_period.return_value = vat_entries mock_models.LedgerVatSummary.get_by_activity_statement_period.return_value = ledger_vat_entries mock_models.StatementAttachment.get_by_statement_period.return_value = [] mock_models.StatementPeriod.get_by_id.return_value = StatementPeriodFactory.build() mock_bulk_create_statement_attachments.return_value = ['yes'] res = logic.create_statement_attachments(statement_period_id, orchard_identity_id) assert res == ['yes'] mock_bulk_create_statement_attachments.assert_called_once_with([ { 'account_id': contracts[0].account_id, 'contract_id': contracts[0].contract_id, 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.DISTRIBUTION_FEE_INVOICE, 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'invoice_number': '4921_1999_0000000003', 'created_by': orchard_identity_id }, { 'account_id': contracts[0].account_id, 'contract_id': contracts[0].contract_id, 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.SELF_BILLING_INVOICE, 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'invoice_number': '4921_1999_SB00000001', 'created_by': orchard_identity_id } ]) @patch('moneyhub.logic.statement_attachment.bulk_create_statement_attachments') @patch('moneyhub.logic.statement_attachment.get_contracts_for_account') @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.sqs') def test_create_collection_summary_documents( mock_sqs, mock_models, mock_get_contracts_for_account, bulk_create_statement_attachments, contract_fixture): """Test collection summary documents.""" account_id = 24601 statement_period_id = 123 correlation_id = 'abcd-1234' orchard_identity_id = 'abcd-1234' expected = [ StatementAttachmentFactory.build(), StatementAttachmentFactory.build(), ] bulk_create_statement_attachments.return_value = expected mock_get_contracts_for_account.return_value = [contract_fixture[1], contract_fixture[2]] mock_models.ReferenceSigningEntity.get_by_contract_id.side_effect = [ ReferenceSigningEntityFactory.build(company_code=KNR_SAP_IDS[0]), ReferenceSigningEntityFactory.build(company_code=KNR_SAP_IDS[0]) ] mock_models.StatementAttachment.get_by_statement_period.return_value = [] mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.return_value = [ statement_period_id] result = logic.create_collection_summary_documents( account_id, statement_period_id, correlation_id, orchard_identity_id) assert result == expected bulk_create_statement_attachments.assert_called_once_with([ { 'account_id': account_id, 'contract_id': contract_fixture[1]['contract_id'], 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.COLLECTION_SUMMARY_LABEL, 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id }, { 'account_id': account_id, 'contract_id': contract_fixture[2]['contract_id'], 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': StatementAttachmentType.COLLECTION_SUMMARY_PERFORMER, 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id } ]) mock_sqs.send_message.assert_called_once_with( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, {'account_id': account_id, 'statement_period_id': statement_period_id} ) @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.sqs') def test_create_legacy_revenue_report_single_contract( mock_sqs, mock_models, contract_fixture ): """Test create legacy revenue reports with a selected contract.""" account_id = 24601 contract_id = 2001 statement_period_id = 123 statement_period_ids = '123,456' orchard_identity_id = 'abcd-1234' report_type = StatementAttachmentType.LEGACY_REVENUE_DETAIL_PHYSICAL file_type = StatementAttachmentFileType.XLS filters = { 'transaction_type_ids': [1, 2, 3], 'transaction_type_group_ids': [4, 5, 6] } number_format = NumberFormat.EU correlation_id = 'abcd-1234' expected = [ StatementAttachmentFactory.build( contract_id=contract_fixture[1]['contract_id'], statement_attachment_type=report_type, file_type=file_type, number_format=number_format, filters=filters ), ] mock_models.StatementPeriodPaymentEntity.are_statement_period_ids_visible.return_value = True mock_models.StatementAttachment.exists.return_value = False mock_models.StatementAttachment.create.side_effect = expected result = logic.create_legacy_revenue_report( account_id, contract_id, statement_period_id, orchard_identity_id, report_type, file_type, number_format, correlation_id, filters=filters, statement_period_ids=statement_period_ids ) assert result == expected mock_models.AccountContract.get_by_account.assert_not_called() mock_sqs.send_message.assert_called_once_with( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, { 'statement_attachment_id': expected[0].statement_attachment_id, } ) @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.sqs') @pytest.mark.parametrize( 'statement_period_id, statement_period_ids, expected_statement_period_ids', [ pytest.param( 123, [123, 456, 789], '123, 456, 789', id='Uses the original statement_period_ids list' ), pytest.param( 123, None, '123', id='Uses the original statement_period_id'), ] ) def test_create_legacy_revenue_report_statement_period_ids( mock_sqs, mock_models, contract_fixture, statement_period_id, statement_period_ids, expected_statement_period_ids ): """Test create legacy revenue reports with a selected contract.""" account_id = 24601 contract_id = 2001 orchard_identity_id = 'abcd-1234' report_type = StatementAttachmentType.LEGACY_REVENUE_DETAIL_PHYSICAL file_type = StatementAttachmentFileType.XLS number_format = NumberFormat.EU correlation_id = 'abcd-1234' expected = [ StatementAttachmentFactory.build( contract_id=contract_fixture[1]['contract_id'], statement_period_ids=statement_period_ids, statement_attachment_type=report_type, file_type=file_type, number_format=number_format ), ] mock_models.StatementPeriodPaymentEntity.are_statement_period_ids_visible.return_value = True mock_models.StatementAttachment.create.side_effect = expected result = logic.create_legacy_revenue_report( account_id, contract_id, statement_period_id, orchard_identity_id, report_type, file_type, number_format, correlation_id, filters=None, statement_period_ids=statement_period_ids ) assert result == expected mock_models.StatementAttachment.create.assert_called_once_with( account_id=account_id, subaccount_id=None, contract_id=contract_id, statement_period_id=statement_period_id, statement_period_ids=expected_statement_period_ids, file_type=file_type, number_format=number_format, created_by=orchard_identity_id, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, statement_attachment_type=report_type, filters=None ) @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.sqs') def test_create_legacy_revenue_report_subaccount( mock_sqs, mock_models, contract_fixture ): """Test create legacy revenue reports with a subaccount.""" account_id = 24601 contract_id = 2001 subaccount_id = 1234 statement_period_id = 123 orchard_identity_id = 'abcd-1234' report_type = StatementAttachmentType.LEGACY_REVENUE_DETAIL_PHYSICAL file_type = StatementAttachmentFileType.XLS number_format = NumberFormat.EU correlation_id = 'abcd-1234' expected = [ StatementAttachmentFactory.build( contract_id=contract_fixture[1]['contract_id'], subaccount_id=subaccount_id, statement_attachment_type=report_type, file_type=file_type, number_format=number_format ), ] mock_models.StatementPeriodPaymentEntity.are_statement_period_ids_visible.return_value = True mock_models.StatementAttachment.create.side_effect = expected result = logic.create_legacy_revenue_report( account_id, contract_id, statement_period_id, orchard_identity_id, report_type, file_type, number_format, correlation_id, subaccount_id=subaccount_id, filters=None ) assert result == expected mock_models.AccountContract.get_by_account.assert_not_called() mock_sqs.send_message.assert_called_once_with( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, { 'statement_attachment_id': expected[0].statement_attachment_id, } ) @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.sqs') @pytest.mark.parametrize('is_visible, report_type, file_type, expected', [ (False, StatementAttachmentType.LEGACY_REVENUE_DETAIL_PHYSICAL, StatementAttachmentFileType.XLS, STATEMENT_PERIOD_NOT_VISIBLE.format(statement_period_ids=[123])), # noqa: E501 (True, StatementAttachmentType.REVENUE_DETAIL, StatementAttachmentFileType.TXT, INVALID_STATEMENT_ATTACHMENT), # noqa: E501 (True, StatementAttachmentType.LEGACY_REVENUE_DETAIL_FULL, StatementAttachmentFileType.CSV, INVALID_FILE_TYPE), # noqa: E501 ]) def test_create_legacy_revenue_with_exceptions( mock_sqs, mock_models, is_visible, report_type, file_type, expected ): """Test create legacy reports with exceptions.""" account_id = 24601 contract_id = 2001 statement_period_id = 123 orchard_identity_id = 'abcd-1234' number_format = NumberFormat.EU correlation_id = 'abcd-1234' mock_models.AccountStatementPeriods.are_statement_period_ids_visible.return_value = is_visible with pytest.raises(HTTPException) as err: logic.create_legacy_revenue_report( account_id, contract_id, statement_period_id, orchard_identity_id, report_type, file_type, number_format, correlation_id, filters=None ) assert err.value.detail == expected mock_models.AccountContract.get_by_account.assert_not_called() mock_sqs.send_message.assert_not_called() @patch('moneyhub.logic.statement_attachment.get_contracts_for_account') @patch('moneyhub.logic.statement_attachment.models') def test_create_collection_summary_period_not_available( mock_models, mock_get_contracts_for_account, contract_fixture): """Test creating collection summary when the period is not available.""" account_id = 24601 statement_period_id = 999 correlation_id = None orchard_identity_id = 'me' mock_get_contracts_for_account.return_value = contract_fixture mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.return_value = [ 1, 2, 3] with pytest.raises(HTTPException) as error: logic.create_collection_summary_documents( account_id, statement_period_id, orchard_identity_id, correlation_id) assert error.value.status_code == 400 assert error.value.detail == STATEMENT_PERIOD_NOT_VISIBLE.format( statement_period_ids=[statement_period_id]) mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.assert_called_with( account_id) @patch('moneyhub.logic.statement_attachment.get_contracts_for_account') @patch('moneyhub.logic.statement_attachment.models') def test_create_collection_summary_unsupported_signing_entity( mock_models, mock_get_contracts_for_account, contract_fixture): """Test creating collection summary when the period is not available.""" account_id = 24601 statement_period_id = 123 correlation_id = None orchard_identity_id = 'me' mock_get_contracts_for_account.return_value = contract_fixture mock_models.ReferenceSigningEntity.get_by_contract_id.side_effect = [ ReferenceSigningEntityFactory.build(company_code='notreal')] mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.return_value = [ statement_period_id] with pytest.raises(HTTPException) as error: logic.create_collection_summary_documents( account_id, statement_period_id, orchard_identity_id, correlation_id) assert error.value.status_code == 400 assert error.value.detail == SIGNING_ENTITY_NOT_SUPPORTED mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.assert_called_with( account_id) @patch('moneyhub.logic.statement_attachment.bulk_create_statement_attachments') @patch('moneyhub.logic.statement_attachment.get_contracts_for_account') @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.sqs') def test_create_collection_summary_existing_attachments( mock_sqs, mock_models, mock_get_contracts_for_account, bulk_create_statement_attachments, contract_fixture ): """Test creating collection summary when attachments exist.""" account_id = 24601 statement_period_id = 123 correlation_id = 'abcd-1234' orchard_identity_id = 'me' existing_attachments = [ StatementAttachmentFactory.build( account_id=account_id, contract_id=contract['contract_id'], statement_attachment_status=StatementAttachmentStatus.COMPLETE, statement_period_id=statement_period_id, ) for contract in contract_fixture ] mock_get_contracts_for_account.return_value = contract_fixture mock_models.ReferenceSigningEntity.get_by_contract_id.return_value = \ ReferenceSigningEntityFactory.build(company_code=KNR_SAP_IDS[0]) mock_models.StatementAttachment.get_by_statement_period.return_value = existing_attachments mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.return_value = [ statement_period_id] result = logic.create_collection_summary_documents( account_id, statement_period_id, orchard_identity_id, correlation_id) assert result == existing_attachments bulk_create_statement_attachments.assert_not_called() mock_sqs.send_message.assert_not_called() mock_models.StatementAttachment.get_by_statement_period.assert_called_once_with( statement_period_id, account_id=account_id, types=[ StatementAttachmentType.COLLECTION_SUMMARY_PERFORMER, StatementAttachmentType.COLLECTION_SUMMARY_LABEL ] ) @patch('moneyhub.logic.statement_attachment.models.StatementAttachment') @patch('moneyhub.logic.statement_attachment.sqs') def test_trigger_statement_attachment_generation(mock_sqs, mock_model): """Test triggering statement attachment generation for statement period.""" account_id = 3 statement_period_id = 3 correlation_id = '123' status = StatementAttachmentStatus.IN_PROGRESS statement_attachment = StatementAttachmentFactory.build( account_id=account_id, statement_attachment_status=status, statement_period_id=statement_period_id ) mock_model.get_by_statement_period.return_value = [statement_attachment] result = logic.trigger_statement_attachment_generation( statement_period_id, None, correlation_id) assert result == { 'status': 'OK', 'messages_sent': 1, } mock_model.get_by_statement_period.assert_called_once_with( statement_period_id, statuses=[ StatementAttachmentStatus.IN_PROGRESS, StatementAttachmentStatus.ERROR, ] ) mock_sqs.send_messages.assert_called_once_with( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, [ { 'account_id': account_id, 'subaccount_id': None, 'statement_period_id': statement_period_id, } ] ) @patch('moneyhub.logic.statement_attachment.models.StatementAttachment') @patch('moneyhub.logic.statement_attachment.sqs') def test_trigger_unique_statement_attachment_generation(mock_sqs, mock_model): """Test triggering unique statement attachment generation for statement period. Only Statement Attachments with unique account_id should be processed. """ account_id = 3 subaccount_id = 10001 statement_period_id = 3 correlation_id = '123' status = StatementAttachmentStatus.IN_PROGRESS statement_attachments = [ StatementAttachmentFactory.build( account_id=account_id, statement_attachment_status=status, statement_period_id=statement_period_id ), StatementAttachmentFactory.build( account_id=account_id + 1, statement_attachment_status=status, statement_period_id=statement_period_id ), StatementAttachmentFactory.build( account_id=account_id + 1, statement_attachment_status=status, statement_period_id=statement_period_id ), StatementAttachmentFactory.build( account_id=account_id + 2, subaccount_id=subaccount_id, statement_attachment_status=status, statement_period_id=statement_period_id ) ] mock_model.get_by_statement_period.return_value = statement_attachments result = logic.trigger_statement_attachment_generation( statement_period_id, None, correlation_id) assert result == { 'status': 'OK', 'messages_sent': 3, } mock_model.get_by_statement_period.assert_called_once_with( statement_period_id, statuses=[ StatementAttachmentStatus.IN_PROGRESS, StatementAttachmentStatus.ERROR, ] ) mock_sqs.send_messages.assert_called_once() (queue_name, correlation_id_param, messages) = mock_sqs.send_messages.call_args[0] assert queue_name == Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME assert correlation_id_param == correlation_id assert sorted(messages, key=lambda x: x['account_id']) == [ { 'account_id': account_id, 'subaccount_id': None, 'statement_period_id': statement_period_id, }, { 'account_id': account_id + 1, 'subaccount_id': None, 'statement_period_id': statement_period_id, }, { 'account_id': account_id + 2, 'subaccount_id': subaccount_id, 'statement_period_id': statement_period_id, }, ] @patch('moneyhub.logic.statement_attachment.models.StatementAttachment') @patch('moneyhub.logic.statement_attachment.sqs') def test_trigger_statement_attachment_generation_with_account(mock_sqs, mock_model): """Test triggering statement attachment generation for a single account.""" account_id = 24601 statement_period_id = 123 correlation_id = 'abcd-1234' mock_model.get_by_statement_period.return_value = [ StatementAttachmentFactory.build( account_id=account_id, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, statement_period_id=statement_period_id ), StatementAttachmentFactory.build( account_id=account_id + 1, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, statement_period_id=statement_period_id ), StatementAttachmentFactory.build( account_id=account_id + 2, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, statement_period_id=statement_period_id ) ] result = logic.trigger_statement_attachment_generation( statement_period_id, account_id, correlation_id) assert result == { 'status': 'OK', 'messages_sent': 1, } mock_model.get_by_statement_period.assert_called_once_with( statement_period_id, statuses=[ StatementAttachmentStatus.IN_PROGRESS, StatementAttachmentStatus.ERROR, ] ) mock_sqs.send_messages.assert_called_once_with( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, [ { 'account_id': account_id, 'subaccount_id': None, 'statement_period_id': statement_period_id, } ] ) @patch('moneyhub.logic.statement_attachment.create_presigned_url') @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.profile_has_access_to_resource') def test_get_invoice_presigned_url( mock_profile_has_access_to_resource, mock_models, mock_create_presigned_url ): """Test getting invoice's presigned url to download.""" account_id = 26 statement_period_id = 26 profile_type = 'MoneyhubProfile' profile_id = 54321 statement_attachment = StatementAttachmentFactory.create( account_id=account_id, statement_period_id=statement_period_id, invoice_number='26_26_01', file_location='s3://invoice-files/2021/invoice_1.pdf' ) expected = 's3://test.pdf' mock_profile_has_access_to_resource.return_value = True mock_models.StatementAttachment.get_by_id_or_error.return_value = \ statement_attachment mock_create_presigned_url.return_value = expected result = logic.get_invoice_presigned_url( statement_attachment.statement_attachment_id, profile_type, profile_id) assert result == expected mock_create_presigned_url.assert_called_once_with( 'invoice-files', '2021/invoice_1.pdf' ) mock_profile_has_access_to_resource.assert_called_once_with( profile_type, profile_id, account_id, None) @patch('moneyhub.logic.statement_attachment.create_presigned_url') @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.profile_has_access_to_resource') def test_get_invoice_presigned_url_subaccount_id( mock_profile_has_access_to_resource, mock_models, mock_create_presigned_url ): """Test getting invoice's presigned URL when the attachment has a subaccount ID.""" account_id = 26 subaccount_id = 54321 profile_type = 'MoneyhubProfile' profile_id = 54321 statement_attachment = StatementAttachmentFactory.create( account_id=account_id, subaccount_id=subaccount_id, file_location='s3://invoice-files/2021/invoice_1.pdf' ) expected = 's3://test.pdf' mock_profile_has_access_to_resource.return_value = True mock_models.StatementAttachment.get_by_id_or_error.return_value = \ statement_attachment mock_create_presigned_url.return_value = expected result = logic.get_invoice_presigned_url( statement_attachment.statement_attachment_id, profile_type, profile_id) assert result == expected mock_profile_has_access_to_resource.assert_called_once_with( profile_type, profile_id, account_id, subaccount_id) @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.profile_has_access_to_resource') def test_get_invoice_presigned_url_failure( mock_profile_has_access_to_resource, mock_models ): """Test to generate invoice's presigned url failed.""" account_id = 27 statement_period_id = 27 statement_attachment = StatementAttachmentFactory.create( account_id=account_id, statement_period_id=statement_period_id, invoice_number='27_27_01', file_location=None ) statement_attachment_id = statement_attachment.statement_attachment_id expected = NO_INVOICE_FILE_LOCATION.format( statement_attachment_id=statement_attachment_id) mock_profile_has_access_to_resource.return_value = True mock_models.StatementAttachment.get_by_id_or_error.return_value = \ statement_attachment with pytest.raises(Exception, match=expected): logic.get_invoice_presigned_url(statement_attachment_id, 'MoneyhubProfile', 54321) @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.profile_has_access_to_resource') def test_get_invoice_presigned_url_forbidden_access( mock_profile_has_access_to_resource, mock_models ): """Test to generate invoice's presigned url when the profile doesn't have access.""" account_id = 28 statement_period_id = 28 statement_attachment = StatementAttachmentFactory.create( account_id=account_id, statement_period_id=statement_period_id, invoice_number='28_28_01', file_location=None ) mock_profile_has_access_to_resource.return_value = False mock_models.StatementAttachment.get_by_id_or_error.return_value = \ statement_attachment with pytest.raises(HTTPException) as error: logic.get_invoice_presigned_url( statement_attachment.statement_attachment_id, 'MoneyhubProfile', 54321) assert error.value.status_code == 403 assert error.value.detail == FORBIDDEN_INVOICE_URL_ACCESS @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.sqs') def test_regenerate_statement_attachments(mock_sqs, mock_models): """Test regenerating statement attachments.""" statement_period_id = 123 correlation_id = 'abcd' existing_attachments = [ StatementAttachmentFactory.build( account_id=24601, contract_id=10001, statement_period_id=statement_period_id, statement_attachment_type=StatementAttachmentType.DISTRIBUTION_FEE_INVOICE, # noqa: E501 statement_attachment_status=StatementAttachmentStatus.COMPLETE, ), StatementAttachmentFactory.build( account_id=24601, contract_id=10001, statement_period_id=statement_period_id, statement_attachment_type=StatementAttachmentType.SELF_BILLING_INVOICE, statement_attachment_status=StatementAttachmentStatus.ERROR, failure_reason='Computer went nutso' ), StatementAttachmentFactory.build( account_id=90210, contract_id=10001, statement_period_id=statement_period_id, statement_attachment_type=StatementAttachmentType.REVENUE_DETAIL, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, ), ] mock_models.StatementAttachment.get_by_statement_period.return_value = existing_attachments result = logic.regenerate_statement_attachments(123, None, None, correlation_id) assert len(result) == len(existing_attachments) assert all( r.statement_attachment_status == StatementAttachmentStatus.IN_PROGRESS and r.failure_reason is None for r in result ) mock_models.StatementAttachment.commit_changes.assert_called_once() mock_models.StatementAttachment.get_by_statement_period.assert_called_once_with( statement_period_id, account_id=None, types=None) mock_sqs.send_messages.assert_called_once_with( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, [ {'account_id': 24601, 'statement_period_id': statement_period_id}, {'account_id': 90210, 'statement_period_id': statement_period_id} ] ) @patch('moneyhub.logic.statement_attachment.models') @patch('moneyhub.logic.statement_attachment.sqs') def test_regenerate_statement_attachments_no_attachments(mock_sqs, mock_models): """Test regenerating attachments when there are no attachments.""" mock_models.StatementAttachment.get_by_statement_period.return_value = [] result = logic.regenerate_statement_attachments( 123, 24601, StatementAttachmentType.REVENUE_DETAIL) assert result == [] mock_models.StatementAttachment.get_by_statement_period.assert_called_once_with( 123, account_id=24601, types=[StatementAttachmentType.REVENUE_DETAIL]) mock_models.StatementAttachment.commit_changes.assert_not_called() mock_sqs.send_messages.assert_not_called() @patch('moneyhub.logic.statement_attachment.models') @pytest.mark.parametrize('statement_attachment_id', [1, 111, 9001]) def test_delete_statement_attachment(mock_models, statement_attachment_id): """Test deleting a statement attachment.""" logic.delete_statement_attachment(statement_attachment_id) mock_models.StatementAttachment.delete_by_id.assert_called_once_with(statement_attachment_id)