diff --git a/src/toil/cwl/cwltoil.py b/src/toil/cwl/cwltoil.py index ab8ea5beb1..3a54659e01 100644 --- a/src/toil/cwl/cwltoil.py +++ b/src/toil/cwl/cwltoil.py @@ -155,6 +155,10 @@ # directories in the current directory and leave them there. DEFAULT_TMPDIR_PREFIX = os.path.join(DEFAULT_TMPDIR, "tmp") +# Memory and disk for internal CWL interpreter jobs. +INTERPRETER_JOB_MEMORY = "1GiB" +INTERPRETER_JOB_DISK = "1MiB" + def cwltoil_was_removed() -> None: """Complain about deprecated entrypoint.""" @@ -179,6 +183,10 @@ class UnresolvedDict(dict[Any, Any]): """Tag to indicate a dict contains promises that must be resolved.""" +# CWL Jobs have inputs and outputs and sometimes we have to do a process of resolution to figure out +# what inputs get values from what outputs. Sometimes, collectively we call these inputs and outputs 'ports'. + + class SkipNull: """ Internal sentinel object. @@ -524,7 +532,7 @@ def pick_value(self, values: list[str | SkipNull] | Any) -> Any: ) -class StepValueFrom: +class ValueFrom: """ A workflow step input which has a valueFrom expression attached to it. @@ -533,7 +541,7 @@ class StepValueFrom: """ def __init__( - self, expr: str, source: Any, req: list[CWLObjectType], container_engine: str + self, expr: str, source: Any | None, req: list[CWLObjectType], container_engine: str, inputs_override: CWLObjectType | None = None ): """ Instantiate an object to carry all know about this valueFrom expression. @@ -548,11 +556,12 @@ def __init__( self.context = None self.req = req self.container_engine = container_engine + self.inputs_override = inputs_override def __repr__(self) -> str: """Allow for debug printing.""" - return f"StepValueFrom({self.expr}, {self.source}, {self.req}, {self.container_engine})" + return f"ValueFrom({self.expr}, {self.source}, {self.req}, {self.container_engine})" def eval_prep( self, step_inputs: CWLObjectType, file_store: AbstractFileStore @@ -560,7 +569,7 @@ def eval_prep( """ Resolve the contents of any file in a set of inputs. - The inputs must be associated with the StepValueFrom object's self.source. + The inputs must be associated with the ValueFrom object's self.source. Called when loadContents is specified. @@ -591,7 +600,7 @@ def resolve(self) -> Any: :return: object that will serve as expression context """ - self.context = self.source.resolve() + self.context = self.source.resolve() if self.source is not None else None return self.context def do_eval(self, inputs: CWLObjectType) -> Any: @@ -603,7 +612,7 @@ def do_eval(self, inputs: CWLObjectType) -> Any: """ return cwl_utils.expression.do_eval( self.expr, - inputs, + self.inputs_override if self.inputs_override is not None else inputs, self.req, None, None, @@ -664,7 +673,7 @@ def resolve(self) -> Any: def resolve_dict_w_promises( - dict_w_promises: UnresolvedDict | CWLObjectType | dict[str, str | StepValueFrom], + dict_w_promises: UnresolvedDict | CWLObjectType | dict[str, str | ValueFrom], file_store: AbstractFileStore | None = None, ) -> CWLObjectType: """ @@ -684,7 +693,7 @@ def resolve_dict_w_promises( result: CWLObjectType = {} for k, v in dict_w_promises.items(): - if isinstance(v, StepValueFrom): + if isinstance(v, ValueFrom): if file_store: v.eval_prep(first_pass_results, file_store) result[k] = v.do_eval(inputs=first_pass_results) @@ -3139,7 +3148,7 @@ def __init__( conditional: Conditional | None, ): """Store our context for later execution.""" - super().__init__(cores=1, memory="1GiB", disk="1MiB", local=True) + super().__init__(cores=1, memory=INTERPRETER_JOB_MEMORY, disk=INTERPRETER_JOB_DISK, local=True) self.step = step self.cwljob = cwljob self.runtime_context = runtime_context @@ -3299,7 +3308,7 @@ def __init__( outputs: Promised[CWLObjectType | list[CWLObjectType]], ): """Collect our context for later gathering.""" - super().__init__(cores=1, memory="1GiB", disk="1MiB", local=True) + super().__init__(cores=1, memory=INTERPRETER_JOB_MEMORY, disk=INTERPRETER_JOB_DISK, local=True) self.step = step self.outputs = outputs @@ -3341,13 +3350,316 @@ def sn(n: Mapping[str, Any] | str) -> str: return outobj +class CWLLoopAccumulate(Job): + """ + Extend a running cwltool:Loop accumulator with one iteration's outputs. + Used by `outputMethod: all_iterations` to build the arrays of each output across all iterations. + """ + + def __init__( + self, + previous_accumulation: Promised[dict[str, list[CWLObjectType]]] | dict[str, list[CWLObjectType]], + iteration_outputs: Promised[CWLObjectType], + output_keys: list[str], + ): + """ + Store the promises needed to extend the accumulator by one iteration. + + :param previous_accumulation: the accumulated outputs from all prior iterations, as a dict mapping output key to list; a plain dict for iteration 0, a Promise thereafter + :param iteration_outputs: promise for the current iteration's output dict + :param output_keys: short-form names of the output ports to accumulate + """ + super().__init__(cores=1, memory=INTERPRETER_JOB_MEMORY, disk=INTERPRETER_JOB_DISK, local=True) + self.previous_accumulation = previous_accumulation + self.iteration_outputs = iteration_outputs + self.output_keys = output_keys + + def run(self, file_store: AbstractFileStore) -> dict[str, list[CWLObjectType]]: + """Append the current iteration's outputs to the prior accumulation for each output key.""" + previous = unwrap(self.previous_accumulation) + this = cast(dict[str, Any], unwrap(self.iteration_outputs)) + result: dict[str, list[Any]] = {} + for k in self.output_keys: + result[k] = previous.get(k, []) + [this.get(k)] + return result + + +class CWLLoop(Job): + """ + Implement the cwltool:Loop workflow-step extension. + + Evaluates `loopWhen` each iteration; if true, spawns the embedded tool + as a child and chains a successor CWLLoop as a follow-on. Terminates when + `loopWhen` returns false, emitting the last iteration's outputs + (outputMethod: last) or the accumulated arrays of each output + (outputMethod: all). + + Validation is the responsibility of cwltool's own loop_checker. + + Loops nest naturally because each iteration spawns an independent + embedded tool job. To combine a loop with scatter, use an intervening + subworkflow — cwltool's loop_checker rejects a step that carries + both requirements directly. + """ + def __init__( + self, + step: cwltool.workflow.WorkflowStep, + cwljob: UnresolvedDict | CWLObjectType, + runtime_context: cwltool.context.RuntimeContext, + parent_name: str | None, + iteration_limit: int, + iteration: int = 0, + previous_outputs: Promised[CWLObjectType] | None = None, + previous_accumulation: Promised[CWLObjectType] | CWLObjectType | None = None, + ): + """ + Store context needed to evaluate one iteration of the loop. + + :param step: the WorkflowStep carrying the cwltool:Loop requirement + :param cwljob: the input object for this iteration, as an UnresolvedDict or plain dict + :param runtime_context: Toil CWL runtime context (filesystem, container engine, etc.) + :param parent_name: human-readable name prefix used for child job naming + :param iteration_limit: maximum number of iterations before raising an error + :param iteration: zero-based index of the current iteration we would execute + :param previous_outputs: promise for the previous iteration's output dict; + should be None for iteration 0 + :param previous_accumulation: accumulated outputs from all prior iterations; + used for outputMethod: all_iterations; should be None for iteration 0 + """ + super().__init__(cores=1, memory=INTERPRETER_JOB_MEMORY, disk=INTERPRETER_JOB_DISK, local=True) + self.step = step + self.cwljob = cwljob + self.runtime_context = runtime_context + self.parent_name = parent_name + self.iteration_limit = iteration_limit + self.iteration = iteration + self.previous_outputs = previous_outputs + self.previous_accumulation = previous_accumulation + + def build_next_inputs(self, cwljob: CWLObjectType, body_followon: Job, loop_inputs: list[CWLObjectType]) -> UnresolvedDict: + """ + Build the input object for the next iteration by applying the + ``loop`` rebinding rules from the cwltool:Loop requirement. + + Each LoopInput record names a step input and describes how its + value is updated: + + - ``loopSource`` pulls from one or more of the just-finished + iteration's outputs (with optional ``linkMerge`` and ``pickValue`` + for merging multiple sources) + - ``default`` supplies a fallback if the source is absent + - ``valueFrom`` runs a CWL expression to compute the final value, + with ``self`` set to the resolved source and ``inputs`` set to + this iteration's joborder + + Step inputs not named in any LoopInput carry their current value + forward unchanged. + """ + loop_by_id: dict[str, CWLObjectType] = { + shortname(cast(str, li["id"])): li for li in loop_inputs + } + step_id_short = shortname(cast(str, self.step.tool["id"])) + container_engine = get_container_engine(self.runtime_context) + + result: dict[str, Any] = {} + for k, v in cwljob.items(): + li = loop_by_id.get(k) + if li is None: + # This input isn't rebound by the loop; carry its current value forward unchanged. + result[k] = JustAValue(v) + continue + + source: Any = None + if "outputSource" in li: + raw_sources = aslist(li["outputSource"]) + if len(raw_sources) == 1 and not li.get("linkMerge") and not li.get("pickValue"): + source = JustAValue( + body_followon.rv(shortname(cast(str, raw_sources[0]))) + ) + else: + # ResolveSource expects a CWL input-like dict with an + # "outputSource" key, not the raw LoopInput object (li), + # because li uses "outputSource" to name loop outputs rather + # than workflow-level sources. We forge a minimal input dict + # with just the keys ResolveSource needs. + rs_input: dict[str, Any] = { + "outputSource": [shortname(cast(str, s)) for s in raw_sources], + } + if li.get("linkMerge"): + rs_input["linkMerge"] = li["linkMerge"] + if li.get("pickValue"): + rs_input["pickValue"] = li["pickValue"] + source = ResolveSource( + name=f"{step_id_short}/{k}", + input=rs_input, + source_key="outputSource", + promises={shortname(cast(str, s)): body_followon for s in raw_sources}, + ) + + if "default" in li: + # Shallow-copy the default so mutations in one iteration don't affect subsequent ones. + source = DefaultWithSource(copy.copy(li["default"]), source) + + if "valueFrom" in li: + # The presence of a valueFrom expression requires that the StepInputExpressionRequirement be present in the step's or workflow's requirements. + # https://github.com/common-workflow-language/cwltool/blob/60dfe96952119f472a57524c19ed214e831b21a4/cwltool/extensions-v1.2.yml#L175-L176 + # Note: the step's requirements have already included the workflow's requirements + # https://github.com/common-workflow-language/cwltool/blob/1bf74499ca1c4a5f98e7cffb0ad4aa89aa98cb9e/cwltool/workflow.py#L207 + if not bool(self.step.get_requirement("StepInputExpressionRequirement")[0]): + # Raise the same error cwltool does: + # https://github.com/common-workflow-language/cwltool/blob/1bf74499ca1c4a5f98e7cffb0ad4aa89aa98cb9e/cwltool/workflow_job.py#L622 + raise cwl_utils.errors.WorkflowException( + "Workflow step contains valueFrom but StepInputExpressionRequirement not in requirements" + ) + result[k] = ValueFrom( + expr=cast(str, li["valueFrom"]), + source=source, + req=self.step.requirements, + container_engine=container_engine, + inputs_override={shortname(k): v for k, v in cwljob.items()}, + ) + elif source is not None: + result[k] = source + else: + result[k] = JustAValue(v) + return UnresolvedDict(result) + + def run(self, file_store: AbstractFileStore) -> CWLObjectType | Promised[CWLObjectType]: + """ + Evaluate the loop condition and either spawn the next iteration or return the final output. + + If the `loopWhen` expression evaluates to true and the iteration limit has not been + reached, spawns the embedded tool as a child job and chains a new CWLLoop as a + follow-on. If false, returns the last iteration's outputs directly + (outputMethod: last_iteration) or the accumulated per-port arrays + (outputMethod: all_iterations). + """ + cwljob = resolve_dict_w_promises(self.cwljob, file_store) + + # Validate that loop is present - CWLLoop should only be scheduled for + # loop steps, so this is a programming error if missing. We don't + # validate when separately because cwltool's loop_checker ensures it + # is always present when loop is present. outputMethod uses .get() since + # it has a legitimate default of "last_iteration". + if "loop" not in self.step.tool: + raise RuntimeError( + "CWLLoop scheduled for a step without a cwltool:Loop requirement" + ) + + loop_inputs = cast(list[CWLObjectType], self.step.tool["loop"]) + when_expression = cast(str, self.step.tool["when"]) + output_method = cast(str, self.step.tool.get("outputMethod", "last_iteration")) + + when_val = cwl_utils.expression.do_eval( + when_expression, + {shortname(k): v for k, v in cwljob.items()}, + self.step.requirements, + None, + None, + {}, + container_engine=get_container_engine(self.runtime_context), + ) + if not isinstance(when_val, bool): + raise cwl_utils.errors.WorkflowException( + f"'loopWhen' expression {when_expression!r} did not evaluate " + "to a boolean" + ) + + if when_val is False: + if self.iteration == 0: + output_keys = [shortname(cast(str, o["id"])) for o in self.step.tool["outputs"]] + return {k: ([] if output_method == "all_iterations" else None) for k in output_keys} + if output_method == "all_iterations": + # By this point we've completed at least one iteration, so previous_accumulation is set. + assert self.previous_accumulation is not None + return unwrap(self.previous_accumulation) + # By this point we've completed at least one iteration, so previous_outputs is set. + assert self.previous_outputs is not None + return unwrap(self.previous_outputs) + + if self.iteration >= self.iteration_limit: + raise cwl_utils.errors.WorkflowException( + f"cwltool:Loop on step {shortname(self.step.tool['id'])!r} " + f"exceeded the iteration limit of {self.iteration_limit}. " + "Raise the limit via --cwl-loop-iteration-limit, or check " + "that `loopWhen` is intended to eventually become false." + ) + + body_job, body_followon = makeJob( + tool=self.step.embedded_tool, + jobobj=UnresolvedDict({k: JustAValue(v) for k, v in cwljob.items()}), + runtime_context=self.runtime_context, + parent_name=f"{self.parent_name}.iteration{self.iteration}", + conditional=None, + ) + + self.addChild(body_job) + + # Declare before the if/else so that mypy can see that they are always defined + next_loop_pred: CWLLoopAccumulate | Job + next_accumulation: Promised[CWLObjectType] | None + + # In all_iterations mode, we need to run CWLLoopAccumulate after the + # body job to append this iteration's outputs to the running accumulator + # before the next CWLLoop iteration runs. next_accumulation carries the + # promise for the updated accumulator into the next iteration. + if output_method == "all_iterations": + output_keys = [shortname(cast(str, o["id"])) for o in self.step.tool["outputs"]] + # On iteration 0 there is no prior accumulation, so seed with empty lists. + prev_acc_arg: Promised[dict[str, list[CWLObjectType]]] | dict[str, list[CWLObjectType]] = ( + cast(dict[str, list[CWLObjectType]], {k: [] for k in output_keys}) + if self.iteration == 0 + else cast(Promised[dict[str, list[CWLObjectType]]], self.previous_accumulation) + ) + accumulated_loops = CWLLoopAccumulate( + previous_accumulation=prev_acc_arg, + iteration_outputs=body_followon.rv(), + output_keys=output_keys, + ) + body_followon.addFollowOn(accumulated_loops) + # The next CWLLoop runs after CWLLoopAccumulate, so it sees the + # updated accumulator as its predecessor. + next_loop_pred = accumulated_loops + next_accumulation = accumulated_loops.rv() + + # In last_iteration mode we don't need to accumulate anything - + # the next CWLLoop runs directly after the body job, and + # next_accumulation stays None since we only care about the final + # iteration's outputs. + elif output_method == "last_iteration": + next_loop_pred = body_followon + next_accumulation = None + + else: + raise cwl_utils.errors.WorkflowException( + f"Unsupported cwltool:Loop outputMethod {output_method!r}" + ) + + next_inputs = self.build_next_inputs(cwljob, body_followon, loop_inputs) + + next_loop = CWLLoop( + step=self.step, + cwljob=next_inputs, + runtime_context=self.runtime_context, + parent_name=self.parent_name, + iteration_limit=self.iteration_limit, + iteration=self.iteration + 1, + previous_outputs=body_followon.rv(), + previous_accumulation=next_accumulation, + ) + + # Attach next loop to its predecessor, which depends on the output method + next_loop_pred.addFollowOn(next_loop) + + return next_loop.rv() + class SelfJob(Job): """Fake job object to facilitate implementation of CWLWorkflow.run().""" def __init__(self, j: "CWLWorkflow", v: CWLObjectType): """Record the workflow and dictionary.""" - super().__init__(cores=1, memory="1GiB", disk="1MiB") + super().__init__(cores=1, memory=INTERPRETER_JOB_MEMORY, disk=INTERPRETER_JOB_DISK) self.j = j self.v = v @@ -3467,7 +3779,7 @@ def run( if stepinputs_fufilled: logger.debug("Ready to make job for workflow step %s", step_id) jobobj: dict[ - str, ResolveSource | DefaultWithSource | StepValueFrom + str, ResolveSource | DefaultWithSource | ValueFrom ] = {} for inp in step.tool["inputs"]: @@ -3487,7 +3799,7 @@ def run( ) if "valueFrom" in inp and "scatter" not in step.tool: - jobobj[key] = StepValueFrom( + jobobj[key] = ValueFrom( inp["valueFrom"], jobobj.get(key, JustAValue(None)), self.cwlwf.requirements, @@ -3504,9 +3816,28 @@ def run( requirements=self.cwlwf.requirements, container_engine=get_container_engine(self.runtime_context), ) + + # Declare types for mypy so it can see that they are always defined in the if/else below + wfjob: CWLLoop | CWLScatter | CWLWorkflow | CWLJob | CWLJobWrapper + followOn: CWLLoop | CWLGather | ResolveIndirect | CWLJob | CWLJobWrapper + + if "loop" in step.tool: + wfjob = CWLLoop( + step, + UnresolvedDict(jobobj), + self.runtime_context, + parent_name=parent_name, + iteration_limit=getattr(self.runtime_context, "cwl_loop_iteration_limit", 1000), + ) + followOn = wfjob + logger.debug( + "Is loop with job %s and follow-on %s", + wfjob, + followOn, + ) - if "scatter" in step.tool: - wfjob: CWLScatter | CWLWorkflow | CWLJob | CWLJobWrapper = ( + elif "scatter" in step.tool: + wfjob = ( CWLScatter( step, UnresolvedDict(jobobj), @@ -3515,9 +3846,7 @@ def run( conditional=conditional, ) ) - followOn: ( - CWLGather | ResolveIndirect | CWLJob | CWLJobWrapper - ) = CWLGather(step, wfjob.rv()) + followOn = CWLGather(step, wfjob.rv()) wfjob.addFollowOn(followOn) logger.debug( "Is scatter with job %s and follow-on %s", @@ -4326,6 +4655,7 @@ def main(args: list[str] | None = None, stdout: TextIO = sys.stdout) -> int: runtime_context.outdir = outdir setattr(runtime_context, "cwl_default_ram", options.cwl_default_ram) setattr(runtime_context, "cwl_min_ram", options.cwl_min_ram) + setattr(runtime_context, "cwl_loop_iteration_limit", options.cwl_loop_iteration_limit) runtime_context.move_outputs = "leave" runtime_context.rm_tmpdir = options.rm_tmpdir runtime_context.streaming_allowed = not options.disable_streaming diff --git a/src/toil/options/cwl.py b/src/toil/options/cwl.py index 8fc6b5eddb..a91e120b40 100644 --- a/src/toil/options/cwl.py +++ b/src/toil/options/cwl.py @@ -458,3 +458,13 @@ def add_cwl_options(parser: ArgumentParser, suppress: bool = True) -> None: " Not to be confused with --caching.", dest="cachedir", ) + + parser.add_argument( + "--cwl-loop-iteration-limit", + type=int, + default=1000, + help=suppress_help + or "Maximum number of iterations allowed for a single cwltool:Loop " + "step before Toil aborts. Guards against runaway loopWhen expressions.", + dest="cwl_loop_iteration_limit", + ) diff --git a/src/toil/test/__init__.py b/src/toil/test/__init__.py index feb491514f..a1426bfc8c 100644 --- a/src/toil/test/__init__.py +++ b/src/toil/test/__init__.py @@ -84,6 +84,12 @@ def get_data(filename: str) -> AbstractContextManager[Path]: return _fallback_get_data(filename) +def get_cwltool_data(filename: str) -> AbstractContextManager[Path]: + """Returns an absolute path for a test file from the cwltool package.""" + filename = os.path.normpath(filename) + return as_file(files("cwltool.tests") / filename) + + @pytest.mark.usefixtures("rootpath") class ToilTest(unittest.TestCase): """ diff --git a/src/toil/test/cwl/cwlTest.py b/src/toil/test/cwl/cwlTest.py index 712d553eec..e08978744d 100644 --- a/src/toil/test/cwl/cwlTest.py +++ b/src/toil/test/cwl/cwlTest.py @@ -51,7 +51,7 @@ from toil.fileStores.abstractFileStore import AbstractFileStore from toil.job import WorkerImportJob from toil.lib.threading import cpu_count -from toil.test import get_data +from toil.test import get_data, get_cwltool_data from toil.test import pneeds_aws_s3 as needs_aws_s3 from toil.test import pneeds_cwl as needs_cwl from toil.test import pneeds_docker as needs_docker @@ -566,6 +566,175 @@ def test_required_input_condition_protection(self, tmp_path: Path) -> None: with get_data("test/cwl/not_run_required_input.cwl") as cwl_file: with get_data("test/cwl/empty.json") as inputs_file: self._tester(cwl_file, inputs_file, {}, tmp_path) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_loop_single_variable(self, tmp_path: Path) -> None: + """cwltool:Loop with one rebound input runs to completion (outputMethod=last).""" + with get_cwltool_data("loop-ext/single-var-loop.cwl") as cwl_file: + with get_cwltool_data("loop-ext/single-var-loop-job.yml") as job_file: + self._tester(cwl_file, job_file, {"o1": 10}, tmp_path, main_args=["--enable-ext"]) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_loop_iteration_limit_exceeded(self, tmp_path: Path) -> None: + """--cwl-loop-iteration-limit aborts runaway loops.""" + from toil.cwl import cwltoil + with get_cwltool_data("loop-ext/single-var-loop.cwl") as cwl_file: + with get_cwltool_data("loop-ext/single-var-loop-job.yml") as job_file: + rc = cwltoil.main( + [ + "--enable-ext", + "--cwl-loop-iteration-limit=3", + f"--outdir={tmp_path}", + str(cwl_file), + str(job_file), + ] + ) + assert rc != 0 + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_loop_single_variable_no_iteration(self, tmp_path: Path) -> None: + """cwltool:Loop with loopWhen=false on entry skips the step and yields null.""" + with get_cwltool_data("loop-ext/single-var-loop-no-iteration.cwl") as cwl_file: + with get_cwltool_data("loop-ext/single-var-loop-job.yml") as job_file: + self._tester(cwl_file, job_file, {"o1": None}, tmp_path, main_args=["--enable-ext"]) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_loop_two_variables(self, tmp_path: Path) -> None: + """cwltool:Loop with two rebound inputs runs to completion (outputMethod=last).""" + with get_cwltool_data("loop-ext/two-vars-loop.cwl") as cwl_file: + with get_cwltool_data("loop-ext/two-vars-loop-job.yml") as job_file: + self._tester(cwl_file, job_file, {"o1": 10}, tmp_path, main_args=["--enable-ext"]) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_loop_all_output_method_no_iteration(self, tmp_path: Path) -> None: + """cwltool:Loop with outputMethod=all and loopWhen=false on entry yields empty array.""" + with get_cwltool_data("loop-ext/all-output-loop-no-iteration.cwl") as cwl_file: + with get_cwltool_data("loop-ext/single-var-loop-job.yml") as job_file: + self._tester(cwl_file, job_file, {"o1": []}, tmp_path, main_args=["--enable-ext"]) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_loop_with_all_output_method(self, tmp_path: Path) -> None: + """cwltool:Loop with outputMethod=all accumulates each iteration's outputs into an array.""" + with get_cwltool_data("loop-ext/all-output-loop.cwl") as cwl_file: + with get_cwltool_data("loop-ext/single-var-loop-job.yml") as job_file: + self._tester(cwl_file, job_file, {"o1": [2, 3, 4, 5, 6, 7, 8, 9, 10]}, tmp_path, main_args=["--enable-ext"]) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_loop_value_from(self, tmp_path: Path) -> None: + """cwltool:Loop with a LoopInput valueFrom expression: next input derived from previous joborder.""" + with get_cwltool_data("loop-ext/value-from-loop.cwl") as cwl_file: + with get_cwltool_data("loop-ext/two-vars-loop-job.yml") as job_file: + self._tester(cwl_file, job_file, {"o1": 10}, tmp_path, main_args=["--enable-ext"]) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_default_value_loop(self, tmp_path: Path) -> None: + """cwltool:Loop with a LoopInput default: fallback used when loopSource resolves to None.""" + with get_cwltool_data("loop-ext/default-value-loop.cwl") as cwl_file: + with get_cwltool_data("loop-ext/single-var-loop-job.yml") as job_file: + self._tester(cwl_file, job_file, {"o1": [8, 11, 14, 17, 20]}, tmp_path, main_args=["--enable-ext"]) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_multi_source_loop_input(self, tmp_path: Path) -> None: + """cwltool:Loop with multi-source loopSource and pickValue.""" + with get_cwltool_data("loop-ext/multi-source-loop.cwl") as cwl_file: + with get_cwltool_data("loop-ext/single-var-loop-job.yml") as job_file: + self._tester(cwl_file, job_file, {"o1": [2, 3, 4, 5, 8, 11, 14, 17, 20]}, tmp_path, main_args=["--enable-ext"]) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_nested_loops(self, tmp_path: Path) -> None: + """cwltool:Loop nested inside another cwltool:Loop, inner uses outputMethod=last, outer uses outputMethod=all.""" + with get_cwltool_data("loop-ext/loop-inside-loop.cwl") as cwl_file: + with get_cwltool_data("loop-ext/two-vars-loop-job.yml") as job_file: + self._tester(cwl_file, job_file, {"o1": [2, 3, 4]}, tmp_path, main_args=["--enable-ext"]) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_nested_loops_all(self, tmp_path: Path) -> None: + """cwltool:Loop nested inside another cwltool:Loop, both outputMethod=all.""" + with get_cwltool_data("loop-ext/loop-inside-loop-all.cwl") as cwl_file: + with get_cwltool_data("loop-ext/two-vars-loop-job.yml") as job_file: + self._tester(cwl_file, job_file, {"o1": [[2], [2, 3], [2, 3, 4]]}, tmp_path, main_args=["--enable-ext"]) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_loop_inside_scatter(self, tmp_path: Path) -> None: + """Scatter over a Workflow step whose embedded workflow contains a cwltool:Loop step.""" + with get_cwltool_data("loop-ext/loop-inside-scatter.cwl") as cwl_file: + with get_cwltool_data("loop-ext/loop-inside-scatter-job.yml") as job_file: + self._tester(cwl_file, job_file, {"o1": [10, 10, 10, 10, 10]}, tmp_path, main_args=["--enable-ext"]) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_loop_validate_fail_scatter(self, tmp_path: Path) -> None: + """cwltool:Loop combined with scatter on the same step is rejected at validation time.""" + with get_cwltool_data("loop-ext/invalid-loop-scatter.cwl") as cwl_file: + with get_cwltool_data("loop-ext/two-vars-loop-job.yml") as job_file: + with pytest.raises(ValidationException): + self._tester(cwl_file, job_file, {}, tmp_path, main_args=["--enable-ext"]) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_loop_validate_fail_when(self, tmp_path: Path) -> None: + """cwltool:Loop combined with when on the same step is rejected at validation time.""" + with get_cwltool_data("loop-ext/invalid-loop-when.cwl") as cwl_file: + with get_cwltool_data("loop-ext/two-vars-loop-job.yml") as job_file: + with pytest.raises(ValidationException): + self._tester(cwl_file, job_file, {}, tmp_path, main_args=["--enable-ext"]) + + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_loop_fail_non_boolean_loop_when(self, tmp_path: Path) -> None: + """A loopWhen expression that evaluates to a non-boolean fails at runtime.""" + from toil.cwl import cwltoil + from io import StringIO + with get_cwltool_data("loop-ext/invalid-non-boolean-loopWhen.cwl") as cwl_file: + with get_cwltool_data("loop-ext/two-vars-loop-job.yml") as job_file: + rc = cwltoil.main( + ["--enable-ext", f"--outdir={tmp_path}", str(cwl_file), str(job_file)], + stdout=StringIO(), + ) + assert rc != 0 + + # This test fails due to AssertionError, needs more looking into it. It is disabled for now. + @needs_cwl + @pytest.mark.cwl + @pytest.mark.cwl_small + def test_loop_value_from_fail_no_requirement(self, tmp_path: Path) -> None: + """cwltool:Loop with valueFrom but no StepInputExpressionRequirement is rejected.""" + from toil.cwl import cwltoil + from io import StringIO + with get_cwltool_data("loop-ext/invalid-value-from-loop-no-requirement.cwl") as cwl_file: + with get_cwltool_data("loop-ext/two-vars-loop-job.yml") as job_file: + rc = cwltoil.main( + ["--enable-ext", f"--outdir={tmp_path}", str(cwl_file), str(job_file)], + stdout=StringIO(), + ) + assert rc != 0 @needs_slurm @pytest.mark.slurm