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_',
Copyright (C) 2012-2017 Edgewall Software