/
test_cmd_test.py
123 lines (112 loc) · 4.44 KB
/
test_cmd_test.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
"""Module contains :class:`CmdTestTestCase` - integration tests for the ``test`` command."""
import json
import os
from .test_utils import (
assert_exists,
CliTestCase,
PROJECT_TEMPLATES_DIR,
skip_if_environ,
skip_unless_python_2_7,
TEST_DATA_DIR,
)
class CmdTestTestCase(CliTestCase):
"""Integration tests for the ``test`` command."""
@skip_if_environ("PLANEMO_SKIP_GALAXY_TESTS")
def test_workflow_test_simple_yaml(self):
"""Test testing a simple YAML workflow with Galaxy."""
with self._isolate():
random_lines = os.path.join(PROJECT_TEMPLATES_DIR, "demo", "randomlines.xml")
cat = os.path.join(PROJECT_TEMPLATES_DIR, "demo", "cat.xml")
test_artifact = os.path.join(TEST_DATA_DIR, "wf1.gxwf.yml")
test_command = [
"--verbose",
"test",
]
test_command = self.append_profile_argument_if_needed(test_command)
test_command += [
"--extra_tools", random_lines,
"--extra_tools", cat,
test_artifact,
]
self._check_exit_code(test_command, exit_code=0)
@skip_if_environ("PLANEMO_SKIP_GALAXY_TESTS")
def test_workflow_test_simple_ga(self):
"""Test testing a simple GA workflow with Galaxy."""
with self._isolate():
cat = os.path.join(PROJECT_TEMPLATES_DIR, "demo", "cat.xml")
test_artifact = os.path.join(TEST_DATA_DIR, "wf2.ga")
test_command = [
"--verbose",
"test"
]
test_command = self.append_profile_argument_if_needed(test_command)
test_command += [
"--extra_tools", cat,
test_artifact,
]
# try:
self._check_exit_code(test_command, exit_code=0)
# except Exception:
# with open(os.path.join(f, "tool_test_output.json"), "r") as o:
# print(o.read())
# raise
@skip_unless_python_2_7()
@skip_if_environ("PLANEMO_SKIP_CWLTOOL_TESTS")
def test_cwltool_tool_test(self):
"""Test testing a CWL tool with cwltool."""
with self._isolate() as f:
test_artifact = os.path.join(TEST_DATA_DIR, "int_tool.cwl")
test_command = [
"test",
"--engine", "cwltool",
test_artifact,
]
self._check_exit_code(test_command, exit_code=0)
assert_exists(os.path.join(f, "tool_test_output.json"))
@skip_unless_python_2_7()
@skip_if_environ("PLANEMO_SKIP_CWLTOOL_TESTS")
def test_output_checks(self):
"""Test testing a CWL tool with cwltool."""
with self._isolate() as f:
test_artifact = os.path.join(TEST_DATA_DIR, "output_tests_tool.cwl")
test_command = [
"test",
"--no-container",
"--engine", "cwltool",
test_artifact,
]
self._check_exit_code(test_command, exit_code=1)
output_json_path = os.path.join(f, "tool_test_output.json")
with open(output_json_path, "r") as f:
output = json.load(f)
assert "tests" in output
tests = output["tests"]
# check out tests/data/output_tests_tool_test.yml
expected_statuses = [
"success",
"failure",
"success",
"success",
"failure",
"success",
"failure",
"success",
"failure",
"success",
"failure",
]
for i in range(len(expected_statuses)):
test_i = tests[i]
data = test_i["data"]
expected_status = expected_statuses[i]
assert data["status"] == expected_status
def append_profile_argument_if_needed(self, command):
# Hook into tests to allow leveraging postgres databases to prevent Galaxy locking errors
# while running tests.
profile_name = os.getenv("PLANEMO_TEST_WORKFLOW_RUN_PROFILE", None)
if not profile_name:
command += ["--profile", profile_name]
database_type = os.getenv("PLANEMO_TEST_WORKFLOW_RUN_PROFILE_DATABASE_TYPE", None)
if database_type:
command += ["--database_type", database_type]
return command