cwltool/tests/test_validate.py at 4937dbc78260e2da796a569c05a79dddfd10ed32 · common-workflow-language/cwltool · GitHub

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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
"""Tests --validation."""

import io
import logging
import re

import pytest

from .util import get_data, get_main_output


def test_validate_graph_with_no_default() -> None:
    """Ensure that --validate works on $graph docs that lack a main/#main."""
    exit_code, stdout, stderr = get_main_output(
        ["--validate", get_data("tests/wf/packed_no_main.cwl")]
    )
    assert exit_code == 0
    assert "packed_no_main.cwl#echo is valid CWL" in stdout
    assert "packed_no_main.cwl#cat is valid CWL" in stdout
    assert "packed_no_main.cwl#collision is valid CWL" in stdout
    assert "tests/wf/packed_no_main.cwl is valid CWL" in stdout


def test_validate_with_valid_input_object() -> None:
    """Ensure that --validate with a valid input object."""
    exit_code, stdout, stderr = get_main_output(
        [
            "--validate",
            get_data("tests/wf/1st-workflow.cwl"),
            "--inp",
            get_data("tests/wf/1st-workflow.cwl"),
            "--ex",
            "FOO",
        ]
    )
    assert exit_code == 0
    assert "tests/wf/1st-workflow.cwl is valid CWL. No errors detected in the inputs." in stdout


def test_validate_with_invalid_input_object() -> None:
    """Ensure that --validate with an invalid input object."""
    exit_code, stdout, stderr = get_main_output(
        [
            "--validate",
            get_data("tests/wf/1st-workflow.cwl"),
            get_data("tests/wf/1st-workflow_bad_inputs.yml"),
        ]
    )
    assert exit_code == 1
    stdout = re.sub(r"\s\s+", " ", stdout)
    assert "Invalid job input record" in stdout
    assert (
        "tests/wf/1st-workflow_bad_inputs.yml:2:1: * the 'ex' field is not "
        "valid because the value is not string" in stdout
    )
    assert (
        "tests/wf/1st-workflow_bad_inputs.yml:1:1: * the 'inp' field is not "
        "valid because is not a dict. Expected a File object." in stdout
    )


def test_validate_quiet() -> None:
    """Ensure that --validate --quiet prints the correct amount of information."""
    exit_code, stdout, stderr = get_main_output(
        [
            "--validate",
            "--quiet",
            get_data("tests/CometAdapter.cwl"),
        ]
    )
    assert exit_code == 0
    stdout = re.sub(r"\s\s+", " ", stdout)
    assert "INFO" not in stdout
    assert "INFO" not in stderr
    assert "tests/CometAdapter.cwl:10:3: object id" in stdout
    assert "tests/CometAdapter.cwl#out' previously defined" in stdout


def test_validate_no_warnings() -> None:
    """Ensure that --validate --no-warnings doesn't print any warnings."""
    exit_code, stdout, stderr = get_main_output(
        [
            "--validate",
            "--no-warnings",
            get_data("tests/CometAdapter.cwl"),
        ]
    )
    assert exit_code == 0
    stdout = re.sub(r"\s\s+", " ", stdout)
    stderr = re.sub(r"\s\s+", " ", stderr)
    assert "INFO" not in stdout
    assert "INFO" not in stderr
    assert "WARNING" not in stdout
    assert "WARNING" not in stderr
    assert "tests/CometAdapter.cwl:9:3: object id" not in stdout
    assert "tests/CometAdapter.cwl:9:3: object id" not in stderr
    assert "tests/CometAdapter.cwl#out' previously defined" not in stdout
    assert "tests/CometAdapter.cwl#out' previously defined" not in stderr


def test_validate_custom_logger() -> None:
    """Custom log handling test."""
    custom_log = io.StringIO()
    handler = logging.StreamHandler(custom_log)
    handler.setLevel(logging.DEBUG)
    exit_code, stdout, stderr = get_main_output(
        [
            "--validate",
            get_data("tests/CometAdapter.cwl"),
        ],
        logger_handler=handler,
    )
    custom_log_text = custom_log.getvalue()
    assert exit_code == 0
    custom_log_text = re.sub(r"\s\s+", " ", custom_log_text)
    stdout = re.sub(r"\s\s+", " ", stdout)
    stderr = re.sub(r"\s\s+", " ", stderr)
    assert "INFO" not in stdout
    assert "INFO" not in stderr
    assert "INFO" in custom_log_text
    assert "WARNING" not in stdout
    assert "WARNING" not in stderr
    assert "WARNING" in custom_log_text
    assert "tests/CometAdapter.cwl:10:3: object id" not in stdout
    assert "tests/CometAdapter.cwl:10:3: object id" not in stderr
    assert "tests/CometAdapter.cwl:10:3: object id" in custom_log_text
    assert "tests/CometAdapter.cwl#out' previously defined" not in stdout
    assert "tests/CometAdapter.cwl#out' previously defined" not in stderr
    assert "tests/CometAdapter.cwl#out' previously defined" in custom_log_text


@pytest.mark.parametrize(
    "file", ["tests/wf/bad_resreq_mnmx_clt.cwl", "tests/wf/bad_resreq_mnmx_wf.cwl"]
)
def test_validate_with_invalid_requirements(file: str) -> None:
    """Ensure that --validate returns an error with an invalid resource requirement."""
    exit_code, stdout, stderr = get_main_output(
        [
            "--validate",
            get_data(file),
        ]
    )
    assert exit_code == 1
    assert "cannot be greater than" in stdout