class BrokenRTLinkError(Error):

    def __init__(self, txn, key, link, parsed=True, source=None):

        if parsed:

            msg_fmt = "{} not found in RT: {}"

        else:

            msg_fmt = "{} link is malformed: {}"

        super().__init__(

            msg_fmt.format(key, link),

            txn,

            source,

        )

"""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)

"""Test link checker for RT links"""

# 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 errors as errormod

from conservancy_beancount.plugin import meta_rt_links

METADATA_KEYS = [

    'approval',

    'check',

    'contract',

    'invoice',

    'purchase-order',

    'receipt',

    'rt-id',

    'statement',

]

GOOD_LINKS = [

    'rt:1',

    'rt:1/5',

    'rt://ticket/2',

    'rt://ticket/3/attachments/15',

]

MALFORMED_LINKS = [

    'rt:one',

    'rt:two/three',

    'rt://4',

    'rt://ticket/5/attach/6',

]

NOT_FOUND_LINKS = [

    'rt:1/10',

    'rt:10',

    'rt://ticket/9',

    'rt://ticket/3/attachments/99',

]

MALFORMED_MSG = '{} link is malformed: {}'.format

NOT_FOUND_MSG = '{} not found in RT: {}'.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(rt_client=testutil.RTClient())

    return meta_rt_links.MetaRTLinks(config)

def test_error_with_no_rt():

    config = testutil.TestConfig()

    with pytest.raises(errormod.ConfigurationError):

        meta_rt_links.MetaRTLinks(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))

@pytest.mark.parametrize('link_source,format_error', [

    (MALFORMED_LINKS, MALFORMED_MSG),

    (NOT_FOUND_LINKS, NOT_FOUND_MSG),

])

def test_bad_txn_links(hook, link_source, format_error):

    meta = build_meta(None, link_source)

    txn = testutil.Transaction(**meta, postings=[

        ('Income:Donations', -5),

        ('Assets:Cash', 5),

    ])

    expected = {format_error(key, value) for key, value in meta.items()}

    actual = {error.message for error in hook.run(txn)}

    assert expected == actual

@pytest.mark.parametrize('link_source,format_error', [

    (MALFORMED_LINKS, MALFORMED_MSG),

    (NOT_FOUND_LINKS, NOT_FOUND_MSG),

])

def test_bad_post_links(hook, link_source, format_error):

    meta = build_meta(None, link_source)

    txn = testutil.Transaction(postings=[

        ('Income:Donations', -5, meta.copy()),

        ('Assets:Cash', 5),

    ])

    expected = {format_error(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', [

    'statement.txt',

    'https://example.org/',

])

def test_docs_outside_rt_not_checked(hook, ext_doc):

    txn = testutil.Transaction(

        receipt='{} {} {}'.format(GOOD_LINKS[0], ext_doc, MALFORMED_LINKS[1]),

        postings=[

            ('Income:Donations', -5),

            ('Assets:Cash', 5),

        ])

    expected = {MALFORMED_MSG('receipt', MALFORMED_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(MALFORMED_LINKS[0], GOOD_LINKS[1]),

        postings=[

            ('Income:Donations', -5, {'invoice': '{} {}'.format(*NOT_FOUND_LINKS)}),

            ('Assets:Cash', 5, {'statement': '{} {}'.format(GOOD_LINKS[0], MALFORMED_LINKS[1])}),

        ])

    expected = {

        MALFORMED_MSG('contract', MALFORMED_LINKS[0]),

        NOT_FOUND_MSG('invoice', NOT_FOUND_LINKS[0]),

        NOT_FOUND_MSG('invoice', NOT_FOUND_LINKS[1]),

        MALFORMED_MSG('statement', MALFORMED_LINKS[1]),

    }

    actual = {error.message for error in hook.run(txn)}

    assert expected == actual

from conservancy_beancount import rtutil

    def __init__(self,

                 repo_path=None,

                 rt_client=None,

    ):

        self._rt_client = rt_client

        if rt_client is None:

            self._rt_wrapper = None

        else:

            self._rt_wrapper = rtutil.RT(rt_client)

    def rt_client(self):

        return self._rt_client

    def rt_wrapper(self):

        return self._rt_wrapper