145 lines
5.5 KiB
Python
145 lines
5.5 KiB
Python
import json
|
|
import re
|
|
from buildbot.util import bytes2unicode
|
|
from buildbot.www.hooks.base import BaseHookHandler
|
|
|
|
from twisted.python import log
|
|
from dateutil.parser import parse as dateparse
|
|
|
|
_HEADER_EVENT_TYPE = 'X-Gitea-Event'
|
|
|
|
|
|
class GiteaHandler(BaseHookHandler):
|
|
|
|
def processPushEvent(self, payload, event_type, codebase):
|
|
refname = payload["ref"]
|
|
|
|
changes = []
|
|
|
|
# We only care about regular heads or tags
|
|
match = re.match(r"^refs/(heads|tags)/(.+)$", refname)
|
|
if not match:
|
|
log.msg("Ignoring refname '{}': Not a branch or tag".format(refname))
|
|
return changes
|
|
|
|
branch = match.group(2)
|
|
|
|
repository = payload['repository']
|
|
repo_url = repository['ssh_url']
|
|
project = repository['full_name']
|
|
|
|
commits = payload['commits']
|
|
if self.options.get('onlyIncludePushCommit', False):
|
|
commits = commits[:1]
|
|
|
|
for commit in commits:
|
|
timestamp = dateparse(commit['timestamp'])
|
|
change = {
|
|
'author': '{} <{}>'.format(commit['author']['name'],
|
|
commit['author']['email']),
|
|
'comments': commit['message'],
|
|
'revision': commit['id'],
|
|
'when_timestamp': timestamp,
|
|
'branch': branch,
|
|
'revlink': commit['url'],
|
|
'repository': repo_url,
|
|
'project': project,
|
|
'category': event_type,
|
|
'properties': {
|
|
'event': event_type,
|
|
'repository_name': repository['name'],
|
|
'owner': repository["owner"]["username"]
|
|
},
|
|
}
|
|
if codebase is not None:
|
|
change['codebase'] = codebase
|
|
changes.append(change)
|
|
return changes
|
|
|
|
def processPullRequestEvent(self, payload, event_type, codebase):
|
|
action = payload['action']
|
|
|
|
# Only handle potential new stuff, ignore close/.
|
|
# Merge itself is handled by the regular branch push message
|
|
if action not in ['opened', 'synchronized', 'edited', 'reopened']:
|
|
log.msg("Gitea Pull Request event '{}' ignored".format(action))
|
|
return []
|
|
pull_request = payload['pull_request']
|
|
if not pull_request['mergeable']:
|
|
log.msg("Gitea Pull Request ignored because it is not mergeable.")
|
|
return []
|
|
if pull_request['merged']:
|
|
log.msg("Gitea Pull Request ignored because it is already merged.")
|
|
return []
|
|
timestamp = dateparse(pull_request['updated_at'])
|
|
base = pull_request['base']
|
|
head = pull_request['head']
|
|
repository = payload['repository']
|
|
change = {
|
|
'author': '{} <{}>'.format(pull_request['user']['full_name'],
|
|
pull_request['user']['email']),
|
|
'comments': 'PR#{}: {}\n\n{}'.format(
|
|
pull_request['number'],
|
|
pull_request['title'],
|
|
pull_request['body']),
|
|
'revision': head['sha'],
|
|
'when_timestamp': timestamp,
|
|
'branch': head['ref'],
|
|
'revlink': pull_request['html_url'],
|
|
'repository': repository['ssh_url'],
|
|
'project': repository['full_name'],
|
|
'category': event_type,
|
|
'properties': {
|
|
'event': event_type,
|
|
'base_branch': base['ref'],
|
|
'base_sha': base['sha'],
|
|
'base_repo_id': base['repo_id'],
|
|
'base_repository': base['repo']['clone_url'],
|
|
'base_git_ssh_url': base['repo']['ssh_url'],
|
|
'head_branch': head['ref'],
|
|
'head_sha': head['sha'],
|
|
'head_repo_id': head['repo_id'],
|
|
'head_repository': head['repo']['clone_url'],
|
|
'head_git_ssh_url': head['repo']['ssh_url'],
|
|
'pr_id': pull_request['id'],
|
|
'pr_number': pull_request['number'],
|
|
'repository_name': repository['name'],
|
|
'owner': repository["owner"]["username"],
|
|
},
|
|
}
|
|
if codebase is not None:
|
|
change['codebase'] = codebase
|
|
return [change]
|
|
|
|
def getChanges(self, request):
|
|
secret = None
|
|
if isinstance(self.options, dict):
|
|
secret = self.options.get('secret')
|
|
try:
|
|
content = request.content.read()
|
|
payload = json.loads(bytes2unicode(content))
|
|
except Exception as exception:
|
|
raise ValueError('Error loading JSON: ' + str(exception))
|
|
if secret is not None and secret != payload['secret']:
|
|
raise ValueError('Invalid secret')
|
|
event_type = bytes2unicode(request.getHeader(_HEADER_EVENT_TYPE))
|
|
log.msg("Received event '{}' from gitea".format(event_type))
|
|
|
|
codebases = request.args.get('codebase', [None])
|
|
codebase = bytes2unicode(codebases[0])
|
|
changes = []
|
|
if event_type == 'push':
|
|
changes = self.processPushEvent(
|
|
payload, event_type, codebase)
|
|
elif event_type == 'pull_request':
|
|
changes = self.processPullRequestEvent(
|
|
payload, event_type, codebase)
|
|
else:
|
|
log.msg("Ignoring gitea event '{}'".format(event_type))
|
|
|
|
return (changes, 'git')
|
|
|
|
|
|
# Plugin name
|
|
gitea = GiteaHandler
|