filters: Add filter_meta_equal function.

conservancy_beancount/filters.py

@@ -0,0 +1,35 @@
+"""filters.py - Common filters for postings"""
+# 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/>.
+
+from . import data
+
+from typing import (
+    Iterable,
+)
+from .beancount_types import (
+    MetaKey,
+    MetaValue,
+)
+
+Postings = Iterable[data.Posting]
+
+def filter_meta_equal(postings: Postings, key: MetaKey, value: MetaValue) -> Postings:
+    for post in postings:
+        try:
+            if post.meta[key] == value:
+                yield post
+        except KeyError:
+            pass

tests/test_filters.py

@@ -0,0 +1,77 @@
+"""test_filters - Unit tests for filter functions"""
+# 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 itertools
+
+import pytest
+
+from . import testutil
+
+from conservancy_beancount import data
+from conservancy_beancount import filters
+
+MISSING_POSTING = testutil.Posting('<Missing Posting>', 0)
+
+@pytest.fixture
+def cc_txn_pair():
+    dates = testutil.date_seq()
+    txn_meta = {
+        'payee': 'Smith-Dakota',
+        'rt-id': 'rt:550',
+    }
+    return [
+        testutil.Transaction(
+            **txn_meta,
+            date=next(dates),
+            receipt='CCReceipt.pdf',
+            postings=[
+                ('Liabilities:CreditCard', -36),
+                ('Expenses:Other', 35),
+                ('Expenses:Tax:Sales', 1),
+            ],
+        ),
+        testutil.Transaction(
+            **txn_meta,
+            date=next(dates),
+            receipt='CCPayment.pdf',
+            postings=[
+                ('Liabilities:CreditCard', 36),
+                ('Assets:Checking', -36, {'statement': 'CheckingStatement.pdf'}),
+            ],
+        ),
+    ]
+
+def check_filter(actual, entries, expected_indexes):
+    postings = [post for txn in entries for post in txn.postings]
+    expected = (postings[ii] for ii in expected_indexes)
+    for actual_post, expected_post in itertools.zip_longest(
+            actual, expected, fillvalue=MISSING_POSTING,
+    ):
+        assert actual_post[:-1] == expected_post[:-1]
+
+@pytest.mark.parametrize('key,value,expected_indexes', [
+    ('entity', 'Smith-Dakota', range(5)),
+    ('receipt', 'CCReceipt.pdf', range(3)),
+    ('receipt', 'CCPayment.pdf', range(3, 5)),
+    ('receipt', 'CC', ()),
+    ('statement', 'CheckingStatement.pdf', [4]),
+    ('BadKey', '', ()),
+    ('emptykey', '', ()),
+])
+def test_filter_meta_equal(cc_txn_pair, key, value, expected_indexes):
+    postings = data.Posting.from_entries(cc_txn_pair)
+    actual = filters.filter_meta_equal(postings, key, value)
+    check_filter(actual, cc_txn_pair, expected_indexes)