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

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author shellac
date Mon, 22 Mar 2021 18:12:50 +0000
parents
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/env/lib/python3.9/site-packages/cwltool/tests/util.py	Mon Mar 22 18:12:50 2021 +0000
@@ -0,0 +1,115 @@
+import contextlib
+import distutils.spawn  # pylint: disable=no-name-in-module,import-error
+import functools
+import io
+import os
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any, Generator, List, Mapping, Optional, Tuple, Union
+
+import pytest
+from pkg_resources import Requirement, ResolutionError, resource_filename
+
+from cwltool.context import LoadingContext, RuntimeContext
+from cwltool.executors import JobExecutor
+from cwltool.factory import Factory
+from cwltool.main import main
+from cwltool.singularity import is_version_2_6, is_version_3_or_newer
+from cwltool.utils import onWindows, windows_default_container_id
+
+
+def get_windows_safe_factory(
+    runtime_context: Optional[RuntimeContext] = None,
+    loading_context: Optional[LoadingContext] = None,
+    executor: Optional[JobExecutor] = None,
+) -> Factory:
+    if onWindows():
+        if not runtime_context:
+            runtime_context = RuntimeContext()
+        runtime_context.find_default_container = functools.partial(
+            force_default_container, windows_default_container_id
+        )
+        runtime_context.use_container = True
+        runtime_context.default_container = windows_default_container_id
+    return Factory(executor, loading_context, runtime_context)
+
+
+def force_default_container(default_container_id: str, _: str) -> str:
+    return default_container_id
+
+
+def get_data(filename: str) -> str:
+    # normalizing path depending on OS or else it will cause problem when joining path
+    filename = os.path.normpath(filename)
+    filepath = None
+    try:
+        filepath = resource_filename(Requirement.parse("cwltool"), filename)
+    except ResolutionError:
+        pass
+    if not filepath or not os.path.isfile(filepath):
+        filepath = os.path.join(os.path.dirname(__file__), os.pardir, filename)
+    return str(Path(filepath).resolve())
+
+
+needs_docker = pytest.mark.skipif(
+    not bool(distutils.spawn.find_executable("docker")),
+    reason="Requires the docker executable on the system path.",
+)
+
+needs_singularity = pytest.mark.skipif(
+    not bool(distutils.spawn.find_executable("singularity")),
+    reason="Requires the singularity executable on the system path.",
+)
+
+needs_singularity_2_6 = pytest.mark.skipif(
+    not bool(distutils.spawn.find_executable("singularity") and is_version_2_6()),
+    reason="Requires that version 2.6.x of singularity executable version is on the system path.",
+)
+
+needs_singularity_3_or_newer = pytest.mark.skipif(
+    (not bool(distutils.spawn.find_executable("singularity")))
+    or (not is_version_3_or_newer()),
+    reason="Requires that version 3.x of singularity executable version is on the system path.",
+)
+
+
+windows_needs_docker = pytest.mark.skipif(
+    onWindows() and not bool(distutils.spawn.find_executable("docker")),
+    reason="Running this test on MS Windows requires the docker executable "
+    "on the system path.",
+)
+
+
+def get_main_output(
+    args: List[str],
+    env: Union[
+        Mapping[bytes, Union[bytes, str]], Mapping[str, Union[bytes, str]], None
+    ] = None,
+    monkeypatch: Any = None,
+) -> Tuple[Optional[int], str, str]:
+    stdout = io.StringIO()
+    stderr = io.StringIO()
+    if env is not None:
+        assert monkeypatch is not None
+        monkeypatch.setattr(os, "environ", env)
+    try:
+        rc = main(argsl=args, stdout=stdout, stderr=stderr)
+    except SystemExit as e:
+        rc = e.code
+    return (
+        rc,
+        stdout.getvalue(),
+        stderr.getvalue(),
+    )
+
+
+@contextlib.contextmanager
+def working_directory(path: Union[str, Path]) -> Generator[None, None, None]:
+    """Change working directory and returns to previous on exit."""
+    prev_cwd = Path.cwd()
+    os.chdir(path)
+    try:
+        yield
+    finally:
+        os.chdir(prev_cwd)