Source code for bodhi_utils

# - 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
# 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 <>
#     Martin Krizek <>

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
        server = autoqa_conf.get('resources', 'bodhi_server')
    except ConfigParser.Error, e:
        server = ''

    # get auth credentials
    user = 'autoqa'
    pswd = ''
        fas_conf = SingleConfigParser()'/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)


[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 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 bodhi_already_commented(update, testname, arch): '''Check if AutoQA comment is already posted. Args: update -- Bodhi update object --or-- update title --or-- update ID --or-- package NVR testname -- the name of the test arch -- tested architecture Note: Only NVR allowed, not ENVR. See Returns: Tuple containing old result and time when the last comment was posted. If no comment is posted yet, or it is, but the update has been modified since, tuple will contain two empty strings. Throws: ValueError -- if no such update can be found ''' # if we received update title or ID, let's convert it to update object first if isinstance(update, str): u = query_update(update) if u: update = u else: raise ValueError("No such update: %s" % update) comment_re = r'AutoQA:[\s]+%s[\s]+test[\s]+(\w+)[\s]+on[\s]+%s' % (testname, arch) old_result = '' comment_time = '' autoqa_comments = [comment for comment in update['comments'] if comment['author'] == user] for comment in autoqa_comments: m = re.match(comment_re, comment['text']) if m is None: continue old_result = comment_time = comment['timestamp'] # check whether update was modified after the last posted comment if update['date_modified'] > comment_time: return ('','') return (old_result, comment_time)
[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 ='test_name') result ='result') arch ='arch') result_url ='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',, 'PASSED', time_span) == False print 'Passed' print '2. Test:', assert _is_bodhi_testresult_needed('FAILED',, 'PASSED', time_span) == True print 'Passed' print '3. Test:', assert _is_bodhi_testresult_needed('PASSED',, '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()