Mercurial > repos > shellac > guppy_basecaller
diff env/lib/python3.7/site-packages/cwltool/tests/test_toolargparse.py @ 2:6af9afd405e9 draft
"planemo upload commit 0a63dd5f4d38a1f6944587f52a8cd79874177fc1"
author | shellac |
---|---|
date | Thu, 14 May 2020 14:56:58 -0400 |
parents | 26e78fe6e8c4 |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/env/lib/python3.7/site-packages/cwltool/tests/test_toolargparse.py Thu May 14 14:56:58 2020 -0400 @@ -0,0 +1,138 @@ +import os +import sys +from tempfile import NamedTemporaryFile +from io import BytesIO, StringIO + +import pytest +from cwltool.main import main +import cwltool.executors + +from .util import get_data, needs_docker + +script_a = ''' +#!/usr/bin/env cwl-runner +cwlVersion: v1.0 +class: CommandLineTool +doc: "This tool is developed for SMC-RNA Challenge for detecting gene fusions (STAR fusion)" +inputs: + #Give it a list of input files + - id: input + type: File + inputBinding: + position: 0 +outputs: + - id: output + type: File + outputBinding: + glob: test.txt +stdout: test.txt +baseCommand: [cat] +''' + +script_b = ''' +#!/usr/bin/env cwl-runner +cwlVersion: v1.0 +class: CommandLineTool +inputs: + - id: bdg + type: "boolean" +outputs: + - id: output + type: File + outputBinding: + glob: foo +baseCommand: + - echo + - "ff" +stdout: foo +''' + +script_c = ''' +#!/usr/bin/env cwl-runner + +cwlVersion: v1.0 +class: ExpressionTool + +inputs: + foo: + type: + type: record + fields: + one: File + two: string + +expression: $(inputs.foo.two) + +outputs: [] +''' + +scripts_argparse_params = [ + ('help', script_a, + lambda x: ["--debug", x, '--input', get_data('tests/echo.cwl')] + ), + ('boolean', script_b, lambda x: [x, '--help'] + ), + ('help with c', script_c, lambda x: [x, '--help']), + ('foo with c', script_c, + lambda x: [x, '--foo.one', get_data('tests/echo.cwl'), '--foo.two', 'test'] + ) +] + +@needs_docker +@pytest.mark.parametrize('name,script_contents,params', scripts_argparse_params) +def test_argparse(name, script_contents, params, tmpdir): + script = None + try: + script = NamedTemporaryFile(mode='w', delete=False) + script.write(script_contents) + script.close() + + my_params = ["--outdir", str(tmpdir)] + my_params.extend(params(script.name)) + assert main(my_params) == 0, name + + except SystemExit as err: + assert err.code == 0, name + finally: + if script and script.name and os.path.exists(script.name): + os.unlink(script.name) + + +class NoopJobExecutor(cwltool.executors.JobExecutor): + def run_jobs(self, + process, # type: Process + job_order_object, # type: Dict[Text, Any] + logger, # type: logging.Logger + runtime_context # type: RuntimeContext + ): # type: (...) -> None + pass + + def execute(self, + process, # type: Process + job_order_object, # type: Dict[Text, Any] + runtime_context, # type: RuntimeContext + logger=None, # type: logging.Logger + ): # type: (...) -> Tuple[Optional[Union[Dict[Text, Any], List[Dict[Text, Any]]]], Text] + return {}, "success" + +def test_dont_require_inputs(): + if sys.version_info[0] < 3: + stream = BytesIO() + else: + stream = StringIO() + + script = None + try: + script = NamedTemporaryFile(mode='w', delete=False) + script.write(script_a) + script.close() + + assert main(argsl=["--debug", script.name, "--input", script.name], executor=NoopJobExecutor(), stdout=stream) == 0 + assert main(argsl=["--debug", script.name], executor=NoopJobExecutor(), stdout=stream) == 2 + assert main(argsl=["--debug", script.name], executor=NoopJobExecutor(), input_required=False, stdout=stream) == 0 + + except SystemExit as err: + assert err.code == 0, name + finally: + if script and script.name and os.path.exists(script.name): + os.unlink(script.name)