Mercurial > repos > shellac > guppy_basecaller
comparison env/lib/python3.7/site-packages/cwltool/tests/util.py @ 0:26e78fe6e8c4 draft
"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 from __future__ import absolute_import | |
2 | |
3 import distutils.spawn # pylint: disable=no-name-in-module,import-error | |
4 import functools | |
5 import os | |
6 import sys | |
7 import shutil | |
8 import contextlib | |
9 import tempfile | |
10 import shutil | |
11 from typing import Text | |
12 | |
13 from pkg_resources import (Requirement, ResolutionError, # type: ignore | |
14 resource_filename) | |
15 import pytest | |
16 | |
17 from cwltool.context import LoadingContext, RuntimeContext | |
18 from cwltool.factory import Factory | |
19 from cwltool.resolver import Path | |
20 from cwltool.utils import onWindows, subprocess, windows_default_container_id | |
21 from cwltool.singularity import is_version_2_6, is_version_3_or_newer | |
22 | |
23 | |
24 def get_windows_safe_factory(runtime_context=None, # type: RuntimeContext | |
25 loading_context=None, # type: LoadingContext | |
26 executor=None # type: Any | |
27 ): # type: (...) -> Factory | |
28 if onWindows(): | |
29 if not runtime_context: | |
30 runtime_context = RuntimeContext() | |
31 runtime_context.find_default_container = functools.partial( | |
32 force_default_container, windows_default_container_id) | |
33 runtime_context.use_container = True | |
34 runtime_context.default_container = windows_default_container_id | |
35 return Factory(executor, loading_context, runtime_context) | |
36 | |
37 def force_default_container(default_container_id, _): | |
38 return default_container_id | |
39 | |
40 def get_data(filename): | |
41 # normalizing path depending on OS or else it will cause problem when joining path | |
42 filename = os.path.normpath(filename) | |
43 filepath = None | |
44 try: | |
45 filepath = resource_filename( | |
46 Requirement.parse("cwltool"), filename) | |
47 except ResolutionError: | |
48 pass | |
49 if not filepath or not os.path.isfile(filepath): | |
50 filepath = os.path.join(os.path.dirname(__file__), os.pardir, filename) | |
51 return Text(Path(filepath).resolve()) | |
52 | |
53 | |
54 needs_docker = pytest.mark.skipif(not bool(distutils.spawn.find_executable('docker')), | |
55 reason="Requires the docker executable on the " | |
56 "system path.") | |
57 | |
58 needs_singularity = pytest.mark.skipif( | |
59 not bool(distutils.spawn.find_executable('singularity')), | |
60 reason="Requires the singularity executable on the system path.") | |
61 | |
62 needs_singularity_2_6 = pytest.mark.skipif( | |
63 not (distutils.spawn.find_executable('singularity') and is_version_2_6()), | |
64 reason="Requires that version 2.6.x of singularity executable version is on the system path.") | |
65 | |
66 needs_singularity_3_or_newer = pytest.mark.skipif( | |
67 not (distutils.spawn.find_executable('singularity') and is_version_3_or_newer), | |
68 reason="Requires that version 2.6.x of singularity executable version is on the system path.") | |
69 | |
70 | |
71 windows_needs_docker = pytest.mark.skipif( | |
72 onWindows() and not bool(distutils.spawn.find_executable('docker')), | |
73 reason="Running this test on MS Windows requires the docker executable " | |
74 "on the system path.") | |
75 | |
76 def get_main_output(args, env=None): | |
77 process = subprocess.Popen( | |
78 [sys.executable, "-m", "cwltool"] + args, | |
79 stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) | |
80 | |
81 stdout, stderr = process.communicate() | |
82 return process.returncode, stdout.decode(), stderr.decode() | |
83 | |
84 @contextlib.contextmanager | |
85 def temp_dir(suffix=""): | |
86 c_dir = tempfile.mkdtemp(suffix, dir=os.curdir) | |
87 try: | |
88 yield c_dir | |
89 finally: | |
90 shutil.rmtree(c_dir, ignore_errors=True) | |
91 | |
92 @contextlib.contextmanager | |
93 def working_directory(path): | |
94 """Change working directory and returns to previous on exit.""" | |
95 prev_cwd = Path.cwd() | |
96 # before python 3.6 chdir doesn't support paths from pathlib | |
97 os.chdir(str(path)) | |
98 try: | |
99 yield | |
100 finally: | |
101 os.chdir(str(prev_cwd)) |