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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
import operator
import xml.etree.ElementTree as ET
from junit_utils import add_junit_property
class MutedTestCheck:
def __init__(self, fn=None):
self.classes = set()
self.methods = set()
if fn:
self.populate(fn)
def populate(self, fn):
with open(fn, "r") as fp:
for line in fp:
line = line.strip()
if not line:
continue
if "::" in line:
cls, method = line.split("::", maxsplit=1)
self.methods.add((cls, method))
else:
self.classes.add(line)
def __call__(self, cls, method=None):
if cls in self.classes:
return True
if method and (cls, method) in self.methods:
return True
return False
@property
def has_rules(self):
return len(self.classes) or len(self.methods)
class MutedShardCheck:
def __init__(self, fn=None):
self.muted = set()
if fn:
self.populate(fn)
def populate(self, fn):
with open(fn, "rt") as fp:
for line in fp:
target = line.strip()
if target:
self.muted.add(target)
def __call__(self, target):
return target in self.muted
def mute_target(node):
failure = node.find("failure")
if failure is None:
return False
skipped = ET.Element("skipped", {"message": failure.attrib["message"]})
node.remove(failure)
node.append(skipped)
add_junit_property(node, "mute", "automatically muted based on rules")
return True
def remove_failure(node):
failure = node.find("failure")
if failure is not None:
node.remove(failure)
return True
return False
def op_attr(node, attr, op, value):
v = int(node.get(attr, 0))
node.set(attr, str(op(v, value)))
def inc_attr(node, attr, value):
return op_attr(node, attr, operator.add, value)
def dec_attr(node, attr, value):
return op_attr(node, attr, operator.sub, value)
def update_suite_info(root, n_remove_failures=None, n_skipped=None):
if n_remove_failures:
dec_attr(root, "failures", n_remove_failures)
if n_skipped:
inc_attr(root, "skipped", n_skipped)
def recalc_suite_info(suite):
tests = failures = skipped = 0
elapsed = 0.0
for case in suite.findall("testcase"):
tests += 1
elapsed += float(case.get("time"))
if case.find("skipped"):
skipped += 1
if case.find("failure"):
failures += 1
suite.set("tests", str(tests))
suite.set("failures", str(failures))
suite.set("skipped", str(skipped))
suite.set("time", str(elapsed))
|
