pytest_wdl.executors package

Submodules

pytest_wdl.executors.cromwell module

class pytest_wdl.executors.cromwell.CromwellExecutor(import_dirs: Optional[List[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_config_file: Union[str, pathlib.Path, None] = None, cromwell_args: Optional[str] = None)

Bases: pytest_wdl.executors.Executor

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.

static get_cromwell_outputs(output) → dict

get_workflow_imports(imports_file: Optional[pathlib.Path] = None) → pathlib.Path

Creates a ZIP file with all WDL files to be imported.

Parameters

imports_file – Text file naming import directories/files - one per line.

Returns

Path to the ZIP file.

run_workflow(wdl_path: Union[str, pathlib.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.

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

• Exception – 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[List[pathlib.Path]] = None)

Bases: pytest_wdl.executors.Executor

Manages the running of WDL workflows using Cromwell.

run_workflow(wdl_path: pathlib.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.

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: 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 Cromwell 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 Cromwell

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

pytest_wdl.executors.miniwdl.log_source(logger: logging.Logger, exn)

Module contents

class pytest_wdl.executors.Executor(import_dirs: Optional[List[pathlib.Path]] = None)

Bases: object

Base class for WDL workflow executors.

Parameters

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

abstract run_workflow(wdl_path: pathlib.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.

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

• Exception – if there was an error executing the workflow

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

pytest_wdl.executors.compare_output_values(expected_value, actual_value, name: str) → None

Compare two values and raise an error if they are not equal.

Parameters

• expected_value –

• actual_value –

• name – Name of the output being compared

Raises

AssertionError –

pytest_wdl.executors.get_workflow_inputs(inputs_dict: Optional[dict] = None, inputs_file: Optional[pathlib.Path] = None, namespace: Optional[str] = None) → Tuple[dict, pathlib.Path]

Persist workflow inputs to a file, or load workflow inputs from a file.

Parameters

• inputs_dict – Dict of input names/values.

• inputs_file – JSON file with workflow inputs.

• namespace – Name of the workflow; used to prefix the input parameters when creating the inputs file from the inputs dict.

Returns

A tuple (inputs_dict, inputs_file)

pytest_wdl.executors.make_serializable(value)

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 make serializable.

Returns

The serializable value.

pytest_wdl.executors.validate_outputs(outputs: dict, expected: dict, target: str) → None

Validate expected and actual outputs are equal.

Parameters

• outputs – Actual outputs

• expected – Expected outputs

• target – Execution target (i.e. workflow name)

Raises

AssertionError –