Files @ 4874a107e834
Branch filter:

Location: NPO-Accounting/conservancy_beancount/conservancy_beancount/plugin/meta_rt_links.py

Brett Smith
meta_rt_links: Start hook.
"""meta_rt_links - Check that RT 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/>.

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 MetaRTLinks(core.TransactionHook):
    HOOK_GROUPS = frozenset(['linkcheck', 'network', 'rt'])
    LINK_METADATA = data.LINK_METADATA.union(['rt-id'])

    def __init__(self, config: configmod.Config) -> None:
        rt_wrapper = config.rt_wrapper()
        if rt_wrapper is None:
            raise errormod.ConfigurationError("can't log in to RT")
        self.rt = rt_wrapper

    def _check_links(self,
                     txn: Transaction,
                     meta: Mapping[MetaKey, MetaValue],
    ) -> errormod.Iter:
        for key in self.LINK_METADATA.intersection(meta):
            for link in str(meta[key]).split():
                if not link.startswith('rt:'):
                    continue
                parsed = self.rt.parse(link)
                if parsed is None or not self.rt.exists(*parsed):
                    yield errormod.BrokenRTLinkError(txn, key, link, parsed)

    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)