pytest_wdl package¶
Subpackages¶
Submodules¶
pytest_wdl.config module¶
-
class
pytest_wdl.config.
UserConfiguration
(config_file: Optional[pathlib.Path] = None, cache_dir: Optional[pathlib.Path] = None, remove_cache_dir: Optional[bool] = None, execution_dir: Optional[pathlib.Path] = None, proxies: Optional[Dict[str, Union[str, Dict[str, str]]]] = None, http_headers: Optional[List[dict]] = None, show_progress: Optional[bool] = None, executors: Optional[str] = None, executor_defaults: Optional[Dict[str, dict]] = None, provider_defaults: Optional[Dict[str, dict]] = None)[source]¶ Bases:
object
Stores pytest-wdl configuration. If configuration options are specified both in the config file and as arguments to the constructor, the latter take precedence.
- Parameters
config_file – JSON (or YAML) file from which to load default values.
cache_dir – The directory in which to cache localized files; defaults to using a temporary directory that is specific to each module and deleted afterwards.
remove_cache_dir – Whether to remove the cache directory; if None, takes the value True if a temp directory is used for caching, and False, if a value for cache_dir is specified.
execution_dir – The directory in which to run workflows. Defaults to None, which signals that a different temporary directory should be used for each workflow run.
proxies – Mapping of proxy type (typically ‘http’ or ‘https’ to either an environment variable, or a dict with either/both keys ‘env’ and ‘value’, where the value is taken from the environment variable (‘env’) first, and from ‘value’ if the environment variable is not specified or is unset.
http_headers – A list of dicts, each of which defines a header. The allowed keys are ‘pattern’, ‘name’, ‘env’, and ‘value’, where pattern is a URL pattern to match, ‘name’ is the header name and ‘env’ and ‘value’ are interpreted the same as for proxies. If no pattern is provided, the header is used for all URLs.
show_progress – Whether to show progress bars when downloading remote test data files.
executors – Default set of executors to run.
executor_defaults – Mapping of executor name to dict of executor-specific configuration options.
-
cleanup
() → None[source]¶ Preforms cleanup operations, such as deleting the cache directory if self.remove_cache_dir is True.
-
get_executor_defaults
(executor_name: str) → dict[source]¶ Get default configuration values for the given executor.
- Parameters
executor_name – The executor name
- Returns
A dict with the executor configuration values, if any.
pytest_wdl.core module¶
-
pytest_wdl.core.
DATA_TYPES
= {'bam': <pytest_wdl.plugins.PluginFactory object>, 'json': <pytest_wdl.plugins.PluginFactory object>, 'vcf': <pytest_wdl.plugins.PluginFactory object>}¶ Data type plugin modules from the discovered entry points.
-
class
pytest_wdl.core.
DataDirs
(basedir: pathlib.Path, module: Union[str, Any, None] = None, function: Union[str, Callable, None] = None, cls: Union[str, Type, None] = None)[source]¶ Bases:
object
Provides data files from test data directory structure as defined by the datadir and datadir-ng plugins. Paths are resolved lazily upon first request.
-
property
paths
¶
-
property
-
class
pytest_wdl.core.
DataManager
(data_resolver: pytest_wdl.core.DataResolver, datadirs: pytest_wdl.core.DataDirs)[source]¶ Bases:
object
Manages test data, which is defined in a test_data.json file.
- Parameters
data_resolver – Module-level config.
datadirs – Data directories to search for the data file.
-
get_dict
(*names, **params) → dict[source]¶ Creates a dict with one or more entries from this DataManager.
- Parameters
*names – Names of test data entries to add to the dict.
**params – Mapping of workflow parameter names to test data entry names.
- Returns
Dict mapping parameter names to test data entries for all specified names.
-
class
pytest_wdl.core.
DataResolver
(data_descriptors: dict, user_config: pytest_wdl.config.UserConfiguration)[source]¶ Bases:
object
Resolves data files that may need to be localized.
-
pytest_wdl.core.
EXECUTORS
= {'cromwell': <pytest_wdl.plugins.PluginFactory object>, 'cromwell-server': <pytest_wdl.plugins.PluginFactory object>, 'dxwdl': <pytest_wdl.plugins.PluginFactory object>, 'miniwdl': <pytest_wdl.plugins.PluginFactory object>}¶ Executor plugin modules from the discovered entry points.
-
pytest_wdl.core.
create_data_file
(user_config: pytest_wdl.config.UserConfiguration, type: Union[str, dict, None] = 'default', name: Optional[str] = None, path: Union[str, pathlib.Path, None] = None, url: Optional[str] = None, contents: Union[str, dict, None] = None, env: Optional[str] = None, http_headers: Optional[dict] = None, digests: Optional[dict] = None, datadirs: Optional[pytest_wdl.core.DataDirs] = None, **kwargs) → pytest_wdl.data_types.DataFile[source]¶
pytest_wdl.fixtures module¶
Fixtures for writing tests that execute WDL workflows using Cromwell.
Note: This library is being transitioned to python3 only, and to use `pathlib.Path`s instead of string paths. For backward compatibility fixtures that produce a path may still return string paths, but this support will be dropped in a future version.
-
class
pytest_wdl.fixtures.
WorkflowRunner
(wdl_search_paths: Sequence[pathlib.Path], import_dirs: Sequence[pathlib.Path], user_config: pytest_wdl.config.UserConfiguration, subtests: pytest_subtests.SubTests, default_executors: Optional[Sequence[str]] = None)[source]¶ Bases:
object
Callable object that runs tests.
-
pytest_wdl.fixtures.
default_executors
(user_config: pytest_wdl.config.UserConfiguration) → Sequence[str][source]¶
-
pytest_wdl.fixtures.
import_dirs
(request: _pytest.fixtures.FixtureRequest, project_root: Union[str, pathlib.Path], import_paths: Union[str, pathlib.Path, None]) → List[Union[str, pathlib.Path]][source]¶ Fixture that provides a list of directories containing WDL scripts to make avaialble as imports. Uses the file provided by import_paths fixture if it is not None, otherwise returns a list containing the parent directory of the test module.
- Parameters
request – A FixtureRequest object
project_root – Project root directory
import_paths – File listing paths to imports, one per line
-
pytest_wdl.fixtures.
import_paths
(request: _pytest.fixtures.FixtureRequest) → Union[str, pathlib.Path, None][source]¶ Fixture that provides the path to a file that lists directories containing WDL scripts to make available as imports. This looks for the file at “tests/import_paths.txt” by default, and returns None if that file doesn’t exist.
-
pytest_wdl.fixtures.
project_root
(request: _pytest.fixtures.FixtureRequest, project_root_files: List[str]) → Union[str, pathlib.Path][source]¶ Fixture that provides the root directory of the project. By default, this assumes that the project has one subdirectory per task, and that this framework is being run from the test subdirectory of a task diretory, and therefore looks for the project root two directories up.
-
pytest_wdl.fixtures.
project_root_files
() → List[str][source]¶ Fixture that provides a list of filenames that are found in the project root directory. Used by the project_root fixture to locate the project root directory.
-
pytest_wdl.fixtures.
user_config
(user_config_file: Optional[pathlib.Path]) → pytest_wdl.config.UserConfiguration[source]¶
-
pytest_wdl.fixtures.
user_config_file
() → Optional[pathlib.Path][source]¶ Fixture that provides the value of ‘user_config’ environment variable. If not specified, looks in the default location ($HOME/pytest_user_config.json).
- Returns
Path to the confif file, or None if not specified.
-
pytest_wdl.fixtures.
workflow_data
(request: _pytest.fixtures.FixtureRequest, workflow_data_resolver: pytest_wdl.core.DataResolver) → pytest_wdl.core.DataManager[source]¶ Provides an accessor for test data files, which may be local or in a remote repository.
- Parameters
request – FixtureRequest object
workflow_data_resolver – Module-level test data configuration
Examples
- def workflow_data_descriptor_file():
return “tests/test_data.json”
- def test_workflow(workflow_data):
print(workflow_data[“myfile”])
-
pytest_wdl.fixtures.
workflow_data_descriptor_file
(request: _pytest.fixtures.FixtureRequest) → Union[str, pathlib.Path][source]¶ Fixture that provides the path to the JSON file that describes test data files.
- Parameters
request – A FixtureRequest object
-
pytest_wdl.fixtures.
workflow_data_descriptors
(request: _pytest.fixtures.FixtureRequest, project_root: Union[str, pathlib.Path], workflow_data_descriptor_file: Union[str, pathlib.Path]) → dict[source]¶ Fixture that provides a mapping of test data names to values. If workflow_data_descriptor_file is relative, it is searched first relative to the current test context directory and then relative to the project root.
- Parameters
workflow_data_descriptor_file – Path to the data descriptor JSON file.
- Returns
A dict with keys as test data names and each value either a primitive, a map describing a data file, or a DataFile object.
-
pytest_wdl.fixtures.
workflow_data_resolver
(workflow_data_descriptors: dict, user_config: pytest_wdl.config.UserConfiguration) → pytest_wdl.core.DataResolver[source]¶ Provides access to test data files for tests in a module.
- Parameters
workflow_data_descriptors – workflow_data_descriptors fixture.
user_config –
-
pytest_wdl.fixtures.
workflow_runner
(request: _pytest.fixtures.FixtureRequest, project_root: Union[str, pathlib.Path], import_dirs: List[Union[str, pathlib.Path]], user_config: pytest_wdl.config.UserConfiguration, default_executors: Sequence[str], subtests: pytest_subtests.SubTests)[source]¶ Provides a callable that runs a workflow. The callable has the same signature as Executor.run_workflow, but takes an additional keyword argument executors, a sequence of strings, which allows overriding the names of the executors to use.
If multiple executors are specified, the tests are run using the subtests fixture of the pytest-subtests plugin.
- Parameters
request – A FixtureRequest object.
project_root – Project root directory.
import_dirs – Directories from which to import WDL scripts.
user_config – A UserConfiguration object.
default_executors – Names of executors to use when executor name isn’t passed to the workflow_runner callable.
subtests – A SubTests object.
- Returns
A generator over the results of calling the workflow with each executor. Each value is a tuple (executor_name, execution_dir, outputs), where execution_dir is the root directory where the task/workflow was run (the structure of the directory is executor-dependent) and outputs is a dict of the task/workflow outputs.
pytest_wdl.loader module¶
-
class
pytest_wdl.loader.
JsonWdlTestsModule
(fspath: py._path.local.LocalPath, parent=None, config=None, session=None, nodeid=None)[source]¶
-
class
pytest_wdl.loader.
TestItem
(parent, data: Optional[dict] = None, name: Optional[str] = None, wdl: Optional[str] = None, inputs: Optional[dict] = None, expected: Optional[dict] = None, tags: Optional[Sequence] = None, **kwargs)[source]¶ Bases:
_pytest.nodes.Item
-
setup
()[source]¶ This method is black magic - uses internal pytest APIs to create a FixtureRequest that can be used to access fixtures in runtest(). Copied from https://github.com/pytest-dev/pytest/blob/master/src/_pytest/doctest.py.
-
-
class
pytest_wdl.loader.
WdlTestsModule
(fspath: py._path.local.LocalPath, parent=None, config=None, session=None, nodeid=None)[source]¶ Bases:
_pytest.python.Module
-
class
pytest_wdl.loader.
YamlWdlTestsModule
(fspath: py._path.local.LocalPath, parent=None, config=None, session=None, nodeid=None)[source]¶
pytest_wdl.localizers module¶
-
class
pytest_wdl.localizers.
LinkLocalizer
(source: pathlib.Path)[source]¶ Bases:
pytest_wdl.localizers.Localizer
Localizes a file to another destination using a symlink.
-
class
pytest_wdl.localizers.
Localizer
[source]¶ Bases:
object
Abstract base of classes that implement file localization.
-
class
pytest_wdl.localizers.
StringLocalizer
(contents: str)[source]¶ Bases:
pytest_wdl.localizers.Localizer
Localizes a string by writing it to a file.
-
class
pytest_wdl.localizers.
UrlLocalizer
(url: str, user_config: pytest_wdl.config.UserConfiguration, http_headers: Optional[dict] = None, digests: Optional[dict] = None)[source]¶ Bases:
pytest_wdl.localizers.Localizer
Localizes a file specified by a URL.
-
property
http_headers
¶
-
localize
(destination: pathlib.Path)[source]¶ Localize a resource to destination.
- Parameters
destination – Path to file where the non-local resource is to be localized.
-
property
proxies
¶
-
property
pytest_wdl.plugins module¶
-
class
pytest_wdl.plugins.
PluginFactory
(entry_point: pkg_resources.EntryPoint, return_type: Type[T])[source]¶ Bases:
typing.Generic
Lazily loads a plugin class associated with a data type.
-
pytest_wdl.plugins.
plugin_factory_map
(return_type: Type[T], group: Optional[str] = None, entry_points: Optional[Iterable[pkg_resources.EntryPoint]] = None) → Dict[str, pytest_wdl.plugins.PluginFactory[~T][T]][source]¶ Creates a mapping of entry point name to PluginFactory for all discovered entry points in the specified group.
- Parameters
group – Entry point group name
return_type – Expected return type
entry_points –
- Returns
Dict mapping entry point name to PluginFactory instances
pytest_wdl.url_schemes module¶
-
class
pytest_wdl.url_schemes.
BaseResponse
[source]¶ Bases:
pytest_wdl.url_schemes.Response
-
download_file
(destination: pathlib.Path, show_progress: bool = False, digests: Optional[dict] = None)[source]¶ Download a file to a specific destination.
- Parameters
destination – Destination path
show_progress – Whether to show a progress bar
digests – Optional dict mapping hash names to digests. These are used to validate the downloaded file.
- Raises
-
-
class
pytest_wdl.url_schemes.
Method
(src_attr, dest_pattern)[source]¶ Bases:
enum.Enum
An enumeration.
-
OPEN
= ('urlopen', '{}_open')¶
-
REQUEST
= ('request', '{}_request')¶
-
RESPONSE
= ('response', '{}_response')¶
-
-
class
pytest_wdl.url_schemes.
Response
[source]¶ Bases:
object
-
abstract
download_file
(destination: pathlib.Path, show_progress: bool = False, digests: Optional[dict] = None)[source]¶ Download a file to a specific destination.
- Parameters
destination – Destination path
show_progress – Whether to show a progress bar
digests – Optional dict mapping hash names to digests. These are used to validate the downloaded file.
- Raises
-
abstract
pytest_wdl.utils module¶
-
exception
pytest_wdl.utils.
MaxCallException
(last=None)[source]¶ Bases:
pytest_wdl.utils.PollingException
Exception raised if maximum number of iterations is exceeded
-
exception
pytest_wdl.utils.
PollingException
(last=None)[source]¶ Bases:
Exception
Base exception that stores the last result seen.
-
exception
pytest_wdl.utils.
TimeoutException
(last=None)[source]¶ Bases:
pytest_wdl.utils.PollingException
Exception raised if polling function times out
-
pytest_wdl.utils.
chdir
(todir: pathlib.Path)[source]¶ Context manager that temporarily changes directories.
- Parameters
todir – The directory to change to.
-
pytest_wdl.utils.
compare_files_with_hash
(file1: pathlib.Path, file2: pathlib.Path, hash_name: str = 'md5')[source]¶
-
pytest_wdl.utils.
context_dir
(path: Optional[pathlib.Path] = None, change_dir: bool = False, cleanup: Optional[bool] = None) → pathlib.Path[source]¶ Context manager that looks for a specific environment variable to specify a directory. If the environment variable is not set, a temporary directory is created and cleaned up upon return from the yield.
- Parameters
path – The environment variable to look for.
change_dir – Whether to change to the directory.
cleanup – Whether to delete the directory when exiting the context. If None, the directory is only deleted if a temporary directory is created.
- Yields
A directory path.
-
pytest_wdl.utils.
ensure_path
(path: Union[str, py._path.local.LocalPath, pathlib.Path], search_paths: Optional[Sequence[pathlib.Path]] = None, canonicalize: bool = True, exists: Optional[bool] = None, is_file: Optional[bool] = None, executable: Optional[bool] = None, create: bool = False) → pathlib.Path[source]¶ Converts a string path or
py.path.local.LocalPath
to apathlib.Path
.- Parameters
path – The path to convert.
search_paths – Directories to search for path if it is not already absolute. If exists is True, looks for the first search path that contains the file, otherwise just uses the first search path.
canonicalize – Whether to return the canonicalized version of the path - expand home directory shortcut (~), make absolute, and resolve symlinks.
exists – If True, raise an exception if the path does not exist; if False, raise an exception if the path does exist.
is_file – If True, raise an exception if the path is not a file; if False, raise an exception if the path is not a directory.
executable – If True and is_file is True and the file exists, raise an exception if it is not executable.
create – Create the directory (or parent, if is_file = True) if it does not exist. Ignored if exists is True.
- Returns
A pathlib.Path object.
-
pytest_wdl.utils.
env_map
(d: dict) → dict[source]¶ Given a mapping of keys to value descriptors, creates a mapping of the keys to the described values.
-
pytest_wdl.utils.
find_executable_path
(executable: str, search_path: Optional[Sequence[pathlib.Path]] = None) → Optional[pathlib.Path][source]¶ Finds ‘executable’ in search_path.
- Parameters
executable – The name of the executable to find.
search_path – The list of directories to search. If None, the system search path (defined by the $PATH environment variable) is used.
- Returns
Absolute path of the executable, or None if no matching executable was found.
-
pytest_wdl.utils.
find_in_classpath
(glob: str) → Optional[pathlib.Path][source]¶ Attempts to find a .jar file matching the specified glob pattern in the Java classpath.
- Parameters
glob – JAR filename pattern
- Returns
Path to the JAR file, or None if a matching file is not found.
-
pytest_wdl.utils.
find_project_path
(*filenames, start: Optional[pathlib.Path] = None, return_parent: bool = False, assert_exists: bool = False) → Optional[pathlib.Path][source]¶ Starting from path folder and moving upwards, search for any of filenames and return the first path containing any one of them.
- Parameters
*filenames – Filenames to search. Either a string filename, or a sequence of string path elements.
start – Starting folder
return_parent – Whether to return the containing folder or the discovered file.
assert_exists – Whether to raise an exception if a file cannot be found.
- Returns
A Path, or None if no folder is found that contains any of filenames. If return_parent is False and more than one of the files is found one of the files is randomly selected for return.
- Raises
FileNotFoundError if the file cannot be found and assert_exists is True. –
-
pytest_wdl.utils.
is_executable
(path: pathlib.Path) → bool[source]¶ Checks if a path is executable.
- Parameters
path – The path to check
- Returns
True if path exists and is executable by the user, otherwise False.
-
pytest_wdl.utils.
poll
(target: Callable, step: int = 1, args: Optional[Sequence] = None, kwargs: Optional[dict] = None, timeout: Optional[int] = None, max_tries: Optional[int] = None, check_success: Callable = <class 'bool'>, step_function: Optional[Callable[[int, int], int]] = None, ignore_exceptions: Sequence = ())[source]¶ Poll by calling a target function until a certain condition is met. You must specify at least a target function to be called and the step – base wait time between each function call.
Vendored from the [polling](https://github.com/justiniso/polling) package.
- Parameters
target – The target callable
step – Step defines the amount of time to wait (in seconds)
args – Arguments to be passed to the target function
kwargs – Keyword arguments to be passed to the target function
timeout – The target function will be called until the time elapsed is greater than the maximum timeout (in seconds). NOTE that the actual execution time of the function can exceed the time specified in the timeout. For instance, if the target function takes 10 seconds to execute and the timeout is 21 seconds, the polling function will take a total of 30 seconds (two iterations of the target –20s which is less than the timeout–21s, and a final iteration)
max_tries – Maximum number of times the target function will be called before failing
check_success – A callback function that accepts the return value of the target function. It must return true if you want the polling function to stop and return this value. It must return false if you want to continue polling. You may also use this function to collect non-success values. The default is a callback that tests for truthiness (anything not False, 0, or empty collection).
step_function –
A callback function that accepts two arguments: current_step, num_tries; and returns the next step value. By default, this is constant, but you can also pass a function that will increase or decrease the step. As an example, you can increase the wait time between calling the target function by 10 seconds every iteration until the step is 100 seconds–at which point it should remain constant at 100 seconds
>>> def my_step_function(current_step: int, num_tries: int) -> int: >>> return max(current_step + 10, 100)
ignore_exceptions – You can specify a tuple of exceptions that should be caught and ignored on every iteration. If the target function raises one of these exceptions, it will be caught and the exception instance will be pushed to the queue of values collected during polling. Any other exceptions raised will be raised as normal.
- Returns
The first value from the target function that meets the condions of the check_success callback. By default, this will be the first value that is not None, 0, False, ‘’, or an empty collection.
-
pytest_wdl.utils.
resolve_file
(filename: Union[str, pathlib.Path], project_root: pathlib.Path, assert_exists: bool = True) → Optional[pathlib.Path][source]¶ Finds filename under project_root or in the project path.
- Parameters
filename – The filename, relative path, or absolute path to resolve.
project_root – The project root dir.
assert_exists – Whether to raise an error if the file cannot be found.
- Returns
A pathlib.Path object, or None if the file cannot be found and assert_exists is False.
- Raises
FileNotFoundError if the file cannot be found and assert_exists is True. –
-
pytest_wdl.utils.
resolve_value_descriptor
(value_descriptor: Union[str, dict]) → Optional[source]¶ Resolves the value of a value descriptor, which may be an environment variable name, or a map with keys env (the environment variable name) and value (the value to use if env is not specified or if the environment variable is unset.
- Parameters
value_descriptor –
Returns:
-
pytest_wdl.utils.
safe_string
(s: str, replacement: str = '_') → str[source]¶ Makes a string safe by replacing non-word characters.
- Parameters
s – The string to make safe
replacement – The replacement stringj
- Returns
The safe string
-
pytest_wdl.utils.
tempdir
(change_dir: bool = False, tmproot: Optional[pathlib.Path] = None, cleanup: Optional[bool] = True) → pathlib.Path[source]¶ Context manager that creates a temporary directory, yields it, and then deletes it after return from the yield.
- Parameters
change_dir – Whether to temporarily change to the temp dir.
tmproot – Root directory in which to create temporary directories.
cleanup – Whether to delete the temporary directory before exiting the context.
Module contents¶
Fixtures for writing tests that execute WDL workflows using Cromwell. For testability purposes, the implementaiton of these fixtures is done in the pytest_wdl.fixtures module.