/usr/share/pyshared/bitten/notify.py is in trac-bitten 0.6+final-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | #-*- coding: utf-8 -*-
#
# Copyright (C) 2007 Ole Trenner, <ole@jayotee.de>
# All rights reserved.
#
# This software is licensed as described in the file COPYING, which
# you should have received as part of this distribution.
from genshi.template.text import NewTextTemplate
from trac.core import Component, implements
from trac.web.chrome import ITemplateProvider, Chrome
from trac.config import BoolOption
from trac.notification import NotifyEmail
from bitten.api import IBuildListener
from bitten.model import Build, BuildStep, BuildLog, TargetPlatform
class BittenNotify(Component):
"""Sends notifications on build status by mail."""
implements(IBuildListener, ITemplateProvider)
notify_on_failure = BoolOption('notification',
'notify_on_failed_build', 'true',
"""Notify if bitten build fails.""")
notify_on_success = BoolOption('notification',
'notify_on_successful_build', 'false',
"""Notify if bitten build succeeds.""")
def __init__(self):
self.log.debug('Initializing BittenNotify plugin')
def notify(self, build=None):
self.log.info('BittenNotify invoked for build %r', build)
self.log.debug('build status: %s', build.status)
if not self._should_notify(build):
return
self.log.info('Sending notification for build %r', build)
try:
email = BuildNotifyEmail(self.env)
email.notify(build)
except Exception, e:
self.log.exception("Failure sending notification for build "
"%s: %s", build.id, e)
def _should_notify(self, build):
if build.status == Build.FAILURE:
return self.notify_on_failure
elif build.status == Build.SUCCESS:
return self.notify_on_success
else:
return False
# IBuildListener methods
def build_started(self, build):
"""build started"""
self.notify(build)
def build_aborted(self, build):
"""build aborted"""
self.notify(build)
def build_completed(self, build):
"""build completed"""
self.notify(build)
# ITemplateProvider methods
def get_templates_dirs(self):
"""Return a list of directories containing the provided template
files."""
from pkg_resources import resource_filename
return [resource_filename(__name__, 'templates')]
def get_htdocs_dirs(self):
"""Return the absolute path of a directory containing additional
static resources (such as images, style sheets, etc)."""
return []
class BuildNotifyEmail(NotifyEmail):
"""Notification of failed builds."""
readable_states = {
Build.SUCCESS: 'Successful',
Build.FAILURE: 'Failed',
}
template_name = 'bitten_notify_email.txt'
from_email = 'bitten@localhost'
def __init__(self, env):
NotifyEmail.__init__(self, env)
# Override the template type to always use NewTextTemplate
if not isinstance(self.template, NewTextTemplate):
self.template = Chrome(env).templates.load(
self.template.filepath, cls=NewTextTemplate)
def notify(self, build):
self.build = build
self.data.update(self.template_data())
subject = '[%s Build] %s [%s] %s' % (self.readable_states[build.status],
self.env.project_name,
self.build.rev,
self.build.config)
NotifyEmail.notify(self, self.build.id, subject)
def get_recipients(self, resid):
to = [self.get_author()]
cc = []
return (to, cc)
def send(self, torcpts, ccrcpts):
mime_headers = {
'X-Trac-Build-ID': str(self.build.id),
'X-Trac-Build-URL': self.build_link(),
}
NotifyEmail.send(self, torcpts, ccrcpts, mime_headers)
def build_link(self):
return self.env.abs_href.build(self.build.config, self.build.id)
def template_data(self):
failed_steps = BuildStep.select(self.env, build=self.build.id,
status=BuildStep.FAILURE)
platform = TargetPlatform.fetch(self.env, id=self.build.platform)
change = self.get_changeset()
return {
'build': {
'id': self.build.id,
'status': self.readable_states[self.build.status],
'link': self.build_link(),
'config': self.build.config,
'platform': getattr(platform, 'name', 'unknown'),
'slave': self.build.slave,
'failed_steps': [{
'name': step.name,
'description': step.description,
'errors': step.errors,
'log_messages': self.get_all_log_messages_for_step(step),
} for step in failed_steps],
},
'change': {
'rev': change.rev,
'link': self.env.abs_href.changeset(change.rev),
'author': change.author,
},
}
def get_all_log_messages_for_step(self, step):
messages = []
for log in BuildLog.select(self.env, build=self.build.id,
step=step.name):
messages.extend(log.messages)
return messages
def get_changeset(self):
repos = self.env.get_repository()
assert repos, 'No "(default)" Repository: Add a repository or alias ' \
'named "(default)" to Trac.'
return repos.get_changeset(self.build.rev)
def get_author(self):
return self.get_changeset().author
|