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_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"), "r") 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"}
def test_argparser_without_doc():
"""The `desription` field is None if `doc` field is not provided."""
loadingContext = LoadingContext()
tool = load_tool(get_data("tests/without_doc.cwl"), loadingContext)
p = argparse.ArgumentParser()
parser = generate_parser(p, tool, {}, [], False)
assert parser.description is None
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 fetch_document(
argsworkflow: Union[str, CWLObjectType],
loadingContext: Optional[LoadingContext] = None,
) -> Tuple[LoadingContext, CommentedMap, str]:
"""Retrieve a CWL document."""
if loadingContext is None:
loadingContext = LoadingContext()
loadingContext.loader = default_loader()
else:
loadingContext = loadingContext.copy()
if loadingContext.loader is None:
loadingContext.loader = default_loader(
loadingContext.fetcher_constructor,
enable_dev=loadingContext.enable_dev,
doc_cache=loadingContext.doc_cache,
)
if isinstance(argsworkflow, str):
uri, fileuri = resolve_tool_uri(
argsworkflow,
resolver=loadingContext.resolver,
document_loader=loadingContext.loader,
)
conf_file = getattr(options,
"beta_dependency_resolvers_configuration", None)
use_conda_dependencies = getattr(options, "beta_conda_dependencies", None)
job_script_provider = None
if conf_file or use_conda_dependencies:
dependencies_configuration = DependenciesConfiguration(options)
job_script_provider = dependencies_configuration
options.default_container = None
runtime_context = cwltool.context.RuntimeContext(vars(options))
runtime_context.find_default_container = functools.partial(
find_default_container, options)
runtime_context.workdir = workdir
runtime_context.move_outputs = "leave"
runtime_context.rm_tmpdir = False
loading_context = cwltool.context.LoadingContext(vars(options))
if options.provenance:
research_obj = cwltool.provenance.ResearchObject(
temp_prefix_ro=options.tmp_outdir_prefix, orcid=options.orcid,
full_name=options.cwl_full_name,
fsaccess=runtime_context.make_fs_access(''))
runtime_context.research_obj = research_obj
with Toil(options) as toil:
if options.restart:
outobj = toil.restart()
else:
loading_context.hints = [{
"class": "ResourceRequirement",
"coresMin": toil.config.defaultCores,
"ramMin": toil.config.defaultMemory / (2**20),
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)
visit_class(toolpath_object, ('File', 'Directory'), addLocation)
return workflow.default_make_tool(toolpath_object, *args, **kwargs)
argv = sys.argv
sys.argv = ['cwltool']
# Keep all environment variables.
runtime_context = RuntimeContext(
kwargs={
'rm_tmpdir': False,
'move_outputs': 'leave',
'preserve_entire_environment': True,
}
)
loading_context = LoadingContext(
kwargs={
'construct_tool_object': construct_tool_object,
}
)
factory = cwltool.factory.Factory(
loading_context=loading_context,
runtime_context=runtime_context,
)
process = factory.make(os.path.relpath(str(output_file)))
outputs = process()
sys.argv = argv
# Move outputs to correct location in the repository.
output_dirs = process.factory.executor.output_dirs
logger.debug('Loading CWL file "{}"'.format(fname))
# Fetching, preprocessing and validating cwl
# Older versions of cwltool
if legacy_cwltool:
try:
(document_loader, workflowobj, uri) = fetch_document(fname)
(document_loader, _, processobj, metadata, uri) = \
validate_document(document_loader, workflowobj, uri)
except TypeError:
from cwltool.context import LoadingContext, getdefault
from cwltool import workflow
from cwltool.resolver import tool_resolver
from cwltool.load_tool import resolve_tool_uri
loadingContext = LoadingContext()
loadingContext.construct_tool_object = getdefault(
loadingContext.construct_tool_object,
workflow.default_make_tool)
loadingContext.resolver = getdefault(loadingContext.resolver,
tool_resolver)
uri, tool_file_uri = resolve_tool_uri(
fname, resolver=loadingContext.resolver,
fetcher_constructor=loadingContext.fetcher_constructor)
document_loader, workflowobj, uri = fetch_document(
uri, resolver=loadingContext.resolver,
fetcher_constructor=loadingContext.fetcher_constructor)
document_loader, avsc_names, processobj, metadata, uri = \
validate_document(
document_loader, workflowobj, uri,
def load_cwl(cwl_file, default_args):
load.loaders = {}
loading_context = LoadingContext(default_args)
loading_context.construct_tool_object = default_make_tool
loading_context.resolver = tool_resolver
tool = load_tool(cwl_file, loading_context)
it_is_workflow = tool.tool["class"] == "Workflow"
return tool, it_is_workflow