Mercurial > bitten > bitten-test
changeset 533:7c1919719538
Fix line endings and trailing whitespace in new notification files
| author | mgood |
|---|---|
| date | Sun, 22 Mar 2009 23:47:36 +0000 |
| parents | e9a22dc21e29 |
| children | 79dd34e914a7 |
| files | bitten/notify.py bitten/templates/bitten_notify_email.txt bitten/tests/notify.py |
| diffstat | 3 files changed, 373 insertions(+), 373 deletions(-) [+] |
line wrap: on
line diff
--- a/bitten/notify.py +++ b/bitten/notify.py @@ -1,188 +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) - +#-*- 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) +
--- a/bitten/templates/bitten_notify_email.txt +++ b/bitten/templates/bitten_notify_email.txt @@ -1,15 +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} - +${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/notify.py +++ b/bitten/tests/notify.py @@ -1,170 +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') +#-*- 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')