trac-hudson / 0.10 / HudsonTrac /

Full commit
# -*- coding: utf-8 -*-
A Trac plugin which interfaces with the Hudson Continuous integration server

You can configure this component via the
[wiki:TracIni#hudson-section "[hudson]"]
section in the trac.ini file.

See also:

import time
import urllib2
import base64
from datetime import datetime
from trac.core import *
from trac.config import Option, BoolOption
from trac.perm import IPermissionRequestor
from trac.util import Markup, format_datetime, pretty_timedelta
from trac.util.text import unicode_quote
from import INavigationContributor, ITemplateProvider
from import add_stylesheet
    from trac.timeline.api import ITimelineEventProvider
except ImportError:
    from trac.Timeline import ITimelineEventProvider
    from ast import literal_eval
except ImportError:
    def literal_eval(str):
        return eval(str, {"__builtins__":None}, {"True":True, "False":False})

class HudsonTracPlugin(Component):
    Display Hudson results in the timeline and an entry in the main navigation

    implements(INavigationContributor, ITimelineEventProvider,
               ITemplateProvider, IPermissionRequestor)

    disp_mod = BoolOption('hudson', 'display_modules', 'false',
                          'Display status of modules in the timeline too. ')
    job_url  = Option('hudson', 'job_url', 'http://localhost/hudson/',
                      'The url of the top-level hudson page if you want to '
                      'display all jobs, or a job or module url (such as '
                      'http://localhost/hudson/job/build_foo/) if you want '
                      'only display builds from a single job or module. '
                      'This must be an absolute url.')
    username = Option('hudson', 'username', '',
                      'The username to use to access hudson')
    password = Option('hudson', 'password', '',
                      'The password to use to access hudson')
    nav_url  = Option('hudson', 'main_page', '/hudson/',
                      'The url of the hudson main page to which the trac nav '
                      'entry should link; if empty, no entry is created in '
                      'the nav bar. This may be a relative url.')
    disp_tab = BoolOption('hudson', 'display_in_new_tab', 'false',
                          'Open hudson page in new tab/window')
    alt_succ = BoolOption('hudson', 'alternate_success_icon', 'false',
                          'Use an alternate success icon (green ball instead '
                          'of blue)')
    use_desc = BoolOption('hudson', 'display_build_descriptions', 'true',
                          'Whether to display the build descriptions for '
                          'each build instead of the canned "Build finished '
                          'successfully" etc messages.')

    def __init__(self):
        # get base api url
        api_url = unicode_quote(self.job_url, '/%:@')
        if api_url and api_url[-1] != '/':
            api_url += '/'
        api_url += 'api/python'

        # set up http authentication
        pwd_mgr = urllib2.HTTPPasswordMgrWithDefaultRealm()
        pwd_mgr.add_password(None, api_url, self.username, self.password)

        b_auth = urllib2.HTTPBasicAuthHandler(pwd_mgr)
        d_auth = urllib2.HTTPDigestAuthHandler(pwd_mgr)

        self.url_opener = urllib2.build_opener(b_auth, d_auth,

        self.env.log.debug("registered auth-handler for '%s', username='%s'",
                           api_url, self.username)

        # construct tree=... parameter to query for the desired items
        tree = '%(b)s'
        if self.disp_mod:
            tree += ',modules[%(b)s]'
        if '/job/' not in api_url:
            tree = 'jobs[' + tree + ']'

        tree = tree % {'b':

        # assemble final url
        self.info_url = '%s?tree=%s' % (api_url, tree)

        self.env.log.debug("Build-info url: '%s'", self.info_url)

    # IPermissionRequestor methods  

    def get_permission_actions(self):
        return ['BUILD_VIEW']

    # INavigationContributor methods

    def get_active_navigation_item(self, req):
        return 'builds'

    def get_navigation_items(self, req):
        if self.nav_url and req.perm.has_permission('BUILD_VIEW'):
            yield 'mainnav', 'builds', Markup('<a href="%s"%s>Builds</a>' % \
                    (self.nav_url, self.disp_tab and ' target="hudson"' or ''))

    # ITemplateProvider methods
    def get_templates_dirs(self):
        return []

    def get_htdocs_dirs(self):
        from pkg_resources import resource_filename
        return [('HudsonTrac', resource_filename(__name__, 'htdocs'))]

    # ITimelineEventProvider methods

    def get_timeline_filters(self, req):
        if req.perm.has_permission('BUILD_VIEW'):
            yield ('build', 'Hudson Builds')

    def get_timeline_events(self, req, start, stop, filters):
        if 'build' not in filters or not req.perm.has_permission('BUILD_VIEW'):

        # helper to extract all values in a nested structure of lists and dicts
        def __find_all(d, paths):
            if not isinstance(paths, basestring):
                for path in paths:
                    for item in __find_all(d, path):
                        yield item

            parts = paths.split('.', 1)
            key = parts[0]
            if key in d:
                if len(parts) > 1:
                    for item in __find_all(d[key], parts[1]):
                        yield item
                    yield d[key]
            elif not isinstance(d, dict) and not isinstance(d, basestring):
                for elem in d:
                    for item in __find_all(elem, paths):
                        yield item

        def __find_first(d, paths):
            l = list(__find_all(d, paths))
            return len(l) > 0 and l[0] or None

        # extract individual builds; what we may get from hudson is:
        # {'jobs': [{'modules': [{'builds': [{'building': False, ...
        # {'jobs': [{'builds': [{'building': False, ...
        # {'modules': [{'builds': [{'building': False, ...
        # {'builds': [{'building': False, ...
        def __get_builds(info):
            for arr in __find_all(info, ['builds', 'modules.builds',
                for item in arr:
                    yield item

        # Support both Trac 0.10 and 0.11
        if isinstance(start, datetime): # Trac>=0.11
            from trac.util.datefmt import to_timestamp
            start = to_timestamp(start)
            stop = to_timestamp(stop)

        add_stylesheet(req, 'HudsonTrac/hudsontrac.css')

        # get and parse the build-info
            local_exc = False
                resp =
                cset ='charset') or 'ISO-8859-1'

                ct   =
                if ct != 'text/x-python':
                    local_exc = True
                    raise IOError(
                        "Error getting build info from '%s': returned document "
                        "has unexpected type '%s' (expected 'text/x-python'). "
                        "The returned text is:\n%s" %
                        (self.info_url, ct, unicode(, cset)))

                info = literal_eval(
            except Exception:
                if local_exc:

                import sys
                self.env.log.exception("Error getting build info from '%s'",
                raise IOError(
                    "Error getting build info from '%s': %s: %s. This most "
                    "likely means you configured a wrong job_url, username, "
                    "or password." %
                    (self.info_url, sys.exc_info()[0].__name__,

        # extract all build entries
        for entry in __get_builds(info):
            # ignore builds that are still running
            if entry['building'] == 'true':

            # get start/stop times
            started   = entry['timestamp']
            completed = started + entry['duration']
            started   /= 1000
            completed /= 1000

            if started < start or started > stop:

            # get message
            result = entry['result']
            message, kind = {
                'SUCCESS': ('Build finished successfully',
                'UNSTABLE': ('Build unstable', 'build-unstable'),
                'ABORTED': ('Build aborted', 'build-aborted'),
                }.get(result, ('Build failed', 'build-failed'))

            if self.use_desc:
                message = entry['description'] and \
                            unicode(entry['description'], cset) or message

            # format response
            comment = "%s at %s, duration %s" % (
                          message, format_datetime(completed),
                          pretty_timedelta(started, completed))

            href  = entry['url']
            title = 'Build "%s" (%s)' % \
                    (unicode(entry['fullDisplayName'], cset), result.lower())

            yield kind, href, title, completed, None, comment

    class HudsonFormLoginHandler(urllib2.BaseHandler):
        def __init__(self, parent):
            self.p = parent

        def http_error_403(self, req, fp, code, msg, headers):
            for h in self.p.url_opener.handlers:
                if isinstance(h, self.p.HTTPOpenHandlerBasicAuthNoChallenge):

                "registered auth-handler for form-based authentication")


    class HTTPOpenHandlerBasicAuthNoChallenge(urllib2.BaseHandler):

        auth_header = 'Authorization'

        def __init__(self, username, password):
            raw = "%s:%s" % (username, password)
            self.auth = 'Basic %s' % base64.b64encode(raw).strip()

        def default_open(self, req):
            req.add_header(self.auth_header, self.auth)