Changeset - fe3560b748fa
[Not reviewed]
0 3 2
Brett Smith - 4 years ago 2021-02-11 18:38:11
brettcsmith@brettcsmith.org
meta_tax_reporting: New plugin validation.
5 files changed with 247 insertions and 7 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/data.py
Show inline comments
...
 
@@ -45,64 +45,65 @@ from .beancount_types import (
 
    Meta,
 
    MetaKey,
 
    MetaValue,
 
    Open,
 
    OptionsMap,
 
    Posting as BasePosting,
 
    Transaction,
 
)
 

	
 
DecimalCompat = Union[decimal.Decimal, int]
 

	
 
EQUITY_ACCOUNTS = frozenset([
 
    'Equity',
 
    'Expenses',
 
    'Income',
 
])
 
FUND_ACCOUNTS = EQUITY_ACCOUNTS | frozenset([
 
    'Assets:Prepaid',
 
    'Assets:Receivable',
 
    'Liabilities:Payable',
 
    'Liabilities:UnearnedIncome',
 
])
 
LINK_METADATA = frozenset([
 
    'approval',
 
    'bank-statement',
 
    'check',
 
    'contract',
 
    'invoice',
 
    'purchase-order',
 
    'receipt',
 
    'rt-id',
 
    'statement',
 
    'tax-reporting',
 
    'tax-statement',
 
])
 

	
 
class AccountMeta(MutableMapping[MetaKey, MetaValue]):
 
    """Access account metadata
 

	
 
    This class provides a consistent interface to all the metadata provided by
 
    Beancount's ``open`` and ``close`` directives: open and close dates,
 
    used currencies, booking method, and the metadata associated with each.
 

	
 
    For convenience, you can use this class as a Mapping to access the ``open``
 
    directive's metadata directly.
 
    """
 
    __slots__ = ('_opening', '_closing')
 

	
 
    def __init__(self, opening: Open, closing: Optional[Close]=None) -> None:
 
        self._opening = opening
 
        self._closing: Optional[Close] = None
 
        if closing is not None:
 
            self.add_closing(closing)
 

	
 
    def add_closing(self, closing: Close) -> None:
 
        if self._closing is not None and self._closing is not closing:
 
            raise ValueError(f"{self.account} already closed by {self._closing!r}")
 
        elif closing.account != self.account:
 
            raise ValueError(f"cannot close {self.account} with {closing.account}")
 
        elif closing.date < self.open_date:
 
            raise ValueError(f"close date {closing.date} predates open date {self.open_date}")
 
        else:
 
            self._closing = closing
 

	
 
    def __iter__(self) -> Iterator[MetaKey]:
conservancy_beancount/plugin/__init__.py
Show inline comments
...
 
@@ -23,76 +23,77 @@ from typing import (
 
)
 
from ..beancount_types import (
 
    ALL_DIRECTIVES,
 
    Directive,
 
    Entries,
 
    Errors,
 
    OptionsMap,
 
)
 
from .. import config as configmod
 
from .core import (
 
    Hook,
 
    HookName,
 
)
 
from ..errors import (
 
    ConfigurationError,
 
    Error,
 
)
 

	
 
__plugins__ = ['run']
 

	
 
logger = logging.getLogger('conservancy_beancount.plugin')
 

	
 
class HookRegistry:
 
    INCLUDED_HOOKS: Dict[str, Optional[List[str]]] = {
 
        '.meta_approval': None,
 
        '.meta_entity': None,
 
        '.meta_expense_type': None,
 
        '.meta_income_type': None,
 
        '.meta_invoice': None,
 
        # Enforcing this hook would be premature as of May 2020.  --brett
 
        # '.meta_payable_documentation': None,
 
        '.meta_paypal_id': ['MetaPayPalID'],
 
        '.meta_project': None,
 
        '.meta_receipt': None,
 
        '.meta_receivable_documentation': None,
 
        '.meta_repo_links': None,
 
        '.meta_rt_links': ['MetaRTLinks'],
 
        '.meta_tax_implication': None,
 
        '.meta_payroll_type': [
 
            'HealthInsuranceHook',
 
            'OtherBenefitsHook',
 
            'SalaryHook',
 
            'TaxHook',
 
        ],
 
        '.meta_project': None,
 
        '.meta_receipt': None,
 
        '.meta_receivable_documentation': None,
 
        '.meta_repo_links': None,
 
        '.meta_rt_links': ['MetaRTLinks'],
 
        '.meta_tax_implication': None,
 
        '.meta_tax_reporting': None,
 
        '.txn_date': ['TransactionDate'],
 
    }
 

	
 
    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
 

	
 
    def import_hooks(self,
 
                     mod_name: str,
 
                     *hook_names: str,
 
                     package: Optional[str]=None,
 
    ) -> None:
 
        if not hook_names:
 
            _, _, hook_name = mod_name.rpartition('.')
 
            hook_names = (hook_name.title().replace('_', ''),)
 
        module = importlib.import_module(mod_name, package)
 
        for hook_name in hook_names:
 
            self.add_hook(getattr(module, hook_name))
 

	
 
    def load_included_hooks(self) -> None:
 
        for mod_name, hook_names in self.INCLUDED_HOOKS.items():
 
            self.import_hooks(mod_name, *(hook_names or []), package=self.__module__)
 

	
conservancy_beancount/plugin/meta_tax_reporting.py
Show inline comments
 
new file 100644
 
"""meta_tax_reporting - Validate tax-reporting metadata links"""
 
# Copyright © 2021  Brett Smith
 
# License: AGPLv3-or-later WITH Beancount-Plugin-Additional-Permission-1.0
 
#
 
# Full copyright and licensing details can be found at toplevel file
 
# LICENSE.txt in the repository.
 

	
 
import datetime
 

	
 
from . import core
 
from .. import config as configmod
 
from .. import data
 
from .. import errors as errormod
 
from ..beancount_types import (
 
    Transaction,
 
)
 

	
 
from .meta_tax_implication import MetaTaxImplication
 
from ..ranges import DateRange
 

	
 
class MetaTaxReporting(core._RequireLinksPostingMetadataHook):
 
    CHECKED_IMPLICATIONS = frozenset(
 
        # We load values through the MetadataEnum to future-proof against
 
        # changes to tax-implication. This ensures that the set contains
 
        # canonical values, or else this code will crash if canonical values
 
        # can't be found.
 
        MetaTaxImplication.VALUES_ENUM[value] for value in [
 
            '1099-MISC-Other',
 
            '1099-NEC',
 
            'Foreign-Grantee',
 
            'Foreign-Individual-Contractor',
 
            'USA-501c3',
 
            'USA-Grantee',
 
        ])
 
    CHECKED_METADATA = ['tax-reporting']
 
    SKIP_FLAGS = '!'
 
    TXN_DATE_RANGE = DateRange(datetime.date(2020, 3, 1), datetime.date.max)
 

	
 
    def __init__(self, config: configmod.Config) -> None:
 
        self._implication_hook = MetaTaxImplication(config)
 
        # Yes, we create our own MetaTaxImplication hook. This is a little
 
        # weird but it does two things for us:
 
        # 1. We can check MetaTaxImplication._run_on_post() as part of our own
 
        # implementation without duplicating the logic.
 
        # 2. We can canonicalize values through the hook. We don't strictly
 
        # need an instance for that, but we have it anyway so doing it this way
 
        # is nicer.
 

	
 
    def _run_on_post(self, txn: Transaction, post: data.Posting) -> bool:
 
        if not self._implication_hook._run_on_post(txn, post):
 
            return False
 
        implication = str(post.meta.get('tax-implication') or '')
 
        normalized = self._implication_hook.VALUES_ENUM.get(implication)
 
        return normalized in self.CHECKED_IMPLICATIONS
setup.py
Show inline comments
 
#!/usr/bin/env python3
 

	
 
from setuptools import setup
 

	
 
setup(
 
    name='conservancy_beancount',
 
    description="Plugin, library, and reports for reading Conservancy's books",
 
    version='1.17.1',
 
    version='1.18.0',
 
    author='Software Freedom Conservancy',
 
    author_email='info@sfconservancy.org',
 
    license='GNU AGPLv3+',
 

	
 
    install_requires=[
 
        'babel>=2.6',  # Debian:python3-babel
 
        'beancount>=2.2',  # Debian:beancount
 
        'GitPython>=2.0',  # Debian:python3-git
 
        # 1.4.1 crashes when trying to save some documents.
 
        'odfpy>=1.4.0,!=1.4.1',  # Debian:python3-odf
 
        'pdfminer.six>=20200101',
 
        'PyYAML>=3.0',  # Debian:python3-yaml
 
        'regex',  # Debian:python3-regex
 
        'rt>=2.0',
 
    ],
 
    setup_requires=[
 
        'pytest-mypy',
 
        'pytest-runner',  # Debian:python3-pytest-runner
 
    ],
 
    tests_require=[
 
        'mypy>=0.770',  # Debian:python3-mypy
 
        'pytest',  # Debian:python3-pytest
 
    ],
 

	
 
    packages=[
 
        'conservancy_beancount',
 
        'conservancy_beancount.pdfforms',
 
        'conservancy_beancount.pdfforms.extract',
 
        'conservancy_beancount.plugin',
 
        'conservancy_beancount.reconcile',
 
        'conservancy_beancount.reports',
 
        'conservancy_beancount.tools',
tests/test_meta_tax_reporting.py
Show inline comments
 
new file 100644
 
"""test_meta_tax_reporting.py - Unit tests for tax-reporting metadata validation"""
 
# Copyright © 2021  Brett Smith
 
# License: AGPLv3-or-later WITH Beancount-Plugin-Additional-Permission-1.0
 
#
 
# Full copyright and licensing details can be found at toplevel file
 
# LICENSE.txt in the repository.
 

	
 
import pytest
 

	
 
from . import testutil
 

	
 
from conservancy_beancount.plugin import meta_tax_reporting
 

	
 
TEST_KEY = 'tax-reporting'
 
IMPLICATION_KEY = 'tax-implication'
 

	
 
REQUIRED_ACCOUNTS = {
 
    'Assets:Checking',
 
    'Assets:Bank:Savings',
 
}
 

	
 
NON_REQUIRED_ACCOUNTS = {
 
    'Assets:Prepaid:Expenses',
 
    'Assets:Receivable:Accounts',
 
    'Liabilities:CreditCard',
 
}
 

	
 
REQUIRED_AMOUNTS = {-50, -500}
 
NON_REQUIRED_AMOUNTS = {-5, 500}
 

	
 
REQUIRED_IMPLICATIONS = {
 
    '1099',
 
    '1099-Misc-Other',
 
    'foreign-grantee',
 
    'Foreign-Individual-Contractor',
 
    'USA-501c3',
 
    'US-Grantee',
 
}
 

	
 
NON_REQUIRED_IMPLICATIONS = {
 
    'Bank-Transfer',
 
    'chargeback',
 
    'Foreign-Corp',
 
    'Loan',
 
    'refund',
 
    'Reimbursement',
 
    'retirement-pretax',
 
    'Tax-Payment',
 
    'us-corp',
 
    'w2',
 
}
 

	
 
@pytest.fixture(scope='module')
 
def hook():
 
    config = testutil.TestConfig(payment_threshold=10)
 
    return meta_tax_reporting.MetaTaxReporting(config)
 

	
 
@pytest.mark.parametrize('account,amount,implication,value', testutil.combine_values(
 
    REQUIRED_ACCOUNTS,
 
    REQUIRED_AMOUNTS,
 
    REQUIRED_IMPLICATIONS,
 
    testutil.LINK_METADATA_STRINGS,
 
))
 
def test_pass_on_txn(hook, account, amount, implication, value):
 
    txn_meta = {
 
        IMPLICATION_KEY: implication,
 
        TEST_KEY: value,
 
    }
 
    txn = testutil.Transaction(**txn_meta, postings=[
 
        (account, amount),
 
        ('Expenses:Other', -amount),
 
    ])
 
    assert not list(hook.run(txn))
 

	
 
@pytest.mark.parametrize('account,amount,implication,value', testutil.combine_values(
 
    REQUIRED_ACCOUNTS,
 
    REQUIRED_AMOUNTS,
 
    REQUIRED_IMPLICATIONS,
 
    testutil.LINK_METADATA_STRINGS,
 
))
 
def test_pass_on_post(hook, account, amount, implication, value):
 
    post_meta = {
 
        IMPLICATION_KEY: implication,
 
        TEST_KEY: value,
 
    }
 
    txn = testutil.Transaction(postings=[
 
        (account, amount, post_meta),
 
        ('Expenses:Other', -amount),
 
    ])
 
    assert not list(hook.run(txn))
 

	
 
@pytest.mark.parametrize('account,amount,implication', testutil.combine_values(
 
    REQUIRED_ACCOUNTS,
 
    REQUIRED_AMOUNTS,
 
    REQUIRED_IMPLICATIONS,
 
))
 
def test_error_when_missing(hook, account, amount, implication):
 
    txn = testutil.Transaction(postings=[
 
        (account, amount, {IMPLICATION_KEY: implication}),
 
        ('Expenses:Other', -amount),
 
    ])
 
    assert list(hook.run(txn))
 

	
 
@pytest.mark.parametrize('account,amount,implication,value', testutil.combine_values(
 
    REQUIRED_ACCOUNTS,
 
    REQUIRED_AMOUNTS,
 
    REQUIRED_IMPLICATIONS,
 
    testutil.NON_LINK_METADATA_STRINGS,
 
))
 
def test_error_when_empty(hook, account, amount, implication, value):
 
    txn_meta = {
 
        IMPLICATION_KEY: implication,
 
        TEST_KEY: value,
 
    }
 
    txn = testutil.Transaction(**txn_meta, postings=[
 
        (account, amount),
 
        ('Expenses:Other', -amount),
 
    ])
 
    assert list(hook.run(txn))
 

	
 
@pytest.mark.parametrize('account,amount,implication,value', testutil.combine_values(
 
    REQUIRED_ACCOUNTS,
 
    REQUIRED_AMOUNTS,
 
    REQUIRED_IMPLICATIONS,
 
    testutil.NON_STRING_METADATA_VALUES,
 
))
 
def test_error_when_wrong_type(hook, account, amount, implication, value):
 
    txn_meta = {
 
        IMPLICATION_KEY: implication,
 
        TEST_KEY: value,
 
    }
 
    txn = testutil.Transaction(**txn_meta, postings=[
 
        (account, amount),
 
        ('Expenses:Other', -amount),
 
    ])
 
    assert list(hook.run(txn))
 

	
 
@pytest.mark.parametrize('account,amount,implication', testutil.combine_values(
 
    NON_REQUIRED_ACCOUNTS,
 
    REQUIRED_AMOUNTS,
 
    REQUIRED_IMPLICATIONS,
 
))
 
def test_skip_by_account(hook, account, amount, implication):
 
    txn = testutil.Transaction(postings=[
 
        (account, amount, {IMPLICATION_KEY: implication}),
 
        ('Expenses:Other', -amount),
 
    ])
 
    assert not list(hook.run(txn))
 

	
 
@pytest.mark.parametrize('account,amount,implication', testutil.combine_values(
 
    REQUIRED_ACCOUNTS,
 
    NON_REQUIRED_AMOUNTS,
 
    REQUIRED_IMPLICATIONS,
 
))
 
def test_skip_by_amount(hook, account, amount, implication):
 
    txn = testutil.Transaction(postings=[
 
        (account, amount, {IMPLICATION_KEY: implication}),
 
        ('Expenses:Other', -amount),
 
    ])
 
    assert not list(hook.run(txn))
 

	
 
@pytest.mark.parametrize('account,amount,implication', testutil.combine_values(
 
    REQUIRED_ACCOUNTS,
 
    REQUIRED_AMOUNTS,
 
    NON_REQUIRED_IMPLICATIONS,
 
))
 
def test_skip_by_implication(hook, account, amount, implication):
 
    txn = testutil.Transaction(postings=[
 
        (account, amount, {IMPLICATION_KEY: implication}),
 
        ('Expenses:Other', -amount),
 
    ])
 
    assert not list(hook.run(txn))
 

	
 
@pytest.mark.parametrize('account,amount,implication', testutil.combine_values(
 
    REQUIRED_ACCOUNTS,
 
    REQUIRED_AMOUNTS,
 
    REQUIRED_IMPLICATIONS,
 
))
 
def test_skip_by_flag(hook, account, amount, implication):
 
    txn = testutil.Transaction(flag='!', postings=[
 
        (account, amount, {IMPLICATION_KEY: implication}),
 
        ('Expenses:Other', -amount),
 
    ])
 
    assert not list(hook.run(txn))
0 comments (0 inline, 0 general)