NPO-Accounting/conservancy_beancount Changeset - 701ccdc19250

Changeset - 701ccdc19250

Parent rev.

Child rev.

[Not reviewed]

0 6 0

Brett Smith - 4 years ago 2020-04-28 19:33:30
brettcsmith@brettcsmith.org

tests: Test where Transactions are real NamedTuples.

This makes methods like _replace available in real code, and caught the
bug where we can't use @functools.lru_cache with Transaction arguments,
because they're unhashable due to their mutable members.

6 files changed with 79 insertions and 65 deletions:

conservancy_beancount/data.py

tests/test_data_is_opening_balance_txn.py

tests/test_meta_entity.py

tests/test_meta_invoice.py

tests/test_meta_project.py

tests/testutil.py

0 comments (0 inline, 0 general)

conservancy_beancount/data.py

➞

Show inline comments

@@ ... / @@ -22,29 +22,31 @@ throughout Conservancy tools. @@
 import collections
 import datetime
 import decimal
 import functools
 from beancount.core import account as bc_account
 from beancount.core import amount as bc_amount
 from beancount.core import convert as bc_convert
 from typing import (
     cast,
     Callable,
     Hashable,
     Iterable,
     Iterator,
     MutableMapping,
     Optional,
     Sequence,
     TypeVar,
     Union,
+)
 from .beancount_types import (
     Directive,
     MetaKey,
     MetaValue,
     Posting as BasePosting,
     Transaction,
+)
 DecimalCompat = Union[decimal.Decimal, int]
@@ ... / @@ -301,24 +303,37 @@ class Posting(BasePosting): @@
         """Yield an enhanced Posting object for every posting in these entries"""
         for entry in entries:
             # Because Beancount's own Transaction class isn't type-checkable,
             # we can't statically check this. Might as well rely on duck
             # typing while we're at it: just try to yield postings from
             # everything, and ignore entries that lack a postings attribute.
             try:
                 yield from cls.from_txn(entry)  # type:ignore[arg-type]
             except AttributeError:
                 pass
 _KT = TypeVar('_KT', bound=Hashable)
 _VT = TypeVar('_VT')
 class _SizedDict(collections.OrderedDict, MutableMapping[_KT, _VT]):
     def __init__(self, maxsize: int=128) -> None:
         self.maxsize = maxsize
         super().__init__()
     def __setitem__(self, key: _KT, value: _VT) -> None:
         super().__setitem__(key, value)
         for _ in range(self.maxsize, len(self)):
             self.popitem(last=False)
 def balance_of(txn: Transaction,
                *preds: Callable[[Account], Optional[bool]],
 ) -> Amount:
     """Return the balance of specified postings in a transaction.
     Given a transaction and a series of account predicates, balance_of
     returns the balance of the amounts of all postings with accounts that
     match any of the predicates.
     balance_of uses the "weight" of each posting, so the return value will
     use the currency of the postings' cost when available.
     """
@@ ... / @@ -326,21 +341,32 @@ def balance_of(txn: Transaction, @@
                    if any(pred(post.account) for pred in preds)]
     number = decimal.Decimal(0)
     if not match_posts:
         currency = ''
     else:
         weights: Sequence[Amount] = [
             bc_convert.get_weight(post) for post in match_posts
+        ]
         number = sum((wt.number for wt in weights), number)
         currency = weights[0].currency
     return Amount(number, currency)
-@functools.lru_cache()
+_opening_balance_cache: MutableMapping[str, bool] = _SizedDict()
 def is_opening_balance_txn(txn: Transaction) -> bool:
     key = '\0'.join(
         f'{post.account}={post.units}' for post in txn.postings
+    )
     try:
         return _opening_balance_cache[key]
     except KeyError:
         pass
     opening_equity = balance_of(txn, Account.is_opening_equity)
     if not opening_equity.currency:
         return False
     rest = balance_of(txn, lambda acct: not acct.is_opening_equity())
     if not rest.currency:
         return False
     return abs(opening_equity.number + rest.number) < decimal.Decimal('.01')
         retval = False
     else:
         rest = balance_of(txn, lambda acct: not acct.is_opening_equity())
         if not rest.currency:
             retval = False
         else:
             retval = abs(opening_equity.number + rest.number) < decimal.Decimal('.01')
     _opening_balance_cache[key] = retval
     return retval

tests/test_data_is_opening_balance_txn.py

➞

Show inline comments

@@ ... / @@ -14,38 +14,38 @@ @@
 # 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 decimal import Decimal
 import pytest
 from . import testutil
 from conservancy_beancount import data
 def test_typical_opening():
-    txn = testutil.Transaction.opening_balance()
+    txn = testutil.OpeningBalance()
     assert data.is_opening_balance_txn(txn)
 def test_multiacct_opening():
     txn = testutil.Transaction(postings=[
         ('Assets:Receivable:Accounts', 100),
         (next(testutil.OPENING_EQUITY_ACCOUNTS), -100),
         ('Liabilities:Payable:Accounts', -150),
         (next(testutil.OPENING_EQUITY_ACCOUNTS), 150),
     ])
     assert data.is_opening_balance_txn(txn)
 def test_opening_with_fx():
-    txn = testutil.Transaction.opening_balance()
+    txn = testutil.OpeningBalance()
     equity_post = txn.postings[-1]
     txn.postings[-1] = equity_post._replace(
         units=testutil.Amount(equity_post.units.number * Decimal('.9'), 'EUR'),
         cost=testutil.Cost('1.11111'),
+    )
     assert data.is_opening_balance_txn(txn)
 @pytest.mark.parametrize('acct1,acct2,number', [
     ('Assets:Receivable:Accounts', 'Income:Donations', 100),
     ('Expenses:Other', 'Liabilities:Payable:Accounts', 200),
     ('Expenses:Other', 'Equity:Retained:Costs', 300),
     # Release from restriction

tests/test_meta_entity.py

➞

Show inline comments

@@ ... @@
         ('Assets:Checking', -25),
         (account, 25),
     ])
     errors = list(hook.run(txn))
     if not required:
         assert not errors
     else:
         assert errors
         assert any(error.message == "{} missing entity".format(account)
                    for error in errors)
 def test_not_required_on_opening(hook):
-    txn = testutil.Transaction.opening_balance()
+    txn = testutil.OpeningBalance()
     assert not list(hook.run(txn))

tests/test_meta_invoice.py

➞

Show inline comments

@@ ... @@
     REQUIRED_ACCOUNTS,
     NON_REQUIRED_ACCOUNTS,
 ))
 def test_missing_invoice(hook, acct1, acct2):
     txn = testutil.Transaction(postings=[
         (acct2, -25),
         (acct1, 25),
     ])
     actual = {error.message for error in hook.run(txn)}
     assert actual == {"{} missing {}".format(acct1, TEST_KEY)}
 def test_not_required_on_opening(hook):
-    txn = testutil.Transaction.opening_balance()
+    txn = testutil.OpeningBalance()
     assert not list(hook.run(txn))

tests/test_meta_project.py

➞

Show inline comments

@@ ... / @@ -149,20 +149,20 @@ def test_missing_project_data(repo_path): @@
         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.Transaction.opening_balance('Equity:Funds:Unrestricted')
+    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.Transaction.opening_balance(acct)
+    txn = testutil.OpeningBalance(acct)
     actual = {error.message for error in hook.run(txn)}
     assert actual == {f'{acct} missing project'}

tests/testutil.py

➞

Show inline comments

@@ ... / @@ -84,24 +84,55 @@ def Posting(account, number, @@
         cost = Cost(*cost)
     if not meta:
         meta = None
     return 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 = {
     '',
     ' ',
     '     ',
+}
@@ ... / @@ -121,77 +152,34 @@ OPENING_EQUITY_ACCOUNTS = itertools.cycle([ @@
 def balance_map(source=None, **kwargs):
     # The source and/or kwargs should map currency name strings to
     # things you can pass to Decimal (a decimal string, an int, etc.)
     # This returns a dict that maps currency name strings to Amount instances.
     retval = {}
     if source is not None:
         retval.update((currency, Amount(number, currency))
                       for currency, number in source)
     if kwargs:
         retval.update(balance_map(kwargs.items()))
     return retval
 class Transaction:
     def __init__(self,
                  date=FY_MID_DATE, flag='*', payee=None,
                  narration='', tags=None, links=None, postings=None,
                  **meta):
         if isinstance(date, str):
             date = parse_date(date)
         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)
         if postings is not None:
             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)
     @classmethod
     def opening_balance(cls, acct=None, **txn_meta):
         if acct is None:
             acct = next(OPENING_EQUITY_ACCOUNTS)
         return cls(**txn_meta, postings=[
             ('Assets:Receivable:Accounts', 100),
             ('Assets:Receivable:Loans', 200),
             ('Liabilities:Payable:Accounts', -15),
             ('Liabilities:Payable:Vacation', -25),
             (acct, -260),
         ])
 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 TestConfig:
     def __init__(self, *,
                  payment_threshold=0,
                  repo_path=None,
                  rt_client=None,
     ):
         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

0 comments (0 inline, 0 general)