Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/cwltool/tests/test_subgraph.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
-1:000000000000 | 0:4f3585e2f14b |
---|---|
1 import json | |
2 from pathlib import Path | |
3 from typing import Any | |
4 | |
5 from schema_salad.utils import convert_to_dict | |
6 | |
7 from cwltool.context import LoadingContext | |
8 from cwltool.load_tool import load_tool | |
9 from cwltool.subgraph import get_step, get_subgraph | |
10 from cwltool.workflow import Workflow, default_make_tool | |
11 | |
12 from .test_fetch import norm | |
13 from .util import get_data | |
14 | |
15 | |
16 def test_get_subgraph() -> None: | |
17 loadingContext = LoadingContext({"construct_tool_object": default_make_tool}) | |
18 wf = norm(Path(get_data("tests/subgraph/count-lines1-wf.cwl")).as_uri()) | |
19 loadingContext.do_update = False | |
20 tool = load_tool(wf, loadingContext) | |
21 | |
22 sg = norm(Path(get_data("tests/subgraph")).as_uri()) | |
23 | |
24 def clean(val: Any) -> Any: | |
25 if isinstance(val, str): | |
26 if val.startswith(sg): | |
27 return val[len(sg) + 1 :] | |
28 if isinstance(val, dict): | |
29 return {k: clean(v) for k, v in val.items()} | |
30 if isinstance(val, list): | |
31 return [clean(v) for v in val] | |
32 return val | |
33 | |
34 for a in ( | |
35 "file1", | |
36 "file2", | |
37 "file3", | |
38 "count_output", | |
39 "output3", | |
40 "output4", | |
41 "output5", | |
42 "step1", | |
43 "step2", | |
44 "step3", | |
45 "step4", | |
46 "step5", | |
47 ): | |
48 assert isinstance(tool, Workflow) | |
49 extracted = get_subgraph([wf + "#" + a], tool) | |
50 with open(get_data("tests/subgraph/extract_" + a + ".json")) as f: | |
51 assert json.load(f) == clean(convert_to_dict(extracted)) | |
52 | |
53 | |
54 def test_get_step() -> None: | |
55 loadingContext = LoadingContext({"construct_tool_object": default_make_tool}) | |
56 wf = norm(Path(get_data("tests/subgraph/count-lines1-wf.cwl")).as_uri()) | |
57 loadingContext.do_update = False | |
58 tool = load_tool(wf, loadingContext) | |
59 assert isinstance(tool, Workflow) | |
60 | |
61 sg = norm(Path(get_data("tests/subgraph")).as_uri()) | |
62 | |
63 def clean(val: Any) -> Any: | |
64 if isinstance(val, str): | |
65 if val.startswith(sg): | |
66 return val[len(sg) + 1 :] | |
67 if isinstance(val, dict): | |
68 return {k: clean(v) for k, v in val.items()} | |
69 if isinstance(val, list): | |
70 return [clean(v) for v in val] | |
71 return val | |
72 | |
73 for a in ( | |
74 "step1", | |
75 "step2", | |
76 "step3", | |
77 "step4", | |
78 "step5", | |
79 ): | |
80 extracted = get_step(tool, wf + "#" + a) | |
81 with open(get_data("tests/subgraph/single_" + a + ".json")) as f: | |
82 assert json.load(f) == clean(convert_to_dict(extracted)) |