Slack status push for Buildbot

slack.py

from buildbot.util import httpclientservice
from buildbot.process.properties import Interpolate, Properties
from buildbot.reporters.http import HttpStatusPushBase
from buildbot.process.results import (CANCELLED, EXCEPTION, FAILURE, RETRY,
                                      SKIPPED, SUCCESS, WARNINGS)
from twisted.internet import defer
from urllib.parse import urlparse


def format_timedelta(td):
    seconds = td.seconds
    minutes, seconds = seconds // 60, seconds % 60
    hours, minutes = minutes // 60, minutes % 60

    res = ""
    if seconds:
        res = "%ds" % seconds
    if minutes:
        res = "%dm %s" % (minutes, res)
    if hours:
        res = "%dh %s" % (hours, res)
    return res


class StatusPush(HttpStatusPushBase):
    name = "Railnova Slack status push"
    neededDetails = {'wantProperties': True}

    templates = {
        SUCCESS: 'succeeded',
        WARNINGS: 'succeeded with warnings',
        FAILURE: 'failed',
        SKIPPED: 'has been skipped',
        EXCEPTION: 'errored',
        RETRY: 'pending',
        CANCELLED: 'cancelled',
    }

    colors = {
        SUCCESS: 'good',
        WARNINGS: 'warning',
        FAILURE: 'danger',
        SKIPPED: '',
        EXCEPTION: 'danger',
        RETRY: '',
        CANCELLED: '',
    }

    icons = {
        SUCCESS: ':white_check_mark:',
        WARNINGS: ':grey_exclamation:',
        FAILURE: ':exclamation:',
        SKIPPED: ':arrow_right_hook:',
        EXCEPTION: ':bangbang:',
        RETRY: ':arrows_counterclockwise:',
        CANCELLED: ':no_entry:',
    }

    @defer.inlineCallbacks
    def reconfigService(self, webhook_url, debug=True, ignore_success=False, **kwargs):
        yield HttpStatusPushBase.reconfigService(self, **kwargs)
        url = urlparse(webhook_url)
        self._http = yield httpclientservice.HTTPClientService.getService(
            self.master,
            "{}://{}".format(url.scheme, url.netloc),
            debug=debug,
            verify=True
        )
        self.path = url.path
        self.ignore_success = ignore_success

    @defer.inlineCallbacks
    def send(self, build):
        if build['complete']:
            res = build['results']
            if res == SUCCESS and self.ignore_success:
                return
            status, icon = self.templates[res], self.icons[res]
            color = self.colors[res]

            build_dt = build['complete_at'] - build['started_at']
            build_time = format_timedelta(build_dt)

            branchs = ' - '.join(set(
                x['branch'] for x in build['buildset']['sourcestamps']
            ))

            props = Properties.fromDict(build['properties'])
            tpl = Interpolate(
                '%(kw:icon)s Build '
                '<%(kw:url)s|%(prop:buildername)s#%(prop:buildnumber)s> '
                '%(kw:status)s (%(kw:branchs)s)',
                url=build['url'], icon=icon, status=status, branchs=branchs
            )

            text = yield props.render(tpl)
            yield self._http.post(self.path, json={
                "attachments": [{
                    "text": text,
                    "color": color,
                    "footer": "%s in %s" % (
                        build['buildset']['sourcestamps'][-1]['revision'][:10],
                        build_time
                    ),
                }]
            })