configuration/buildbot/message_notifier.py

# Copyright 2014 the Melange authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
  
import cgi
import collections
import datetime
import re

from buildbot.status import builder

_LOG_LIMIT_LINES = 100
_GOOGLE_PROJECT_HOSTING_LINK_PREFIX = 'https://code.google.com/p/soc/source/detail?r='
_DETAILED_LOG_OF_STEP_HTML = u'<i>Detailed log of %s step:</i> <a href="%s">%s</a>'
_LAST_LINES_OF_LOG_HTML = u'<h4>Last %d lines of "%s"</h4>'
_LINE_BREAK_HTML = u'<br>'
_TABLE_CLOSING_TAG_HTML = u'</table>'
_HTML_TAGS_CLEAN =re.compile('<.*?>')
_TESTS_SUMMARY_LOGS = 'Tests summary'
_TESTS_RUN_LOGS = 'stdio'


class _LogInfo(collections.namedtuple('_LogInfo', ['log_url',
    'log_body', 'log_status'])):
  """_LogInfo class containing useful information about a log.
    
  Attributes:
    log_url(string): The URL where the log lives.
    log_body(list): The list of strings containing the log's stdio.
    log_status(string): One of the constants in buildbot.status.builder.
  """
    
      
def get_logs_from_build(build, master_status):
  """Pulls logs related to a build, and extracts meaningful and usable 
  information out of them.
  
  Args:
    build(zope.interface.Interface.IBuildStatus): The build status.
    master_status(buildbot.status.builder.Status): The BuildMaster status.
  
  Returns:
    logs(dictionary): A dictionary with Log name identifier mapped to 
        _LogInfo objects.
  """
  logs = dict()
  for log in build.getLogs():
    log_name = "%s.%s" % (log.getStep().getName(), log.getName())
    log_name_identifier = log_name.split('.')[-1]
    if 'Test run' in log_name:
      log_status, _ = log.getStep().getResults()
      log_body = log.getText().splitlines()
      log_url = '%s/steps/%s/logs/%s' % (master_status.getURLForThing(build),
          log.getStep().getName(),
          log.getName())
      log_info = _LogInfo(log_url=log_url,
          log_body=log_body, log_status=log_status)
      logs[log_name_identifier] = log_info
  return logs


def build_html_for_test_summary(logs):
  """Constructs html string list for test summary logs.
  
  Args:
    logs(dictionary): A dictionary with Log name identifier mapped to 
        _LogInfo objects.
    
  Returns:
    html_string_list(list): A list containing HTML strings which are processed 
    logs from Tests Summary Step.
  """
  html_string_list = list()
  test_summary_log = logs[_TESTS_SUMMARY_LOGS]
  name = _TESTS_SUMMARY_LOGS
  content = [line for line in test_summary_log.log_body[0].split('</br>') if line != ' ']
  url = test_summary_log.log_url
  html_string_list.append(_DETAILED_LOG_OF_STEP_HTML % (name, url, url))
  html_string_list.append(_LINE_BREAK_HTML)
  html_string_list.append(_LAST_LINES_OF_LOG_HTML % (_LOG_LIMIT_LINES, name))
  unilist = list()
  if len(content) < _LOG_LIMIT_LINES:
    for line in content:
      line = re.sub(_HTML_TAGS_CLEAN, '', line)
      unilist.append(cgi.escape(unicode(line,'utf-8')))
  else:
    for line in content[len(content)-_LOG_LIMIT_LINES:]:
      line = re.sub(_HTML_TAGS_CLEAN, '', line)
      unilist.append(cgi.escape(unicode(line,'utf-8')))
  html_string_list.append(u'<pre>'.join(unilist))
  html_string_list.append(u'</pre>')
  html_string_list.append(_LINE_BREAK_HTML)
  return html_string_list


def build_html_for_test_run(logs):
  """Constructs html string list for test run logs.
  
  Args:
    logs(dictionary): Log name identifier mapped to useful information from 
        logs related to a particular build.
    
  Returns:
    html_string_list(list): A list containing HTML strings which are processed 
        logs from Tests Run Step.
  """
  html_string_list = list()
  test_run_log = logs[_TESTS_RUN_LOGS]
  name = _TESTS_RUN_LOGS
  content = test_run_log.log_body
  url = test_run_log.log_url
  html_string_list.append(_DETAILED_LOG_OF_STEP_HTML % (name, url, url))
  html_string_list.append(_LINE_BREAK_HTML)
  html_string_list.append(_LAST_LINES_OF_LOG_HTML % (_LOG_LIMIT_LINES, name))
  unilist = list()
  if len(content) < _LOG_LIMIT_LINES:
    for line in content:
      unilist.append(cgi.escape(unicode(line,'utf-8')))
  else:
    for line in content[len(content)-LOG_LIMIT_LINES:]:
      unilist.append(cgi.escape(unicode(line,'utf-8')))
  html_string_list.append(u'<pre>'.join(unilist))
  html_string_list.append(u'</pre>')
  return html_string_list


def build_html_for_changelist(source_stamps):
  """Construct html string list for latest changes from a build SourceStamp.
  
  Args:
    source_stamps: buildbot.status.builder.BuildStatus.getSourceStamps
  
  Returns:
    html_string_list(list): A list containing HTML strings which are processed 
        logs for Recent changes based on latest sourcestamps.
  """
  html_string_list = list()
  if source_stamps.changes:
    html_string_list.append(u'<h4>Recent Changes:</h4>')
    for change in source_stamps.changes:
      change_dict = change.asDict()
      when = datetime.datetime.fromtimestamp(change_dict['when']).ctime()
      html_string_list.append(u'<table cellspacing="10">')
      html_string_list.append(
          u'<tr><td>Repository:</td><td>%s</td></tr>' % change_dict['repository'])
      revision_link = _GOOGLE_PROJECT_HOSTING_LINK_PREFIX + change_dict['revision']
      html_string_list.append(
          u'<tr><td>Revision:</td><td><a href=%s>%s</a></td></tr>' % (
          revision_link, change_dict['revision']))
      html_string_list.append(
          u'<tr><td>Project:</td><td>%s</td></tr>' % change_dict['project'])
      html_string_list.append(u'<tr><td>Time:</td><td>%s</td></tr>' % when)
      html_string_list.append(
          u'<tr><td>Changed by:</td><td>%s</td></tr>' % change_dict['who'])
      html_string_list.append(
          u'<tr><td>Comments:</td><td>%s</td></tr>' % change_dict['comments'])
      html_string_list.append(_TABLE_CLOSING_TAG_HTML)
      files = change_dict['files']
      if files:
        html_string_list.append(
            u'<table cellspacing="10"><tr><th align="left">Files</th></tr>')
        for file in files:
          html_string_list.append(u'<tr><td>%s</td></tr>' % file['name'] )
        html_string_list.append(_TABLE_CLOSING_TAG_HTML)
  return html_string_list


def html_message_formatter(mode, name, build, results, master_status):
  """Provide a customized message to Buildbot's MailNotifier.
  The last 100 lines of the log are provided as well as the changes
  relevant to the build.  Message content is formatted as html.
  
  Args:
    mode(string): MailNotifier mode - one of 'all', 'failing', 'problem', 
        'change', or 'passing'.
    name(string): Name of the builder that generated this event.
    build(zope.interface.Interface.IBuildStatus): The build status.
    results(string): The result code - one of 'success', 'warnings', 
         'failure', 'skipped', or 'exception'.
    master_status(buildbot.status.builder.Status): The BuildMaster status.
    
  Returns:
    None - if no url is available for the build object.
    Otherwise - A dictionary with two keys:
      type(string): Message type - 'html' or 'plain'.
      body(string): Complete text of message. 
  """
  result = builder.Results[results]

  logs = get_logs_from_build(build, master_status)

  text = list()
  text.append(u'<h4>Build status: %s</h4>' % result.upper())
  text.append(u'<table cellspacing="10"><tr>')
  text.append(
      u"<td>Buildslave for this Build:</td><td><b>%s</b></td></tr>" % build.getSlavename())
  if master_status.getURLForThing(build):
    text.append(u'<tr><td>Complete logs for all build steps:</td><td><a href="%s">%s</a></td></tr>'
        % (master_status.getURLForThing(build),
        master_status.getURLForThing(build)))
    text.append(u'<tr><td>Build Reason:</td><td>%s</td></tr>' % build.getReason())
    source = u""
    revision_main = u""
    for source_stamps in build.getSourceStamps():
      if source_stamps.codebase:
        source += u'%s: ' % source_stamps.codebase
      if source_stamps.branch:
        source += u"[branch %s] " % source_stamps.branch
      if source_stamps.revision:
        source +=  source_stamps.revision
        revision_main = source_stamps.revision
        revision_link = _GOOGLE_PROJECT_HOSTING_LINK_PREFIX + revision_main
      else:
        source += u"HEAD"
      if source_stamps.patch:
        source += u" (plus patch)"
      if source_stamps.patch_info:
        source += u" (%s)" % source_stamps.patch_info[1]
    text.append(
        u"<tr><td>Build Source Stamp:</td><td><b><a href =%s>%s</a></b></td></tr>" % (
        revision_link, source))
    text.append(
        u"<tr><td>Blamelist:</td><td>%s</td></tr>" % ",".join(
        build.getResponsibleUsers()))
    text.append(_TABLE_CLOSING_TAG_HTML)
    text.append(_LINE_BREAK_HTML)
    
    # get log for the tests summary.
    test_summary_logs = build_html_for_test_summary(logs)
    text += test_summary_logs

    # get log for the tests run.
    test_run_logs = build_html_for_test_run(logs)
    text += test_run_logs

    # get log for latest changes.
    latest_changes = build_html_for_changelist(source_stamps)
    text += latest_changes
    
    text.append(u'<br><br>')
    text.append(u'<b>-The Buildbot</b>')
    return {
        'body': u"\n".join(text),
        'type': 'html'}
  else:
    return None