1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
|
#!/usr/bin/env python
# ***** BEGIN LICENSE BLOCK *****
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
# ***** END LICENSE BLOCK *****
"""Code to integration with automation.
"""
try:
import simplejson as json
assert json
except ImportError:
import json
from mozharness.base.log import ERROR, INFO, WARNING
TBPL_SUCCESS = "SUCCESS"
TBPL_WARNING = "WARNING"
TBPL_FAILURE = "FAILURE"
TBPL_EXCEPTION = "EXCEPTION"
TBPL_RETRY = "RETRY"
TBPL_STATUS_DICT = {
TBPL_SUCCESS: INFO,
TBPL_WARNING: WARNING,
TBPL_FAILURE: ERROR,
TBPL_EXCEPTION: ERROR,
TBPL_RETRY: WARNING,
}
EXIT_STATUS_DICT = {
TBPL_SUCCESS: 0,
TBPL_WARNING: 1,
TBPL_FAILURE: 2,
TBPL_EXCEPTION: 3,
TBPL_RETRY: 4,
}
TBPL_WORST_LEVEL_TUPLE = (
TBPL_RETRY,
TBPL_EXCEPTION,
TBPL_FAILURE,
TBPL_WARNING,
TBPL_SUCCESS,
)
class AutomationMixin(object):
worst_status = TBPL_SUCCESS
properties = {}
def tryserver_email(self):
pass
def record_status(self, tbpl_status, level=None, set_return_code=True):
if tbpl_status not in TBPL_STATUS_DICT:
self.error("record_status() doesn't grok the status %s!" % tbpl_status)
else:
if not level:
level = TBPL_STATUS_DICT[tbpl_status]
self.worst_status = self.worst_level(
tbpl_status, self.worst_status, TBPL_WORST_LEVEL_TUPLE
)
if self.worst_status != tbpl_status:
self.info(
"Current worst status %s is worse; keeping it." % self.worst_status
)
if set_return_code:
self.return_code = EXIT_STATUS_DICT[self.worst_status]
def add_failure(self, key, message="%(key)s failed.", level=ERROR):
if key not in self.failures:
self.failures.append(key)
self.add_summary(message % {"key": key}, level=level)
self.record_status(TBPL_FAILURE)
def query_failure(self, key):
return key in self.failures
def query_is_nightly(self):
"""returns whether or not the script should run as a nightly build."""
return bool(self.config.get("nightly_build"))
|
