-
Notifications
You must be signed in to change notification settings - Fork 651
/
test_util.py
180 lines (146 loc) · 4.57 KB
/
test_util.py
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
import errno
import logging
import os
import subprocess
import time
from xml.etree import ElementTree
import six
from py import util
class TestCase(object):
def __init__(self, class_name="", name=""):
self.class_name = class_name
self.name = name
# Time in seconds of the test.
self.time = None
# String describing the failure.
self.failure = None
class TestSuite(object):
"""A suite of test cases."""
def __init__(self, class_name):
self._cases = {}
self._class_name = class_name
def create(self, name):
"""Create a new TestCase with the specified name.
Args:
name: Name for the newly created TestCase.
Returns:
TestCase: The newly created test case.
Raises:
ValueError: If a test case with the specified name already exists.
"""
if name in self._cases:
raise ValueError("TestSuite already has a test named %s" % name)
self._cases[name] = TestCase()
self._cases[name].class_name = self._class_name
self._cases[name].name = name
return self._cases[name]
def get(self, name):
"""Get the specified test case.
Args:
name: Name of the test case to return.
Returns:
TestCase: The requested test case.
Raises:
KeyError: If no test with that name exists.
"""
if not name in self._cases:
raise KeyError("No TestCase named %s" % name)
return self._cases[name]
def __iter__(self):
"""Return an iterator of TestCases."""
return six.itervalues(self._cases)
def wrap_test(test_func, test_case):
"""Wrap a test func.
Test_func is a callable that contains the commands to perform a particular
test.
Args:
test_func: The callable to invoke.
test_case: A TestCase to be populated.
Raises:
Exceptions are reraised to indicate test failure.
"""
start = time.time()
try:
test_func()
except subprocess.CalledProcessError as e:
test_case.failure = (
"Subprocess failed;\n{0}".format(e.output))
raise
except Exception as e:
test_case.failure = "Test failed; " + e.message
raise
finally:
test_case.time = time.time() - start
def create_xml(test_cases):
"""Create an Element tree representing the test cases.
Args:
test_cases: TestSuite or List of test case objects.
Returns:
ElementTree: representing the elements.
"""
total_time = 0
failures = 0
for c in test_cases:
if c.time:
total_time += c.time
if c.failure:
failures += 1
attrib = {"failures": "{0}".format(failures), "tests": "{0}".format(len(test_cases)),
"time": "{0}".format(total_time)}
root = ElementTree.Element("testsuite", attrib)
for c in test_cases:
attrib = {
"classname": c.class_name,
"name": c.name,
}
if c.time:
attrib["time"] = "{0}".format(c.time)
# If the time isn't set and no message is set we interpret that as
# the test not being run.
if not c.time and not c.failure:
c.failure = "Test was not run."
e = ElementTree.Element("testcase", attrib)
root.append(e)
if c.failure:
f = ElementTree.Element("failure")
f.text = c.failure
e.append(f)
t = ElementTree.ElementTree(root)
return t
def create_junit_xml_file(test_cases, output_path, gcs_client=None):
"""Create a JUnit XML file.
The junit schema is specified here:
https://www.ibm.com/support/knowledgecenter/en/SSQ2R2_9.5.0/com.ibm.rsar.analysis.codereview.cobol.doc/topics/cac_useresults_junit.html
Args:
test_cases: TestSuite or List of test case objects.
output_path: Path to write the XML
gcs_client: GCS client to use if output is GCS.
"""
t = create_xml(test_cases)
logging.info("Creating %s", output_path)
if output_path.startswith("gs://"):
b = six.StringIO()
t.write(b)
bucket_name, path = util.split_gcs_uri(output_path)
bucket = gcs_client.get_bucket(bucket_name)
blob = bucket.blob(path)
blob.upload_from_string(b.getvalue())
else:
dir_name = os.path.dirname(output_path)
if not os.path.exists(dir_name):
logging.info("Creating directory %s", dir_name)
try:
os.makedirs(dir_name)
except OSError as e:
if e.errno == errno.EEXIST:
# The path already exists. This is probably a race condition
# with some other test creating the directory.
# We should just be able to continue
pass
else:
raise
t.write(output_path)
def get_num_failures(xml_string):
"""Return the number of failures based on the XML string."""
e = ElementTree.fromstring(xml_string)
return int(e.attrib.get("failures", 0))