b7aae7b3c0
This makes the output more useful for broad searches like on an entity. Invoices that cross FY boundaries will appear to be paid without being accrued, and so would appear when we were just filtering zeroed-out invoices. If we integrate the aging report into this module in the future, that'll need to follow different logic, and just filter out zeroed-out invoices. But the basic balance report and outgoing report are more workaday tools, where more filtering makes them more useful.
419 lines
14 KiB
Python
419 lines
14 KiB
Python
"""test_reports_accrual - Unit tests for accrual report"""
|
|
# Copyright © 2020 Brett Smith
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
|
import collections
|
|
import copy
|
|
import io
|
|
import itertools
|
|
import re
|
|
|
|
import pytest
|
|
|
|
from . import testutil
|
|
|
|
from beancount import loader as bc_loader
|
|
from conservancy_beancount import data
|
|
from conservancy_beancount import rtutil
|
|
from conservancy_beancount.reports import accrual
|
|
from conservancy_beancount.reports import core
|
|
|
|
_accruals_load = bc_loader.load_file(testutil.test_path('books/accruals.beancount'))
|
|
ACCRUALS_COUNT = sum(
|
|
1
|
|
for entry in _accruals_load[0]
|
|
for post in getattr(entry, 'postings', ())
|
|
if post.account.startswith(('Assets:Receivable:', 'Liabilities:Payable:'))
|
|
)
|
|
|
|
ACCOUNTS = [
|
|
'Assets:Receivable:Accounts',
|
|
'Assets:Receivable:Loans',
|
|
'Liabilities:Payable:Accounts',
|
|
'Liabilities:Payable:Vacation',
|
|
]
|
|
|
|
CONSISTENT_METADATA = [
|
|
'contract',
|
|
'entity',
|
|
'purchase-order',
|
|
]
|
|
|
|
class RTClient(testutil.RTClient):
|
|
TICKET_DATA = {
|
|
'40': [
|
|
('400', 'invoice feb.csv', 'text/csv', '40.4k'),
|
|
],
|
|
'44': [
|
|
('440', 'invoice feb.csv', 'text/csv', '40.4k'),
|
|
],
|
|
'490': [],
|
|
'505': [],
|
|
'510': [
|
|
('4000', 'contract.pdf', 'application/pdf', '1.4m'),
|
|
('5100', 'invoice april.pdf', 'application/pdf', '1.5m'),
|
|
('5105', 'payment.png', 'image/png', '51.5k'),
|
|
('6100', 'invoice may.pdf', 'application/pdf', '1.6m'),
|
|
],
|
|
'515': [],
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def accrual_entries():
|
|
return copy.deepcopy(_accruals_load[0])
|
|
|
|
@pytest.fixture
|
|
def accrual_postings():
|
|
entries = copy.deepcopy(_accruals_load[0])
|
|
return data.Posting.from_entries(entries)
|
|
|
|
def check_link_regexp(regexp, match_s, first_link_only=False):
|
|
assert regexp
|
|
assert re.search(regexp, match_s)
|
|
assert re.search(regexp, match_s + ' postlink')
|
|
assert re.search(regexp, match_s + '0') is None
|
|
assert re.search(regexp, '1' + match_s) is None
|
|
end_match = re.search(regexp, 'prelink ' + match_s)
|
|
if first_link_only:
|
|
assert end_match is None
|
|
else:
|
|
assert end_match
|
|
|
|
@pytest.mark.parametrize('link_fmt', [
|
|
'{}',
|
|
'rt:{}',
|
|
'rt://ticket/{}',
|
|
])
|
|
def test_search_term_parse_rt_shortcuts(link_fmt):
|
|
key, regexp = accrual.SearchTerm.parse(link_fmt.format(220))
|
|
assert key == 'rt-id'
|
|
check_link_regexp(regexp, 'rt:220', first_link_only=True)
|
|
check_link_regexp(regexp, 'rt://ticket/220', first_link_only=True)
|
|
|
|
@pytest.mark.parametrize('link_fmt', [
|
|
'{}/{}',
|
|
'rt:{}/{}',
|
|
'rt://ticket/{}/attachments/{}',
|
|
])
|
|
def test_search_term_parse_invoice_shortcuts(link_fmt):
|
|
key, regexp = accrual.SearchTerm.parse(link_fmt.format(330, 660))
|
|
assert key == 'invoice'
|
|
check_link_regexp(regexp, 'rt:330/660')
|
|
check_link_regexp(regexp, 'rt://ticket/330/attachments/660')
|
|
|
|
@pytest.mark.parametrize('key', [
|
|
'approval',
|
|
'contract',
|
|
'invoice',
|
|
])
|
|
def test_search_term_parse_metadata_rt_shortcut(key):
|
|
actual_key, regexp = accrual.SearchTerm.parse(f'{key}=440/420')
|
|
assert actual_key == key
|
|
check_link_regexp(regexp, 'rt:440/420')
|
|
check_link_regexp(regexp, 'rt://ticket/440/attachments/420')
|
|
|
|
@pytest.mark.parametrize('key', [
|
|
None,
|
|
'approval',
|
|
'contract',
|
|
'invoice',
|
|
])
|
|
def test_search_term_parse_repo_link(key):
|
|
document = '1234.pdf'
|
|
if key is None:
|
|
key = 'invoice'
|
|
search = document
|
|
else:
|
|
search = f'{key}={document}'
|
|
actual_key, regexp = accrual.SearchTerm.parse(search)
|
|
assert actual_key == key
|
|
check_link_regexp(regexp, document)
|
|
|
|
@pytest.mark.parametrize('search,unmatched', [
|
|
('1234.pdf', '1234_pdf'),
|
|
])
|
|
def test_search_term_parse_regexp_escaping(search, unmatched):
|
|
_, regexp = accrual.SearchTerm.parse(search)
|
|
assert re.search(regexp, unmatched) is None
|
|
|
|
@pytest.mark.parametrize('search_terms,expect_count,check_func', [
|
|
([], ACCRUALS_COUNT, lambda post: post.account.is_under(
|
|
'Assets:Receivable:', 'Liabilities:Payable:',
|
|
)),
|
|
([('invoice', '^rt:505/5050$')], 2, lambda post: post.meta['entity'] == 'DonorA'),
|
|
([('rt-id', r'^rt:\D+515$')], 1, lambda post: post.meta['entity'] == 'DonorB'),
|
|
([('entity', '^Lawyer$')], 3, lambda post: post.meta['rt-id'] == 'rt:510'),
|
|
([('entity', '^Lawyer$'), ('contract', '^rt:510/')], 2,
|
|
lambda post: post.meta['invoice'].startswith('rt:510/')),
|
|
([('rt-id', '^rt:510$'), ('approval', '.')], 0, lambda post: False),
|
|
])
|
|
def test_filter_search(accrual_postings, search_terms, expect_count, check_func):
|
|
actual = list(accrual.filter_search(accrual_postings, search_terms))
|
|
if expect_count < ACCRUALS_COUNT:
|
|
assert ACCRUALS_COUNT > len(actual) >= expect_count
|
|
else:
|
|
assert len(actual) == ACCRUALS_COUNT
|
|
for post in actual:
|
|
assert check_func(post)
|
|
|
|
@pytest.mark.parametrize('arg,expected', [
|
|
('balance', accrual.balance_report),
|
|
('outgoing', accrual.outgoing_report),
|
|
('bal', accrual.balance_report),
|
|
('out', accrual.outgoing_report),
|
|
('outgoings', accrual.outgoing_report),
|
|
])
|
|
def test_report_type_by_name(arg, expected):
|
|
assert accrual.ReportType.by_name(arg.lower()) is expected
|
|
assert accrual.ReportType.by_name(arg.title()) is expected
|
|
assert accrual.ReportType.by_name(arg.upper()) is expected
|
|
|
|
@pytest.mark.parametrize('arg', [
|
|
'unknown',
|
|
'blance',
|
|
'outgong',
|
|
])
|
|
def test_report_type_by_unknown_name(arg):
|
|
# Raising ValueError helps argparse generate good messages.
|
|
with pytest.raises(ValueError):
|
|
accrual.ReportType.by_name(arg)
|
|
|
|
@pytest.mark.parametrize('meta_key,account', testutil.combine_values(
|
|
CONSISTENT_METADATA,
|
|
ACCOUNTS,
|
|
))
|
|
def test_consistency_check_when_consistent(meta_key, account):
|
|
invoice = f'test-{meta_key}-invoice'
|
|
meta = {
|
|
'invoice': invoice,
|
|
meta_key: f'test-{meta_key}-value',
|
|
}
|
|
txn = testutil.Transaction(postings=[
|
|
(account, 100, meta),
|
|
(account, -100, meta),
|
|
])
|
|
related = core.RelatedPostings(data.Posting.from_txn(txn))
|
|
assert not list(accrual.consistency_check({invoice: related}))
|
|
|
|
@pytest.mark.parametrize('meta_key,account', testutil.combine_values(
|
|
['approval', 'fx-rate', 'statement'],
|
|
ACCOUNTS,
|
|
))
|
|
def test_consistency_check_ignored_metadata(meta_key, account):
|
|
invoice = f'test-{meta_key}-invoice'
|
|
txn = testutil.Transaction(postings=[
|
|
(account, 100, {'invoice': invoice, meta_key: 'credit'}),
|
|
(account, -100, {'invoice': invoice, meta_key: 'debit'}),
|
|
])
|
|
related = core.RelatedPostings(data.Posting.from_txn(txn))
|
|
assert not list(accrual.consistency_check({invoice: related}))
|
|
|
|
@pytest.mark.parametrize('meta_key,account', testutil.combine_values(
|
|
CONSISTENT_METADATA,
|
|
ACCOUNTS,
|
|
))
|
|
def test_consistency_check_when_inconsistent(meta_key, account):
|
|
invoice = f'test-{meta_key}-invoice'
|
|
txn = testutil.Transaction(postings=[
|
|
(account, 100, {'invoice': invoice, meta_key: 'credit', 'lineno': 1}),
|
|
(account, -100, {'invoice': invoice, meta_key: 'debit', 'lineno': 2}),
|
|
])
|
|
related = core.RelatedPostings(data.Posting.from_txn(txn))
|
|
errors = list(accrual.consistency_check({invoice: related}))
|
|
for exp_lineno, (actual, exp_msg) in enumerate(itertools.zip_longest(errors, [
|
|
f'inconsistent {meta_key} for invoice {invoice}: credit',
|
|
f'inconsistent {meta_key} for invoice {invoice}: debit',
|
|
]), 1):
|
|
assert actual.message == exp_msg
|
|
assert actual.entry is txn
|
|
assert actual.source.get('lineno') == exp_lineno
|
|
|
|
def check_output(output, expect_patterns):
|
|
output.seek(0)
|
|
testutil.check_lines_match(iter(output), expect_patterns)
|
|
|
|
@pytest.mark.parametrize('invoice,expected', [
|
|
('rt:505/5050', "Zero balance outstanding since 2020-05-05"),
|
|
('rt:510/5100', "Zero balance outstanding since 2020-05-10"),
|
|
('rt:510/6100', "-280.00 USD outstanding since 2020-06-10"),
|
|
('rt://ticket/515/attachments/5150', "1500.00 USD outstanding since 2020-05-15",),
|
|
])
|
|
def test_balance_report(accrual_postings, invoice, expected):
|
|
related = core.RelatedPostings(
|
|
post for post in accrual_postings
|
|
if post.meta.get('invoice') == invoice
|
|
and post.account.is_under('Assets:Receivable', 'Liabilities:Payable')
|
|
)
|
|
output = io.StringIO()
|
|
accrual.balance_report({invoice: related}, output)
|
|
check_output(output, [invoice, expected])
|
|
|
|
def test_outgoing_report(accrual_postings):
|
|
invoice = 'rt:510/6100'
|
|
related = core.RelatedPostings(
|
|
post for post in accrual_postings
|
|
if post.meta.get('invoice') == invoice
|
|
and post.account.is_under('Assets:Receivable', 'Liabilities:Payable')
|
|
)
|
|
output = io.StringIO()
|
|
errors = io.StringIO()
|
|
rt_client = RTClient()
|
|
rt_cache = rtutil.RT(rt_client)
|
|
accrual.outgoing_report({invoice: related}, output, errors, rt_client, rt_cache)
|
|
assert not errors.getvalue()
|
|
rt_url = rt_client.DEFAULT_URL[:-9]
|
|
rt_id_url = rf'\b{re.escape(f"{rt_url}Ticket/Display.html?id=510")}\b'
|
|
contract_url = rf'\b{re.escape(f"{rt_url}Ticket/Attachment/4000/4000/contract.pdf")}\b'
|
|
print(output.getvalue())
|
|
check_output(output, [
|
|
r'^PAYMENT FOR APPROVAL:$',
|
|
r'^REQUESTOR: Mx\. 510 <mx510@example\.org>$',
|
|
r'^TOTAL TO PAY: 280\.00 USD$',
|
|
fr'^AGREEMENT: {contract_url}',
|
|
r'^PAYMENT TO: Hon\. Mx\. 510$',
|
|
r'^PAYMENT METHOD: payment method 510$',
|
|
r'^BEANCOUNT ENTRIES:$',
|
|
# For each transaction, check for the date line, a metadata, and the
|
|
# Expenses posting.
|
|
r'^\s*2020-06-10\s',
|
|
fr'^\s+rt-id: "{rt_id_url}"$',
|
|
r'^\s+Expenses:Services:Legal\s+220\.00 USD$',
|
|
r'^\s*2020-06-12\s',
|
|
fr'^\s+contract: "{contract_url}"$',
|
|
r'^\s+Expenses:FilingFees\s+60\.00 USD$',
|
|
])
|
|
|
|
def test_outgoing_report_custom_field_fallbacks(accrual_postings):
|
|
invoice = 'rt:510/6100'
|
|
related = core.RelatedPostings(
|
|
post for post in accrual_postings
|
|
if post.meta.get('invoice') == invoice
|
|
and post.account.is_under('Assets:Receivable', 'Liabilities:Payable')
|
|
)
|
|
output = io.StringIO()
|
|
errors = io.StringIO()
|
|
rt_client = RTClient(want_cfs=False)
|
|
rt_cache = rtutil.RT(rt_client)
|
|
accrual.outgoing_report({invoice: related}, output, errors, rt_client, rt_cache)
|
|
assert not errors.getvalue()
|
|
check_output(output, [
|
|
r'^PAYMENT FOR APPROVAL:$',
|
|
r'^REQUESTOR: <mx510@example\.org>$',
|
|
r'^PAYMENT TO:\s*$',
|
|
r'^PAYMENT METHOD:\s*$',
|
|
])
|
|
|
|
def run_main(arglist, config=None):
|
|
if config is None:
|
|
config = testutil.TestConfig(
|
|
books_path=testutil.test_path('books/accruals.beancount'),
|
|
rt_client=RTClient(),
|
|
)
|
|
output = io.StringIO()
|
|
errors = io.StringIO()
|
|
retcode = accrual.main(arglist, output, errors, config)
|
|
return retcode, output, errors
|
|
|
|
def check_main_fails(arglist, config, error_flags, error_patterns):
|
|
retcode, output, errors = run_main(arglist, config)
|
|
assert retcode > 16
|
|
assert (retcode - 16) & error_flags
|
|
check_output(errors, error_patterns)
|
|
assert not output.getvalue()
|
|
|
|
@pytest.mark.parametrize('arglist', [
|
|
['--report-type=balance'],
|
|
['--report-type=outgoing'],
|
|
['entity=EarlyBird'],
|
|
])
|
|
def test_output_excludes_payments(arglist):
|
|
retcode, output, errors = run_main(arglist)
|
|
assert not errors.getvalue()
|
|
assert retcode == 0
|
|
output.seek(0)
|
|
for line in output:
|
|
assert not re.match(r'\brt:4\d\b', line)
|
|
|
|
@pytest.mark.parametrize('arglist,expect_invoice', [
|
|
(['40'], 'rt:40/400'),
|
|
(['44/440'], 'rt:44/440'),
|
|
])
|
|
def test_output_payments_when_only_match(arglist, expect_invoice):
|
|
retcode, output, errors = run_main(arglist)
|
|
assert not errors.getvalue()
|
|
assert retcode == 0
|
|
check_output(output, [
|
|
rf'^{re.escape(expect_invoice)}:$',
|
|
r' outstanding since ',
|
|
])
|
|
|
|
@pytest.mark.parametrize('arglist', [
|
|
['--report-type=outgoing'],
|
|
['510'],
|
|
['510/6100'],
|
|
['entity=Lawyer'],
|
|
])
|
|
def test_main_outgoing_report(arglist):
|
|
retcode, output, errors = run_main(arglist)
|
|
assert not errors.getvalue()
|
|
assert retcode == 0
|
|
rt_url = RTClient.DEFAULT_URL[:-9]
|
|
rt_id_url = re.escape(f'<{rt_url}Ticket/Display.html?id=510>')
|
|
contract_url = re.escape(f'<{rt_url}Ticket/Attachment/4000/4000/contract.pdf>')
|
|
check_output(output, [
|
|
r'^REQUESTOR: Mx\. 510 <mx510@example\.org>$',
|
|
r'^TOTAL TO PAY: 280\.00 USD$',
|
|
r'^\s*2020-06-12\s',
|
|
r'^\s+Expenses:FilingFees\s+60\.00 USD$',
|
|
])
|
|
|
|
@pytest.mark.parametrize('arglist', [
|
|
['-t', 'balance'],
|
|
['515'],
|
|
['515/5150'],
|
|
['entity=DonorB'],
|
|
])
|
|
def test_main_balance_report(arglist):
|
|
retcode, output, errors = run_main(arglist)
|
|
assert not errors.getvalue()
|
|
assert retcode == 0
|
|
check_output(output, [
|
|
r'\brt://ticket/515/attachments/5150:$',
|
|
r'^\s+1500\.00 USD outstanding since 2020-05-15$',
|
|
])
|
|
|
|
def test_main_no_books():
|
|
check_main_fails([], testutil.TestConfig(), 1 | 8, [
|
|
r':1: +no books to load in configuration\b',
|
|
])
|
|
|
|
@pytest.mark.parametrize('arglist', [
|
|
['499'],
|
|
['505/99999'],
|
|
['entity=NonExistent'],
|
|
])
|
|
def test_main_no_matches(arglist):
|
|
check_main_fails(arglist, None, 8, [
|
|
r'^warning: no matching entries found to report$',
|
|
])
|
|
|
|
def test_main_no_rt():
|
|
config = testutil.TestConfig(
|
|
books_path=testutil.test_path('books/accruals.beancount'),
|
|
)
|
|
check_main_fails(['-t', 'out'], config, 4, [
|
|
r'^error: unable to generate outgoing report: RT client is required\b',
|
|
])
|