Mercurial > repos > shellac > sam_consensus_v3
diff 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 |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/env/lib/python3.9/site-packages/cwltool/tests/test_subgraph.py Mon Mar 22 18:12:50 2021 +0000 @@ -0,0 +1,82 @@ +import json +from pathlib import Path +from typing import Any + +from schema_salad.utils import convert_to_dict + +from cwltool.context import LoadingContext +from cwltool.load_tool import load_tool +from cwltool.subgraph import get_step, get_subgraph +from cwltool.workflow import Workflow, default_make_tool + +from .test_fetch import norm +from .util import get_data + + +def test_get_subgraph() -> None: + loadingContext = LoadingContext({"construct_tool_object": default_make_tool}) + wf = norm(Path(get_data("tests/subgraph/count-lines1-wf.cwl")).as_uri()) + loadingContext.do_update = False + tool = load_tool(wf, loadingContext) + + sg = norm(Path(get_data("tests/subgraph")).as_uri()) + + def clean(val: Any) -> Any: + if isinstance(val, str): + if val.startswith(sg): + return val[len(sg) + 1 :] + if isinstance(val, dict): + return {k: clean(v) for k, v in val.items()} + if isinstance(val, list): + return [clean(v) for v in val] + return val + + for a in ( + "file1", + "file2", + "file3", + "count_output", + "output3", + "output4", + "output5", + "step1", + "step2", + "step3", + "step4", + "step5", + ): + assert isinstance(tool, Workflow) + extracted = get_subgraph([wf + "#" + a], tool) + with open(get_data("tests/subgraph/extract_" + a + ".json")) as f: + assert json.load(f) == clean(convert_to_dict(extracted)) + + +def test_get_step() -> None: + loadingContext = LoadingContext({"construct_tool_object": default_make_tool}) + wf = norm(Path(get_data("tests/subgraph/count-lines1-wf.cwl")).as_uri()) + loadingContext.do_update = False + tool = load_tool(wf, loadingContext) + assert isinstance(tool, Workflow) + + sg = norm(Path(get_data("tests/subgraph")).as_uri()) + + def clean(val: Any) -> Any: + if isinstance(val, str): + if val.startswith(sg): + return val[len(sg) + 1 :] + if isinstance(val, dict): + return {k: clean(v) for k, v in val.items()} + if isinstance(val, list): + return [clean(v) for v in val] + return val + + for a in ( + "step1", + "step2", + "step3", + "step4", + "step5", + ): + extracted = get_step(tool, wf + "#" + a) + with open(get_data("tests/subgraph/single_" + a + ".json")) as f: + assert json.load(f) == clean(convert_to_dict(extracted))