Mercurial > repos > shellac > sam_consensus_v3

comparison env/lib/python3.9/site-packages/cwltool/tests/util.py @ 0:4f3585e2f14b draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author shellac
date Mon, 22 Mar 2021 18:12:50 +0000 (2021-03-22)
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:4f3585e2f14b
1 import contextlib
2 import distutils.spawn # pylint: disable=no-name-in-module,import-error
3 import functools
4 import io
5 import os
6 import subprocess
7 import sys
8 from pathlib import Path
9 from typing import Any, Generator, List, Mapping, Optional, Tuple, Union
10
11 import pytest
12 from pkg_resources import Requirement, ResolutionError, resource_filename
13
14 from cwltool.context import LoadingContext, RuntimeContext
15 from cwltool.executors import JobExecutor
16 from cwltool.factory import Factory
17 from cwltool.main import main
18 from cwltool.singularity import is_version_2_6, is_version_3_or_newer
19 from cwltool.utils import onWindows, windows_default_container_id
20
21
22 def get_windows_safe_factory(
23 runtime_context: Optional[RuntimeContext] = None,
24 loading_context: Optional[LoadingContext] = None,
25 executor: Optional[JobExecutor] = None,
26 ) -> Factory:
27 if onWindows():
28 if not runtime_context:
29 runtime_context = RuntimeContext()
30 runtime_context.find_default_container = functools.partial(
31 force_default_container, windows_default_container_id
32 )
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
38 def force_default_container(default_container_id: str, _: str) -> str:
39 return default_container_id
40
41
42 def get_data(filename: str) -> str:
43 # normalizing path depending on OS or else it will cause problem when joining path
44 filename = os.path.normpath(filename)
45 filepath = None
46 try:
47 filepath = resource_filename(Requirement.parse("cwltool"), filename)
48 except ResolutionError:
49 pass
50 if not filepath or not os.path.isfile(filepath):
51 filepath = os.path.join(os.path.dirname(__file__), os.pardir, filename)
52 return str(Path(filepath).resolve())
53
54
55 needs_docker = pytest.mark.skipif(
56 not bool(distutils.spawn.find_executable("docker")),
57 reason="Requires the docker executable on the system path.",
58 )
59
60 needs_singularity = pytest.mark.skipif(
61 not bool(distutils.spawn.find_executable("singularity")),
62 reason="Requires the singularity executable on the system path.",
63 )
64
65 needs_singularity_2_6 = pytest.mark.skipif(
66 not bool(distutils.spawn.find_executable("singularity") and is_version_2_6()),
67 reason="Requires that version 2.6.x of singularity executable version is on the system path.",
68 )
69
70 needs_singularity_3_or_newer = pytest.mark.skipif(
71 (not bool(distutils.spawn.find_executable("singularity")))
72 or (not is_version_3_or_newer()),
73 reason="Requires that version 3.x of singularity executable version is on the system path.",
74 )
75
76
77 windows_needs_docker = pytest.mark.skipif(
78 onWindows() and not bool(distutils.spawn.find_executable("docker")),
79 reason="Running this test on MS Windows requires the docker executable "
80 "on the system path.",
81 )
82
83
84 def get_main_output(
85 args: List[str],
86 env: Union[
87 Mapping[bytes, Union[bytes, str]], Mapping[str, Union[bytes, str]], None
88 ] = None,
89 monkeypatch: Any = None,
90 ) -> Tuple[Optional[int], str, str]:
91 stdout = io.StringIO()
92 stderr = io.StringIO()
93 if env is not None:
94 assert monkeypatch is not None
95 monkeypatch.setattr(os, "environ", env)
96 try:
97 rc = main(argsl=args, stdout=stdout, stderr=stderr)
98 except SystemExit as e:
99 rc = e.code
100 return (
101 rc,
102 stdout.getvalue(),
103 stderr.getvalue(),
104 )
105
106
107 @contextlib.contextmanager
108 def working_directory(path: Union[str, Path]) -> Generator[None, None, None]:
109 """Change working directory and returns to previous on exit."""
110 prev_cwd = Path.cwd()
111 os.chdir(path)
112 try:
113 yield
114 finally:
115 os.chdir(prev_cwd)