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)) |
