configuration/buildbot/message_notifier.py - soc - Git at Google

 # 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
 _SOURCECODE_LINK_PREFIX = 'https://melange.googlesource.com/soc/+/'
 _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('<.*?>')
 _PYUNIT_LOGS = 'PyUnit Tests.stdio'
 _FUNCTIONAL_LOGS = 'Functional Tests.stdio'
 _PYLINT_LOGS = 'PyLint.stdio'
 _PYUNIT_NAME = 'PyUnit'
 _FUNCTIONAL_NAME = 'Functional'
 _PYLINT_NAME = 'PyLint'


 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())
     if (_PYUNIT_NAME in log_name) or (_FUNCTIONAL_NAME in log_name) or(
         _PYLINT_NAME 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] = log_info
   return logs


 def build_html_for_tests(logs, type):
   """Constructs html string list for a test given it's type.

   Args:
     logs(dictionary): A dictionary with Log name identifier mapped to
         _LogInfo objects.
     type: String, indicating the type of test - PyUnit or Functional.

   Returns:
     html_string_list(list): A list containing HTML strings which are processed
     logs from TestCommand Steps.
   """
   html_string_list = list()
   test_summary_log = logs[type]
   content = test_summary_log.log_body
   url = test_summary_log.log_url
   html_string_list.append(_DETAILED_LOG_OF_STEP_HTML % (type, url, url))
   html_string_list.append(_LINE_BREAK_HTML)
   html_string_list.append(_LAST_LINES_OF_LOG_HTML % (_LOG_LIMIT_LINES, type))
   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')))
   for line in unilist:
     html_string_list.append(u'<pre>' + line + u'</pre>')
   html_string_list.append(_LINE_BREAK_HTML)
   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 = _SOURCECODE_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>'
         u'<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 = _SOURCECODE_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>"
         u"<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 PyLint tests.
     pylint_logs = build_html_for_tests(logs, _PYLINT_LOGS)
     text += pylint_logs

     # get log for the PyUnit tests.
     pyunit_logs = build_html_for_tests(logs, _PYUNIT_LOGS)
     text += pyunit_logs

     # get log for the Functional tests.
     functional_logs = build_html_for_tests(logs, _FUNCTIONAL_LOGS)
     text += functional_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
	# 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
	_SOURCECODE_LINK_PREFIX = 'https://melange.googlesource.com/soc/+/'
	_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('<.*?>')
	_PYUNIT_LOGS = 'PyUnit Tests.stdio'
	_FUNCTIONAL_LOGS = 'Functional Tests.stdio'
	_PYLINT_LOGS = 'PyLint.stdio'
	_PYUNIT_NAME = 'PyUnit'
	_FUNCTIONAL_NAME = 'Functional'
	_PYLINT_NAME = 'PyLint'


	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())
	if (_PYUNIT_NAME in log_name) or (_FUNCTIONAL_NAME in log_name) or(
	_PYLINT_NAME 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] = log_info
	return logs


	def build_html_for_tests(logs, type):
	"""Constructs html string list for a test given it's type.

	Args:
	logs(dictionary): A dictionary with Log name identifier mapped to
	_LogInfo objects.
	type: String, indicating the type of test - PyUnit or Functional.

	Returns:
	html_string_list(list): A list containing HTML strings which are processed
	logs from TestCommand Steps.
	"""
	html_string_list = list()
	test_summary_log = logs[type]
	content = test_summary_log.log_body
	url = test_summary_log.log_url
	html_string_list.append(_DETAILED_LOG_OF_STEP_HTML % (type, url, url))
	html_string_list.append(_LINE_BREAK_HTML)
	html_string_list.append(_LAST_LINES_OF_LOG_HTML % (_LOG_LIMIT_LINES, type))
	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')))
	for line in unilist:
	html_string_list.append(u'<pre>' + line + u'</pre>')
	html_string_list.append(_LINE_BREAK_HTML)
	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 = _SOURCECODE_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>'
	u'<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 = _SOURCECODE_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>"
	u"<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 PyLint tests.
	pylint_logs = build_html_for_tests(logs, _PYLINT_LOGS)
	text += pylint_logs

	# get log for the PyUnit tests.
	pyunit_logs = build_html_for_tests(logs, _PYUNIT_LOGS)
	text += pyunit_logs

	# get log for the Functional tests.
	functional_logs = build_html_for_tests(logs, _FUNCTIONAL_LOGS)
	text += functional_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