override task runtime values from input JSON

Author: mlin

WDL/CLI.py

@@ -971,6 +971,7 @@ def bold(line):
         ans.append(bold(f"  {str(b.value.type)} {b.name}"))
         add_wrapped_parameter_meta(target, b.name, ans)
     optional_inputs = target.available_inputs.subtract(target.required_inputs)
+    optional_inputs = optional_inputs.filter(lambda b: not b.value.name.startswith("_"))
     if target.inputs is None:
         # if the target doesn't have an input{} section (pre WDL 1.0), exclude
         # declarations bound to a non-constant expression (heuristic)

WDL/Tree.py

@@ -316,6 +316,11 @@ def available_inputs(self) -> Env.Bindings[Decl]:
         Each input is at the top level of the Env, with no namespace.
         """
         ans = Env.Bindings()
+
+        if self.effective_wdl_version not in ["draft-2", "1.0"]:
+            # synthetic placeholder to expose runtime & hints overrides
+            ans = ans.bind("_runtime", Decl(self.pos, Type.Any(), "_runtime"))
+
         for decl in reversed(self.inputs if self.inputs is not None else self.postinputs):
             ans = ans.bind(decl.name, decl)
         return ans
@@ -333,7 +338,7 @@ def required_inputs(self) -> Env.Bindings[Decl]:
         for b in reversed(list(self.available_inputs)):
             assert isinstance(b, Env.Binding)
             d: Decl = b.value
-            if d.expr is None and d.type.optional is False:
+            if d.expr is None and d.type.optional is False and not d.name.startswith("_"):
                 ans = Env.Bindings(b, ans)
         return ans
 

WDL/__init__.py

@@ -242,17 +242,33 @@ def values_from_json(
             key2 = key
             if namespace and key.startswith(namespace):
                 key2 = key[len(namespace) :]
-            if key2 not in available:
-                # attempt to simplify <call>.<subworkflow>.<input>
+
+            ty = None
+            if key2 in available:
+                ty = available[key2]
+            else:
                 key2parts = key2.split(".")
-                if len(key2parts) == 3 and key2parts[0] and key2parts[1] and key2parts[2]:
+
+                runtime_idx = next(
+                    (i for i, term in enumerate(key2parts) if term in ("runtime", "hints")), -1
+                )
+                if (
+                    runtime_idx >= 0
+                    and len(key2parts) > (runtime_idx + 1)
+                    and ".".join(key2parts[:runtime_idx] + ["_runtime"]) in available
+                ):
+                    # allow arbitrary keys for runtime/hints
+                    ty = Type.Any()
+                elif len(key2parts) == 3 and key2parts[0] and key2parts[1] and key2parts[2]:
+                    # attempt to simplify <call>.<subworkflow>.<input> from old Cromwell JSON
                     key2 = ".".join([key2parts[0], key2parts[2]])
-            try:
-                ty = available[key2]
-            except KeyError:
+                    if key2 in available:
+                        ty = available[key2]
+            if not ty:
                 raise Error.InputError("unknown input/output: " + key) from None
             if isinstance(ty, Tree.Decl):
                 ty = ty.type
+
             assert isinstance(ty, Type.Base)
             try:
                 ans = ans.bind(key2, Value.from_json(ty, values_json[key]))

WDL/runtime/task.py

@@ -167,7 +167,7 @@ def run_local_task(
                 # evaluate runtime fields
                 stdlib = InputStdLib(task.effective_wdl_version, logger, container)
                 container.runtime_values = _eval_task_runtime(
-                    cfg, logger, task, container, container_env, stdlib
+                    cfg, logger, task, posix_inputs, container, container_env, stdlib
                 )
 
                 # interpolate command
@@ -409,6 +409,7 @@ def _eval_task_runtime(
     cfg: config.Loader,
     logger: logging.Logger,
     task: Tree.Task,
+    inputs: Env.Bindings[Value.Base],
     container: TaskContainer,
     env: Env.Bindings[Value.Base],
     stdlib: StdLib.Base,
@@ -421,8 +422,10 @@ def _eval_task_runtime(
             runtime_values[key] = Value.Int(v)
         else:
             raise Error.InputError(f"invalid default runtime setting {key} = {v}")
-    for key, expr in task.runtime.items():
+    for key, expr in task.runtime.items():  # evaluate expressions in source code
         runtime_values[key] = expr.eval(env, stdlib)
+    for b in inputs.enter_namespace("runtime"):
+        runtime_values[b.name] = b.value  # input overrides
     logger.debug(_("runtime values", **dict((key, str(v)) for key, v in runtime_values.items())))
     ans = {}
 

WDL/runtime/workflow.py

@@ -371,7 +371,14 @@ def _do_job(
             assert isinstance(job.node.callee, (Tree.Task, Tree.Workflow))
             callee_inputs = job.node.callee.available_inputs
             call_inputs = call_inputs.map(
-                lambda b: Env.Binding(b.name, b.value.coerce(callee_inputs[b.name].type))
+                lambda b: Env.Binding(
+                    b.name,
+                    (
+                        b.value.coerce(callee_inputs[b.name].type)
+                        if b.name in callee_inputs
+                        else b.value
+                    ),
+                )
             )
             # check input files against whitelist
             disallowed_filenames = _fspaths(call_inputs) - self.filename_whitelist

tests/test_7runner.py

@@ -678,6 +678,43 @@ def test_directory(self, capture):
         logs += new_logs
 
 
+class RuntimeOverride(RunnerTestCase):
+    def test_runtime_override(self):
+        wdl = """
+        version development
+        workflow w {
+            input {
+                String who
+            }
+            call t {
+                input:
+                    who = who
+            }
+        }
+        task t {
+            input {
+                String who
+            }
+            command {
+                cp /etc/issue issue
+                echo "Hello, ~{who}!"
+            }
+            output {
+                String msg = read_string(stdout())
+                String issue = read_string("issue")
+            }
+            runtime {
+                docker: "ubuntu:20.04"
+            }
+        }
+        """
+        outp = self._run(wdl, {
+            "who": "Alice",
+            "t.runtime.docker": "ubuntu:20.10"
+        })
+        assert "20.10" in outp["t.issue"]
+
+
 class MiscRegressionTests(RunnerTestCase):
     def test_repeated_file_rewriting(self):
         wdl = """