Toolfactory (#1133)

* Tool Factory experiment.

* Pass through inputs by default

* Fix mypy

* Rename ToolFactory -> WorkflowGenerator

Add a minimal test

* Add workflow generator test files

* Windows supports '/' as a path separator.

* Rename WorkflowGenerator -> ProcessGenerator

* Fix tests

* Fix manifest

* Bugfix and test for create_file_and_add_volume with file literals.

* Undo side effect of changing CWLTOOL_OPTIONS for predictable test behavior

Author: Peter Amstutz

MANIFEST.in

@@ -10,6 +10,7 @@ include tests/override/*
 include tests/checker_wf/*
 include tests/subgraph/*
 include tests/trs/*
+include tests/wf/generator/*
 include cwltool/schemas/v1.0/*.yml
 include cwltool/schemas/v1.0/*.yml
 include cwltool/schemas/v1.0/*.md

cwltool/command_line_tool.py

@@ -169,9 +169,9 @@ def revmap_file(builder, outdir, f):
 
         if revmap_f and not builder.pathmapper.mapper(revmap_f[0]).type.startswith("Writable"):
             f["location"] = revmap_f[1]
-        elif uripath == outdir or uripath.startswith(outdir+os.sep):
+        elif uripath == outdir or uripath.startswith(outdir+os.sep) or uripath.startswith(outdir+'/'):
             f["location"] = file_uri(path)
-        elif path == builder.outdir or path.startswith(builder.outdir+os.sep):
+        elif path == builder.outdir or path.startswith(builder.outdir+os.sep) or path.startswith(builder.outdir+'/'):
             f["location"] = builder.fs_access.join(outdir, path[len(builder.outdir) + 1:])
         elif not os.path.isabs(path):
             f["location"] = builder.fs_access.join(outdir, path)
@@ -413,7 +413,7 @@ def job(self,
 
                 def update_status_output_callback(
                         output_callbacks,  # type: Callable[[List[Dict[Text, Any]], Text], None]
-                        jobcachelock,      # type: IO[Any] 
+                        jobcachelock,      # type: IO[Any]
                         outputs,           # type: List[Dict[Text, Any]]
                         processStatus      # type: Text
                         ):  # type: (...) -> None

cwltool/extensions.yml

@@ -135,3 +135,22 @@ $graph:
         "_type": "@vocab"
     - name: networkAccess
       type: [boolean, string]
+
+- name: ProcessGenerator
+  type: record
+  inVocab: true
+  extends: cwl:Process
+  documentRoot: true
+  fields:
+    - name: class
+      jsonldPredicate:
+        "_id": "@type"
+        "_type": "@vocab"
+      type: string
+    - name: run
+      type: [string, cwl:Process]
+      jsonldPredicate:
+        _id: "cwl:run"
+        _type: "@id"
+      doc: |
+        Specifies the process to run.

cwltool/job.py

@@ -543,7 +543,7 @@ def create_file_and_add_volume(self,
             tmp_dir, tmp_prefix = os.path.split(tmpdir_prefix)
             new_file = os.path.join(
                 tempfile.mkdtemp(prefix=tmp_prefix, dir=tmp_dir),
-                os.path.basename(volume.resolved))
+                os.path.basename(volume.target))
         writable = True if volume.type == "CreateWritableFile" else False
         if secret_store:
             contents = secret_store.retrieve(volume.resolved)

cwltool/main.py

@@ -51,6 +51,7 @@
                       scandeps, shortname, use_custom_schema,
                       use_standard_schema, CWL_IANA)
 from .workflow import Workflow
+from .procgenerator import ProcessGenerator
 from .provenance import ResearchObject
 from .resolver import ga4gh_tool_registries, tool_resolver
 from .secrets import SecretStore
@@ -851,17 +852,6 @@ def main(argsl=None,                   # type: Optional[List[str]]
 
         runtimeContext.secret_store = getdefault(runtimeContext.secret_store, SecretStore())
         runtimeContext.make_fs_access = getdefault(runtimeContext.make_fs_access, StdFsAccess)
-        try:
-            initialized_job_order_object = init_job_order(
-                job_order_object, args, tool, jobloader, stdout,
-                print_input_deps=args.print_input_deps,
-                relative_deps=args.relative_deps,
-                make_fs_access=runtimeContext.make_fs_access,
-                input_basedir=input_basedir,
-                secret_store=runtimeContext.secret_store,
-                input_required=input_required)
-        except SystemExit as err:
-            return err.code
 
         if not executor:
             if args.parallel:
@@ -875,6 +865,32 @@ def main(argsl=None,                   # type: Optional[List[str]]
 
         try:
             runtimeContext.basedir = input_basedir
+
+            if isinstance(tool, ProcessGenerator):
+                tfjob_order = {}  # type: MutableMapping[Text, Any]
+                if loadingContext.jobdefaults:
+                    tfjob_order.update(loadingContext.jobdefaults)
+                if job_order_object:
+                    tfjob_order.update(job_order_object)
+                tfout, tfstatus = real_executor(tool.embedded_tool, tfjob_order, runtimeContext)
+                if tfstatus != "success":
+                    raise WorkflowException("ProcessGenerator failed to generate workflow")
+                tool, job_order_object = tool.result(tfjob_order, tfout, runtimeContext)
+                if not job_order_object:
+                    job_order_object = None
+
+            try:
+                initialized_job_order_object = init_job_order(
+                    job_order_object, args, tool, jobloader, stdout,
+                    print_input_deps=args.print_input_deps,
+                    relative_deps=args.relative_deps,
+                    make_fs_access=runtimeContext.make_fs_access,
+                    input_basedir=input_basedir,
+                    secret_store=runtimeContext.secret_store,
+                    input_required=input_required)
+            except SystemExit as err:
+                return err.code
+
             del args.workflow
             del args.job_order
 

cwltool/procgenerator.py

@@ -0,0 +1,122 @@
+import copy
+from .process import Process, shortname
+from .load_tool import load_tool
+from schema_salad import validate
+from .errors import WorkflowException
+from .context import RuntimeContext, LoadingContext
+from typing import (Any, Callable, Dict, Generator, Iterable, List,
+                    Mapping, MutableMapping, MutableSequence,
+                    Optional, Tuple, Union, cast)
+from .loghandler import _logger
+from typing_extensions import Text  # pylint: disable=unused-import
+
+class ProcessGeneratorJob(object):
+    def __init__(self, procgenerator):
+        # type: (ProcessGenerator) -> None
+        self.procgenerator = procgenerator
+        self.jobout = None         # type: Optional[Dict[Text, Any]]
+        self.processStatus = None  # type: Optional[Text]
+
+    def receive_output(self, jobout, processStatus):
+        # type: (Dict[Text, Any], Text) -> None
+        self.jobout = jobout
+        self.processStatus = processStatus
+
+    def job(self,
+            job_order,         # type: Mapping[Text, Any]
+            output_callbacks,  # type: Callable[[Any, Any], Any]
+            runtimeContext     # type: RuntimeContext
+           ):  # type: (...) -> Generator[Any, None, None]
+        # FIXME: Declare base type for what Generator yields
+
+        try:
+            for tool in self.procgenerator.embedded_tool.job(
+                    job_order,
+                    self.receive_output,
+                    runtimeContext):
+                yield tool
+
+            while self.processStatus is None:
+                yield None
+
+            if self.processStatus != "success":
+                output_callbacks(self.jobout, self.processStatus)
+                return
+
+            if self.jobout is None:
+                raise WorkflowException("jobout should not be None")
+
+            created_tool, runinputs = self.procgenerator.result(job_order, self.jobout, runtimeContext)
+
+            for tool in created_tool.job(
+                    runinputs,
+                    output_callbacks,
+                    runtimeContext):
+                yield tool
+
+        except WorkflowException:
+            raise
+        except Exception as exc:
+            _logger.exception("Unexpected exception")
+            raise WorkflowException(Text(exc))
+
+
+class ProcessGenerator(Process):
+    def __init__(self,
+                 toolpath_object,      # type: MutableMapping[Text, Any]
+                 loadingContext        # type: LoadingContext
+    ):  # type: (...) -> None
+        super(ProcessGenerator, self).__init__(
+            toolpath_object, loadingContext)
+        self.loadingContext = loadingContext  # type: LoadingContext
+        try:
+            if isinstance(toolpath_object["run"], MutableMapping):
+                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 validate.ValidationException as vexc:
+            if loadingContext.debug:
+                _logger.exception("Validation exception")
+            raise WorkflowException(
+                u"Tool definition %s failed validation:\n%s" %
+                (toolpath_object["run"], validate.indent(str(vexc))))
+
+    def job(self,
+            job_order,         # type: Mapping[Text, Text]
+            output_callbacks,  # type: Callable[[Any, Any], Any]
+            runtimeContext     # type: RuntimeContext
+           ):  # type: (...) -> Generator[Any, None, None]
+        # FIXME: Declare base type for what Generator yields
+        return ProcessGeneratorJob(self).job(job_order, output_callbacks, runtimeContext)
+
+    def result(self,
+            job_order,      # type: Mapping[Text, Any]
+            jobout,         # type: Mapping[Text, Any]
+            runtimeContext  # type: RuntimeContext
+    ):  # type: (...) -> Tuple[Process, MutableMapping[Text, Any]]
+        try:
+            loadingContext = self.loadingContext.copy()
+            loadingContext.metadata = {}
+            embedded_tool = load_tool(
+                jobout["runProcess"]["location"], loadingContext)
+        except validate.ValidationException as vexc:
+            if runtimeContext.debug:
+                _logger.exception("Validation exception")
+            raise WorkflowException(
+                u"Tool definition %s failed validation:\n%s" %
+                (jobout["runProcess"], validate.indent(str(vexc))))
+
+        if "runInputs" in jobout:
+            runinputs = cast(MutableMapping[Text, Any], jobout["runInputs"])
+        else:
+            runinputs = cast(MutableMapping[Text, Any], copy.deepcopy(job_order))
+            for i in self.embedded_tool.tool["inputs"]:
+                if shortname(i["id"]) in runinputs:
+                    del runinputs[shortname(i["id"])]
+            if "id" in runinputs:
+                del runinputs["id"]
+
+        return embedded_tool, runinputs

cwltool/workflow.py

@@ -22,7 +22,7 @@
 from typing_extensions import Text  # pylint: disable=unused-import
 # move to a regular typing import when Python 3.3-3.6 is no longer supported
 
-from . import command_line_tool, context, expression
+from . import command_line_tool, context, expression, procgenerator
 from .command_line_tool import CallbackJob, ExpressionTool
 from .job import JobBase
 from .builder import content_limit_respected_read
@@ -56,6 +56,8 @@ def default_make_tool(toolpath_object,      # type: MutableMapping[Text, Any]
             return command_line_tool.ExpressionTool(toolpath_object, loadingContext)
         if toolpath_object["class"] == "Workflow":
             return Workflow(toolpath_object, loadingContext)
+        if toolpath_object["class"] == "ProcessGenerator":
+            return procgenerator.ProcessGenerator(toolpath_object, loadingContext)
 
     raise WorkflowException(
         u"Missing or invalid 'class' field in "
@@ -574,7 +576,7 @@ def make_workflow_step(self,
         return WorkflowStep(toolpath_object, pos, loadingContext, parentworkflowProv)
 
     def job(self,
-            job_order,         # type: Mapping[Text, Text]
+            job_order,         # type: Mapping[Text, Any]
             output_callbacks,  # type: Callable[[Any, Any], Any]
             runtimeContext     # type: RuntimeContext
            ):  # type: (...) -> Generator[Union[WorkflowJob, ExpressionTool.ExpressionJob, JobBase, CallbackJob, None], None, None]
@@ -596,7 +598,7 @@ def job(self,
         for wjob in job.job(builder.job, output_callbacks, runtimeContext):
             yield wjob
 
-    def visit(self, op):  # type: (Callable[[MutableMapping[Text, Any]], Any]) -> None 
+    def visit(self, op):  # type: (Callable[[MutableMapping[Text, Any]], Any]) -> None
         op(self.tool)
         for step in self.steps:
             step.visit(op)

tests/test_docker.py

@@ -31,3 +31,15 @@ def test_docker_incorrect_image_pull():
         ['--default-container', 'non-existant-weird-image',
          get_data("tests/wf/hello-workflow.cwl"), "--usermessage", "hello"])
     assert result_code != 0
+
+@needs_docker
+def test_docker_file_mount():
+    # test for bug in
+    # ContainerCommandLineJob.create_file_and_add_volume()
+    #
+    # the bug was that it would use the file literal contents as the
+    # temporary file name, which can easily result in a file name that
+    # is too long or otherwise invalid.  This test case uses ".."
+    result_code = main(
+        [get_data("tests/wf/literalfile.cwl"), get_data("tests/wf/literalfile-job.yml")])
+    assert result_code == 0

tests/test_ext.py

@@ -29,11 +29,17 @@ def test_listing_deep():
 
 @needs_docker
 def test_cwltool_options():
-    os.environ["CWLTOOL_OPTIONS"] = "--enable-ext"
-    params = [get_data('tests/wf/listing_deep.cwl'),
-              get_data('tests/listing-job.yml')]
-    assert main(params) == 0
-    del os.environ["CWLTOOL_OPTIONS"]
+    try:
+        opt = os.environ.get("CWLTOOL_OPTIONS")
+        os.environ["CWLTOOL_OPTIONS"] = "--enable-ext"
+        params = [get_data('tests/wf/listing_deep.cwl'),
+                  get_data('tests/listing-job.yml')]
+        assert main(params) == 0
+    finally:
+        if opt is not None:
+            os.environ["CWLTOOL_OPTIONS"] = opt
+        else:
+            del os.environ["CWLTOOL_OPTIONS"]
 
 @needs_docker
 def test_listing_shallow():

tests/test_procgenerator.py

@@ -0,0 +1,28 @@
+import pytest
+import os
+from .util import get_data, windows_needs_docker
+from cwltool.main import main
+
+@windows_needs_docker
+def test_missing_enable_ext():
+    # Requires --enable-ext and --enable-dev
+    try:
+        opt = os.environ.get("CWLTOOL_OPTIONS")
+
+        if "CWLTOOL_OPTIONS" in os.environ:
+            del os.environ["CWLTOOL_OPTIONS"]
+        assert main([get_data('tests/wf/generator/zing.cwl'),
+                     "--zing", "zipper"]) == 1
+
+        assert main(["--enable-ext", "--enable-dev",
+                     get_data('tests/wf/generator/zing.cwl'),
+                     "--zing", "zipper"]) == 0
+
+        os.environ["CWLTOOL_OPTIONS"] = "--enable-ext --enable-dev"
+        assert main([get_data('tests/wf/generator/zing.cwl'),
+                     "--zing", "zipper"]) == 0
+    finally:
+        if opt is not None:
+            os.environ["CWLTOOL_OPTIONS"] = opt
+        else:
+            del os.environ["CWLTOOL_OPTIONS"]

tests/wf/generator/pytoolgen.cwl

@@ -0,0 +1,31 @@
+cwlVersion: v1.0
+$namespaces:
+  cwltool: "http://commonwl.org/cwltool#"
+class: ProcessGenerator
+inputs:
+  script: string
+  dir: Directory
+outputs: {}
+run:
+  class: CommandLineTool
+  inputs:
+    script: string
+    dir: Directory
+  outputs:
+    runProcess:
+      type: File
+      outputBinding:
+        glob: main.cwl
+  requirements:
+    InlineJavascriptRequirement: {}
+    cwltool:LoadListingRequirement:
+      loadListing: shallow_listing
+    InitialWorkDirRequirement:
+      listing: |
+        ${
+         var v = inputs.dir.listing;
+         v.push({entryname: "inp.py", entry: inputs.script});
+         return v;
+        }
+  arguments: [python, inp.py]
+  stdout: main.cwl

tests/wf/generator/zing.cwl

@@ -0,0 +1,13 @@
+#!/usr/bin/env cwltool
+{cwl:tool: pytoolgen.cwl, script: {$include: "#attachment-1"}, dir: {class: Directory, location: .}}
+--- |
+import os
+import sys
+print("""
+cwlVersion: v1.0
+class: CommandLineTool
+inputs:
+  zing: string
+outputs: {}
+arguments: [echo, $(inputs.zing)]
+""")

tests/wf/literalfile-job.yml

@@ -0,0 +1,3 @@
+a1:
+  class: File
+  contents: ".."