jenkins.py

import time
import notify2

from abc import ABCMeta
from jenkinsapi.jenkins import Jenkins


class JenkinsPoller():

    __metaclass__ = ABCMeta

    def __init__(self, address, build_names=[], polling_iterval=300):
        super(JenkinsPoller, self).__init__()
        self.daemon = True

        # start connection to Jenkins
        self._jenkins_server = Jenkins(address)
        self._interval = polling_iterval
        self._build_names = build_names
        self._last_failures = {name: -1 for name in self._build_names}
        self._last_sucesses = {name: -1 for name in self._build_names}

    def run(self):
        try:
            self._poll_jenkins()
            time.sleep(self._interval)
        except (KeyboardInterrupt, SystemExit):
            # if we catch either exception, just quit the program
            print "\nQuitting..."

    def _poll_jenkins(self):
        for name in self._build_names:
            build = self._jenkins_server[name].get_last_completed_build()
            build_good = build.is_good()
            build_number = build.buildno

            if build_number > self._last_failures[name]:
                if not build_good:
                    self._last_failures[name] = build_number
                    self._show_build_failure(name)

            if build_number > self._last_sucesses[name]:
                if build_good:
                    self._last_sucesses[name] = build_number
                    self._show_build_success(name)


class JenkinsNotify2(JenkinsPoller):
    def _show_build_failure(self, name=''):
        n = notify2.Notification("Build Failed", name, "notification-message-im")
        n.set_urgency(notify2.URGENCY_CRITICAL)
        n.show()

    def _show_build_success(self, name=''):
        n = notify2.Notification("Build Passed", name, "notification-message-im")
        n.show()


if __name__ == "__main__":
    notify2.init('JenkinsWatcher')
    j = JenkinsNotify2("http://builds.mantidproject.org/", build_names=['develop_systemtests_rhel6', 'develop_incremental'])
    j.run()