Changeset - aef00ce83f20
[Not reviewed]
0 3 0
Brett Smith - 4 years ago 2020-05-30 14:35:29
brettcsmith@brettcsmith.org
accrual: Check the consistency of accruals' cost.
3 files changed with 34 insertions and 10 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/reports/accrual.py
Show inline comments
...
 
@@ -58,24 +58,25 @@ option::
 
# 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 argparse
 
import collections
 
import datetime
 
import enum
 
import logging
 
import re
 
import sys
 

	
 
from typing import (
 
    Any,
 
    Callable,
 
    Dict,
 
    Iterable,
 
    Iterator,
 
    Mapping,
 
    NamedTuple,
 
    Optional,
 
    Sequence,
 
    Set,
 
    TextIO,
 
    Tuple,
 
)
...
 
@@ -319,37 +320,46 @@ class SearchTerm(NamedTuple):
 
            pattern = r'(?:^|\s){}(?:\s|$)'.format(re.escape(raw_link))
 
        else:
 
            ticket_id, attachment_id = rt_ids
 
            if key is None:
 
                key = 'rt-id' if attachment_id is None else 'invoice'
 
            pattern = rtutil.RT.metadata_regexp(
 
                ticket_id,
 
                attachment_id,
 
                first_link_only=key == 'rt-id' and attachment_id is None,
 
            )
 
        return cls(key, pattern)
 

	
 
def _consistency_check_one_thing(
 
        key: MetaValue,
 
        related: core.RelatedPostings,
 
        get_name: str,
 
        get_func: Callable[[data.Posting], Any],
 
) -> Iterable[Error]:
 
    values = {get_func(post) for post in related}
 
    if len(values) != 1:
 
        for post in related:
 
            errmsg = f'inconsistent {get_name} for invoice {key}: {get_func(post)}'
 
            yield Error(post.meta, errmsg, post.meta.txn)
 

	
 
def consistency_check(groups: PostGroups) -> Iterable[Error]:
 
    errfmt = 'inconsistent {} for invoice {}: {{}}'
 
    for key, related in groups.items():
 
        yield from _consistency_check_one_thing(
 
            key, related, 'cost', lambda post: post.cost,
 
        )
 
        for checked_meta in ['contract', 'entity', 'purchase-order']:
 
            meta_values = related.meta_values(checked_meta)
 
            if len(meta_values) != 1:
 
                errmsg = f'inconsistent {checked_meta} for invoice {key}'
 
                for post in related:
 
                    yield Error(
 
                        post.meta,
 
                        f'{errmsg}: {post.meta.get(checked_meta)}',
 
                        post.meta.txn,
 
                    )
 
            yield from _consistency_check_one_thing(
 
                key, related, checked_meta, lambda post: post.meta.get(checked_meta),
 
            )
 

	
 
def filter_search(postings: Iterable[data.Posting],
 
                  search_terms: Iterable[SearchTerm],
 
) -> Iterable[data.Posting]:
 
    accounts = tuple(AccrualAccount.account_names())
 
    postings = (post for post in postings if post.account.is_under(*accounts))
 
    for meta_key, pattern in search_terms:
 
        postings = filters.filter_meta_match(postings, meta_key, re.compile(pattern))
 
    return postings
 

	
 
def parse_arguments(arglist: Optional[Sequence[str]]=None) -> argparse.Namespace:
 
    parser = argparse.ArgumentParser(prog=PROGNAME)
setup.py
Show inline comments
 
#!/usr/bin/env python3
 

	
 
from setuptools import setup
 

	
 
setup(
 
    name='conservancy_beancount',
 
    description="Plugin, library, and reports for reading Conservancy's books",
 
    version='1.0.8',
 
    version='1.0.9',
 
    author='Software Freedom Conservancy',
 
    author_email='info@sfconservancy.org',
 
    license='GNU AGPLv3+',
 

	
 
    install_requires=[
 
        'babel>=2.6',  # Debian:python3-babel
 
        'beancount>=2.2',  # Debian:beancount
 
        'PyYAML>=3.0',  # Debian:python3-yaml
 
        'regex',  # Debian:python3-regex
 
        'rt>=2.0',
 
    ],
 
    setup_requires=[
tests/test_reports_accrual.py
Show inline comments
...
 
@@ -241,24 +241,38 @@ def test_consistency_check_when_inconsistent(meta_key, account):
 
        (account, -100, {'invoice': invoice, meta_key: 'debit', 'lineno': 2}),
 
    ])
 
    related = core.RelatedPostings(data.Posting.from_txn(txn))
 
    errors = list(accrual.consistency_check({invoice: related}))
 
    for exp_lineno, (actual, exp_msg) in enumerate(itertools.zip_longest(errors, [
 
            f'inconsistent {meta_key} for invoice {invoice}: credit',
 
            f'inconsistent {meta_key} for invoice {invoice}: debit',
 
    ]), 1):
 
        assert actual.message == exp_msg
 
        assert actual.entry is txn
 
        assert actual.source.get('lineno') == exp_lineno
 

	
 
def test_consistency_check_cost():
 
    account = ACCOUNTS[0]
 
    invoice = 'test-cost-invoice'
 
    txn = testutil.Transaction(postings=[
 
        (account, 100, 'EUR', ('1.1251', 'USD'), {'invoice': invoice, 'lineno': 1}),
 
        (account, -100, 'EUR', ('1.125', 'USD'), {'invoice': invoice, 'lineno': 2}),
 
    ])
 
    related = core.RelatedPostings(data.Posting.from_txn(txn))
 
    errors = list(accrual.consistency_check({invoice: related}))
 
    for post, err in itertools.zip_longest(txn.postings, errors):
 
        assert err.message == f'inconsistent cost for invoice {invoice}: {post.cost}'
 
        assert err.entry is txn
 
        assert err.source.get('lineno') == post.meta['lineno']
 

	
 
def check_output(output, expect_patterns):
 
    output.seek(0)
 
    testutil.check_lines_match(iter(output), expect_patterns)
 

	
 
def run_outgoing(invoice, postings, rt_client=None):
 
    if rt_client is None:
 
        rt_client = RTClient()
 
    if not isinstance(postings, core.RelatedPostings):
 
        postings = relate_accruals_by_meta(postings, invoice)
 
    output = io.StringIO()
 
    report = accrual.OutgoingReport(rt_client, output)
 
    report.run({invoice: postings})
0 comments (0 inline, 0 general)