Mercurial > repos > bcclaywell > argo_navis

comparison venv/lib/python2.7/site-packages/planemo/galaxy_test/structures.py @ 0:d67268158946 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
planemo upload commit a3f181f5f126803c654b3a66dd4e83a48f7e203b
author bcclaywell
date Mon, 12 Oct 2015 17:43:33 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:d67268158946
1 """ Utilities for reasoning about Galaxy test results.
2 """
3 from __future__ import print_function
4 from __future__ import absolute_import
5
6 from collections import namedtuple
7 import json
8 import xml.etree.ElementTree as ET
9
10 RUN_TESTS_CMD = (
11 "sh run_tests.sh --report_file %s %s %s %s"
12 )
13
14
15 class GalaxyTestCommand(object):
16
17 def __init__(
18 self,
19 html_report_file,
20 xunit_report_file,
21 structured_report_file,
22 failed=False,
23 installed=False,
24 ):
25 self.html_report_file = html_report_file
26 self.xunit_report_file = xunit_report_file
27 self.structured_report_file = structured_report_file
28 self.failed = failed
29 self.installed = installed
30
31 def build(self):
32 xunit_report_file = self.xunit_report_file
33 sd_report_file = self.structured_report_file
34 html_report_file = self.html_report_file
35 if xunit_report_file:
36 xunit_arg = "--xunit_report_file %s" % xunit_report_file
37 else:
38 xunit_arg = ""
39 if sd_report_file:
40 sd_arg = "--structured_data_report_file %s" % sd_report_file
41 else:
42 sd_arg = ""
43 if self.installed:
44 tests = "-installed"
45 else:
46 tests = "functional.test_toolbox"
47 if self.failed:
48 sd = StructuredData(self.structured_report_file)
49 failed_ids = sd.failed_ids
50 tests = " ".join(failed_ids)
51 return RUN_TESTS_CMD % (html_report_file, xunit_arg, sd_arg, tests)
52
53
54 class StructuredData(object):
55 """ Abstraction around Galaxy's structured test data output.
56 """
57
58 def __init__(self, json_path):
59 self.json_path = json_path
60 try:
61 with open(json_path, "r") as output_json_f:
62 structured_data = json.load(output_json_f)
63 structured_data_tests = structured_data["tests"]
64 except Exception:
65 print("Warning: Targetting older Galaxy which did not "
66 "produce a structured test results files.")
67 structured_data = {}
68 structured_data_tests = {}
69 self.structured_data = structured_data
70 self.structured_data_tests = structured_data_tests
71 structured_data_by_id = {}
72 for test in self.structured_data_tests:
73 structured_data_by_id[test["id"]] = test["data"]
74 self.structured_data_by_id = structured_data_by_id
75 self.has_details = "summary" in structured_data
76 if self.has_details:
77 self._read_summary()
78
79 def update(self):
80 with open(self.json_path, "w") as out_f:
81 json.dump(self.structured_data, out_f)
82
83 def merge_xunit(self, xunit_root):
84 self.has_details = True
85 xunit_attrib = xunit_root.attrib
86 num_tests = int(xunit_attrib.get("tests", 0))
87 num_failures = int(xunit_attrib.get("failures", 0))
88 num_errors = int(xunit_attrib.get("errors", 0))
89 num_skips = int(xunit_attrib.get("skips", 0))
90 summary = dict(
91 num_tests=num_tests,
92 num_failures=num_failures,
93 num_errors=num_errors,
94 num_skips=num_skips,
95 )
96
97 self.structured_data["summary"] = summary
98 self.num_tests = num_tests
99 self.num_problems = num_skips + num_errors + num_failures
100
101 for testcase_el in xunit_t_elements_from_root(xunit_root):
102 test = case_id(testcase_el)
103 test_data = self.structured_data_by_id.get(test.id)
104 if not test_data:
105 continue
106 problem_el = None
107 for problem_type in ["skip", "failure", "error"]:
108 problem_el = testcase_el.find(problem_type)
109 if problem_el is not None:
110 break
111 if problem_el is not None:
112 status = problem_el.tag
113 test_data["problem_type"] = problem_el.attrib["type"]
114 test_data["problem_log"] = problem_el.text
115 else:
116 status = "success"
117 test_data["status"] = status
118
119 def _read_summary(self):
120 # TODO: read relevant data out of summary object.
121 pass
122
123 @property
124 def failed_ids(self):
125 ids = set([])
126 for test_data in self.structured_data_tests:
127 if test_data["data"]["status"] == "success":
128 continue
129 test_case = test_data["id"].replace(".test_toolbox.", ".test_toolbox:")
130 ids.add(test_case)
131 return ids
132
133
134 class GalaxyTestResults(object):
135 """ Class that combine the test-centric xunit output
136 with the Galaxy centric structured data output - and
137 abstracts away the difference (someday).
138 """
139
140 def __init__(
141 self,
142 output_json_path,
143 output_xml_path,
144 output_html_path,
145 exit_code,
146 ):
147 self.output_html_path = output_html_path
148 self.exit_code = exit_code
149 sd = StructuredData(output_json_path)
150 self.sd = sd
151 self.structured_data = sd.structured_data
152 self.structured_data_tests = sd.structured_data_tests
153 self.structured_data_by_id = sd.structured_data_by_id
154
155 if output_xml_path:
156 self.xunit_tree = parse_xunit_report(output_xml_path)
157 sd.merge_xunit(self._xunit_root)
158 else:
159 self.xunit_tree = ET.fromstring("<testsuite />")
160 self.sd.update()
161
162 @property
163 def has_details(self):
164 return self.sd.has_details
165
166 @property
167 def num_tests(self):
168 return self.sd.num_tests
169
170 @property
171 def num_problems(self):
172 return self.sd.num_problems
173
174 @property
175 def _xunit_root(self):
176 return self.xunit_tree.getroot()
177
178 @property
179 def all_tests_passed(self):
180 return self.sd.num_problems == 0
181
182 @property
183 def xunit_testcase_elements(self):
184 return xunit_t_elements_from_root(self._xunit_root)
185
186
187 def xunit_t_elements_from_root(xunit_root):
188 for testcase_el in find_cases(xunit_root):
189 yield testcase_el
190
191
192 def parse_xunit_report(xunit_report_path):
193 return ET.parse(xunit_report_path)
194
195
196 def find_cases(xunit_root):
197 return xunit_root.findall("testcase")
198
199
200 def case_id(testcase_el):
201 name_raw = testcase_el.attrib["name"]
202 if "TestForTool_" in name_raw:
203 raw_id = name_raw
204 else:
205 class_name = testcase_el.attrib["classname"]
206 raw_id = "{0}.{1}".format(class_name, name_raw)
207
208 name = None
209 num = None
210 if "TestForTool_" in raw_id:
211 tool_and_num = raw_id.split("TestForTool_")[-1]
212 if ".test_tool_" in tool_and_num:
213 name, num_str = tool_and_num.split(".test_tool_", 1)
214 num = _parse_num(num_str)
215 # Tempted to but something human friendly in here like
216 # num + 1 - but then it doesn't match HTML report.
217 else:
218 name = tool_and_num
219 else:
220 name = raw_id
221
222 return TestId(name, num, raw_id)
223
224
225 def _parse_num(num_str):
226 try:
227 num = int(num_str)
228 except ValueError:
229 num = None
230 return num
231
232
233 TestId = namedtuple("TestId", ["name", "num", "id"])
234
235
236 @property
237 def _label(self):
238 if self.num is not None:
239 return "{0}[{1}]".format(self.name, self.num)
240 else:
241 return self.id
242
243
244 TestId.label = _label