NPO-Accounting/conservancy_beancount Changeset - 6a7815090ca0

Changeset - 6a7815090ca0

Parent rev.

Child rev.

[Not reviewed]

0 4 1

Brett Smith - 4 years ago 2020-07-15 14:27:05
brettcsmith@brettcsmith.org

data: Add AccountMeta class.

5 files changed with 301 insertions and 2 deletions:

conservancy_beancount/beancount_types.py

conservancy_beancount/data.py

104

tests/test_data_account.py

tests/test_data_account_meta.py

113

tests/testutil.py

0 comments (0 inline, 0 general)

conservancy_beancount/beancount_types.py

➞

Show inline comments

 """Type definitions for Beancount data structures"""
 # 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 beancount.core.data as bc_data
 from typing import (
     TYPE_CHECKING,
     Any,
     FrozenSet,
     Iterable,
     List,
     Mapping,
     NamedTuple,
     Optional,
     Set,
     Tuple,
     Type,
     Union,
+)
 if TYPE_CHECKING:
     from . import errors as errormod
 Account = bc_data.Account
 Currency = bc_data.Currency
 Meta = bc_data.Meta
 MetaKey = str
 MetaValue = Any
 MetaValueEnum = str
 Posting = bc_data.Posting
 class Directive(NamedTuple):
-    meta: bc_data.Meta
     meta: Meta
     date: datetime.date
 class Close(Directive):
     account: str
 class Error(NamedTuple):
     source: Mapping[MetaKey, MetaValue]
     message: str
     entry: Optional[Directive]
 class Open(Directive):
     account: str
     # Beancount's own type declarations say these aren't Optional, but in
     # practice these fields are both None if not defined in the directive.
     currencies: Optional[List[Currency]]
     booking: Optional[bc_data.Booking]
 class Transaction(Directive):
     flag: bc_data.Flag
     payee: Optional[str]
     narration: str
     tags: Set[str]
     links: Set[str]
     postings: List[Posting]
 ALL_DIRECTIVES: FrozenSet[Type[Directive]] = frozenset([
     Transaction,
 ])
 Entries = List[Directive]
 Errors = List[Union[Error, 'errormod.Error']]
 OptionsMap = Mapping[str, Any]
 LoadResult = Tuple[Entries, Errors, OptionsMap]

conservancy_beancount/data.py

➞

Show inline comments

 """Enhanced Beancount data structures for Conservancy
 The classes in this module are interface-compatible with Beancount's core data
 structures, and provide additional business logic that we want to use
 throughout Conservancy tools.
 """
 # 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 collections
 import datetime
 import decimal
 import functools
 import re
 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 beancount.core import data as bc_data
 from beancount.core import position as bc_position
 from typing import (
     cast,
     overload,
     Callable,
     Hashable,
     Iterable,
     Iterator,
     MutableMapping,
     Optional,
     Sequence,
     TypeVar,
     Union,
+)
 from .beancount_types import (
     Close,
     Currency,
     Directive,
     Meta,
     MetaKey,
     MetaValue,
     Open,
     Posting as BasePosting,
     Transaction,
+)
 DecimalCompat = Union[decimal.Decimal, int]
 LINK_METADATA = frozenset([
     'approval',
     'bank-statement',
     'check',
     'contract',
     'invoice',
     'purchase-order',
     'receipt',
     'rt-id',
     'statement',
     '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]:
         return iter(self._opening.meta)
     def __len__(self) -> int:
         return len(self._opening.meta)
     def __getitem__(self, key: MetaKey) -> MetaValue:
         return self._opening.meta[key]
     def __setitem__(self, key: MetaKey, value: MetaValue) -> None:
         self._opening.meta[key] = value
     def __delitem__(self, key: MetaKey) -> None:
         del self._opening.meta[key]
     @property
     def account(self) -> 'Account':
         return Account(self._opening.account)
     @property
     def booking(self) -> Optional[bc_data.Booking]:
         return self._opening.booking
     @property
     def close_date(self) -> Optional[datetime.date]:
         return None if self._closing is None else self._closing.date
     @property
     def close_meta(self) -> Optional[Meta]:
         return None if self._closing is None else self._closing.meta
     @property
     def currencies(self) -> Optional[Sequence[Currency]]:
         return self._opening.currencies
     @property
     def open_date(self) -> datetime.date:
         return self._opening.date
     @property
     def open_meta(self) -> Meta:
         return self._opening.meta
 class Account(str):
     """Account name string
     This is a string that names an account, like Assets:Bank:Checking
     or Income:Donations. This class provides additional methods for common
     account name parsing and queries.
     """
     __slots__ = ()
     SEP = bc_account.sep
     _meta_map: MutableMapping[str, AccountMeta] = {}
     @classmethod
     def load_opening(cls, opening: Open) -> None:
         cls._meta_map[opening.account] = AccountMeta(opening)
     @classmethod
     def load_closing(cls, closing: Close) -> None:
         try:
             cls._meta_map[closing.account].add_closing(closing)
         except KeyError:
             raise ValueError(
                 f"tried to load {closing.account} close directive before open",
             ) from None
     @classmethod
     def load_openings_and_closings(cls, entries: Iterable[Directive]) -> None:
         for entry in entries:
             # type ignores because Beancount's directives aren't type-checkable.
             if isinstance(entry, bc_data.Open):
                 cls.load_opening(entry)  # type:ignore[arg-type]
             elif isinstance(entry, bc_data.Close):
                 cls.load_closing(entry)  # type:ignore[arg-type]
     @property
     def meta(self) -> AccountMeta:
         return self._meta_map[self]
     def is_cash_equivalent(self) -> bool:
         return (
             self.is_under('Assets:') is not None
             and self.is_under('Assets:Prepaid', 'Assets:Receivable') is None
+        )
     def is_checking(self) -> bool:
         return self.is_cash_equivalent() and ':Check' in self
     def is_credit_card(self) -> bool:
         return self.is_under('Liabilities:CreditCard') is not None
     def is_opening_equity(self) -> bool:
         return self.is_under('Equity:Funds', 'Equity:OpeningBalance') is not None
     def is_under(self, *acct_seq: str) -> Optional[str]:
         """Return a match if this account is "under" a part of the hierarchy
         Pass in any number of account name strings as arguments. If this
         account is under one of those strings in the account hierarchy, the
         first matching string will be returned. Otherwise, None is returned.
         You can use the return value of this method as a boolean if you don't
         care which account string is matched.
         An account is considered to be under itself:
           Account('Expenses:Tax').is_under('Expenses:Tax') # returns 'Expenses:Tax'
         To do a "strictly under" search, end your search strings with colons:
           Account('Expenses:Tax').is_under('Expenses:Tax:') # returns None
           Account('Expenses:Tax').is_under('Expenses:') # returns 'Expenses:'
         This method does check that all the account boundaries match:
           Account('Expenses:Tax').is_under('Exp') # returns None
         """
         for prefix in acct_seq:
             if self.startswith(prefix) and (
                 prefix.endswith(self.SEP)
                 or self == prefix
                 or self[len(prefix)] == self.SEP
             ):
                 return prefix
         return None
     def _find_part_slice(self, index: int) -> slice:
         if index < 0:
             raise ValueError(f"bad part index {index!r}")
         start = 0
         for _ in range(index):
             try:
                 start = self.index(self.SEP, start) + 1
             except ValueError:
                 raise IndexError("part index {index!r} out of range") from None
         try:
             stop = self.index(self.SEP, start + 1)
         except ValueError:
             stop = len(self)
         return slice(start, stop)
     def count_parts(self) -> int:
         return self.count(self.SEP) + 1
     @overload
     def slice_parts(self, start: None=None, stop: None=None) -> Sequence[str]: ...
     @overload
     def slice_parts(self, start: slice, stop: None=None) -> Sequence[str]: ...
     @overload
     def slice_parts(self, start: int, stop: int) -> Sequence[str]: ...
     @overload
     def slice_parts(self, start: int, stop: None=None) -> str: ...
     def slice_parts(self,
                     start: Optional[Union[int, slice]]=None,
                     stop: Optional[int]=None,
     ) -> Sequence[str]:
         """Slice the account parts like they were a list
         Given a single index, return that part of the account name as a string.
         Otherwise, return a list of part names sliced according to the arguments.
         """
         if start is None:
             part_slice = slice(None)
         elif isinstance(start, slice):
             part_slice = start
         elif stop is None:
             return self[self._find_part_slice(start)]
         else:
             part_slice = slice(start, stop)
         return self.split(self.SEP)[part_slice]
     def root_part(self, count: int=1) -> str:
         """Return the first part(s) of the account name as a string"""
         try:
             stop = self._find_part_slice(count - 1).stop
         except IndexError:
             return self
         else:
             return self[:stop]
 class Amount(bc_amount.Amount):
     """Beancount amount after processing
     Beancount's native Amount class declares number to be Optional[Decimal],
     because the number is None when Beancount first parses a posting that does
     not have an amount, because the user wants it to be automatically balanced.
     As part of the loading process, Beancount replaces those None numbers
     with the calculated amount, so it will always be a Decimal. This class
     overrides the type declaration accordingly, so the type checker knows
     that our code doesn't have to consider the possibility that number is
     None.
     """
     number: decimal.Decimal
     # beancount.core._Amount is the plain namedtuple.
     # beancore.core.Amount adds instance methods to it.
     # b.c.Amount.__New__ calls `b.c._Amount.__new__`, which confuses type
     # checking. See <https://github.com/python/mypy/issues/1279>.
     # It works fine if you use super(), which is better practice anyway.
     # So we override __new__ just to call _Amount.__new__ this way.
     def __new__(cls, number: decimal.Decimal, currency: str) -> 'Amount':
         return super(bc_amount.Amount, Amount).__new__(cls, number, currency)
 class Metadata(MutableMapping[MetaKey, MetaValue]):
     """Transaction or posting metadata
     This class wraps a Beancount metadata dictionary with additional methods
     for common parsing and query tasks.
     """
     __slots__ = ('meta',)
     _HUMAN_NAMES: MutableMapping[MetaKey, str] = {
         # Initialize this dict with special cases.
         # We use it as a cache for other metadata names as they're queried.
         'check-id': 'Check Number',
         'paypal-id': 'PayPal ID',
         'rt-id': 'Ticket',
+    }
     def __init__(self, source: MutableMapping[MetaKey, MetaValue]) -> None:
         self.meta = source
     def __iter__(self) -> Iterator[MetaKey]:
         return iter(self.meta)
     def __len__(self) -> int:
         return len(self.meta)
     def __getitem__(self, key: MetaKey) -> MetaValue:
         return self.meta[key]
     def __setitem__(self, key: MetaKey, value: MetaValue) -> None:
         self.meta[key] = value
     def __delitem__(self, key: MetaKey) -> None:
         del self.meta[key]
     def get_links(self, key: MetaKey) -> Sequence[str]:
         try:
             value = self.meta[key]
         except KeyError:
             return ()
         if isinstance(value, str):
             return value.split()
         else:
             raise TypeError("{} metadata is a {}, not str".format(
                 key, type(value).__name__,
             ))
     def report_links(self, key: MetaKey) -> Sequence[str]:
         """Return a sequence of link strings under the named metadata key
         get_links raises a TypeError if the metadata is not a string.
         This method simply returns the empty sequence.
         Validation code (like in the plugin) usually uses get_links()
         while reporting code uses report_links().
         """
         try:
             return self.get_links(key)
         except TypeError:
             return ()
     @overload
     def first_link(self, key: MetaKey, default: None=None) -> Optional[str]: ...
     @overload
     def first_link(self, key: MetaKey, default: str) -> str: ...
     def first_link(self, key: MetaKey, default: Optional[str]=None) -> Optional[str]:
         try:
             return self.get_links(key)[0]
         except (IndexError, TypeError):
             return default
     @classmethod
     def human_name(cls, key: MetaKey) -> str:
         """Return the "human" version of a metadata name
         This is usually the metadata key with punctuation replaced with spaces,
         and then titlecased, with a few special cases. The return value is
         suitable for using in reports.
         """
         try:
             retval = cls._HUMAN_NAMES[key]
         except KeyError:
             retval = key.replace('-', ' ').title()
             retval = re.sub(r'\bId$', 'ID', retval)
             cls._HUMAN_NAMES[key] = retval
         return retval
 class PostingMeta(Metadata):
     """Combined access to posting metadata with its parent transaction metadata
     This lets you access posting metadata through a single dict-like object.
     If you try to look up metadata that doesn't exist on the posting, it will
     look for the value in the parent transaction metadata instead.
     You can set and delete metadata as well. Changes only affect the metadata
     of the posting, never the transaction. Changes are propagated to the
     underlying Beancount data structures.
     Functionally, you can think of this as identical to:
       collections.ChainMap(post.meta, txn.meta)
     Under the hood, this class does a little extra work to avoid creating
     posting metadata if it doesn't have to.
     """
     __slots__ = ('txn', 'index', 'post')
     def __init__(self,
                  txn: Transaction,
                  index: int,
                  post: Optional[BasePosting]=None,
     ) -> None:
         if post is None:
             post = txn.postings[index]
         self.txn = txn
         self.index = index
         self.post = post
         if post.meta is None:
             self.meta = self.txn.meta
         else:
             self.meta = collections.ChainMap(post.meta, txn.meta)
     def __getitem__(self, key: MetaKey) -> MetaValue:
         try:
             return super().__getitem__(key)
         except KeyError:
             if key == 'entity' and self.txn.payee is not None:
                 return self.txn.payee
             else:
                 raise
     def __setitem__(self, key: MetaKey, value: MetaValue) -> None:
         if self.post.meta is None:
             self.post = self.post._replace(meta={key: value})
             self.txn.postings[self.index] = self.post
             # mypy complains that self.post.meta could be None, but we know
             # from two lines up that it's not.
             self.meta = collections.ChainMap(self.post.meta, self.txn.meta)  # type:ignore[arg-type]
         else:
             super().__setitem__(key, value)
     def __delitem__(self, key: MetaKey) -> None:
         if self.post.meta is None:
             raise KeyError(key)
         else:
             super().__delitem__(key)
     # This is arguably cheating a litttle bit, but I'd argue the date of
     # the parent transaction still qualifies as posting metadata, and
     # it's something we want to access so often it's good to have it
     # within easy reach.
     @property
     def date(self) -> datetime.date:
         return self.txn.date
 class Posting(BasePosting):
     """Enhanced Posting objects
     This class is a subclass of Beancount's native Posting class where
     specific fields are replaced with enhanced versions:
     * The `account` field is an Account object
     * The `units` field is our Amount object (which simply declares that the
       number is always a Decimal—see that docstring for details)
     * The `meta` field is a PostingMeta object
     """
     __slots__ = ()
     account: Account
     units: Amount
     cost: Optional[bc_position.Cost]
     # mypy correctly complains that our MutableMapping is not compatible
     # with Beancount's meta type declaration of Optional[Dict]. IMO
     # Beancount's type declaration is a smidge too specific: I think its type
     # declaration should also use MutableMapping, because it would be very
     # unusual for code to specifically require a Dict over that.
     # If it did, this declaration would pass without issue.
     meta: PostingMeta  # type:ignore[assignment]
     @classmethod
     def from_beancount(cls,
                        txn: Transaction,
                        index: int,
                        post: Optional[BasePosting]=None,
     ) -> 'Posting':
         if post is None:
             post = txn.postings[index]
         return cls(
             Account(post.account),
             *post[1:5],
             # see rationale above about Posting.meta
             PostingMeta(txn, index, post), # type:ignore[arg-type]
+        )
     @classmethod
     def from_txn(cls, txn: Transaction) -> Iterator['Posting']:
         """Yield an enhanced Posting object for every posting in the transaction"""
         for index, post in enumerate(txn.postings):
             yield cls.from_beancount(txn, index, post)
     @classmethod
     def from_entries(cls, entries: Iterable[Directive]) -> Iterator['Posting']:
         """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
     def at_cost(self) -> Amount:
         if self.cost is None:
             return self.units
         else:
             return Amount(self.units.number * self.cost.number, self.cost.currency)
 _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.
     """
     match_posts = [post for post in Posting.from_txn(txn)
                    if any(pred(post.account) for pred in preds)]
     number = decimal.Decimal(0)
     if not match_posts:
         currency = ''
     else:

tests/test_data_account.py

➞

Show inline comments

 """Test Account class"""
 # 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
 from datetime import date as Date
 from beancount.core.data import Open, Close, Booking
 from conservancy_beancount import data
 clean_account_meta = pytest.fixture()(testutil.clean_account_meta)
 def check_account_meta(acct_meta, opening, closing=None):
     if isinstance(acct_meta, str):
         acct_meta = data.Account(acct_meta).meta
     assert acct_meta == opening.meta
     assert acct_meta.account == opening.account
     assert acct_meta.booking == opening.booking
     assert acct_meta.currencies == opening.currencies
     assert acct_meta.open_date == opening.date
     assert acct_meta.open_meta == opening.meta
     if closing is None:
         assert acct_meta.close_date is None
         assert acct_meta.close_meta is None
     else:
         assert acct_meta.close_date == closing.date
         assert acct_meta.close_meta == closing.meta
 @pytest.mark.parametrize('acct_name,under_arg,expected', [
     ('Expenses:Tax:Sales', 'Expenses:Tax:Sales:', False),
     ('Expenses:Tax:Sales', 'Expenses:Tax:Sales', True),
     ('Expenses:Tax:Sales', 'Expenses:Tax:', True),
     ('Expenses:Tax:Sales', 'Expenses:Tax', True),
     ('Expenses:Tax:Sales', 'Expenses:', True),
     ('Expenses:Tax:Sales', 'Expenses', True),
     ('Expenses:Tax:Sales', 'Expense', False),
     ('Expenses:Tax:Sales', 'Equity:', False),
     ('Expenses:Tax:Sales', 'Equity', False),
 ])
 def test_is_under_one_arg(acct_name, under_arg, expected):
     expected = under_arg if expected else None
     assert data.Account(acct_name).is_under(under_arg) == expected
 @pytest.mark.parametrize('acct_name,expected', [
     ('Assets:Cash', None),
     ('Assets:Checking', None),
     ('Assets:Prepaid:Expenses', 'Assets:Prepaid'),
     ('Assets:Receivable:Accounts', 'Assets:Receivable'),
 ])
 def test_is_under_multi_arg(acct_name, expected):
     assert expected == data.Account(acct_name).is_under(
         'Assets:Prepaid', 'Assets:Receivable',
+    )
     if expected:
         expected += ':'
     assert expected == data.Account(acct_name).is_under(
         'Assets:Prepaid:', 'Assets:Receivable:',
+    )
 @pytest.mark.parametrize('acct_name,expected', [
     ('Assets:Bank:Checking', True),
     ('Assets:Cash', True),
     ('Assets:Cash:EUR', True),
     ('Assets:Prepaid:Expenses', False),
     ('Assets:Prepaid:Vacation', False),
     ('Assets:Receivable:Accounts', False),
     ('Assets:Receivable:Fraud', False),
     ('Expenses:Other', False),
     ('Equity:OpeningBalance', False),
     ('Income:Other', False),
     ('Liabilities:CreditCard', False),
 ])
 def test_is_cash_equivalent(acct_name, expected):
     assert data.Account(acct_name).is_cash_equivalent() == expected
 @pytest.mark.parametrize('acct_name,expected', [
     ('Assets:Bank:Check9999', True),
     ('Assets:Bank:CheckCard', True),
     ('Assets:Bank:Checking', True),
     ('Assets:Bank:Savings', False),
     ('Assets:Cash', False),
     ('Assets:Check9999', True),
     ('Assets:CheckCard', True),
     ('Assets:Checking', True),
     ('Assets:Prepaid:Expenses', False),
     ('Assets:Receivable:Accounts', False),
     ('Expenses:Other', False),
     ('Equity:OpeningBalance', False),
     ('Income:Other', False),
     ('Liabilities:CreditCard', False),
 ])
 def test_is_checking(acct_name, expected):
     assert data.Account(acct_name).is_checking() == expected
 @pytest.mark.parametrize('acct_name,expected', [
     ('Assets:Cash', False),
     ('Assets:Prepaid:Expenses', False),
     ('Assets:Receivable:Accounts', False),
     ('Expenses:Other', False),
     ('Equity:OpeningBalance', False),
     ('Income:Other', False),
     ('Liabilities:CreditCard', True),
     ('Liabilities:CreditCard:Visa', True),
     ('Liabilities:Payable:Accounts', False),
     ('Liabilities:UnearnedIncome:Donations', False),
 ])
 def test_is_credit_card(acct_name, expected):
     assert data.Account(acct_name).is_credit_card() == expected
 @pytest.mark.parametrize('acct_name,expected', [
     ('Assets:Cash', False),
     ('Assets:Prepaid:Expenses', False),
     ('Assets:Receivable:Accounts', False),
     ('Expenses:Other', False),
     ('Equity:Funds:Restricted', True),
     ('Equity:Funds:Unrestricted', True),
     ('Equity:OpeningBalance', True),
     ('Equity:Retained:Costs', False),
     ('Income:Other', False),
     ('Liabilities:CreditCard', False),
     ('Liabilities:Payable:Accounts', False),
     ('Liabilities:UnearnedIncome:Donations', False),
 ])
 def test_is_opening_equity(acct_name, expected):
     assert data.Account(acct_name).is_opening_equity() == expected
 @pytest.mark.parametrize('acct_name', [
     'Assets:Cash',
     'Assets:Receivable:Accounts',
     'Expenses:Other',
     'Equity:Funds:Restricted',
     'Income:Other',
     'Liabilities:CreditCard',
     'Liabilities:Payable:Accounts',
 ])
 def test_slice_parts_no_args(acct_name):
     account = data.Account(acct_name)
     assert account.slice_parts() == acct_name.split(':')
 @pytest.mark.parametrize('acct_name', [
     'Assets:Cash',
     'Assets:Receivable:Accounts',
     'Expenses:Other',
     'Equity:Funds:Restricted',
     'Income:Other',
     'Liabilities:CreditCard',
     'Liabilities:Payable:Accounts',
 ])
 def test_slice_parts_index(acct_name):
     account = data.Account(acct_name)
     parts = acct_name.split(':')
     for index, expected in enumerate(parts):
         assert account.slice_parts(index) == expected
     with pytest.raises(IndexError):
         account.slice_parts(index + 1)
 @pytest.mark.parametrize('acct_name', [
     'Assets:Cash',
     'Assets:Receivable:Accounts',
     'Expenses:Other',
     'Equity:Funds:Restricted',
     'Income:Other',
     'Liabilities:CreditCard',
     'Liabilities:Payable:Accounts',
 ])
 def test_slice_parts_range(acct_name):
     account = data.Account(acct_name)
     parts = acct_name.split(':')
     for start, stop in zip([0, 0, 1, 1], [2, 3, 2, 3]):
         assert account.slice_parts(start, stop) == parts[start:stop]
 @pytest.mark.parametrize('acct_name', [
     'Assets:Cash',
     'Assets:Receivable:Accounts',
     'Expenses:Other',
     'Equity:Funds:Restricted',
     'Income:Other',
     'Liabilities:CreditCard',
     'Liabilities:Payable:Accounts',
 ])
 def test_slice_parts_slice(acct_name):
     account = data.Account(acct_name)
     parts = acct_name.split(':')
     for start, stop in zip([0, 0, 1, 1], [2, 3, 2, 3]):
         sl = slice(start, stop)
         assert account.slice_parts(sl) == parts[start:stop]
 @pytest.mark.parametrize('acct_name', [
     'Assets:Cash',
     'Assets:Receivable:Accounts',
     'Expenses:Other',
     'Equity:Funds:Restricted',
     'Income:Other',
     'Liabilities:CreditCard',
     'Liabilities:Payable:Accounts',
 ])
 def test_count_parts(acct_name):
     account = data.Account(acct_name)
     assert account.count_parts() == acct_name.count(':') + 1
 @pytest.mark.parametrize('acct_name', [
     'Assets:Cash',
     'Assets:Receivable:Accounts',
     'Expenses:Other',
     'Equity:Funds:Restricted',
     'Income:Other',
     'Liabilities:CreditCard',
     'Liabilities:Payable:Accounts',
 ])
 def test_root_part(acct_name):
     account = data.Account(acct_name)
     parts = acct_name.split(':')
     assert account.root_part() == parts[0]
     assert account.root_part(1) == parts[0]
     assert account.root_part(2) == ':'.join(parts[:2])
 def test_load_opening(clean_account_meta):
     opening = Open({'lineno': 210}, Date(2010, 2, 1), 'Assets:Cash', None, None)
     data.Account.load_opening(opening)
     check_account_meta('Assets:Cash', opening)
 def test_load_closing(clean_account_meta):
     name = 'Assets:Checking'
     opening = Open({'lineno': 230}, Date(2010, 10, 1), name, None, None)
     closing = Close({'lineno': 235}, Date(2010, 11, 1), name)
     data.Account.load_opening(opening)
     data.Account.load_closing(closing)
     check_account_meta(name, opening, closing)
 def test_load_closing_without_opening(clean_account_meta):
     closing = Close({'lineno': 245}, Date(2010, 3, 1), 'Assets:Cash')
     with pytest.raises(ValueError):
         data.Account.load_closing(closing)
 def test_load_openings_and_closings(clean_account_meta):
     entries = [
         Open({'lineno': 1, 'income-type': 'Donations'},
              Date(2000, 3, 1), 'Income:Donations', None, None),
         Open({'lineno': 2},
              Date(2000, 3, 1), 'Income:Other', None, None),
         Open({'lineno': 3, 'asset-type': 'Cash equivalent'},
              Date(2000, 4, 1), 'Assets:Checking', ['USD', 'EUR'], Booking.STRICT),
         testutil.Transaction(date=Date(2000, 4, 10), postings=[
             ('Income:Donations', -10),
             ('Assets:Checking', 10),
         ]),
         Close({'lineno': 30, 'why': 'Changed banks'},
               Date(2000, 5, 1), 'Assets:Checking')
+    ]
     data.Account.load_openings_and_closings(iter(entries))
     check_account_meta('Income:Donations', entries[0])
     check_account_meta('Income:Other', entries[1])
     check_account_meta('Assets:Checking', entries[2], entries[-1])

tests/test_data_account_meta.py

➞

Show inline comments

 new file 100644
 """Test AccountMeta class"""
 # 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 itertools
 import pytest
 from datetime import date as Date
 from beancount.core.data import Open, Close, Booking
 from conservancy_beancount import data
 def test_attributes():
     open_date = Date(2019, 6, 1)
     name = 'Assets:Bank:Checking'
     currencies = ['USD', 'EUR']
     booking = Booking.STRICT
     actual = data.AccountMeta(Open({}, open_date, name, list(currencies), booking))
     assert actual.open_date == open_date
     assert actual.account == name
     assert isinstance(actual.account, data.Account)
     assert actual.currencies == currencies
     assert actual.booking == booking
 def test_mapping():
     src_meta = {'filename': 'maptest', 'lineno': 10}
     actual = data.AccountMeta(Open(
         src_meta.copy(), Date(2019, 6, 1), 'Income:Donations', None, None,
     ))
     assert len(actual) == 2
     assert set(actual) == set(src_meta)  # Test __iter__
     for key, expected in src_meta.items():
         assert actual[key] == expected
 def test_close_attributes_without_closing():
     actual = data.AccountMeta(Open(
         {}, Date(2019, 6, 1), 'Assets:Cash', None, None,
     ))
     assert actual.close_date is None
     assert actual.close_meta is None
 def test_close_at_init():
     src_meta = {'filename': 'initclose', 'lineno': 50}
     close_date = Date(2020, 6, 1)
     name = 'Assets:Bank:MoneyMarket'
     actual = data.AccountMeta(
         Open({}, Date(2019, 6, 1), name, None, None),
         Close(src_meta.copy(), close_date, name),
+    )
     assert actual.close_date == close_date
     assert actual.close_meta == src_meta
 def test_add_closing():
     src_meta = {'filename': 'laterclose', 'lineno': 65}
     close_date = Date(2020, 1, 1)
     name = 'Assets:Bank:EUR'
     actual = data.AccountMeta(Open({}, Date(2019, 6, 1), name, None, None))
     assert actual.close_date is None
     assert actual.close_meta is None
     actual.add_closing(Close(src_meta.copy(), close_date, name))
     assert actual.close_date == close_date
     assert actual.close_meta == src_meta
 def test_add_closing_already_inited():
     name = 'Assets:Bank:Savings'
     actual = data.AccountMeta(
         Open({}, Date(2019, 6, 1), name, None, None),
         Close({}, Date(2019, 7, 1), name),
+    )
     with pytest.raises(ValueError):
         actual.add_closing(Close({}, Date(2019, 8, 1), name))
 def test_add_closing_called_twice():
     name = 'Assets:Bank:FX'
     actual = data.AccountMeta(Open({}, Date(2019, 6, 1), name, None, None))
     actual.add_closing(Close({}, Date(2019, 7, 1), name))
     with pytest.raises(ValueError):
         actual.add_closing(Close({}, Date(2019, 8, 1), name))
 @pytest.mark.parametrize('close_date,close_name', [
     (Date(2020, 6, 1), 'Income:Grants'),  # Account name doesn't match
     (Date(2010, 6, 1), 'Income:Donations'),  # Close predates Open
 ])
 def test_bad_closing_at_init(close_date, close_name):
     with pytest.raises(ValueError):
         data.AccountMeta(
             Open({}, Date(2019, 6, 1), 'Income:Donations', None, None),
             Close({}, close_date, close_name),
+        )
 @pytest.mark.parametrize('close_date,close_name', [
     (Date(2020, 6, 1), 'Income:Grants'),  # Account name doesn't match
     (Date(2010, 6, 1), 'Income:Donations'),  # Close predates Open
 ])
 def test_add_closing_wrong_account(close_date, close_name):
     actual = data.AccountMeta(
         Open({}, Date(2019, 6, 1), 'Income:Donations', None, None),
+    )
     with pytest.raises(ValueError):
         actual.add_closing(Close({}, close_date, close_name))

tests/testutil.py

➞

Show inline comments

 """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 itertools
 import re
 import beancount.core.amount as bc_amount
 import beancount.core.data as bc_data
 import beancount.loader as bc_loader
 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, rtutil
+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 primarily used as a 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():
     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),
+]
 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,
     ):
         if books_path is None:
             self._books_loader = None
         else:
             self._books_loader = TestBooksLoader(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):
         return self._books_loader
     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
 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
         att_id = str(attachment_id)
         multipart_id = None
         for attch in att_seq:
             if attch[0] == att_id:
                 break
             elif attch[2].startswith('multipart/'):
                 multipart_id = attch[0]
         else:
             return None
         tx_id = multipart_id or att_id
         if attch[1] == '(Unnamed)':
             filename = ''
         else:
             filename = attch[1]
         return {
             'id': att_id,
             'ContentType': attch[2],
             'Filename': filename,
             'Transaction': tx_id,
+        }
     def get_ticket(self, ticket_id):
         ticket_id_s = str(ticket_id)
         if ticket_id_s not in self.TICKET_DATA:
             return None
         retval = {
             'id': 'ticket/{}'.format(ticket_id_s),
             'numerical_id': ticket_id_s,
             'Requestors': [
                 f'mx{ticket_id_s}@example.org',
                 'requestor2@example.org',
             ],
+        }
         if self.want_cfs:
             retval['CF.{payment-amount}'] = ''
             retval['CF.{payment-method}'] = ''
             retval['CF.{payment-to}'] = f'Hon. Mx. {ticket_id_s}'
         return retval
     def edit_ticket(self, ticket_id, **kwargs):
         self.edits.setdefault(str(ticket_id), {}).update(kwargs)
         return True
     def get_user(self, user_id):
         user_id_s = str(user_id)
         match = re.search(r'(\d+)@', user_id_s)
         if match is None:
             email = f'mx{user_id_s}@example.org'
             user_id_num = int(user_id_s)
         else:
             email = user_id_s
             user_id_num = int(match.group(1))
         retval = {
             'id': f'user/{user_id_num}',

0 comments (0 inline, 0 general)