# 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/>.

from typing import (

    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,

        )

#

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

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

    def add_hook(self, hook_cls: Type[Hook]) -> Type[Hook]:

        self.group_name_map['all'].add(hook_cls)

        self.group_name_map[hook_cls.DIRECTIVE.__name__].add(hook_cls)

        for key in hook_cls.HOOK_GROUPS:

            self.group_name_map.setdefault(key, set()).add(hook_cls)

        return hook_cls  # to allow use as a decorator

    # This method is too dynamic to typecheck.

    def import_hooks(self, mod_name, *hook_names, package=__module__):  # type:ignore

        module = importlib.import_module(mod_name, package)

        for hook_name in hook_names:

            self.add_hook(getattr(module, hook_name))

    def group_by_directive(self, config_str: str='') -> Iterable[Tuple[HookName, Type[Hook]]]:

        config_str = config_str.strip()

        available_hooks: Set[Type[Hook]] = set()

        for token in config_str.split():

            if token.startswith('-'):

                update_available = available_hooks.difference_update

                key = token[1:]

            else:

                update_available = available_hooks.update

                key = token

            try:

                update_set = self.group_name_map[key]

            except KeyError:

                raise ValueError("configuration refers to unknown hooks {!r}".format(key)) from None

            else:

                update_available(update_set)

        for directive in ALL_DIRECTIVES:

            key = directive.__name__

            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

"""Base classes for plugin checks"""

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

    Directive,

    MetaKey,

    MetaValue,

    MetaValueEnum,

    Transaction,

    Type,

)

### CONSTANTS

# 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

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

CT = TypeVar('CT', bound=LessComparable)

class _GenericRange(Generic[CT]):

    """Convenience class to check whether a value is within a range.

    `foo in generic_range` is equivalent to `start <= foo < stop`.

    Since we have multiple user-configurable ranges, having the check

    encapsulated in an object helps implement the check consistently, and

    makes it easier for subclasses to override.

    """

    def __init__(self, start: CT, stop: CT) -> None:

        self.start = start

        self.stop = stop

# 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.plugin import meta_expense_allocation

VALID_VALUES = {

    'program': 'program',

    'administration': 'administration',

    '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}),

    ])

    errors = list(hook.run(txn))

    assert errors

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

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

def test_valid_values_on_transactions(hook, src_value, set_value):

    txn = testutil.Transaction(**{TEST_KEY: src_value}, postings=[

        ('Assets:Cash', -25),

        ('Expenses:General', 25),

    ])

    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)

# 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.plugin import meta_income_type

VALID_VALUES = {

    'Donations': 'Donations',

    'Payable-Derecognition': 'Payable-Derecognition',

    '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}),

    ])

    errors = list(hook.run(txn))

    assert errors

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

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

def test_valid_values_on_transactions(hook, src_value, set_value):

    txn = testutil.Transaction(**{TEST_KEY: src_value}, postings=[

        ('Assets:Cash', 25),

        ('Income:Other', -25),

    ])

    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)

VALID_VALUES = {

    '1099': '1099',

    'Accountant-Advises-No-1099': 'Accountant-Advises-No-1099',

    'Bank-Transfer': 'Bank-Transfer',

    'Foreign-Corporation': 'Foreign-Corporation',

    'Foreign-Individual-Contractor': 'Foreign-Individual-Contractor',

    'Fraud': 'Fraud',

    'HSA-Contribution': 'HSA-Contribution',

    'Loan': 'Loan',

    'Payroll': 'Payroll',

    'Refund': 'Refund',

    'Reimbursement': 'Reimbursement',

    'Retirement-Pretax': 'Retirement-Pretax',

    'Tax-Payment': 'Tax-Payment',

    'USA-501c3': 'USA-501c3',

    '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}),

    ])

    errors = list(hook.run(txn))

    assert errors

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

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

def test_valid_values_on_transactions(hook, src_value, set_value):

    txn = testutil.Transaction(**{TEST_KEY: src_value}, postings=[

        ('Accrued:AccountsPayable', 25),

        ('Assets:Cash', -25),

    ])

    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)

"""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

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)

    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)

"""Mock Beancount objects for testing"""

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

            }

            assert actual == expected

def parse_date(s, fmt='%Y-%m-%d'):

    return datetime.datetime.strptime(s, fmt).date()

def Posting(account, number,

            currency='USD', cost=None, price=None, flag=None,

            **meta):

    if not meta:

        meta = None

    return bc_data.Posting(

        account,

        bc_amount.Amount(Decimal(number), currency),

        cost,

        price,

        flag,

        self.date = date

        self.flag = flag

        self.payee = payee

        self.narration = narration

        self.tags = set(tags or '')

        self.links = set(links or '')

        self.postings = []

        self.meta = {

            'filename': '<test>',

            'lineno': 0,

        }

        self.meta.update(meta)

        for posting in postings:

            self.add_posting(*posting)

    def add_posting(self, arg, *args, **kwargs):

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