"""Tests for custom report logic.""" from unittest.mock import patch from fastapi import HTTPException import pytest from moneyhub.config import Config from moneyhub.constants.constants import NumberFormat from moneyhub.constants.constants import ReportCustomColumnDimension from moneyhub.constants.constants import ReportCustomFileType from moneyhub.constants.constants import ReportCustomRowDimension from moneyhub.constants.constants import ReportCustomStatus from moneyhub.constants.constants import RevenueDisplayType from moneyhub.constants.constants import RevenueType from moneyhub.constants.error import NO_REPORT_FILE_LOCATION from moneyhub.constants.error import STATEMENT_DATA_NOT_AVAILABLE from moneyhub.logic import report_custom as logic from moneyhub.schemas import ReportCustomCreateSchema from moneyhub.schemas.report_custom import ReportCustomFiltersSchema from tests.utils.factories import ReportCustomFactory from tests.utils.factories import StatementPeriodFactory @patch('moneyhub.logic.report_custom.models') @patch('moneyhub.logic.report_custom.sqs') def test_create_report(mock_sqs, mock_models): """Test creating custom report.""" account_id = 111 statement_period_ids = [1, 3, 2] dimension_column = ReportCustomColumnDimension.TERRITORY dimension_row = ReportCustomRowDimension.PRODUCT number_format = NumberFormat.EU file_type = ReportCustomFileType.XLS correlation_id = 'abc' revenue_type = RevenueType.DISTRIBUTION revenue_display_type = RevenueDisplayType.GROSS orchard_identity_id = 'orchard-identity' created_report = ReportCustomFactory.build() filters = ReportCustomFiltersSchema( store_ids=[286, 1, 348], country_codes=['US', 'GB'], transaction_type_ids=[1, 2, 3], imprint_ids=[1, 2, 3], activity_period_ids=[1, 2, 3], artist_ids=[2, 3, 4], product_ids=[3, 4, 5], project_ids=[1, 2, 3], subaccount_ids=[4, 5, 6], recording_ids=['5', '6', '7'], track_unique_ids=[6, 7, 8], ) payload = ReportCustomCreateSchema( statement_period_ids=statement_period_ids, revenue_type=revenue_type, revenue_display_type=revenue_display_type, dimension_column=dimension_column, dimension_row=dimension_row, filters=filters, number_format=number_format, file_type=file_type, ) mock_models.ReportCustom.build.return_value = created_report mock_models.AccountStatementPeriods.get_by_account_id.return_value = ([ StatementPeriodFactory.build( statement_period_id=1 ), StatementPeriodFactory.build( statement_period_id=2 ), StatementPeriodFactory.build( statement_period_id=3 ), StatementPeriodFactory.build( statement_period_id=4 ), StatementPeriodFactory.build( statement_period_id=5 ), ], 5) result = logic.create_report(account_id, payload, orchard_identity_id, correlation_id) assert result == created_report mock_models.ReportCustom.build.assert_called_once_with( account_id=account_id, contract_id=None, subaccount_id=None, statement_period_ids='1,2,3', revenue_type=revenue_type, revenue_display_type=revenue_display_type, report_custom_status=ReportCustomStatus.IN_PROGRESS, dimension_column=dimension_column, dimension_row=dimension_row, filters={ 'store_ids': [286, 1, 348], 'country_codes': ['US', 'GB'], 'transaction_type_ids': [1, 2, 3], 'imprint_ids': [1, 2, 3], 'activity_period_ids': [1, 2, 3], 'artist_ids': [2, 3, 4], 'product_ids': [3, 4, 5], 'subaccount_ids': [4, 5, 6], 'recording_ids': ['5', '6', '7'], 'track_unique_ids': [6, 7, 8], 'project_ids': [1, 2, 3], }, number_format=number_format, file_type=file_type, created_by=orchard_identity_id) mock_models.ReportCustom.commit_changes.assert_called_once() mock_sqs.send_message.assert_called_once_with( Config.SQS_MH_CUSTOM_REPORTS_QUEUE_NAME, correlation_id, {'report_custom_id': created_report.report_custom_id} ) mock_models.AccountStatementPeriods.get_by_account_id \ .assert_called_once_with(account_id, None) @patch('moneyhub.logic.report_custom.models') @patch('moneyhub.logic.report_custom.sqs') def test_create_report_defaults_revenue_display_type_to_net(mock_sqs, mock_models): """Test that revenue_display_type defaults to NET when omitted.""" account_id = 111 statement_period_ids = [1] created_report = ReportCustomFactory.build() payload = ReportCustomCreateSchema( statement_period_ids=statement_period_ids, revenue_type=RevenueType.DISTRIBUTION, dimension_column=ReportCustomColumnDimension.TERRITORY, dimension_row=ReportCustomRowDimension.PRODUCT, ) mock_models.ReportCustom.build.return_value = created_report mock_models.AccountStatementPeriods.get_by_account_id.return_value = ([ StatementPeriodFactory.build(statement_period_id=1), ], 1) logic.create_report(account_id, payload, 'orchard-identity', 'abc') _, build_kwargs = mock_models.ReportCustom.build.call_args assert build_kwargs['revenue_display_type'] == RevenueDisplayType.NET @patch('moneyhub.logic.report_custom.models') def test_create_report_statement_period_data_not_available(mock_models): """Test creating custom report if a statement period data is not available.""" account_id = 111 statement_period_ids = [1, 2] revenue_type = RevenueType.DISTRIBUTION dimension_column = ReportCustomColumnDimension.TERRITORY dimension_row = ReportCustomRowDimension.PRODUCT number_format = NumberFormat.EU file_type = ReportCustomFileType.XLS correlation_id = 'abc' created_report = ReportCustomFactory.build() payload = ReportCustomCreateSchema( statement_period_ids=statement_period_ids, revenue_type=revenue_type, dimension_column=dimension_column, dimension_row=dimension_row, number_format=number_format, file_type=file_type, ) mock_models.ReportCustom.build.return_value = created_report mock_models.AccountStatementPeriods.get_by_account_id.return_value = ([ StatementPeriodFactory.build( statement_period_id=1 ), StatementPeriodFactory.build( statement_period_id=3 ), StatementPeriodFactory.build( statement_period_id=4 ), ], 3) with pytest.raises(HTTPException) as error: logic.create_report( account_id, payload, 'orchard_identity_id', correlation_id) assert error.value.status_code == 400 assert error.value.detail == STATEMENT_DATA_NOT_AVAILABLE mock_models.ReportCustom.build.assert_not_called() @patch('moneyhub.logic.report_custom.models') def test_get_custom_report(mock_models): """Test fetching a custom report.""" report_id = 24601 report = ReportCustomFactory.build() mock_models.ReportCustom.get_by_id_or_error.return_value = report result = logic.get_custom_report(report_id) assert result == report mock_models.ReportCustom.get_by_id_or_error.assert_called_once_with(report_id) @patch('moneyhub.logic.report_custom.models') def test_get_custom_reports(mock_models): """Test fetching custom reports.""" account_id = 111 contract_id = 123 reports = [ReportCustomFactory.build()] mock_models.ReportCustom.get_by_account_id.return_value = reports result = logic.get_custom_reports(account_id, contract_id) assert result == reports mock_models.ReportCustom.get_by_account_id.assert_called_once_with( account_id, contract_id, None, False) @patch('moneyhub.logic.report_custom.models') def test_get_custom_reports_with_subaccount(mock_models): """Test fetching custom reports filtered by subaccount_id.""" account_id = 111 contract_id = 123 subaccount_id = 456 reports = [ReportCustomFactory.build()] mock_models.ReportCustom.get_by_account_id.return_value = reports result = logic.get_custom_reports(account_id, contract_id, subaccount_id, True) assert result == reports mock_models.ReportCustom.get_by_account_id.assert_called_once_with( account_id, contract_id, subaccount_id, True) @patch('moneyhub.logic.report_custom.create_presigned_url') @patch('moneyhub.logic.report_custom.models') def test_get_report_presigned_url( mock_models, mock_url ): """Test getting report's presigned url to download.""" account_id = 26 statement_period_id = 26 report_custom = ReportCustomFactory.build( report_custom_id=123, account_id=account_id, statement_period_ids=[statement_period_id], revenue_type='distribution', report_custom_status=ReportCustomStatus.COMPLETE, dimension_column=ReportCustomColumnDimension.TERRITORY, dimension_row=ReportCustomRowDimension.PRODUCT, file_location='s3://report-files/2021/report_1.pdf' ) report_custom_id = report_custom.report_custom_id mock_models.ReportCustom.get_by_id_or_error.return_value = \ report_custom logic.get_report_presigned_url(report_custom_id) mock_url.assert_called_once_with( 'report-files', '2021/report_1.pdf' ) @patch('moneyhub.logic.report_custom.models') def test_get_report_presigned_url_failure(mock_models): """Test to generate report's presigned url failed.""" account_id = 27 statement_period_id = 27 report_custom = ReportCustomFactory.build( report_custom_id=124, account_id=account_id, statement_period_ids=[statement_period_id], revenue_type='distribution', report_custom_status=ReportCustomStatus.COMPLETE, dimension_column=ReportCustomColumnDimension.TERRITORY, dimension_row=ReportCustomRowDimension.PRODUCT, file_location=None ) report_custom_id = report_custom.report_custom_id mock_models.ReportCustom.get_by_id_or_error.return_value = \ report_custom with pytest.raises( Exception, match=NO_REPORT_FILE_LOCATION.format( report_custom_id=report_custom_id ) ): logic.get_report_presigned_url(report_custom_id) @patch('moneyhub.logic.report_custom.models') def test_update_custom_report(mock_models): """Test fetching a custom report.""" report_id = 24601 file_location = 's3://test.csv' report_custom_status = ReportCustomStatus.COMPLETE payload = { 'file_location': file_location, 'report_custom_status': report_custom_status, } report = ReportCustomFactory.build( file_location=payload['file_location'], report_custom_status=payload['report_custom_status'] ) mock_models.ReportCustom.update_custom_report.return_value = report result = logic.update_custom_report(report_id, payload) assert result == report mock_models.ReportCustom.update_custom_report.assert_called_once_with( report_id, payload) @patch('moneyhub.logic.report_custom.models') def test_delete_custom_report(mock_models): """Test fetching custom report.""" report_custom_id = 11 file_location = 's3://test.csv' report_custom_status = ReportCustomStatus.COMPLETE report = ReportCustomFactory.build( file_location=file_location, report_custom_status=report_custom_status ) mock_models.ReportCustom.delete_by_id.return_value = report result = logic.delete_custom_report(report_custom_id) assert result == report mock_models.ReportCustom.delete_by_id.assert_called_once_with(report_custom_id) @patch('moneyhub.logic.report_custom.models') @patch('moneyhub.logic.report_custom.sqs') def test_regenerate_custom_reports(mock_sqs, mock_models): """Test regenerating custom reports.""" statement_period_id = 123 correlation_id = 'abcd' existing_reports = [ ReportCustomFactory.build( report_custom_id=1, account_id=24601, contract_id=10001, statement_period_ids=f'{statement_period_id}', report_custom_status=ReportCustomStatus.COMPLETE, ), ReportCustomFactory.build( report_custom_id=2, account_id=24601, contract_id=10001, statement_period_ids=f'{statement_period_id},999', report_custom_status=ReportCustomStatus.ERROR, ), ReportCustomFactory.build( report_custom_id=3, account_id=90210, contract_id=10001, statement_period_ids='999', report_custom_status=ReportCustomStatus.IN_PROGRESS, ), ] mock_models.ReportCustom.get_by_statement_period.return_value = existing_reports result = logic.regenerate_custom_reports(123, correlation_id) assert len(result) == len(existing_reports) assert all( r.report_custom_status == ReportCustomStatus.IN_PROGRESS for r in result ) mock_models.ReportCustom.commit_changes.assert_called_once() mock_models.ReportCustom.get_by_statement_period.assert_called_once_with( statement_period_id, None) mock_sqs.send_messages.assert_called_once_with( Config.SQS_MH_CUSTOM_REPORTS_QUEUE_NAME, correlation_id, [ {'report_custom_id': 1}, {'report_custom_id': 2}, {'report_custom_id': 3}, ] ) @patch('moneyhub.logic.report_custom.models') @patch('moneyhub.logic.report_custom.sqs') def test_regenerate_custom_reports_no_reports(mock_sqs, mock_models): """Test regenerating reports when there are no reports.""" statement_period_id = 123 mock_models.ReportCustom.get_by_statement_period.return_value = [] result = logic.regenerate_custom_reports(statement_period_id) assert result == [] mock_models.ReportCustom.get_by_statement_period.assert_called_once_with( statement_period_id, None) mock_models.ReportCustom.commit_changes.assert_not_called() mock_sqs.send_messages.assert_not_called()