# bodhi_utils.py - utility functions for dealing with bodhi
#
# Copyright 2010, Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Authors:
# Will Woods <wwoods@redhat.com>
# Martin Krizek <mkrizek@redhat.com>
import sys
import re
import time
import ConfigParser
import util
import fedora.client
from datetime import datetime
from autoqa import bodhi_update_state
from autoqa.config import SingleConfigParser, getbool, autoqa_conf
# global variables
user = None
pswd = None
bodhi = None
def _init():
'''Initialize this module'''
global user, pswd, bodhi
try:
server = autoqa_conf.get('resources', 'bodhi_server')
except ConfigParser.Error, e:
server = ''
# get auth credentials
user = 'autoqa'
pswd = ''
try:
fas_conf = SingleConfigParser()
fas_conf.read('/etc/autoqa/fas.conf')
fas = fas_conf.get_section('fas')
user = fas['username'] or user
pswd = fas['password']
except (ConfigParser.Error, KeyError), e:
print >> sys.stderr, 'Missing or invalid fas.conf: %s' % e
# XXX: python-2.4 doesn't support the following simplified syntax. When
# python-2.4 support is no longer required, use the following instead.
# ops = {'base_url': server} if server else {}
ops = {}
if server:
ops['base_url'] = server
bodhi = fedora.client.bodhi.BodhiClient(username=user, password=pswd, **ops)
_init()
[docs]def bodhitime(timestamp):
'''Convert timestamp (seconds since Epoch, assumed to be local time) to a
Bodhi-approved time string ('%Y-%m-%d %H:%M:%S')'''
return time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(timestamp))
[docs]def parse_bodhitime(bodhitimestr):
'''Convert a Bodhi-approved time string ('%Y-%m-%d %H:%M:%S') to a
float-represented number of seconds since Epoch (local time)'''
return time.mktime(time.strptime(bodhitimestr,'%Y-%m-%d %H:%M:%S'))
[docs]def query_update(package):
'''Get the last Bodhi update matching criteria.
Args:
package -- package NVR --or-- package name --or-- update title --or-- update ID
Note: Only NVR allowed, not ENVR. See https://fedorahosted.org/bodhi/ticket/592.
Returns: Most recent Bodhi update object matching criteria, or None if
no such available.
'''
res = bodhi.query(package=package)
if res['updates']:
return res['updates'][0]
else:
return None
[docs]def bodhi_list(params, limit=100):
'''Perform a bodhi 'list' method call, with pagination handling. This method
accepts more search params than the standard 'query' method.
Args:
params -- dictionary of params that will be passed to the 'list' method call.
E.g.: {'package':'foo-1.1.fc14'}
limit -- the maximum number of results returned
Returns: List of update objects matching criteria
'''
params['tg_paginate_limit'] = limit
params['tg_paginate_no'] = 1
updates = list()
num_items = 1 # this is a lie to get us in the loop
while num_items > len(updates):
r = bodhi.send_request('list', params)
num_items = r['num_items']
updates += r['updates']
params['tg_paginate_no'] += 1
return updates
[docs]def _is_bodhi_testresult_needed(old_result, comment_time, result, time_span):
'''Check if the comment is meant to be posted.
Args:
old_result -- the result of the last test
comment_time -- the comment time of the last test
result -- the result of the test
time_span -- waiting period before posting the same comment
Returns:
True if the comment will be posted, False otherwise.
'''
# the first comment or a comment with different result, post it
if not old_result or old_result != result:
return True
# If we got here, it means that the comment with the same result has been
# already posted, we now need to determine whether we can post the
# comment again or not.
# If the previous result is *not* 'FAILED', we won't post it in order not to
# spam developers.
# If the previous result *is* 'FAILED', we will need to check whether given
# time span expired, if so, we will post the same comment again to remind
# a developer about the issue.
if result != 'FAILED':
return False
posted_datetime = datetime.strptime(comment_time, '%Y-%m-%d %H:%M:%S')
delta = (datetime.utcnow() - posted_datetime)
# total_seconds() is introduced in python 2.7, until 2.7 is everywhere...
total_seconds = (delta.microseconds + (delta.seconds + delta.days * 24 * 3600) * 10**6) / 10**6
minutes = total_seconds/60.0
if minutes < time_span:
return False
return True
[docs]def bodhi_post_testresult(update, testname, result, url, arch = 'noarch',
karma = 0):
'''Post comment and karma to bodhi
Args:
update -- the *title* of the update comment on
testname -- the name of the test
result -- the result of the test
url -- url of the result of the test
arch -- tested architecture (default 'noarch')
karma -- karma points (default 0)
Returns:
True if comment was posted successfully or comment wasn't meant to be
posted (either posting is turned off or comment was already posted),
False otherwise.
'''
# TODO when new bodhi releases, add update identification by UPDATEID support
err_msg = 'Could not post a comment to bodhi'
if not update or not testname or not result or url == None:
sys.stderr.write('Incomplete arguments!\n%s\n' % err_msg)
return False
try:
if not getbool(autoqa_conf.get('notifications', 'send_bodhi_comments')):
raise ValueError
except ValueError:
print 'Sending bodhi comments is turned off. Test result will NOT be sent.'
# it's either False or not a bool -> it's false, do not send it (but
# return True since it's intentional, not an error)
return True
if not user or not pswd:
sys.stderr.write('Conf file containing FAS credentials is incomplete!'\
'\n%s\n' % err_msg)
return False
comment = 'AutoQA: %s test %s on %s. Result log: %s ' \
'(results are informative only)' % (testname, result, arch, url)
try:
(old_result, comment_time) = bodhi_already_commented(update, testname, arch)
time_span = int(autoqa_conf.get('notifications', 'bodhi_posting_comments_span'))
if not _is_bodhi_testresult_needed(old_result, comment_time, result, time_span):
print 'The test result already posted to bodhi.'
return True
bodhi_update = query_update(update)
parsed_results = []
for comment in bodhi_update['comments']:
if comment['author'] == user:
parsed_results.append(_parse_result_from_comment(comment))
kojitag = 'updates-testing'
if bodhi_update['request'] == 'stable' or bodhi_update['status'] == 'stable':
kojitag = 'updates'
new_result = {'time':datetime.utcnow(), 'testname':testname,
'result':result, 'arch':arch}
send_email = _is_bodhi_comment_email_needed(update, parsed_results,
new_result, kojitag=kojitag)
# log email decision
if send_email:
print "Bodhi email will be sent"
else:
print "Bodhi email will NOT be sent"
print 'update: %s \n comment: %s \n karma: %d' % (update, comment, karma)
print 'update: %s \n comment: %s \n karma: %s \n send_email: %s' % (type(update), type(comment), type(karma), type(send_email))
if not bodhi.comment(update, comment, karma, send_email):
sys.stderr.write('%s\n') % err_msg
return False
print 'The test result was sent to bodhi successfully.'
except Exception, e:
sys.stderr.write('An error occured: %s\n' % e)
sys.stderr.write('Could not connect to bodhi!\n%s\n' % err_msg)
return False
return True
[docs]def build2update(builds, strict=False):
'''Find matching Bodhi updates for provided builds.
@param builds builds to search for; iterable of strings
@param strict if False, incomplete Bodhi updates are allowed.
if True, every Bodhi update will be compared with the set of
provided builds. If there is an Bodhi update which contains
builds not provided in @param builds, that update is marked
as incomplete and removed from the result - i.e. all builds
from @param builds that were part of this incomplete update
are placed in the second dictionary of the result tuple.
@return tuple of two dictionaries:
* The first dict provides mapping between builds and their updates
where no error occured. {build (string): Bodhi update (Bunch)}
* The second dict provides mapping between builds and their updates
where some error occured.
{build (string): Bodhi update (Bunch) or None}
The value is None if the matching Bodhi update could not be found
(the only possible cause of failure if @param strict=False). Or
the value is a Bodhi update that was incomplete (happens only if
@param strict=True).
'''
build2update = {}
failures = {}
# Bodhi works with NVR only, but we have to ensure we receive and return
# even other formats, like ENVRA. So we need to convert internally.
builds_nvr = set([util.envra(build, 'nvr') for build in builds])
for build in sorted(builds):
# if already found answer for this build, skip it
if build in build2update or build in failures:
continue
# get Bodhi update object
print 'Searching Bodhi updates for: %s' % build
update = query_update(util.envra(build, 'nvr'))
if update is None:
# no such Bodhi update
failures[build] = None
continue
# all builds listed in the update
bodhi_builds = set([build['nvr'] for build in update['builds']])
# builds *not* provided in @param builds but part of the update (NVRs)
missing_builds = bodhi_builds.difference(builds_nvr)
# builds provided in @param builds and part of the update
matched_builds = [build for build in builds if util.envra(build, 'nvr')
in bodhi_builds]
# reject incomplete updates
if missing_builds and strict:
for matched_build in matched_builds:
failures[matched_build] = update
continue
# create item in build2update
for build in matched_builds:
build2update[build] = update
# some build might have been added to both @var failures and @var build2update
# if that happens, remove it from @var build2update
for build in failures:
build2update.pop(build, None)
return (build2update, failures)
def _is_bodhi_comment_email_needed(update_name, parsed_comments, new_result,
kojitag):
'''
Determines whether or not to send an email with the comment posted to bodhi.
Uses previous comments on update in order to determine current state
Args:
update_name -- name of the update to be tested
parsed_comments -- already existing AutoQA comments for the update
new_result -- the new result to be posted
kojitag -- the koji tag being tested (affects the expected tests)
'''
print "checking email needed for %s (%s)" % (update_name, kojitag)
# compute current state
update_state = bodhi_update_state.BodhiUpdateState(update_name,
kojitag=kojitag)
for result in parsed_comments:
update_state.add_result(result)
current_state = update_state.get_state()
update_state.add_result(new_result)
new_state = update_state.get_state()
update_state_change = current_state != new_state
email_config = _get_bodhi_email_config()
# Can't really do anything if there is no email config information
# available, just return False to not send an email
if email_config is None:
return False
# time to figure out whether or not to send an email
send_email = False
print 'update state : %s to %s' % (str(current_state), str(new_state))
# if we always send an email, just return true
if getbool(email_config['email_always_send']):
return True
# by default, we want to send email on state change, so start from there
if update_state_change:
send_email = True
# if configured, don't send an email if all tests are passed unless the
# state change is from fail to pass, then we want to leave the current
# state alone.
if not getbool(email_config['email_all_passed']):
if new_state is bodhi_update_state.PASS \
and current_state is not bodhi_update_state.FAIL:
print 'not sending email due to configuration for email_all_passed'
send_email = False
# send emails on test state changes if configured
if getbool(email_config['email_test_state_change']):
if update_state.did_test_change():
print 'sending email due to configuration for email_test_state_change'
send_email = True
# don't send results from updates_testing if not configured
if not getbool(email_config['email_updates_testing']):
if kojitag.endswith('updates-testing'):
print 'not sending email due to configuration for email_updates_testing'
send_email = False
return send_email
def _parse_result_from_comment(comment):
'''
Parses timestamp and results from bodhi comments
Args:
comment -- the string containing comment
'''
comment_time = datetime.strptime(comment['timestamp'], '%Y-%m-%d %H:%M:%S')
# comment form to match:
# 'AutoQA: %s test %s on %s. Result log: %s (results are informative only)' \
# note that it isn't looking for the autoqa user right now, that needs to
# be done in any code that calls this
comment_match = re.match(r'AutoQA: (?P<test_name>\w+) test (?P<result>\w+)'\
r' on (?P<arch>\w+)\. Result log:[\t \n]+'\
r'(?P<result_url>[/:\w]+).*', comment['text'])
test_name = ''
result = ''
arch = ''
result_url = ''
if comment_match:
test_name = comment_match.group('test_name')
result = comment_match.group('result')
arch = comment_match.group('arch')
result_url = comment_match.group('result_url')
else:
print >> sys.stderr, 'Failed to parse bodhi comment: %s' % comment['text']
return {'time':comment_time, 'testname':test_name, 'result':result, 'arch':arch}
def _get_bodhi_email_config():
'''
Retrieve email configuration settings from autoqa.conf
'''
bodhi_conf = {}
try:
bodhi_conf = autoqa_conf.get_section('bodhi_email')
except ConfigParser.Error, e:
print >> sys.stderr, 'Error getting bodhi configuration email from'\
' autoqa.conf: %s' % e
# not sure how this error condition should be handled, return None for now
return None
return bodhi_conf
def _self_test():
'''
Simple self test.
'''
from datetime import timedelta
time_span = int(autoqa_conf.get('notifications', 'bodhi_posting_comments_span'))
try:
print '1. Test:',
assert _is_bodhi_testresult_needed('PASSED', datetime.now, 'PASSED', time_span) == False
print 'Passed'
print '2. Test:',
assert _is_bodhi_testresult_needed('FAILED', datetime.now, 'PASSED', time_span) == True
print 'Passed'
print '3. Test:',
assert _is_bodhi_testresult_needed('PASSED', datetime.now, 'FAILED', time_span) == True
print 'Passed'
print '4. Test:',
date = (datetime.utcnow() - timedelta(minutes=time_span)).\
strftime('%Y-%m-%d %H:%M:%S')
assert _is_bodhi_testresult_needed('FAILED', date, 'FAILED', time_span) == True
print 'Passed'
print '5. Test:',
date = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')
assert _is_bodhi_testresult_needed('FAILED', date, 'FAILED', time_span) == False
print 'Passed'
print '6. Test:',
assert _is_bodhi_testresult_needed('', '', 'FAILED', time_span) == True
print 'Passed'
except AssertionError:
print 'Failed [!!!]'
if __name__ == '__main__':
_self_test()