Testing evaluating requirements and hints eagerly

Author: kinow

cwltool/workflow.py

@@ -35,6 +35,7 @@
     aslist,
 )
 from .workflow_job import WorkflowJob
+from cwl_utils import expression
 
 
 def default_make_tool(
@@ -156,6 +157,43 @@ def job(
         output_callbacks: Optional[OutputCallbackType],
         runtimeContext: RuntimeContext,
     ) -> JobsGeneratorType:
+        # TODO: This is not very efficient and could be improved.
+        #
+        # See issue #1330. We needed to eager-load the requirements
+        # at the workflow level, so that it was not re-evaluated at
+        # each step level (since a command-line tool, for instance,
+        # won't have the inputs from args, but instead will have the
+        # empty/default inputs of a workflow step).
+        #
+        # The solution below evaluates the requirements and hints for
+        # the workflow, for each step of the workflow, and for the
+        # embedded tool (command-line or expression tools) of each
+        # step.
+        #
+        # This prevents evaluation at the step level (i.e. the values
+        # were already loaded earlier/eagerly).
+        def _fix_hints_and_requirements(
+            hints_or_requirements: List[CWLObjectType],
+        ) -> None:
+            for hint_or_requirement in hints_or_requirements:
+                for key, value in hint_or_requirement.items():
+                    hint_or_requirement[key] = expression.do_eval(
+                        ex=value,
+                        jobinput=job_order,
+                        requirements=self.requirements,
+                        outdir=runtimeContext.outdir,
+                        tmpdir=runtimeContext.tmpdir,
+                        resources={},
+                        context=None,
+                        timeout=runtimeContext.eval_timeout,
+                    )
+
+        for key in ["hints", "requirements"]:
+            _fix_hints_and_requirements(getattr(self, key))
+            for step in self.steps:
+                _fix_hints_and_requirements(getattr(step, key))
+                _fix_hints_and_requirements(getattr(step.embedded_tool, key))
+
         builder = self._init_job(job_order, runtimeContext)
 
         if runtimeContext.research_obj is not None: