ad81c45f2a
This module basically reimplements the old payment-report+income-report, in a single tool (after setting aside some of the checks that have moved to the plugin). The aging report can be implemented here too when we need that.
410 lines
14 KiB
Python
410 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 = {
|
|
'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('invoice,expected', [
|
|
# No outstanding balance
|
|
('rt:505/5050', accrual.balance_report),
|
|
('rt:510/5100', accrual.balance_report),
|
|
# Outstanding receivable
|
|
('rt://ticket/515/attachments/5150', accrual.balance_report),
|
|
# Outstanding payable
|
|
('rt:510/6100', accrual.outgoing_report),
|
|
])
|
|
def test_default_report_type(accrual_postings, invoice, expected):
|
|
related = core.RelatedPostings()
|
|
for post in accrual_postings:
|
|
if (post.account.is_under('Assets:Receivable', 'Liabilities:Payable')
|
|
and post.meta.get('invoice') == invoice):
|
|
related.add(post)
|
|
groups = {invoice: related}
|
|
report_type, report_groups = accrual.ReportType.default_for(groups)
|
|
assert report_type is expected
|
|
assert report_groups == groups
|
|
|
|
@pytest.mark.parametrize('entity,exp_type,exp_invoices', [
|
|
('^Lawyer$', accrual.outgoing_report, {'rt:510/6100'}),
|
|
('^Donor[AB]$', accrual.balance_report, {'rt://ticket/515/attachments/5150'}),
|
|
('^(Lawyer|DonorB)$', accrual.balance_report,
|
|
{'rt:510/6100', 'rt://ticket/515/attachments/5150'}),
|
|
])
|
|
def test_default_report_type_multi_invoices(accrual_postings, entity, exp_type, exp_invoices):
|
|
groups = core.RelatedPostings.group_by_meta((
|
|
post for post in accrual_postings
|
|
if post.account.is_under('Assets:Receivable', 'Liabilities:Payable')
|
|
and re.match(entity, post.meta.get('entity', ''))
|
|
), 'invoice')
|
|
report_type, report_groups = accrual.ReportType.default_for(groups)
|
|
assert report_type is exp_type
|
|
assert set(report_groups.keys()) == exp_invoices
|
|
assert all(len(related) > 0 for related in report_groups.values())
|
|
|
|
@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)
|
|
for pattern in expect_patterns:
|
|
assert any(re.search(pattern, line) for line in output), \
|
|
f"{pattern!r} not found in output"
|
|
|
|
@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 = 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'^PAYMENT FOR APPROVAL:$',
|
|
r'^REQUESTOR: Mx\. 510 <mx510@example\.org>$',
|
|
r'^TOTAL TO PAY: 280\.00 USD$',
|
|
fr'^AGREEMENT: {contract_url}',
|
|
r'^PAYMENT TO: 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 run_main(arglist, config):
|
|
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=outgoing'],
|
|
['510'],
|
|
['510/6100'],
|
|
['entity=Lawyer'],
|
|
])
|
|
def test_main_outgoing_report(arglist):
|
|
rt_client = RTClient()
|
|
config = testutil.TestConfig(
|
|
books_path=testutil.test_path('books/accruals.beancount'),
|
|
rt_client=rt_client,
|
|
)
|
|
retcode, output, errors = run_main(arglist, config)
|
|
assert not errors.getvalue()
|
|
assert retcode == 0
|
|
rt_url = rt_client.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):
|
|
config = testutil.TestConfig(
|
|
books_path=testutil.test_path('books/accruals.beancount'),
|
|
)
|
|
retcode, output, errors = run_main(arglist, config)
|
|
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):
|
|
config = testutil.TestConfig(
|
|
books_path=testutil.test_path('books/accruals.beancount'),
|
|
)
|
|
check_main_fails(arglist, config, 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',
|
|
])
|