Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/cwltool/tests/test_ext.py @ 2:6af9afd405e9 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 0a63dd5f4d38a1f6944587f52a8cd79874177fc1"
author shellac
date Thu, 14 May 2020 14:56:58 -0400
parents 26e78fe6e8c4
children
comparison
equal deleted inserted replaced
1:75ca89e9b81c 2:6af9afd405e9
1 from __future__ import absolute_import
2
3 import os
4 import shutil
5 import tempfile
6 import sys
7 import re
8 from io import StringIO
9
10 import pytest
11
12 from cwltool.main import main
13 import cwltool.process
14
15
16 from .util import get_data, needs_docker, temp_dir, windows_needs_docker
17
18
19 @needs_docker
20 def test_missing_enable_ext():
21 # Require that --enable-ext is provided.
22 assert main([get_data('tests/wf/listing_deep.cwl'), get_data('tests/listing-job.yml')]) != 0
23
24 @needs_docker
25 def test_listing_deep():
26 params = ["--enable-ext", get_data('tests/wf/listing_deep.cwl'),
27 get_data('tests/listing-job.yml')]
28 assert main(params) == 0
29
30 @needs_docker
31 def test_cwltool_options():
32 try:
33 opt = os.environ.get("CWLTOOL_OPTIONS")
34 os.environ["CWLTOOL_OPTIONS"] = "--enable-ext"
35 params = [get_data('tests/wf/listing_deep.cwl'),
36 get_data('tests/listing-job.yml')]
37 assert main(params) == 0
38 finally:
39 if opt is not None:
40 os.environ["CWLTOOL_OPTIONS"] = opt
41 else:
42 del os.environ["CWLTOOL_OPTIONS"]
43
44 @needs_docker
45 def test_listing_shallow():
46 # This fails on purpose, because it tries to access listing in a subdirectory
47 # the same way that listing_deep does, but it shouldn't be expanded.
48 params = ["--enable-ext", get_data('tests/wf/listing_shallow.cwl'),
49 get_data('tests/listing-job.yml')]
50 assert main(params) != 0
51
52 @needs_docker
53 def test_listing_none():
54 # This fails on purpose, because it tries to access listing but it shouldn't be there.
55 params = ["--enable-ext", get_data('tests/wf/listing_none.cwl'),
56 get_data('tests/listing-job.yml')]
57 assert main(params) != 0
58
59 @needs_docker
60 def test_listing_v1_0():
61 # Default behavior in 1.0 is deep expansion.
62 assert main([get_data('tests/wf/listing_v1_0.cwl'), get_data('tests/listing-job.yml')]) == 0
63
64 @pytest.mark.skip(reason="This is not the default behaviour yet")
65 @needs_docker
66 def test_listing_v1_1():
67 # Default behavior in 1.1 will be no expansion
68 assert main([get_data('tests/wf/listing_v1_1.cwl'), get_data('tests/listing-job.yml')]) != 0
69
70 @needs_docker
71 def test_double_overwrite(tmpdir):
72 with temp_dir() as tmp:
73 tmp_name = os.path.join(tmp, "value")
74
75 before_value, expected_value = "1", "3"
76
77 with open(tmp_name, "w") as f:
78 f.write(before_value)
79
80 assert main(["--enable-ext", "--outdir", str(tmpdir),
81 get_data('tests/wf/mut2.cwl'), "-a", tmp_name]) == 0
82
83 with open(tmp_name, "r") as f:
84 actual_value = f.read()
85
86 assert actual_value == expected_value
87
88 @needs_docker
89 def test_disable_file_overwrite_without_ext():
90 with temp_dir() as tmp:
91 with temp_dir() as out:
92 tmp_name = os.path.join(tmp, "value")
93 out_name = os.path.join(out, "value")
94
95 before_value, expected_value = "1", "2"
96
97 with open(tmp_name, "w") as f:
98 f.write(before_value)
99
100 assert main(["--outdir", out, get_data('tests/wf/updateval.cwl'), "-r", tmp_name]) == 0
101
102 with open(tmp_name, "r") as f:
103 tmp_value = f.read()
104 with open(out_name, "r") as f:
105 out_value = f.read()
106
107 assert tmp_value == before_value
108 assert out_value == expected_value
109
110 @needs_docker
111 def test_disable_dir_overwrite_without_ext():
112 with temp_dir() as tmp:
113 with temp_dir() as out:
114
115 assert main(["--outdir", out, get_data('tests/wf/updatedir.cwl'), "-r", tmp]) == 0
116
117 assert not os.listdir(tmp)
118 assert os.listdir(out)
119
120 @needs_docker
121 def test_disable_file_creation_in_outdir_with_ext():
122 with temp_dir() as tmp:
123 with temp_dir() as out:
124
125 tmp_name = os.path.join(tmp, "value")
126 out_name = os.path.join(out, "value")
127
128 before_value, expected_value = "1", "2"
129
130 with open(tmp_name, "w") as f:
131 f.write(before_value)
132
133 params = ["--enable-ext", "--leave-outputs", "--outdir",
134 out, get_data('tests/wf/updateval_inplace.cwl'), "-r", tmp_name]
135 assert main(params) == 0
136
137 with open(tmp_name, "r") as f:
138 tmp_value = f.read()
139
140 assert tmp_value == expected_value
141 assert not os.path.exists(out_name)
142
143 @needs_docker
144 def test_disable_dir_creation_in_outdir_with_ext():
145 with temp_dir() as tmp:
146 with temp_dir() as out:
147 params = ["--enable-ext", "--leave-outputs", "--outdir",
148 out, get_data('tests/wf/updatedir_inplace.cwl'), "-r", tmp]
149 assert main(params) == 0
150
151 assert os.listdir(tmp)
152 assert not os.listdir(out)
153
154 @needs_docker
155 def test_write_write_conflict():
156 with temp_dir('tmp') as tmp:
157 tmp_name = os.path.join(tmp, "value")
158
159 before_value, expected_value = "1", "2"
160
161 with open(tmp_name, "w") as f:
162 f.write(before_value)
163
164 assert main(["--enable-ext", get_data('tests/wf/mut.cwl'), "-a", tmp_name]) != 0
165
166 with open(tmp_name, "r") as f:
167 tmp_value = f.read()
168
169 assert tmp_value == expected_value
170
171 @pytest.mark.skip(reason="This test is non-deterministic")
172 def test_read_write_conflict():
173 with temp_dir('tmp') as tmp:
174 tmp_name = os.path.join(tmp, "value")
175
176 with open(tmp_name, "w") as f:
177 f.write("1")
178
179 assert main(["--enable-ext", get_data('tests/wf/mut3.cwl'), "-a", tmp_name]) != 0
180
181 @needs_docker
182 def test_require_prefix_networkaccess():
183 assert main(["--enable-ext", get_data('tests/wf/networkaccess.cwl')]) == 0
184 assert main([get_data('tests/wf/networkaccess.cwl')]) != 0
185 assert main(["--enable-ext", get_data('tests/wf/networkaccess-fail.cwl')]) != 0
186
187 @needs_docker
188 def test_require_prefix_workreuse(tmpdir):
189 assert main(["--enable-ext", '--outdir', str(tmpdir), get_data('tests/wf/workreuse.cwl')]) == 0
190 assert main([get_data('tests/wf/workreuse.cwl')]) != 0
191 assert main(["--enable-ext", get_data('tests/wf/workreuse-fail.cwl')]) != 0
192
193 @windows_needs_docker
194 def test_require_prefix_timelimit():
195 assert main(["--enable-ext", get_data('tests/wf/timelimit.cwl')]) == 0
196 assert main([get_data('tests/wf/timelimit.cwl')]) != 0
197 assert main(["--enable-ext", get_data('tests/wf/timelimit-fail.cwl')]) != 0
198
199 def test_warn_large_inputs():
200 was = cwltool.process.FILE_COUNT_WARNING
201 try:
202 stream = StringIO()
203
204 cwltool.process.FILE_COUNT_WARNING = 3
205 main([get_data('tests/wf/listing_v1_0.cwl'), get_data('tests/listing2-job.yml')],
206 stderr=stream)
207
208 assert "Recursive directory listing has resulted in a large number of File" in re.sub("\n *", " ", stream.getvalue())
209 finally:
210 cwltool.process.FILE_COUNT_WARNING = was