pytest_wdl/executors/cromwell_server.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, _ = read_write_inputs(
            inputs_file=kwargs.get("inputs_file"),
            inputs_dict=inputs,
            namespace=target,
            write_formatted_inputs=False
        )

        payload = {}
        payload_files = []

        def open_payload_file(path: Path, mode: str = "r") -> IO:
            open_file = open(path, mode)
            payload_files.append(open_file)
            return open_file

        try:
            payload["workflowSource"] = open_payload_file(wdl_path)

            if inputs_dict:
                payload["workflowInputs"] = json.dumps(inputs_dict, default=str)

            imports_file = self._get_workflow_imports(
                self._import_dirs, kwargs.get("imports_file")
            )

            if imports_file:
                payload["workflowDependencies"] = open_payload_file(imports_file, "rb")

            if self._cromwell_config_file:
                if isinstance(inputs_dict, dict):
                    payload["workflowOptions"] = json.dumps(
                        self._cromwell_config_file, default=str
                    )
                else:
                    payload["workflowOptions"] = open_payload_file(
                        self._cromwell_config_file
                    )

            LOG.info(
                f"Executing cromwell server '{self._cromwell_api_url}' with inputs "
                f"{json.dumps(inputs_dict, default=str)}"
            )

            with requests.post(
                self._cromwell_api_url, files=payload, auth=self._auth
            ) as resp:
                status_object = self._resp_to_json(resp, target, inputs_dict)
                run_id = status_object["id"]
                LOG.info(
                    f"Executing on cromwell with id {run_id}. Waiting until terminal "
                    f"state is reached"
                )
        finally:
            for fh in payload_files:
                try:
                    fh.close()
                except:
                    LOG.exception("Error closing file %s", fh)

        self._poll_until_terminal(
            run_id, target, inputs_dict, kwargs.get("timeout", DEFAULT_POLLING_TIMEOUT)
        )

        metadata_url = f"{self._cromwell_api_url}/{run_id}/metadata"
        outputs = None

        with requests.get(metadata_url, auth=self._auth) as metadata_response:
            metadata = self._resp_to_json(metadata_response, target, inputs_dict)

            if metadata["status"] == "Succeeded":
                outputs = metadata["outputs"]
            else:
                error_kwargs = {
                    "executor": "cromwell",
                    "target": target,
                    "status": "Failed",
                    "inputs": inputs_dict,
                }
                self._parse_metadata_errors(
                    metadata, target=target, error_kwargs=error_kwargs
                )
                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, _ = read_write_inputs(
            inputs_file=kwargs.get("inputs_file"),
            inputs_dict=inputs,
            namespace=target,
            write_formatted_inputs=False
        )
        payload = {}
        payload_files = []
        def open_payload_file(path: Path, mode: str = "r") -> IO:
            open_file = open(path, mode)
            payload_files.append(open_file)
            return open_file
        try:
            payload["workflowSource"] = open_payload_file(wdl_path)
            if inputs_dict:
                payload["workflowInputs"] = json.dumps(inputs_dict, default=str)
            imports_file = self._get_workflow_imports(
                self._import_dirs, kwargs.get("imports_file")
            )
            if imports_file:
                payload["workflowDependencies"] = open_payload_file(imports_file, "rb")
            if self._cromwell_config_file:
                if isinstance(inputs_dict, dict):
                    payload["workflowOptions"] = json.dumps(
                        self._cromwell_config_file, default=str
                    )
                else:
                    payload["workflowOptions"] = open_payload_file(
                        self._cromwell_config_file
                    )
            LOG.info(
                f"Executing cromwell server '{self._cromwell_api_url}' with inputs "
                f"{json.dumps(inputs_dict, default=str)}"
            )
            with requests.post(
                self._cromwell_api_url, files=payload, auth=self._auth
            ) as resp:
                status_object = self._resp_to_json(resp, target, inputs_dict)
                run_id = status_object["id"]
                LOG.info(
                    f"Executing on cromwell with id {run_id}. Waiting until terminal "
                    f"state is reached"
                )
        finally:
            for fh in payload_files:
                try:
                    fh.close()
                except:
                    LOG.exception("Error closing file %s", fh)
        self._poll_until_terminal(
            run_id, target, inputs_dict, kwargs.get("timeout", DEFAULT_POLLING_TIMEOUT)
        )
        metadata_url = f"{self._cromwell_api_url}/{run_id}/metadata"
        outputs = None
        with requests.get(metadata_url, auth=self._auth) as metadata_response:
            metadata = self._resp_to_json(metadata_response, target, inputs_dict)
            if metadata["status"] == "Succeeded":
                outputs = metadata["outputs"]
            else:
                error_kwargs = {
                    "executor": "cromwell",
                    "target": target,
                    "status": "Failed",
                    "inputs": inputs_dict,
                }
                self._parse_metadata_errors(
                    metadata, target=target, error_kwargs=error_kwargs
                )
                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 , _ = read_write_inputs ( inputs_file = kwargs . get ( "inputs_file" ) , inputs_dict = inputs , namespace = target , write_formatted_inputs = False ) payload = { } payload_files = [ ] def open_payload_file ( path : Path , mode : str = "r" ) -> IO : open_file = open ( path , mode ) payload_files . append ( open_file ) return open_file try : payload [ "workflowSource" ] = open_payload_file ( wdl_path ) if inputs_dict : payload [ "workflowInputs" ] = json . dumps ( inputs_dict , default = str ) imports_file = self . _get_workflow_imports ( self . _import_dirs , kwargs . get ( "imports_file" ) ) if imports_file : payload [ "workflowDependencies" ] = open_payload_file ( imports_file , "rb" ) if self . _cromwell_config_file : if isinstance ( inputs_dict , dict ) : payload [ "workflowOptions" ] = json . dumps ( self . _cromwell_config_file , default = str ) else : payload [ "workflowOptions" ] = open_payload_file ( self . _cromwell_config_file ) LOG . info ( f"Executing cromwell server '{self._cromwell_api_url}' with inputs " f"{json.dumps(inputs_dict, default=str)}" ) with requests . post ( self . _cromwell_api_url , files = payload , auth = self . _auth ) as resp : status_object = self . _resp_to_json ( resp , target , inputs_dict ) run_id = status_object [ "id" ] LOG . info ( f"Executing on cromwell with id {run_id}. Waiting until terminal " f"state is reached" ) finally : for fh in payload_files : try : fh . close ( ) except : LOG . exception ( "Error closing file %s" , fh ) self . _poll_until_terminal ( run_id , target , inputs_dict , kwargs . get ( "timeout" , DEFAULT_POLLING_TIMEOUT ) ) metadata_url = f"{self._cromwell_api_url}/{run_id}/metadata" outputs = None with requests . get ( metadata_url , auth = self . _auth ) as metadata_response : metadata = self . _resp_to_json ( metadata_response , target , inputs_dict ) if metadata [ "status" ] == "Succeeded" : outputs = metadata [ "outputs" ] else : error_kwargs = { "executor" : "cromwell" , "target" : target , "status" : "Failed" , "inputs" : inputs_dict , } self . _parse_metadata_errors ( metadata , target = target , error_kwargs = error_kwargs ) 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 ` .