conservancy_beancount/tests/test_plugin_run.py

"""Test main plugin run loop"""
# 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 plugin, _typing

CONFIG_MAP = {}
HOOK_REGISTRY = plugin.HookRegistry()

@HOOK_REGISTRY.add_hook
class TransactionCounter:
    DIRECTIVE = _typing.Transaction
    HOOK_GROUPS = frozenset()

    def run(self, txn):
        return ['txn:{}'.format(id(txn))]


@HOOK_REGISTRY.add_hook
class PostingCounter(TransactionCounter):
    DIRECTIVE = _typing.Transaction
    HOOK_GROUPS = frozenset(['posting'])

    def run(self, txn):
        return ['post:{}'.format(id(post)) for post in txn.postings]


def map_errors(errors):
    retval = {}
    for errkey in errors:
        key, _, errid = errkey.partition(':')
        retval.setdefault(key, set()).add(errid)
    return retval

def test_with_multiple_hooks():
    in_entries = [
        testutil.Transaction(postings=[
            ('Income:Donations', -25),
            ('Assets:Cash', 25),
        ]),
        testutil.Transaction(postings=[
            ('Expenses:General', 10),
            ('Liabilites:CreditCard', -10),
        ]),
    ]
    out_entries, errors = plugin.run(in_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_with_posting_hooks_only():
    in_entries = [
        testutil.Transaction(postings=[
            ('Income:Donations', -25),
            ('Assets:Cash', 25),
        ]),
        testutil.Transaction(postings=[
            ('Expenses:General', 10),
            ('Liabilites:CreditCard', -10),
        ]),
    ]
    out_entries, errors = plugin.run(in_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