Changeset - b28646aa12e1
[Not reviewed]
0 2 0
Brett Smith - 4 years ago 2020-04-12 19:18:19
brettcsmith@brettcsmith.org
core.RelatedPostings: Add iter_with_balance method.

payment-report and accrual-report query to find the last date a
series of postings had a non/zero balance. This method is a good
building block for that.
2 files changed with 87 insertions and 49 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/reports/core.py
Show inline comments
...
 
@@ -121,8 +121,16 @@ class RelatedPostings(Sequence[data.Posting]):
 
    def add(self, post: data.Posting) -> None:
 
        self._postings.append(post)
 

	
 
    def balance(self) -> Balance:
 
    def iter_with_balance(self) -> Iterable[Tuple[data.Posting, Balance]]:
 
        balance = MutableBalance()
 
        for post in self:
 
            balance.add_amount(post.units)
 
        return balance
 
            yield post, balance
 

	
 
    def balance(self) -> Balance:
 
        for _, balance in self.iter_with_balance():
 
            pass
 
        try:
 
            return balance
 
        except NameError:
 
            return Balance()
tests/test_reports_related_postings.py
Show inline comments
...
 
@@ -14,6 +14,7 @@
 
# 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 itertools
 

	
...
 
@@ -26,58 +27,87 @@ from . import testutil
 
from conservancy_beancount import data
 
from conservancy_beancount.reports import core
 

	
 
_day_counter = itertools.count(1)
 
def next_date():
 
    return testutil.FY_MID_DATE + datetime.timedelta(next(_day_counter))
 
def date_seq(date=testutil.FY_MID_DATE, step=1):
 
    while True:
 
        yield date
 
        date = date + datetime.timedelta(days=step)
 

	
 
def txn_pair(acct, src_acct, dst_acct, amount, date=None, txn_meta={}, post_meta={}):
 
    if date is None:
 
        date = next_date()
 
    src_txn = testutil.Transaction(date=date, **txn_meta, postings=[
 
        (acct, amount, post_meta.copy()),
 
        (src_acct, -amount),
 
    ])
 
    dst_date = date + datetime.timedelta(days=1)
 
    dst_txn = testutil.Transaction(date=dst_date, **txn_meta, postings=[
 
        (acct, -amount, post_meta.copy()),
 
        (dst_acct, amount),
 
    ])
 
    return (src_txn, dst_txn)
 
def accruals_and_payments(acct, src_acct, dst_acct, start_date, *amounts):
 
    dates = date_seq(start_date)
 
    for amt, currency in amounts:
 
        yield testutil.Transaction(date=next(dates), postings=[
 
            (acct, amt, currency),
 
            (dst_acct if amt < 0 else src_acct, -amt, currency),
 
        ])
 

	
 
def donation(amount, currency='USD', date=None, other_acct='Assets:Cash', **meta):
 
    if date is None:
 
        date = next_date()
 
    return testutil.Transaction(date=date, postings=[
 
        ('Income:Donations', -amount, currency, meta),
 
        (other_acct, amount, currency),
 
    ])
 
@pytest.fixture
 
def credit_card_cycle():
 
    return list(accruals_and_payments(
 
        'Liabilities:CreditCard',
 
        'Assets:Checking',
 
        'Expenses:Other',
 
        datetime.date(2020, 4, 1),
 
        (-110, 'USD'),
 
        (110, 'USD'),
 
        (-120, 'USD'),
 
        (120, 'USD'),
 
    ))
 

	
 
def test_balance():
 
    related = core.RelatedPostings()
 
    related.add(data.Posting.from_beancount(donation(10), 0))
 
    assert related.balance() == testutil.balance_map(USD=-10)
 
    related.add(data.Posting.from_beancount(donation(15), 0))
 
    assert related.balance() == testutil.balance_map(USD=-25)
 
    related.add(data.Posting.from_beancount(donation(20), 0))
 
    assert related.balance() == testutil.balance_map(USD=-45)
 
@pytest.fixture
 
def two_accruals_three_payments():
 
    return list(accruals_and_payments(
 
        'Assets:Receivable:Accounts',
 
        'Income:Donations',
 
        'Assets:Checking',
 
        datetime.date(2020, 4, 10),
 
        (440, 'USD'),
 
        (-230, 'USD'),
 
        (550, 'EUR'),
 
        (-210, 'USD'),
 
        (-550, 'EUR'),
 
    ))
 

	
 
def test_balance_zero():
 
    related = core.RelatedPostings()
 
    related.add(data.Posting.from_beancount(donation(10), 0))
 
    related.add(data.Posting.from_beancount(donation(-10), 0))
 
    assert related.balance().is_zero()
 
def test_balance_empty():
 
    balance = core.RelatedPostings().balance()
 
    assert not balance
 
    assert balance.is_zero()
 

	
 
def test_balance_multiple_currencies():
 
def test_balance_credit_card(credit_card_cycle):
 
    related = core.RelatedPostings()
 
    related.add(data.Posting.from_beancount(donation(10, 'GBP'), 0))
 
    related.add(data.Posting.from_beancount(donation(15, 'GBP'), 0))
 
    related.add(data.Posting.from_beancount(donation(20, 'EUR'), 0))
 
    related.add(data.Posting.from_beancount(donation(25, 'EUR'), 0))
 
    assert related.balance() == testutil.balance_map(EUR=-45, GBP=-25)
 
    assert related.balance() == testutil.balance_map()
 
    expected = Decimal()
 
    for txn in credit_card_cycle:
 
        post = txn.postings[0]
 
        expected += post.units.number
 
        related.add(post)
 
        assert related.balance() == testutil.balance_map(USD=expected)
 
    assert expected == 0
 

	
 
def test_balance_multiple_currencies_one_zero():
 
def check_iter_with_balance(entries):
 
    expect_posts = [txn.postings[0] for txn in entries]
 
    expect_balances = []
 
    balance_tally = collections.defaultdict(Decimal)
 
    related = core.RelatedPostings()
 
    related.add(data.Posting.from_beancount(donation(10, 'EUR'), 0))
 
    related.add(data.Posting.from_beancount(donation(15, 'USD'), 0))
 
    related.add(data.Posting.from_beancount(donation(-10, 'EUR'), 0))
 
    assert related.balance() == testutil.balance_map(EUR=0, USD=-15)
 
    for post in expect_posts:
 
        number, currency = post.units
 
        balance_tally[currency] += number
 
        expect_balances.append(testutil.balance_map(balance_tally.items()))
 
        related.add(post)
 
    for (post, balance), exp_post, exp_balance in zip(
 
            related.iter_with_balance(),
 
            expect_posts,
 
            expect_balances,
 
    ):
 
        assert post is exp_post
 
        assert balance == exp_balance
 
    assert post is expect_posts[-1]
 
    assert related.balance() == expect_balances[-1]
 

	
 
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)
0 comments (0 inline, 0 general)