Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/cwltool/tests/test_toolargparse.py @ 0:26e78fe6e8c4 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author shellac
date Sat, 02 May 2020 07:14:21 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:26e78fe6e8c4
1 import os
2 import sys
3 from tempfile import NamedTemporaryFile
4 from io import BytesIO, StringIO
5
6 import pytest
7 from cwltool.main import main
8 import cwltool.executors
9
10 from .util import get_data, needs_docker
11
12 script_a = '''
13 #!/usr/bin/env cwl-runner
14 cwlVersion: v1.0
15 class: CommandLineTool
16 doc: "This tool is developed for SMC-RNA Challenge for detecting gene fusions (STAR fusion)"
17 inputs:
18 #Give it a list of input files
19 - id: input
20 type: File
21 inputBinding:
22 position: 0
23 outputs:
24 - id: output
25 type: File
26 outputBinding:
27 glob: test.txt
28 stdout: test.txt
29 baseCommand: [cat]
30 '''
31
32 script_b = '''
33 #!/usr/bin/env cwl-runner
34 cwlVersion: v1.0
35 class: CommandLineTool
36 inputs:
37 - id: bdg
38 type: "boolean"
39 outputs:
40 - id: output
41 type: File
42 outputBinding:
43 glob: foo
44 baseCommand:
45 - echo
46 - "ff"
47 stdout: foo
48 '''
49
50 script_c = '''
51 #!/usr/bin/env cwl-runner
52
53 cwlVersion: v1.0
54 class: ExpressionTool
55
56 inputs:
57 foo:
58 type:
59 type: record
60 fields:
61 one: File
62 two: string
63
64 expression: $(inputs.foo.two)
65
66 outputs: []
67 '''
68
69 scripts_argparse_params = [
70 ('help', script_a,
71 lambda x: ["--debug", x, '--input', get_data('tests/echo.cwl')]
72 ),
73 ('boolean', script_b, lambda x: [x, '--help']
74 ),
75 ('help with c', script_c, lambda x: [x, '--help']),
76 ('foo with c', script_c,
77 lambda x: [x, '--foo.one', get_data('tests/echo.cwl'), '--foo.two', 'test']
78 )
79 ]
80
81 @needs_docker
82 @pytest.mark.parametrize('name,script_contents,params', scripts_argparse_params)
83 def test_argparse(name, script_contents, params, tmpdir):
84 script = None
85 try:
86 script = NamedTemporaryFile(mode='w', delete=False)
87 script.write(script_contents)
88 script.close()
89
90 my_params = ["--outdir", str(tmpdir)]
91 my_params.extend(params(script.name))
92 assert main(my_params) == 0, name
93
94 except SystemExit as err:
95 assert err.code == 0, name
96 finally:
97 if script and script.name and os.path.exists(script.name):
98 os.unlink(script.name)
99
100
101 class NoopJobExecutor(cwltool.executors.JobExecutor):
102 def run_jobs(self,
103 process, # type: Process
104 job_order_object, # type: Dict[Text, Any]
105 logger, # type: logging.Logger
106 runtime_context # type: RuntimeContext
107 ): # type: (...) -> None
108 pass
109
110 def execute(self,
111 process, # type: Process
112 job_order_object, # type: Dict[Text, Any]
113 runtime_context, # type: RuntimeContext
114 logger=None, # type: logging.Logger
115 ): # type: (...) -> Tuple[Optional[Union[Dict[Text, Any], List[Dict[Text, Any]]]], Text]
116 return {}, "success"
117
118 def test_dont_require_inputs():
119 if sys.version_info[0] < 3:
120 stream = BytesIO()
121 else:
122 stream = StringIO()
123
124 script = None
125 try:
126 script = NamedTemporaryFile(mode='w', delete=False)
127 script.write(script_a)
128 script.close()
129
130 assert main(argsl=["--debug", script.name, "--input", script.name], executor=NoopJobExecutor(), stdout=stream) == 0
131 assert main(argsl=["--debug", script.name], executor=NoopJobExecutor(), stdout=stream) == 2
132 assert main(argsl=["--debug", script.name], executor=NoopJobExecutor(), input_required=False, stdout=stream) == 0
133
134 except SystemExit as err:
135 assert err.code == 0, name
136 finally:
137 if script and script.name and os.path.exists(script.name):
138 os.unlink(script.name)