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

Add tests for repository notification api

Browse source
This commit is contained in:
Evan Cordell 2017-07-17 17:56:32 -04:00
parent 9dad44e93d
commit 57517adef3
7 changed files with 239 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

30
endpoints/api/repositorynotification_models_pre_oci.py
View file

@ -2,12 +2,13 @@ import json
from app import notification_queue
from data import model
from data.model import InvalidNotificationException
from endpoints.api.repositorynotification_models_interface import RepoNotificationInterface, RepositoryNotification
from endpoints.notificationevent import NotificationEvent
from endpoints.notificationhelper import build_notification_data
class PreOCIModel(RepoNotificationInterface):
class RepoNotificationPreOCIModel(RepoNotificationInterface):
def create_repo_notification(self, namespace_name, repository_name, event_name, method_name, method_config, event_config, title=None):
repository = model.repository.get_repository(namespace_name, repository_name)
@ -23,25 +24,40 @@ class PreOCIModel(RepoNotificationInterface):
for n in model.notification.list_repo_notifications(namespace_name, repository_name, event_name)]
def get_repo_notification(self, uuid):
return self._notification(model.notification.get_repo_notification(uuid))
try:
found = model.notification.get_repo_notification(uuid)
except InvalidNotificationException:
return None
return self._notification(found)
def delete_repo_notification(self, namespace_name, repository_name, uuid):
return self._notification(
self.model.notification.delete_repo_notification(namespace_name, repository_name, uuid))
try:
found = model.notification.delete_repo_notification(namespace_name, repository_name, uuid)
except InvalidNotificationException:
return None
return self._notification(found)
def reset_notification_number_of_failures(self, namespace_name, repository_name, uuid):
return self._notification(
model.notification.reset_notification_number_of_failures(namespace_name, repository_name, uuid))
def queue_test_notification(self, notification):
def queue_test_notification(self, uuid):
try:
notification = model.notification.get_repo_notification(uuid)
except InvalidNotificationException:
return None
event_info = NotificationEvent.get_event(notification.event.name)
sample_data = event_info.get_sample_data(notification)
notification_data = build_notification_data(notification, sample_data)
notification_queue.put([notification.repository.namespace_user.username, notification,
notification_queue.put([notification.repository.namespace_user.username, notification.uuid,
notification.event.name], json.dumps(notification_data))
return self._notification(notification)
def _notification(self, notification):
if not notification:
return None
return RepositoryNotification(uuid=notification.uuid,
title=notification.title,
event_name=notification.event.name,
@ -52,4 +68,4 @@ class PreOCIModel(RepoNotificationInterface):
pre_oci_model = PreOCIModel()
pre_oci_model = RepoNotificationPreOCIModel()