On Thu, 2011-01-20 at 13:22 +0100, Kamil Páral wrote:
Also add more docstrings and collapse methods _bodhi_already_commented() and _check_already_commented() into a single method bodhi_alread_commented(). Raise ValueError if we try to comment on update that does not exist.
Seems good to me, but I haven't done any testing in conjunction with depcheck yet. Once the depcheck branch is updated to accommodate this and other fixes ... I'll be happy to run a few tests.
Thanks, James
lib/python/bodhi_utils.py | 106 +++++++++++++++++++++++++++------------------ 1 files changed, 64 insertions(+), 42 deletions(-)
diff --git a/lib/python/bodhi_utils.py b/lib/python/bodhi_utils.py index 1bca502..2df40f8 100644 --- a/lib/python/bodhi_utils.py +++ b/lib/python/bodhi_utils.py @@ -28,26 +28,31 @@ from datetime import datetime from autoqa.config import SingleConfigParser, getbool, autoqa_conf import ConfigParser
-try:
- server = autoqa_conf.get('resources', 'bodhi_server')
-except ConfigParser.Error, e:
- server = ''
-try:
- fas_conf = SingleConfigParser()
- fas_conf.read_single(['fas.conf', '/etc/autoqa/fas.conf'])
- fas = fas_conf.get_section('fas')
- user = fas['username']
- pswd = fas['password']
- if not user or not pswd:
raise IOError
-except (IOError, ConfigParser.Error), e:
- # if there's no fas.conf or it's incomplete, we just log in anonymously.
- user = ''
- pswd = ''
+def _init():
- '''Initialize this module'''
- global user, pswd, bodhi
- try:
server = autoqa_conf.get('resources', 'bodhi_server')
- except ConfigParser.Error, e:
server = ''
-ops = {'base_url': server} if server else {} -bodhi = fedora.client.bodhi.BodhiClient(username=user, password=pswd, **ops)
- try:
fas_conf = SingleConfigParser()
fas_conf.read_single(['fas.conf', '/etc/autoqa/fas.conf'])
fas = fas_conf.get_section('fas')
user = fas['username']
pswd = fas['password']
if not user or not pswd:
raise IOError
- except (IOError, ConfigParser.Error), e:
# if there's no fas.conf or it's incomplete, we just log in anonymously.
user = ''
pswd = ''
- ops = {'base_url': server} if server else {}
- bodhi = fedora.client.bodhi.BodhiClient(username=user, password=pswd, **ops)
+_init()
def bodhitime(timestamp): '''Convert timestamp (seconds since Epoch, assumed to be local time) to a @@ -59,8 +64,32 @@ def parse_bodhitime(bodhitimestr): float-represented number of seconds since Epoch (local time)''' return time.mktime(time.strptime(bodhitimestr,'%Y-%m-%d %H:%M:%S'))
+def query_update(package):
- '''Get the last Bodhi update matching criteria.
Args:
package -- package NVR --or-- package name --or-- update title --or-- update ID
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
def bodhi_list(params, limit=100):
- '''Perform a bodhi 'list' method call, with pagination handling'''
- '''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
@@ -73,37 +102,30 @@ def bodhi_list(params, limit=100): params['tg_paginate_no'] += 1 return updates
-def _bodhi_already_commented(update, user, testname, arch): +def bodhi_already_commented(update, user, testname, arch): '''Check if the comment is already posted.
Args:
update -- The *title* of the update
update -- Bodhi update object --or-- update title --or-- update ID user -- username that posted the comment testname -- the name of the test arch -- tested architecture Returns:
Tuple containing old result and time when the last comment was posted,
if none comment is posted already, tuple will be empty.
- '''
- params = dict()
- params['package'] = update
- res = bodhi_list(params)
- return _check_already_commented(res[0], user, testname, arch)
-def _check_already_commented(update, user, testname, arch):
- '''Check if the comment is already posted.
Tuple containing old result and time when the last comment was posted.
If no comment is posted yet, tuple will contain two empty strings.
Args:
update -- Bodhi update object
user -- username that posted the comment
testname -- the name of the test
arch -- tested architecture
Returns:
Tuple containing old result and time when the last comment was posted,
if none comment is posted already, tuple will be empty.
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 = ''
@@ -197,7 +219,7 @@ def bodhi_post_testresult(update, testname, result, url, arch = 'noarch', karma comment = 'AutoQA: %s test %s on %s. The result can be found at: %s.' \ % (testname, result, arch, url) try:
(old_result, comment_time) = _bodhi_already_commented(update, user, testname, arch)
(old_result, comment_time) = bodhi_already_commented(update, user, 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):