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
|
"""
Checks loading of some real world tools and workflows found in the wild (e.g. dockstore)
run individually as py.test -k tests/test_real_cwl.py
"""
from typing import Any, Dict, Optional, Union
import pytest
from schema_salad.avro.schema import Names, SchemaParseException
from schema_salad.exceptions import ValidationException
from schema_salad.ref_resolver import Loader
from schema_salad.schema import load_and_validate, load_schema
from .util import get_data
test_dir_name = "tests/test_real_cwl/"
class TestRealWorldCWL:
document_loader: Loader
avsc_names: Union[Names, SchemaParseException, None] = None
schema_metadata: Optional[Dict[str, Any]] = None
metaschema_loader: Optional[Loader] = None
@classmethod
def setup_class(cls) -> None:
path = get_data("tests/test_schema/CommonWorkflowLanguage.yml")
assert path
(
cls.document_loader,
cls.avsc_names,
schema_metadata,
metaschema_loader,
) = load_schema(path)
def load_cwl(self, src: str) -> None:
path = get_data(test_dir_name + src)
assert path
assert isinstance(self.avsc_names, Names)
with pytest.raises(ValidationException):
try:
load_and_validate(
self.document_loader,
self.avsc_names,
path,
True,
)
except ValidationException as e:
# msgs = to_one_line_messages(str(e)).splitlines()
print("\n", e)
raise
def test_topmed_single_doc(self) -> None:
"""TOPMed Variant Calling Pipeline CWL1"""
self.load_cwl(src="topmed/topmed_variant_calling_pipeline.cwl")
def test_h3agatk_WES(self) -> None:
"""H3ABioNet GATK Germline Workflow"""
self.load_cwl(src="h3agatk/GATK-complete-WES-Workflow-h3abionet.cwl")
def test_h3agatk_SNP(self) -> None:
"""H3ABioNet SNPs Workflow"""
self.load_cwl(src="h3agatk/GATK-Sub-Workflow-h3abionet-snp.cwl")
def test_icgc_pancan(self) -> None:
"""ICGC PanCan"""
self.load_cwl(src="ICGC-TCGA-PanCancer/preprocess_vcf.cwl")
|
