NPO-Accounting/conservancy_beancount Changeset - ccc3a829da9e

Changeset - ccc3a829da9e

Parent rev.

Child rev.

[Not reviewed]

0 2 0

Brett Smith - 4 years ago 2020-06-15 14:38:26
brettcsmith@brettcsmith.org

reports: Add RelatedPostings.balance_at_cost_by_date() method.

2 files changed with 40 insertions and 0 deletions:

conservancy_beancount/reports/core.py

tests/test_reports_related_postings.py

0 comments (0 inline, 0 general)

conservancy_beancount/reports/core.py

➞

Show inline comments

@@ ... / @@ -307,128 +307,136 @@ class RelatedPostings(Sequence[data.Posting]): @@
         """
         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]:
         for post in self:
             try:
                 yield from post.meta.get_links(key)
             except TypeError:
                 pass
     def all_meta_links(self, key: MetaKey) -> Iterator[str]:
         return filters.iter_unique(self._all_meta_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 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)

tests/test_reports_related_postings.py

➞

Show inline comments

@@ ... / @@ -136,128 +136,160 @@ def check_iter_with_balance(entries): @@
 def test_iter_with_balance_empty():
     assert not list(core.RelatedPostings().iter_with_balance())
 def test_iter_with_balance_credit_card(credit_card_cycle):
     check_iter_with_balance(credit_card_cycle)
 def test_iter_with_balance_two_acccruals(two_accruals_three_payments):
     check_iter_with_balance(two_accruals_three_payments)
 def test_balance_at_cost_mixed():
     txn = testutil.Transaction(postings=[
         ('Expenses:Other', '22'),
         ('Expenses:Other', '30', 'EUR', ('1.1',)),
         ('Expenses:Other', '40', 'EUR'),
         ('Expenses:Other', '50', 'USD', ('1.1', 'EUR')),
     ])
     related = core.RelatedPostings(data.Posting.from_txn(txn))
     balance = related.balance_at_cost()
     amounts = set(balance.values())
     assert amounts == {testutil.Amount(55, 'USD'), testutil.Amount(95, 'EUR')}
 def test_balance_at_single_currency_cost():
     txn = testutil.Transaction(postings=[
         ('Expenses:Other', '22'),
         ('Expenses:Other', '30', 'EUR', ('1.1',)),
         ('Expenses:Other', '40', 'GBP', ('1.1',)),
     ])
     related = core.RelatedPostings(data.Posting.from_txn(txn))
     balance = related.balance_at_cost()
     amounts = set(balance.values())
     assert amounts == {testutil.Amount(99)}
 def test_balance_at_cost_zeroed_out():
     txn = testutil.Transaction(postings=[
         ('Income:Other', '-22'),
         ('Assets:Receivable:Accounts', '20', 'EUR', ('1.1',)),
     ])
     related = core.RelatedPostings(data.Posting.from_txn(txn))
     balance = related.balance_at_cost()
     assert balance.is_zero()
 def test_balance_at_cost_singleton():
     txn = testutil.Transaction(postings=[
         ('Assets:Receivable:Accounts', '20', 'EUR', ('1.1',)),
     ])
     related = core.RelatedPostings(data.Posting.from_txn(txn))
     balance = related.balance_at_cost()
     amounts = set(balance.values())
     assert amounts == {testutil.Amount(22)}
 def test_balance_at_cost_singleton_without_cost():
     txn = testutil.Transaction(postings=[
         ('Assets:Receivable:Accounts', '20'),
     ])
     related = core.RelatedPostings(data.Posting.from_txn(txn))
     balance = related.balance_at_cost()
     amounts = set(balance.values())
     assert amounts == {testutil.Amount(20)}
 def test_balance_at_cost_empty():
     related = core.RelatedPostings()
     balance = related.balance_at_cost()
     assert balance.is_zero()
 @pytest.mark.parametrize('date,expected', [
     (testutil.FY_MID_DATE - datetime.timedelta(days=1), 0),
     (testutil.FY_MID_DATE, 0),
     (testutil.FY_MID_DATE + datetime.timedelta(days=1), 25),
     (testutil.FY_MID_DATE + datetime.timedelta(days=2), 70),
     (testutil.FY_MID_DATE + datetime.timedelta(days=3), 135),
     (testutil.FY_MID_DATE + datetime.timedelta(days=4), 135),
 ])
 def test_balance_at_cost_by_date(date, expected):
     dates = testutil.date_seq()
     jpy_cost = ('0.01', 'USD')
     entries = [
         testutil.Transaction(date=next(dates), postings=[
             ('Assets:Cash', 1000, 'JPY', jpy_cost),
             ('Assets:Cash', 15),
         ]),
         testutil.Transaction(date=next(dates), postings=[
             ('Assets:Cash', 2000, 'JPY', jpy_cost),
             ('Assets:Cash', 25),
         ]),
         testutil.Transaction(date=next(dates), postings=[
             ('Assets:Cash', 3000, 'JPY', jpy_cost),
             ('Assets:Cash', 35),
         ]),
+    ]
     related = core.RelatedPostings(data.Posting.from_entries(entries))
     actual = related.balance_at_cost_by_date(date)
     if not expected:
         assert actual.is_zero()
     else:
         assert actual == {'USD': testutil.Amount(expected)}
 def test_meta_values_empty():
     related = core.RelatedPostings()
     assert related.meta_values('key') == set()
 def test_meta_values_no_match():
     related = core.RelatedPostings([
         testutil.Posting('Income:Donations', -1, metakey='metavalue'),
     ])
     assert related.meta_values('key') == {None}
 def test_meta_values_no_match_default_given():
     related = core.RelatedPostings([
         testutil.Posting('Income:Donations', -1, metakey='metavalue'),
     ])
     assert related.meta_values('key', '') == {''}
 def test_meta_values_one_match():
     related = core.RelatedPostings([
         testutil.Posting('Income:Donations', -1, key='metavalue'),
     ])
     assert related.meta_values('key') == {'metavalue'}
 def test_meta_values_some_match():
     related = core.RelatedPostings([
         testutil.Posting('Income:Donations', -1, key='1'),
         testutil.Posting('Income:Donations', -2, metakey='2'),
     ])
     assert related.meta_values('key') == {'1', None}
 def test_meta_values_some_match_default_given():
     related = core.RelatedPostings([
         testutil.Posting('Income:Donations', -1, key='1'),
         testutil.Posting('Income:Donations', -2, metakey='2'),
     ])
     assert related.meta_values('key', '') == {'1', ''}
 def test_meta_values_all_match():
     related = core.RelatedPostings([
         testutil.Posting('Income:Donations', -1, key='1'),
         testutil.Posting('Income:Donations', -2, key='2'),
     ])
     assert related.meta_values('key') == {'1', '2'}
 def test_meta_values_all_match_one_value():
     related = core.RelatedPostings([
         testutil.Posting('Income:Donations', -1, key='1'),
         testutil.Posting('Income:Donations', -2, key='1'),
     ])
     assert related.meta_values('key') == {'1'}
 def test_meta_values_all_match_default_given():
     related = core.RelatedPostings([
         testutil.Posting('Income:Donations', -1, key='1'),
         testutil.Posting('Income:Donations', -2, key='2'),
     ])
     assert related.meta_values('key', '') == {'1', '2'}
 def test_meta_values_many_types():
     expected = {
         datetime.date(2020, 4, 1),
         Decimal(42),
         testutil.Amount(5),
         'rt:42',
+    }

0 comments (0 inline, 0 general)