Differential D67

Bodhi Comment directive
ClosedPublic
Actions

Authored by jskladan on Apr 23 2014, 11:11 AM.

Details

Reviewers
kparal
tflink
Maniphest Tasks
T46: bodhi comment directive module
Summary

Updated version of the Bodhi Comment directive

Test Plan

Unit tests appended. I still need to write tests for the _is_email_needed stuff, though.

Diff Detail

Branch
arcpatch-D67
Lint
No Linters Available
Unit
No Unit Test Coverage
jskladan updated this revision.Apr 29 2014, 12:42 PM

Added some tests

jskladan updated this revision.Apr 29 2014, 12:47 PM

Trying to post proper diff

tflink requested changes to this revision.Apr 29 2014, 3:18 PM

Minor change request for the test class name.

If you've tested this with a local taskotron runner and fake_fedorainfra, I'm ok with the code. Bodhi comments need to go away soon and I'm not all that worried about fully unit tested code as long as it works.

Just to make sure I understood you over IRC, you did test this with fake_fedorainfra, right?

testing/test_bodhi_comment_directive.py
16

Shouldn't this be TestBodhiCommentReport or something that isn't Resultsdb?

jskladan added a comment.Apr 29 2014, 3:24 PM

I did test this with fake-fedorainfra, but not with local taskotron instance. I have not yet one set up'n'running. I'll try and set one up, but could you run it in the meantime?

Thanks

testing/test_bodhi_comment_directive.py
16

Ouch, thanks for the catch. Ye'ol' copy-paste error strikes again.

tflink added a comment.Apr 29 2014, 3:37 PM

I did test this with fake-fedorainfra, but not with local taskotron instance. I have not yet one set up'n'running. I'll try and set one up, but could you run it in the meantime?

I'm not so worried about the differences between running tasks locally through the command line and running them in the full stg system. We're eventually going to test it in stg or a dev system but in theory, that should be almost the same as running the runner locally while pointed at a fake_fedorainfra instance

Can you go into more detail on how you tested this code?

jskladan added a comment.Apr 29 2014, 4:01 PM

I used a code very similar to this (managed to delete it in the mean time, as it was even nastier than this is):

def test_with_boji(self):
    import fedora.client
    Config = config.get_config()
    Config.fas_password = "secret"
    Config.bodhi_server = "http://virt_resultsdb:5000/bodhi/"
    Config.reporting_enabled = True
    Config.report_to_bodhi = True

    mock_bodhi = fedora.client.bodhi.BodhiClient(username=Config.fas_username,
                                            password=Config.fas_password,
                                            base_url=Config.bodhi_server)

    directive = bodhi_comment_directive.BodhiCommentDirective()
    directive.bodhi_api.bodhi = mock_bodhi

    ref_envdata = {"checkname": "depcheck"}
    ref_input = {"doreport": "onchange"}

    check_details = [check.CheckDetail(
                item = 'perl-Term-ProgressBar-2.15-1.fc19',
                report_type = check.ReportType.BODHI_UPDATE,
                outcome = result,
                arch = 'x86_64') for result in ['FAILED', 'FAILED', 'PASSED', 'PASSED', 'FAILED']]


    for detail in check_details:
        ref_input['results'] = "TAP version 13\n1..1\n%s" % check.export_TAP(detail)
        directive.process(ref_input, ref_envdata)

And run it with different checkname, doreport and arch values.

tflink added a comment.Apr 30 2014, 6:34 AM

I'm having trouble running this against a fakefedorainfra instance. I'm getting tracebacks like:

[libtaskotron:logger.py:34] 2014-04-30 00:16:02 CRITICAL Traceback (most recent call last):
  File "runtask.py", line 4, in <module>
    runner.main()
  File "/home/tflink/code/taskotron/libtaskotron/libtaskotron/runner.py", line 184, in main
    task_runner.run()
  File "/home/tflink/code/taskotron/libtaskotron/libtaskotron/runner.py", line 33, in run
    self.do_actions()
  File "/home/tflink/code/taskotron/libtaskotron/libtaskotron/runner.py", line 107, in do_actions
    self.do_single_action(action)
  File "/home/tflink/code/taskotron/libtaskotron/libtaskotron/runner.py", line 93, in do_single_action
    directive_callable = getattr(directive_object, directive_classname)()
  File "/home/tflink/code/taskotron/libtaskotron/libtaskotron/directives/bodhi_comment_directive.py", line 381, in __init__
    self.bodhi_api = bodhi_utils()
TypeError: 'module' object is not callable

Using the command (substituting where appropriate):

python runtask.py -i <update_id> -t bodhi_id /path/to/task-examplebodhi/examplebodhi.yml -a x86_64

And the task yaml file {F1856}

jskladan added a comment.Apr 30 2014, 7:59 AM

Tim, are you sure you have the right pieces of code? Since the line you get tracebacks on:

self.bodhi_api = bodhi_utils()

is a tiny bit different in the diff provided here:

self.bodhi_api = bodhi_utils
tflink added a comment.Apr 30 2014, 1:15 PM

Tim, are you sure you have the right pieces of code? Since the line you get tracebacks on is a tiny bit different in the diff provided here

Sorry, I grabbed the wrong traceback. I tried a few simple things to see if I could make it work and must have not scrolled back far enough.

[libtaskotron:bodhi_comment_directive.py:360] 2014-04-30 00:13:34 ERROR   'module' object has no attribute 'query_update'
Traceback (most recent call last):
  File "/home/tflink/code/taskotron/libtaskotron/libtaskotron/directives/bodhi_comment_directive.py", line 327, in _post_testresult
    (old_result, comment_time) = _already_commented(bodhi_api, update, testname, arch)
  File "/home/tflink/code/taskotron/libtaskotron/libtaskotron/directives/bodhi_comment_directive.py", line 224, in _already_commented
    u = bodhi_api.query_update(update)

Were you able to get that task yaml file to work with libtaskotron and a fakefedorainfra instance?

jskladan added a comment.Apr 30 2014, 1:34 PM

*sigh* I know where the problem is - apparently the bodhi_utils.py changed last Thursday... 7b24f6fe00f087bca9ba20498ca7c7adc91178c9 and the bodhi_comment_directive is written based on how it looked the whole time before it.

jskladan updated this revision.Apr 30 2014, 2:45 PM
  • T46 - Fixes to work with new bodhi_utils
jskladan updated this revision.Apr 30 2014, 2:48 PM
  • T46 - Fixed the name of test class
tflink accepted this revision.Apr 30 2014, 3:34 PM

It works with a local runner and fake_fedorainfra once I apply the patch from D79.

Please fix the missing docs about checkname and merge into develop

libtaskotron/directives/bodhi_comment_directive.py
373

This isn't quite right: it needs checkname as well

jskladan closed this revision.May 6 2014, 7:52 AM

Revision Contents

Diff 217

View Options

libtaskotron/config_defaults.py

Show All 36 Lines29 def __init__(self):
37 37​
38 self.koji_url = 'http://koji.fedoraproject.org/kojihub' 38​ self.koji_url = 'http://koji.fedoraproject.org/kojihub'
39 self.pkg_url = 'http://kojipkgs.fedoraproject.org/packages' 39​ self.pkg_url = 'http://kojipkgs.fedoraproject.org/packages'
40 self.bodhi_server = 'https://admin.fedoraproject.org/updates/' 40​ self.bodhi_server = 'https://admin.fedoraproject.org/updates/'
41 self.resultsdb_server = \ 41​ self.resultsdb_server = \
42 'http://resultsdb.qa.fedoraproject.org/resultsdb/api/v1.0/' 42​ 'http://resultsdb.qa.fedoraproject.org/resultsdb/api/v1.0/'
43 43​
44 self.bodhi_email_failed_span = 4320 44​ self.bodhi_email_failed_span = 4320
45​ self.bodhi_posting_comments_span = 4320
45 46​
46 self.tmpdir = '/var/tmp/taskotron' 47​ self.tmpdir = '/var/tmp/taskotron'
47 self.logdir = '/var/log/taskotron' 48​ self.logdir = '/var/log/taskotron'
48 self.logfile = os.path.join(self.logdir, 'taskotron.log') 49​ self.logfile = os.path.join(self.logdir, 'taskotron.log')
49 50​
50 self.fas_username = 'taskotron' 51​ self.fas_username = 'taskotron'
51 self.fas_password = '' 52​ self.fas_password = ''
52 53​
▲ Show 20 Lines • Show All 45 Lines • Show Last 20 Lines
View Options

libtaskotron/directives/bodhi_comment_directive.py

  • This file was added.
1​# -*- coding: utf-8 -*-
2​# Copyright 2014, Red Hat, Inc.
3​#
4​# This program is free software; you can redistribute it and/or modify
5​# it under the terms of the GNU General Public License as published by
6​# the Free Software Foundation; either version 2 of the License, or
7​# (at your option) any later version.
8​#
9​# This program is distributed in the hope that it will be useful,
10​# but WITHOUT ANY WARRANTY; without even the implied warranty of
11​# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12​# GNU General Public License for more details.
13​#
14​# You should have received a copy of the GNU General Public License along
15​# with this program; if not, write to the Free Software Foundation, Inc.,
16​# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17​#
18​# Authors:
19​# Tim Flink <tflink@redhat.com>
20​# Josef Skladanka <jskladan@redhat.com>
21​
22​directive_class = 'BodhiCommentDirective'
23​
24​
25​from libtaskotron.directives import BaseDirective
26​
27​from libtaskotron.logger import log
28​from libtaskotron import check
29​from libtaskotron import config
30​from libtaskotron import bodhi_utils
31​
32​from libtaskotron.exceptions import TaskotronDirectiveError, TaskotronValueError
33​from libtaskotron.rpm_utils import basearch
34​
35​import datetime
36​import re
37​
38​RE_COMMENT = re.compile(r'Taskotron: (?P<test_name>\w+) test (?P<result>\w+)'\
39​ r' on (?P<arch>\w+)\. Result log:[\t \n]+'\
40​ r'(?P<result_url>[/:\w]+).*')
41​
42​
43​# symbols used for consistant declaration of state
44​PASS = 'state_PASS'# <%s>' % __name__
45​FAIL = 'state_FAIL'# <%s>'% __name__
46​INCOMPLETE = 'state_INCOMPLETE'# <%s>' % __name__
47​UNKNOWN = 'state_UNKNOWN'# <%s>' % __name__
48​
49​class BodhiUpdateState:
50​ def __init__(self, name, kojitag='updates'):
51​ self.test_states = { 'depcheck_32':'NORUN',
52​ 'depcheck_64':'NORUN',
53​ 'upgradepath':'NORUN'
54​ }
55​
56​ # since upgradepath is only run on updates-pending,
57​ # set it to passed if we're working on updates-testing
58​ if kojitag == 'updates-testing':
59​ self.test_states['upgradepath'] = 'PASSED'
60​
61​ self.result_history = []
62​ self.update_name = name
63​ self.test_change = False
64​
65​ def num_results(self):
66​ ''' Retrieve the number of results currently held '''
67​ return len(self.result_history)
68​
69​ def add_result(self, new_result):
70​ ''' Update the current state with a new result.
71​
72​ Args:
73​ new_result: a dictionary containing the following keys:
74​ time (datetime object), testname, result, arch
75​ '''
76​
77​ self.result_history.append(new_result)
78​
79​ # copy the current test state so that we can compare later
80​ old_state = self.test_states.copy()
81​
82​ # for depcheck, we only care about i386 and x86_64 arches
83​ # anything else will be ignored
84​ if new_result['testname'] == 'depcheck':
85​ arch = basearch(new_result['arch'])
86​
87​ if arch == 'i386':
88​ self.test_states['depcheck_32'] = new_result['result']
89​
90​ elif arch == 'x86_64':
91​ self.test_states['depcheck_64'] = new_result['result']
92​
93​ else:
94​ log.info("Arch %s is not valid for depcheck. Ignoring." \
95​ % new_result['arch'])
96​
97​ # don't care about the arch for upgrade path. It should be noarch
98​ # but the results aren't changed if it isn't
99​ elif new_result['testname'] == 'upgradepath':
100​ self.test_states['upgradepath'] = new_result['result']
101​
102​ # detect if any tests changed from the last result
103​ if old_state != self.test_states:
104​ self.test_change = True
105​ else:
106​ self.test_change = False
107​
108​ def get_state(self):
109​ ''' Determine state based on currently known results '''
110​
111​ # print out the current state for debugging purposes
112​ state_str = "Current State of update %s :" % self.update_name
113​ state_str += '\n %s' % self.test_states
114​ for result in self.result_history:
115​ state_str += '\n %s' % result
116​
117​ log.info(state_str)
118​
119​ test_values = set(self.test_states.values())
120​ if 'NORUN' in test_values:
121​ return INCOMPLETE
122​ if 'FAILED' in test_values:
123​ return FAIL
124​ if 'PASSED' in test_values and len(test_values) == 1:
125​ return PASS
126​
127​ return UNKNOWN
128​
129​ def did_test_change(self):
130​ ''' Determine whether the state of any tests changed with the addition
131​ of the last result'''
132​ return self.test_change
133​
134​
135​def _is_comment_email_needed(update_name, parsed_comments, new_result,
136​ kojitag):
137​ '''
138​ Determines whether or not to send an email with the comment posted to bodhi.
139​ Uses previous comments on update in order to determine current state
140​
141​ Args:
142​ update_name -- name of the update to be tested
143​ parsed_comments -- already existing AutoQA comments for the update
144​ new_result -- the new result to be posted
145​ kojitag -- the koji tag being tested (affects the expected tests)
146​ '''
147​
148​ Config = config.get_config()
149​
150​ log.info("checking email needed for %s (%s)", update_name, kojitag)
151​ # compute current state
152​ update_state = BodhiUpdateState(update_name, kojitag=kojitag)
153​ for result in parsed_comments:
154​ update_state.add_result(result)
155​
156​ current_state = update_state.get_state()
157​ update_state.add_result(new_result)
158​ new_state = update_state.get_state()
159​
160​ log.info('update state : %s to %s', current_state, new_state)
161​
162​ # by default, we want to send email on state change, so start from there
163​ send_email = True
164​
165​ if new_state is PASS and current_state is not FAIL:
166​ send_email = False
167​
168​ return send_email
169​
170​def _parse_result_from_comment(comment):
171​ '''
172​ Parses timestamp and results from bodhi comment
173​
174​ Args:
175​ comment -- the 'comment' part of bodhi update object
176​ '''
177​
178​ comment_time = datetime.datetime.strptime(comment['timestamp'], '%Y-%m-%d %H:%M:%S')
179​
180​ # comment form to match:
181​ # 'Taskotron: %s test %s on %s. Result log: %s'
182​ # note that it isn't looking for the autoqa user right now, that needs to
183​ # be done in any code that calls this
184​ comment_match = RE_COMMENT.match(comment['text'])
185​
186​ test_name = ''
187​ result = ''
188​ arch = ''
189​
190​ if comment_match:
191​ test_name = comment_match.group('test_name')
192​ result = comment_match.group('result')
193​ arch = comment_match.group('arch')
194​ else:
195​ log.warning('Failed to parse bodhi comment: %r', comment['text'])
196​
197​ return {'time':comment_time, 'testname':test_name,
198​ 'result':result, 'arch':arch}
199​
200​
201​def _already_commented(bodhi_api, update, testname, arch):
202​ '''Check if Taskotron comment is already posted.
203​
204​ Args:
205​ update -- Bodhi update object --or-- update title --or-- update ID --or-- package NVR
206​ testname -- the name of the test
207​ arch -- tested architecture
208​
209​ Note: Only NVR allowed, not ENVR. See https://fedorahosted.org/bodhi/ticket/592.
210​
211​ Returns:
212​ Tuple containing old result and time when the last comment was posted.
213​ If no comment is posted yet, or it is, but the update
214​ has been modified since, tuple will contain two empty strings.
215​
216​ Throws:
217​ TaskotronValueError -- if no such update can be found
218​ '''
219​
220​ Config = config.get_config()
221​
222​ # if we received update title or ID, let's convert it to update object first
223​ if isinstance(update, unicode) or isinstance(update, str):
224​ u = bodhi_api.query_update(update)
225​ if u:
226​ update = u
227​ else:
228​ raise TaskotronValueError("No such update: %s" % update)
229​
230​ comment_re = r'Taskotron:[\s]+%s[\s]+test[\s]+(\w+)[\s]+on[\s]+%s' % (testname, arch)
231​ old_result = ''
232​ comment_time = ''
233​
234​ taskotron_comments = [comment for comment in update['comments']
235​ if comment['author'] == Config.fas_username]
236​ for comment in taskotron_comments:
237​ m = re.match(comment_re, comment['text'])
238​ if m is None:
239​ continue
240​ old_result = m.group(1)
241​ comment_time = comment['timestamp']
242​ # check whether update was modified after the last posted comment
243​ if update['date_modified'] > comment_time:
244​ return ('','')
245​ return (old_result, comment_time)
246​
247​def _is_comment_needed(old_result, comment_time, result, time_span = None):
248​ '''Check if the comment is meant to be posted.
249​
250​ Args:
251​ old_result -- the result of the last test
252​ comment_time -- the comment time of the last test
253​ result -- the result of the test
254​ time_span -- waiting period (in minutes) before posting the same comment
255​
256​ Returns:
257​ True if the comment will be posted, False otherwise.
258​ '''
259​ # the first comment or a comment with different result, post it
260​ if not old_result or old_result != result:
261​ return True
262​
263​ # If we got here, it means that the comment with the same result has been
264​ # already posted, we now need to determine whether we can post the
265​ # comment again or not.
266​ # If the previous result is *not* 'FAILED', we won't post it in order not to
267​ # spam developers.
268​ # If the previous result *is* 'FAILED', we will need to check whether given
269​ # time span expired, if so, we will post the same comment again to remind
270​ # a developer about the issue.
271​
272​ if result != 'FAILED':
273​ return False
274​
275​ if time_span is None:
276​ Config = config.get_config()
277​ time_span = Config.bodhi_posting_comments_span
278​
279​
280​ posted_datetime = datetime.datetime.strptime(comment_time, '%Y-%m-%d %H:%M:%S')
281​
282​ delta = (datetime.datetime.utcnow() - posted_datetime)
283​ # total_seconds() is introduced in python 2.7, until 2.7 is everywhere...
284​ delta_minutes = (delta.seconds + delta.days * 24 * 3600) / 60
285​
286​ if delta_minutes < time_span:
287​ return False
288​
289​ return True
290​
291​
292​def _post_testresult(bodhi_api, update, testname, result, url,
293​ arch = 'noarch', karma = 0, doreport='onchange'):
294​ '''Post comment and karma to bodhi
295​
296​ Args:
297​ update -- the *title* of the update comment on
298​ testname -- the name of the test
299​ result -- the result of the test
300​ url -- url of the result of the test
301​ arch -- tested architecture (default 'noarch')
302​ karma -- karma points (default 0)
303​ doreport -- set to 'all' to force posting bodhi comment
304​
305​ Returns:
306​ True if comment was posted successfully or comment wasn't meant to be
307​ posted (either posting is turned off or comment was already posted),
308​ False otherwise.
309​ '''
310​
311​ # TODO when new bodhi releases, add update identification by UPDATEID support
312​
313​ Config = config.get_config()
314​
315​ if not update or not testname or not result or url == None:
316​ log.error("bodhi_post_testresults() requires non-empty update, "\
317​ "testname, result and url arguments.")
318​ return False
319​
320​ if not Config.fas_username or not Config.fas_password:
321​ log.error('Conf file containing FAS credentials is incomplete!')
322​ return False
323​
324​ comment = 'Taskotron: %s test %s on %s. Result log: %s ' \
325​ '(results are informative only)' % (testname, result, arch, url)
326​ try:
327​ (old_result, comment_time) = _already_commented(bodhi_api, update, testname, arch)
328​
329​ comment_needed = _is_comment_needed(old_result, comment_time, result)
330​
331​ if not comment_needed and doreport != 'all':
332​ log.info('Current test result is already present in bodhi.')
333​ return True
334​
335​ bodhi_update = bodhi_api.query_update(update)
336​ parsed_results = []
337​ for found_comment in bodhi_update['comments']:
338​ if found_comment['author'] == Config.fas_username:
339​ parsed_results.append(_parse_result_from_comment(found_comment))
340​
341​ kojitag = 'updates-testing'
342​ if bodhi_update['request'] == 'stable' or bodhi_update['status'] == 'stable':
343​ kojitag = 'updates'
344​
345​ new_result = {'time':datetime.datetime.utcnow(), 'testname':testname,
346​ 'result':result, 'arch':arch}
347​
348​ send_email = _is_comment_email_needed(update, parsed_results,
349​ new_result, kojitag)
350​
351​
352​ if not bodhi_api.client.comment(update, comment, karma, send_email):
353​ log.error('Could not post a comment to Bodhi. %r, %r, %r, %r',
354​ update, comment, karma, send_email)
355​ return False
356​
357​ log.info('The test result was sent to bodhi successfully.')
358​
359​ except Exception as e:
360​ log.exception(e)
361​ return False
362​
363​ return True
364​
365​
366​
367​class BodhiCommentDirective(BaseDirective):
368​ """The bodhi_comment directive interfaces with Bodhi to create
369​ a comment in Bodhi
370​
371​ format: "bodhi_comment:
372​ results: <TAP13>
373​ doreport: [all, onchange]"
tflinkUnsubmitted
Not Done

This isn't quite right: it needs checkname as well

374​ """
375​ def __init__(self, bodhi_api=None):
376​
377​ if bodhi_api:
378​ self.bodhi_api = bodhi_api
379​ else:
380​ self.bodhi_api = bodhi_utils.BodhiUtils()
381​
382​ def process(self, input_data, env_data):
383​ output_data = []
384​
385​
386​ Config = config.get_config()
387​ if not (Config.reporting_enabled and Config.report_to_bodhi):
388​ log.info("Reporting to Bodhi is disabled.")
389​ return
390​
391​
392​ if not ('doreport' in input_data and 'results' in input_data):
393​ detected_args = ', '.join(input_data.keys())
394​ raise TaskotronDirectiveError("The bodhi_comment directive "\
395​ "requires 'doreport' and 'results' arguments. "\
396​ "Detected arguments: %s" % detected_args)
397​
398​ if input_data['doreport'] not in ['all', 'onchange']:
399​ raise TaskotronDirectiveError("The argument 'doreport' is"\
400​ "set to invalid value '%s'. Valid values are: "\
401​ "'all', 'onchange'.")
402​
403​ if 'checkname' not in env_data:
404​ raise TaskotronDirectiveError("The bodhi_comment directive "\
405​ "requires checkname.")
406​
407​ try:
408​ check_details = check.import_TAP(input_data['results'])
409​ except TaskotronValueError as e:
410​ log.exception(e)
411​ raise TaskotronDirectiveError("Failed to load the 'results':"\
412​ "%s" % e)
413​
414​ # Filter the results of "BODHI_UPDATE" type
415​ check_details = [r for r in check_details
416​ if r.report_type == check.ReportType.BODHI_UPDATE]
417​
418​ # ? Log when no results of the type are found ?
419​
420​ for detail in check_details:
421​ #TODO: replace url with proper URL
422​ outcome = _post_testresult(self.bodhi_api, detail.item,
423​ env_data['checkname'],
424​ detail.outcome,
425​ url = "http://example.com",
426​ doreport = input_data['doreport'],
427​ arch = detail.keyvals.get('arch', 'noarch'),
428​ )
429​
430​ if not outcome:
431​ log.warning("Failed to post Bodhi Comment: `%s` `%s` `%s`",
432​ detail.item,
433​ env_data['checkname'],
434​ detail.outcome)
435​
436​ output_data.append(outcome)
437​
438​ return output_data
439​
View Options

testing/test_bodhi_comment_directive.py

  • This file was added.
1​"""Unit tests for libtaskotron/directives/resultsdb_directive.py"""
2​
3​import pytest
4​import datetime
5​import itertools
6​from dingus import Dingus
7​from copy import deepcopy
8​
9​from libtaskotron.directives import bodhi_comment_directive
10​from libtaskotron.exceptions import TaskotronDirectiveError, TaskotronValueError
11​
12​from libtaskotron import check
13​from libtaskotron import config
14​
15​
16​class TestBodhiCommentReport():
tflinkUnsubmitted
Not Done

Shouldn't this be TestBodhiCommentReport or something that isn't Resultsdb?

jskladanAuthorUnsubmitted
Not Done

Ouch, thanks for the catch. Ye'ol' copy-paste error strikes again.

17​
18​ def setup_method(self, method):
19​ '''Run this before every test invocation'''
20​ config._config = None
21​ Config = config.get_config()
22​ Config.reporting_enabled = True
23​ Config.report_to_bodhi = True
24​
25​
26​ self.comment = {
27​ "author": Config.fas_username,
28​ "timestamp": "2014-01-31 10:11:12",
29​ "text": "Taskotron: depcheck test PASSED on x86_64. Result log: http://example.com",
30​ }
31​
32​ self.update = {
33​ 'request': 'stable',
34​ 'comments': [self.comment],
35​ 'date_modified': "2014-01-31 10:00:00"
36​ }
37​
38​ self.stub_bodhi = Dingus(query_update__returns = self.update)
39​ self.directive = bodhi_comment_directive.BodhiCommentDirective(self.stub_bodhi)
40​
41​ self.cd = check.CheckDetail(
42​ item = 'update_title',
43​ report_type = check.ReportType.BODHI_UPDATE,
44​ outcome = 'PASSED',
45​ )
46​
47​
48​ #TODO: remove this ugly hack - it is here because of the bug in bayeux
49​ tap = "TAP version 13\n1..1\n%s" % check.export_TAP(self.cd)
50​
51​
52​ self.ref_input = {
53​ 'results': tap,
54​ 'doreport': 'onchange',
55​ }
56​ self.ref_envdata = {
57​ 'checkname': 'depcheck',
58​ }
59​
60​ def test_config_reporting_disabled(self):
61​ """Checks config options can disable reporting."""
62​ Config = config.get_config()
63​
64​ for r_e, r_t_b in ((False, False), (False, True), (True, False)):
65​ Config.reporting_enabled = r_e
66​ Config.report_to_bodhi = r_t_b
67​
68​ assert self.directive.process(self.ref_input, self.ref_envdata) is None
69​
70​ def test_missing_arguments_in_inputdata(self):
71​ """Check if arguments raise exception"""
72​ with pytest.raises(TaskotronDirectiveError):
73​ input_data = deepcopy(self.ref_input)
74​ del(input_data['results'])
75​ self.directive.process(input_data, self.ref_envdata)
76​
77​ with pytest.raises(TaskotronDirectiveError):
78​ input_data = deepcopy(self.ref_input)
79​ del(input_data['doreport'])
80​ self.directive.process(input_data, self.ref_envdata)
81​
82​ def test_wrong_doreport_value(self):
83​ """Checks whether unsupported doreport value"""
84​ with pytest.raises(TaskotronDirectiveError):
85​ input_data = deepcopy(self.ref_input)
86​ input_data['doreport'] = 'FooBar'
87​ self.directive.process(input_data, self.ref_envdata)
88​
89​ def test_missing_checkname(self, monkeypatch):
90​ """Checks if missing checkname raises exception"""
91​ with pytest.raises(TaskotronDirectiveError):
92​ envdata = deepcopy(self.ref_envdata)
93​ del(envdata['checkname'])
94​
95​ self.directive.process(self.ref_input, envdata)
96​
97​ def test_failed_tap_import(self, monkeypatch):
98​ """Checks if failed TAP import raises exception"""
99​ def mock_raise(self, *args, **kwargs):
100​ raise TaskotronValueError("Testing Error")
101​
102​ monkeypatch.setattr(check, 'import_TAP', mock_raise)
103​
104​ with pytest.raises(TaskotronDirectiveError):
105​ self.directive.process(self.ref_input, self.ref_envdata)
106​
107​
108​ def test_parse_result_from_comment(self):
109​ """Checks parsing of the bodhi comments"""
110​ output = bodhi_comment_directive._parse_result_from_comment(self.comment)
111​
112​ assert output['time'] == datetime.datetime(2014, 1, 31, 10, 11, 12)
113​ assert output['testname'] == 'depcheck'
114​ assert output['result'] == 'PASSED'
115​ assert output['arch'] == 'x86_64'
116​
117​ def test_already_commented_nonexistent_update(self):
118​ """Checks whether non-existing updates throws
119​ an error in _already_commented."""
120​ stub_bodhi = Dingus(query_update__returns = None)
121​
122​ with pytest.raises(TaskotronValueError):
123​ bodhi_comment_directive._already_commented(stub_bodhi, '', '', '')
124​
125​
126​ def test_already_commented(self):
127​ """Tests general _already_commented behaviour."""
128​
129​ # Test comment found
130​ outcome = bodhi_comment_directive._already_commented(None, self.update, 'depcheck', 'x86_64')
131​ assert outcome == ('PASSED', self.comment['timestamp'])
132​
133​ # Test no comment
134​ update = deepcopy(self.update)
135​ update['comments'] = []
136​ outcome = bodhi_comment_directive._already_commented(None, update, 'depcheck', 'x86_64')
137​ assert outcome == ('', '')
138​
139​ # Test modified after commented
140​ update = deepcopy(self.update)
141​ update['date_modified'] = "2014-01-31 15:00:00"
142​ outcome = bodhi_comment_directive._already_commented(None, update, 'depcheck', 'x86_64')
143​ assert outcome == ('', '')
144​
145​ def test_is_comment_needed(self):
146​ """Checks whether _is_comment_needed returns proper results in
147​ various situations."""
148​ # Test first comment
149​ outcome = bodhi_comment_directive._is_comment_needed(None, None, None, None)
150​ assert outcome == True
151​
152​ # Test different result
153​ outcome = bodhi_comment_directive._is_comment_needed('PASSED', None, 'FAILED', None)
154​ assert outcome == True
155​
156​ # Test same result, but not FAILED
157​ outcome = bodhi_comment_directive._is_comment_needed('PASSED', None, 'PASSED', None)
158​ assert outcome == False
159​
160​ def test_is_comment_needed_decide_repost(self, monkeypatch):
161​ """Checks whether time_span has an impact on comment posting."""
162​ date_utcnow = datetime.datetime(2014, 01, 10, 12, 05)
163​ date_posted = datetime.datetime(2014, 01, 10, 12, 00)
164​
165​ class FakeDatetime(datetime.datetime):
166​ @classmethod
167​ def utcnow(cls):
168​ return date_utcnow
169​
170​
171​ monkeypatch.setattr(datetime, 'datetime', FakeDatetime)
172​
173​ # Test same result, FAILED, delta < time_span
174​ time_span = 1000
175​ comment_time = date_posted.strftime('%Y-%m-%d %H:%M:%S')
176​ outcome = bodhi_comment_directive._is_comment_needed('FAILED', comment_time, 'FAILED', time_span)
177​ assert outcome == False
178​
179​ # Test same result, FAILED, delta > time_span
180​ time_span = 0
181​ comment_time = date_posted.strftime('%Y-%m-%d %H:%M:%S')
182​ outcome = bodhi_comment_directive._is_comment_needed('FAILED', comment_time, 'FAILED', time_span)
183​ assert outcome == True
184​
185​ def test_post_testresult_params_and_config(self):
186​ """Checks whether missing parameters/config values terminate
187​ _post_testresults."""
188​ # Test empty parameters
189​ outcome = bodhi_comment_directive._post_testresult(Dingus(),
190​ update = None,
191​ testname = None,
192​ result = None,
193​ url = None)
194​ assert outcome == False
195​
196​ # Test configuration
197​ Config = config.get_config()
198​ Config.fas_username = None
199​ Config.fas_password = None
200​ outcome = bodhi_comment_directive._post_testresult(Dingus(),
201​ update = True,
202​ testname = True,
203​ result = True,
204​ url = True)
205​ config._config = None
206​ assert outcome == False
207​
208​ @pytest.fixture
209​ def prepare_for_bodhi_comment(self, monkeypatch):
210​ """Sets up environment for fake bodhi comment calls"""
211​ Config = config.get_config()
212​ Config.fas_password = "secret!"
213​
214​ stub_already_commented = lambda *args, **kwargs: ('', '')
215​ stub_return_true = lambda *args, **kwargs: True
216​ stub_return_false = lambda *args, **kwargs: False
217​
218​ monkeypatch.setattr(bodhi_comment_directive, '_already_commented', stub_already_commented)
219​ monkeypatch.setattr(bodhi_comment_directive, '_is_comment_needed', stub_return_true)
220​ monkeypatch.setattr(bodhi_comment_directive, '_is_comment_email_needed', stub_return_false)
221​
222​
223​ def test_post_testresult_comment_not_needed(self, monkeypatch, prepare_for_bodhi_comment):
224​ """Checks the behaviour when comment is not needed."""
225​ self.stub_bodhi.client = Dingus('bodhi_api.bodhi')
226​
227​ monkeypatch.setattr(bodhi_comment_directive, '_is_comment_needed', lambda *a, **k: False)
228​
229​ outcome = bodhi_comment_directive._post_testresult(self.stub_bodhi,
230​ update = True,
231​ testname = True,
232​ result = True,
233​ url = True,
234​ doreport = 'onchange')
235​ assert outcome == True
236​ assert len(self.stub_bodhi.client.calls()) == 0
237​
238​ def test_post_testresult(self, prepare_for_bodhi_comment):
239​ """Checks whether _post_testresult maps arguments correctly to
240​ the bodhi comment call."""
241​ self.stub_bodhi.client = Dingus('bodhi_api.bodhi')
242​
243​ outcome = bodhi_comment_directive._post_testresult(self.stub_bodhi,
244​ update = "update_title",
245​ testname = "depcheck",
246​ result = "PASSED",
247​ url = "http://example.com")
248​
249​ # Select the first call of "comment" method.
250​ call = [call for call in self.stub_bodhi.client.calls() if call[0] == 'comment'][0]
251​ # Select the positional arguments of that call
252​ call_data = call[1]
253​
254​ assert call_data[0] == "update_title"
255​ assert call_data[1] == "Taskotron: depcheck test PASSED on noarch. Result log: http://example.com (results are informative only)"
256​ assert call_data[2] == 0
257​ assert call_data[3] == False
258​
259​ def test_tap_mapping(self, prepare_for_bodhi_comment):
260​ """Checks whether TAP output is mapped correctly to
261​ the bodhi comment call."""
262​ self.stub_bodhi.client = Dingus('bodhi_api.bodhi')
263​
264​ self.directive.process(self.ref_input, self.ref_envdata)
265​
266​ # Select the first call of "comment" method.
267​ call = [call for call in self.stub_bodhi.client.calls() if call[0] == 'comment'][0]
268​ # Select the positional arguments of that call
269​ call_data = call[1]
270​
271​ assert call_data[0] == "update_title"
272​ assert call_data[1] == "Taskotron: depcheck test PASSED on noarch. Result log: http://example.com (results are informative only)"
273​ assert call_data[2] == 0
274​ assert call_data[3] == False
275​
276​ def test_is_comment_email_needed(self, monkeypatch):
277​ def mock_BodhiUpdateState_factory(current_state, new_state):
278​
279​ class MockBodhiUpdateState(Dingus):
280​ def __init__(self, *args, **kwargs):
281​ Dingus.__init__(self, *args, **kwargs)
282​ self.state = current_state
283​
284​ def add_result(self, *args, **kwargs):
285​ self.state = new_state
286​
287​ def get_state(self):
288​ return self.state
289​
290​ return MockBodhiUpdateState
291​
292​ PASS = bodhi_comment_directive.PASS
293​ FAIL = bodhi_comment_directive.FAIL
294​ INCOMPLETE = bodhi_comment_directive.INCOMPLETE
295​ UNKNOWN = bodhi_comment_directive.UNKNOWN
296​
297​
298​ # if new_state is PASS and the current_state was not 'FAIL', do not send email
299​ for current_state in (PASS, INCOMPLETE, UNKNOWN):
300​ mock_bus = mock_BodhiUpdateState_factory(current_state, PASS)
301​ monkeypatch.setattr(bodhi_comment_directive, 'BodhiUpdateState', mock_bus)
302​
303​ outcome = bodhi_comment_directive._is_comment_email_needed(None, [], None, None)
304​ assert outcome == False
305​
306​ # If current_state is FAIL, send email
307​ for new_state in (PASS, FAIL, INCOMPLETE, UNKNOWN):
308​ mock_bus = mock_BodhiUpdateState_factory(FAIL, new_state)
309​ monkeypatch.setattr(bodhi_comment_directive, 'BodhiUpdateState', mock_bus)
310​
311​ outcome = bodhi_comment_directive._is_comment_email_needed(None, [], None, None)
312​ assert outcome == True
313​
314​ # If new_state is not PASS and current_state is not FAIL, send email
315​ new_states = (FAIL, INCOMPLETE, UNKNOWN)
316​ current_states = (PASS, INCOMPLETE, UNKNOWN)
317​
318​ for current_state, new_state in itertools.product(current_states, new_states):
319​ mock_bus = mock_BodhiUpdateState_factory(current_state, new_state)
320​ monkeypatch.setattr(bodhi_comment_directive, 'BodhiUpdateState', mock_bus)
321​
322​ outcome = bodhi_comment_directive._is_comment_email_needed(None, [], None, None)
323​ assert outcome == True
324​
325​