NPO-Accounting/conservancy_beancount Changeset - 6213bc1e5da9

Changeset - 6213bc1e5da9

Parent rev.

Child rev.

[Not reviewed]

0 2 1

Brett Smith - 4 years ago 2020-06-20 23:04:53
brettcsmith@brettcsmith.org

reports: Add PeriodPostings class.

This is AccountPostings from the ledger report, cleaned up to be more
general.

3 files changed with 126 insertions and 20 deletions:

conservancy_beancount/reports/core.py

conservancy_beancount/reports/ledger.py

tests/test_reports_period_postings.py

0 comments (0 inline, 0 general)

conservancy_beancount/reports/core.py

➞

Show inline comments

@@ ... / @@ -300,192 +300,243 @@ class RelatedPostings(Sequence[data.Posting]): @@
         This method takes an iterable of postings and returns a mapping.
         The keys of the mapping are the values of post.meta.get(key, default).
         The values are RelatedPostings instances that contain all the postings
         that had that same metadata value.
         """
         def key_func(post: data.Posting) -> Optional[MetaValue]:
             return post.meta.get(key, default)
         return cls._group_by(postings, key_func)
     @classmethod
     def group_by_first_meta_link(
             cls: Type[RelatedType],
             postings: Iterable[data.Posting],
             key: MetaKey,
     ) -> Iterator[Tuple[Optional[str], RelatedType]]:
         """Relate postings by the first link in metadata
         This method takes an iterable of postings and returns a mapping.
         The keys of the mapping are the values of
         post.meta.first_link(key, None).
         The values are RelatedPostings instances that contain all the postings
         that had that same first metadata link.
         """
         def key_func(post: data.Posting) -> Optional[MetaValue]:
             return post.meta.first_link(key, None)
         return cls._group_by(postings, key_func)
     def __repr__(self) -> str:
         return f'<{type(self).__name__} {self._postings!r}>'
     @overload
     def __getitem__(self: RelatedType, index: int) -> data.Posting: ...
     @overload
     def __getitem__(self: RelatedType, s: slice) -> RelatedType: ...
     def __getitem__(self: RelatedType,
                     index: Union[int, slice],
     ) -> Union[data.Posting, RelatedType]:
         if isinstance(index, slice):
             return type(self)(self._postings[index], _can_own=True)
         else:
             return self._postings[index]
     def __len__(self) -> int:
         return len(self._postings)
     def all_meta_links(self, key: MetaKey) -> Iterator[str]:
         return filters.iter_unique(
             link for post in self for link in post.meta.report_links(key)
+        )
     @overload
     def first_meta_links(self, key: MetaKey, default: str='') -> Iterator[str]: ...
     @overload
     def first_meta_links(self, key: MetaKey, default: None) -> Iterator[Optional[str]]: ...
     def first_meta_links(self,
                          key: MetaKey,
                          default: Optional[str]='',
     ) -> Iterator[Optional[str]]:
         retval = filters.iter_unique(
             post.meta.first_link(key, default) for post in self
+        )
         if default == '':
             retval = (s for s in retval if s)
         return retval
     def iter_with_balance(self) -> Iterator[Tuple[data.Posting, Balance]]:
         balance = MutableBalance()
         for post in self:
             balance += post.units
             yield post, balance
     def balance(self) -> Balance:
         return Balance(post.units for post in self)
     def balance_at_cost(self) -> Balance:
         return Balance(post.at_cost() for post in self)
     def balance_at_cost_by_date(self, date: datetime.date) -> Balance:
         for index, post in enumerate(self):
             if post.meta.date >= date:
                 break
         else:
             index += 1
         return Balance(post.at_cost() for post in self._postings[:index])
     def meta_values(self,
                     key: MetaKey,
                     default: Optional[MetaValue]=None,
     ) -> Set[Optional[MetaValue]]:
         return {post.meta.get(key, default) for post in self}
 class PeriodPostings(RelatedPostings):
     """Postings filtered and balanced over a date range
     Create a subclass with ``PeriodPostings.with_start_date(date)``.
     Note that there is no explicit stop date. The expectation is that the
     caller has already filtered out posts past the stop date from the input.
     Instances of that subclass will have three Balance attributes:
     * ``start_bal`` is the balance at cost of postings to your start date
     * ``period_bal`` is the balance at cost of postings from your start date
     * ``stop_bal`` is the balance at cost of all postings
     Use this subclass when your report includes a lot of balances over time to
     help you get the math right.
     """
     __slots__ = (
         'begin_bal',
         'end_bal',
         'period_bal',
         'start_bal',
         'stop_bal',
+    )
     START_DATE = datetime.date(datetime.MINYEAR, 1, 1)
     def __init__(self,
                  source: Iterable[data.Posting]=(),
                  *,
                  _can_own: bool=False,
     ) -> None:
         start_posts: List[data.Posting] = []
         period_posts: List[data.Posting] = []
         for post in source:
             if post.meta.date < self.START_DATE:
                 start_posts.append(post)
             else:
                 period_posts.append(post)
         super().__init__(period_posts, _can_own=True)
         self.start_bal = RelatedPostings(start_posts, _can_own=True).balance_at_cost()
         self.period_bal = self.balance_at_cost()
         self.stop_bal = self.start_bal + self.period_bal
         # Convenience aliases
         self.begin_bal = self.start_bal
         self.end_bal = self.stop_bal
     @classmethod
     def with_start_date(cls: Type[RelatedType], start_date: datetime.date) -> Type[RelatedType]:
         name = f'BalancePostings{start_date.strftime("%Y%m%d")}'
         return type(name, (cls,), {'START_DATE': start_date})
 class BaseSpreadsheet(Generic[RT, ST], metaclass=abc.ABCMeta):
     """Abstract base class to help write spreadsheets
     This class provides the very core logic to write an arbitrary set of data
     rows to arbitrary output. It calls hooks when it starts writing the
     spreadsheet, starts a new "section" of rows, ends a section, and ends the
     spreadsheet.
     RT is the type of the input data rows. ST is the type of the section
     identifier that you create from each row. If you don't want to use the
     section logic at all, set ST to None and define section_key to return None.
     """
     @abc.abstractmethod
     def section_key(self, row: RT) -> ST:
         """Return the section a row belongs to
         Given a data row, this method should return some identifier for the
         "section" the row belongs to. The write method uses this to
         determine when to call start_section and end_section.
         If your spreadsheet doesn't need sections, define this to return None.
         """
         ...
     @abc.abstractmethod
     def write_row(self, row: RT) -> None:
         """Write a data row to the output spreadsheet
         This method is called once for each data row in the input.
         """
         ...
     # The next four methods are all called by the write method when the name
     # says. You may override them to output headers or sums, record
     # state, etc. The default implementations are all noops.
     def start_spreadsheet(self) -> None:
         pass
     def start_section(self, key: ST) -> None:
         pass
     def end_section(self, key: ST) -> None:
         pass
     def end_spreadsheet(self) -> None:
         pass
     def write(self, rows: Iterable[RT]) -> None:
         prev_section: Optional[ST] = None
         self.start_spreadsheet()
         for row in rows:
             section = self.section_key(row)
             if section != prev_section:
                 if prev_section is not None:
                     self.end_section(prev_section)
                 self.start_section(section)
                 prev_section = section
             self.write_row(row)
         try:
             should_end = section is not None
         except NameError:
             should_end = False
         if should_end:
             self.end_section(section)
         self.end_spreadsheet()
 class BaseODS(BaseSpreadsheet[RT, ST], metaclass=abc.ABCMeta):
     """Abstract base class to help write OpenDocument spreadsheets
     This class provides the very core logic to write an arbitrary set of data
     rows to an OpenDocument spreadsheet. It provides helper methods for
     building sheets, rows, and cells.
     See also the BaseSpreadsheet base class for additional documentation about
     methods you must and can define, the definition of RT and ST, etc.
     """
     # Defined in the XSL spec, "Definitions of Units of Measure"
     MEASUREMENT_UNITS = frozenset([
         'cm',
         'em',
         'in',
         'mm',
         'pc',
         'pt',
         'px',
     ])
     MEASUREMENT_RE = re.compile(
         r'([-+]?(?:\d+\.?|\.\d+|\d+\.\d+))({})'.format('|'.join(MEASUREMENT_UNITS)),
         re.ASCII,
+    )
     def __init__(self, rt_wrapper: Optional[rtutil.RT]=None) -> None:
         self.rt_wrapper = rt_wrapper

conservancy_beancount/reports/ledger.py

➞

Show inline comments

 """ledger.py - General ledger report from Beancount
 This tool produces a spreadsheet that shows postings in Beancount, organized
 by account.
 Specify the date range you want to report with the ``--begin`` and ``--end``
 options.
 Select the accounts you want to report with the ``--account`` option. You can
 specify this option multiple times. The report will include at least one sheet
 for each account you specify. Subaccounts will be reported on that sheet as
 well.
 Select the postings you want to report by passing metadata search terms in
 ``name=value`` format.
 Run ``ledger-report --help`` for abbreviations and other options.
 Examples
 --------
 Report all activity related to a given project::
     ledger-report project=NAME
 Get all Assets postings for a given month to help with reconciliation::
     ledger-report -a Assets -b 2018-05-01 -e 2018-06-01
 """
 # 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 argparse
 import collections
 import datetime
 import enum
 import itertools
 import operator
 import logging
 import sys
 from typing import (
     Callable,
     Dict,
     Iterable,
     Iterator,
     List,
     Mapping,
     Optional,
     Sequence,
     TextIO,
     Tuple,
     Union,
+)
 from pathlib import Path
 import odf.table  # type:ignore[import]
 from beancount.core import data as bc_data
 from beancount.parser import printer as bc_printer
 from . import core
 from .. import books
 from .. import cliutil
 from .. import config as configmod
 from .. import data
 from .. import ranges
 from .. import rtutil
 PostTally = List[Tuple[int, data.Account]]
 PROGNAME = 'ledger-report'
 logger = logging.getLogger('conservancy_beancount.reports.ledger')
 class AccountPostings(core.RelatedPostings):
     START_DATE: datetime.date
     def __init__(self,
                  source: Iterable[data.Posting]=(),
                  *,
                  _can_own: bool=False,
     ) -> None:
         super().__init__(source, _can_own=_can_own)
         self.start_bal = self.balance_at_cost_by_date(self.START_DATE)
         self.stop_bal = self.balance_at_cost()
         self.period_bal = self.stop_bal - self.start_bal
 class LedgerODS(core.BaseODS[data.Posting, data.Account]):
     CORE_COLUMNS: Sequence[str] = [
         'Date',
         data.Metadata.human_name('entity'),
         'Description',
         'Original Amount',
         'Booked Amount',
+    ]
     ACCOUNT_COLUMNS: Dict[str, Sequence[str]] = collections.OrderedDict([
         ('Income', ['project', 'rt-id', 'receipt', 'income-type']),
         ('Expenses', ['project', 'rt-id', 'receipt', 'approval', 'expense-allocation']),
         ('Equity', ['rt-id']),
         ('Assets:Receivable', ['project', 'rt-id', 'invoice', 'approval', 'contract', 'purchase-order']),
         ('Liabilities:Payable', ['project', 'rt-id', 'invoice', 'approval', 'contract', 'purchase-order']),
         ('Assets:PayPal', ['rt-id', 'paypal-id', 'receipt', 'approval']),
         ('Assets', ['rt-id', 'receipt', 'approval', 'bank-statement']),
         ('Liabilities', ['rt-id', 'receipt', 'approval', 'bank-statement']),
     ])
     # Excel 2003 was limited to 65,536 rows per worksheet.
     # While we can probably count on all our users supporting more modern
     # formats (Excel 2007 supports over 1 million rows per worksheet),
     # keeping the default limit conservative seems good to avoid running into
     # other limits (like the number of hyperlinks per worksheet), plus just
     # better for human organization and readability.
     SHEET_SIZE = 65000
     def __init__(self,
                  start_date: datetime.date,
                  stop_date: datetime.date,
                  sheet_names: Optional[Sequence[str]]=None,
                  rt_wrapper: Optional[rtutil.RT]=None,
                  sheet_size: Optional[int]=None,
     ) -> None:
         if sheet_names is None:
             sheet_names = list(self.ACCOUNT_COLUMNS)
         if sheet_size is None:
             sheet_size = self.SHEET_SIZE
         super().__init__(rt_wrapper)
         self.date_range = ranges.DateRange(start_date, stop_date)
         self.required_sheet_names = sheet_names
         self.sheet_size = sheet_size
     def init_styles(self) -> None:
         super().init_styles()
         self.amount_column = self.column_style(1.2)
         self.default_column = self.column_style(1.5)
         self.column_styles: Mapping[str, Union[str, odf.style.Style]] = {
             'Date': '',
             'Description': self.column_style(2),
             'Original Amount': self.amount_column,
             'Booked Amount': self.amount_column,
             data.Metadata.human_name('project'): self.amount_column,
             data.Metadata.human_name('rt-id'): self.amount_column,
+        }
     @classmethod
     def _group_tally(
             cls,
             tally_by_account: PostTally,
             key: Callable[[data.Account], Optional[str]],
     ) -> Dict[str, PostTally]:
         retval: Dict[str, PostTally] = collections.defaultdict(list)
         for count, account in tally_by_account:
             item_key = key(account)
             if item_key is not None:
                 retval[item_key].append((count, account))
         return retval
     @classmethod
     def _split_sheet(
             cls,
             tally_by_account: PostTally,
             sheet_size: int,
             sheet_name: str,
     ) -> Iterator[str]:
         total = 0
         for index, (count, account) in enumerate(tally_by_account):
             total += count
             if total > sheet_size:
                 break
         else:
             # All the accounts fit in this sheet.
             yield sheet_name
             return
         if index == 0 and len(tally_by_account) == 1:
             # With one account, we can't split any further, so warn and stop.
             logger.warning(
                 "%s has %s rows, over size %s",
                 account, f'{count:,g}', f'{sheet_size:,g}',
+            )
             yield sheet_name
             return
         group_func = operator.methodcaller('root_part', sheet_name.count(':') + 2)
         maybe_split = cls._group_tally(tally_by_account[:index], group_func)
         must_split = cls._group_tally(tally_by_account[index:], group_func)
         for subkey, must_split_tally in sorted(must_split.items()):
@@ ... @@
                 assert not any(split_names)
         if maybe_split:
             yield sheet_name
     @classmethod
     def plan_sheets(
             cls,
             tally_by_account: Mapping[data.Account, int],
             base_sheets: Sequence[str],
             sheet_size: int,
     ) -> Sequence[str]:
         sorted_tally: PostTally = [
             (count, account)
             for account, count in tally_by_account.items()
+        ]
         sorted_tally.sort()
         split_tally = cls._group_tally(
             sorted_tally,
             operator.methodcaller('is_under', *base_sheets),
+        )
         return [
             sheet_name
             for key in base_sheets
             for sheet_name in cls._split_sheet(split_tally[key], sheet_size, key)
+        ]
     def section_key(self, row: data.Posting) -> data.Account:
         return row.account
     def start_sheet(self, sheet_name: str) -> None:
         self.use_sheet(sheet_name.replace(':', ' '))
         columns_key = data.Account(sheet_name).is_under(*self.ACCOUNT_COLUMNS)
         # columns_key must not be None because ACCOUNT_COLUMNS has an entry
         # for all five root accounts.
         assert columns_key is not None
         self.metadata_columns = self.ACCOUNT_COLUMNS[columns_key]
         self.sheet_columns: Sequence[str] = [
             *self.CORE_COLUMNS,
             *(data.Metadata.human_name(meta_key) for meta_key in self.metadata_columns),
+        ]
         for col_name in self.sheet_columns:
             self.sheet.addElement(odf.table.TableColumn(
                 stylename=self.column_styles.get(col_name, self.default_column),
             ))
         self.add_row(*(
             self.string_cell(col_name, stylename=self.style_bold)
             for col_name in self.sheet_columns
         ))
         self.lock_first_row()
     def _report_section_balance(self, key: data.Account, date_key: str) -> None:
         uses_opening = key.is_under('Assets', 'Equity', 'Liabilities')
         related = self.account_groups[key]
         if date_key == 'start':
             if not uses_opening:
                 return
             date = self.date_range.start
             balance = related.start_bal
             description = "Opening Balance"
         else:
             date = self.date_range.stop
             if uses_opening:
                 balance = related.stop_bal
                 description = "Ending Balance"
             else:
                 balance = related.period_bal
                 description = "Period Total"
         self.add_row(
             self.date_cell(date, stylename=self.merge_styles(
                 self.style_bold, self.style_date,
             )),
             odf.table.TableCell(),
             self.string_cell(description, stylename=self.style_bold),
             odf.table.TableCell(),
             self.balance_cell(self.norm_func(balance), stylename=self.style_bold),
+        )
     def start_section(self, key: data.Account) -> None:
         self.add_row()
         self.add_row(
             odf.table.TableCell(),
             self.string_cell(
                 f"{key} Ledger"
                 f" From {self.date_range.start.isoformat()}"
                 f" To {self.date_range.stop.isoformat()}",
                 stylename=self.style_bold,
                 numbercolumnsspanned=len(self.sheet_columns) - 1,
             ),
+        )
         self.norm_func = core.normalize_amount_func(key)
         self._report_section_balance(key, 'start')
     def end_section(self, key: data.Account) -> None:
         self._report_section_balance(key, 'stop')
     def write_row(self, row: data.Posting) -> None:
         if row.meta.date not in self.date_range:
             return
         elif row.cost is None:
         if row.cost is None:
             amount_cell = odf.table.TableCell()
         else:
             amount_cell = self.currency_cell(self.norm_func(row.units))
         self.add_row(
             self.date_cell(row.meta.date),
             self.string_cell(row.meta.get('entity') or ''),
             self.string_cell(row.meta.txn.narration),
             amount_cell,
             self.currency_cell(self.norm_func(row.at_cost())),
             *(self.meta_links_cell(row.meta.report_links(key))
               if key in data.LINK_METADATA
               else self.string_cell(row.meta.get(key, ''))
               for key in self.metadata_columns),
+        )
     def _combined_balance_row(self,
                               balance_accounts: Sequence[str],
                               attr_name: str,
     ) -> None:
         date = getattr(self.date_range, attr_name)
         balance_attrname = f'{attr_name}_bal'
         balance = -sum((
             getattr(related, balance_attrname)
             for account, related in self.account_groups.items()
             if account.is_under(*balance_accounts)
         ), core.MutableBalance())
         self.add_row(
             self.string_cell(
                 f"Balance as of {date.isoformat()}",
                 stylename=self.merge_styles(self.style_bold, self.style_endtext),
             ),
             self.balance_cell(balance, stylename=self.style_bold),
+        )
     def write_balance_sheet(self) -> None:
         balance_accounts = ['Equity', 'Income', 'Expenses']
         self.use_sheet("Balance")
         self.sheet.addElement(odf.table.TableColumn(stylename=self.column_style(3)))
         self.sheet.addElement(odf.table.TableColumn(stylename=self.column_style(1.5)))
         self.add_row(
             self.string_cell("Account", stylename=self.style_bold),
             self.string_cell("Balance", stylename=self.style_bold),
+        )
         self.lock_first_row()
         self.add_row()
         self.add_row(self.string_cell(
             f"Ledger From {self.date_range.start.isoformat()}"
             f" To {self.date_range.stop.isoformat()}",
             stylename=self.merge_styles(self.style_centertext, self.style_bold),
             numbercolumnsspanned=2,
         ))
         self.add_row()
         self._combined_balance_row(balance_accounts, 'start')
         for _, account in core.sort_and_filter_accounts(
                 self.account_groups, balance_accounts,
         ):
             balance = self.account_groups[account].period_bal
             if not balance.is_zero():
                 self.add_row(
                     self.string_cell(account, stylename=self.style_endtext),
                     self.balance_cell(-balance),
+                )
         self._combined_balance_row(balance_accounts, 'stop')
     def write(self, rows: Iterable[data.Posting]) -> None:
         AccountPostings.START_DATE = self.date_range.start
         self.account_groups = dict(AccountPostings.group_by_account(
         related_cls = core.PeriodPostings.with_start_date(self.date_range.start)
         self.account_groups = dict(related_cls.group_by_account(
             post for post in rows if post.meta.date < self.date_range.stop
         ))
         self.write_balance_sheet()
         tally_by_account_iter = (
-            (account, sum(1 for post in related if post.meta.date in self.date_range))
+            (account, len(related))
             for account, related in self.account_groups.items()
+        )
         tally_by_account = {
             account: count
             for account, count in tally_by_account_iter
             if count
+        }
         sheet_names = self.plan_sheets(
             tally_by_account, self.required_sheet_names, self.sheet_size,
+        )
         using_sheet_index = -1
         for sheet_index, account in core.sort_and_filter_accounts(
                 tally_by_account, sheet_names,
         ):
             while using_sheet_index < sheet_index:
                 using_sheet_index += 1
                 self.start_sheet(sheet_names[using_sheet_index])
             super().write(self.account_groups[account])
         for index in range(using_sheet_index + 1, len(sheet_names)):
             self.start_sheet(sheet_names[index])
 class ReturnFlag(enum.IntFlag):
     LOAD_ERRORS = 1
     NOTHING_TO_REPORT = 8
 def parse_arguments(arglist: Optional[Sequence[str]]=None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(prog=PROGNAME)
     cliutil.add_version_argument(parser)
     parser.add_argument(
         '--begin', '--start', '-b',
         dest='start_date',
         metavar='DATE',
         type=cliutil.date_arg,
         help="""Date to start reporting entries, inclusive, in YYYY-MM-DD format.
 The default is one year ago.
 """)
     parser.add_argument(
         '--end', '--stop', '-e',
         dest='stop_date',
         metavar='DATE',
         type=cliutil.date_arg,
         help="""Date to stop reporting entries, exclusive, in YYYY-MM-DD format.
 The default is a year after the start date, or 30 days from today if the start
 date was also not specified.
 """)
     parser.add_argument(
         '--account', '-a',
         dest='sheet_names',
         metavar='ACCOUNT',
         action='append',
         help="""Show this account in the report. You can specify this option
 multiple times. If not specified, the default set adapts to your search
 criteria.
 """)
     parser.add_argument(
         '--sheet-size', '--size',
         metavar='SIZE',
         type=int,
         default=LedgerODS.SHEET_SIZE,
         help="""Try to limit sheets to this many rows. The report will
 automatically create new sheets to make this happen. When that's not possible,
 it will issue a warning.
 """)
     parser.add_argument(
         '--output-file', '-O',
         metavar='PATH',
         type=Path,
         help="""Write the report to this file, or stdout when PATH is `-`.
 The default is stdout for the balance and outgoing reports, and a generated
 filename for other reports.
 """)
     cliutil.add_loglevel_argument(parser)
     parser.add_argument(
         'search_terms',
         metavar='FILTER',
         type=cliutil.SearchTerm.arg_parser('project', 'rt-id'),
         nargs=argparse.ZERO_OR_MORE,
         help="""Report on postings that match this criteria. The format is
 NAME=TERM. TERM is a link or word that must exist in a posting's NAME
 metadata to match. A single ticket number is a shortcut for
 `rt-id=rt:NUMBER`. Any other word is a shortcut for `project=TERM`.
 """)
     args = parser.parse_args(arglist)
     if args.sheet_names is None:
         if any(term.meta_key == 'project' for term in args.search_terms):
             args.sheet_names = ['Income', 'Expenses', 'Assets:Receivable', 'Liabilities:Payable']
         else:
             args.sheet_names = list(LedgerODS.ACCOUNT_COLUMNS)
     return args
 def diff_year(date: datetime.date, diff: int) -> datetime.date:
     new_year = date.year + diff
     try:
         return date.replace(year=new_year)

tests/test_reports_period_postings.py

➞

Show inline comments

 new file 100644
 """test_reports_period_postings - Unit tests for PeriodPostings"""
 # 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 operator
 from decimal import Decimal
 import pytest
 from . import testutil
 from conservancy_beancount import data
 from conservancy_beancount.reports import core
 def check_balance_attrs(balance_postings, start_usd, stop_usd):
     if start_usd:
         expected = {'USD': testutil.Amount(start_usd)}
         assert balance_postings.start_bal == expected
         assert balance_postings.begin_bal == expected
     else:
         assert balance_postings.start_bal.is_zero()
         assert balance_postings.begin_bal.is_zero()
     expected = {'USD': testutil.Amount(stop_usd)}
     assert balance_postings.stop_bal == expected
     assert balance_postings.end_bal == expected
     expected = {'USD': testutil.Amount(stop_usd - start_usd)}
     assert balance_postings.period_bal == expected
 @pytest.mark.parametrize('start_date,expect_start_bal', [
     (datetime.date(2019, 2, 1), 0),
     (datetime.date(2019, 4, 1), 30),
     (datetime.date(2019, 6, 1), 120),
 ])
 def test_balance_postings_attrs(start_date, expect_start_bal):
     entries = [testutil.Transaction(date=datetime.date(2019, n, 15), postings=[
         ('Income:Donations', -n * 10),
         ('Assets:Cash', n * 10),
     ]) for n in range(3, 7)]
     cls = core.PeriodPostings.with_start_date(start_date)
     actual = dict(cls.group_by_account(data.Posting.from_entries(entries)))
     assert len(actual) == 2
     check_balance_attrs(actual['Assets:Cash'], expect_start_bal, 180)
     check_balance_attrs(actual['Income:Donations'], -expect_start_bal, -180)
 @pytest.mark.parametrize('start_date,expect_count', [
     (datetime.date(2019, 2, 1), 4),
     (datetime.date(2019, 4, 1), 3),
     (datetime.date(2019, 6, 1), 1),
 ])
 def test_balance_postings_filter(start_date, expect_count):
     entries = [testutil.Transaction(date=datetime.date(2019, n, 15), postings=[
         ('Income:Donations', -n * 10),
         ('Assets:Cash', n * 10),
     ]) for n in range(3, 7)]
     cls = core.PeriodPostings.with_start_date(start_date)
     for _, related in cls.group_by_account(data.Posting.from_entries(entries)):
         assert len(related) == expect_count

0 comments (0 inline, 0 general)