pytest_wdl.executors package

Submodules

pytest_wdl.executors.cromwell_local module

class pytest_wdl.executors.cromwell_local.CromwellLocalExecutor(import_dirs: Optional[Sequence[pathlib.Path]] = None, java_bin: Union[str, pathlib.Path, None] = None, java_args: Optional[str] = None, cromwell_jar_file: Union[str, pathlib.Path, None] = None, cromwell_configuration: Union[str, pathlib.Path, dict, None] = None, cromwell_args: Optional[str] = None, cromwell_config_file: Union[str, pathlib.Path, None] = None)[source]

Bases: pytest_wdl.executors.JavaExecutor, pytest_wdl.executors._cromwell.CromwellHelperMixin

Manages the running of WDL workflows using Cromwell.

Parameters

  • import_dirs – Relative or absolute paths to directories containing WDL scripts that should be available as imports.

  • java_bin – Path to the java executable.

  • java_args – Default Java arguments to use; can be overidden by passing java_args=… to run_workflow.

  • cromwell_jar_file – Path to the Cromwell JAR file.

  • cromwell_args – Default Cromwell arguments to use; can be overridden by passing cromwell_args=… to run_workflow.

run_workflow(wdl_path: pathlib.Path, inputs: Optional[dict] = None, expected: Optional[dict] = None, **kwargs) → dict[source]

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

Parameters

  • 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

pytest_wdl.executors.cromwell_server module

class pytest_wdl.executors.cromwell_server.CromwellServerExecutor(import_dirs: Optional[Sequence[pathlib.Path]] = None, cromwell_api_url: Optional[str] = 'http://localhost:8000/api/workflows/v1', cromwell_api_username: Optional[str] = None, cromwell_api_password: Optional[str] = None, cromwell_configuration: Union[str, pathlib.Path, dict, None] = None)[source]

Bases: pytest_wdl.executors.Executor, pytest_wdl.executors._cromwell.CromwellHelperMixin

Manages the running of WDL workflows using a remote Cromwell running in Server mode.

Parameters

  • import_dirs – Relative or absolute paths to directories containing WDL scripts that should be available as imports.

  • cromwell_api_url – The full URL where this cromwell exists http://localhost:8000/api/workflows/v1

  • cromwell_api_username – The username to pass to the cromwell API if protected by basic auth

  • cromwell_api_password – The password to pass to the cromwell API if protected by basic auth

  • cromwell_configuration – A config file that will be passed to Cromwell

run_workflow(wdl_path: pathlib.Path, inputs: Optional[dict] = None, expected: Optional[dict] = None, **kwargs) → dict[source]

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

Parameters

  • 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

pytest_wdl.executors.miniwdl module

class pytest_wdl.executors.miniwdl.MiniwdlExecutor(import_dirs: Optional[Sequence[pathlib.Path]] = None)[source]

Bases: pytest_wdl.executors.Executor

Manages the running of WDL workflows using miniwdl.

run_workflow(wdl_path: pathlib.Path, inputs: Optional[dict] = None, expected: Optional[dict] = None, **kwargs) → dict[source]

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

Parameters

  • wdl_path – The WDL script to execute.

  • inputs – Object that will be serialized to JSON and provided to miniwdl as the workflow inputs.

  • expected – Dict mapping output parameter names to expected values.

  • kwargs –

    Additional keyword arguments, mostly for debugging: * workflow_name: Name of the workflow to run. * task_name: Name of the task to run if a workflow isn’t defined. * inputs_file: Path to the miniwdl inputs file to use. Inputs are

    written to this file only if it doesn’t exist.

Returns

Dict of outputs.

Raises

  • Exception – if there was an error executing miniwdl

  • AssertionError – if the actual outputs don’t match the expected outputs

Module contents

exception pytest_wdl.executors.ExecutionFailedError(executor: str, target: str, status: str, inputs: Optional[dict] = None, executor_stdout: Optional[str] = None, executor_stderr: Optional[str] = None, failed_task: Optional[str] = None, failed_task_exit_status: Optional[int] = None, failed_task_stdout: Optional[str] = None, failed_task_stderr: Optional[str] = None, msg: Optional[str] = None)[source]

Bases: pytest_wdl.executors.ExecutorError

property exit_status_str
class pytest_wdl.executors.Executor[source]

Bases: object

Base class for WDL workflow executors.

abstract run_workflow(wdl_path: pathlib.Path, inputs: Optional[dict] = None, expected: Optional[dict] = None, **kwargs) → dict[source]

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

Parameters

  • 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 executor-specific keyword arguments (mostly for debugging)

Returns

Dict of outputs.

Raises

  • ExecutionFailedError – if there was an error executing the workflow

  • AssertionError – if the actual outputs don’t match the expected outputs

exception pytest_wdl.executors.ExecutorError(executor: str, msg: Optional[str] = None)[source]

Bases: Exception

class pytest_wdl.executors.InputsFormatter[source]

Bases: object

format_inputs(inputs_dict: dict, namespace: Optional[str] = None) → dict[source]
format_value(value: Any) → Any[source]

Convert a primitive, DataFile, Sequence, or Dict to a JSON-serializable object. Currently, arbitrary objects can be serialized by implementing an as_dict() method, otherwise they are converted to strings.

Parameters

value – The value to format.

Returns

The serializable value.

classmethod get_instance() → pytest_wdl.executors.InputsFormatter[source]
class pytest_wdl.executors.JavaExecutor(java_bin: Union[str, pathlib.Path, None] = None, java_args: Optional[str] = None)[source]

Bases: pytest_wdl.executors.Executor

Manages the running of WDL workflows using a Java-based executor.

Parameters

  • java_bin – Path to the java executable.

  • java_args – Default Java arguments to use; can be overidden by passing java_args=… to run_workflow.

static resolve_jar_file(file_name_pattern: str, jar_path: Optional[pathlib.Path] = None, env_var: Optional[str] = None)[source]
pytest_wdl.executors.get_target_name(wdl_path: Optional[pathlib.Path] = None, wdl_doc: Optional[WDL.Tree.Document] = None, task_name: Optional[str] = None, workflow_name: Optional[str] = None, **kwargs) → Tuple[str, bool][source]

Get the execution target. The order of priority is:

  • task_name

  • workflow_name

  • wdl_doc.workflow.name

  • wdl_doc.task[0].name

  • wdl_file.stem

Parameters

  • wdl_path – Path to a WDL file

  • wdl_doc – A miniwdl-parsed WDL document

  • task_name – The task name

  • workflow_name – The workflow name

  • **kwargs – Additional keyword arguments to pass to parse_wdl

Returns

A tuple (target, is_task), where is_task is a boolean indicating whether the target is a task (True) or a workflow (False).

Raises

  • ValueError if 1) neither task_name nor workflow_name is specified and the –

  • WDL document contains no workflow and multiple tasks; or 2) all of the –

  • parameters are None. –

pytest_wdl.executors.parse_wdl(wdl_path: pathlib.Path, import_dirs: Optional[Sequence[pathlib.Path]] = (), check_quant: bool = False, **_) → WDL.Tree.Document[source]
pytest_wdl.executors.read_write_inputs(inputs_file: Union[str, pathlib.Path, None] = None, inputs_dict: Optional[dict] = None, inputs_formatter: Optional[pytest_wdl.executors.InputsFormatter] = <pytest_wdl.executors.InputsFormatter object>, write_formatted_inputs: bool = True, **kwargs) → Tuple[dict, Optional[pathlib.Path]][source]

If inputs_file is specified and it exists, read its contents. Otherwise, if inputs_dict is specified, format it using inputs_formatter (if specified) and write it to inputs_file or a temporary file.

Parameters

  • inputs_file –

  • inputs_dict –

  • inputs_formatter –

  • write_formatted_inputs –

  • kwargs –

Returns

The (formatted) inputs dict and the resolved inputs file. If both inputs_dict and inputs_file are None, returns ({}, None).