Files
@ 84d8adb7f664
Branch filter:
Location: NPO-Accounting/conservancy_beancount/tests/test_plugin.py
84d8adb7f664
3.3 KiB
text/x-python
plugin: Move hook initialization from HookRegistry to run().
Makes more sense here so run can report errors in hook configuration.
Makes more sense here so run can report errors in hook configuration.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | """Test main plugin"""
# 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 pytest
from . import testutil
from conservancy_beancount import beancount_types, plugin
HOOK_REGISTRY = plugin.HookRegistry()
class TransactionHook:
DIRECTIVE = beancount_types.Transaction
HOOK_GROUPS = frozenset()
def run(self, txn):
assert False, "something called base class run method"
@HOOK_REGISTRY.add_hook
class TransactionError(TransactionHook):
HOOK_GROUPS = frozenset(['configured'])
def run(self, txn):
return ['txn:{}'.format(id(txn))]
@HOOK_REGISTRY.add_hook
class PostingError(TransactionHook):
HOOK_GROUPS = frozenset(['configured', 'posting'])
def run(self, txn):
return ['post:{}'.format(id(post)) for post in txn.postings]
@pytest.fixture
def config_map():
return {}
@pytest.fixture
def easy_entries():
return [
testutil.Transaction(postings=[
('Income:Donations', -25),
('Assets:Cash', 25),
]),
testutil.Transaction(postings=[
('Expenses:General', 10),
('Liabilites:CreditCard', -10),
]),
]
def map_errors(errors):
retval = {}
for errkey in errors:
key, _, errid = errkey.partition(':')
retval.setdefault(key, set()).add(errid)
return retval
@pytest.mark.parametrize('group_str,expected', [
(None, [TransactionError, PostingError]),
('', [TransactionError, PostingError]),
('all', [TransactionError, PostingError]),
('Transaction', [TransactionError, PostingError]),
('-posting', [TransactionError]),
('-configured posting', [PostingError]),
('configured -posting', [TransactionError]),
])
def test_registry_group_by_directive(group_str, expected):
args = () if group_str is None else (group_str,)
actual = {hook for _, hook in HOOK_REGISTRY.group_by_directive(*args)}
assert actual.issuperset(expected)
if len(expected) == 1:
assert not (TransactionError in actual and PostingError in actual)
def test_registry_unknown_group_name():
with pytest.raises(ValueError):
next(HOOK_REGISTRY.group_by_directive('UnKnownTestGroup'))
def test_run_with_multiple_hooks(easy_entries, config_map):
out_entries, errors = plugin.run(easy_entries, config_map, '', HOOK_REGISTRY)
assert len(out_entries) == 2
errmap = map_errors(errors)
assert len(errmap.get('txn', '')) == 2
assert len(errmap.get('post', '')) == 4
def test_run_with_one_hook(easy_entries, config_map):
out_entries, errors = plugin.run(easy_entries, config_map, 'posting', HOOK_REGISTRY)
assert len(out_entries) == 2
errmap = map_errors(errors)
assert len(errmap.get('txn', '')) == 0
assert len(errmap.get('post', '')) == 4
|