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
|
import csv
import subprocess
import string
import os
import StringIO
import re
import datetime
class SubunitResults(object):
def __init__(self, subunit_stream, all_tests_file):
self.subunit_stream = subunit_stream
self.all_tests_file = all_tests_file
# clean up a test id
def clean_test_id(self, test_id):
if '[' in test_id:
test_id = test_id[:test_id.index('[')]
return re.sub('[^-0-9A-Za-z_.]', '-', test_id)
# extract a "class name" from a test id
def get_class_name(self, test_id, class_depth):
clean_name = self.clean_test_id(test_id)
words = clean_name.split(".")
return ".".join(words[1:class_depth])
# convert subunit2csv data to json
def csv_to_json(self, input_csv):
f = StringIO.StringIO(input_csv)
reader = csv.DictReader(f)
result = []
for row in reader:
test_id = row["test"]
# clean up the test name (get rid of the [uuid]) part
row["test"] = self.clean_test_id(test_id)
# add an entry for the class
row["class"] = self.get_class_name(test_id, 3)
result.append(row)
return result
def get_tests_run(self):
p1 = subprocess.Popen(["cat", self.subunit_stream], stdout=subprocess.PIPE)
p2 = subprocess.Popen(["subunit2csv", "--no-passthrough"], stdin=p1.stdout, stdout=subprocess.PIPE)
p1.stdout.close()
output = p2.communicate()[0]
return self.csv_to_json(output)
def get_failing_tests(self):
p1 = subprocess.Popen(["cat", self.subunit_stream], stdout = subprocess.PIPE)
p2 = subprocess.Popen(["subunit-filter", "--only-genuine-failures", "--no-passthrough"], stdin=p1.stdout, stdout=subprocess.PIPE)
p3 = subprocess.Popen(["subunit2csv", "--no-passthrough"], stdin=p2.stdout, stdout=subprocess.PIPE)
p1.stdout.close()
p2.stdout.close()
output = p3.communicate()[0]
return self.csv_to_json(output)
def get_failing_tests_xml(self):
p1 = subprocess.Popen(["cat", self.subunit_stream], stdout = subprocess.PIPE)
p2 = subprocess.Popen(["subunit-filter", "--only-genuine-failures", "--passthrough"], stdin=p1.stdout, stdout=subprocess.PIPE)
p3 = subprocess.Popen(["subunit2junitxml"], stdin=p2.stdout, stdout=subprocess.PIPE)
p1.stdout.close()
p2.stdout.close()
output = p3.communicate()[0]
return output
def get_passing_tests(self):
p1 = subprocess.Popen(["cat", self.subunit_stream], stdout = subprocess.PIPE)
p2 = subprocess.Popen(["subunit-filter", "--no-skip", "--no-failure", "--success", "--no-passthrough"], stdin=p1.stdout, stdout=subprocess.PIPE)
p3 = subprocess.Popen(["subunit2csv", "--no-passthrough"], stdin=p2.stdout, stdout=subprocess.PIPE)
p1.stdout.close()
p2.stdout.close()
output = p3.communicate()[0]
return self.csv_to_json(output)
def get_skipped_tests(self):
p1 = subprocess.Popen(["cat", self.subunit_stream], stdout = subprocess.PIPE)
p2 = subprocess.Popen(["subunit-filter", "--no-error", "--no-failure", "--no-success", "--no-xfail", "--no-passthrough"], stdin=p1.stdout, stdout=subprocess.PIPE)
p3 = subprocess.Popen(["subunit2csv", "--no-passthrough"], stdin=p2.stdout, stdout=subprocess.PIPE)
p1.stdout.close()
p2.stdout.close()
output = p3.communicate()[0]
return self.csv_to_json(output)
def get_all_tests(self):
with open(self.all_tests_file) as f:
return [self.clean_test_id(line.rstrip('\n')) for line in f]
|
