quay/endpoints/notificationevent.py

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