Changeset - 8d7a2b1eea35
[Not reviewed]
0 2 0
Brett Smith - 4 years ago 2020-06-11 18:22:11
brettcsmith@brettcsmith.org
accrual: Add AccrualPostings.rt_id property.

This is like the existing attributes, but it only supports the outgoings
report, so don't build it at __init__ time.
2 files changed with 28 insertions and 0 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/reports/accrual.py
Show inline comments
...
 
@@ -229,64 +229,68 @@ class AccrualPostings(core.RelatedPostings):
 
                post.meta.get('invoice') or 'BlankInvoice'
 
            )
 
            post_entity = self.entity if entity_ok else (
 
                post.meta.get('entity') or 'BlankEntity'
 
            )
 
            groups[f'{post.account} {post_invoice} {post_entity}'].append(post)
 
        type_self = type(self)
 
        for group_key, posts in groups.items():
 
            yield group_key, type_self(posts, _can_own=True)
 

	
 
    def is_paid(self, default: Optional[bool]=None) -> Optional[bool]:
 
        if self.accrual_type is None:
 
            return default
 
        else:
 
            return self.end_balance.le_zero()
 

	
 
    def is_zero(self, default: Optional[bool]=None) -> Optional[bool]:
 
        if self.accrual_type is None:
 
            return default
 
        else:
 
            return self.end_balance.is_zero()
 

	
 
    def since_last_nonzero(self) -> 'AccrualPostings':
 
        for index, (post, balance) in enumerate(self.iter_with_balance()):
 
            if balance.is_zero():
 
                start_index = index
 
        try:
 
            empty = start_index == index
 
        except NameError:
 
            empty = True
 
        return self if empty else self[start_index + 1:]
 

	
 
    @property
 
    def rt_id(self) -> Union[str, None, Sentinel]:
 
        return self._single_item(self.first_meta_links('rt-id', None))
 

	
 

	
 
class BaseReport:
 
    def __init__(self, out_file: TextIO) -> None:
 
        self.out_file = out_file
 
        self.logger = logger.getChild(type(self).__name__)
 

	
 
    def _report(self, posts: AccrualPostings, index: int) -> Iterable[str]:
 
        raise NotImplementedError("BaseReport._report")
 

	
 
    def run(self, groups: PostGroups) -> None:
 
        for index, invoice in enumerate(groups):
 
            for line in self._report(groups[invoice], index):
 
                print(line, file=self.out_file)
 

	
 

	
 
class AgingODS(core.BaseODS[AccrualPostings, Optional[data.Account]]):
 
    COLUMNS = [
 
        'Date',
 
        'Entity',
 
        'Invoice Amount',
 
        'Booked Amount',
 
        'Project',
 
        'Ticket',
 
        'Invoice',
 
        'Approval',
 
        'Contract',
 
        'Purchase Order',
 
    ]
 
    COL_COUNT = len(COLUMNS)
 

	
 
    def __init__(self,
 
                 rt_client: rt.Rt,
tests/test_reports_accrual.py
Show inline comments
...
 
@@ -280,64 +280,88 @@ def test_accrual_postings_entity():
 
def test_accrual_postings_entities():
 
    txn = testutil.Transaction(postings=[
 
        (ACCOUNTS[0], 25, {'entity': 'Accruee'}),
 
        (ACCOUNTS[0], -15, {'entity': 'Payee15'}),
 
        (ACCOUNTS[0], -10, {'entity': 'Payee10'}),
 
    ])
 
    related = accrual.AccrualPostings(data.Posting.from_txn(txn))
 
    actual = related.entities()
 
    assert next(actual, None) == 'Accruee'
 
    assert set(actual) == {'Payee10', 'Payee15'}
 

	
 
def test_accrual_postings_entities_no_duplicates():
 
    txn = testutil.Transaction(postings=[
 
        (ACCOUNTS[0], 25, {'entity': 'Accruee'}),
 
        (ACCOUNTS[0], -15, {'entity': 'Accruee'}),
 
        (ACCOUNTS[0], -10, {'entity': 'Other'}),
 
    ])
 
    related = accrual.AccrualPostings(data.Posting.from_txn(txn))
 
    actual = related.entities()
 
    assert next(actual, None) == 'Accruee'
 
    assert next(actual, None) == 'Other'
 
    assert next(actual, None) is None
 

	
 
def test_accrual_postings_inconsistent_account():
 
    meta = {'invoice': 'invoice.pdf'}
 
    txn = testutil.Transaction(postings=[
 
        (acct_name, index, meta)
 
        for index, acct_name in enumerate(ACCOUNTS)
 
    ])
 
    related = accrual.AccrualPostings(data.Posting.from_txn(txn))
 
    assert related.account is related.INCONSISTENT
 

	
 
def test_accrual_postings_rt_id():
 
    txn = testutil.Transaction(postings=[
 
        (ACCOUNTS[0], 10, {'rt-id': 'rt:90'}),
 
        (ACCOUNTS[0], 10, {'rt-id': 'rt:90 rt:92'}),
 
        (ACCOUNTS[0], 10, {'rt-id': 'rt:90 rt:94 rt:92'}),
 
    ])
 
    related = accrual.AccrualPostings(data.Posting.from_txn(txn))
 
    assert related.rt_id == 'rt:90'
 

	
 
def test_accrual_postings_rt_id_inconsistent():
 
    txn = testutil.Transaction(postings=[
 
        (ACCOUNTS[0], 10, {'rt-id': 'rt:96'}),
 
        (ACCOUNTS[0], 10, {'rt-id': 'rt:98 rt:96'}),
 
    ])
 
    related = accrual.AccrualPostings(data.Posting.from_txn(txn))
 
    assert related.rt_id is related.INCONSISTENT
 

	
 
def test_accrual_postings_rt_id_none():
 
    txn = testutil.Transaction(postings=[
 
        (ACCOUNTS[0], 10),
 
    ])
 
    related = accrual.AccrualPostings(data.Posting.from_txn(txn))
 
    assert related.rt_id is None
 

	
 
@pytest.mark.parametrize('acct_name,invoice,day', testutil.combine_values(
 
    ACCOUNTS,
 
    ['FIXME', '', None, *testutil.NON_STRING_METADATA_VALUES],
 
    itertools.count(1),
 
))
 
def test_make_consistent_bad_invoice(acct_name, invoice, day):
 
    txn = testutil.Transaction(date=datetime.date(2019, 1, day), postings=[
 
        (acct_name, index * 10, {'invoice': invoice, 'entity': f'BadInvoice{day}'})
 
        for index in range(1, 4)
 
    ])
 
    related = accrual.AccrualPostings(data.Posting.from_txn(txn))
 
    consistent = dict(related.make_consistent())
 
    assert len(consistent) == 1
 
    key = next(iter(consistent))
 
    assert acct_name in key
 
    if invoice:
 
        assert str(invoice) in key
 
    actual = consistent[key]
 
    assert actual
 
    assert len(actual) == 3
 
    for act_post, exp_post in zip(actual, txn.postings):
 
        assert act_post.units == exp_post.units
 
        assert act_post.meta.get('invoice') == invoice
 

	
 
def test_make_consistent_across_accounts():
 
    invoice = 'Invoices/CrossAccount.pdf'
 
    txn = testutil.Transaction(date=datetime.date(2019, 2, 1), postings=[
 
        (acct_name, 100, {'invoice': invoice, 'entity': 'CrossAccount'})
 
        for acct_name in ACCOUNTS
 
    ])
 
    related = accrual.AccrualPostings(data.Posting.from_txn(txn))
 
    consistent = dict(related.make_consistent())
0 comments (0 inline, 0 general)