227 lines
6.4 KiB
Python
227 lines
6.4 KiB
Python
import logging
|
|
|
|
from notificationhelper import build_event_data
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class InvalidNotificationEventException(Exception):
|
|
pass
|
|
|
|
class NotificationEvent(object):
|
|
def __init__(self):
|
|
pass
|
|
|
|
def get_level(self, event_data, notification_data):
|
|
"""
|
|
Returns a 'level' representing the severity of the event.
|
|
Valid values are: 'info', 'warning', 'error', 'primary'
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def get_summary(self, event_data, notification_data):
|
|
"""
|
|
Returns a human readable one-line summary for the given notification data.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def get_message(self, event_data, notification_data):
|
|
"""
|
|
Returns a human readable HTML message for the given notification data.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def get_sample_data(self, repository=None):
|
|
"""
|
|
Returns sample data for testing the raising of this notification, with an optional
|
|
repository.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@classmethod
|
|
def event_name(cls):
|
|
"""
|
|
Particular event implemented by subclasses.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@classmethod
|
|
def get_event(cls, eventname):
|
|
for subc in cls.__subclasses__():
|
|
if subc.event_name() == eventname:
|
|
return subc()
|
|
|
|
raise InvalidNotificationEventException('Unable to find event: %s' % eventname)
|
|
|
|
|
|
class RepoPushEvent(NotificationEvent):
|
|
@classmethod
|
|
def event_name(cls):
|
|
return 'repo_push'
|
|
|
|
def get_level(self, event_data, notification_data):
|
|
return 'info'
|
|
|
|
def get_summary(self, event_data, notification_data):
|
|
return 'Repository %s updated' % (event_data['repository'])
|
|
|
|
def get_message(self, event_data, notification_data):
|
|
if not event_data.get('updated_tags', {}).keys():
|
|
html = """
|
|
Repository <a href="%s">%s</a> has been updated via a push.
|
|
""" % (event_data['homepage'],
|
|
event_data['repository'])
|
|
else:
|
|
html = """
|
|
Repository <a href="%s">%s</a> has been updated via a push.
|
|
<br><br>
|
|
Tags Updated: %s
|
|
""" % (event_data['homepage'],
|
|
event_data['repository'],
|
|
', '.join(event_data['updated_tags'].keys()))
|
|
|
|
return html
|
|
|
|
def get_sample_data(self, repository):
|
|
return build_event_data(repository, {
|
|
'updated_tags': {'latest': 'someimageid', 'foo': 'anotherimage'},
|
|
'pushed_image_count': 10,
|
|
'pruned_image_count': 3
|
|
})
|
|
|
|
|
|
class BuildQueueEvent(NotificationEvent):
|
|
@classmethod
|
|
def event_name(cls):
|
|
return 'build_queued'
|
|
|
|
def get_level(self, event_data, notification_data):
|
|
return 'info'
|
|
|
|
def get_sample_data(self, repository):
|
|
build_uuid = 'fake-build-id'
|
|
|
|
return build_event_data(repository, {
|
|
'is_manual': False,
|
|
'build_id': build_uuid,
|
|
'build_name': 'some-fake-build',
|
|
'docker_tags': ['latest', 'foo', 'bar'],
|
|
'trigger_kind': 'GitHub'
|
|
}, subpage='/build?current=%s' % build_uuid)
|
|
|
|
def get_summary(self, event_data, notification_data):
|
|
return 'Build queued for repository %s' % (event_data['repository'])
|
|
|
|
def get_message(self, event_data, notification_data):
|
|
is_manual = event_data['is_manual']
|
|
if is_manual:
|
|
html = """
|
|
A <a href="%s">new build</a> has been manually queued to start on repository %s.
|
|
<br><br>
|
|
Build ID: %s
|
|
""" % (event_data['homepage'], event_data['repository'], event_data['build_id'])
|
|
else:
|
|
html = """
|
|
A <a href="%s">new build</a> has been queued via a %s trigger to start on repository %s.
|
|
<br><br>
|
|
Build ID: %s
|
|
""" % (event_data['homepage'], event_data['trigger_kind'],
|
|
event_data['repository'], event_data['build_id'])
|
|
|
|
return html
|
|
|
|
|
|
|
|
class BuildStartEvent(NotificationEvent):
|
|
@classmethod
|
|
def event_name(cls):
|
|
return 'build_start'
|
|
|
|
def get_level(self, event_data, notification_data):
|
|
return 'info'
|
|
|
|
def get_sample_data(self, repository):
|
|
build_uuid = 'fake-build-id'
|
|
|
|
return build_event_data(repository, {
|
|
'build_id': build_uuid,
|
|
'build_name': 'some-fake-build',
|
|
'docker_tags': ['latest', 'foo', 'bar'],
|
|
'trigger_kind': 'GitHub'
|
|
}, subpage='/build?current=%s' % build_uuid)
|
|
|
|
def get_summary(self, event_data, notification_data):
|
|
return 'Build started for repository %s' % (event_data['repository'])
|
|
|
|
def get_message(self, event_data, notification_data):
|
|
html = """
|
|
A <a href="%s">new build</a> has started on repository %s.
|
|
<br><br>
|
|
Build ID: %s
|
|
""" % (event_data['homepage'], event_data['repository'], event_data['build_id'])
|
|
|
|
return html
|
|
|
|
|
|
class BuildSuccessEvent(NotificationEvent):
|
|
@classmethod
|
|
def event_name(cls):
|
|
return 'build_success'
|
|
|
|
def get_level(self, event_data, notification_data):
|
|
return 'primary'
|
|
|
|
def get_sample_data(self, repository):
|
|
build_uuid = 'fake-build-id'
|
|
|
|
return build_event_data(repository, {
|
|
'build_id': build_uuid,
|
|
'build_name': 'some-fake-build',
|
|
'docker_tags': ['latest', 'foo', 'bar'],
|
|
'trigger_kind': 'GitHub'
|
|
}, subpage='/build?current=%s' % build_uuid)
|
|
|
|
def get_summary(self, event_data, notification_data):
|
|
return 'Build succeeded for repository %s' % (event_data['repository'])
|
|
|
|
def get_message(self, event_data, notification_data):
|
|
html = """
|
|
A <a href="%s">build</a> has finished on repository %s.
|
|
<br><br>
|
|
Build ID: %s
|
|
""" % (event_data['homepage'], event_data['repository'], event_data['build_id'])
|
|
|
|
return html
|
|
|
|
|
|
class BuildFailureEvent(NotificationEvent):
|
|
@classmethod
|
|
def event_name(cls):
|
|
return 'build_failure'
|
|
|
|
def get_level(self, event_data, notification_data):
|
|
return 'error'
|
|
|
|
def get_sample_data(self, repository):
|
|
build_uuid = 'fake-build-id'
|
|
|
|
return build_event_data(repository, {
|
|
'build_id': build_uuid,
|
|
'build_name': 'some-fake-build',
|
|
'docker_tags': ['latest', 'foo', 'bar'],
|
|
'trigger_kind': 'GitHub',
|
|
'error_message': 'This is a fake error message'
|
|
}, subpage='/build?current=%s' % build_uuid)
|
|
|
|
def get_summary(self, event_data, notification_data):
|
|
return 'Build failure for repository %s' % (event_data['repository'])
|
|
|
|
def get_message(self, event_data, notification_data):
|
|
html = """
|
|
A <a href="%s">build</a> has failed on repository %s.
|
|
<br><br>
|
|
Reason: %s<br>
|
|
Build ID: %s<br>
|
|
""" % (event_data['homepage'], event_data['repository'],
|
|
event_data['error_message'], event_data['build_id'])
|
|
|
|
return html
|