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

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

0:4f3585e2f14b

"""Test that all temporary directories respect the --tmpdir-prefix and --tmp-outdir-prefix options."""
import subprocess
import sys
from pathlib import Path
from typing import List, cast

from _pytest.monkeypatch import MonkeyPatch
from ruamel.yaml.comments import CommentedMap
from schema_salad.avro import schema
from schema_salad.sourceline import cmap

from cwltool.builder import Builder
from cwltool.command_line_tool import CommandLineTool
from cwltool.context import LoadingContext, RuntimeContext
from cwltool.docker import DockerCommandLineJob
from cwltool.job import JobBase
from cwltool.pathmapper import MapperEnt, PathMapper
from cwltool.stdfsaccess import StdFsAccess
from cwltool.update import INTERNAL_VERSION
from cwltool.utils import create_tmp_dir

from .util import get_data, needs_docker


def test_docker_commandLineTool_job_tmpdir_prefix(tmp_path: Path) -> None:
    """Test that docker enabled CommandLineTool respects temp directory directives."""
    loading_context = LoadingContext(
        {
            "metadata": {
                "cwlVersion": INTERNAL_VERSION,
                "http://commonwl.org/cwltool#original_cwlVersion": INTERNAL_VERSION,
            }
        }
    )
    clt = CommandLineTool(
        cast(
            CommentedMap,
            cmap(
                {
                    "cwlVersion": INTERNAL_VERSION,
                    "class": "CommandLineTool",
                    "inputs": [],
                    "outputs": [],
                    "requirements": [
                        {"class": "DockerRequirement", "dockerPull": "debian:stable"}
                    ],
                }
            ),
        ),
        loading_context,
    )
    tmpdir_prefix = str(tmp_path / "1")
    tmp_outdir_prefix = str(tmp_path / "2")
    runtime_context = RuntimeContext(
        {
            "tmpdir_prefix": tmpdir_prefix,
            "tmp_outdir_prefix": tmp_outdir_prefix,
        }
    )
    job = next(clt.job({}, None, runtime_context))
    assert isinstance(job, JobBase)
    assert job.stagedir and job.stagedir.startswith(tmpdir_prefix)
    assert job.tmpdir and job.tmpdir.startswith(tmpdir_prefix)
    assert job.outdir and job.outdir.startswith(tmp_outdir_prefix)


def test_commandLineTool_job_tmpdir_prefix(tmp_path: Path) -> None:
    """Test that non-docker enabled CommandLineTool respects temp directory directives."""
    loading_context = LoadingContext(
        {
            "metadata": {
                "cwlVersion": INTERNAL_VERSION,
                "http://commonwl.org/cwltool#original_cwlVersion": INTERNAL_VERSION,
            }
        }
    )
    clt = CommandLineTool(
        cast(
            CommentedMap,
            cmap(
                {
                    "cwlVersion": INTERNAL_VERSION,
                    "class": "CommandLineTool",
                    "inputs": [],
                    "outputs": [],
                    "requirements": [],
                }
            ),
        ),
        loading_context,
    )
    tmpdir_prefix = str(tmp_path / "1")
    tmp_outdir_prefix = str(tmp_path / "2")
    runtime_context = RuntimeContext(
        {
            "tmpdir_prefix": tmpdir_prefix,
            "tmp_outdir_prefix": tmp_outdir_prefix,
        }
    )
    job = next(clt.job({}, None, runtime_context))
    assert isinstance(job, JobBase)
    assert job.stagedir and job.stagedir.startswith(tmpdir_prefix)
    assert job.tmpdir and job.tmpdir.startswith(tmpdir_prefix)
    assert job.outdir and job.outdir.startswith(tmp_outdir_prefix)


@needs_docker
def test_dockerfile_tmpdir_prefix(tmp_path: Path, monkeypatch: MonkeyPatch) -> None:
    """Test that DockerCommandLineJob.get_image respects temp directory directives."""
    monkeypatch.setattr(
        target=subprocess, name="check_call", value=lambda *args, **kwargs: True
    )
    tmp_outdir_prefix = tmp_path / "1"
    assert DockerCommandLineJob.get_image(
        {
            "class": "DockerRequirement",
            "dockerFile": "FROM debian:stable",
            "dockerImageId": sys._getframe().f_code.co_name,
        },
        pull_image=True,
        force_pull=True,
        tmp_outdir_prefix=str(tmp_outdir_prefix),
    )
    children = sorted(tmp_outdir_prefix.parent.glob("*"))
    assert len(children) == 1
    subdir = tmp_path / children[0]
    assert len(sorted(subdir.glob("*"))) == 1
    assert (subdir / "Dockerfile").exists()


def test_docker_tmpdir_prefix(tmp_path: Path) -> None:
    """Test that DockerCommandLineJob respects temp directory directives."""
    (tmp_path / "3").mkdir()
    tmpdir_prefix = str(tmp_path / "3" / "ttmp")
    runtime_context = RuntimeContext(
        {"tmpdir_prefix": tmpdir_prefix, "user_space_docker_cmd": None}
    )
    builder = Builder(
        {},
        [],
        [],
        {},
        schema.Names(),
        [],
        [],
        {},
        None,
        None,
        StdFsAccess,
        StdFsAccess(""),
        None,
        0.1,
        False,
        False,
        False,
        "",
        runtime_context.get_outdir(),
        runtime_context.get_tmpdir(),
        runtime_context.get_stagedir(),
        INTERNAL_VERSION,
    )
    job = DockerCommandLineJob(builder, {}, PathMapper, [], [], "")
    runtime: List[str] = []

    volume_writable_file = MapperEnt(
        resolved=get_data("tests/2.fastq"), target="foo", type=None, staged=None
    )
    (tmp_path / "1").mkdir()
    job.add_writable_file_volume(
        runtime, volume_writable_file, None, str(tmp_path / "1" / "writable_file")
    )
    children = sorted((tmp_path / "1").glob("*"))
    assert len(children) == 1
    subdir = tmp_path / children[0]
    assert subdir.name.startswith("writable_file")
    assert len(sorted(subdir.glob("*"))) == 1
    assert (subdir / "2.fastq").exists()

    resolved_writable_dir = tmp_path / "data_orig"
    resolved_writable_dir.mkdir(parents=True)
    volume_dir = MapperEnt(
        resolved=str(resolved_writable_dir), target="bar", type=None, staged=None
    )
    (tmp_path / "2").mkdir()
    job.add_writable_directory_volume(
        runtime, volume_dir, None, str(tmp_path / "2" / "dir")
    )
    children = sorted((tmp_path / "2").glob("*"))
    assert len(children) == 1
    subdir = tmp_path / "2" / children[0]
    assert subdir.name.startswith("dir")
    assert len(sorted(subdir.glob("*"))) == 1
    assert (subdir / "data_orig").exists()

    cidfile = job.create_runtime({}, runtime_context)[1]
    assert cidfile and cidfile.startswith(tmpdir_prefix)

    volume_file = MapperEnt(resolved="Hoopla!", target="baz", type=None, staged=None)
    (tmp_path / "4").mkdir()
    job.create_file_and_add_volume(
        runtime, volume_file, None, None, str(tmp_path / "4" / "file")
    )
    children = sorted((tmp_path / "4").glob("*"))
    assert len(children) == 1
    subdir = tmp_path / "4" / children[0]
    assert subdir.name.startswith("file")
    assert len(sorted(subdir.glob("*"))) == 1
    assert (subdir / "baz").exists()


def test_runtimeContext_respects_tmpdir_prefix(tmp_path: Path) -> None:
    """Test that RuntimeContext helper methods respects tmpdir_prefix."""
    tmpdir_prefix = str(tmp_path / "foo")
    runtime_context = RuntimeContext({"tmpdir_prefix": tmpdir_prefix})
    assert runtime_context.get_tmpdir().startswith(tmpdir_prefix)
    assert runtime_context.get_stagedir().startswith(tmpdir_prefix)
    assert runtime_context.create_tmpdir().startswith(tmpdir_prefix)
    assert create_tmp_dir(tmpdir_prefix).startswith(tmpdir_prefix)


def test_runtimeContext_respects_tmp_outdir_prefix(tmp_path: Path) -> None:
    """Test that RuntimeContext helper methods respects tmp_outdir_prefix."""
    tmpdir_prefix = str(tmp_path / "foo")
    runtime_context = RuntimeContext({"tmp_outdir_prefix": tmpdir_prefix})
    assert runtime_context.get_outdir().startswith(tmpdir_prefix)
    assert runtime_context.create_outdir().startswith(tmpdir_prefix)