pytest_wdl/executors/cromwell_local.py
def run_workflow(
        self,
        wdl_path: Path,
        inputs: Optional[dict] = None,
        expected: Optional[dict] = None,
        **kwargs,
    ) -> dict:
        """
        Run a WDL workflow on given inputs, and check that the output matches
        given expected values.

        Args:
            wdl_path: The WDL script to execute.
            inputs: Object that will be serialized to JSON and provided to Cromwell
                as the workflow inputs.
            expected: Dict mapping output parameter names to expected values.
            kwargs: Additional keyword arguments, mostly for debugging:
                * workflow_name: The name of the workflow in the WDL script. If None,
                    the name of the WDL script is used (without the .wdl extension).
                * inputs_file: Path to the Cromwell inputs file to use. Inputs are
                    written to this file only if it doesn't exist.
                * imports_file: Path to the WDL imports file to use. Imports are
                    written to this file only if it doesn't exist.
                * java_args: Additional arguments to pass to Java runtime.
                * cromwell_args: Additional arguments to pass to `cromwell run`.

        Returns:
            Dict of outputs.

        Raises:
            ExecutionFailedError: if there was an error executing Cromwell
            AssertionError: if the actual outputs don't match the expected outputs
        """
        target, is_task = get_target_name(
            wdl_path=wdl_path, import_dirs=self._import_dirs, **kwargs
        )

        if is_task:
            raise ValueError(
                "Cromwell cannot execute tasks independently of a workflow"
            )

        inputs_dict, inputs_file = read_write_inputs(
            inputs_file=kwargs.get("inputs_file"), inputs_dict=inputs, namespace=target
        )
        imports_file = self._get_workflow_imports(
            self._import_dirs, kwargs.get("imports_file")
        )

        inputs_arg = f"-i {inputs_file}" if inputs_file else ""
        imports_zip_arg = f"-p {imports_file}" if imports_file else ""
        java_args = kwargs.get("java_args", self.java_args) or ""
        cromwell_args = kwargs.get("cromwell_args", self._cromwell_args) or ""
        metadata_file = Path.cwd() / "metadata.json"

        cmd = (
            f"{self.java_bin} {java_args} -jar {self._cromwell_jar_file} run "
            f"-m {metadata_file} {cromwell_args} {inputs_arg} {imports_zip_arg} "
            f"{wdl_path}"
        )
        LOG.info(
            f"Executing cromwell command '{cmd}' with inputs "
            f"{json.dumps(inputs_dict, default=str)}"
        )

        exe = subby.run(cmd, raise_on_error=False)

        metadata = None

        if metadata_file.exists():
            with open(metadata_file, "rt") as inp:
                metadata = json.load(inp)

        if exe.ok:
            if metadata:
                assert metadata["status"] == "Succeeded"
                outputs = metadata["outputs"]
            else:
                LOG.warning(
                    f"Cromwell command completed successfully but did not generate "
                    f"a metadata file at {metadata_file}"
                )
                outputs = self._get_cromwell_outputs(exe.output)
        else:
            error_kwargs = {
                "executor": "cromwell",
                "target": target,
                "status": "Failed",
                "inputs": inputs_dict,
                "executor_stdout": exe.output,
                "executor_stderr": exe.error,
            }
            if metadata:
                error_kwargs = {
                    "executor": "cromwell",
                    "target": target,
                    "status": "Failed",
                    "inputs": inputs_dict,
                }
                self._parse_metadata_errors(
                    metadata, target=target, error_kwargs=error_kwargs
                )
            else:
                error_kwargs["msg"] = (
                    f"Cromwell command failed but did not generate a metadata "
                    f"file at {metadata_file}"
                )

            raise ExecutionFailedError(**error_kwargs)

        if expected:
            self._validate_outputs(outputs, expected, target)

        return outputs

        Run a WDL workflow on given inputs, and check that the output matches
        given expected values.

        Args:
            wdl_path: The WDL script to execute.
            inputs: Object that will be serialized to JSON and provided to Cromwell
                as the workflow inputs.
            expected: Dict mapping output parameter names to expected values.
            kwargs: Additional keyword arguments, mostly for debugging:
                * workflow_name: The name of the workflow in the WDL script. If None,
                    the name of the WDL script is used (without the .wdl extension).
                * inputs_file: Path to the Cromwell inputs file to use. Inputs are
                    written to this file only if it doesn't exist.
                * imports_file: Path to the WDL imports file to use. Imports are
                    written to this file only if it doesn't exist.
                * java_args: Additional arguments to pass to Java runtime.
                * cromwell_args: Additional arguments to pass to `cromwell run`.

        Returns:
            Dict of outputs.

        Raises:
            ExecutionFailedError: if there was an error executing Cromwell
            AssertionError: if the actual outputs don't match the expected outputs
        
Run a WDL workflow on given inputs, and check that the output matches
given expected values.
Run a WDL workflow on given inputs and check that the output matches given expected values .
def run_workflow(
        self,
        wdl_path: Path,
        inputs: Optional[dict] = None,
        expected: Optional[dict] = None,
        **kwargs,
    ) -> dict:
        target, is_task = get_target_name(
            wdl_path=wdl_path, import_dirs=self._import_dirs, **kwargs
        )
        if is_task:
            raise ValueError(
                "Cromwell cannot execute tasks independently of a workflow"
            )
        inputs_dict, inputs_file = read_write_inputs(
            inputs_file=kwargs.get("inputs_file"), inputs_dict=inputs, namespace=target
        )
        imports_file = self._get_workflow_imports(
            self._import_dirs, kwargs.get("imports_file")
        )
        inputs_arg = f"-i {inputs_file}" if inputs_file else ""
        imports_zip_arg = f"-p {imports_file}" if imports_file else ""
        java_args = kwargs.get("java_args", self.java_args) or ""
        cromwell_args = kwargs.get("cromwell_args", self._cromwell_args) or ""
        metadata_file = Path.cwd() / "metadata.json"
        cmd = (
            f"{self.java_bin} {java_args} -jar {self._cromwell_jar_file} run "
            f"-m {metadata_file} {cromwell_args} {inputs_arg} {imports_zip_arg} "
            f"{wdl_path}"
        )
        LOG.info(
            f"Executing cromwell command '{cmd}' with inputs "
            f"{json.dumps(inputs_dict, default=str)}"
        )
        exe = subby.run(cmd, raise_on_error=False)
        metadata = None
        if metadata_file.exists():
            with open(metadata_file, "rt") as inp:
                metadata = json.load(inp)
        if exe.ok:
            if metadata:
                assert metadata["status"] == "Succeeded"
                outputs = metadata["outputs"]
            else:
                LOG.warning(
                    f"Cromwell command completed successfully but did not generate "
                    f"a metadata file at {metadata_file}"
                )
                outputs = self._get_cromwell_outputs(exe.output)
        else:
            error_kwargs = {
                "executor": "cromwell",
                "target": target,
                "status": "Failed",
                "inputs": inputs_dict,
                "executor_stdout": exe.output,
                "executor_stderr": exe.error,
            }
            if metadata:
                error_kwargs = {
                    "executor": "cromwell",
                    "target": target,
                    "status": "Failed",
                    "inputs": inputs_dict,
                }
                self._parse_metadata_errors(
                    metadata, target=target, error_kwargs=error_kwargs
                )
            else:
                error_kwargs["msg"] = (
                    f"Cromwell command failed but did not generate a metadata "
                    f"file at {metadata_file}"
                )
            raise ExecutionFailedError(**error_kwargs)
        if expected:
            self._validate_outputs(outputs, expected, target)
        return outputs
def run_workflow ( self , wdl_path : Path , inputs : Optional [ dict ] = None , expected : Optional [ dict ] = None , ** kwargs , ) -> dict : target , is_task = get_target_name ( wdl_path = wdl_path , import_dirs = self . _import_dirs , ** kwargs ) if is_task : raise ValueError ( "Cromwell cannot execute tasks independently of a workflow" ) inputs_dict , inputs_file = read_write_inputs ( inputs_file = kwargs . get ( "inputs_file" ) , inputs_dict = inputs , namespace = target ) imports_file = self . _get_workflow_imports ( self . _import_dirs , kwargs . get ( "imports_file" ) ) inputs_arg = f"-i {inputs_file}" if inputs_file else "" imports_zip_arg = f"-p {imports_file}" if imports_file else "" java_args = kwargs . get ( "java_args" , self . java_args ) or "" cromwell_args = kwargs . get ( "cromwell_args" , self . _cromwell_args ) or "" metadata_file = Path . cwd ( ) / "metadata.json" cmd = ( f"{self.java_bin} {java_args} -jar {self._cromwell_jar_file} run " f"-m {metadata_file} {cromwell_args} {inputs_arg} {imports_zip_arg} " f"{wdl_path}" ) LOG . info ( f"Executing cromwell command '{cmd}' with inputs " f"{json.dumps(inputs_dict, default=str)}" ) exe = subby . run ( cmd , raise_on_error = False ) metadata = None if metadata_file . exists ( ) : with open ( metadata_file , "rt" ) as inp : metadata = json . load ( inp ) if exe . ok : if metadata : assert metadata [ "status" ] == "Succeeded" outputs = metadata [ "outputs" ] else : LOG . warning ( f"Cromwell command completed successfully but did not generate " f"a metadata file at {metadata_file}" ) outputs = self . _get_cromwell_outputs ( exe . output ) else : error_kwargs = { "executor" : "cromwell" , "target" : target , "status" : "Failed" , "inputs" : inputs_dict , "executor_stdout" : exe . output , "executor_stderr" : exe . error , } if metadata : error_kwargs = { "executor" : "cromwell" , "target" : target , "status" : "Failed" , "inputs" : inputs_dict , } self . _parse_metadata_errors ( metadata , target = target , error_kwargs = error_kwargs ) else : error_kwargs [ "msg" ] = ( f"Cromwell command failed but did not generate a metadata " f"file at {metadata_file}" ) raise ExecutionFailedError ( ** error_kwargs ) if expected : self . _validate_outputs ( outputs , expected , target ) return outputs
Run a WDL workflow on given inputs, and check that the output matches
given expected values.
Run a WDL workflow on given inputs and check that the output matches given expected values .
"""
        Run a WDL workflow on given inputs, and check that the output matches
        given expected values.

        Args:
            wdl_path: The WDL script to execute.
            inputs: Object that will be serialized to JSON and provided to Cromwell
                as the workflow inputs.
            expected: Dict mapping output parameter names to expected values.
            kwargs: Additional keyword arguments, mostly for debugging:
                * workflow_name: The name of the workflow in the WDL script. If None,
                    the name of the WDL script is used (without the .wdl extension).
                * inputs_file: Path to the Cromwell inputs file to use. Inputs are
                    written to this file only if it doesn't exist.
                * imports_file: Path to the WDL imports file to use. Imports are
                    written to this file only if it doesn't exist.
                * java_args: Additional arguments to pass to Java runtime.
                * cromwell_args: Additional arguments to pass to `cromwell run`.

        Returns:
            Dict of outputs.

        Raises:
            ExecutionFailedError: if there was an error executing Cromwell
            AssertionError: if the actual outputs don't match the expected outputs
        """
Dict of outputs.
Dict of outputs .
if there was an error executing Cromwell
if there was an error executing Cromwell
ExecutionFailedError
if the actual outputs don't match the expected outputs
if the actual outputs don ' t match the expected outputs
AssertionError
self
wdl_path
Path
The WDL script to execute.
The WDL script to execute .
inputs
Optional[dict]
Object that will be serialized to JSON and provided to Cromwell
as the workflow inputs.
Object that will be serialized to JSON and provided to Cromwell as the workflow inputs .
expected
Optional[dict]
Dict mapping output parameter names to expected values.
Dict mapping output parameter names to expected values .
kwargs
Additional keyword arguments, mostly for debugging:
workflow_name: The name of the workflow in the WDL script. If None,
the name of the WDL script is used (without the .wdl extension).
inputs_file: Path to the Cromwell inputs file to use. Inputs are
written to this file only if it doesn't exist.
imports_file: Path to the WDL imports file to use. Imports are
written to this file only if it doesn't exist.
java_args: Additional arguments to pass to Java runtime.
cromwell_args: Additional arguments to pass to `cromwell run`.
Additional keyword arguments mostly for debugging : workflow_name : The name of the workflow in the WDL script . If None the name of the WDL script is used ( without the . wdl extension ) . inputs_file : Path to the Cromwell inputs file to use . Inputs are written to this file only if it doesn ' t exist . imports_file : Path to the WDL imports file to use . Imports are written to this file only if it doesn ' t exist . java_args : Additional arguments to pass to Java runtime . cromwell_args : Additional arguments to pass to ` cromwell run ` .