"""Ledger adjustment logic tests.""" from collections import namedtuple from decimal import Decimal from unittest.mock import patch from moneyhub.constants.constants import FLOWTHROUGH_ADJUSTMENT_TYPE_ID from moneyhub.constants.constants import GroupBy from moneyhub.constants.constants import PAGINATION_TYPE_STANDARD from moneyhub.logic import ledger_adjustment as logic from moneyhub.schemas import PaginationSchema from moneyhub.schemas.ledger_adjustment import BreakdownItemSchema from moneyhub.schemas.ledger_adjustment import LedgerAdjustmentAppliedSchema from moneyhub.schemas.ledger_adjustment import LedgerAdjustmentBreakdownItemSchema from moneyhub.schemas.ledger_adjustment import LedgerAdjustmentExpenseSchema from moneyhub.schemas.ledger_adjustment import PaginatedExpenseSchema from tests.utils.factories import AdjustmentsByTypeFactory from tests.utils.factories import CombinedAdjustmentsFactory from tests.utils.factories import ExpensesByArtistFactory from tests.utils.factories import ExpensesByImprintFactory from tests.utils.factories import ExpensesBySubaccountFactory from tests.utils.factories import ExpensesFactory from tests.utils.factories import ReferenceAdjustmentTypeFactory ExtendedBreakdownItemSchema = namedtuple( 'ExtendedBreakdownItemSchema', [ 'adjustment_amount_payee_currency', 'adjustment_payee_currency_code', 'statement_period_id', 'reference_adjustment_type_id', 'reference_adjustment_type_name' ] ) ExpenseGroupedByExpenseTypeId = namedtuple( 'ExpenseGroupedByExpenseTypeId', [ 'account_id', 'reference_adjustment_type_id', 'adjustment_amount_payee_currency', 'adjustment_payee_currency_code', 'reference_adjustment_type_name' ] ) ExtendedWorksheetAdjustment = namedtuple( 'ExtendedWorksheetAdjustment', [ 'worksheet_adjustment_id', 'abacus_event_id', 'account_id', 'activity_statement_period_id', 'adjustment_amount', 'adjustment_currency_code', 'adjustment_payee_currency_code', 'adjustment_amount_payee_currency', 'apply_to_statement_period_id', 'contract_id', 'reference_adjustment_type_id', 'reference_adjustment_type_name', 'note', 'upc', 'distribution_type', ] ) ExtendedWorksheetAdjustmentDetail = namedtuple( 'ExtendedWorksheetAdjustmentDetail', [ 'worksheet_adjustment_detail_id', 'worksheet_adjustment_id', 'abacus_event_id', 'account_id', 'activity_statement_period_id', 'adjustment_amount', 'adjustment_currency_code', 'adjustment_payee_currency_code', 'adjustment_amount_payee_currency', 'apply_to_statement_period_id', 'contract_id', 'reference_adjustment_type_id', 'reference_adjustment_type_name', 'note', 'upc', 'distribution_type', ] ) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_by_account_id(mock_models): """Test get_by_account_id method.""" account_id = 24601 contract_id = 1234 adjustment_items = [ CombinedAdjustmentsFactory.build( worksheet_adjustment_id=1, account_id=account_id, activity_statement_period_id=285, adjustment_payee_currency_code='AUD', adjustment_amount_payee_currency=Decimal('1290.99'), apply_to_statement_period_id=286, contract_id=contract_id, reference_adjustment_type_id=1, reference_adjustment_type_name='Label Earnings', note='mock adjustment' ), CombinedAdjustmentsFactory.build( worksheet_adjustment_id=2, account_id=account_id, activity_statement_period_id=285, adjustment_amount_payee_currency=Decimal('1315.99'), adjustment_payee_currency_code='AUD', apply_to_statement_period_id=286, contract_id=contract_id, reference_adjustment_type_id=1, reference_adjustment_type_name='Label Earnings', note='mock adjustment' ), CombinedAdjustmentsFactory.build( worksheet_adjustment_id=3, account_id=account_id, activity_statement_period_id=286, adjustment_amount_payee_currency=Decimal('9001.12'), adjustment_payee_currency_code='USD', apply_to_statement_period_id=287, contract_id=contract_id, reference_adjustment_type_id=1, reference_adjustment_type_name='Label Earnings', note='mock adjustment' ), ] visible_statement_periods = [285, 286] mock_models.CombinedAdjustments.get_by_account_id.return_value = adjustment_items mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.return_value = \ visible_statement_periods result = logic.get_by_account_id( account_id, contract_id, None, None, None) assert result == [ adjustment_items[0], adjustment_items[1], ] mock_models.CombinedAdjustments.get_by_account_id.assert_called_once_with( account_id, contract_id, None, None, None) mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.assert_called_once_with( # noqa: E501 account_id) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_expenses_by_account_id_grouped_by_subaccount(mock_models): """Test get_expenses_by_account_id method grouped_by_subaccount.""" account_id = 24601 contract_id = 1234 group_by = GroupBy.SUBACCOUNT_ID total_records = 1 expense1 = ExpensesBySubaccountFactory.build( adjustment_amount_payee_currency=Decimal('1315.99'), adjustment_payee_currency_code='AUD', account_id=account_id ) expenses = [expense1] expected_response = PaginatedExpenseSchema( items=[expense1], pagination=PaginationSchema( pagination_type=PAGINATION_TYPE_STANDARD, total_records=1 ) ) mock_models.ExpensesBySubaccount.get_by_subaccount.return_value = (expenses, total_records) result = logic.get_expenses_by_account_id( account_id, None, None, contract_id, group_by=group_by) assert result == expected_response mock_models.ExpensesBySubaccount.get_by_subaccount.assert_called_once_with( account_id, None, None, contract_id, None, None, None, None, None, None) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_adjustments_by_account_and_statement_periods_empty(mock_models): """Test get_adjustments_by_account_and_statement_periods method when empty.""" account_id = 123 statement_period_id = 247 adjustment_items = [] expected_response = [] mock_models.AdjustmentsByType.get_by_account_id.return_value = adjustment_items mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.return_value = [ statement_period_id] res = logic.get_adjustments_by_account_and_statement_periods( account_id, None, [statement_period_id]) assert res == expected_response mock_models.AdjustmentsByType.get_by_account_id.assert_called_once_with( account_id, None, [statement_period_id]) mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.assert_called_once_with( # noqa: E501 account_id) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_adjustments_by_account_and_statement_periods_not_visible(mock_models): """Test getting adjustments when the statement periods are not all visible.""" account_id = 123 statement_period_id = 247 contract_id = 1334 mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.return_value = [ statement_period_id - 1, statement_period_id, statement_period_id + 1, ] mock_models.AdjustmentsByType.get_by_account_id.return_value = [ AdjustmentsByTypeFactory.build(statement_period_id=1), ] result = logic.get_adjustments_by_account_and_statement_periods( account_id, contract_id, [statement_period_id, 290]) assert result == [] mock_models.AdjustmentsByType.get_by_account_id.assert_called_once_with( account_id, contract_id, [statement_period_id, 290]) mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.assert_called_once_with( # noqa: E501 account_id) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_adjustments_by_account_and_statement_periods(mock_models): """Test getting adjustments when using combined adjustments.""" account_id = 24601 statement_period_id = 123 contract_id = 1337 adjustment_items = [ AdjustmentsByTypeFactory.build( account_id=account_id, contract_id=contract_id, statement_period_id=statement_period_id, adjustment_amount_payee_currency=Decimal('4581.98'), adjustment_payee_currency_code='NOK', reference_adjustment_type_id=1, reference_adjustment_type_name='Label Earnings' ), ] expected_response = [ LedgerAdjustmentAppliedSchema( adjustment_total_payee_currency=Decimal('4581.98'), adjustment_payee_currency_code='NOK', breakdown_items=[ LedgerAdjustmentBreakdownItemSchema( adjustment_total_payee_currency=Decimal('4581.98'), adjustment_payee_currency_code='NOK', contract_id=None, reference_adjustment_type_id=1, reference_adjustment_type_name='Label Earnings' ) ], statement_period_id=statement_period_id ), ] mock_models.AdjustmentsByType.get_by_account_id.return_value = adjustment_items mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.return_value = [ statement_period_id] res = logic.get_adjustments_by_account_and_statement_periods( account_id, contract_id, [statement_period_id]) assert res == expected_response mock_models.AdjustmentsByType.get_by_account_id.assert_called_once_with( account_id, contract_id, [statement_period_id]) mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.assert_called_once_with( # noqa: E501 account_id) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_flowthrough_adjustments_by_account_and_statement_periods(mock_models): """Test getting flowthrough adjustments when using combined adjustments.""" account_id = 24601 statement_period_id = 123 adjustment_items = [ AdjustmentsByTypeFactory.build( account_id=account_id, contract_id=10001, statement_period_id=statement_period_id, adjustment_amount_payee_currency=Decimal('100.00'), adjustment_payee_currency_code='NOK', reference_adjustment_type_id=FLOWTHROUGH_ADJUSTMENT_TYPE_ID, reference_adjustment_type_name='Flowthrough' ), AdjustmentsByTypeFactory.build( account_id=account_id, contract_id=10002, statement_period_id=statement_period_id, adjustment_amount_payee_currency=Decimal('-100.00'), adjustment_payee_currency_code='NOK', reference_adjustment_type_id=FLOWTHROUGH_ADJUSTMENT_TYPE_ID, reference_adjustment_type_name='Flowthrough' ), ] expected_response = [ LedgerAdjustmentAppliedSchema( adjustment_total_payee_currency=Decimal('0'), adjustment_payee_currency_code='NOK', breakdown_items=[ LedgerAdjustmentBreakdownItemSchema( adjustment_total_payee_currency=Decimal('100.00'), adjustment_payee_currency_code='NOK', contract_id=10001, reference_adjustment_type_id=FLOWTHROUGH_ADJUSTMENT_TYPE_ID, reference_adjustment_type_name='Flowthrough' ), LedgerAdjustmentBreakdownItemSchema( adjustment_total_payee_currency=Decimal('-100.00'), adjustment_payee_currency_code='NOK', contract_id=10002, reference_adjustment_type_id=FLOWTHROUGH_ADJUSTMENT_TYPE_ID, reference_adjustment_type_name='Flowthrough' ) ], statement_period_id=statement_period_id ), ] mock_models.AdjustmentsByType.get_by_account_id.return_value = adjustment_items mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.return_value = [ statement_period_id] res = logic.get_adjustments_by_account_and_statement_periods( account_id, None, [statement_period_id]) assert res == expected_response mock_models.AdjustmentsByType.get_by_account_id.assert_called_once_with( account_id, None, [statement_period_id]) mock_models.StatementPeriodPaymentEntity.get_visible_statement_period_ids.assert_called_once_with( # noqa: E501 account_id) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_expenses_by_account_and_statement_periods( mock_models): """Test get_expenses_by_account_and_statement_periods method ff enabled.""" account_id = 123 contract_id = 34423 statement_period_id_1 = 247 statement_period_id_2 = 248 item1 = ExtendedBreakdownItemSchema( adjustment_amount_payee_currency=Decimal('5581.98'), adjustment_payee_currency_code='NOK', statement_period_id=statement_period_id_1, reference_adjustment_type_id=1, reference_adjustment_type_name='Label Earnings' ) item2 = ExtendedBreakdownItemSchema( adjustment_amount_payee_currency=Decimal('3000.00'), adjustment_payee_currency_code='NOK', statement_period_id=statement_period_id_2, reference_adjustment_type_id=1, reference_adjustment_type_name='Label Earnings' ) expense_items = [item1, item2] expected_response = [ LedgerAdjustmentExpenseSchema( amount=Decimal('5581.98'), currency_code='NOK', statement_period_id=statement_period_id_1, breakdown_items=[ BreakdownItemSchema( adjustment_total_payee_currency=Decimal('5581.98'), adjustment_payee_currency_code='NOK', reference_adjustment_type_id=1, reference_adjustment_type_name='Label Earnings' ) ] ), LedgerAdjustmentExpenseSchema( amount=Decimal('3000.00'), currency_code='NOK', statement_period_id=statement_period_id_2, breakdown_items=[ BreakdownItemSchema( adjustment_total_payee_currency=Decimal('3000.00'), adjustment_payee_currency_code='NOK', reference_adjustment_type_id=1, reference_adjustment_type_name='Label Earnings' ) ] ) ] mock_models.Expenses. \ get_expenses_by_account_and_statement_periods.return_value = \ expense_items res = logic.get_expenses_by_account_and_statement_periods( account_id, [statement_period_id_1, statement_period_id_2], contract_id) assert res == expected_response mock_models.Expenses. \ get_expenses_by_account_and_statement_periods. \ assert_called_once_with( account_id, [statement_period_id_1, statement_period_id_2], contract_id) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_ledger_adjustments_types_by_account_id(mock_models): """Test getting ledger adjustment types by account id.""" account_id = 24601 adjustment_types = [ { 'reference_adjustment_type_id': 9, 'type_name': 'Monthly Royalty Payment' }, { 'reference_adjustment_type_id': 81, 'type_name': 'VAX tax', } ] mock_models.CombinedAdjustments.get_adjustments_types_by_account_id.return_value \ = adjustment_types result = logic.get_ledger_adjustments_types_by_account_id(account_id) assert result == adjustment_types mock_models.CombinedAdjustments.get_adjustments_types_by_account_id.assert_called_once_with( account_id) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_expenses_artists_by_account_id(mock_models): """Test getting expense artists by account id.""" account_id = 24601 expected = [1, 2, 3] mock_models.Expenses.get_artists_by_account.return_value = expected result = logic.get_expenses_artists_by_account_id(account_id) assert result == expected mock_models.Expenses.get_artists_by_account.assert_called_once_with(account_id) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_expenses_subaccounts_by_account_id( mock_models): """Test getting expense subaccounts by account id.""" account_id = 24601 expected = [ { 'account_id': account_id, 'subaccount_id': 7111, 'subaccount_name': 'Another Subaccount Name', }, { 'account_id': account_id, 'subaccount_id': 7116, 'subaccount_name': 'Subaccount Name' } ] mock_models.Expenses.get_expenses_subaccounts_by_account_id.return_value = expected result = logic.get_expenses_subaccounts_by_account_id(account_id) assert result == expected mock_models.Expenses.get_expenses_subaccounts_by_account_id.assert_called_once_with( account_id) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_expenses_upcs_by_account_id(mock_models): """Test getting expense UPCs by account id ff enabled.""" account_id = 24601 mock_models.Expenses.get_upcs_by_account.return_value = [1, 2, 3] result = logic.get_expenses_upcs_by_account_id(account_id) assert result == [1, 2, 3] mock_models.Expenses.get_upcs_by_account.assert_called_once_with(account_id) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_expenses_types_by_account_id( mock_models): """Test getting expense types by account id.""" account_id = 24601 expenses_types = [ ReferenceAdjustmentTypeFactory.build( reference_adjustment_type_id=100, type_name='Mechanicals', oa_category_name=None ), ReferenceAdjustmentTypeFactory.build( reference_adjustment_type_id=81, type_name='VAX tax', oa_category_name=None ) ] mock_models.Expenses.get_expenses_types_by_account_id.return_value \ = expenses_types result = logic.get_expenses_types_by_account_id(account_id) assert result == [ expenses_types[0], expenses_types[1] ] mock_models.Expenses.get_expenses_types_by_account_id.assert_called_once_with( account_id) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_expenses_by_account_id_grouped_by_expense_type_id(mock_models): """Test get_by_account_id method grouped by expense_type_id.""" account_id = 24601 contract_id = 1234 statement_period_id_start = 250 statement_period_id_end = 300 group_by = GroupBy.EXPENSE_TYPE_ID total_records = 1 payload = { 'items': [ { 'account_id': account_id, 'reference_adjustment_type_id': 1, 'adjustment_amount_payee_currency': Decimal('4581.98'), 'adjustment_payee_currency_code': 'NOK', 'reference_adjustment_type_name': 'Physical Rework' } ], 'pagination': { 'pagination_type': 'standard', 'total_records': 1, }, } adjustment_items = ExpenseGroupedByExpenseTypeId( account_id=account_id, reference_adjustment_type_id=1, adjustment_amount_payee_currency=Decimal('4581.98'), adjustment_payee_currency_code='NOK', reference_adjustment_type_name='Physical Rework' ) mock_models.Expenses.get_by_account_id.return_value = ( [adjustment_items._asdict()], total_records) result = logic.get_expenses_by_account_id( account_id, 50, 0, contract_id, statement_period_id_start, statement_period_id_end, None, None, None, None, None, group_by) assert result.__class__ == PaginatedExpenseSchema assert result == PaginatedExpenseSchema(**payload) mock_models.Expenses.get_by_account_id.assert_called_once_with( account_id, 50, 0, contract_id, statement_period_id_start, statement_period_id_end, None, None, None, None, None, group_by) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_expenses_by_account_id(mock_models): """Test get_by_account_id method when using expenses dbt.""" account_id = 24601 contract_id = 1234 artist_id = 12341231 total_records = 1 expenses = ExpensesFactory.build( account_id=account_id, contract_id=1234, adjustment_amount_payee_currency=Decimal('60.00')) payload = { 'items': [ { 'account_id': account_id, 'reference_adjustment_type_id': 1, 'apply_to_statement_period_id': 123, 'adjustment_amount_payee_currency': Decimal('60.00'), 'adjustment_payee_currency_code': 'USD', 'contract_id': 1234, 'upc': '101010', 'reference_adjustment_type_name': 'Freight' } ], 'pagination': { 'pagination_type': 'standard', 'total_records': 1, }, } mock_models.Expenses.get_by_account_id.return_value = ( [expenses.to_dict()], total_records) result = logic.get_expenses_by_account_id(account_id, 50, 0, contract_id, artist_id=artist_id) assert result.__class__ == PaginatedExpenseSchema assert result == PaginatedExpenseSchema(**payload) mock_models.Expenses.get_by_account_id.assert_called_once_with( account_id, 50, 0, contract_id, None, None, None, None, None, artist_id, None, None) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_expenses_by_imprint(mock_models): """Test getting expense by imprint id.""" account_id = 24601 imprint_name = 'Bad Seed Ltd' amount = 20 total_records = 1 group_by = GroupBy.IMPRINT_ID payload = { 'items': [ { 'account_id': account_id, 'adjustment_amount_payee_currency': Decimal('20.00'), 'adjustment_payee_currency_code': 'USD', 'imprint': 'Bad Seed Ltd', 'imprint_id': 123456 } ], 'pagination': { 'pagination_type': 'standard', 'total_records': 1, }, } fake_imprint = ExpensesByImprintFactory.build( account_id=account_id, imprint=imprint_name, adjustment_amount_payee_currency=amount) mock_models.ExpensesByImprint.get_by_imprint_account_id.return_value = ( [fake_imprint.to_dict()], total_records) result = logic.get_expenses_by_account_id(account_id, 50, 0, group_by=group_by) assert result.__class__ == PaginatedExpenseSchema assert result == PaginatedExpenseSchema(**payload) mock_models.ExpensesByImprint.get_by_imprint_account_id.assert_called_once_with( account_id, 50, 0, None, None, None, None, None, None, None) @patch('moneyhub.logic.ledger_adjustment.models') def test_get_expenses_by_artist(mock_models): """Test getting expense by artist id.""" account_id = 24601 artist_name = 'Best Artist' amount = 20 total_records = 1 group_by = GroupBy.ARTIST_ID payload = { 'items': [ { 'account_id': account_id, 'adjustment_amount_payee_currency': Decimal('20.00'), 'adjustment_payee_currency_code': 'USD', 'artist_name': 'Best Artist', 'artist_id': 123456 } ], 'pagination': { 'pagination_type': 'standard', 'total_records': 1, }, } fake_artist = ExpensesByArtistFactory.build( account_id=account_id, artist_name=artist_name, adjustment_amount_payee_currency=amount) mock_models.ExpensesByArtist.get_by_artist_by_account_id.return_value = ( [fake_artist.to_dict()], total_records) result = logic.get_expenses_by_account_id(account_id, 50, 0, group_by=group_by) assert result.__class__ == PaginatedExpenseSchema assert result == PaginatedExpenseSchema(**payload) mock_models.ExpensesByArtist.get_by_artist_by_account_id.assert_called_once_with( account_id, 50, 0, None, None, None, None, None, None, None)