1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
# SPDX-License-Identifier: Apache-2.0
"""Tests for cwl-docker-extract."""
from pathlib import Path
import pytest
from cwl_utils.docker_extract import arg_parser, run
from .util import get_data, needs_docker, needs_podman, needs_singularity
@pytest.mark.parametrize(
("target", "engine"),
[
pytest.param("testdata/md5sum.cwl", "docker", marks=needs_docker),
pytest.param("testdata/md5sum_v11.cwl", "docker", marks=needs_docker),
pytest.param("testdata/md5sum.cwl", "podman", marks=needs_podman),
pytest.param("testdata/md5sum_v11.cwl", "podman", marks=needs_podman),
pytest.param("testdata/md5sum.cwl", "singularity", marks=needs_singularity),
pytest.param("testdata/md5sum_v11.cwl", "singularity", marks=needs_singularity),
],
)
def test_container_extraction(target: str, engine: str, tmp_path: Path) -> None:
"""Test container extraction tool."""
args = ["--dir", str(tmp_path), get_data(target), "--container-engine", engine]
if engine == "singularity":
args.append("--singularity")
reqs = run(arg_parser().parse_args(args))
assert len(reqs) == 1
assert len(list(tmp_path.iterdir())) == 1
@pytest.mark.parametrize(
("engine"),
[
pytest.param("docker", marks=needs_docker),
pytest.param("podman", marks=needs_podman),
pytest.param("singularity", marks=needs_singularity),
],
)
def test_container_extraction_force(engine: str, tmp_path: Path) -> None:
"""Test force pull container extraction."""
args = [
"--dir",
str(tmp_path),
get_data("testdata/md5sum.cwl"),
"--container-engine",
engine,
]
if engine == "singularity":
args.append("--singularity")
reqs = run(arg_parser().parse_args(args))
assert len(reqs) == 1
assert len(list(tmp_path.iterdir())) == 1
args = [
"--dir",
str(tmp_path),
get_data("testdata/md5sum.cwl"),
"--container-engine",
engine,
"--force-download",
]
if engine == "singularity":
args.append("--singularity")
reqs = run(arg_parser().parse_args(args))
assert len(reqs) == 1
assert len(list(tmp_path.iterdir())) == 1
@pytest.mark.parametrize(
("engine"),
[
pytest.param("docker", marks=needs_docker),
pytest.param("podman", marks=needs_podman),
pytest.param("singularity", marks=needs_singularity),
],
)
def test_container_extraction_no_dockerPull(
engine: str, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""Test container extraction tool when dockerPull is missing."""
args = [
"--dir",
str(tmp_path),
get_data("testdata/debian_image_id.cwl"),
"--container-engine",
engine,
]
if engine == "singularity":
args.append("--singularity")
reqs = run(arg_parser().parse_args(args))
assert len(reqs) == 1
assert len(list(tmp_path.iterdir())) == 0
captured = capsys.readouterr()
assert (
captured.err
== """Unable to save image from due to lack of 'dockerPull':
class: DockerRequirement
dockerImageId: 'debian:stable-slim.img'
"""
)
@pytest.mark.parametrize(
("engine"),
[
pytest.param("docker", marks=needs_docker),
pytest.param("podman", marks=needs_podman),
pytest.param("singularity", marks=needs_singularity),
],
)
def test_container_extraction_embedded_step(engine: str, tmp_path: Path) -> None:
"""Test container extraction tool."""
args = [
"--dir",
str(tmp_path),
get_data("testdata/workflows/count-lines16-wf.cwl"),
"--container-engine",
engine,
]
if engine == "singularity":
args.append("--singularity")
reqs = run(arg_parser().parse_args(args))
assert len(reqs) == 1
assert len(list(tmp_path.iterdir())) == 1
|
