vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Convert over to notifications system. Note this is incomplete

Browse source
This commit is contained in:
Joseph Schorr 2014-07-17 22:51:58 -04:00
parent de8e898ad0
commit 8d7493cb86
17 changed files with 432 additions and 166 deletions
Download patch file Download diff file Expand all files Collapse all files

BIN
test/data/test.db
View file

Binary file not shown.

30
test/test_api_security.py
View file

@ -18,7 +18,7 @@ from endpoints.api.robot import UserRobotList, OrgRobot, OrgRobotList, UserRobot
from endpoints.api.trigger import (BuildTriggerActivate, BuildTriggerSources, BuildTriggerSubdirs,
TriggerBuildList, ActivateBuildTrigger, BuildTrigger,
BuildTriggerList, BuildTriggerAnalyze)
from endpoints.api.webhook import Webhook, WebhookList
from endpoints.api.repositorynotification import RepositoryNotification, RepositoryNotificationList
from endpoints.api.user import (PrivateRepositories, ConvertToOrganization, Recovery, Signout,
Signin, User, UserAuthorizationList, UserAuthorization)
from endpoints.api.repotoken import RepositoryToken, RepositoryTokenList
@ -1883,10 +1883,10 @@ class TestBuildTriggerD6tiBuynlargeOrgrepo(ApiTestCase):
self._run_test('DELETE', 404, 'devtable', None)
class TestWebhookQfatPublicPublicrepo(ApiTestCase):
class TestRepositoryNotificationQfatPublicPublicrepo(ApiTestCase):
def setUp(self):
ApiTestCase.setUp(self)
self._set_url(Webhook, public_id="QFAT", repository="public/publicrepo")
self._set_url(RepositoryNotification, uuid="QFAT", repository="public/publicrepo")
def test_get_anonymous(self):
self._run_test('GET', 401, None, None)
@ -1913,10 +1913,10 @@ class TestWebhookQfatPublicPublicrepo(ApiTestCase):
self._run_test('DELETE', 403, 'devtable', None)
class TestWebhookQfatDevtableShared(ApiTestCase):
class TestRepositoryNotificationQfatDevtableShared(ApiTestCase):
def setUp(self):
ApiTestCase.setUp(self)
self._set_url(Webhook, public_id="QFAT", repository="devtable/shared")
self._set_url(RepositoryNotification, uuid="QFAT", repository="devtable/shared")
def test_get_anonymous(self):
self._run_test('GET', 401, None, None)
@ -1943,10 +1943,10 @@ class TestWebhookQfatDevtableShared(ApiTestCase):
self._run_test('DELETE', 400, 'devtable', None)
class TestWebhookQfatBuynlargeOrgrepo(ApiTestCase):
class TestRepositoryNotificationQfatBuynlargeOrgrepo(ApiTestCase):
def setUp(self):
ApiTestCase.setUp(self)
self._set_url(Webhook, public_id="QFAT", repository="buynlarge/orgrepo")
self._set_url(RepositoryNotification, uuid="QFAT", repository="buynlarge/orgrepo")
def test_get_anonymous(self):
self._run_test('GET', 401, None, None)
@ -2529,10 +2529,10 @@ class TestBuildTriggerListBuynlargeOrgrepo(ApiTestCase):
self._run_test('GET', 200, 'devtable', None)
class TestWebhookListPublicPublicrepo(ApiTestCase):
class TestRepositoryNotificationListPublicPublicrepo(ApiTestCase):
def setUp(self):
ApiTestCase.setUp(self)
self._set_url(WebhookList, repository="public/publicrepo")
self._set_url(RepositoryNotificationList, repository="public/publicrepo")
def test_get_anonymous(self):
self._run_test('GET', 401, None, None)
@ -2559,10 +2559,10 @@ class TestWebhookListPublicPublicrepo(ApiTestCase):
self._run_test('POST', 403, 'devtable', {})
class TestWebhookListDevtableShared(ApiTestCase):
class TestRepositoryNotificationListDevtableShared(ApiTestCase):
def setUp(self):
ApiTestCase.setUp(self)
self._set_url(WebhookList, repository="devtable/shared")
self._set_url(RepositoryNotificationList, repository="devtable/shared")
def test_get_anonymous(self):
self._run_test('GET', 401, None, None)
@ -2586,13 +2586,13 @@ class TestWebhookListDevtableShared(ApiTestCase):
self._run_test('POST', 403, 'reader', {})
def test_post_devtable(self):
self._run_test('POST', 201, 'devtable', {})
self._run_test('POST', 201, 'devtable', {'event': 'repo_push', 'method': 'email', 'config': {}})
class TestWebhookListBuynlargeOrgrepo(ApiTestCase):
class TestRepositoryNotificationListBuynlargeOrgrepo(ApiTestCase):
def setUp(self):
ApiTestCase.setUp(self)
self._set_url(WebhookList, repository="buynlarge/orgrepo")
self._set_url(RepositoryNotificationList, repository="buynlarge/orgrepo")
def test_get_anonymous(self):
self._run_test('GET', 401, None, None)
@ -2616,7 +2616,7 @@ class TestWebhookListBuynlargeOrgrepo(ApiTestCase):
self._run_test('POST', 403, 'reader', {})
def test_post_devtable(self):
self._run_test('POST', 201, 'devtable', {})
self._run_test('POST', 201, 'devtable', {'event': 'repo_push', 'method': 'email', 'config': {}})
class TestRepositoryTokenListPublicPublicrepo(ApiTestCase):

45
test/test_api_usage.py
View file

@ -20,7 +20,7 @@ from endpoints.api.robot import UserRobotList, OrgRobot, OrgRobotList, UserRobot
from endpoints.api.trigger import (BuildTriggerActivate, BuildTriggerSources, BuildTriggerSubdirs,
TriggerBuildList, ActivateBuildTrigger, BuildTrigger,
BuildTriggerList, BuildTriggerAnalyze)
from endpoints.api.webhook import Webhook, WebhookList
from endpoints.api.repositorynotification import RepositoryNotification, RepositoryNotificationList
from endpoints.api.user import (PrivateRepositories, ConvertToOrganization, Signout, Signin, User,
UserAuthorizationList, UserAuthorization)
@ -1073,41 +1073,44 @@ class TestRequestRepoBuild(ApiTestCase):
class TestWebhooks(ApiTestCase):
class TestRepositoryNotifications(ApiTestCase):
def test_webhooks(self):
self.login(ADMIN_ACCESS_USER)
# Add a webhook.
json = self.postJsonResponse(WebhookList,
# Add a notification.
json = self.postJsonResponse(RepositoryNotificationList,
params=dict(repository=ADMIN_ACCESS_USER + '/simple'),
data=dict(url='http://example.com'),
data=dict(config={'url': 'http://example.com'}, event='repo_push', method='webhook'),
expected_code=201)
self.assertEquals('http://example.com', json['parameters']['url'])
wid = json['public_id']
self.assertEquals('repo_push', json['event'])
self.assertEquals('webhook', json['method'])
self.assertEquals('http://example.com', json['config']['url'])
wid = json['uuid']
# Get the webhook.
json = self.getJsonResponse(Webhook,
params=dict(repository=ADMIN_ACCESS_USER + '/simple', public_id=wid))
# Get the notification.
json = self.getJsonResponse(RepositoryNotification,
params=dict(repository=ADMIN_ACCESS_USER + '/simple', uuid=wid))
self.assertEquals(wid, json['public_id'])
self.assertEquals('http://example.com', json['parameters']['url'])
self.assertEquals(wid, json['uuid'])
self.assertEquals('repo_push', json['event'])
self.assertEquals('webhook', json['method'])
# Verify the webhook is listed.
json = self.getJsonResponse(WebhookList,
# Verify the notification is listed.
json = self.getJsonResponse(RepositoryNotificationList,
params=dict(repository=ADMIN_ACCESS_USER + '/simple'))
ids = [w['public_id'] for w in json['webhooks']]
ids = [w['uuid'] for w in json['notifications']]
assert wid in ids
# Delete the webhook.
self.deleteResponse(Webhook,
params=dict(repository=ADMIN_ACCESS_USER + '/simple', public_id=wid),
# Delete the notification.
self.deleteResponse(RepositoryNotification,
params=dict(repository=ADMIN_ACCESS_USER + '/simple', uuid=wid),
expected_code=204)
# Verify the webhook is gone.
self.getResponse(Webhook,
params=dict(repository=ADMIN_ACCESS_USER + '/simple', public_id=wid),
# Verify the notification is gone.
self.getResponse(RepositoryNotification,
params=dict(repository=ADMIN_ACCESS_USER + '/simple', uuid=wid),
expected_code=404)