Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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 j 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 j in tool.job(joborder, None, RuntimeContext()):
pass
def test_argparser_with_doc():
"""The `desription` field is set if `doc` field is provided."""
loadingContext = LoadingContext()
tool = load_tool(get_data("tests/with_doc.cwl"), loadingContext)
p = argparse.ArgumentParser()
parser = generate_parser(p, tool, {}, [], False)
assert parser.description is not None
def main(wf_file, json_file):
main_wf = cwltool.load_tool.load_tool(wf_file, cwltool.workflow.defaultMakeTool)
main_wf_dict = _wf_to_dict(main_wf)
nf_file = "%s.nf" % os.path.splitext(wf_file)[0]
with open(nf_file, "w") as out_handle:
for step in main_wf_dict["steps"]:
nf_step_to_process(step, out_handle)
def load_cwl(cwl_file):
load.loaders = {}
loading_context = cwltool.context.LoadingContext(get_default_args())
loading_context.construct_tool_object = default_make_tool
loading_context.resolver = tool_resolver
return load.load_tool(cwl_file, loading_context)
def make(self, cwl):
"""Instantiate a CWL object from a CWl document."""
load = load_tool.load_tool(cwl, self.loading_context)
if isinstance(load, int):
raise Exception("Error loading tool")
return Callable(load, self)
def result(
self,
job_order: CWLObjectType,
jobout: CWLObjectType,
runtimeContext: RuntimeContext,
) -> Tuple[Process, CWLObjectType]:
try:
loadingContext = self.loadingContext.copy()
loadingContext.metadata = {}
embedded_tool = load_tool(
cast(Dict[str, str], jobout["runProcess"])["location"], loadingContext
)
except ValidationException as vexc:
if runtimeContext.debug:
_logger.exception("Validation exception")
raise WorkflowException(
"Tool definition %s failed validation:\n%s"
% (jobout["runProcess"], indent(str(vexc)))
)
if "runInputs" in jobout:
runinputs = cast(CWLObjectType, jobout["runInputs"])
else:
runinputs = copy.deepcopy(job_order)
for i in self.embedded_tool.tool["inputs"]:
if shortname(i["id"]) in runinputs:
).get("requirements", []),
)
)
hints = copy.deepcopy(getdefault(loadingContext.hints, []))
hints.extend(toolpath_object.get("hints", []))
loadingContext.hints = hints
try:
if isinstance(toolpath_object["run"], CommentedMap):
self.embedded_tool = loadingContext.construct_tool_object(
toolpath_object["run"], loadingContext
) # type: Process
else:
loadingContext.metadata = {}
self.embedded_tool = load_tool(toolpath_object["run"], loadingContext)
except ValidationException as vexc:
if loadingContext.debug:
_logger.exception("Validation exception")
raise WorkflowException(
"Tool definition %s failed validation:\n%s"
% (toolpath_object["run"], indent(str(vexc)))
) from vexc
validation_errors = []
self.tool = toolpath_object = copy.deepcopy(toolpath_object)
bound = set()
for stepfield, toolfield in (("in", "inputs"), ("out", "outputs")):
toolpath_object[toolfield] = []
for index, step_entry in enumerate(toolpath_object[stepfield]):
if isinstance(step_entry, str):
param = CommentedMap() # type: CommentedMap
def main(wf_file, json_file):
wf_file = os.path.abspath(wf_file)
out_dir = os.path.dirname(wf_file).replace("-workflow", "-wdl")
if not os.path.exists(out_dir):
os.makedirs(out_dir)
main_wf = cwltool.load_tool.load_tool(wf_file, cwltool.workflow.defaultMakeTool)
records = {}
main_wf_dict, records = _wf_to_dict(main_wf, records)
main_wf_dict["structs"] = records
main_wf_class = cwl2wdl_classes.Workflow(main_wf_dict)
for wf_class in [x.task_definition for x in main_wf_class.subworkflows] + [main_wf_class]:
wdl_file = os.path.join(out_dir, "%s.wdl" % wf_class.name)
wdl_doc = generators.WdlWorkflowGenerator(wf_class).generate_wdl()
with open(wdl_file, "w") as out_handle:
out_handle.write(wdl_doc)
_validate(wdl_file)