NPO-Accounting/conservancy_beancount
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

filters: Add filter_for_rt_id function.

Browse source
This commit is contained in:
Brett Smith 2020-04-22 10:59:12 -04:00
parent 4420873c96
commit 54a1bc4600
2 changed files with 50 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
conservancy_beancount/filters.py
View file

@ -46,3 +46,12 @@ def filter_meta_match(postings: Postings, key: MetaKey, regexp: Regexp) -> Posti
yield post yield post
except (KeyError, TypeError): except (KeyError, TypeError):
pass pass
def filter_for_rt_id(postings: Postings, ticket_id: Union[int, str]) -> Postings:
"""Filter postings with a primary RT ticket
This functions yields postings where the *first* rt-id matches the given
ticket number.
"""
regexp = r'^\s*rt:(?://ticket/)?{}\b'.format(re.escape(str(ticket_id)))
return filter_meta_match(postings, 'rt-id', regexp)

41
tests/test_filters.py
View file

@ -95,3 +95,44 @@ def test_filter_meta_match(cc_txn_pair, key, regexp, expected_indexes):
postings = data.Posting.from_entries(cc_txn_pair) postings = data.Posting.from_entries(cc_txn_pair)
actual = filters.filter_meta_match(postings, key, regexp) actual = filters.filter_meta_match(postings, key, regexp)
check_filter(actual, cc_txn_pair, expected_indexes) check_filter(actual, cc_txn_pair, expected_indexes)
@pytest.mark.parametrize('ticket_id,expected_indexes', [
(550, range(5)),
('550', range(5)),
(55, ()),
('55', ()),
(50, ()),
('.', ()),
])
def test_filter_for_rt_id(cc_txn_pair, ticket_id, expected_indexes):
postings = data.Posting.from_entries(cc_txn_pair)
actual = filters.filter_for_rt_id(postings, ticket_id)
check_filter(actual, cc_txn_pair, expected_indexes)
@pytest.mark.parametrize('rt_id', [
'rt:450/',
'rt:450/678',
' rt:450 rt:540',
'rt://ticket/450',
'rt://ticket/450/',
'rt://ticket/450/678',
' rt://ticket/450',
'rt://ticket/450 rt://ticket/540',
])
def test_filter_for_rt_id_syntax_variations(rt_id):
entries = [testutil.Transaction(**{'rt-id': rt_id}, postings=[
('Income:Donations', -10),
('Assets:Cash', 10),
])]
postings = data.Posting.from_entries(entries)
actual = filters.filter_for_rt_id(postings, 450)
check_filter(actual, entries, range(2))
def test_filter_for_rt_id_uses_first_link_only():
entries = [testutil.Transaction(postings=[
('Income:Donations', -10, {'rt-id': 'rt:1 rt:350'}),
('Assets:Cash', 10, {'rt-id': 'rt://ticket/2 rt://ticket/350'}),
])]
postings = data.Posting.from_entries(entries)
actual = filters.filter_for_rt_id(postings, 350)
check_filter(actual, entries, ()),