Mercurial > repos > bcclaywell > argo_navis

comparison venv/lib/python2.7/site-packages/planemo/galaxy_test/actions.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 import os
2
3 import click
4
5 from . import structures as test_structures
6 from planemo.io import info, warn
7 from planemo import galaxy_run
8 from planemo.reports import build_report
9
10
11 from galaxy.tools.deps.commands import shell
12
13 XUNIT_UPGRADE_MESSAGE = ("This version of Galaxy does not support xUnit - "
14 "please update to newest development brach.")
15 NO_XUNIT_MESSAGE = ("Cannot locate xUnit report option for tests - update "
16 "Galaxy for more detailed breakdown.")
17 NO_JSON_MESSAGE = ("Cannot locate json report option for tests - update "
18 "Galaxy for more detailed breakdown.")
19 NO_TESTS_MESSAGE = "No tests were executed - see Galaxy output for details."
20 ALL_TESTS_PASSED_MESSAGE = "All %d test(s) executed passed."
21 PROBLEM_COUNT_MESSAGE = ("There were problems with %d test(s) - out of %d "
22 "test(s) executed. See %s for detailed breakdown.")
23 GENERIC_PROBLEMS_MESSAGE = ("One or more tests failed. See %s for detailed "
24 "breakdown.")
25 GENERIC_TESTS_PASSED_MESSAGE = "No failing tests encountered."
26
27
28 def run_in_config(ctx, config, **kwds):
29 config_directory = config.config_directory
30 html_report_file = kwds["test_output"]
31
32 job_output_files = kwds.get("job_output_files", None)
33 if job_output_files is None:
34 job_output_files = os.path.join(config_directory, "jobfiles")
35
36 xunit_supported, xunit_report_file = __xunit_state(kwds, config)
37 structured_report_file = __structured_report_file(kwds, config)
38
39 info("Testing using galaxy_root %s", config.galaxy_root)
40 # TODO: Allow running dockerized Galaxy here instead.
41 server_ini = os.path.join(config_directory, "galaxy.ini")
42 config.env["GALAXY_CONFIG_FILE"] = server_ini
43 config.env["GALAXY_TEST_VERBOSE_ERRORS"] = "true"
44 config.env["GALAXY_TEST_SAVE"] = job_output_files
45
46 cd_to_galaxy_command = "cd %s" % config.galaxy_root
47 test_cmd = test_structures.GalaxyTestCommand(
48 html_report_file,
49 xunit_report_file,
50 structured_report_file,
51 failed=kwds.get("failed", False),
52 installed=kwds.get("installed", False),
53 ).build()
54 cmd = "; ".join([
55 cd_to_galaxy_command,
56 galaxy_run.ACTIVATE_COMMAND, # TODO: this should be moved to
57 # run_tests.sh to match run.sh.
58 test_cmd,
59 ])
60 action = "Testing tools"
61 return_code = galaxy_run.run_galaxy_command(
62 ctx,
63 cmd,
64 config.env,
65 action
66 )
67 if kwds.get('update_test_data', False):
68 update_cp_args = (job_output_files, config.test_data_dir)
69 shell('cp -r "%s"/* "%s"' % update_cp_args)
70
71 if xunit_report_file and (not os.path.exists(xunit_report_file)):
72 warn(NO_XUNIT_MESSAGE)
73 xunit_report_file = None
74
75 test_results = test_structures.GalaxyTestResults(
76 structured_report_file,
77 xunit_report_file,
78 html_report_file,
79 return_code,
80 )
81
82 try:
83 test_data = test_results.structured_data
84
85 if 'test_output' in kwds:
86 with open(kwds['test_output'], 'w') as handle:
87 handle.write(build_report.build_report(test_data))
88
89 for kw_name in ('markdown', 'text'):
90 if 'test_output_%s' % kw_name in kwds:
91 with open(kwds['test_output_%s' % kw_name], 'w') as handle:
92 handle.write(build_report.build_report(test_data, report_type=kw_name))
93
94 except Exception:
95 pass
96
97 __handle_summary(
98 test_results,
99 **kwds
100 )
101
102 return return_code
103
104
105 def __handle_summary(
106 test_results,
107 **kwds
108 ):
109 summary_style = kwds.get("summary")
110 if summary_style == "none":
111 return
112
113 if test_results.has_details:
114 __summarize_tests_full(
115 test_results,
116 **kwds
117 )
118 else:
119 if test_results.exit_code:
120 warn(GENERIC_PROBLEMS_MESSAGE % test_results.output_html_path)
121 else:
122 info(GENERIC_TESTS_PASSED_MESSAGE)
123
124
125 def __summarize_tests_full(
126 test_results,
127 **kwds
128 ):
129 num_tests = test_results.num_tests
130 num_problems = test_results.num_problems
131
132 if num_tests == 0:
133 warn(NO_TESTS_MESSAGE)
134 return
135
136 if num_problems == 0:
137 info(ALL_TESTS_PASSED_MESSAGE % num_tests)
138
139 if num_problems:
140 html_report_file = test_results.output_html_path
141 message_args = (num_problems, num_tests, html_report_file)
142 message = PROBLEM_COUNT_MESSAGE % message_args
143 warn(message)
144
145 for testcase_el in test_results.xunit_testcase_elements:
146 structured_data_tests = test_results.structured_data_tests
147 __summarize_test_case(structured_data_tests, testcase_el, **kwds)
148
149
150 def __summarize_test_case(structured_data, testcase_el, **kwds):
151 summary_style = kwds.get("summary")
152 test_id = test_structures.case_id(testcase_el)
153 passed = len(list(testcase_el)) == 0
154 if not passed:
155 state = click.style("failed", bold=True, fg='red')
156 else:
157 state = click.style("passed", bold=True, fg='green')
158 click.echo(test_id.label + ": " + state)
159 if summary_style != "minimal":
160 __print_command_line(structured_data, test_id)
161
162
163 def __print_command_line(structured_data, test_id):
164 try:
165 test = [d for d in structured_data if d["id"] == test_id.id][0]["data"]
166 except (KeyError, IndexError):
167 # Failed to find structured data for this test - likely targetting
168 # and older Galaxy version.
169 return
170
171 execution_problem = test.get("execution_problem", None)
172 if execution_problem:
173 click.echo("| command: *could not execute job, no command generated* ")
174 return
175
176 try:
177 command = test["job"]["command_line"]
178 except (KeyError, IndexError):
179 click.echo("| command: *failed to determine command for job* ")
180 return
181
182 click.echo("| command: %s" % command)
183
184
185 def __xunit_state(kwds, config):
186 xunit_supported = True
187 if shell("grep -q xunit '%s'/run_tests.sh" % config.galaxy_root):
188 xunit_supported = False
189
190 xunit_report_file = kwds.get("test_output_xunit", None)
191 if xunit_report_file is None and xunit_supported:
192 xunit_report_file = os.path.join(config.config_directory, "xunit.xml")
193 elif xunit_report_file is not None and not xunit_supported:
194 warn(XUNIT_UPGRADE_MESSAGE)
195 xunit_report_file = None
196
197 return xunit_supported, xunit_report_file
198
199
200 def __structured_report_file(kwds, config):
201 structured_data_supported = True
202 if shell("grep -q structured_data '%s'/run_tests.sh" % config.galaxy_root):
203 structured_data_supported = False
204
205 structured_report_file = None
206 structured_report_file = kwds.get("test_output_json", None)
207 if structured_report_file is None and structured_data_supported:
208 conf_dir = config.config_directory
209 structured_report_file = os.path.join(conf_dir, "structured_data.json")
210 elif structured_report_file is not None and not structured_data_supported:
211 warn(NO_JSON_MESSAGE)
212 structured_report_file = None
213
214 return structured_report_file