# -*- coding: utf-8 -*-
# Copyright (C) 2012-2018 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/
import colander
import string
import collections
import logging
from mako import exceptions
from rhodecode.translation import _
log = logging.getLogger(__name__)
class IntegrationTypeBase(object):
""" Base class for IntegrationType plugins """
is_dummy = False
description = ''
@classmethod
def icon(cls):
return '''
'''
def __init__(self, settings):
"""
:param settings: dict of settings to be used for the integration
"""
self.settings = settings
def settings_schema(self):
"""
A colander schema of settings for the integration type
"""
return colander.Schema()
class EEIntegration(IntegrationTypeBase):
description = 'Integration available in RhodeCode EE edition.'
is_dummy = True
def __init__(self, name, key, settings=None):
self.display_name = name
self.key = key
super(EEIntegration, self).__init__(settings)
# Helpers #
# updating this required to update the `common_vars` as well.
WEBHOOK_URL_VARS = [
('event_name', 'Unique name of the event type, e.g pullrequest-update'),
('repo_name', 'Full name of the repository'),
('repo_type', 'VCS type of repository'),
('repo_id', 'Unique id of repository'),
('repo_url', 'Repository url'),
# extra repo fields
('extra:', 'Extra repo variables, read from its settings.'),
# special attrs below that we handle, using multi-call
('branch', 'Name of each branch submitted, if any.'),
('branch_head', 'Head ID of pushed branch (full sha of last commit), if any.'),
('commit_id', 'ID (full sha) of each commit submitted, if any.'),
# pr events vars
('pull_request_id', 'Unique ID of the pull request.'),
('pull_request_title', 'Title of the pull request.'),
('pull_request_url', 'Pull request url.'),
('pull_request_shadow_url', 'Pull request shadow repo clone url.'),
('pull_request_commits_uid', 'Calculated UID of all commits inside the PR. '
'Changes after PR update'),
# user who triggers the call
('username', 'User who triggered the call.'),
('user_id', 'User id who triggered the call.'),
]
# common vars for url template used for CI plugins. Shared with webhook
CI_URL_VARS = WEBHOOK_URL_VARS
class CommitParsingDataHandler(object):
def aggregate_branch_data(self, branches, commits):
branch_data = collections.OrderedDict()
for obj in branches:
branch_data[obj['name']] = obj
branches_commits = collections.OrderedDict()
for commit in commits:
if commit.get('git_ref_change'):
# special case for GIT that allows creating tags,
# deleting branches without associated commit
continue
commit_branch = commit['branch']
if commit_branch not in branches_commits:
_branch = branch_data[commit_branch] \
if commit_branch else commit_branch
branch_commits = {'branch': _branch,
'branch_head': '',
'commits': []}
branches_commits[commit_branch] = branch_commits
branch_commits = branches_commits[commit_branch]
branch_commits['commits'].append(commit)
branch_commits['branch_head'] = commit['raw_id']
return branches_commits
class WebhookDataHandler(CommitParsingDataHandler):
name = 'webhook'
def __init__(self, template_url, headers):
self.template_url = template_url
self.headers = headers
def get_base_parsed_template(self, data):
"""
initially parses the passed in template with some common variables
available on ALL calls
"""
# note: make sure to update the `WEBHOOK_URL_VARS` if this changes
common_vars = {
'repo_name': data['repo']['repo_name'],
'repo_type': data['repo']['repo_type'],
'repo_id': data['repo']['repo_id'],
'repo_url': data['repo']['url'],
'username': data['actor']['username'],
'user_id': data['actor']['user_id'],
'event_name': data['name']
}
extra_vars = {}
for extra_key, extra_val in data['repo']['extra_fields'].items():
extra_vars['extra__{}'.format(extra_key)] = extra_val
common_vars.update(extra_vars)
template_url = self.template_url.replace('${extra:', '${extra__')
return string.Template(template_url).safe_substitute(**common_vars)
def repo_push_event_handler(self, event, data):
url = self.get_base_parsed_template(data)
url_calls = []
branches_commits = self.aggregate_branch_data(
data['push']['branches'], data['push']['commits'])
if '${branch}' in url or '${branch_head}' in url or '${commit_id}' in url:
# call it multiple times, for each branch if used in variables
for branch, commit_ids in branches_commits.items():
branch_url = string.Template(url).safe_substitute(branch=branch)
if '${branch_head}' in branch_url:
# last commit in the aggregate is the head of the branch
branch_head = commit_ids['branch_head']
branch_url = string.Template(branch_url).safe_substitute(
branch_head=branch_head)
# call further down for each commit if used
if '${commit_id}' in branch_url:
for commit_data in commit_ids['commits']:
commit_id = commit_data['raw_id']
commit_url = string.Template(branch_url).safe_substitute(
commit_id=commit_id)
# register per-commit call
log.debug(
'register %s call(%s) to url %s',
self.name, event, commit_url)
url_calls.append(
(commit_url, self.headers, data))
else:
# register per-branch call
log.debug(
'register %s call(%s) to url %s',
self.name, event, branch_url)
url_calls.append(
(branch_url, self.headers, data))
else:
log.debug(
'register %s call(%s) to url %s', self.name, event, url)
url_calls.append((url, self.headers, data))
return url_calls
def repo_create_event_handler(self, event, data):
url = self.get_base_parsed_template(data)
log.debug(
'register %s call(%s) to url %s', self.name, event, url)
return [(url, self.headers, data)]
def pull_request_event_handler(self, event, data):
url = self.get_base_parsed_template(data)
log.debug(
'register %s call(%s) to url %s', self.name, event, url)
url = string.Template(url).safe_substitute(
pull_request_id=data['pullrequest']['pull_request_id'],
pull_request_title=data['pullrequest']['title'],
pull_request_url=data['pullrequest']['url'],
pull_request_shadow_url=data['pullrequest']['shadow_url'],
pull_request_commits_uid=data['pullrequest']['commits_uid'],
)
return [(url, self.headers, data)]
def __call__(self, event, data):
from rhodecode import events
if isinstance(event, events.RepoPushEvent):
return self.repo_push_event_handler(event, data)
elif isinstance(event, events.RepoCreateEvent):
return self.repo_create_event_handler(event, data)
elif isinstance(event, events.PullRequestEvent):
return self.pull_request_event_handler(event, data)
else:
raise ValueError(
'event type `%s` not in supported list: %s' % (
event.__class__, events))
def get_auth(settings):
from requests.auth import HTTPBasicAuth
username = settings.get('username')
password = settings.get('password')
if username and password:
return HTTPBasicAuth(username, password)
return None
def get_web_token(settings):
return settings['secret_token']
def get_url_vars(url_vars):
return '\n'.join(
'{} - {}'.format('${' + key + '}', explanation)
for key, explanation in url_vars)
def render_with_traceback(template, *args, **kwargs):
try:
return template.render(*args, **kwargs)
except Exception:
log.error(exceptions.text_error_template().render())
raise