diff --git a/tests/test_reports_ledger.py b/tests/test_reports_ledger.py index db95bfb50a27154b5812ecb4af77bcd84de5c695..6d179917fe3c7b868cbf00af5d0dcff7c5cb4d0d 100644 --- a/tests/test_reports_ledger.py +++ b/tests/test_reports_ledger.py @@ -19,6 +19,7 @@ import contextlib import copy import datetime import io +import itertools import re import pytest @@ -63,6 +64,12 @@ START_DATE = datetime.date(2018, 3, 1) MID_DATE = datetime.date(2019, 3, 1) STOP_DATE = datetime.date(2020, 3, 1) +REPORT_KWARGS = [ + {'report_class': ledger.LedgerODS}, + *({'report_class': ledger.TransactionODS, 'txn_filter': flags} + for flags in ledger.TransactionFilter), +] + @pytest.fixture def ledger_entries(): return copy.deepcopy(_ledger_load[0]) @@ -101,13 +108,17 @@ class ExpectedPostings(core.RelatedPostings): cls.find_section(ods, data.Account(account)) @classmethod - def check_in_report(cls, ods, account, start_date=START_DATE, end_date=STOP_DATE): + def check_in_report(cls, ods, account, + start_date=START_DATE, end_date=STOP_DATE, txn_filter=None): date = end_date + datetime.timedelta(days=1) txn = testutil.Transaction(date=date, postings=[ (account, 0), ]) related = cls(data.Posting.from_txn(txn)) - related.check_report(ods, start_date, end_date) + if txn_filter is None: + related.check_report(ods, start_date, end_date) + else: + related.check_txn_report(ods, txn_filter, start_date, end_date) def slice_date_range(self, start_date, end_date): postings = enumerate(self) @@ -156,6 +167,64 @@ class ExpectedPostings(core.RelatedPostings): empty = '$0.00' if expect_posts else '0' assert closing_row[4].text == closing_bal.format(None, empty=empty, sep='\0') + def _post_data_from_row(self, row): + if row[4].text: + number = row[4].value + match = re.search(r'([A-Z]{3})\d*Cell', row[4].getAttribute('stylename') or '') + assert match + currency = match.group(1) + else: + number = row[5].value + currency = 'USD' + return (row[2].text, row[3].text, number, currency) + + def _post_data_from_post(self, post, norm_func): + return ( + post.account, + post.meta.get('entity') or '', + norm_func(post.units.number), + post.units.currency, + ) + + def check_txn_report(self, ods, txn_filter, start_date, end_date, expect_totals=True): + account = self[0].account + norm_func = core.normalize_amount_func(account) + open_bal, expect_posts = self.slice_date_range(start_date, end_date) + open_bal = norm_func(open_bal) + period_bal = core.MutableBalance() + rows = self.find_section(ods, account) + if (expect_totals + and txn_filter == ledger.TransactionFilter.ALL + and account.is_under('Assets', 'Liabilities')): + opening_row = testutil.ODSCell.from_row(next(rows)) + assert opening_row[0].value == start_date + assert opening_row[5].text == open_bal.format(None, empty='0', sep='\0') + period_bal += open_bal + last_txn = None + for post in expect_posts: + txn = post.meta.txn + post_flag = ledger.TransactionFilter.post_flag(post) + if txn is last_txn or (not txn_filter & post_flag): + continue + last_txn = txn + row1 = testutil.ODSCell.from_row(next(rows)) + assert row1[0].value == txn.date + assert row1[1].text == (txn.narration or '') + expected = {self._post_data_from_post(post, norm_func) + for post in txn.postings} + actual = {self._post_data_from_row(testutil.ODSCell.from_row(row)) + for row in itertools.islice(rows, len(txn.postings) - 1)} + actual.add(self._post_data_from_row(row1)) + assert actual == expected + for post_acct, _, number, currency in expected: + if post_acct == account: + period_bal += testutil.Amount(number, currency) + if expect_totals: + closing_row = testutil.ODSCell.from_row(next(rows)) + assert closing_row[0].value == end_date + empty = '$0.00' if period_bal else '0' + assert closing_row[5].text == period_bal.format(None, empty=empty, sep='\0') + def get_sheet_names(ods): return [sheet.getAttribute('name').replace(' ', ':') @@ -259,14 +328,16 @@ def test_plan_sheets_full_split_required(caplog): assert actual == ['Assets:Bank:Checking', 'Assets:Bank:Savings', 'Assets'] assert not caplog.records -def build_report(ledger_entries, start_date, stop_date, *args, **kwargs): +def build_report(ledger_entries, start_date, stop_date, *args, + report_class=ledger.LedgerODS, **kwargs): postings = list(data.Posting.from_entries(iter(ledger_entries))) with clean_account_meta(): data.Account.load_openings_and_closings(iter(ledger_entries)) - report = ledger.LedgerODS(start_date, stop_date, *args, **kwargs) + report = report_class(start_date, stop_date, *args, **kwargs) report.write(iter(postings)) return postings, report +@pytest.mark.parametrize('report_kwargs', iter(REPORT_KWARGS)) @pytest.mark.parametrize('start_date,stop_date', [ (START_DATE, STOP_DATE), (START_DATE, MID_DATE), @@ -274,50 +345,77 @@ def build_report(ledger_entries, start_date, stop_date, *args, **kwargs): (START_DATE.replace(month=6), START_DATE.replace(month=12)), (STOP_DATE, STOP_DATE.replace(month=12)), ]) -def test_date_range_report(ledger_entries, start_date, stop_date): - postings, report = build_report(ledger_entries, start_date, stop_date) +def test_date_range_report(ledger_entries, start_date, stop_date, report_kwargs): + txn_filter = report_kwargs.get('txn_filter') + postings, report = build_report(ledger_entries, start_date, stop_date, **report_kwargs) expected = dict(ExpectedPostings.group_by_account(postings)) for account in iter_accounts(ledger_entries): try: related = expected[account] except KeyError: - ExpectedPostings.check_in_report(report.document, account, start_date, stop_date) + ExpectedPostings.check_in_report( + report.document, account, start_date, stop_date, txn_filter, + ) else: - related.check_report(report.document, start_date, stop_date) + if txn_filter is None: + related.check_report(report.document, start_date, stop_date) + else: + related.check_txn_report( + report.document, txn_filter, start_date, stop_date, + ) +@pytest.mark.parametrize('report_kwargs', iter(REPORT_KWARGS)) @pytest.mark.parametrize('tot_accts', [ (), ('Assets', 'Liabilities'), ('Income', 'Expenses'), ('Assets', 'Liabilities', 'Income', 'Expenses'), ]) -def test_report_filter_totals(ledger_entries, tot_accts): +def test_report_filter_totals(ledger_entries, tot_accts, report_kwargs): + txn_filter = report_kwargs.get('txn_filter') postings, report = build_report(ledger_entries, START_DATE, STOP_DATE, totals_with_entries=tot_accts, - totals_without_entries=tot_accts) + totals_without_entries=tot_accts, + **report_kwargs) expected = dict(ExpectedPostings.group_by_account(postings)) for account in iter_accounts(ledger_entries): expect_totals = account.startswith(tot_accts) if account in expected and expected[account][-1].meta.date >= START_DATE: - expected[account].check_report(report.document, START_DATE, STOP_DATE, - expect_totals=expect_totals) + if txn_filter is None: + expected[account].check_report( + report.document, START_DATE, STOP_DATE, expect_totals=expect_totals, + ) + else: + expected[account].check_txn_report( + report.document, txn_filter, + START_DATE, STOP_DATE, expect_totals=expect_totals, + ) elif expect_totals: - ExpectedPostings.check_in_report(report.document, account) + ExpectedPostings.check_in_report( + report.document, account, START_DATE, STOP_DATE, txn_filter, + ) else: ExpectedPostings.check_not_in_report(report.document, account) +@pytest.mark.parametrize('report_kwargs', iter(REPORT_KWARGS)) @pytest.mark.parametrize('accounts', [ ('Income', 'Expenses'), ('Assets:Receivable', 'Liabilities:Payable'), ]) -def test_account_names_report(ledger_entries, accounts): - postings, report = build_report(ledger_entries, START_DATE, STOP_DATE, accounts) +def test_account_names_report(ledger_entries, accounts, report_kwargs): + txn_filter = report_kwargs.get('txn_filter') + postings, report = build_report(ledger_entries, START_DATE, STOP_DATE, + accounts, **report_kwargs) expected = dict(ExpectedPostings.group_by_account(postings)) for account in iter_accounts(ledger_entries): - if account.startswith(accounts): + if not account.startswith(accounts): + ExpectedPostings.check_not_in_report(report.document, account) + elif txn_filter is None: expected[account].check_report(report.document, START_DATE, STOP_DATE) else: - ExpectedPostings.check_not_in_report(report.document, account) + expected[account].check_txn_report( + report.document, txn_filter, START_DATE, STOP_DATE, + ) def run_main(arglist, config=None): if config is None: @@ -424,6 +522,30 @@ def test_main_project_report(ledger_entries, project, start_date, stop_date): except KeyError: ExpectedPostings.check_not_in_report(ods, account) +@pytest.mark.parametrize('flag', [ + '--disbursements', + '--receipts', +]) +def test_main_cash_report(ledger_entries, flag): + if flag == '--receipts': + txn_filter = ledger.TransactionFilter.CREDIT + else: + txn_filter = ledger.TransactionFilter.DEBIT + retcode, output, errors = run_main([ + flag, + '-b', START_DATE.isoformat(), + '-e', STOP_DATE.isoformat(), + ]) + assert not errors.getvalue() + assert retcode == 0 + ods = odf.opendocument.load(output) + postings = data.Posting.from_entries(ledger_entries) + for account, expected in ExpectedPostings.group_by_account(postings): + if account == 'Assets:Checking' or account == 'Assets:PayPal': + expected.check_txn_report(ods, txn_filter, START_DATE, STOP_DATE) + else: + expected.check_not_in_report(ods) + @pytest.mark.parametrize('arg', [ 'Assets:NoneSuchBank', 'Funny money',