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