Mercurial > repos > shellac > sam_consensus_v3
diff env/lib/python3.9/site-packages/cwltool/tests/test_load_tool.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_load_tool.py Mon Mar 22 18:12:50 2021 +0000 @@ -0,0 +1,103 @@ +from pathlib import Path + +import pytest + +from cwltool.context import LoadingContext, RuntimeContext +from cwltool.errors import WorkflowException +from cwltool.load_tool import load_tool +from cwltool.process import use_custom_schema, use_standard_schema +from cwltool.update import INTERNAL_VERSION +from cwltool.utils import CWLObjectType + +from .test_fetch import norm +from .util import get_data, windows_needs_docker + + +@windows_needs_docker +def test_check_version() -> None: + """ + It is permitted to load without updating, but not execute. + + Attempting to execute without updating to the internal version should raise an error. + """ + joborder = {"inp": "abc"} # type: CWLObjectType + loadingContext = LoadingContext({"do_update": True}) + tool = load_tool(get_data("tests/echo.cwl"), loadingContext) + for _ in tool.job(joborder, None, RuntimeContext()): + pass + + loadingContext = LoadingContext({"do_update": False}) + tool = load_tool(get_data("tests/echo.cwl"), loadingContext) + with pytest.raises(WorkflowException): + for _ in tool.job(joborder, None, RuntimeContext()): + pass + + +def test_use_metadata() -> None: + """Use the version from loadingContext.metadata if cwlVersion isn't present in the document.""" + loadingContext = LoadingContext({"do_update": False}) + tool = load_tool(get_data("tests/echo.cwl"), loadingContext) + + loadingContext = LoadingContext() + loadingContext.metadata = tool.metadata + tooldata = tool.tool.copy() + del tooldata["cwlVersion"] + load_tool(tooldata, loadingContext) + + +def test_checklink_outputSource() -> None: + """Is outputSource resolved correctly independent of value of do_validate.""" + outsrc = ( + norm(Path(get_data("tests/wf/1st-workflow.cwl")).as_uri()) + + "#argument/classfile" + ) + + loadingContext = LoadingContext({"do_validate": True}) + tool = load_tool(get_data("tests/wf/1st-workflow.cwl"), loadingContext) + assert norm(tool.tool["outputs"][0]["outputSource"]) == outsrc + + loadingContext = LoadingContext({"do_validate": False}) + tool = load_tool(get_data("tests/wf/1st-workflow.cwl"), loadingContext) + assert norm(tool.tool["outputs"][0]["outputSource"]) == outsrc + + +def test_load_graph_fragment() -> None: + """Reloading from a dictionary without a cwlVersion.""" + loadingContext = LoadingContext() + uri = Path(get_data("tests/wf/scatter-wf4.cwl")).as_uri() + "#main" + tool = load_tool(uri, loadingContext) + + loader = tool.doc_loader + assert loader + rs, metadata = loader.resolve_ref(uri) + # Reload from a dict (in 'rs'), not a URI. The dict is a fragment + # of original document and doesn't have cwlVersion set, so test + # that it correctly looks up the root document to get the + # cwlVersion. + assert isinstance(rs, str) or isinstance(rs, dict) + tool = load_tool(rs, loadingContext) + assert tool.metadata["cwlVersion"] == INTERNAL_VERSION + + +def test_load_graph_fragment_from_packed() -> None: + """Loading a fragment from packed with update.""" + loadingContext = LoadingContext() + uri = Path(get_data("tests/wf/packed-with-loadlisting.cwl")).as_uri() + "#main" + try: + with open(get_data("cwltool/extensions.yml")) as res: + use_custom_schema("v1.0", "http://commonwl.org/cwltool", res.read()) + + # The updater transforms LoadListingRequirement from an + # extension (in v1.0) to a core feature (in v1.1) but there + # was a bug when loading a packed workflow and loading a + # specific fragment it would get the un-updated document. + # This recreates that case and asserts that we are using the + # updated document like we should. + + tool = load_tool(uri, loadingContext) + + assert tool.tool["requirements"] == [ + {"class": "LoadListingRequirement", "loadListing": "no_listing"} + ] + finally: + use_standard_schema("v1.0")