conservancy_beancount/tests/test_reports_related_postings.py

"""test_reports_related_postings - Unit tests for RelatedPostings"""
# 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 itertools

from decimal import Decimal

import pytest

from . import testutil

from conservancy_beancount import data
from conservancy_beancount.reports import core

def accruals_and_payments(acct, src_acct, dst_acct, start_date, *amounts):
    dates = testutil.date_seq(start_date)
    for amt, currency in amounts:
        post_meta = {'metanumber': amt, 'metacurrency': currency}
        yield testutil.Transaction(date=next(dates), postings=[
            (acct, amt, currency, post_meta),
            (dst_acct if amt < 0 else src_acct, -amt, currency, post_meta),
        ])

@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'),
    ))

@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_initialize_with_list(credit_card_cycle):
    related = core.RelatedPostings(credit_card_cycle[0].postings)
    assert len(related) == 2

def test_initialize_with_iterable(two_accruals_three_payments):
    related = core.RelatedPostings(
        post for txn in two_accruals_three_payments
        for post in txn.postings
        if post.account == 'Assets:Receivable:Accounts'
    )
    assert len(related) == 5

def test_balance_empty():
    balance = core.RelatedPostings().balance()
    assert not balance
    assert balance.is_zero()

def test_balance_credit_card(credit_card_cycle):
    related = core.RelatedPostings()
    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_clear_after_add():
    related = core.RelatedPostings()
    related.add(testutil.Posting('Income:Donations', -10))
    assert related.balance()
    related.clear()
    assert not related.balance()

def test_clear_after_initialization():
    related = core.RelatedPostings([
        testutil.Posting('Income:Donations', -12),
    ])
    assert related.balance()
    related.clear()
    assert not related.balance()

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()
    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)

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()

def test_meta_values_empty():
    related = core.RelatedPostings()
    assert related.meta_values('key') == set()

def test_meta_values_no_match():
    related = core.RelatedPostings()
    related.add(testutil.Posting('Income:Donations', -1, metakey='metavalue'))
    assert related.meta_values('key') == {None}

def test_meta_values_no_match_default_given():
    related = core.RelatedPostings()
    related.add(testutil.Posting('Income:Donations', -1, metakey='metavalue'))
    assert related.meta_values('key', '') == {''}

def test_meta_values_one_match():
    related = core.RelatedPostings()
    related.add(testutil.Posting('Income:Donations', -1, key='metavalue'))
    assert related.meta_values('key') == {'metavalue'}

def test_meta_values_some_match():
    related = core.RelatedPostings()
    related.add(testutil.Posting('Income:Donations', -1, key='1'))
    related.add(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()
    related.add(testutil.Posting('Income:Donations', -1, key='1'))
    related.add(testutil.Posting('Income:Donations', -2, metakey='2'))
    assert related.meta_values('key', '') == {'1', ''}

def test_meta_values_all_match():
    related = core.RelatedPostings()
    related.add(testutil.Posting('Income:Donations', -1, key='1'))
    related.add(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()
    related.add(testutil.Posting('Income:Donations', -1, key='1'))
    related.add(testutil.Posting('Income:Donations', -2, key='1'))
    assert related.meta_values('key') == {'1'}

def test_meta_values_all_match_default_given():
    related = core.RelatedPostings()
    related.add(testutil.Posting('Income:Donations', -1, key='1'))
    related.add(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',
    }
    related = core.RelatedPostings()
    for index, value in enumerate(expected):
        related.add(testutil.Posting('Income:Donations', -index, key=value))
    assert related.meta_values('key') == expected

@pytest.mark.parametrize('count', range(3))
def test_all_meta_links_zero(count):
    postings = (
        testutil.Posting('Income:Donations', -n, testkey=str(n))
        for n in range(count)
    )
    related = core.RelatedPostings(
        post._replace(meta=data.Metadata(post.meta))
        for post in postings
    )
    assert related.all_meta_links('approval') == set()

def test_all_meta_links_singletons():
    postings = (
        testutil.Posting('Income:Donations', -10, statement=value)
        for value in itertools.chain(
            testutil.NON_LINK_METADATA_STRINGS,
            testutil.LINK_METADATA_STRINGS,
            testutil.NON_STRING_METADATA_VALUES,
        ))
    related = core.RelatedPostings(
        post._replace(meta=data.Metadata(post.meta))
        for post in postings
    )
    assert related.all_meta_links('statement') == testutil.LINK_METADATA_STRINGS

def test_all_meta_links_multiples():
    postings = (
        testutil.Posting('Income:Donations', -10, approval=' '.join(value))
        for value in itertools.permutations(testutil.LINK_METADATA_STRINGS, 2)
    )
    related = core.RelatedPostings(
        post._replace(meta=data.Metadata(post.meta))
        for post in postings
    )
    assert related.all_meta_links('approval') == testutil.LINK_METADATA_STRINGS

def test_group_by_meta_zero():
    assert len(core.RelatedPostings.group_by_meta([], 'metacurrency')) == 0

def test_group_by_meta_key_error():
    # Make sure the return value doesn't act like a defaultdict.
    with pytest.raises(KeyError):
        core.RelatedPostings.group_by_meta([], 'metakey')['metavalue']

def test_group_by_meta_one(credit_card_cycle):
    posting = next(post for post in data.Posting.from_entries(credit_card_cycle)
                   if post.account.is_credit_card())
    actual = core.RelatedPostings.group_by_meta([posting], 'metacurrency')
    assert set(actual) == {'USD'}

def test_group_by_meta_many(two_accruals_three_payments):
    postings = [post for post in data.Posting.from_entries(two_accruals_three_payments)
                if post.account == 'Assets:Receivable:Accounts']
    actual = core.RelatedPostings.group_by_meta(postings, 'metacurrency')
    assert set(actual) == {'USD', 'EUR'}
    for key, group in actual.items():
        assert 2 <= len(group) <= 3
        assert group.balance().is_zero()

def test_group_by_meta_many_single_posts(two_accruals_three_payments):
    postings = [post for post in data.Posting.from_entries(two_accruals_three_payments)
                if post.account == 'Assets:Receivable:Accounts']
    actual = core.RelatedPostings.group_by_meta(postings, 'metanumber')
    assert set(actual) == {post.units.number for post in postings}
    assert len(actual) == len(postings)