import logging
import io
import os.path
import tarfile
import base64
logger = logging.getLogger(__name__)
class InvalidNotificationEventException(Exception):
pass
class NotificationEvent(object):
def __init__(self):
pass
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_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', []):
html = """
Repository %s has been updated via a push.
""" % (event_data['homepage'],
event_data['repository'])
else:
html = """
Repository %s has been updated via a push.
Tags Updated: %s
""" % (event_data['homepage'],
event_data['repository'],
event_data['updated_tags'])
return html
def get_sample_data(self, repository):
repo_string = '%s/%s' % (repository.namespace, repository.name)
event_data = {
'repository': repo_string,
'namespace': repository.namespace,
'name': repository.name,
'docker_url': 'quay.io/%s' % repo_string,
'homepage': 'https://quay.io/repository/%s' % repo_string,
'visibility': repository.visibility.name,
'updated_tags': ['latest', 'foo', 'bar'],
'pushed_image_count': 10,
'pruned_image_count': 3
}
return event_data
class BuildQueueEvent(NotificationEvent):
@classmethod
def event_name(cls):
return 'build_queued'
def get_sample_data(self, repository):
build_uuid = 'fake-build-id'
repo_string = '%s/%s' % (repository.namespace, repository.name)
event_data = {
'repository': repo_string,
'namespace': repository.namespace,
'name': repository.name,
'docker_url': 'quay.io/%s' % repo_string,
'homepage': 'https://quay.io/repository/%s/build/%s' % (repo_string, build_uuid),
'is_manual': False,
'build_id': build_uuid,
'build_name': 'some-fake-build',
'docker_tags': ['latest', 'foo', 'bar'],
'trigger_kind': 'GitHub'
}
return event_data
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 new build has been manually queued to start on repository %s.
Build ID: %s
""" % (event_data['homepage'], event_data['repository'], event_data['build_id'])
else:
html = """
A new build has been queued via a %s trigger to start on repository %s.
Build ID: %s
""" % (event_data['homepage'], event_data['repository'],
event_data['trigger_kind'], event_data['build_id'])
return html
class BuildStartEvent(NotificationEvent):
@classmethod
def event_name(cls):
return 'build_start'
def get_sample_data(self, repository):
build_uuid = 'fake-build-id'
repo_string = '%s/%s' % (repository.namespace, repository.name)
event_data = {
'repository': repo_string,
'namespace': repository.namespace,
'name': repository.name,
'docker_url': 'quay.io/%s' % repo_string,
'homepage': 'https://quay.io/repository/%s/build?current=%s' % (repo_string, build_uuid),
'build_id': build_uuid,
'build_name': 'some-fake-build',
'docker_tags': ['latest', 'foo', 'bar'],
'trigger_kind': 'GitHub'
}
return event_data
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 new build has started on repository %s.
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_sample_data(self, repository):
build_uuid = 'fake-build-id'
repo_string = '%s/%s' % (repository.namespace, repository.name)
event_data = {
'repository': repo_string,
'namespace': repository.namespace,
'name': repository.name,
'docker_url': 'quay.io/%s' % repo_string,
'homepage': 'https://quay.io/repository/%s/build?current=%s' % (repo_string, build_uuid),
'build_id': build_uuid,
'build_name': 'some-fake-build',
'docker_tags': ['latest', 'foo', 'bar'],
'trigger_kind': 'GitHub'
}
return event_data
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 build has finished on repository %s.
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_sample_data(self, repository):
build_uuid = 'fake-build-id'
repo_string = '%s/%s' % (repository.namespace, repository.name)
event_data = {
'repository': repo_string,
'namespace': repository.namespace,
'name': repository.name,
'docker_url': 'quay.io/%s' % repo_string,
'homepage': 'https://quay.io/repository/%s/build?current=%s' % (repo_string, build_uuid),
'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'
}
return event_data
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 build has failed on repository %s.
Reason: %s
Build ID: %s
""" % (event_data['homepage'], event_data['repository'],
event_data['error_message'], event_data['build_id'])
return html