NPO-Accounting/conservancy_beancount Changeset - fdd9f2847b78

Changeset - fdd9f2847b78

Parent rev.

Child rev.

[Not reviewed]

0 10 0

Brett Smith - 4 years ago 2021-01-29 14:38:37
brettcsmith@brettcsmith.org

plugin: Skip enum value checks with a flag+FIXME.

We've long supported skipping documentation checks by flagging the
transaction. We haven't done the same for enumerated metadata because we
need it less often, and bad values tend to do more damage to reports.

However, occasionally when something very off-process happens, we do need it
as a matter of expediency. So support it.

In order to skip validation of these fields, the plugin requires that the
value start with the string "FIXME". This helps ensure that reports have a
consistent way to detect and warn about unfilled values in flagged
transactions.

10 files changed with 71 insertions and 10 deletions:

conservancy_beancount/plugin/core.py

conservancy_beancount/plugin/meta_paypal_id.py

conservancy_beancount/plugin/meta_tax_implication.py

setup.py

tests/test_meta_expense_type.py

tests/test_meta_income_type.py

tests/test_meta_paypal_id.py

tests/test_meta_project.py

tests/test_meta_tax_implication.py

tests/testutil.py

0 comments (0 inline, 0 general)

conservancy_beancount/plugin/core.py

➞

Show inline comments

 """Base classes for plugin checks"""
 # Copyright © 2020  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 abc
 import datetime
 import re
 from .. import config as configmod
 from .. import data
 from .. import errors as errormod
 from .. import ranges
 from typing import (
     Any,
     Container,
     Dict,
     FrozenSet,
     Generic,
     Iterable,
     Iterator,
     Mapping,
     Optional,
     Sequence,
     Type,
     TypeVar,
+)
 from ..beancount_types import (
     Account,
     Directive,
     MetaKey,
     MetaValue,
     MetaValueEnum,
     Transaction,
+)
 ### 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.
 # The default start date is the beginning of FY19, minus one day to cover FY19's
 # opening balances.
 DEFAULT_START_DATE = datetime.date(2019, 2, 28)
 # 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.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()
     def __init__(self, config: configmod.Config) -> None:
         pass
         # Subclasses that need configuration should override __init__ to check
         # and store it.
     @abc.abstractmethod
     def run(self, entry: Entry) -> errormod.Iter: ...
 ### HELPER CLASSES
 class MetadataEnum:
     """Map acceptable metadata values to their normalized forms.
     When a piece of metadata uses a set of allowed values, use this class to
     define them. You can also specify aliases that hooks will normalize to
     the primary values.
     """
     def __init__(self,
                  key: MetaKey,
                  standard_values: Iterable[MetaValueEnum],
                  aliases_map: Optional[Mapping[MetaValueEnum, MetaValueEnum]]=None,
     ) -> None:
         """Specify allowed values and aliases for this metadata.
         Arguments:
         * key: The name of the metadata key that uses this enum.
         * standard_values: A sequence of strings that enumerate the standard
           values for this metadata.
         * aliases_map: A mapping of strings to strings. The keys are
           additional allowed metadata values. The values are standard values
           that each key will evaluate to. The code asserts that all values are
           in standard_values.
         """
         self.key = key
         self._stdvalues = frozenset(standard_values)
         self._aliases: Dict[MetaValueEnum, MetaValueEnum] = dict(aliases_map or ())
         assert self._stdvalues.issuperset(self._aliases.values())
         self._aliases.update((v, v) for v in standard_values)
     def __repr__(self) -> str:
         return "{}<{}>".format(type(self).__name__, self.key)
     def __contains__(self, key: MetaValueEnum) -> bool:
         """Returns true if `key` is a standard value or alias."""
         return key in self._aliases
     def __getitem__(self, key: MetaValueEnum) -> MetaValueEnum:
         """Return the standard value for `key`.
         Raises KeyError if `key` is not a known value or alias.
         """
         return self._aliases[key]
     def __iter__(self) -> Iterator[MetaValueEnum]:
         """Iterate over standard values."""
         return iter(self._stdvalues)
     def get(self,
             key: MetaValueEnum,
             default_key: Optional[MetaValueEnum]=None,
     ) -> Optional[MetaValueEnum]:
         """Return self[key], or a default fallback if that doesn't exist.
         default_key is another key to look up, *not* a default value to return.
         This helps ensure you always get a standard value.
         """
         try:
             return self[key]
         except KeyError:
             if default_key is None:
                 return None
             else:
                 return self[default_key]
 ### HOOK SUBCLASSES
 class TransactionHook(Hook[Transaction]):
     DIRECTIVE = Transaction
     SKIP_FLAGS: Container[str] = frozenset()
     TXN_DATE_RANGE = ranges.DateRange(DEFAULT_START_DATE, DEFAULT_STOP_DATE)
     def _run_on_txn(self, txn: Transaction) -> bool:
         """Check whether we should run on a given transaction
         This method implements our usual checks for whether or not a hook
         should run on a given transaction. It's here for subclasses to use in
         their own implementations. See _PostingHook below for an example.
         """
         return (
             txn.flag not in self.SKIP_FLAGS
             and txn.date in self.TXN_DATE_RANGE
             and not data.is_opening_balance_txn(txn)
+        )
 class _PostingHook(TransactionHook, metaclass=abc.ABCMeta):
     def __init_subclass__(cls) -> None:
         cls.HOOK_GROUPS = cls.HOOK_GROUPS.union(['posting'])
     def _is_flagged_fixme(self, post: data.Posting, value: MetaValue) -> bool:
         return (
             post.meta.txn.flag == '!'
             and isinstance(value, str)
             and value.startswith('FIXME')
+        )
     def _run_on_post(self, txn: Transaction, post: data.Posting) -> bool:
         return True
     def run(self, txn: Transaction) -> errormod.Iter:
         if self._run_on_txn(txn):
             for post in data.Posting.from_txn(txn):
                 if self._run_on_post(txn, post):
                     yield from self.post_run(txn, post)
     @abc.abstractmethod
     def post_run(self, txn: Transaction, post: data.Posting) -> errormod.Iter: ...
 class _NormalizePostingMetadataHook(_PostingHook):
     """Base class to normalize posting metadata from an enum."""
     # This class provides basic functionality to filter postings, normalize
     # metadata values, and set default values.
     METADATA_KEY: MetaKey
     VALUES_ENUM: MetadataEnum
     def __init_subclass__(cls) -> None:
         super().__init_subclass__()
         cls.METADATA_KEY = cls.VALUES_ENUM.key
         cls.HOOK_GROUPS = cls.HOOK_GROUPS.union(['metadata', cls.METADATA_KEY])
     # If the posting does not specify METADATA_KEY, the hook calls
     # _default_value to get a default. This method should either return
     # a value string from METADATA_ENUM, or else raise InvalidMetadataError.
     # This base implementation does the latter.
     def _default_value(self, txn: Transaction, post: data.Posting) -> MetaValueEnum:
         raise errormod.InvalidMetadataError(txn, self.METADATA_KEY, None, post)
     def post_run(self, txn: Transaction, post: data.Posting) -> errormod.Iter:
         source_value = post.meta.get(self.METADATA_KEY)
         set_value = source_value
         error: Optional[errormod.Error] = None
         if source_value is None:
             try:
                 set_value = self._default_value(txn, post)
             except errormod.Error as error_:
                 error = error_
         else:
             try:
                 set_value = self.VALUES_ENUM[source_value]
             except KeyError:
                 error = errormod.InvalidMetadataError(
                     txn, self.METADATA_KEY, source_value, post,
+                )
                 if not self._is_flagged_fixme(post, source_value):
                     error = errormod.InvalidMetadataError(
                         txn, self.METADATA_KEY, source_value, post,
+                    )
         if error is None:
             post.meta[self.METADATA_KEY] = set_value
         else:
             yield error
 class _RequireLinksPostingMetadataHook(_PostingHook):
     """Base class to require that posting metadata include links"""
     # This base class confirms that a posting's metadata has one or more links
     # under one of the metadata keys listed in CHECKED_METADATA.
     # Most subclasses only need to define CHECKED_METADATA and _run_on_post.
     CHECKED_METADATA: Sequence[MetaKey]
     def __init_subclass__(cls) -> None:
         super().__init_subclass__()
         cls.HOOK_GROUPS = cls.HOOK_GROUPS.union(cls.CHECKED_METADATA).union('metadata')
     def _check_metadata(self,
                         txn: Transaction,
                         post: data.Posting,
                         keys: Sequence[MetaKey],
     ) -> Iterator[errormod.InvalidMetadataError]:
         have_docs = False
         for key in keys:
             try:
                 links = post.meta.get_links(key)
             except TypeError as error:
                 yield errormod.InvalidMetadataError(txn, key, post.meta[key], post)
             else:
                 have_docs = have_docs or any(links)
         if not have_docs:
             yield errormod.InvalidMetadataError(txn, '/'.join(keys), None, post)
     def post_run(self, txn: Transaction, post: data.Posting) -> errormod.Iter:
         return self._check_metadata(txn, post, self.CHECKED_METADATA)

conservancy_beancount/plugin/meta_paypal_id.py

➞

Show inline comments

 """meta_paypal_id - Validate paypal-id metadata"""
 # Copyright © 2020  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 re
 from . import core
 from .. import data
 from .. import errors as errormod
 from ..beancount_types import (
     Transaction,
+)
 class MetaPayPalID(core._PostingHook):
     METADATA_KEY = 'paypal-id'
     HOOK_GROUPS = frozenset(['metadata', METADATA_KEY])
     TXN_ID_RE = re.compile(r'^[A-Z0-9]{17}$')
     INVOICE_ID_RE = re.compile(r'^INV2(?:-[A-Z0-9]{4}){4}$')
     def _run_on_post(self, txn: Transaction, post: data.Posting) -> bool:
         if post.account.is_under('Assets:PayPal'):
             return True
         elif post.account.is_under('Assets:Receivable'):
             return self.METADATA_KEY in post.meta
         else:
             return False
     def post_run(self, txn: Transaction, post: data.Posting) -> errormod.Iter:
         if post.account.is_under('Assets:Receivable'):
             regexp = self.INVOICE_ID_RE
         else:
             regexp = self.TXN_ID_RE
         value = post.meta.get(self.METADATA_KEY)
         try:
             # A bad argument type is okay because we catch the TypeError.
             match = regexp.match(value)  # type:ignore[arg-type]
         except TypeError:
             match = None
         if match is None:
+        if match is None and not self._is_flagged_fixme(post, value):
             yield errormod.InvalidMetadataError(txn, self.METADATA_KEY, value, post)

conservancy_beancount/plugin/meta_tax_implication.py

➞

Show inline comments

 """meta_tax_implication - Validate tax-implication metadata"""
 # Copyright © 2020  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 decimal
 from . import core
 from .. import config as configmod
 from .. import data
 from typing import (
     Iterator,
     Optional,
     Tuple,
+)
 from ..beancount_types import (
     Transaction,
+)
 def _make_aliases(s: str, stdname: Optional[str]=None) -> Iterator[Tuple[str, str]]:
     if stdname is None:
         stdname = s
     elif s != stdname:
         yield (s, stdname)
     yield (s.lower(), stdname)
     if s.startswith('1099-'):
         yield from _make_aliases(f'1099{s[5:]}', stdname)
     elif s.startswith('USA-'):
         yield from _make_aliases(f'US-{s[4:]}', stdname)
     if s.endswith('-Corporation'):
         yield from _make_aliases(f'{s[:-12]}-Corp', stdname)
 class MetaTaxImplication(core._NormalizePostingMetadataHook):
     _STDNAMES = [
         '1099-MISC-Other',
         '1099-NEC',
         'Bank-Transfer',
         'Chargeback',
         'Foreign-Corporation',
         'Foreign-Grantee',
         'Foreign-Individual-Contractor',
         'Loan',
         'Refund',
         'Reimbursement',
         'Retirement-Pretax',
         'Tax-Payment',
         'USA-501c3',
         'USA-Corporation',
         'USA-Grantee',
         'W2',
+    ]
     _ALIASES = dict(
         alias for value in _STDNAMES for alias in _make_aliases(value)
+    )
     _ALIASES['1099'] = '1099-NEC'
     VALUES_ENUM = core.MetadataEnum('tax-implication', _STDNAMES, _ALIASES)
     del _STDNAMES, _ALIASES
     # Sometimes we accrue a payment before we have determined the recipient's
     # tax status.
     SKIP_FLAGS = '!'
     def __init__(self, config: configmod.Config) -> None:
         self.payment_threshold = -config.payment_threshold()
     def _run_on_post(self, txn: Transaction, post: data.Posting) -> bool:
         return (
             post.account.is_cash_equivalent()
             and post.units.number < self.payment_threshold
+        )

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.16.1',
+    version='1.16.2',
     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',
     ],
     entry_points={
         'console_scripts': [
             'accrual-report = conservancy_beancount.reports.accrual:entry_point',
             'assemble-audit-reports = conservancy_beancount.tools.audit_report:entry_point',
             'balance-sheet-report = conservancy_beancount.reports.balance_sheet:entry_point',
             'budget-report = conservancy_beancount.reports.budget:entry_point',
             'bean-sort = conservancy_beancount.tools.sort_entries:entry_point',
             'extract-odf-links = conservancy_beancount.tools.extract_odf_links:entry_point',
             'fund-report = conservancy_beancount.reports.fund:entry_point',
             'ledger-report = conservancy_beancount.reports.ledger:entry_point',
             'opening-balances = conservancy_beancount.tools.opening_balances:entry_point',
             'pdfform-extract = conservancy_beancount.pdfforms.extract:entry_point',
             'pdfform-extract-irs990scheduleA = conservancy_beancount.pdfforms.extract.irs990scheduleA:entry_point',
             'pdfform-fill = conservancy_beancount.pdfforms.fill:entry_point',
             'reconcile-paypal = conservancy_beancount.reconcile.paypal:entry_point',
             'split-ods-links = conservancy_beancount.tools.split_ods_links:entry_point',
         ],
     },
+)

tests/test_meta_expense_type.py

➞

Show inline comments

 """Test handling of expense-type metadata"""
 # Copyright © 2020  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_expense_type
 VALID_VALUES = {
     'program': 'program',
     'fundraising': 'fundraising',
     'management': 'management',
     'admin': 'management',
     'administration': 'management',
     'mgmt': 'management',
+}
 INVALID_VALUES = {
     'porgram',
     'mangement',
     'fundrasing',
     '',
     *testutil.NON_STRING_METADATA_VALUES,
+}
 TEST_KEY = 'expense-type'
 PROJECT_KEY = 'project'
 UNRESTRICTED_FUND = 'Conservancy'
 RESTRICTED_FUND = 'Alpha'
 @pytest.fixture(scope='module')
 def hook():
     config = testutil.TestConfig()
     return meta_expense_type.MetaExpenseType(config)
 @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)
 def test_invalid_values_on_transactions(hook, src_value):
     txn = testutil.Transaction(**{TEST_KEY: src_value}, postings=[
         ('Assets:Cash', -25),
         ('Expenses:General', 25),
     ])
     errors = list(hook.run(txn))
     assert errors
     testutil.check_post_meta(txn, None, None)
 @pytest.mark.parametrize('account', [
     'Assets:Cash',
     'Assets:Receivable:Accounts',
     'Equity:OpeningBalance',
     'Income:Other',
     'Liabilities:CreditCard',
     'Liabilities:Payable:Vacation',
 ])
 def test_non_expense_accounts_skipped(hook, account):
     meta = {TEST_KEY: 'program'}
     txn = testutil.Transaction(postings=[
         (account, -25),
         ('Expenses:General', 25, meta.copy()),
     ])
     errors = list(hook.run(txn))
     assert not errors
     testutil.check_post_meta(txn, None, meta)
 @pytest.mark.parametrize('account,exp_unrestricted,exp_restricted', [
     ('Expenses:Accounting', 'management', 'management'),
     ('Expenses:BadDebt', 'management', 'program'),
     ('Expenses:BankingFees', 'management', 'management'),
     ('Expenses:ComputerEquipment', 'management', 'program'),
     ('Expenses:Fines', 'management', 'program'),
     ('Expenses:FilingFees', 'management', 'program'),
     ('Expenses:Hosting', 'management', 'program'),
     ('Expenses:Insurance', 'management', 'management'),
     ('Expenses:Office', 'management', 'program'),
     ('Expenses:Other', 'management', 'program'),
     ('Expenses:Phones', 'management', 'program'),
     ('Expenses:Postage', 'management', 'program'),
     ('Expenses:ProfessionalMemberships', 'management', 'program'),
     ('Expenses:Services:Accounting', 'management', 'management'),
     ('Expenses:Services:Administration', 'management', 'management'),
     ('Expenses:Services:Advocacy', 'program', 'program'),
     ('Expenses:Services:Development', 'program', 'program'),
     ('Expenses:Services:Fundraising', 'fundraising', 'fundraising'),
     ('Expenses:Travel', 'management', 'management'),
 ])
 def test_default_values(hook, account, exp_unrestricted, exp_restricted):
     txn = testutil.Transaction(postings=[
         ('Liabilites:CreditCard', -25),
         (account, 20, {PROJECT_KEY: UNRESTRICTED_FUND}),
         (account, 5, {PROJECT_KEY: RESTRICTED_FUND}),
     ])
     errors = list(hook.run(txn))
     assert not errors
     testutil.check_post_meta(
         txn, None,
         {TEST_KEY: exp_unrestricted, PROJECT_KEY: UNRESTRICTED_FUND},
         {TEST_KEY: exp_restricted, PROJECT_KEY: RESTRICTED_FUND},
+    )
 @pytest.mark.parametrize('date,set_value', [
     (testutil.EXTREME_FUTURE_DATE, None),
     (testutil.FUTURE_DATE, 'program'),
     (testutil.FY_START_DATE, 'program'),
     (testutil.FY_MID_DATE, 'program'),
     (testutil.PAST_DATE, None),
 ])
 def test_default_value_set_in_date_range(hook, date, set_value):
     txn = testutil.Transaction(date=date, postings=[
         ('Liabilites:CreditCard', -25),
         ('Expenses:General', 25),
     ])
     errors = list(hook.run(txn))
     assert not errors
     expect_meta = None if set_value is None else {TEST_KEY: set_value}
     testutil.check_post_meta(txn, None, expect_meta)
 @pytest.mark.parametrize('src_value', INVALID_VALUES)
 def test_flagged_txn_checked(hook, src_value):
     txn = testutil.Transaction(flag='!', 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', testutil.FIXME_VALUES)
 def test_flagged_fixme_ok(hook, src_value):
     txn = testutil.Transaction(flag='!', 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: src_value})

tests/test_meta_income_type.py

➞

Show inline comments

 """Test handling of income-type metadata"""
 # Copyright © 2020  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_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():
     config = testutil.TestConfig()
     return meta_income_type.MetaIncomeType(config)
 @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)
 def test_invalid_values_on_transactions(hook, src_value):
     txn = testutil.Transaction(**{TEST_KEY: src_value}, postings=[
         ('Assets:Cash', 25),
         ('Income:Other', -25),
     ])
     errors = list(hook.run(txn))
     assert errors
     testutil.check_post_meta(txn, None, None)
 @pytest.mark.parametrize('account', [
     'Assets:Cash',
     'Assets:Receivable:Accounts',
     'Equity:OpeningBalance',
     'Expenses:Other',
     'Liabilities:CreditCard',
     'Liabilities:Payable:Vacation',
 ])
 def test_non_income_accounts_skipped(hook, account):
     meta = {TEST_KEY: 'RBI'}
     txn = testutil.Transaction(postings=[
         (account, 25),
         ('Income:Other', -25, meta.copy()),
     ])
     errors = list(hook.run(txn))
     assert not errors
     testutil.check_post_meta(txn, None, meta)
 @pytest.mark.parametrize('account,set_value', [
     ('Income:Conferences:Registrations', 'RBI'),
     ('Income:Conferences:Sponsorship', 'RBI'),
     ('Income:Donations', 'Donations'),
     ('Income:Honoraria', 'RBI'),
     ('Income:Interest', 'RBI'),
     ('Income:Interest:Dividend', 'RBI'),
     ('Income:Royalties', 'RBI'),
     ('Income:Sales', 'RBI'),
     ('Income:SoftwareDevelopment', 'RBI'),
     ('Income:TrademarkLicensing', 'RBI'),
     ('Income:TrademarkSales', 'RBI'),
 ])
 def test_default_values(hook, account, set_value):
     txn = testutil.Transaction(postings=[
         ('Assets:Cash', 25),
         (account, -25),
     ])
     errors = list(hook.run(txn))
     assert not errors
     testutil.check_post_meta(txn, None, {TEST_KEY: set_value})
 @pytest.mark.parametrize('account', [
     'Income:Other',
 ])
 def test_no_default_value(hook, account):
     txn = testutil.Transaction(postings=[
         ('Assets:Cash', 25),
         (account, -25),
     ])
     errors = list(hook.run(txn))
     assert errors
     testutil.check_post_meta(txn, None, None)
 @pytest.mark.parametrize('date,set_value', [
     (testutil.EXTREME_FUTURE_DATE, None),
     (testutil.FUTURE_DATE, 'Donations'),
     (testutil.FY_START_DATE, 'Donations'),
     (testutil.FY_MID_DATE, 'Donations'),
     (testutil.PAST_DATE, None),
 ])
 def test_default_value_set_in_date_range(hook, date, set_value):
     txn = testutil.Transaction(date=date, postings=[
         ('Assets:Cash', 25),
         ('Income:Donations', -25),
     ])
     errors = list(hook.run(txn))
     assert not errors
     expect_meta = None if set_value is None else {TEST_KEY: set_value}
     testutil.check_post_meta(txn, None, expect_meta)
 @pytest.mark.parametrize('src_value', INVALID_VALUES)
 def test_flagged_txn_checked(hook, src_value):
     txn = testutil.Transaction(flag='!', 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', testutil.FIXME_VALUES)
 def test_flagged_fixme_ok(hook, src_value):
     txn = testutil.Transaction(flag='!', 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: src_value})

tests/test_meta_paypal_id.py

➞

Show inline comments

@@ ... / @@ -3,192 +3,204 @@ @@
 # 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 itertools
 import pytest
 from . import testutil
 from conservancy_beancount.plugin import meta_paypal_id
 VALID_TXN_IDS = {
     # Basically normal ids.
     'A1234567890BCDEFG',
     '12345HIJKLNMO6780',
     # All numeric
     '05678901234567890',
     # All alphabetic
     'CDEFHSNOECRUHOECR',
+}
 VALID_INVOICE_IDS = {
     'INV2-ABCD-1234-EFGH-6789',
     # All numeric
     'INV2-1010-2020-3030-4567',
     # All alphabetic
     'INV2-ABCD-EFGH-IJKL-MNOP',
+}
 VALID_VALUES = VALID_TXN_IDS | VALID_INVOICE_IDS
 INVALID_VALUES = {
     # Empty
     '',
     ' ',
     # Punctuation and whitespace
     'Z12345-67890QRSTU',
     'Y12345.67890QRSTU',
     'X12345_67890QRSTU',
     'W12345 67890QRSTU',
     'INV2-ABCD.1234-EFGH-7890',
     'INV2-ABCD-1234_EFGH-7890',
     'INV2-ABCD-1234-EFGH 7890',
     # Too short
     'Q1234567890RSTUV',
     'INV2-ABCD-1234-EFGH-789',
     # Too long
     'V123456789012345WX',
     'INV2-ABCD-1234-EFGH-78900',
     'INV2-ABCD-1234-EFGH-7890-IJKL',
     # Bad cadence
     'INV2-ABCD-1234-EFG-H7890',
     'INV2ABCD-123-EFG-456-789',
     'INV2ABCDEFGHIJKLMNOPQRST',
+}
 ACCOUNTS = itertools.cycle([
     'Assets:PayPal',
     'Assets:Receivable:Accounts',
 ])
 TEST_KEY = 'paypal-id'
 INVALID_MSG = f"{{}} has invalid {TEST_KEY}: {{}}".format
 BAD_TYPE_MSG = f"{{}} has wrong type of {TEST_KEY}: expected str but is a {{}}".format
 @pytest.fixture(scope='module')
 def hook():
     config = testutil.TestConfig()
     return meta_paypal_id.MetaPayPalID(config)
 def paypal_account_for_id(paypal_id):
     if paypal_id.startswith('INV'):
         return 'Assets:Receivable:Accounts'
     else:
         return 'Assets:PayPal'
 @pytest.mark.parametrize('src_value', VALID_VALUES)
 def test_valid_values_on_postings(hook, src_value):
     txn = testutil.Transaction(postings=[
         ('Income:Donations', -25),
         (paypal_account_for_id(src_value), 25, {TEST_KEY: src_value}),
     ])
     assert not list(hook.run(txn))
 @pytest.mark.parametrize('src_value', INVALID_VALUES)
 def test_invalid_values_on_postings(hook, src_value):
     acct = paypal_account_for_id(src_value)
     txn = testutil.Transaction(postings=[
         ('Income:Donations', -25),
         (acct, 25, {TEST_KEY: src_value}),
     ])
     actual = {error.message for error in hook.run(txn)}
     assert actual == {INVALID_MSG(acct, src_value)}
 @pytest.mark.parametrize('src_value,acct', testutil.combine_values(
     testutil.NON_STRING_METADATA_VALUES,
     ACCOUNTS,
 ))
 def test_bad_type_values_on_postings(hook, src_value, acct):
     txn = testutil.Transaction(postings=[
         ('Income:Donations', -25),
         (acct, 25, {TEST_KEY: src_value}),
     ])
     actual = {error.message for error in hook.run(txn)}
     assert actual == {BAD_TYPE_MSG(acct, type(src_value).__name__)}
 @pytest.mark.parametrize('src_value', VALID_VALUES)
 def test_valid_values_on_transactions(hook, src_value):
     txn = testutil.Transaction(**{TEST_KEY: src_value}, postings=[
         ('Income:Donations', -25),
         (paypal_account_for_id(src_value), 25),
     ])
     assert not list(hook.run(txn))
 @pytest.mark.parametrize('src_value', INVALID_VALUES)
 def test_invalid_values_on_transactions(hook, src_value):
     acct = paypal_account_for_id(src_value)
     txn = testutil.Transaction(**{TEST_KEY: src_value}, postings=[
         ('Income:Donations', -25),
         (acct, 25),
     ])
     actual = {error.message for error in hook.run(txn)}
     assert actual == {INVALID_MSG(acct, src_value)}
 @pytest.mark.parametrize('src_value,acct', testutil.combine_values(
     testutil.NON_STRING_METADATA_VALUES,
     ACCOUNTS,
 ))
 def test_bad_type_values_on_transactions(hook, src_value, acct):
     txn = testutil.Transaction(**{TEST_KEY: src_value}, postings=[
         ('Income:Donations', -25),
         (acct, 25),
     ])
     actual = {error.message for error in hook.run(txn)}
     assert actual == {BAD_TYPE_MSG(acct, type(src_value).__name__)}
 @pytest.mark.parametrize('src_value', VALID_INVOICE_IDS)
 def test_invoice_ids_not_accepted_for_non_accruals(hook, src_value):
     acct = 'Assets:PayPal'
     txn = testutil.Transaction(**{TEST_KEY: src_value}, postings=[
         ('Income:Donations', -25),
         (acct, 25),
     ])
     actual = {error.message for error in hook.run(txn)}
     assert actual == {INVALID_MSG(acct, src_value)}
 @pytest.mark.parametrize('src_value', VALID_TXN_IDS)
 def test_transaction_ids_not_accepted_for_accruals(hook, src_value):
     acct = 'Assets:Receivable:Accounts'
     txn = testutil.Transaction(**{TEST_KEY: src_value}, postings=[
         ('Income:Donations', -25),
         (acct, 25),
     ])
     actual = {error.message for error in hook.run(txn)}
     assert actual == {INVALID_MSG(acct, src_value)}
 def test_required_for_assets_paypal(hook):
     acct = 'Assets:PayPal'
     txn = testutil.Transaction(postings=[
         ('Income:Donations', -35),
         (acct, 35),
     ])
     actual = {error.message for error in hook.run(txn)}
     assert actual == {f"{acct} missing {TEST_KEY}"}
 @pytest.mark.parametrize('txn_id,inv_id', testutil.combine_values(
     VALID_TXN_IDS,
     VALID_INVOICE_IDS,
 ))
 def test_invoice_payment_transaction_ok(hook, txn_id, inv_id):
     txn = testutil.Transaction(**{TEST_KEY: txn_id}, postings=[
         ('Assets:Receivable:Accounts', -100, {TEST_KEY: inv_id}),
         ('Assets:PayPal', 97),
         ('Expenses:BankingFees', 3),
     ])
     assert not list(hook.run(txn))
 def test_not_required_on_opening(hook):
     txn = testutil.Transaction(postings=[
         ('Assets:PayPal', 1000),
         (next(testutil.OPENING_EQUITY_ACCOUNTS), -1000),
     ])
     assert not list(hook.run(txn))
 def test_still_required_on_flagged_txn(hook):
     txn = testutil.Transaction(flag='!', postings=[
         ('Assets:PayPal', 1000),
         ('Income:Donations', -1000),
     ])
     assert list(hook.run(txn))
 @pytest.mark.parametrize('account,src_value', testutil.combine_values(
     ACCOUNTS,
     testutil.FIXME_VALUES,
 ))
 def test_flagged_fixme_ok(hook, account, src_value):
     txn = testutil.Transaction(flag='!', postings=[
         (account, 200, {TEST_KEY: src_value}),
         ('Income:Donations', -200),
     ])
     assert not list(hook.run(txn))
     testutil.check_post_meta(txn, {TEST_KEY: src_value}, None)

tests/test_meta_project.py

➞

Show inline comments

@@ ... / @@ -12,192 +12,202 @@ import pytest @@
 from . import testutil
 from conservancy_beancount import errors as errormod
 from conservancy_beancount.plugin import meta_project
 VALID_VALUES = {
     'Conservancy': 'Conservancy',
     'Alpha': 'Alpha',
     'Bravo': 'Bravo',
     'Charles': 'Charlie',
     'Chuck': 'Charlie',
+}
 INVALID_VALUES = {
     'Alhpa',
     'Yankee',
     '',
+}
 TEST_KEY = 'project'
 DEFAULT_VALUE = 'Conservancy'
 @pytest.fixture(scope='module')
 def hook():
     config = testutil.TestConfig(repo_path='repository')
     return meta_project.MetaProject(config)
 @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)
 def test_invalid_values_on_transactions(hook, src_value):
     txn = testutil.Transaction(**{TEST_KEY: src_value}, postings=[
         ('Assets:Cash', -25),
         ('Expenses:General', 25),
     ])
     errors = list(hook.run(txn))
     assert errors
     testutil.check_post_meta(txn, None, None)
 @pytest.mark.parametrize('account,required', [
     ('Assets:Cash', False),
     ('Assets:Receivable:Accounts', True),
     ('Assets:Receivable:Loans', True),
     ('Equity:OpeningBalance', False),
     ('Equity:Funds:Restricted', True),
     ('Equity:Funds:Unrestricted', False),
     ('Expenses:General', True),
     ('Income:Donations', True),
     ('Liabilities:CreditCard', False),
     ('Liabilities:Payable:Accounts', True),
     # We do want a "project" for Lia:Pay:Vacation but it has a default value
     ('Liabilities:Payable:Vacation', False),
     ('Liabilities:UnearnedIncome:Donations', True),
 ])
 def test_which_accounts_required_on(hook, account, required):
     txn = testutil.Transaction(postings=[
         ('Assets:Checking', 25),
         (account, 25),
     ])
     errors = list(hook.run(txn))
     assert required == any(errors)
 @pytest.mark.parametrize('account', [
     'Equity:Funds:Unrestricted',
     'Equity:Realized:CurrencyConversion',
     'Expenses:Payroll:Salary',
     'Expenses:Payroll:Tax',
     'Liabilities:Payable:Vacation',
 ])
 def test_default_values(hook, account):
     txn = testutil.Transaction(postings=[
         ('Assets:Checking', -25),
         (account, 25),
     ])
     errors = list(hook.run(txn))
     assert not errors
     testutil.check_post_meta(txn, None, {TEST_KEY: DEFAULT_VALUE})
 @pytest.mark.parametrize('equity,other_acct,value', testutil.combine_values(
     ['Equity:Funds:Unrestricted', 'Equity:Realized:CurrencyConversion'],
     ['Assets:Checking', 'Liabilities:CreditCard'],
     VALID_VALUES,
 ))
 def test_equity_override_txn_meta(hook, equity, other_acct, value):
     if value == DEFAULT_VALUE:
         value = f'Not{value}'
     txn = testutil.Transaction(**{TEST_KEY: value}, postings=[
         (other_acct, 100),
         (equity, -100),
     ])
     errors = list(hook.run(txn))
     assert not errors
     testutil.check_post_meta(txn, None, {TEST_KEY: DEFAULT_VALUE})
 @pytest.mark.parametrize('equity,other_acct,value', testutil.combine_values(
     ['Equity:Funds:Unrestricted', 'Equity:Realized:CurrencyConversion'],
     ['Assets:Checking', 'Liabilities:CreditCard'],
     VALID_VALUES,
 ))
 def test_equity_override_post_meta(hook, equity, other_acct, value):
     if value == DEFAULT_VALUE:
         value = f'Not{value}'
     txn = testutil.Transaction(postings=[
         (other_acct, 100),
         (equity, -100, {TEST_KEY: value}),
     ])
     actual = {error.message for error in hook.run(txn)}
     assert actual == {f"{equity} has invalid {TEST_KEY}: {value}"}
     testutil.check_post_meta(txn, None, {TEST_KEY: DEFAULT_VALUE})
 @pytest.mark.parametrize('date,required', [
     (testutil.EXTREME_FUTURE_DATE, False),
     (testutil.FUTURE_DATE, True),
     (testutil.FY_START_DATE, True),
     (testutil.FY_MID_DATE, True),
     (testutil.PAST_DATE, None),
 ])
 def test_default_value_set_in_date_range(hook, date, required):
     txn = testutil.Transaction(date=date, postings=[
         ('Expenses:Payroll:Benefits', 25),
         ('Liabilities:Payable:Vacation', -25),
     ])
     errors = list(hook.run(txn))
     assert not errors
     expect_meta = {TEST_KEY: DEFAULT_VALUE} if required else None
     testutil.check_post_meta(txn, expect_meta, expect_meta)
 @pytest.mark.parametrize('repo_path', [
     None,
     '..',
 ])
 def test_missing_project_data(repo_path):
     config = testutil.TestConfig(repo_path=repo_path)
     with pytest.raises(errormod.ConfigurationError):
         meta_project.MetaProject(config)
 @pytest.mark.parametrize('repo_path_s,data_path_s', [
     ('repository', 'Projects/project-list.yml'),
     ('..', 'LICENSE.txt'),
 ])
 def test_invalid_project_data(repo_path_s, data_path_s):
     config = testutil.TestConfig(repo_path=repo_path_s)
     with pytest.raises(errormod.ConfigurationError):
         meta_project.MetaProject(config, Path(data_path_s))
 def test_not_required_on_opening(hook):
     txn = testutil.OpeningBalance('Equity:Funds:Unrestricted')
     assert not list(hook.run(txn))
 def test_always_required_on_restricted_funds(hook):
     acct = 'Equity:Funds:Restricted'
     txn = testutil.OpeningBalance(acct)
     actual = {error.message for error in hook.run(txn)}
     assert actual == {f'{acct} missing project'}
 @pytest.mark.parametrize('src_value', INVALID_VALUES)
 def test_still_required_on_flagged_txn(hook, src_value):
     txn = testutil.Transaction(flag='!', **{TEST_KEY: src_value}, postings=[
         ('Assets:Cash', -25),
         ('Expenses:General', 25),
     ])
     errors = list(hook.run(txn))
     assert errors
     testutil.check_post_meta(txn, None, None)
 @pytest.mark.parametrize('src_value', testutil.FIXME_VALUES)
 def test_flagged_fixme_ok(hook, src_value):
     txn = testutil.Transaction(flag='!', 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: src_value})

tests/test_meta_tax_implication.py

➞

Show inline comments

 """Test handling of tax-implication metadata"""
 # Copyright © 2020  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_implication
 VALID_VALUES = {
     '1099': '1099-NEC',
     '1099-NEC': '1099-NEC',
     '1099nec': '1099-NEC',
     '1099-MISC-Other': '1099-MISC-Other',
     '1099misc-other': '1099-MISC-Other',
     'Bank-Transfer': 'Bank-Transfer',
     'Chargeback': 'Chargeback',
     'Foreign-Corporation': 'Foreign-Corporation',
     'foreign-corp': 'Foreign-Corporation',
     'Foreign-Grantee': 'Foreign-Grantee',
     'Foreign-Individual-Contractor': 'Foreign-Individual-Contractor',
     'Loan': 'Loan',
     'Refund': 'Refund',
     'Reimbursement': 'Reimbursement',
     'Retirement-Pretax': 'Retirement-Pretax',
     'Tax-Payment': 'Tax-Payment',
     'USA-501c3': 'USA-501c3',
     'USA-Corporation': 'USA-Corporation',
     'us-corp': 'USA-Corporation',
     'USA-Grantee': 'USA-Grantee',
     'US-Grantee': 'USA-Grantee',
     'W2': 'W2',
+}
 INVALID_VALUES = {
     '199',
     'W3',
     'Payrol',
     '',
+}
 TEST_KEY = 'tax-implication'
 @pytest.fixture(scope='module')
 def hook():
     config = testutil.TestConfig()
     return meta_tax_implication.MetaTaxImplication(config)
 @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=[
         ('Liabilities:Payable:Accounts', 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=[
         ('Liabilities:Payable:Accounts', 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=[
         ('Liabilities:Payable:Accounts', 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)
 def test_invalid_values_on_transactions(hook, src_value):
     txn = testutil.Transaction(**{TEST_KEY: src_value}, postings=[
         ('Liabilities:Payable:Accounts', 25),
         ('Assets:Cash', -25),
     ])
     errors = list(hook.run(txn))
     assert errors
     testutil.check_post_meta(txn, None, None)
 @pytest.mark.parametrize('count,account', enumerate([
     'Assets:Payable:Accounts',
     'Assets:Prepaid:Expenses',
     'Equity:OpeningBalance',
     'Expenses:Other',
     'Income:Other',
     'Liabilities:CreditCard',
     'Liabilities:Payable:Accounts',
     'Liabilities:UnearnedIncome:Donations',
 ], 1))
 def test_non_payment_accounts_skipped(hook, account, count):
     amount = count * 100
     meta = {TEST_KEY: 'USA-Corporation'}
     txn = testutil.Transaction(postings=[
         (account, amount),
         ('Assets:Checking', -amount, meta.copy()),
     ])
     errors = list(hook.run(txn))
     assert not errors
     testutil.check_post_meta(txn, None, meta)
 def test_asset_credits_skipped(hook):
     txn = testutil.Transaction(postings=[
         ('Income:Donations', -25),
         ('Assets:Cash', 25),
     ])
     errors = list(hook.run(txn))
     assert not errors
     testutil.check_post_meta(txn, None, None)
 @pytest.mark.parametrize('date,need_value', [
     (testutil.EXTREME_FUTURE_DATE, False),
     (testutil.FUTURE_DATE, True),
     (testutil.FY_START_DATE, True),
     (testutil.FY_MID_DATE, True),
     (testutil.PAST_DATE, False),
 ])
 def test_validation_only_in_date_range(hook, date, need_value):
     txn = testutil.Transaction(date=date, postings=[
         ('Liabilites:CreditCard', 25),
         ('Assets:Cash', -25),
     ])
     errors = list(hook.run(txn))
     assert bool(errors) == bool(need_value)
     testutil.check_post_meta(txn, None, None)
 @pytest.mark.parametrize('src_value', INVALID_VALUES)
-def test_flagged_txn_skipped(hook, src_value):
+def test_flagged_txn_checked(hook, src_value):
     txn = testutil.Transaction(flag='!', **{TEST_KEY: src_value}, postings=[
         ('Liabilities:Payable:Accounts', 25),
         ('Assets:Cash', -25),
     ])
     assert list(hook.run(txn))
 @pytest.mark.parametrize('src_value', testutil.FIXME_VALUES)
 def test_flagged_fixme_ok(hook, src_value):
     txn = testutil.Transaction(flag='!', postings=[
         ('Liabilities:Payable:Accounts', 25),
         ('Assets:Cash', -25, {TEST_KEY: src_value}),
     ])
     assert not list(hook.run(txn))
     testutil.check_post_meta(txn, None, {TEST_KEY: src_value})

tests/testutil.py

➞

Show inline comments

@@ ... / @@ -5,384 +5,390 @@ @@
 # Full copyright and licensing details can be found at toplevel file
 # LICENSE.txt in the repository.
 import datetime
 import itertools
 import re
 import unittest.mock
 import beancount.core.amount as bc_amount
 import beancount.core.data as bc_data
 import beancount.loader as bc_loader
 import beancount.parser.options as bc_options
 import git
 import odf.element
 import odf.opendocument
 import odf.table
 from decimal import Decimal
 from pathlib import Path
 from typing import Any, Optional, NamedTuple
 from conservancy_beancount import books, data, rtutil
 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)
 TESTS_DIR = Path(__file__).parent
 # This function is a teardown fixture, but different test files use
 # it with different scopes. Typical usage looks like:
 #   clean_account_meta = pytest.fixture([options])(testutil.clean_account_meta)
 def clean_account_meta():
     try:
         yield
     finally:
         data.Account.load_options_map(bc_options.OPTIONS_DEFAULTS)
         data.Account._meta_map.clear()
 def _ods_cell_value_type(cell):
     assert cell.tagName == 'table:table-cell'
     return cell.getAttribute('valuetype')
 def _ods_cell_value(cell):
     value_type = cell.getAttribute('valuetype')
     if value_type == 'currency' or value_type == 'float':
         return Decimal(cell.getAttribute('value'))
     elif value_type == 'date':
         return datetime.datetime.strptime(
             cell.getAttribute('datevalue'), '%Y-%m-%d',
         ).date()
     else:
         return cell.getAttribute('value')
 def _ods_elem_text(elem):
     if isinstance(elem, odf.element.Text):
         return elem.data
     else:
         return '\0'.join(_ods_elem_text(child) for child in elem.childNodes)
 odf.element.Element.value_type = property(_ods_cell_value_type)
 odf.element.Element.value = property(_ods_cell_value)
 odf.element.Element.text = property(_ods_elem_text)
 def check_lines_match(lines, expect_patterns, source='output'):
     for pattern in expect_patterns:
         assert any(re.search(pattern, line) for line in lines), \
             f"{pattern!r} not found in {source}"
 def check_logs_match(caplog, expected):
     records = iter(caplog.records)
     for exp_level, exp_msg in expected:
         exp_level = exp_level.upper()
         assert any(
             log.levelname == exp_level and log.message == exp_msg for log in records
         ), f"{exp_level} log {exp_msg!r} not found"
 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 combine_values(*value_seqs):
     stop = 0
     for seq in value_seqs:
         try:
             stop = max(stop, len(seq))
         except TypeError:
             pass
     return itertools.islice(
         zip(*(itertools.cycle(seq) for seq in value_seqs)),
         stop,
+    )
 def date_seq(date=FY_MID_DATE, step=1):
     while True:
         yield date
         date += datetime.timedelta(days=step)
 def parse_date(s, fmt='%Y-%m-%d'):
     return datetime.datetime.strptime(s, fmt).date()
 def test_path(s):
     if s is None:
         return s
     s = Path(s)
     if not s.is_absolute():
         s = TESTS_DIR / s
     return s
 def Amount(number, currency='USD'):
     return bc_amount.Amount(Decimal(number), currency)
 def Cost(number, currency='USD', date=FY_MID_DATE, label=None):
     return bc_data.Cost(Decimal(number), currency, date, label)
 def Posting(account, number,
             currency='USD', cost=None, price=None, flag=None,
             _post_type=bc_data.Posting, _meta_type=None, **meta):
     if cost is not None:
         cost = Cost(*cost)
     if not meta:
         meta = None
     elif _meta_type:
         meta = _meta_type(meta)
     return _post_type(
         account,
         Amount(number, currency),
         cost,
         price,
         flag,
         meta,
+    )
 def Transaction(date=FY_MID_DATE, flag='*', payee=None,
                 narration='', tags=None, links=None, postings=(),
                 **meta):
     if isinstance(date, str):
         date = parse_date(date)
     meta.setdefault('filename', '<test>')
     meta.setdefault('lineno', 0)
     real_postings = []
     for post in postings:
         try:
             post.account
         except AttributeError:
             if isinstance(post[-1], dict):
                 args = post[:-1]
                 kwargs = post[-1]
             else:
                 args = post
                 kwargs = {}
             post = Posting(*args, **kwargs)
         real_postings.append(post)
     return bc_data.Transaction(
         meta,
         date,
         flag,
         payee,
         narration,
         set(tags or ''),
         set(links or ''),
         real_postings,
+    )
 LINK_METADATA_STRINGS = {
     'Invoices/304321.pdf',
     'rt:123/456',
     'rt://ticket/234',
+}
 NON_LINK_METADATA_STRINGS = {
     '',
     ' ',
     '     ',
+}
 NON_STRING_METADATA_VALUES = [
     Decimal(5),
     FY_MID_DATE,
     Amount(50),
     Amount(500, None),
+]
 FIXME_VALUES = [
     'FIXME',
     'FIXME loose comment',
     'FIXME: comment with punctuation',
+]
 OPENING_EQUITY_ACCOUNTS = itertools.cycle([
     'Equity:Funds:Unrestricted',
     'Equity:Funds:Restricted',
     'Equity:OpeningBalance',
 ])
 class ODSCell:
     @classmethod
     def from_row(cls, row):
         return row.getElementsByType(odf.table.TableCell)
     @classmethod
     def from_sheet(cls, spreadsheet):
         for row in spreadsheet.getElementsByType(odf.table.TableRow):
             yield list(cls.from_row(row))
     @classmethod
     def from_ods_file(cls, path):
         ods = odf.opendocument.load(path)
         return cls.from_sheet(ods.spreadsheet)
 def OpeningBalance(acct=None, **txn_meta):
     if acct is None:
         acct = next(OPENING_EQUITY_ACCOUNTS)
     return Transaction(**txn_meta, postings=[
         ('Assets:Receivable:Accounts', 100),
         ('Assets:Receivable:Loans', 200),
         ('Liabilities:Payable:Accounts', -15),
         ('Liabilities:Payable:Vacation', -25),
         (acct, -260),
     ])
 class TestBooksLoader(books.Loader):
     def __init__(self, source):
         self.source = source
     def load_all(self, from_year=None):
         return bc_loader.load_file(self.source)
     def load_fy_range(self, from_fy, to_fy=None):
         return self.load_all()
 class TestConfig:
     def __init__(self, *,
                  books_path=None,
                  fiscal_year=(3, 1),
                  payment_threshold=0,
                  repo_path=None,
                  rt_client=None,
     ):
         self._books_path = books_path
         self.fiscal_year = fiscal_year
         self._payment_threshold = Decimal(payment_threshold)
         self.repo_path = test_path(repo_path)
         self._rt_client = rt_client
         if rt_client is None:
             self._rt_wrapper = None
         else:
             self._rt_wrapper = rtutil.RT(rt_client)
     def books_loader(self):
         if self._books_path is None:
             return None
         else:
             return TestBooksLoader(self._books_path)
     def books_path(self):
         return self._books_path
     def books_repo(self):
         return None
     def config_file_path(self):
         return test_path('userconfig/conservancy_beancount/config.ini')
     def fiscal_year_begin(self):
         return books.FiscalYear(*self.fiscal_year)
     def payment_threshold(self):
         return self._payment_threshold
     def repository_path(self):
         return self.repo_path
     def rt_client(self):
         return self._rt_client
     def rt_wrapper(self):
         return self._rt_wrapper
 def TestRepo(head_hexsha='abcd1234', dirty=False):
     retval = unittest.mock.Mock(spec=git.Repo)
     retval.is_dirty.return_value = dirty
     retval.head.commit.hexsha = head_hexsha
     return retval
 class _TicketBuilder:
     MESSAGE_ATTACHMENTS = [
         ('(Unnamed)', 'multipart/alternative', '0b'),
         ('(Unnamed)', 'text/plain', '1.2k'),
         ('(Unnamed)', 'text/html', '1.4k'),
+    ]
     MISC_ATTACHMENTS = [
         ('Forwarded Message.eml', 'message/rfc822', '3.1k'),
         ('photo.jpg', 'image/jpeg', '65.2k'),
         ('ConservancyInvoice-301.pdf', 'application/pdf', '326k'),
         ('Company_invoice-2020030405_as-sent.pdf', 'application/pdf', '50k'),
         ('statement.txt', 'text/plain', '652b'),
         ('screenshot.png', 'image/png', '1.9m'),
+    ]
     def __init__(self):
         self.id_seq = itertools.count(1)
         self.misc_attchs = itertools.cycle(self.MISC_ATTACHMENTS)
     def new_attch(self, attch):
         return (str(next(self.id_seq)), *attch)
     def new_msg_with_attachments(self, attachments_count=1):
         for attch in self.MESSAGE_ATTACHMENTS:
             yield self.new_attch(attch)
         for _ in range(attachments_count):
             yield self.new_attch(next(self.misc_attchs))
     def new_messages(self, messages_count, attachments_count=None):
         for n in range(messages_count):
             if attachments_count is None:
                 att_count = messages_count - n
             else:
                 att_count = attachments_count
             yield from self.new_msg_with_attachments(att_count)
 class RTClient:
     _builder = _TicketBuilder()
     DEFAULT_URL = 'https://example.org/defaultrt/REST/1.0/'
     TICKET_DATA = {
         '1': list(_builder.new_messages(1, 3)),
         '2': list(_builder.new_messages(2, 1)),
         '3': list(_builder.new_messages(3, 0)),
+    }
     del _builder
     def __init__(self,
                  url=DEFAULT_URL,
                  default_login=None,
                  default_password=None,
                  proxy=None,
                  default_queue='General',
                  skip_login=False,
                  verify_cert=True,
                  http_auth=None,
                  want_cfs=True,
     ):
         self.url = url
         if http_auth is None:
             self.user = default_login
             self.password = default_password
             self.auth_method = 'login'
             self.login_result = skip_login or None
         else:
             self.user = http_auth.username
             self.password = http_auth.password
             self.auth_method = type(http_auth).__name__
             self.login_result = True
         self.last_login = None
         self.want_cfs = want_cfs
         self.edits = {}
     def login(self, login=None, password=None):
         if login is None and password is None:
             login = self.user
             password = self.password
         self.login_result = bool(login and password and not password.startswith('bad'))
         self.last_login = (login, password, self.login_result)
         return self.login_result
     def get_attachments(self, ticket_id):
         try:
             return list(self.TICKET_DATA[str(ticket_id)])
         except KeyError:
             return None
     def get_attachment(self, ticket_id, attachment_id):
         try:
             att_seq = iter(self.TICKET_DATA[str(ticket_id)])
         except KeyError:
             return None

0 comments (0 inline, 0 general)