072937eff5
The old loading strategy didn't load options, which yielded some spurious errors. It also created awkward duplication of plugin information in the code as well as the books. Implement a new loading strategy that works by reading one of the "main files" under the books/ subdirectory and includes entries for additional FYs beyond that. This is still not ideal in a lot of ways. In particular, Beancount can't cache any results, causing any load to be slower than it theoretically could be. I expect more commits to follow. But some of them might require restructuring the books, and that should happen separately.
408 lines
14 KiB
Python
408 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)
|
|
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 = 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',
|
|
])
|