Iterable,

)

class Error(Exception):

    def __init__(self, message, entry, source=None):

        self.message = message

        self.entry = entry

        self.source = entry.meta if source is None else source

    def __repr__(self):

        return "{clsname}<{source[filename]}:{source[lineno]}: {message}>".format(

            clsname=type(self).__name__,

            message=self.message,

            source=self.source,

        )

Iter = Iterable[Error]

class InvalidMetadataError(Error):

    def __init__(self, txn, post, key, value=None, source=None):

        if value is None:

            msg_fmt = "{post.account} missing {key}"

        else:

            msg_fmt = "{post.account} has invalid {key}: {value}"

        super().__init__(

            msg_fmt.format(post=post, key=key, value=value),

            txn,

            source,

        )

import beancount.core.data as bc_data

from typing import (

    AbstractSet,

    Any,

    Dict,

    Iterable,

    List,

    Set,

    Tuple,

    Type,

)

from ..beancount_types import (

    ALL_DIRECTIVES,

    Directive,

)

from .core import (

    Hook,

    HookName,

)

from ..errors import (

    Error,

)

__plugins__ = ['run']

class HookRegistry:

    def __init__(self) -> None:

        self.group_name_map: Dict[HookName, Set[Type[Hook]]] = {

            t.__name__: set() for t in ALL_DIRECTIVES

        }

        self.group_name_map['all'] = set()

            for hook in self.group_name_map[key] & available_hooks:

                yield key, hook

HOOK_REGISTRY = HookRegistry()

HOOK_REGISTRY.import_hooks('.meta_expense_allocation', 'MetaExpenseAllocation')

HOOK_REGISTRY.import_hooks('.meta_tax_implication', 'MetaTaxImplication')

def run(

        entries: List[Directive],

        options_map: Dict[str, Any],

        config: str='',

        hook_registry: HookRegistry=HOOK_REGISTRY,

) -> Tuple[List[Directive], List[Error]]:

    errors: List[Error] = []

    hooks: Dict[HookName, List[Hook]] = {}

    for key, hook_type in hook_registry.group_by_directive(config):

    for entry in entries:

        entry_type = type(entry).__name__

        for hook in hooks[entry_type]:

            errors.extend(hook.run(entry))

    return entries, errors

# 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 abc

import datetime

import re

from .. import data

from .. import errors as errormod

from typing import (

    Any,

    Dict,

    FrozenSet,

    Generic,

    Iterable,

    Iterator,

    Mapping,

    Optional,

    TypeVar,

)

from ..beancount_types import (

    Account,

# I expect these will become configurable in the future, which is why I'm

# keeping them outside of a class, but for now constants will do.

DEFAULT_START_DATE: datetime.date = datetime.date(2020, 3, 1)

# The default stop date leaves a little room after so it's easy to test

# dates past the far end of the range.

DEFAULT_STOP_DATE: datetime.date = datetime.date(datetime.MAXYEAR, 1, 1)

### TYPE DEFINITIONS

HookName = str

Entry = TypeVar('Entry', bound=Directive)

class Hook(Generic[Entry], metaclass=abc.ABCMeta):

    DIRECTIVE: Type[Directive]

    HOOK_GROUPS: FrozenSet[HookName] = frozenset()

    @abc.abstractmethod

    def run(self, entry: Entry) -> errormod.Iter: ...

    def __init_subclass__(cls):

        cls.DIRECTIVE = cls.__orig_bases__[0].__args__[0]

TransactionHook = Hook[Transaction]

### HELPER CLASSES

class LessComparable(metaclass=abc.ABCMeta):

    @abc.abstractmethod

    def __le__(self, other: Any) -> bool: ...

    @abc.abstractmethod

    'fundraising': 'fundraising',

    'admin': 'administration',

}

INVALID_VALUES = {

    'invalid',

    'porgram',

    'adimn',

    'fundrasing',

    '',

}

TEST_KEY = 'expense-allocation'

@pytest.fixture(scope='module')

def hook():

@pytest.mark.parametrize('src_value,set_value', VALID_VALUES.items())

def test_valid_values_on_postings(hook, src_value, set_value):

    txn = testutil.Transaction(postings=[

        ('Assets:Cash', -25),

        ('Expenses:General', 25, {TEST_KEY: src_value}),

    ])

    errors = list(hook.run(txn))

    assert not errors

    testutil.check_post_meta(txn, None, {TEST_KEY: set_value})

@pytest.mark.parametrize('src_value', INVALID_VALUES)

def test_invalid_values_on_postings(hook, src_value):

    txn = testutil.Transaction(postings=[

        ('Assets:Cash', -25),

        ('Expenses:General', 25, {TEST_KEY: src_value}),

    'RBI': 'RBI',

    'UBTI': 'UBTI',

}

INVALID_VALUES = {

    'Dontion',

    'Payble-Derecognitoin',

    'RIB',

    'UTBI',

    '',

}

TEST_KEY = 'income-type'

@pytest.fixture(scope='module')

def hook():

@pytest.mark.parametrize('src_value,set_value', VALID_VALUES.items())

def test_valid_values_on_postings(hook, src_value, set_value):

    txn = testutil.Transaction(postings=[

        ('Assets:Cash', 25),

        ('Income:Other', -25, {TEST_KEY: src_value}),

    ])

    errors = list(hook.run(txn))

    assert not errors

    testutil.check_post_meta(txn, None, {TEST_KEY: set_value})

@pytest.mark.parametrize('src_value', INVALID_VALUES)

def test_invalid_values_on_postings(hook, src_value):

    txn = testutil.Transaction(postings=[

        ('Assets:Cash', 25),

        ('Income:Other', -25, {TEST_KEY: src_value}),

    'USA-Corporation': 'USA-Corporation',

    'USA-LLC-No-1099': 'USA-LLC-No-1099',

    'W2': 'W2',

}

INVALID_VALUES = {

    '199',

    'W3',

    'Payrol',

    '',

}

TEST_KEY = 'tax-implication'

@pytest.fixture(scope='module')

def hook():

@pytest.mark.parametrize('src_value,set_value', VALID_VALUES.items())

def test_valid_values_on_postings(hook, src_value, set_value):

    txn = testutil.Transaction(postings=[

        ('Accrued:AccountsPayable', 25),

        ('Assets:Cash', -25, {TEST_KEY: src_value}),

    ])

    errors = list(hook.run(txn))

    assert not errors

    testutil.check_post_meta(txn, None, {TEST_KEY: set_value})

@pytest.mark.parametrize('src_value', INVALID_VALUES)

def test_invalid_values_on_postings(hook, src_value):

    txn = testutil.Transaction(postings=[

        ('Accrued:AccountsPayable', 25),

        ('Assets:Cash', -25, {TEST_KEY: src_value}),

# 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

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):

@HOOK_REGISTRY.add_hook

class PostingError(TransactionHook):

    HOOK_GROUPS = frozenset(['configured', 'posting'])

    def run(self, txn):

@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 = {}

        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)

# (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 datetime

import beancount.core.amount as bc_amount

import beancount.core.data as bc_data

from decimal import Decimal

EXTREME_FUTURE_DATE = datetime.date(datetime.MAXYEAR, 12, 30)

FUTURE_DATE = datetime.date.today() + datetime.timedelta(days=365 * 99)

FY_START_DATE = datetime.date(2020, 3, 1)

FY_MID_DATE = datetime.date(2020, 9, 1)

PAST_DATE = datetime.date(2000, 1, 1)

def check_post_meta(txn, *expected_meta, default=None):

    assert len(txn.postings) == len(expected_meta)

    for post, expected in zip(txn.postings, expected_meta):

        if not expected:

            assert not post.meta

        else:

            actual = None if post.meta is None else {

                key: post.meta.get(key, default) for key in expected

            }

        """Add a posting to this transaction. Use any of these forms:

           txn.add_posting(account, number, …, kwarg=value, …)

           txn.add_posting(account, number, …, posting_kwargs_dict)

           txn.add_posting(posting_object)

        """

        if kwargs:

            posting = Posting(arg, *args, **kwargs)

        elif args:

            if isinstance(args[-1], dict):

                kwargs = args[-1]

                args = args[:-1]

            posting = Posting(arg, *args, **kwargs)

        else:

            posting = arg

        self.postings.append(posting)