# -*- coding: utf-8 -*-
# Copyright (C) 2012-2016 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 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 .
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
from __future__ import unicode_literals
import deform
import logging
import requests
import colander
from celery.task import task
from mako.template import Template
from rhodecode import events
from rhodecode.translation import _
from rhodecode.integrations.types.base import IntegrationTypeBase
log = logging.getLogger(__name__)
class WebhookSettingsSchema(colander.Schema):
url = colander.SchemaNode(
colander.String(),
title=_('Webhook URL'),
description=_('URL of the webhook to receive POST event.'),
missing=colander.required,
required=True,
validator=colander.url,
widget=deform.widget.TextInputWidget(
placeholder='https://www.example.com/webhook'
),
)
secret_token = colander.SchemaNode(
colander.String(),
title=_('Secret Token'),
description=_('String used to validate received payloads.'),
default='',
missing='',
widget=deform.widget.TextInputWidget(
placeholder='secret_token'
),
)
class WebhookIntegrationType(IntegrationTypeBase):
key = 'webhook'
display_name = _('Webhook')
description = _('Post json events to a webhook endpoint')
icon = ''''''
valid_events = [
events.PullRequestCloseEvent,
events.PullRequestMergeEvent,
events.PullRequestUpdateEvent,
events.PullRequestCommentEvent,
events.PullRequestReviewEvent,
events.PullRequestCreateEvent,
events.RepoPushEvent,
events.RepoCreateEvent,
]
def settings_schema(self):
schema = WebhookSettingsSchema()
schema.add(colander.SchemaNode(
colander.Set(),
widget=deform.widget.CheckboxChoiceWidget(
values=sorted(
[(e.name, e.display_name) for e in self.valid_events]
)
),
description="Events activated for this integration",
name='events'
))
return schema
def send_event(self, event):
log.debug('handling event %s with webhook integration %s',
event.name, self)
if event.__class__ not in self.valid_events:
log.debug('event not valid: %r' % event)
return
if event.name not in self.settings['events']:
log.debug('event ignored: %r' % event)
return
data = event.as_dict()
post_to_webhook(data, self.settings)
@task(ignore_result=True)
def post_to_webhook(data, settings):
log.debug('sending event:%s to webhook %s', data['name'], settings['url'])
resp = requests.post(settings['url'], json={
'token': settings['secret_token'],
'event': data
})
resp.raise_for_status() # raise exception on a failed request