Compare commits

..

No commits in common. "master" and "gitea-auth" have entirely different histories.

12 changed files with 668 additions and 783 deletions

View File

@ -1,8 +1,9 @@
# Buildbot Transifex Plugin
# Buildbot Gitea Plugin
[![PyPI version](https://badge.fury.io/py/buildbot-gitea.svg)](https://badge.fury.io/py/buildbot-gitea)
![GitHub](https://img.shields.io/github/license/lab132/buildbot-gitea)
[![Build Status](https://travis-ci.org/lab132/buildbot-gitea.svg?branch=master)](https://travis-ci.org/lab132/buildbot-gitea)
This plugin for buildbot adds integration support with gitea, featuring push hooks, commit status updates and a change source.
@ -122,6 +123,8 @@ The parameters are as follows:
| --- | --- |
| URL | The URL to the gitea instance. |
| `token` | Generate an access token in the profile you want the buildbot to impersonate. Make sure the account in gitea has access to the repositories. |
| `startDescription` | `Renderable` A short description when buildbot starts building on a change. Defaults to `Build started.` |
| `endDescription` | `Renderable` A short description when buildbot stops building on a change. Defaults to `Build done.` |
| `context` | `Renderable` The context is an identifier for this status, allowing to identify from which builder this came, defaults to `Interpolate('buildbot/%(prop:buildername)s')` |
| `context_pr` | `Renderable` The context message to use, when building on a pull request, allowing to identify from which builder this came, defaults to `Interpolate('buildbot/pull_request/%(prop:buildername)s')` |
| `warningAsSuccess` | Treat warnings as build as success to set in the build status of gitea. If false, warnings will be displayed as warnings. |
@ -134,20 +137,21 @@ Gitea supports OAuth2 authentication so it is possible to have buildbot communic
`./master.cfg`
```py
from buildbot.plugins import util
c['www']['auth'] = util.GiteaAuth(
from buildbot_gitea.auth import GiteaAuth
c['www']['auth'] = GiteaAuth(
endpoint="https://your-gitea-host",
client_id='oauth2-client-id',
client_id 'oauth2-client-id',
client_secret='oauth2-client-secret')
```
| Parameter | Value |
| --- | --- |
| `endpoint` | The URL to your Gitea app. Something like `https://gitea.example.com/` |
| `client_id` | The OAuth2 Client ID `GUID`, can be a `Secret`. |
| `client_secret` | The OAuth2 Client Secret provided, when creating the OAuth application in gitea. Can be a `Secret`. |
| `endpoint` | The URL to your Gitea app. |
| `client_id` | The OAuth2 Client ID |
| `client_secret` | The OAuth2 Client Secret |
Resources:
+ [Gitea OAuth2 Provider documentation](https://docs.gitea.io/en-us/oauth2-provider/)
+ [Buildbot OAuth2 documentation](https://docs.buildbot.net/current/developer/cls-auth.html?highlight=oauth2#buildbot.www.oauth2.OAuth2Auth)
+ [Buildbot OAuth2 source](https://github.com/buildbot/buildbot/blob/master/master/buildbot/www/oauth2.py)

19
buildbot_gitea/auth.py Normal file
View File

@ -0,0 +1,19 @@
from buildbot.www.oauth2 import OAuth2Auth
from urllib.parse import urljoin
class GiteaAuth(OAuth2Auth):
name = 'Gitea'
faIcon = 'mug-tea'
AUTH_URL = 'login/oauth/authorize'
TOKEN_URL = 'login/oauth/access_token'
def __init__(self, endpoint, client_id, client_secret):
super(GiteaAuth, self).__init__(client_id, client_secret)
self.resourceEndpoint = endpoint
self.authUri = urljoin(endpoint, self.AUTH_URL)
self.tokenUri = urljoin(endpoint, self.TOKEN_URL)
def getUserInfoFromOAuthClient(self, c):
return self.get(c, '/api/v1/user')

176
buildbot_gitea/reporter.py Normal file
View File

@ -0,0 +1,176 @@
# Based on the gitlab reporter from buildbot
from __future__ import absolute_import
from __future__ import print_function
from twisted.internet import defer
from twisted.python import log
from buildbot.process.properties import Interpolate
from buildbot.process.properties import Properties
from buildbot.process.results import CANCELLED
from buildbot.process.results import EXCEPTION
from buildbot.process.results import FAILURE
from buildbot.process.results import RETRY
from buildbot.process.results import SKIPPED
from buildbot.process.results import SUCCESS
from buildbot.process.results import WARNINGS
from buildbot.reporters import http
from buildbot.util import httpclientservice
import re
class GiteaStatusPush(http.HttpStatusPushBase):
name = "GiteaStatusPush"
neededDetails = dict(wantProperties=True)
ssh_url_match = re.compile(r"(ssh://)?[\w+\-\_]+@[\w\.\-\_]+:?(\d*/)?(?P<owner>[\w_\-\.]+)/(?P<repo_name>[\w_\-\.]+)(\.git)?")
@defer.inlineCallbacks
def reconfigService(self, baseURL, token,
startDescription=None, endDescription=None,
context=None, context_pr=None, verbose=False,
warningAsSuccess=False, **kwargs):
token = yield self.renderSecrets(token)
yield http.HttpStatusPushBase.reconfigService(self, **kwargs)
self.context = context or Interpolate('buildbot/%(prop:buildername)s')
self.context_pr = context_pr or \
Interpolate('buildbot/pull_request/%(prop:buildername)s')
self.startDescription = startDescription or 'Build started.'
self.endDescription = endDescription or 'Build done.'
if baseURL.endswith('/'):
baseURL = baseURL[:-1]
self.baseURL = baseURL
self._http = yield httpclientservice.HTTPClientService.getService(
self.master, baseURL,
headers={'Authorization': 'token {}'.format(token)},
debug=self.debug, verify=self.verify)
self.verbose = verbose
self.project_ids = {}
self.warningAsSuccess = warningAsSuccess
def createStatus(self,
project_owner, repo_name, sha, state, target_url=None,
description=None, context=None):
"""
:param project_owner: username of the owning user or organization
:param repo_name: name of the repository
:param sha: Full sha to create the status for.
:param state: one of the following 'pending', 'success', 'failed'
or 'cancelled'.
:param target_url: Target url to associate with this status.
:param description: Short description of the status.
:param context: Context of the result
:return: A deferred with the result from GitLab.
"""
payload = {'state': state}
if description is not None:
payload['description'] = description
if target_url is not None:
payload['target_url'] = target_url
if context is not None:
payload['context'] = context
return self._http.post(
'/api/v1/repos/{owner}/{repository}/statuses/{sha}'.format(
owner=project_owner,
repository=repo_name,
sha=sha
),
json=payload)
@defer.inlineCallbacks
def send(self, build):
props = Properties.fromDict(build['properties'])
props.master = self.master
if build['complete']:
state = {
SUCCESS: 'success',
WARNINGS: 'success' if self.warningAsSuccess else 'warning',
FAILURE: 'failure',
SKIPPED: 'success',
EXCEPTION: 'error',
RETRY: 'pending',
CANCELLED: 'error'
}.get(build['results'], 'failure')
description = yield props.render(self.endDescription)
else:
state = 'pending'
description = yield props.render(self.startDescription)
if 'pr_id' in props:
context = yield props.render(self.context_pr)
else:
context = yield props.render(self.context)
sourcestamps = build['buildset']['sourcestamps']
for sourcestamp in sourcestamps:
sha = sourcestamp['revision']
if sha is None:
# No special revision for this, so ignore it
continue
if 'repository_name' in props:
repository_name = props['repository_name']
else:
match = re.match(self.ssh_url_match, sourcestamp['repository'])
if match is not None:
repository_name = match.group("repo_name")
else:
log.msg(
"Could not send status, "
"build has no repository_name property for Gitea.")
continue
if 'owner' in props:
repository_owner = props['owner']
else:
match = re.match(self.ssh_url_match, sourcestamp['repository'])
if match is not None:
repository_owner = match.group("owner")
else:
log.msg(
"Could not send status, "
"build has no owner property for Gitea.")
continue
try:
target_url = build['url']
res = yield self.createStatus(
project_owner=repository_owner,
repo_name=repository_name,
sha=sha,
state=state,
target_url=target_url,
context=context,
description=description
)
if res.code not in (200, 201, 204):
message = yield res.json()
message = message.get('message', 'unspecified error')
log.msg(
'Could not send status "{state}" for '
'{repo} at {sha}: {code} : {message}'.format(
state=state,
repo=sourcestamp['repository'], sha=sha,
code=res.code,
message=message))
elif self.verbose:
log.msg(
'Status "{state}" sent for '
'{repo} at {sha}.'.format(
state=state,
repo=sourcestamp['repository'], sha=sha))
except Exception as e:
log.err(
e,
'Failed to send status "{state}" for '
'{repo} at {sha}'.format(
state=state,
repo=sourcestamp['repository'], sha=sha
))

View File

@ -0,0 +1,33 @@
from __future__ import absolute_import
from __future__ import print_function
from twisted.internet import defer
from twisted.python import log
from buildbot.steps.source.git import Git
class Gitea(Git):
"""
Source step that knows how to handle merge requests from
the Gitea webhook
"""
@defer.inlineCallbacks
def _fetch(self, arg):
res = yield super(Gitea, self)._fetch(arg)
if self.build.hasProperty("pr_id"):
remote = yield self._dovccmd(
['config', 'remote.pr_source.url'], collectStdout=True)
if remote is None or remote.strip() is '':
yield self._dovccmd(
['remote', 'add', 'pr_source',
self.build.getProperty("head_git_ssh_url", None)])
else:
yield self._dovccmd(
['remote', 'set-url', 'pr_source',
self.build.getProperty("head_git_ssh_url", None)])
yield self._dovccmd(['fetch', 'pr_source'])
res = yield self._dovccmd(['merge', self.build.getProperty("head_sha", None)])
defer.returnValue(res)

View File

@ -0,0 +1,88 @@
import json
import mock
from buildbot.process.properties import Secret
from buildbot.test.util.config import ConfigErrorsMixin
from buildbot.test.util.misc import TestReactorMixin
from buildbot.test.util import www
from twisted.internet import defer
from twisted.trial import unittest
from buildbot.secrets.manager import SecretManager
from buildbot.test.fake.secrets import FakeSecretStorage
from buildbot_gitea.auth import GiteaAuth
try:
import requests
except ImportError:
requests = None
class FakeResponse:
def __init__(self, _json):
self.json = lambda: _json
self.content = json.dumps(_json)
def raise_for_status(self):
pass
class TestGiteaAuth(TestReactorMixin, www.WwwTestMixin, ConfigErrorsMixin,
unittest.TestCase):
def setUp(self):
self.setUpTestReactor()
if requests is None:
raise unittest.SkipTest("Need to install requests to test oauth2")
self.patch(requests, 'request', mock.Mock(spec=requests.request))
self.patch(requests, 'post', mock.Mock(spec=requests.post))
self.patch(requests, 'get', mock.Mock(spec=requests.get))
self.giteaAuth = GiteaAuth(
'https://gitea.test',
'client-id',
'client-secret')
self._master = master = self.make_master(
url='h:/a/b/', auth=self.giteaAuth)
self.giteaAuth.reconfigAuth(master, master.config)
self.giteaAuth_secret = GiteaAuth(
'https://gitea.test',
Secret("client-id"),
Secret("client-secret"))
self._master = master = self.make_master(
url='h:/a/b/', auth=self.giteaAuth_secret)
fake_storage_service = FakeSecretStorage()
fake_storage_service.reconfigService(
secretdict={
"client-id": "secretClientId",
"client-secret": "secretClientSecret"
})
secret_service = SecretManager()
secret_service.services = [fake_storage_service]
secret_service.setServiceParent(self._master)
self.giteaAuth_secret.reconfigAuth(master, master.config)
@defer.inlineCallbacks
def test_getGiteaLoginURL(self):
res = yield self.giteaAuth.getLoginURL('http://redir')
exp = ("https://gitea.test/login/oauth/authorize?client_id=client-id&"
"redirect_uri=h%3A%2Fa%2Fb%2Fauth%2Flogin&response_type=code&"
"state=redirect%3Dhttp%253A%252F%252Fredir")
self.assertEqual(res, exp)
res = yield self.giteaAuth.getLoginURL(None)
exp = ("https://gitea.test/login/oauth/authorize?client_id=client-id&"
"redirect_uri=h%3A%2Fa%2Fb%2Fauth%2Flogin&response_type=code")
self.assertEqual(res, exp)
@defer.inlineCallbacks
def test_getGiteaLoginURL_with_secret(self):
res = yield self.giteaAuth_secret.getLoginURL('http://redir')
exp = ("https://gitea.test/login/oauth/authorize?client_id=secretClientId&"
"redirect_uri=h%3A%2Fa%2Fb%2Fauth%2Flogin&response_type=code&"
"state=redirect%3Dhttp%253A%252F%252Fredir")
self.assertEqual(res, exp)
res = yield self.giteaAuth_secret.getLoginURL(None)
exp = ("https://gitea.test/login/oauth/authorize?client_id=secretClientId&"
"redirect_uri=h%3A%2Fa%2Fb%2Fauth%2Flogin&response_type=code")
self.assertEqual(res, exp)

View File

@ -0,0 +1,183 @@
# Based on TestGitLabStatusPush from buildbot
from __future__ import absolute_import
from __future__ import print_function
from mock import Mock
from twisted.internet import defer
from twisted.trial import unittest
from buildbot import config
from buildbot.process.properties import Interpolate
from buildbot.process.results import FAILURE
from buildbot.process.results import SUCCESS
from buildbot_gitea.reporter import GiteaStatusPush
from buildbot.test.fake import fakemaster
from buildbot.test.fake import httpclientservice as fakehttpclientservice
from buildbot.test.util import logging
from buildbot.test.util.reporter import ReporterTestMixin
from buildbot.test.util.misc import TestReactorMixin
class TestGiteaStatusPush(
unittest.TestCase,
ReporterTestMixin,
logging.LoggingMixin,
TestReactorMixin):
# repository must be in the form http://gitea/<owner>/<project>
TEST_REPO = u'http://gitea/buildbot/buildbot'
@defer.inlineCallbacks
def setUp(self):
self.setUpTestReactor()
# ignore config error if txrequests is not installed
self.patch(config, '_errors', Mock())
self.master = fakemaster.make_master(testcase=self,
wantData=True, wantDb=True, wantMq=True)
yield self.master.startService()
self._http = yield fakehttpclientservice.HTTPClientService.getFakeService(
self.master, self,
"http://gitea", headers={'Authorization': 'token XXYYZZ'},
debug=None, verify=None)
self.sp = sp = GiteaStatusPush("http://gitea/", Interpolate('XXYYZZ'))
sp.sessionFactory = Mock(return_value=Mock())
yield sp.setServiceParent(self.master)
def tearDown(self):
return self.master.stopService()
def setupProps(self):
self.TEST_PROPS['owner'] = "buildbot"
self.TEST_PROPS['repository_name'] = "buildbot"
@defer.inlineCallbacks
def setupBuildResults(self, buildResults):
self.insertTestData([buildResults], buildResults)
build = yield self.master.data.get(("builds", 20))
defer.returnValue(build)
@defer.inlineCallbacks
def test_basic(self):
self.setupProps()
build = yield self.setupBuildResults(SUCCESS)
# we make sure proper calls to txrequests have been made
self._http.expect(
'post',
'/api/v1/repos/buildbot/buildbot/statuses/d34db33fd43db33f',
json={'state': 'pending',
'target_url': 'http://localhost:8080/#builders/79/builds/0',
'description': 'Build started.', 'context': 'buildbot/Builder0'})
self._http.expect(
'post',
'/api/v1/repos/buildbot/buildbot/statuses/d34db33fd43db33f',
json={'state': 'success',
'target_url': 'http://localhost:8080/#builders/79/builds/0',
'description': 'Build done.', 'context': 'buildbot/Builder0'})
self._http.expect(
'post',
'/api/v1/repos/buildbot/buildbot/statuses/d34db33fd43db33f',
json={'state': 'failure',
'target_url': 'http://localhost:8080/#builders/79/builds/0',
'description': 'Build done.', 'context': 'buildbot/Builder0'})
build['complete'] = False
self.sp.buildStarted(("build", 20, "started"), build)
build['complete'] = True
self.sp.buildFinished(("build", 20, "finished"), build)
build['results'] = FAILURE
self.sp.buildFinished(("build", 20, "finished"), build)
@defer.inlineCallbacks
def test_sshurl(self):
self.setupProps()
self.TEST_REPO = u'git@gitea:buildbot/buildbot.git'
build = yield self.setupBuildResults(SUCCESS)
# we make sure proper calls to txrequests have been made
self._http.expect(
'post',
'/api/v1/repos/buildbot/buildbot/statuses/d34db33fd43db33f',
json={'state': 'pending',
'target_url': 'http://localhost:8080/#builders/79/builds/0',
'description': 'Build started.', 'context': 'buildbot/Builder0'})
build['complete'] = False
self.sp.buildStarted(("build", 20, "started"), build)
@defer.inlineCallbacks
def test_sshurl_noprops(self):
self.TEST_REPO = u'git@gitea:buildbot/buildbot.git'
build = yield self.setupBuildResults(SUCCESS)
# we make sure proper calls to txrequests have been made
self._http.expect(
'post',
'/api/v1/repos/buildbot/buildbot/statuses/d34db33fd43db33f',
json={'state': 'pending',
'target_url': 'http://localhost:8080/#builders/79/builds/0',
'description': 'Build started.', 'context': 'buildbot/Builder0'})
build['complete'] = False
self.sp.buildStarted(("build", 20, "started"), build)
@defer.inlineCallbacks
def test_noowner(self):
self.setUpLogging()
self.setupProps()
del self.TEST_PROPS["owner"]
self.TEST_REPO = u''
build = yield self.setupBuildResults(SUCCESS)
build['complete'] = False
self.sp.buildStarted(("build", 20, "started"), build)
# implicit check that no http request is done
self.assertLogged("Could not send status, "
"build has no owner property for Gitea.")
@defer.inlineCallbacks
def test_noreponame(self):
self.setUpLogging()
self.setupProps()
del self.TEST_PROPS["repository_name"]
self.TEST_REPO = u''
build = yield self.setupBuildResults(SUCCESS)
build['complete'] = False
self.sp.buildStarted(("build", 20, "started"), build)
# implicit check that no http request is done
self.assertLogged("Could not send status, "
"build has no repository_name property for Gitea.")
@defer.inlineCallbacks
def test_senderror(self):
self.setupProps()
self.setUpLogging()
build = yield self.setupBuildResults(SUCCESS)
# we make sure proper calls to txrequests have been made
self._http.expect(
'post',
'/api/v1/repos/buildbot/buildbot/statuses/d34db33fd43db33f',
json={'state': 'pending',
'target_url': 'http://localhost:8080/#builders/79/builds/0',
'description': 'Build started.', 'context': 'buildbot/Builder0'},
content_json={
"message": "sha1 not found: d34db33fd43db33f",
"url": "https://godoc.org/github.com/go-gitea/go-sdk/gitea"
},
code=500)
build['complete'] = False
self.sp.buildStarted(("build", 20, "started"), build)
self.assertLogged(
"Could not send status \"pending\" for "
"http://gitea/buildbot/buildbot at d34db33fd43db33f:"
" 500 : sha1 not found: d34db33fd43db33f")
@defer.inlineCallbacks
def test_badchange(self):
self.setupProps()
self.setUpLogging()
build = yield self.setupBuildResults(SUCCESS)
# we make sure proper calls to txrequests have been made
build['complete'] = False
self.sp.buildStarted(("build", 20, "started"), build)
self.assertLogged("Failed to send status \"pending\" for"
" http://gitea/buildbot/buildbot at d34db33fd43db33f")
self.flushLoggedErrors(AssertionError)

View File

@ -0,0 +1,107 @@
# This file is part of Buildbot. Buildbot is free software: you can
# redistribute it and/or modify it under the terms of the GNU General Public
# License as published by the Free Software Foundation, version 2.
#
# 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 General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright Buildbot Team Members
from __future__ import absolute_import
from __future__ import print_function
from twisted.trial import unittest
from buildbot.process.results import SUCCESS
from buildbot_gitea.step_source import Gitea
from buildbot.test.fake.remotecommand import Expect
from buildbot.test.fake.remotecommand import ExpectShell
from buildbot.test.util import config
from buildbot.test.util import sourcesteps
from buildbot.test.util.misc import TestReactorMixin
class TestGitea(sourcesteps.SourceStepMixin, config.ConfigErrorsMixin, unittest.TestCase, TestReactorMixin):
stepClass = Gitea
def setUp(self):
self.setUpTestReactor()
self.sourceName = self.stepClass.__name__
return self.setUpSourceStep()
def setupStep(self, step, **kwargs):
step = sourcesteps.SourceStepMixin.setupStep(self, step, **kwargs)
step.build.properties.setProperty("pr_id", "1", "gitea pr id")
step.build.properties.setProperty("base_sha", "f6ad368298bd941e934a41f3babc827b2aa95a1d", "gitea source branch")
step.build.properties.setProperty("base_branch", "master", "gitea source branch")
step.build.properties.setProperty("base_git_ssh_url",
"git@gitea.example.com:base/awesome_project.git",
"gitea source git ssh url")
step.build.properties.setProperty("head_sha", "e4cd1224c622d46a8199c85c858485723115d2c8", "gitea target sha")
step.build.properties.setProperty("head_branch", "feature-branch", "gitea target branch")
step.build.properties.setProperty("head_git_ssh_url",
"git@gitea.example.com:target/awesome_project.git",
"gitea target git ssh url")
return step
def tearDown(self):
return self.tearDownSourceStep()
def test_with_merge_branch(self):
self.setupStep(
Gitea(repourl='git@gitea.example.com:base/awesome_project.git',
mode='full', method='clean'))
self.expectCommands(
ExpectShell(workdir='wkdir',
command=['git', '--version'])
+ ExpectShell.log('stdio',
stdout='git version 1.7.5')
+ 0,
Expect('stat', dict(file='wkdir/.buildbot-patched',
logEnviron=True))
+ 1,
Expect('listdir', {'dir': 'wkdir', 'logEnviron': True,
'timeout': 1200})
+ Expect.update('files', ['.git'])
+ 0,
ExpectShell(workdir='wkdir',
command=['git', 'clean', '-f', '-f', '-d'])
+ 0,
# here we always ignore revision, and fetch the merge branch
ExpectShell(workdir='wkdir',
command=['git', 'fetch', '-t',
'git@gitea.example.com:base/awesome_project.git', 'HEAD', '--progress'])
+ 0,
ExpectShell(workdir='wkdir',
command=['git', 'reset', '--hard', 'FETCH_HEAD', '--'])
+ 0,
ExpectShell(workdir='wkdir',
command=['git', 'config', 'remote.pr_source.url'])
+ 0,
ExpectShell(workdir='wkdir',
command=['git', 'remote', 'add', 'pr_source', 'git@gitea.example.com:target/awesome_project.git'])
+ 0,
ExpectShell(workdir='wkdir',
command=['git', 'fetch', 'pr_source'])
+ 0,
ExpectShell(workdir='wkdir',
command=['git', 'merge', 'e4cd1224c622d46a8199c85c858485723115d2c8'])
+ 0,
ExpectShell(workdir='wkdir',
command=['git', 'rev-parse', 'HEAD'])
+ ExpectShell.log('stdio',
stdout='e4cd1224c622d46a8199c85c858485723115d2c8')
+ 0,
)
self.expectOutcome(result=SUCCESS)
self.expectProperty(
'got_revision', 'e4cd1224c622d46a8199c85c858485723115d2c8', 'Gitea')
return self.runStep()

View File

@ -2,18 +2,13 @@ import buildbot.www.change_hook as change_hook
from buildbot.test.fake.web import FakeRequest
from buildbot.test.fake.web import fakeMasterForHooks
from buildbot.test.util.misc import TestReactorMixin
from buildbot.secrets.manager import SecretManager
from buildbot.test.fake.secrets import FakeSecretStorage
from buildbot.process.properties import Secret
from twisted.internet import defer
from twisted.trial import unittest
from buildbot_gitea.webhook import GiteaHandler, _HEADER_EVENT_TYPE, _HEADER_SIGNATURE
giteaJsonPushPayload_Signature = 'b5feb0994ad24c209188d36a30cecfea86666aa9c65a419b068f73f91152e7bc'
from buildbot_gitea.webhook import GiteaHandler, _HEADER_EVENT_TYPE
giteaJsonPushPayload = rb"""
{
@ -116,130 +111,6 @@ giteaJsonPushPayload = rb"""
}
"""
giteaJsonPushModifiedFiles = rb"""
{
"secret": "pass",
"ref": "refs/heads/master",
"before": "92a8bf0e02b2146e6b35b71d6e08c376133b7fc9",
"after": "ea07c3148db428876add8b312256239275c395fb",
"compare_url": "https://git.example.com/Test/test/compare/92a8bf0e02b2146e6b35b71d6e08c376133b7fc9...ea07c3148db428876add8b312256239275c395fb",
"commits": [
{
"id": "ea07c3148db428876add8b312256239275c395fb",
"message": "test123\n",
"url": "https://git.example.com/Test/test/commit/ea07c3148db428876add8b312256239275c395fb",
"author": {
"name": "Test User",
"email": "Test@example.com",
"username": "Test"
},
"committer": {
"name": "Test User",
"email": "Test@example.com",
"username": "Test"
},
"verification": null,
"timestamp": "2021-03-09T20:12:19Z",
"added": [
"testfile2"
],
"removed": [
"testfile3"
],
"modified": [
"testfile1"
]
}
],
"head_commit": null,
"repository": {
"id": 29,
"owner": {
"id": 1,
"login": "Test",
"full_name": "Test User",
"email": "Test@example.com",
"avatar_url": "https://git.example.com/user/avatar/Test/-1",
"language": "en-US",
"is_admin": true,
"last_login": "2021-03-09T20:10:52Z",
"created": "2018-06-05T09:41:06Z",
"username": "Test"
},
"name": "test",
"full_name": "Test/test",
"description": "",
"empty": false,
"private": true,
"fork": false,
"template": false,
"parent": null,
"mirror": false,
"size": 17,
"html_url": "https://git.example.com/Test/test",
"ssh_url": "ssh://git@git.example.com/Test/test.git",
"clone_url": "https://git.example.com/Test/test.git",
"original_url": "",
"website": "",
"stars_count": 0,
"forks_count": 0,
"watchers_count": 1,
"open_issues_count": 0,
"open_pr_counter": 0,
"release_counter": 0,
"default_branch": "master",
"archived": false,
"created_at": "2019-03-03T17:26:23Z",
"updated_at": "2021-03-09T20:12:20Z",
"permissions": {
"admin": true,
"push": true,
"pull": true
},
"has_issues": true,
"internal_tracker": {
"enable_time_tracker": false,
"allow_only_contributors_to_track_time": true,
"enable_issue_dependencies": true
},
"has_wiki": true,
"has_pull_requests": true,
"has_projects": false,
"ignore_whitespace_conflicts": false,
"allow_merge_commits": true,
"allow_rebase": true,
"allow_rebase_explicit": true,
"allow_squash_merge": true,
"avatar_url": "",
"internal": false
},
"pusher": {
"id": 1,
"login": "Test",
"full_name": "Test User",
"email": "Test@example.com",
"avatar_url": "https://git.example.com/user/avatar/Test/-1",
"language": "en-US",
"is_admin": true,
"last_login": "2021-03-09T20:10:52Z",
"created": "2018-06-05T09:41:06Z",
"username": "Test"
},
"sender": {
"id": 1,
"login": "Test",
"full_name": "Test User",
"email": "Test@example.com",
"avatar_url": "https://git.example.com/user/avatar/Test/-1",
"language": "en-US",
"is_admin": true,
"last_login": "2021-03-09T20:10:52Z",
"created": "2018-06-05T09:41:06Z",
"username": "Test"
}
}
"""
giteaInvalidSecretPush = rb"""
{
"secret": "invalidSecret",
@ -324,8 +195,6 @@ giteaInvalidSecretPush = rb"""
}
"""
giteaJsonPullRequestPayload_Signature = '8685905c03fa521dd1eacfb84405195dbca2a08206c3a978a3656399f5dbe01a'
giteaJsonPullRequestPayload = rb"""
{
"secret": "test",
@ -501,9 +370,6 @@ giteaJsonPullRequestPayload = rb"""
}
"""
giteaJsonPullRequestPayloadNotMergeable_Signature = '5552a0cbcbb3fe6286681bc7846754929be0d3f27ccc32914e5fd3ce01f34632'
giteaJsonPullRequestPayloadNotMergeable = rb"""
{
"secret": "test",
@ -679,7 +545,6 @@ giteaJsonPullRequestPayloadNotMergeable = rb"""
}
"""
giteaJsonPullRequestPayloadMerged_Signature = '4d3b1045aea9aa5cce4f7270d549c11d212c55036d9c547d0c9327891d56bf97'
giteaJsonPullRequestPayloadMerged = rb"""
{
"secret": "test",
@ -855,427 +720,6 @@ giteaJsonPullRequestPayloadMerged = rb"""
}
"""
giteaJsonPullRequestFork = rb"""
{
"secret": "test",
"action": "opened",
"number": 4,
"pull_request": {
"id": 36,
"url": "https://git.example.com/testuser/webhook_test/pulls/4",
"number": 4,
"user": {
"id": 1,
"login": "testuser",
"full_name": "testuser name",
"email": "testuser@example.com",
"avatar_url": "https://git.example.com/user/avatar/testuser/-1",
"language": "en-US",
"is_admin": true,
"last_login": "2021-03-27T13:53:28Z",
"created": "2018-06-05T09:41:06Z",
"username": "testuser"
},
"title": "testfork",
"body": "Test PR",
"labels": [],
"milestone": null,
"assignee": null,
"assignees": null,
"state": "open",
"is_locked": false,
"comments": 0,
"html_url": "https://git.example.com/testuser/webhook_test/pulls/4",
"diff_url": "https://git.example.com/testuser/webhook_test/pulls/4.diff",
"patch_url": "https://git.example.com/testuser/webhook_test/pulls/4.patch",
"mergeable": true,
"merged": false,
"merged_at": null,
"merge_commit_sha": null,
"merged_by": null,
"base": {
"label": "master",
"ref": "master",
"sha": "449a5a8ca05607106b5ba41988c1a658a8949a18",
"repo_id": 20,
"repo": {
"id": 20,
"owner": {
"id": 1,
"login": "testuser",
"full_name": "testuser name",
"email": "testuser@example.com",
"avatar_url": "https://git.example.com/user/avatar/testuser/-1",
"language": "en-US",
"is_admin": true,
"last_login": "2021-03-27T13:53:28Z",
"created": "2018-06-05T09:41:06Z",
"username": "testuser"
},
"name": "webhook_test",
"full_name": "testuser/webhook_test",
"description": "",
"empty": false,
"private": true,
"fork": false,
"template": false,
"parent": null,
"mirror": false,
"size": 76,
"html_url": "https://git.example.com/testuser/webhook_test",
"ssh_url": "ssh://git@git.example.com/testuser/webhook_test.git",
"clone_url": "https://git.example.com/testuser/webhook_test.git",
"original_url": "",
"website": "",
"stars_count": 0,
"forks_count": 1,
"watchers_count": 1,
"open_issues_count": 0,
"open_pr_counter": 1,
"release_counter": 0,
"default_branch": "master",
"archived": false,
"created_at": "2018-09-04T10:45:23Z",
"updated_at": "2018-09-04T13:05:51Z",
"permissions": {
"admin": false,
"push": false,
"pull": false
},
"has_issues": true,
"internal_tracker": {
"enable_time_tracker": false,
"allow_only_contributors_to_track_time": true,
"enable_issue_dependencies": true
},
"has_wiki": true,
"has_pull_requests": true,
"has_projects": false,
"ignore_whitespace_conflicts": false,
"allow_merge_commits": true,
"allow_rebase": true,
"allow_rebase_explicit": true,
"allow_squash_merge": true,
"avatar_url": "",
"internal": false
}
},
"head": {
"label": "feature_branch",
"ref": "feature_branch",
"sha": "53e3075cbe468f14c2801d186d703e64b2adee12",
"repo_id": 34,
"repo": {
"id": 34,
"owner": {
"id": 14,
"login": "test_org",
"full_name": "",
"email": "",
"avatar_url": "https://git.example.com/user/avatar/test_org/-1",
"language": "",
"is_admin": false,
"last_login": "1970-01-01T00:00:00Z",
"created": "2018-09-27T11:35:41Z",
"username": "test_org"
},
"name": "webhook_test_fork",
"full_name": "test_org/webhook_test_fork",
"description": "",
"empty": false,
"private": true,
"fork": true,
"template": false,
"parent": {
"id": 20,
"owner": {
"id": 1,
"login": "testuser",
"full_name": "testuser name",
"email": "testuser@example.com",
"avatar_url": "https://git.example.com/user/avatar/testuser/-1",
"language": "en-US",
"is_admin": true,
"last_login": "2021-03-27T13:53:28Z",
"created": "2018-06-05T09:41:06Z",
"username": "testuser"
},
"name": "webhook_test",
"full_name": "testuser/webhook_test",
"description": "",
"empty": false,
"private": true,
"fork": false,
"template": false,
"parent": null,
"mirror": false,
"size": 76,
"html_url": "https://git.example.com/testuser/webhook_test",
"ssh_url": "ssh://git@git.example.com/testuser/webhook_test.git",
"clone_url": "https://git.example.com/testuser/webhook_test.git",
"original_url": "",
"website": "",
"stars_count": 0,
"forks_count": 1,
"watchers_count": 1,
"open_issues_count": 0,
"open_pr_counter": 2,
"release_counter": 0,
"default_branch": "master",
"archived": false,
"created_at": "2018-09-04T10:45:23Z",
"updated_at": "2018-09-04T13:05:51Z",
"permissions": {
"admin": false,
"push": false,
"pull": false
},
"has_issues": true,
"internal_tracker": {
"enable_time_tracker": false,
"allow_only_contributors_to_track_time": true,
"enable_issue_dependencies": true
},
"has_wiki": true,
"has_pull_requests": true,
"has_projects": false,
"ignore_whitespace_conflicts": false,
"allow_merge_commits": true,
"allow_rebase": true,
"allow_rebase_explicit": true,
"allow_squash_merge": true,
"avatar_url": "",
"internal": false
},
"mirror": false,
"size": 19,
"html_url": "https://git.example.com/test_org/webhook_test_fork",
"ssh_url": "ssh://git@git.example.com/test_org/webhook_test_fork.git",
"clone_url": "https://git.example.com/test_org/webhook_test_fork.git",
"original_url": "",
"website": "",
"stars_count": 0,
"forks_count": 0,
"watchers_count": 1,
"open_issues_count": 0,
"open_pr_counter": 0,
"release_counter": 0,
"default_branch": "master",
"archived": false,
"created_at": "2021-03-28T21:40:46Z",
"updated_at": "2021-03-28T21:41:01Z",
"permissions": {
"admin": false,
"push": false,
"pull": false
},
"has_issues": true,
"internal_tracker": {
"enable_time_tracker": false,
"allow_only_contributors_to_track_time": true,
"enable_issue_dependencies": true
},
"has_wiki": true,
"has_pull_requests": true,
"has_projects": true,
"ignore_whitespace_conflicts": false,
"allow_merge_commits": true,
"allow_rebase": true,
"allow_rebase_explicit": true,
"allow_squash_merge": true,
"avatar_url": "",
"internal": false
}
},
"merge_base": "449a5a8ca05607106b5ba41988c1a658a8949a18",
"due_date": null,
"created_at": "2021-03-28T21:41:24Z",
"updated_at": "2021-03-28T21:41:24Z",
"closed_at": null
},
"repository": {
"id": 20,
"owner": {
"id": 1,
"login": "testuser",
"full_name": "testuser name",
"email": "testuser@example.com",
"avatar_url": "https://git.example.com/user/avatar/testuser/-1",
"language": "en-US",
"is_admin": true,
"last_login": "2021-03-27T13:53:28Z",
"created": "2018-06-05T09:41:06Z",
"username": "testuser"
},
"name": "webhook_test",
"full_name": "testuser/webhook_test",
"description": "",
"empty": false,
"private": true,
"fork": false,
"template": false,
"parent": null,
"mirror": false,
"size": 76,
"html_url": "https://git.example.com/testuser/webhook_test",
"ssh_url": "ssh://git@git.example.com/testuser/webhook_test.git",
"clone_url": "https://git.example.com/testuser/webhook_test.git",
"original_url": "",
"website": "",
"stars_count": 0,
"forks_count": 1,
"watchers_count": 1,
"open_issues_count": 0,
"open_pr_counter": 2,
"release_counter": 0,
"default_branch": "master",
"archived": false,
"created_at": "2018-09-04T10:45:23Z",
"updated_at": "2018-09-04T13:05:51Z",
"permissions": {
"admin": true,
"push": true,
"pull": true
},
"has_issues": true,
"internal_tracker": {
"enable_time_tracker": false,
"allow_only_contributors_to_track_time": true,
"enable_issue_dependencies": true
},
"has_wiki": true,
"has_pull_requests": true,
"has_projects": false,
"ignore_whitespace_conflicts": false,
"allow_merge_commits": true,
"allow_rebase": true,
"allow_rebase_explicit": true,
"allow_squash_merge": true,
"avatar_url": "",
"internal": false
},
"sender": {
"id": 1,
"login": "testuser",
"full_name": "testuser name",
"email": "testuser@example.com",
"avatar_url": "https://git.example.com/user/avatar/testuser/-1",
"language": "en-US",
"is_admin": true,
"last_login": "2021-03-27T13:53:28Z",
"created": "2018-06-05T09:41:06Z",
"username": "testuser"
},
"review": null
}
"""
giteaJsonPushEmptyFiles = rb"""
{
"secret": "pass",
"ref": "refs/heads/develop",
"before": "2437bd7c6b0af7b8da570973c02f0cca07ec787d",
"after": "2437bd7c6b0af7b8da570973c02f0cca07ec787d",
"compare_url": "",
"commits": [
{
"id": "2437bd7c6b0af7b8da570973c02f0cca07ec787d",
"message": "snip",
"url": "snip/commit/2437bd7c6b0af7b8da570973c02f0cca07ec787d",
"author": {
"name": "snip",
"email": "snip",
"username": ""
},
"committer": {
"name": "snip",
"email": "snip",
"username": ""
},
"verification": null,
"timestamp": "0001-01-01T00:00:00Z",
"added": null,
"removed": null,
"modified": null
}
],
"head_commit": {
"id": "2437bd7c6b0af7b8da570973c02f0cca07ec787d",
"message": "snip",
"url": "snip/commit/2437bd7c6b0af7b8da570973c02f0cca07ec787d",
"author": {
"name": "snip",
"email": "snip",
"username": ""
},
"committer": {
"name": "snip",
"email": "snip",
"username": ""
},
"verification": null,
"timestamp": "0001-01-01T00:00:00Z",
"added": null,
"removed": null,
"modified": null
},
"repository": {
"id": "snip",
"owner": {"id":"snip","login":"snip","full_name":"snip","email":"snip","avatar_url":"snip","language":"","is_admin":false,"last_login":"0001-01-01T00:00:00Z","created":"2019-07-04T02:15:26+02:00","restricted":false,"active":false,"prohibit_login":false,"location":"","website":"","description":"","visibility":"public","followers_count":0,"following_count":0,"starred_repos_count":0,"username":"snip"},
"name": "snip",
"full_name": "snip",
"description": "snip",
"empty": false,
"private": true,
"fork": false,
"template": false,
"parent": null,
"mirror": false,
"size": 19106,
"html_url": "snip",
"ssh_url": "git@snip.git",
"clone_url": "snip.git",
"original_url": "",
"website": "",
"stars_count": 0,
"forks_count": 0,
"watchers_count": 5,
"open_issues_count": 33,
"open_pr_counter": 1,
"release_counter": 0,
"default_branch": "develop",
"archived": false,
"created_at": "2020-08-25T09:34:29+02:00",
"updated_at": "2022-01-19T15:55:11+01:00",
"permissions": {
"admin": false,
"push": false,
"pull": false
},
"has_issues": true,
"internal_tracker": {
"enable_time_tracker": false,
"allow_only_contributors_to_track_time": true,
"enable_issue_dependencies": true
},
"has_wiki": true,
"has_pull_requests": true,
"has_projects": false,
"ignore_whitespace_conflicts": false,
"allow_merge_commits": true,
"allow_rebase": true,
"allow_rebase_explicit": false,
"allow_squash_merge": true,
"default_merge_style": "merge",
"avatar_url": "",
"internal": false,
"mirror_interval": ""
},
"pusher": {"id":"snip","login":"snip","full_name":"snip","email":"snip","avatar_url":"snip","language":"","is_admin":false,"last_login":"0001-01-01T00:00:00Z","created":"2021-01-22T02:15:29+01:00","restricted":false,"active":false,"prohibit_login":false,"location":"","website":"","description":"","visibility":"public","followers_count":0,"following_count":0,"starred_repos_count":0,"username":"snip"},
"sender": {"id":"snip","login":"snip","full_name":"snip","email":"snip","avatar_url":"snip","language":"","is_admin":false,"last_login":"0001-01-01T00:00:00Z","created":"2021-01-22T02:15:29+01:00","restricted":false,"active":false,"prohibit_login":false,"location":"","website":"","description":"","visibility":"public","followers_count":0,"following_count":0,"starred_repos_count":0,"username":"snip"}
}
"""
class TestChangeHookGiteaPush(unittest.TestCase, TestReactorMixin):
def setUp(self):
@ -1289,23 +733,6 @@ class TestChangeHookGiteaPush(unittest.TestCase, TestReactorMixin):
change = self.changeHook.master.data.updates.changesAdded[0]
self.assertEqual(change['repository'], 'ssh://git@git.example.com/max/webhook_test.git')
self.assertEqual(
change["author"], "Max Mustermann <max@example.com>")
self.assertEqual(
change["revision"], 'ad7157cc4a137b3e1dfe92750ccfb1bbad239f9a')
self.assertEqual(
change["when_timestamp"],
1536063014)
self.assertEqual(
change["comments"], "TestBranch2\n")
self.assertEqual(change["branch"], "feature-branch")
self.assertEqual(change[
"revlink"],
"https://git.example.com/max/webhook_test/commit/ad7157cc4a137b3e1dfe92750ccfb1bbad239f9a")
change = self.changeHook.master.data.updates.changesAdded[1]
self.assertEqual(change['repository'], 'ssh://git@git.example.com/max/webhook_test.git')
self.assertEqual(
change["author"], "Max Mustermann <max@example.com>")
self.assertEqual(
@ -1319,40 +746,22 @@ class TestChangeHookGiteaPush(unittest.TestCase, TestReactorMixin):
self.assertEqual(change[
"revlink"],
"https://git.example.com/max/webhook_test/commit/9d7157cc4a137b3e1dfe92750ccfb1bbad239f99")
def checkFileChanges(self, codebase=None):
self.assertEqual(len(self.changeHook.master.data.updates.changesAdded), 1)
change = self.changeHook.master.data.updates.changesAdded[0]
self.assertEqual(change['repository'], 'ssh://git@git.example.com/Test/test.git')
change = self.changeHook.master.data.updates.changesAdded[1]
self.assertEqual(change['repository'], 'ssh://git@git.example.com/max/webhook_test.git')
self.assertEqual(
change["author"], "Test User <Test@example.com>")
change["author"], "Max Mustermann <max@example.com>")
self.assertEqual(
change["revision"], 'ea07c3148db428876add8b312256239275c395fb')
change["revision"], 'ad7157cc4a137b3e1dfe92750ccfb1bbad239f9a')
self.assertEqual(
change["comments"], "test123\n")
self.assertEqual(change["branch"], "master")
change["when_timestamp"],
1536063014)
self.assertEqual(
change["comments"], "TestBranch2\n")
self.assertEqual(change["branch"], "feature-branch")
self.assertEqual(change[
"revlink"],
"https://git.example.com/Test/test/commit/ea07c3148db428876add8b312256239275c395fb")
self.assertEqual(change["files"], ["testfile2", "testfile1", "testfile3"])
def checkNoFileChanges(self, codebase=None):
self.assertEqual(len(self.changeHook.master.data.updates.changesAdded), 1)
change = self.changeHook.master.data.updates.changesAdded[0]
self.assertEqual(change['repository'], 'git@snip.git')
self.assertEqual(
change["author"], "snip <snip>")
self.assertEqual(
change["revision"], '2437bd7c6b0af7b8da570973c02f0cca07ec787d')
self.assertEqual(
change["comments"], "snip")
self.assertEqual(change["branch"], "develop")
self.assertEqual(change[
"revlink"],
"snip/commit/2437bd7c6b0af7b8da570973c02f0cca07ec787d")
self.assertEqual(change["files"], [])
"https://git.example.com/max/webhook_test/commit/ad7157cc4a137b3e1dfe92750ccfb1bbad239f9a")
def checkChangesFromPullRequest(self, codebase=None):
self.assertEqual(len(self.changeHook.master.data.updates.changesAdded), 1)
@ -1362,13 +771,13 @@ class TestChangeHookGiteaPush(unittest.TestCase, TestReactorMixin):
self.assertEqual(
change["author"], "Max Mustermann <max@example.com>")
self.assertEqual(
change["revision"], '7c5de0796c409e7802abe759113d7fc37e0d6578')
change["revision"], '9d7157cc4a137b3e1dfe92750ccfb1bbad239f99')
self.assertEqual(
change["when_timestamp"],
1536063289)
self.assertEqual(
change["comments"], "PR#1: TestPR\n\n")
self.assertEqual(change["branch"], "master")
self.assertEqual(change["branch"], "feature-branch")
self.assertEqual(change[
"revlink"], "https://git.example.com/max/webhook_test/pulls/1")
properties = change["properties"]
@ -1385,86 +794,30 @@ class TestChangeHookGiteaPush(unittest.TestCase, TestReactorMixin):
self.assertEqual(properties["pr_id"], 8)
self.assertEqual(properties["pr_number"], 1)
def checkChangesFromPullRequestFork(self, codebase=None):
self.assertEqual(len(self.changeHook.master.data.updates.changesAdded), 1)
change = self.changeHook.master.data.updates.changesAdded[0]
self.assertEqual(change['repository'], 'ssh://git@git.example.com/testuser/webhook_test.git')
self.assertEqual(
change["author"], "testuser name <testuser@example.com>")
self.assertEqual(
change["revision"], '449a5a8ca05607106b5ba41988c1a658a8949a18')
self.assertEqual(change["branch"], "master")
self.assertEqual(change[
"revlink"], "https://git.example.com/testuser/webhook_test/pulls/4")
properties = change["properties"]
self.assertEqual(properties["base_branch"], "master")
self.assertEqual(properties["base_sha"], "449a5a8ca05607106b5ba41988c1a658a8949a18")
self.assertEqual(properties["base_repository"], "https://git.example.com/testuser/webhook_test.git")
self.assertEqual(properties["base_git_ssh_url"], "ssh://git@git.example.com/testuser/webhook_test.git")
self.assertEqual(properties["head_branch"], "feature_branch")
self.assertEqual(properties["head_sha"], "53e3075cbe468f14c2801d186d703e64b2adee12")
self.assertEqual(properties["head_repository"], "https://git.example.com/test_org/webhook_test_fork.git")
self.assertEqual(properties["head_git_ssh_url"], "ssh://git@git.example.com/test_org/webhook_test_fork.git")
self.assertEqual(properties["pr_id"], 36)
self.assertEqual(properties["pr_number"], 4)
@defer.inlineCallbacks
def testPushEvent(self):
self.request = FakeRequest(content=giteaJsonPushPayload)
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b"push"
self.request.received_headers[_HEADER_SIGNATURE] = giteaJsonPushPayload_Signature
res = yield self.request.test_render(self.changeHook)
self.checkChangesFromPush(res)
@defer.inlineCallbacks
def testChangedFiles(self):
self.request = FakeRequest(content=giteaJsonPushModifiedFiles)
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b"push"
res = yield self.request.test_render(self.changeHook)
self.checkFileChanges(res)
@defer.inlineCallbacks
def testNoChangedFiles(self):
self.request = FakeRequest(content=giteaJsonPushEmptyFiles)
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b"push"
res = yield self.request.test_render(self.changeHook)
self.checkNoFileChanges(res)
@defer.inlineCallbacks
def testPullRequestEvent(self):
self.request = FakeRequest(content=giteaJsonPullRequestPayload)
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b"pull_request"
self.request.received_headers[_HEADER_SIGNATURE] = giteaJsonPullRequestPayload_Signature
res = yield self.request.test_render(self.changeHook)
self.checkChangesFromPullRequest(res)
@defer.inlineCallbacks
def testPullRequestForkEvent(self):
self.request = FakeRequest(content=giteaJsonPullRequestFork)
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b"pull_request"
res = yield self.request.test_render(self.changeHook)
self.checkChangesFromPullRequestFork(res)
@defer.inlineCallbacks
def testPullRequestNotMergeableEvent(self):
self.request = FakeRequest(content=giteaJsonPullRequestPayloadNotMergeable)
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b"pull_request"
self.request.received_headers[_HEADER_SIGNATURE] = giteaJsonPullRequestPayloadNotMergeable_Signature
yield self.request.test_render(self.changeHook)
self.assertEqual(len(self.changeHook.master.data.updates.changesAdded), 0)
@ -1474,7 +827,6 @@ class TestChangeHookGiteaPush(unittest.TestCase, TestReactorMixin):
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b"pull_request"
self.request.received_headers[_HEADER_SIGNATURE] = giteaJsonPullRequestPayloadMerged_Signature
yield self.request.test_render(self.changeHook)
self.assertEqual(len(self.changeHook.master.data.updates.changesAdded), 0)
@ -1511,7 +863,6 @@ class TestChangeHookGiteaPushOnlySingle(unittest.TestCase, TestReactorMixin):
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b"push"
self.request.received_headers[_HEADER_SIGNATURE] = giteaJsonPushPayload_Signature
res = yield self.request.test_render(self.changeHook)
self.checkChangesFromPush(res)
@ -1529,7 +880,6 @@ class TestChangeHookGiteaSecretPhrase(unittest.TestCase, TestReactorMixin):
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b"push"
self.request.received_headers[_HEADER_SIGNATURE] = giteaJsonPushPayload_Signature
yield self.request.test_render(self.changeHook)
self.assertEqual(len(self.changeHook.master.data.updates.changesAdded), 2)
@ -1539,43 +889,6 @@ class TestChangeHookGiteaSecretPhrase(unittest.TestCase, TestReactorMixin):
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b"push"
self.request.received_headers[_HEADER_SIGNATURE] = giteaJsonPushPayload_Signature
yield self.request.test_render(self.changeHook)
self.assertEqual(len(self.changeHook.master.data.updates.changesAdded), 0)
class TestChangeHookGiteaSecretPhraseProvider(unittest.TestCase, TestReactorMixin):
def setUp(self):
self.setUpTestReactor()
self.master = fakeMasterForHooks(self)
self.changeHook = change_hook.ChangeHookResource(
dialects={'gitea': {"secret": Secret("token")}},
master=self.master)
fake_storage_service = FakeSecretStorage()
secret_service = SecretManager()
secret_service.services = [fake_storage_service]
secret_service.setServiceParent(self.master)
fake_storage_service.reconfigService(secretdict={"token": "test"})
@defer.inlineCallbacks
def testValidSecret(self):
self.request = FakeRequest(content=giteaJsonPushPayload)
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b"push"
self.request.received_headers[_HEADER_SIGNATURE] = giteaJsonPushPayload_Signature
yield self.request.test_render(self.changeHook)
self.assertEqual(len(self.changeHook.master.data.updates.changesAdded), 2)
@defer.inlineCallbacks
def testInvalidSecret(self):
self.request = FakeRequest(content=giteaInvalidSecretPush)
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b"push"
self.request.received_headers[_HEADER_SIGNATURE] = giteaJsonPushPayload_Signature
yield self.request.test_render(self.changeHook)
self.assertEqual(len(self.changeHook.master.data.updates.changesAdded), 0)
@ -1608,7 +921,6 @@ class TestChangeHookGiteaClass(unittest.TestCase, TestReactorMixin):
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b'push'
self.request.received_headers[_HEADER_SIGNATURE] = giteaJsonPushPayload_Signature
yield self.request.test_render(self.changeHook)
self.checkChanges()
@ -1619,7 +931,6 @@ class TestChangeHookGiteaClass(unittest.TestCase, TestReactorMixin):
self.request.uri = b'/change_hook/gitea'
self.request.method = b'POST'
self.request.received_headers[_HEADER_EVENT_TYPE] = b'release'
self.request.received_headers[_HEADER_SIGNATURE] = giteaJsonPushPayload_Signature
yield self.request.test_render(self.changeHook)
self.checkChanges()

View File

@ -1,25 +1,17 @@
import base64
import json
import re
import hmac
import hashlib
from buildbot.process.properties import Properties
from buildbot.util import bytes2unicode, unicode2bytes
from buildbot.util import bytes2unicode
from buildbot.www.hooks.base import BaseHookHandler
from twisted.internet import defer
from twisted.python import log
from dateutil.parser import parse as dateparse
_HEADER_USER_AGENT = 'User-Agent'
_HEADER_SIGNATURE = 'X-TX-Signature'
_EVENT_KEY = 'event'
_HEADER_EVENT_TYPE = 'X-Gitea-Event'
class TransifexHandler(BaseHookHandler):
class GiteaHandler(BaseHookHandler):
def process_translation_completed(self, payload, event_type, codebase):
def process_push(self, payload, event_type, codebase):
refname = payload["ref"]
changes = []
@ -41,14 +33,10 @@ class TransifexHandler(BaseHookHandler):
commits = commits[:1]
for commit in commits:
files = []
for kind in ('added', 'modified', 'removed'):
files.extend(commit.get(kind, []) or [])
timestamp = dateparse(commit['timestamp'])
change = {
'author': '{} <{}>'.format(commit['author']['name'],
commit['author']['email']),
'files': files,
'comments': commit['message'],
'revision': commit['id'],
'when_timestamp': timestamp,
@ -65,10 +53,10 @@ class TransifexHandler(BaseHookHandler):
}
if codebase is not None:
change['codebase'] = codebase
changes.insert(0, change)
changes.append(change)
return changes
def process_review_compoleted(self, payload, event_type, codebase):
def process_pull_request(self, payload, event_type, codebase):
action = payload['action']
# Only handle potential new stuff, ignore close/.
@ -94,11 +82,11 @@ class TransifexHandler(BaseHookHandler):
pull_request['number'],
pull_request['title'],
pull_request['body']),
'revision': base['sha'],
'revision': head['sha'],
'when_timestamp': timestamp,
'branch': base['ref'],
'branch': head['ref'],
'revlink': pull_request['html_url'],
'repository': base['repo']['ssh_url'],
'repository': repository['ssh_url'],
'project': repository['full_name'],
'category': event_type,
'properties': {
@ -113,8 +101,6 @@ class TransifexHandler(BaseHookHandler):
'head_repo_id': head['repo_id'],
'head_repository': head['repo']['clone_url'],
'head_git_ssh_url': head['repo']['ssh_url'],
'head_owner': head['repo']['owner']['username'],
'head_reponame': head['repo']['name'],
'pr_id': pull_request['id'],
'pr_number': pull_request['number'],
'repository_name': repository['name'],
@ -125,69 +111,47 @@ class TransifexHandler(BaseHookHandler):
change['codebase'] = codebase
return [change]
def _transform_variables(payload):
retval = {
project: payload.get('project'),
repository = [payload.get('resource')],
branch = payload.get('language')
}
return retval
@defer.inlineCallbacks
def getChanges(self, request):
secret = None
if isinstance(self.options, dict):
secret = self.options.get('secret')
try:
content = request.content.read()
content_text = bytes2unicode(content)
payload = json.loads(content_text)
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))
if secret is not None:
p = Properties()
p.master = self.master
rendered_secret = yield p.render(secret)
signature = hmac.new(
unicode2bytes(rendered_secret),
unicode2bytes(content_text.strip()),
digestmod=hashlib.sha256)
header_signature = bytes2unicode(
request.getHeader(_HEADER_SIGNATURE))
http_verb = 'POST'
http_url_path = request.headers('X-TX-Url')
http_gmt_date = request.headers('Date')
content_md5 = hashlib.md5(content).hexdigest()
msg = b'\n'.join([
http_verb, http_url_path, http_gmt_date, content_md5
])
tx_signature = base64.b64encode(
hmac.new(
key=rendered_secret,
msg=msg,
digestmod=hashlib.sha256
).digest()
)
if tx_signature() != header_signature:
raise ValueError('Invalid secret')
event_type = bytes2unicode(payload.get(_EVENT_KEY), "None")
log.msg("Received event '{}' from transifex".format(event_type))
codebase = ""
codebases = request.args.get('codebase', [None])
codebase = bytes2unicode(codebases[0])
changes = []
handler_function = getattr(self, 'process_{}'.format(event_type), None)
if not handler_function:
log.msg("Ignoring transifex event '{}'".format(event_type))
log.msg("Ignoring gitea event '{}'".format(event_type))
else:
changes = handler_function(payload, event_type, codebase)
return (changes, 'transifex')
return (changes, 'git')
class GiteaHandlerPlugin(BaseHookHandler):
def __init__(self, master, options):
if not options:
options = {}
super().__init__(master, options)
handler_class = options.get('class', GiteaHandler)
if 'class' in options:
del options['class']
self.handler = handler_class(master, options)
def getChanges(self, request):
return self.handler.getChanges(request)
# Plugin name
transifex = TransifexHandler
gitea = GiteaHandlerPlugin

View File

@ -6,7 +6,7 @@ from setuptools import setup
with open("README.md", "r") as fh:
long_description = fh.read()
VERSION = "1.7.2"
VERSION = "1.1.0"
setup(name='buildbot-gitea',
version=VERSION,
@ -17,8 +17,8 @@ setup(name='buildbot-gitea',
long_description=long_description,
long_description_content_type="text/markdown",
packages=['buildbot_gitea'],
install_requires=[
"buildbot>=3.0.0"
requires=[
"buildbot (>=2.0.0)"
],
entry_points={
"buildbot.webhooks": [