NPO-Accounting/conservancy_beancount Changeset - 5c6043311b8c

Changeset - 5c6043311b8c

Parent rev.

Child rev.

[Not reviewed]

0 2 2

Brett Smith - 4 years ago 2020-03-20 20:47:06
brettcsmith@brettcsmith.org

meta_repo_links: Start hook.

4 files changed with 212 insertions and 0 deletions:

conservancy_beancount/data.py

conservancy_beancount/errors.py

conservancy_beancount/plugin/meta_repo_links.py

tests/test_meta_repo_links.py

135

0 comments (0 inline, 0 general)

conservancy_beancount/data.py

➞

Show inline comments

@@ ... / @@ -16,48 +16,58 @@ throughout Conservancy tools. @@
 # 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.abc
 from beancount.core import account as bc_account
 from typing import (
     Iterable,
     Iterator,
     MutableMapping,
     Optional,
+)
 from .beancount_types import (
     MetaKey,
     MetaValue,
     Posting as BasePosting,
     Transaction,
+)
 LINK_METADATA = frozenset([
     'approval',
     'check',
     'contract',
     'invoice',
     'purchase-order',
     'receipt',
     'statement',
 ])
 class Account(str):
     """Account name string
     This is a string that names an account, like Accrued:AccountsPayable
     or Income:Donations. This class provides additional methods for common
     account name parsing and queries.
     """
     SEP = bc_account.sep
     def is_income(self) -> bool:
         return self.is_under('Income:', 'UnearnedIncome:') is not None
     def is_real_asset(self) -> bool:
         return bool(
             self.is_under('Assets:')
             and not self.is_under('Assets:PrepaidExpenses', 'Assets:PrepaidVacation')
+        )
     def is_under(self, *acct_seq: str) -> Optional[str]:
         """Return a match if this account is "under" a part of the hierarchy
         Pass in any number of account name strings as arguments. If this
         account is under one of those strings in the account hierarchy, the
         first matching string will be returned. Otherwise, None is returned.

conservancy_beancount/errors.py

➞

Show inline comments

@@ ... / @@ -17,43 +17,51 @@ @@
 from typing import (
     Iterable,
+)
 class Error(Exception):
     def __init__(self, message, entry, source=None):
         self.message = message
         self.entry = entry
         self.source = entry.meta if source is None else source
     def __repr__(self):
         return "{clsname}<{source[filename]}:{source[lineno]}: {message}>".format(
             clsname=type(self).__name__,
             message=self.message,
             source=self.source,
+        )
     def _fill_source(self, source, filename='conservancy_beancount', lineno=0):
         source.setdefault('filename', filename)
         source.setdefault('lineno', lineno)
 Iter = Iterable[Error]
 class BrokenLinkError(Error):
     def __init__(self, txn, key, link, source=None):
         super().__init__(
             "{} not found in repository: {}".format(key, link),
             txn,
             source,
+        )
 class ConfigurationError(Error):
     def __init__(self, message, entry=None, source=None):
         if source is None:
             source = {}
         self._fill_source(source)
         super().__init__(message, entry, source)
 class InvalidMetadataError(Error):
     def __init__(self, txn, post, key, value=None, source=None):
         if value is None:
             msg_fmt = "{post.account} missing {key}"
         else:
             msg_fmt = "{post.account} has invalid {key}: {value}"
         super().__init__(
             msg_fmt.format(post=post, key=key, value=value),
             txn,
             source,
+        )

conservancy_beancount/plugin/meta_repo_links.py

➞

Show inline comments

 new file 100644
 """meta_repo_links - Check that repository links are valid"""
 # 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 re
 from . import core
 from .. import config as configmod
 from .. import data
 from .. import errors as errormod
 from ..beancount_types import (
     MetaKey,
     MetaValue,
     Transaction,
+)
 from typing import (
     Mapping,
+)
 class MetaRepoLinks(core.TransactionHook):
     HOOK_GROUPS = frozenset(['linkcheck'])
     PATH_PUNCT_RE = re.compile(r'[:/]')
     def __init__(self, config: configmod.Config) -> None:
         repo_path = config.repository_path()
         if repo_path is None:
             raise errormod.ConfigurationError("no repository configured")
         self.repo_path = repo_path
     def _check_links(self,
                      txn: Transaction,
                      meta: Mapping[MetaKey, MetaValue],
     ) -> errormod.Iter:
         for key in data.LINK_METADATA.intersection(meta):
             for link in str(meta[key]).split():
                 match = self.PATH_PUNCT_RE.search(link)
                 if match and match.group(0) == ':':
                     pass
                 elif not (self.repo_path / link).exists():
                     yield errormod.BrokenLinkError(txn, key, link)
     def run(self, txn: Transaction) -> errormod.Iter:
         yield from self._check_links(txn, txn.meta)
         for post in txn.postings:
             if post.meta is not None:
                 yield from self._check_links(txn, post.meta)

tests/test_meta_repo_links.py

➞

Show inline comments

 new file 100644
 """Test link checker for repository documents"""
 # 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
 from pathlib import Path
 import pytest
 from . import testutil
 from conservancy_beancount import errors as errormod
 from conservancy_beancount.plugin import meta_repo_links
 METADATA_KEYS = [
     'approval',
     'check',
     'contract',
     'invoice',
     'purchase-order',
     'receipt',
     'statement',
+]
 GOOD_LINKS = [Path(s) for s in [
     'Projects/project-data.yml',
     'Projects/project-list.yml',
 ]]
 BAD_LINKS = [Path(s) for s in [
     'NonexistentDirectory/NonexistentFile1.txt',
     'NonexistentDirectory/NonexistentFile2.txt',
 ]]
 NOT_FOUND_MSG = '{} not found in repository: {}'.format
 def build_meta(keys=None, *sources):
     if keys is None:
         keys = iter(METADATA_KEYS)
     sources = (itertools.cycle(src) for src in sources)
     return {key: ' '.join(str(x) for x in rest)
             for key, *rest in zip(keys, *sources)}
 @pytest.fixture(scope='module')
 def hook():
     config = testutil.TestConfig(repo_path='repository')
     return meta_repo_links.MetaRepoLinks(config)
 def test_error_with_no_repository():
     config = testutil.TestConfig(repo_path=None)
     with pytest.raises(errormod.ConfigurationError):
         meta_repo_links.MetaRepoLinks(config)
 def test_good_txn_links(hook):
     meta = build_meta(None, GOOD_LINKS)
     txn = testutil.Transaction(**meta, postings=[
         ('Income:Donations', -5),
         ('Assets:Cash', 5),
     ])
     assert not list(hook.run(txn))
 def test_good_post_links(hook):
     meta = build_meta(None, GOOD_LINKS)
     txn = testutil.Transaction(postings=[
         ('Income:Donations', -5, meta),
         ('Assets:Cash', 5),
     ])
     assert not list(hook.run(txn))
 def test_bad_txn_links(hook):
     meta = build_meta(None, BAD_LINKS)
     txn = testutil.Transaction(**meta, postings=[
         ('Income:Donations', -5),
         ('Assets:Cash', 5),
     ])
     expected = {NOT_FOUND_MSG(key, value) for key, value in meta.items()}
     actual = {error.message for error in hook.run(txn)}
     assert expected == actual
 def test_bad_post_links(hook):
     meta = build_meta(None, BAD_LINKS)
     txn = testutil.Transaction(postings=[
         ('Income:Donations', -5, meta.copy()),
         ('Assets:Cash', 5),
     ])
     expected = {NOT_FOUND_MSG(key, value) for key, value in meta.items()}
     actual = {error.message for error in hook.run(txn)}
     assert expected == actual
 @pytest.mark.parametrize('ext_doc', [
     'rt:123',
     'rt:456/789',
     'rt://ticket/23',
     'rt://ticket/34/attachments/567890',
 ])
 def test_docs_outside_repository_not_checked(hook, ext_doc):
     txn = testutil.Transaction(
         receipt='{} {} {}'.format(GOOD_LINKS[0], ext_doc, BAD_LINKS[1]),
         postings=[
             ('Income:Donations', -5),
             ('Assets:Cash', 5),
         ])
     expected = {NOT_FOUND_MSG('receipt', BAD_LINKS[1])}
     actual = {error.message for error in hook.run(txn)}
     assert expected == actual
 def test_mixed_results(hook):
     txn = testutil.Transaction(
         approval='{} {}'.format(*GOOD_LINKS),
         contract='{} {}'.format(BAD_LINKS[0], GOOD_LINKS[1]),
         postings=[
             ('Income:Donations', -5, {'invoice': '{} {}'.format(*BAD_LINKS)}),
             ('Assets:Cash', 5, {'statement': '{} {}'.format(GOOD_LINKS[0], BAD_LINKS[1])}),
         ])
     expected = {
         NOT_FOUND_MSG('contract', BAD_LINKS[0]),
         NOT_FOUND_MSG('invoice', BAD_LINKS[0]),
         NOT_FOUND_MSG('invoice', BAD_LINKS[1]),
         NOT_FOUND_MSG('statement', BAD_LINKS[1]),
+    }
     actual = {error.message for error in hook.run(txn)}
     assert expected == actual

0 comments (0 inline, 0 general)