Mercurial > bitten > bitten-test
changeset 531:5b4a1f1872d3
Import of bitten notify from Ole Trenner. Imported from http://trac.3dbits.de/bittennotify, revision [31]. Source code license change to bsd verified by Ole. Everything looks great. Thanks for the patch.
author | wbell |
---|---|
date | Sat, 21 Mar 2009 19:05:41 +0000 |
parents | 81e43e3770e6 |
children | e9a22dc21e29 |
files | bitten/notify.py bitten/templates/bitten_notify_email.txt bitten/tests/__init__.py bitten/tests/master.py bitten/tests/notify.py setup.py |
diffstat | 6 files changed, 380 insertions(+), 5 deletions(-) [+] |
line wrap: on
line diff
new file mode 100644 --- /dev/null +++ b/bitten/notify.py @@ -0,0 +1,188 @@ +#-*- 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 trac.core import * +from trac.web.chrome import ITemplateProvider +from trac.config import BoolOption +from trac.notification import NotifyEmail +from bitten.api import IBuildListener +from bitten.model import Build, BuildStep, BuildLog + + +CONFIG_SECTION = 'notification' +NOTIFY_ON_FAILURE = 'notify_on_failed_build' +NOTIFY_ON_SUCCESS = 'notify_on_successful_build' + + +class BittenNotify(Component): + notify_on_failure = BoolOption(CONFIG_SECTION, NOTIFY_ON_FAILURE, 'true', + """Notify if bitten build fails.""") + + notify_on_success = BoolOption(CONFIG_SECTION, NOTIFY_ON_SUCCESS, 'false', + """Notify if bitten build succeeds.""") + + def __init__(self): + self.log.debug('Initializing BittenNotify plugin') + + +class BittenNotifyDispatcher(Component): + """Sends notifications on build status by mail.""" + + implements(IBuildListener, ITemplateProvider) + + def __init__(self): + self.log.debug('Initializing BittenNotify Dispatcher') + self.email = BittenNotifyEmail(self.env) + + def notify(self, build = None): + self.log.info('BittenNotify invoked for build %r' % build) + self.log.debug('build status: %s' % build.status) + if self._should_notify(build): + self.log.info('Sending notification for build %r' % build) + build_info = BuildInfo(self.env, build) + self.email.notify(build_info) + + def _should_notify(self, build): + notify_on_failure = self.config.getbool(CONFIG_SECTION, + NOTIFY_ON_FAILURE) + notify_on_success = self.config.getbool(CONFIG_SECTION, + NOTIFY_ON_SUCCESS) + build_is_failure = (build.status == Build.FAILURE) + build_is_success = (build.status == Build.SUCCESS) + return (build_is_failure and notify_on_failure) or \ + (build_is_success and notify_on_success) + + # 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 BuildInfo(dict): + """Wraps a Build instance and exposes properties conveniently""" + + readable_states = {Build.SUCCESS:'Successful', Build.FAILURE:'Failed'} + + def __init__(self, env, build): + dict.__init__(self) + self.build = build + self.env = env + self['project_name'] = self.env.project_name + self['id'] = self.build.id + self['status'] = self.readable_states[self.build.status] + self['link'] = self.env.abs_href.build(self.build.config, + self.build.id) + self['config'] = self.build.config + self['slave'] = self.build.slave + self['changeset'] = self.build.rev + self['changesetlink'] = self.env.abs_href.changeset(self.build.rev) + self['author'] = self.get_author(build) + self['errors'] = self.get_errors(build) + self['faillog'] = self.get_faillog(build) + + def get_author(self, build): + if build and build.rev: + changeset = self.env.get_repository().get_changeset(build.rev) + return changeset.author + + def get_failed_steps(self, build): + build_steps = BuildStep.select(self.env, + build=build.id, + status=BuildStep.FAILURE) + return build_steps + + def get_errors(self, build): + errors = '' + for step in self.get_failed_steps(build): + errors += ', '.join(['%s: %s' % (step.name, error) \ + for error in step.errors]) + return errors + + def get_faillog(self, build): + faillog = '' + for step in self.get_failed_steps(build): + build_logs = BuildLog.select(self.env, + build=build.id, + step=step.name) + for log in build_logs: + faillog += '\n'.join(['%5s: %s' % (level, msg) \ + for level, msg in log.messages]) + return faillog + + def __getattr__(self, attr): + return dict.__getitem__(self,attr) + + def __repr__(self): + repr = '' + for k, v in self.items(): + repr += '%s: %s\n' % (k, v) + return repr + + def __str__(self): + return self.repr() + + +class BittenNotifyEmail(NotifyEmail): + """Notification of failed builds.""" + + template_name = 'bitten_notify_email.txt' + from_email = 'bitten@localhost' + + def __init__(self, env): + NotifyEmail.__init__(self, env) + + def notify(self, build_info): + self.build_info = build_info + self.data = self.build_info + subject = '[%s Build] %s [%s] %s' % (self.build_info.status, + self.env.project_name, + self.build_info.changeset, + self.build_info.config) + stream = self.template.generate(**self.data) + body = stream.render('text') + self.env.log.debug('notification: %s' % body ) + NotifyEmail.notify(self, self.build_info.id, subject) + + def get_recipients(self, resid): + author = self.build_info.author + users = {} + [users.__setitem__(username, email) for username, name, email in self.env.get_known_users(None)] + if (author in users.keys() and users[author]): + author = users[author] + torecipients = [author] + ccrecipients = [] + return (torecipients, ccrecipients) + + def send(self, torcpts, ccrcpts, mime_headers={}): + mime_headers = {} + mime_headers['X-Trac-Build-ID'] = str(self.build_info.id) + mime_headers['X-Trac-Build-URL'] = self.build_info.link + NotifyEmail.send(self, torcpts, ccrcpts, mime_headers) +
new file mode 100644 --- /dev/null +++ b/bitten/templates/bitten_notify_email.txt @@ -0,0 +1,15 @@ +${status} build of ${project_name} [${changeset}] +--------------------------------------------------------------------- + + Changeset: ${changeset} - <${changesetlink}> + Committed by: ${author} + + Build Configuration: ${config} + Build Slave: ${slave} + Build Number: ${id} - <${link}> + + Failed Steps: ${errors} + Failure Log: + +${faillog} +
--- a/bitten/tests/__init__.py +++ b/bitten/tests/__init__.py @@ -9,8 +9,7 @@ # are also available at http://bitten.edgewall.org/wiki/License. import unittest - -from bitten.tests import admin, master, model, recipe, queue, slave, web_ui +from bitten.tests import admin, master, model, recipe, queue, slave, web_ui, notify from bitten.build import tests as build from bitten.report import tests as report from bitten.util import tests as util @@ -27,6 +26,7 @@ suite.addTest(build.suite()) suite.addTest(report.suite()) suite.addTest(util.suite()) + suite.addTest(notify.suite()) return suite if __name__ == '__main__':
--- a/bitten/tests/master.py +++ b/bitten/tests/master.py @@ -48,7 +48,7 @@ for stmt in connector.to_sql(table): cursor.execute(stmt) - self.repos = Mock() + self.repos = Mock(get_changeset=lambda rev: Mock(author = 'author')) self.env.get_repository = lambda authname=None: self.repos def tearDown(self):
new file mode 100644 --- /dev/null +++ b/bitten/tests/notify.py @@ -0,0 +1,170 @@ +#-*- 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. + +import logging +import os +import sys +import unittest + +from trac.db import DatabaseManager +from trac.test import EnvironmentStub, Mock +from bitten.model import * +from bitten.notify import * + +ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')) + +class BittenNotifyBaseTest(unittest.TestCase): + def setUp(self): + self.set_up_env() + + def set_up_env(self): + self.env = EnvironmentStub() + self.env.get_templates_dir = lambda *args: os.path.join(ROOT, 'bitten', 'templates') + self.env.path = '' + self.repos = Mock(get_changeset=lambda rev: Mock(author = 'author')) + self.env.get_repository = lambda authname = None: self.repos + + db = self.env.get_db_cnx() + cursor = db.cursor() + connector, _ = DatabaseManager(self.env)._get_connector() + for table in schema: + for stmt in connector.to_sql(table): + cursor.execute(stmt) + db.commit() + +class BittenNotifyDispatcherTest(BittenNotifyBaseTest): + """unit tests for BittenNotify dispatcher class""" + def setUp(self): + BittenNotifyBaseTest.setUp(self) + #fixture + self.state = [False] + self.email = Mock(notify = lambda buildInfo: \ + self.state.__setitem__(0,True)) + self.dispatcher = BittenNotifyDispatcher(self.env) + self.dispatcher.email = self.email + self.failed_build = Build(self.env, status = Build.FAILURE) + self.successful_build = Build(self.env, status = Build.SUCCESS) + + def test_do_notify_on_failed_build(self): + self.env.config.set(CONFIG_SECTION, NOTIFY_ON_FAILURE, 'true') + self.dispatcher.notify(self.failed_build) + self.assertTrue(self.state[0], + 'notifier should be called for failed builds.') + + def test_do_not_notify_on_failed_build(self): + self.env.config.set(CONFIG_SECTION, NOTIFY_ON_FAILURE, 'false') + self.dispatcher.notify(self.failed_build) + self.assertFalse(self.state[0], + 'notifier should not be called for failed build.') + + def test_do_notify_on_successful_build(self): + self.env.config.set(CONFIG_SECTION, NOTIFY_ON_SUCCESS, 'true') + self.dispatcher.notify(self.successful_build) + self.assertTrue(self.state[0], + 'notifier should be called for successful builds when configured.') + + def test_do_not_notify_on_successful_build(self): + self.env.config.set(CONFIG_SECTION, NOTIFY_ON_SUCCESS, 'false') + self.dispatcher.notify(self.successful_build) + self.assertFalse(self.state[0], + 'notifier shouldn\'t be called for successful build.') + + +class BuildInfoTest(BittenNotifyBaseTest): + """unit tests for BuildInfo class""" + + def setUp(self): + BittenNotifyBaseTest.setUp(self) + #fixture + self.failed_build = Build(self.env, + config = 'config', + slave = 'slave', + rev = 10, + status = Build.FAILURE) + self.failed_build.id = 1 + self.successful_build = Build(self.env, status = Build.SUCCESS) + self.successful_build.id = 2 + step = BuildStep(self.env, + build = 1, + name = 'test', + status = BuildStep.FAILURE) + step.errors = ['msg'] + step.insert() + log = BuildLog(self.env, build = 1, step = 'test') + log.messages = [('info','msg')] + log.insert() + + def test_exposed_properties(self): + build_info = BuildInfo(self.env, self.failed_build) + self.assertEquals(self.failed_build.id, build_info.id) + self.assertEquals('Failed', build_info.status) + self.assertEquals('http://example.org/trac.cgi/build/config/1', + build_info.link) + self.assertEquals('config', build_info.config) + self.assertEquals('slave', build_info.slave) + self.assertEquals('10', build_info.changeset) + self.assertEquals('http://example.org/trac.cgi/changeset/10', + build_info.changesetlink) + self.assertEquals('author', build_info.author) + self.assertEquals('test: msg', build_info.errors) + self.assertEquals(' info: msg', build_info.faillog) + + def test_exposed_properties_on_successful_build(self): + build_info = BuildInfo(self.env, self.successful_build) + self.assertEquals(self.successful_build.id, build_info.id) + self.assertEquals('Successful', build_info.status) + + +class BittenNotifyEmailTest(BittenNotifyBaseTest): + """unit tests for BittenNotify dispatcher class""" + def setUp(self): + BittenNotifyBaseTest.setUp(self) + self.env.config.set('notification','smtp_enabled','true') + #fixture + self.state = [[]] + self.email = BittenNotifyEmail(self.env) + empty = lambda *a, **k : None + self.email.begin_send = empty + self.email.finish_send = empty + self.email.send = lambda to, cc, hdrs = {} : \ + self.state.__setitem__(0,to) + self.build_info = BuildInfo(self.env, Build(self.env, + status = Build.SUCCESS)) + self.build_info['author'] = 'author' + + def test_notification_uses_default_address(self): + self.email.notify(self.build_info) + self.assertTrue('author' in self.state[0], + 'recipient list should contain plain author') + + def test_notification_uses_custom_address(self): + self.env.get_known_users = lambda cnx = None : [('author', + 'Author\'s Name', + 'author@email.com')] + self.email.notify(self.build_info) + self.assertTrue('author@email.com' in self.state[0], + 'recipient list should contain custom author\'s email') + + def test_notification_discards_invalid_address(self): + self.env.get_known_users = lambda cnx = None : [('author', + 'Author\'s Name', + '')] + self.email.notify(self.build_info) + self.assertTrue('author' in self.state[0], + 'recipient list should only use valid custom address') + + +def suite(): + suite = unittest.TestSuite() + suite.addTest(unittest.makeSuite(BittenNotifyDispatcherTest,'test')) + suite.addTest(unittest.makeSuite(BuildInfoTest,'test')) + suite.addTest(unittest.makeSuite(BittenNotifyEmailTest,'test')) + return suite + +if __name__ == '__main__': + unittest.main(defaultTest='suite')
--- a/setup.py +++ b/setup.py @@ -38,7 +38,8 @@ package_data = { 'bitten': ['htdocs/*.*', 'htdocs/charts_library/*.swf', - 'templates/*.html'] + 'templates/*.html', + 'templates/*.txt'] }, test_suite = 'bitten.tests.suite', entry_points = { @@ -54,7 +55,8 @@ 'bitten.master = bitten.master', 'bitten.web_ui = bitten.web_ui', 'bitten.testing = bitten.report.testing', - 'bitten.coverage = bitten.report.coverage' + 'bitten.coverage = bitten.report.coverage', + 'bitten.notify = bitten.notify' ], 'bitten.recipe_commands': [ NS + 'sh#exec = bitten.build.shtools:exec_',