data: Move iter_posting into Posting class methods.
As I move into reporting code, having Posting.from_beancount() is handy, and then from_txn() might as well come along for the ride.
This commit is contained in:
parent
eb7f73e644
commit
14a87e792b
4 changed files with 52 additions and 28 deletions
|
@ -264,6 +264,27 @@ class Posting(BasePosting):
|
||||||
# If it did, this declaration would pass without issue.
|
# If it did, this declaration would pass without issue.
|
||||||
meta: Metadata # type:ignore[assignment]
|
meta: Metadata # 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) -> Iterable['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)
|
||||||
|
|
||||||
|
|
||||||
def balance_of(txn: Transaction,
|
def balance_of(txn: Transaction,
|
||||||
*preds: Callable[[Account], Optional[bool]],
|
*preds: Callable[[Account], Optional[bool]],
|
||||||
|
@ -277,7 +298,7 @@ def balance_of(txn: Transaction,
|
||||||
balance_of uses the "weight" of each posting, so the return value will
|
balance_of uses the "weight" of each posting, so the return value will
|
||||||
use the currency of the postings' cost when available.
|
use the currency of the postings' cost when available.
|
||||||
"""
|
"""
|
||||||
match_posts = [post for post in iter_postings(txn)
|
match_posts = [post for post in Posting.from_txn(txn)
|
||||||
if any(pred(post.account) for pred in preds)]
|
if any(pred(post.account) for pred in preds)]
|
||||||
number = decimal.Decimal(0)
|
number = decimal.Decimal(0)
|
||||||
if not match_posts:
|
if not match_posts:
|
||||||
|
@ -299,13 +320,3 @@ def is_opening_balance_txn(txn: Transaction) -> bool:
|
||||||
if not rest.currency:
|
if not rest.currency:
|
||||||
return False
|
return False
|
||||||
return abs(opening_equity.number + rest.number) < decimal.Decimal('.01')
|
return abs(opening_equity.number + rest.number) < decimal.Decimal('.01')
|
||||||
|
|
||||||
def iter_postings(txn: Transaction) -> Iterator[Posting]:
|
|
||||||
"""Yield an enhanced Posting object for every posting in the transaction"""
|
|
||||||
for index, source in enumerate(txn.postings):
|
|
||||||
yield Posting(
|
|
||||||
Account(source.account),
|
|
||||||
*source[1:5],
|
|
||||||
# see rationale above about Posting.meta
|
|
||||||
PostingMeta(txn, index, source), # type:ignore[arg-type]
|
|
||||||
)
|
|
||||||
|
|
|
@ -195,7 +195,7 @@ class _PostingHook(TransactionHook, metaclass=abc.ABCMeta):
|
||||||
|
|
||||||
def run(self, txn: Transaction) -> errormod.Iter:
|
def run(self, txn: Transaction) -> errormod.Iter:
|
||||||
if self._run_on_txn(txn):
|
if self._run_on_txn(txn):
|
||||||
for post in data.iter_postings(txn):
|
for post in data.Posting.from_txn(txn):
|
||||||
if self._run_on_post(txn, post):
|
if self._run_on_post(txn, post):
|
||||||
yield from self.post_run(txn, post)
|
yield from self.post_run(txn, post)
|
||||||
|
|
||||||
|
|
|
@ -60,7 +60,7 @@ class MetaEntity(core.TransactionHook):
|
||||||
txn_entity_ok = False
|
txn_entity_ok = False
|
||||||
if txn_entity_ok is False:
|
if txn_entity_ok is False:
|
||||||
yield errormod.InvalidMetadataError(txn, self.METADATA_KEY, txn_entity)
|
yield errormod.InvalidMetadataError(txn, self.METADATA_KEY, txn_entity)
|
||||||
for post in data.iter_postings(txn):
|
for post in data.Posting.from_txn(txn):
|
||||||
if not post.account.is_under(
|
if not post.account.is_under(
|
||||||
'Assets:Receivable',
|
'Assets:Receivable',
|
||||||
'Expenses',
|
'Expenses',
|
||||||
|
|
|
@ -1,4 +1,4 @@
|
||||||
"""Test data.iter_postings function"""
|
"""Test Posting methods"""
|
||||||
# Copyright © 2020 Brett Smith
|
# Copyright © 2020 Brett Smith
|
||||||
#
|
#
|
||||||
# This program is free software: you can redistribute it and/or modify
|
# This program is free software: you can redistribute it and/or modify
|
||||||
|
@ -27,20 +27,33 @@ def simple_txn(index=None, key=None):
|
||||||
('Income:Donations', -5, {'note': 'donation love', 'extra': 'Extra'}),
|
('Income:Donations', -5, {'note': 'donation love', 'extra': 'Extra'}),
|
||||||
])
|
])
|
||||||
|
|
||||||
def test_iter_postings(simple_txn):
|
def test_from_beancount():
|
||||||
for source, post in zip(simple_txn.postings, data.iter_postings(simple_txn)):
|
txn = testutil.Transaction(payee='Smith-Dakota', postings=[
|
||||||
|
('Income:Donations', -50),
|
||||||
|
('Assets:Cash', 50, {'receipt': 'cash-donation.pdf'}),
|
||||||
|
])
|
||||||
|
post = data.Posting.from_beancount(txn, 1)
|
||||||
|
# We don't just want to assert isinstance(post.attr, data.SomeClass);
|
||||||
|
# we also want to double-check that attributes were instantiated correctly.
|
||||||
|
assert post.account.is_under('Assets:Cash')
|
||||||
|
assert post.meta['receipt'] == 'cash-donation.pdf'
|
||||||
|
assert post.meta['entity'] == 'Smith-Dakota'
|
||||||
|
assert post.meta.date == testutil.FY_MID_DATE
|
||||||
|
|
||||||
|
def test_setting_metadata_propagates_to_source(simple_txn):
|
||||||
|
src_post = simple_txn.postings[1]
|
||||||
|
post = data.Posting.from_beancount(simple_txn, 1)
|
||||||
|
post.meta['edited'] = 'yes'
|
||||||
|
assert src_post.meta['edited'] == 'yes'
|
||||||
|
assert not isinstance(src_post.meta, data.PostingMeta)
|
||||||
|
|
||||||
|
def test_deleting_metadata_propagates_to_source(simple_txn):
|
||||||
|
post = data.Posting.from_beancount(simple_txn, 1)
|
||||||
|
del post.meta['extra']
|
||||||
|
assert 'extra' not in simple_txn.postings[1].meta
|
||||||
|
|
||||||
|
def test_from_txn(simple_txn):
|
||||||
|
for source, post in zip(simple_txn.postings, data.Posting.from_txn(simple_txn)):
|
||||||
assert all(source[x] == post[x] for x in range(len(source) - 1))
|
assert all(source[x] == post[x] for x in range(len(source) - 1))
|
||||||
assert isinstance(post.account, data.Account)
|
assert isinstance(post.account, data.Account)
|
||||||
assert post.meta['note'] # Only works with PostingMeta
|
assert post.meta['note'] # Only works with PostingMeta
|
||||||
|
|
||||||
def test_setting_metadata_propagates_to_source(simple_txn):
|
|
||||||
for index, post in enumerate(data.iter_postings(simple_txn)):
|
|
||||||
post.meta['edited'] = str(index)
|
|
||||||
for index, post in enumerate(simple_txn.postings):
|
|
||||||
assert post.meta['edited'] == str(index)
|
|
||||||
assert not isinstance(post.meta, data.PostingMeta)
|
|
||||||
|
|
||||||
def test_deleting_metadata_propagates_to_source(simple_txn):
|
|
||||||
posts = list(data.iter_postings(simple_txn))
|
|
||||||
del posts[1].meta['extra']
|
|
||||||
assert 'extra' not in simple_txn.postings[1].meta
|
|
Loading…
Reference in a new issue