#       Licensed to the Apache Software Foundation (ASF) under one
#       or more contributor license agreements.  See the NOTICE file
#       distributed with this work for additional information
#       regarding copyright ownership.  The ASF licenses this file
#       to you under the Apache License, Version 2.0 (the
#       "License"); you may not use this file except in compliance
#       with the License.  You may obtain a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#       Unless required by applicable law or agreed to in writing,
#       software distributed under the License is distributed on an
#       "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
#       KIND, either express or implied.  See the License for the
#       specific language governing permissions and limitations
#       under the License.

import json
import hmac
import hashlib
import datetime as dt

from mock import Mock, MagicMock, patch, call
import pytest
from formencode import Invalid
from ming.odm import session
from tg import tmpl_context as c
from tg import config

from allura import model as M
from allura.lib import helpers as h
from allura.webhooks import (
    WebhookValidator,
    WebhookController,
    send_webhook,
    RepoPushWebhookSender,
    SendWebhookHelper,
)
from allura.tests import decorators as td
from alluratest.controller import (
    setup_basic_test,
    TestController,
    TestRestApiBase,
)
import six


# important to be distinct from 'test' and 'test2' which ForgeGit and
# ForgeImporter use, so that the tests can run in parallel and not clobber each
# other
test_project_with_repo = 'adobe-1'
with_git = td.with_tool(test_project_with_repo, 'git', 'src', 'Git')
with_git2 = td.with_tool(test_project_with_repo, 'git', 'src2', 'Git2')


class TestWebhookBase:
    def setup_method(self, method):
        setup_basic_test()
        self.patches = self.monkey_patch()
        for p in self.patches:
            p.start()
        self.setup_with_tools()
        self.project = M.Project.query.get(shortname=test_project_with_repo)
        self.git = self.project.app_instance('src')
        self.wh = M.Webhook(
            type='repo-push',
            app_config_id=self.git.config._id,
            hook_url='http://httpbin.org/post',
            secret='secret')
        session(self.wh).flush(self.wh)

    def teardown_method(self, method):
        for p in self.patches:
            p.stop()

    @with_git
    def setup_with_tools(self):
        pass

    def monkey_patch(self):
        # we don't need actual repo here, and this avoids test conflicts when
        # running in parallel
        repo_init = patch.object(M.Repository, 'init', autospec=True)
        return [repo_init]


class TestValidators(TestWebhookBase):
    @with_git2
    def test_webhook_validator(self):
        sender = Mock(type='repo-push')
        app = self.git
        invalid_app = self.project.app_instance('src2')
        v = WebhookValidator(sender=sender, app=app, not_empty=True)
        with pytest.raises(Invalid) as cm:
            v.to_python(None)
        assert cm.value.msg == 'Please enter a value'
        with pytest.raises(Invalid) as cm:
            v.to_python('invalid id')
        assert cm.value.msg == 'Invalid webhook'

        wh = M.Webhook(type='invalid type',
                       app_config_id=invalid_app.config._id,
                       hook_url='http://hooks.slack.com',
                       secret='secret')
        session(wh).flush(wh)
        # invalid type
        with pytest.raises(Invalid) as cm:
            v.to_python(wh._id)
        assert cm.value.msg == 'Invalid webhook'

        wh.type = 'repo-push'
        session(wh).flush(wh)
        # invalild app
        with pytest.raises(Invalid) as cm:
            v.to_python(wh._id)
        assert cm.value.msg == 'Invalid webhook'

        wh.app_config_id = app.config._id
        session(wh).flush(wh)
        assert v.to_python(wh._id) == wh
        assert v.to_python(str(wh._id)) == wh


class TestWebhookController(TestController):

    def setup_method(self, method):
        super().setup_method(method)
        self.patches = self.monkey_patch()
        for p in self.patches:
            p.start()
        self.setup_with_tools()
        self.project = M.Project.query.get(shortname=test_project_with_repo)
        self.git = self.project.app_instance('src')
        self.url = str(self.git.admin_url + 'webhooks')

    def teardown_method(self, method):
        super().teardown_method(method)
        for p in self.patches:
            p.stop()

    @with_git
    def setup_with_tools(self):
        pass

    def monkey_patch(self):
        gen_secret = patch.object(
            WebhookController,
            'gen_secret',
            return_value='super-secret',
            autospec=True)
        # we don't need actual repo here, and this avoids test conflicts when
        # running in parallel
        repo_init = patch.object(M.Repository, 'init', autospec=True)
        return [gen_secret, repo_init]

    def create_webhook(self, data, url=None):
        url = url or self.url
        r = self.app.post(url + '/repo-push/create', data)
        wf = json.loads(self.webflash(r))
        assert wf['status'] == 'ok'
        assert wf['message'] == 'Created successfully'
        return r

    def find_error(self, r, field, msg, form_type='create'):
        form = r.html.find('form', attrs={'action': form_type})
        if field == '_the_form':
            error = form.findPrevious('div', attrs={'class': 'error'})
        else:
            error = form.find('input', attrs={'name': field})
            error = error.findNext('div', attrs={'class': 'error'})
        if error:
            assert msg in error.getText()
        else:
            assert False, 'Validation error not found'

    def test_AAAA_WORKAROUND__edit(self):
        """
        This must run first in this test class for unknown reasons ever since
            https://github.com/TurboGears/tg2/commit/02fb49b14e70fdd8ac16973488fb3637e5e59114

        If any test runs the self.app.post from create_webhook before this one, then this test will fail on:
            with td.audits(msg):
                r = form.submit()
        because WebhookValidator's `value` will be "create" instead of an objectid str

        Maybe something to do with WebhookControllerMeta setup of `validate` decorators?
        """
        data1 = {'url': 'http://httpbin.org/post',
                 'secret': 'secret'}
        data2 = {'url': 'http://example.com/hook',
                 'secret': 'secret2'}
        self.create_webhook(data1).follow()
        self.create_webhook(data2).follow()
        assert M.Webhook.query.find().count() == 2
        wh1 = M.Webhook.query.get(hook_url=data1['url'])
        r = self.app.get(self.url + '/repo-push/%s' % wh1._id)
        form = r.forms[0]
        assert form['url'].value == data1['url']
        assert form['secret'].value == data1['secret']
        assert form['webhook'].value == str(wh1._id)
        form['url'] = 'http://host.org/hook'
        form['secret'] = 'new secret'
        msg = 'edit webhook repo-push\n{} => {}\n{}'.format(
            data1['url'], form['url'].value, 'secret changed')
        with td.audits(msg):
            r = form.submit()
        wf = json.loads(self.webflash(r))
        assert wf['status'] == 'ok'
        assert wf['message'] == 'Edited successfully'
        assert M.Webhook.query.find().count() == 2
        wh1 = M.Webhook.query.get(_id=wh1._id)
        assert wh1.hook_url == 'http://host.org/hook'
        assert wh1.app_config_id == self.git.config._id
        assert wh1.secret == 'new secret'
        assert wh1.type == 'repo-push'

        # Duplicates
        r = self.app.get(self.url + '/repo-push/%s' % wh1._id)
        form = r.forms[0]
        form['url'] = data2['url']
        r = form.submit()
        self.find_error(r, '_the_form',
                        '"repo-push" webhook already exists for Git http://example.com/hook',
                        form_type='edit')

    def test_access(self):
        self.app.get(self.url + '/repo-push/')
        self.app.get(self.url + '/repo-push/',
                     extra_environ={'username': 'test-user'},
                     status=403)
        r = self.app.get(self.url + '/repo-push/',
                         extra_environ={'username': '*anonymous'},
                         status=302)
        assert (r.location ==
                     'http://localhost/auth/'
                     '?return_to=%2Fadobe%2Fadobe-1%2Fadmin%2Fsrc%2Fwebhooks%2Frepo-push%2F')

    def test_invalid_hook_type(self):
        self.app.get(self.url + '/invalid-hook-type/', status=404)

    def test_create(self):
        assert M.Webhook.query.find().count() == 0
        r = self.app.get(self.url)
        assert '<h1>repo-push</h1>' in r
        assert 'http://httpbin.org/post' not in r
        data = {'url': 'http://httpbin.org/post',
                'secret': ''}
        msg = 'add webhook repo-push {} {}'.format(
            data['url'], self.git.config.url())
        with td.audits(msg):
            r = self.create_webhook(data).follow()
        assert 'http://httpbin.org/post' in r

        hooks = M.Webhook.query.find().all()
        assert len(hooks) == 1
        assert hooks[0].type == 'repo-push'
        assert hooks[0].hook_url == 'http://httpbin.org/post'
        assert hooks[0].app_config_id == self.git.config._id
        assert hooks[0].secret == 'super-secret'

        # Try to create duplicate
        with td.out_audits(msg):
            r = self.app.post(self.url + '/repo-push/create', data)
        self.find_error(r, '_the_form',
                        '"repo-push" webhook already exists for Git http://httpbin.org/post')
        assert M.Webhook.query.find().count() == 1

    def test_create_limit_reached(self):
        assert M.Webhook.query.find().count() == 0
        limit = json.dumps({'git': 1})
        with h.push_config(config, **{'webhook.repo_push.max_hooks': limit}):
            data = {'url': 'http://httpbin.org/post',
                    'secret': ''}
            r = self.create_webhook(data).follow()
            assert M.Webhook.query.find().count() == 1

            r = self.app.post(self.url + '/repo-push/create', data)
            wf = json.loads(self.webflash(r))
            assert wf['status'] == 'error'
            assert (
                wf['message'] ==
                'You have exceeded the maximum number of webhooks '
                'you are allowed to create for this project/app')
            assert M.Webhook.query.find().count() == 1

    def test_create_validation(self):
        assert M.Webhook.query.find().count() == 0
        r = self.app.post(
            self.url + '/repo-push/create', {}, status=404)

        data = {'url': '', 'secret': ''}
        r = self.app.post(self.url + '/repo-push/create', data)
        self.find_error(r, 'url', 'Please enter a value')

        data = {'url': 'qwer', 'secret': 'qwe'}
        r = self.app.post(self.url + '/repo-push/create', data)
        self.find_error(r, 'url',
                        'You must provide a full domain name (like qwer.com)')

    def test_edit_validation(self):
        invalid = M.Webhook(
            type='invalid type',
            app_config_id=None,
            hook_url='http://httpbin.org/post',
            secret='secret')
        session(invalid).flush(invalid)
        self.app.get(self.url + '/repo-push/%s' % invalid._id, status=404)

        data = {'url': 'http://httpbin.org/post',
                'secret': 'secret'}
        self.create_webhook(data).follow()
        wh = M.Webhook.query.get(hook_url=data['url'], type='repo-push')

        # invalid id in hidden field, just in case
        r = self.app.get(self.url + '/repo-push/%s' % wh._id)
        data = {k: v[0].value for (k, v) in r.forms[0].fields.items() if k}
        data['webhook'] = str(invalid._id)
        self.app.post(self.url + '/repo-push/edit', data, status=404)

        # empty values
        data = {'url': '', 'secret': '', 'webhook': str(wh._id)}
        r = self.app.post(self.url + '/repo-push/edit', data)
        self.find_error(r, 'url', 'Please enter a value', 'edit')

        data = {'url': 'qwe', 'secret': 'qwe', 'webhook': str(wh._id)}
        r = self.app.post(self.url + '/repo-push/edit', data)
        self.find_error(r, 'url',
                        'You must provide a full domain name (like qwe.com)', 'edit')

    def test_delete(self):
        data = {'url': 'http://httpbin.org/post',
                'secret': 'secret'}
        self.create_webhook(data).follow()
        assert M.Webhook.query.find().count() == 1
        wh = M.Webhook.query.get(hook_url=data['url'])
        data = {'webhook': str(wh._id)}
        msg = 'delete webhook repo-push {} {}'.format(
            wh.hook_url, self.git.config.url())
        with td.audits(msg):
            r = self.app.post(self.url + '/repo-push/delete', data)
        assert r.json == {'status': 'ok'}
        assert M.Webhook.query.find().count() == 0

    def test_delete_validation(self):
        invalid = M.Webhook(
            type='invalid type',
            app_config_id=None,
            hook_url='http://httpbin.org/post',
            secret='secret')
        session(invalid).flush(invalid)
        assert M.Webhook.query.find().count() == 1

        data = {'webhook': ''}
        self.app.post(self.url + '/repo-push/delete', data, status=404)

        data = {'webhook': str(invalid._id)}
        self.app.post(self.url + '/repo-push/delete', data, status=404)
        assert M.Webhook.query.find().count() == 1

    @with_git2
    def test_list_webhooks(self):
        git2 = self.project.app_instance('src2')
        url2 = str(git2.admin_url + 'webhooks')
        data1 = {'url': 'http://httpbin.org/post',
                 'secret': 'secret'}
        data2 = {'url': 'http://another-host.org/',
                 'secret': 'secret2'}
        data3 = {'url': 'http://another-app.org/',
                 'secret': 'secret3'}
        self.create_webhook(data1).follow()
        self.create_webhook(data2).follow()
        self.create_webhook(data3, url=url2).follow()
        wh1 = M.Webhook.query.get(hook_url=data1['url'])
        wh2 = M.Webhook.query.get(hook_url=data2['url'])

        r = self.app.get(self.url)
        assert '<h1>repo-push</h1>' in r
        rows = r.html.find('table').findAll('tr')
        assert len(rows) == 2
        rows = sorted((self._format_row(row) for row in rows), key=lambda rows: rows[0]['text'])
        expected_rows = sorted([
            [{'text': wh1.hook_url},
             {'text': wh1.secret},
             {'href': self.url + '/repo-push/' + str(wh1._id),
              'text': 'Edit'},
             {'href': self.url + '/repo-push/delete',
              'data-id': str(wh1._id)}],
            [{'text': wh2.hook_url},
             {'text': wh2.secret},
             {'href': self.url + '/repo-push/' + str(wh2._id),
              'text': 'Edit'},
             {'href': self.url + '/repo-push/delete',
              'data-id': str(wh2._id)}],
        ], key=lambda rows: rows[0]['text'])
        assert rows == expected_rows
        # make sure webhooks for another app is not visible
        assert 'http://another-app.org/' not in r
        assert 'secret3' not in r

    def _format_row(self, row):
        def link(td):
            a = td.find('a')
            return {'href': a.get('href'), 'text': a.getText()}

        def text(td):
            return {'text': td.getText()}

        def delete_btn(td):
            a = td.find('a')
            return {'href': a.get('href'), 'data-id': a.get('data-id')}

        tds = row.findAll('td')
        return [text(tds[0]), text(tds[1]), link(tds[2]), delete_btn(tds[3])]


class TestSendWebhookHelper(TestWebhookBase):
    def setup_method(self, method):
        super().setup_method(method)
        self.payload = {'some': ['data', 23]}
        self.h = SendWebhookHelper(self.wh, self.payload)

    def test_timeout(self):
        assert self.h.timeout == 30
        with h.push_config(config, **{'webhook.timeout': 10}):
            assert self.h.timeout == 10

    def test_retries(self):
        assert self.h.retries == [60, 120, 240]
        with h.push_config(config, **{'webhook.retry': '1 2 3 4 5 6'}):
            assert self.h.retries == [1, 2, 3, 4, 5, 6]

    def test_sign(self):
        json_payload = json.dumps(self.payload)
        signature = hmac.new(
            self.wh.secret.encode('utf-8'),
            json_payload.encode('utf-8'),
            hashlib.sha1)
        signature = 'sha1=' + signature.hexdigest()
        assert self.h.sign(json_payload) == signature

    def test_log_msg(self):
        assert (
            self.h.log_msg('OK') ==
            'OK: repo-push http://httpbin.org/post /adobe/adobe-1/src/')
        response = Mock(
            status_code=500,
            text='that is why',
            headers={'Content-Type': 'application/json'})
        assert (
            self.h.log_msg('Error', response=response) ==
            "Error: repo-push http://httpbin.org/post /adobe/adobe-1/src/ 500 "
            "that is why {'Content-Type': 'application/json'}")

    @patch('allura.webhooks.SendWebhookHelper', autospec=True)
    def test_send_webhook_task(self, swh):
        send_webhook(self.wh._id, self.payload)
        swh.assert_called_once_with(self.wh, self.payload)

    @patch('allura.webhooks.requests', autospec=True)
    @patch('allura.webhooks.log', autospec=True)
    def test_send(self, log, requests):
        requests.post.return_value = Mock(status_code=200)
        self.h.sign = Mock(return_value='sha1=abc')
        self.h.send()
        headers = {'content-type': 'application/json',
                   'User-Agent': 'Allura Webhook (https://allura.apache.org/)',
                   'X-Allura-Signature': 'sha1=abc'}
        requests.post.assert_called_once_with(
            self.wh.hook_url,
            data=json.dumps(self.payload),
            headers=headers,
            timeout=30)
        log.info.assert_called_once_with(
            'Webhook successfully sent: {} {} {}'.format(
                self.wh.type, self.wh.hook_url, self.wh.app_config.url()))

    @patch('allura.webhooks.time', autospec=True)
    @patch('allura.webhooks.requests', autospec=True)
    @patch('allura.webhooks.log', autospec=True)
    def test_send_error_response_status(self, log, requests, time):
        requests.post.return_value = Mock(status_code=500)
        self.h.send()
        assert requests.post.call_count == 4  # initial call + 3 retries
        assert (time.sleep.call_args_list ==
                     [call(60), call(120), call(240)])
        assert log.info.call_args_list == [
            call('Retrying webhook in: %s', [60, 120, 240]),
            call('Retrying webhook in %s seconds', 60),
            call('Retrying webhook in %s seconds', 120),
            call('Retrying webhook in %s seconds', 240)]
        assert log.error.call_count == 4
        log.error.assert_called_with(
            'Webhook send error: {} {} {} {} {} {}'.format(
                self.wh.type, self.wh.hook_url,
                self.wh.app_config.url(),
                requests.post.return_value.status_code,
                requests.post.return_value.text,
                requests.post.return_value.headers))

    @patch('allura.webhooks.time', autospec=True)
    @patch('allura.webhooks.requests', autospec=True)
    @patch('allura.webhooks.log', autospec=True)
    def test_send_error_no_retries(self, log, requests, time):
        requests.post.return_value = Mock(status_code=500)
        with h.push_config(config, **{'webhook.retry': ''}):
            self.h.send()
            assert requests.post.call_count == 1
            assert time.call_count == 0
            log.info.assert_called_once_with('Retrying webhook in: %s', [])
            assert log.error.call_count == 1
            log.error.assert_called_with(
                'Webhook send error: {} {} {} {} {} {}'.format(
                    self.wh.type, self.wh.hook_url,
                    self.wh.app_config.url(),
                    requests.post.return_value.status_code,
                    requests.post.return_value.text,
                    requests.post.return_value.headers))


class TestRepoPushWebhookSender(TestWebhookBase):
    @patch('allura.webhooks.send_webhook', autospec=True)
    def test_send(self, send_webhook):
        sender = RepoPushWebhookSender()
        sender.get_payload = Mock()
        with h.push_config(c, app=self.git):
            sender.send(dict(arg1=1, arg2=2))
        send_webhook.post.assert_called_once_with(
            self.wh._id,
            sender.get_payload.return_value)

    @patch('allura.webhooks.send_webhook', autospec=True)
    def test_send_with_list(self, send_webhook):
        sender = RepoPushWebhookSender()
        sender.get_payload = Mock(side_effect=[1, 2])
        self.wh.enforce_limit = Mock(return_value=True)
        with h.push_config(c, app=self.git):
            sender.send([dict(arg1=1, arg2=2), dict(arg1=3, arg2=4)])
        assert send_webhook.post.call_count == 2
        assert (send_webhook.post.call_args_list ==
                     [call(self.wh._id, 1), call(self.wh._id, 2)])
        assert self.wh.enforce_limit.call_count == 1

    @patch('allura.webhooks.log', autospec=True)
    @patch('allura.webhooks.send_webhook', autospec=True)
    def test_send_limit_reached(self, send_webhook, log):
        sender = RepoPushWebhookSender()
        sender.get_payload = Mock()
        self.wh.enforce_limit = Mock(return_value=False)
        with h.push_config(c, app=self.git):
            sender.send(dict(arg1=1, arg2=2))
        assert send_webhook.post.call_count == 0
        log.warn.assert_called_once_with(
            'Webhook fires too often: %s. Skipping', self.wh)

    @patch('allura.webhooks.send_webhook', autospec=True)
    def test_send_no_configured_webhooks(self, send_webhook):
        self.wh.delete()
        session(self.wh).flush(self.wh)
        sender = RepoPushWebhookSender()
        with h.push_config(c, app=self.git):
            sender.send(dict(arg1=1, arg2=2))
        assert send_webhook.post.call_count == 0

    def test_get_payload(self):
        sender = RepoPushWebhookSender()
        _ci = lambda x: MagicMock(webhook_info={'id': str(x)}, parent_ids=['0'])
        with patch.object(self.git.repo, 'commit', new=_ci):
            with h.push_config(c, app=self.git):
                result = sender.get_payload(commit_ids=['1', '2', '3'], ref='ref')
        expected_result = {
            'size': 3,
            'commits': [{'id': '1'}, {'id': '2'}, {'id': '3'}],
            'ref': 'ref',
            'after': '1',
            'before': '0',
            'repository': {
                'full_name': '/adobe/adobe-1/src/',
                'name': 'Git',
                'url': 'http://localhost/adobe/adobe-1/src/',
            },
        }
        assert result == expected_result

    def test_enforce_limit(self):
        def add_webhooks(suffix, n):
            for i in range(n):
                webhook = M.Webhook(
                    type='repo-push',
                    app_config_id=self.git.config._id,
                    hook_url=f'http://httpbin.org/{suffix}/{i}',
                    secret='secret')
                session(webhook).flush(webhook)

        sender = RepoPushWebhookSender()
        # default
        assert sender.enforce_limit(self.git) is True
        add_webhooks('one', 3)
        assert sender.enforce_limit(self.git) is False

        # config
        limit = json.dumps({'git': 5})
        with h.push_config(config, **{'webhook.repo_push.max_hooks': limit}):
            assert sender.enforce_limit(self.git) is True
            add_webhooks('two', 3)
            assert sender.enforce_limit(self.git) is False

    def test_before(self):
        sender = RepoPushWebhookSender()
        with patch.object(self.git.repo, 'commit', autospec=True) as _ci:
            assert sender._before(self.git.repo, ['3', '2', '1']) == ''
            _ci.return_value.parent_ids = ['0']
            assert sender._before(self.git.repo, ['3', '2', '1']) == '0'

    def test_after(self):
        sender = RepoPushWebhookSender()
        assert sender._after([]) == ''
        assert sender._after(['3', '2', '1']) == '3'

    def test_convert_id(self):
        sender = RepoPushWebhookSender()
        assert sender._convert_id('') == ''
        assert sender._convert_id('a433fa9') == 'a433fa9'
        assert sender._convert_id('a433fa9:13') == 'r13'


class TestModels(TestWebhookBase):
    def test_webhook_url(self):
        assert (self.wh.url() ==
                     f'/adobe/adobe-1/admin/src/webhooks/repo-push/{self.wh._id}')

    def test_webhook_enforce_limit(self):
        self.wh.last_sent = None
        assert self.wh.enforce_limit() is True
        # default value
        self.wh.last_sent = dt.datetime.utcnow() - dt.timedelta(seconds=31)
        assert self.wh.enforce_limit() is True
        self.wh.last_sent = dt.datetime.utcnow() - dt.timedelta(seconds=15)
        assert self.wh.enforce_limit() is False
        # value from config
        with h.push_config(config, **{'webhook.repo_push.limit': 100}):
            self.wh.last_sent = dt.datetime.utcnow() - dt.timedelta(seconds=101)
            assert self.wh.enforce_limit() is True
            self.wh.last_sent = dt.datetime.utcnow() - dt.timedelta(seconds=35)
            assert self.wh.enforce_limit() is False

    @patch('allura.model.webhook.dt', autospec=True)
    def test_update_limit(self, dt_mock):
        _now = dt.datetime(2015, 2, 2, 13, 39)
        dt_mock.datetime.utcnow.return_value = _now
        assert self.wh.last_sent is None
        self.wh.update_limit()
        session(self.wh).expunge(self.wh)
        assert M.Webhook.query.get(_id=self.wh._id).last_sent == _now

    def test_json(self):
        expected = {
            '_id': str(self.wh._id),
            'url': 'http://localhost/rest/adobe/adobe-1/admin'
                   '/src/webhooks/repo-push/{}'.format(self.wh._id),
            'type': 'repo-push',
            'hook_url': 'http://httpbin.org/post',
            'mod_date': self.wh.mod_date,
        }
        assert self.wh.__json__() == expected


class TestWebhookRestController(TestRestApiBase):
    def setup_method(self, method):
        super().setup_method(method)
        self.patches = self.monkey_patch()
        for p in self.patches:
            p.start()
        self.setup_with_tools()
        self.project = M.Project.query.get(shortname=test_project_with_repo)
        self.git = self.project.app_instance('src')
        self.url = '/rest' + self.git.admin_url + 'webhooks'
        self.webhooks = []
        for i in range(3):
            webhook = M.Webhook(
                type='repo-push',
                app_config_id=self.git.config._id,
                hook_url=f'http://httpbin.org/post/{i}',
                secret=f'secret-{i}')
            session(webhook).flush(webhook)
            self.webhooks.append(webhook)

    def teardown_method(self, method):
        super().teardown_method(method)
        for p in self.patches:
            p.stop()

    @with_git
    def setup_with_tools(self):
        pass

    def monkey_patch(self):
        gen_secret = patch.object(
            WebhookController,
            'gen_secret',
            return_value='super-secret',
            autospec=True)
        # we don't need actual repo here, and this avoids test conflicts when
        # running in parallel
        repo_init = patch.object(M.Repository, 'init', autospec=True)
        return [gen_secret, repo_init]

    def test_webhooks_list(self):
        r = self.api_get(self.url)
        webhooks = [{
            '_id': str(wh._id),
            'url': 'http://localhost/rest/adobe/adobe-1/admin'
                   '/src/webhooks/repo-push/{}'.format(wh._id),
            'type': 'repo-push',
            'hook_url': f'http://httpbin.org/post/{n}',
            'mod_date': str(wh.mod_date),
        } for n, wh in enumerate(self.webhooks)]
        expected = {
            'webhooks': webhooks,
            'limits': {'repo-push': {'max': 3, 'used': 3}},
        }
        assert r.json == expected

    def test_webhook_GET_404(self):
        r = self.api_get(self.url + '/repo-push/invalid', status=404)

    def test_webhook_GET(self):
        webhook = self.webhooks[0]
        r = self.api_get(f'{self.url}/repo-push/{webhook._id}')
        expected = {
            '_id': str(webhook._id),
            'url': 'http://localhost/rest/adobe/adobe-1/admin'
                   '/src/webhooks/repo-push/{}'.format(webhook._id),
            'type': 'repo-push',
            'hook_url': 'http://httpbin.org/post/0',
            'mod_date': str(webhook.mod_date),
        }
        assert r.status_int == 200
        assert r.json == expected

    def test_create_validation(self):
        assert M.Webhook.query.find().count() == len(self.webhooks)
        r = self.api_get(self.url + '/repo-push', status=405)

        r = self.api_post(self.url + '/repo-push', status=400)
        expected = {
            'result': 'error',
            'error': {'url': 'Please enter a value'},
        }
        assert r.json == expected

        data = {'url': 'qwer', 'secret': 'qwe'}
        r = self.api_post(self.url + '/repo-push', status=400, **data)
        expected = {
            'result': 'error',
            'error': {
                'url': 'You must provide a full domain name (like qwer.com)'
            },
        }
        assert r.json == expected
        assert M.Webhook.query.find().count() == len(self.webhooks)

    def test_create(self):
        assert M.Webhook.query.find().count() == len(self.webhooks)
        data = {'url': 'http://hook.slack.com/abcd'}
        limit = json.dumps({'git': 10})
        with h.push_config(config, **{'webhook.repo_push.max_hooks': limit}):
            msg = 'add webhook repo-push {} {}'.format(
                data['url'], self.git.config.url())
            with td.audits(msg):
                r = self.api_post(self.url + '/repo-push', status=201, **data)
        webhook = M.Webhook.query.get(hook_url=data['url'])
        assert webhook.secret == 'super-secret'  # secret generated
        expected = {
            '_id': str(webhook._id),
            'url': 'http://localhost/rest/adobe/adobe-1/admin'
                   '/src/webhooks/repo-push/{}'.format(webhook._id),
            'type': 'repo-push',
            'hook_url': data['url'],
            'mod_date': str(webhook.mod_date),
        }
        assert r.json == expected
        assert M.Webhook.query.find().count() == len(self.webhooks) + 1

    def test_create_duplicates(self):
        assert M.Webhook.query.find().count() == len(self.webhooks)
        data = {'url': self.webhooks[0].hook_url}
        limit = json.dumps({'git': 10})
        with h.push_config(config, **{'webhook.repo_push.max_hooks': limit}):
            r = self.api_post(self.url + '/repo-push', status=400, **data)
        expected = {'result': 'error',
                    'error': '_the_form: "repo-push" webhook already '
                              'exists for Git http://httpbin.org/post/0'}
        assert r.json == expected
        assert M.Webhook.query.find().count() == len(self.webhooks)

    def test_create_limit_reached(self):
        assert M.Webhook.query.find().count() == len(self.webhooks)
        data = {'url': 'http://hook.slack.com/abcd'}
        r = self.api_post(self.url + '/repo-push', status=400, **data)
        expected = {
            'result': 'error',
            'limits': {'max': 3, 'used': 3},
            'error': 'You have exceeded the maximum number of webhooks '
                      'you are allowed to create for this project/app'}
        assert r.json == expected
        assert M.Webhook.query.find().count() == len(self.webhooks)

    def test_edit_validation(self):
        webhook = self.webhooks[0]
        url = f'{self.url}/repo-push/{webhook._id}'
        data = {'url': 'qwe', 'secret': 'qwe'}
        r = self.api_post(url, status=400, **data)
        expected = {
            'result': 'error',
            'error': {
                'url': 'You must provide a full domain name (like qwe.com)'
            },
        }
        assert r.json == expected

    def test_edit(self):
        webhook = self.webhooks[0]
        url = f'{self.url}/repo-push/{webhook._id}'
        # change only url
        data = {'url': 'http://hook.slack.com/abcd'}
        msg = ('edit webhook repo-push\n'
               'http://httpbin.org/post/0 => http://hook.slack.com/abcd\n')
        with td.audits(msg):
            r = self.api_post(url, status=200, **data)
        webhook = M.Webhook.query.get(_id=webhook._id)
        assert webhook.hook_url == data['url']
        assert webhook.secret == 'secret-0'
        expected = {
            '_id': str(webhook._id),
            'url': 'http://localhost/rest/adobe/adobe-1/admin'
                   '/src/webhooks/repo-push/{}'.format(webhook._id),
            'type': 'repo-push',
            'hook_url': data['url'],
            'mod_date': str(webhook.mod_date),
        }
        assert r.json == expected

        # change only secret
        data = {'secret': 'new-secret'}
        msg = ('edit webhook repo-push\n'
               'http://hook.slack.com/abcd => http://hook.slack.com/abcd\n'
               'secret changed')
        with td.audits(msg):
            r = self.api_post(url, status=200, **data)
        webhook = M.Webhook.query.get(_id=webhook._id)
        assert webhook.hook_url == 'http://hook.slack.com/abcd'
        assert webhook.secret == 'new-secret'
        expected = {
            '_id': str(webhook._id),
            'url': 'http://localhost/rest/adobe/adobe-1/admin'
                   '/src/webhooks/repo-push/{}'.format(webhook._id),
            'type': 'repo-push',
            'hook_url': 'http://hook.slack.com/abcd',
            'mod_date': str(webhook.mod_date),
        }
        assert r.json == expected

    def test_edit_duplicates(self):
        webhook = self.webhooks[0]
        url = f'{self.url}/repo-push/{webhook._id}'
        data = {'url': 'http://httpbin.org/post/1'}
        r = self.api_post(url, status=400, **data)
        expected = {'result': 'error',
                    'error': '_the_form: "repo-push" webhook already '
                              'exists for Git http://httpbin.org/post/1'}
        assert r.json == expected

    def test_delete_validation(self):
        url = f'{self.url}/repo-push/invalid'
        self.api_delete(url, status=404)

    def test_delete(self):
        assert M.Webhook.query.find().count() == 3
        webhook = self.webhooks[0]
        url = f'{self.url}/repo-push/{webhook._id}'
        msg = 'delete webhook repo-push {} {}'.format(
            webhook.hook_url, self.git.config.url())
        with td.audits(msg):
            r = self.api_delete(url, status=200)
        assert r.json == {'result': 'ok'}
        assert M.Webhook.query.find().count() == 2
        assert M.Webhook.query.get(_id=webhook._id) is None

    def test_permissions(self):
        self.api_get(self.url, user='test-user', status=403)
        self.api_get(self.url, user='*anonymous', status=401)
        url = self.url + '/repo-push/'
        self.api_post(url, user='test-user', status=403)
        self.api_post(url, user='*anonymous', status=401)
        url = self.url + '/repo-push/' + str(self.webhooks[0]._id)
        self.api_get(url, user='test-user', status=403)
        self.api_get(url, user='*anonymous', status=401)
        self.api_post(url, user='test-user', status=403)
        self.api_post(url, user='*anonymous', status=401)
        self.api_delete(url, user='test-user', status=403)
        self.api_delete(url, user='*anonymous', status=401)