Test Wrappers

class concoursetools.testing.SimpleTestResourceWrapper(inner_resource, directory_dict=None, one_off_build=False, instance_vars=None, **env_vars)[source]

A simplistic resource wrapper designed to reduce test code.

The only real functionality of this wrapper is to mock arguments such as the Build Metadata and the directory paths.

Tip

This is best to use if want to invoke your resource code with as little effort as possible.

Parameters:

  • inner_resource (ConcourseResource[Version]) – The resource to be wrapped.

  • directory_dict (dict[str, Any] | None) – The initial state of the resource directory. See TemporaryDirectoryState

  • one_off_build (bool) – Set to True if you are testing a one-off build.

  • instance_vars (dict[str, str] | None) – Pass optional instance vars to emulate an instanced pipeline.

  • env_vars (str) – Pass additional environment variables, or overload the default ones.

fetch_new_versions(previous_version=None)[source]

Fetch new versions of the resource.

Calls the inner resource whilst capturing debugging output.

Parameters:

previous_version (Optional[Version]) – The most recent version of the resource. This will be set to None if the resource has never been run before.

Return type:

list[Version]

Returns:

A list of new versions.

download_version(version, **params)[source]

Download a version and place its files within the resource directory in your pipeline.

Calls the inner resource whilst capturing debugging output and passing a temporary directory as the destination directory.

Parameters:

  • version (Version) – The version to be downloaded.

  • params (object) – Additional keyword parameters passed to the inner resource.

Return type:

tuple[Version, dict[str, str]]

Returns:

The version (most likely unchanged), and a dictionary of metadata.

publish_new_version(**params)[source]

Update a resource by publishing a new version.

Calls the inner resource whilst capturing debugging output and passing a temporary directory as the sources directory.

Parameters:

params (object) – Additional keyword parameters passed to the inner resource.

Return type:

tuple[Version, dict[str, str]]

Returns:

The new version, and a dictionary of metadata.

class concoursetools.testing.JSONTestResourceWrapper(inner_resource_type, inner_resource_config, directory_dict=None, one_off_build=False, instance_vars=None, **env_vars)[source]

A resource wrapper which interfaces directly with JSON configuration and Main Scripts.

Tip

This is best to use if you are concerned about the conversion of your resource and versions.

Parameters:

  • inner_resource_type (type[ConcourseResource[Version]]) – The ConcourseResource subclass corresponding to the resource.

  • inner_resource_config (dict[str, Any]) – The JSON configuration for the resource.

  • directory_dict (dict[str, Any] | None) – The initial state of the resource directory. See TemporaryDirectoryState

  • one_off_build (bool) – Set to True if you are testing a one-off build.

  • instance_vars (dict[str, str] | None) – Pass optional instance vars to emulate an instanced pipeline.

  • env_vars (str) – Pass additional environment variables, or overload the default ones.

fetch_new_versions(previous_version_config=None)[source]

Fetch new versions of the resource.

Mocks the environment and calls check_main().

Caution

No environment variables are available to the check script.

Parameters:

previous_version_config (dict[str, str] | None) – The JSON configuration of the most recent version of the resource. This will be set to None if the resource has never been run before.

Return type:

list[dict[str, str]]

Returns:

A list of new version configurations.

download_version(version_config, params=None)[source]

Download a version and place its files within the resource directory in your pipeline.

Mocks the environment and calls in_main().

Parameters:

  • version_config (dict[str, str]) – The JSON configuration of the version.

  • params (dict[str, Any] | None) – A mapping of additional keyword parameters passed to the inner resource.

Return type:

tuple[dict[str, str], list[MetadataPair]]

Returns:

The version configuration (most likely unchanged), and a list of metadata pairs.

publish_new_version(params=None)[source]

Update a resource by publishing a new version.

Mocks the environment and calls out_main().

Parameters:

params (dict[str, Any] | None) – A mapping of additional keyword parameters passed to the inner resource.

Return type:

tuple[dict[str, str], list[MetadataPair]]

Returns:

The new version configuration, and a list of metadata pairs.

class concoursetools.testing.ConversionTestResourceWrapper(inner_resource_type, inner_resource_config, directory_dict=None, one_off_build=False, instance_vars=None, **env_vars)[source]

A resource wrapper based on JSONTestResourceWrapper, yet with the interface of SimpleTestResourceWrapper.

All inputs and outputs are instances, but are converted to and from JSON internally.

Tip

This is best to use if you want easier testing, but are still concerned about the conversion of your resource and versions.

Parameters:

  • inner_resource_type (type[ConcourseResource[Version]]) – The ConcourseResource subclass corresponding to the resource.

  • inner_resource_config (dict[str, Any]) – The JSON configuration for the resource.

  • directory_dict (dict[str, Any] | None) – The initial state of the resource directory. See TemporaryDirectoryState

  • one_off_build (bool) – Set to True if you are testing a one-off build.

  • instance_vars (dict[str, str] | None) – Pass optional instance vars to emulate an instanced pipeline.

  • env_vars (str) – Pass additional environment variables, or overload the default ones.

fetch_new_versions(previous_version=None)[source]

Fetch new versions of the resource.

Converts the version (if it exists) to JSON, and then invokes fetch_new_versions(). The response is then converted back to Version instances.

Parameters:

previous_version (Optional[Version]) – The most recent version of the resource. This will be set to None if the resource has never been run before.

Return type:

list[Version]

Returns:

A list of new versions.

download_version(version, **params)[source]

Download a version and place its files within the resource directory in your pipeline.

Converts the version to JSON, and then invokes download_version() with the additional params as a dict. The returned version configuration is then converted back to a Version instance, and the metadata pairs converted to a standard dict.

Parameters:

  • version (Version) – The version to be downloaded.

  • params (object) – Additional keyword parameters passed to the inner resource.

Return type:

tuple[Version, dict[str, str]]

Returns:

The version (most likely unchanged), and a dictionary of metadata.

publish_new_version(**params)[source]

Update a resource by publishing a new version.

Invokes publish_new_version() with the additional params as a dict. The returned version configuration is then converted to a Version instance, and the metadata pairs converted to a standard dict.

Parameters:

params (object) – Additional keyword parameters passed to the inner resource.

Return type:

tuple[Version, dict[str, str]]

Returns:

The new version, and a dictionary of metadata.

class concoursetools.testing.FileTestResourceWrapper(inner_resource_config, check_script=None, in_script=None, out_script=None, directory_dict=None, one_off_build=False, instance_vars=None, **env_vars)[source]

A resource wrapper which calls arbitrary scripts.

Tip

This is best to use if your resource uses external scripts for any of the steps.

Parameters:

Caution

If any of the paths for the scripts do not resolve, the corresponding methods will raise NotImplementedError.

fetch_new_versions(previous_version_config=None)[source]

Fetch new versions of the resource.

Calls the external check script using run_script() with the correct environment.

Caution

No environment variables are available to the check script.

Parameters:

previous_version_config (dict[str, str] | None) – The JSON configuration of the most recent version of the resource. This will be set to None if the resource has never been run before.

Return type:

list[dict[str, str]]

Returns:

A list of new version configurations.

download_version(version_config, params=None)[source]

Download a version and place its files within the resource directory in your pipeline.

Calls the external in script using run_script() with the correct environment.

Parameters:

  • version_config (dict[str, str]) – The JSON configuration of the version.

  • params (dict[str, Any] | None) – A mapping of additional keyword parameters passed to the inner resource.

Return type:

tuple[dict[str, str], list[MetadataPair]]

Returns:

The version configuration (most likely unchanged), and a list of metadata pairs.

publish_new_version(params=None)[source]

Update a resource by publishing a new version.

Calls the external out script using run_script() with the correct environment.

Parameters:

params (dict[str, Any] | None) – A mapping of additional keyword parameters passed to the inner resource.

Return type:

tuple[dict[str, str], list[MetadataPair]]

Returns:

The new version configuration, and a list of metadata pairs.

classmethod from_assets_dir(inner_resource_config, assets_dir, directory_dict=None, one_off_build=False, instance_vars=None, **env_vars)[source]

Create an instance from a single folder of asset files.

Parameters:

  • inner_resource_config (dict[str, Any]) – The JSON configuration for the resource.

  • assets_dir (Path) – The path to a folder containing the external script files. Files are expected to be named check, in and out. If not found, then no error will be raised unless the corresponding method is called.

  • directory_dict (dict[str, Any] | None) – The initial state of the resource directory. See TemporaryDirectoryState

  • one_off_build (bool) – Set to True if you are testing a one-off build.

  • instance_vars (dict[str, str] | None) – Pass optional instance vars to emulate an instanced pipeline.

  • env_vars (str) – Pass additional environment variables, or overload the default ones.

Return type:

FileTestResourceWrapper

class concoursetools.testing.FileConversionTestResourceWrapper(inner_resource_type, inner_resource_config, executable, permissions=493, encoding=None, directory_dict=None, one_off_build=False, instance_vars=None, **env_vars)[source]

A resource wrapper which converts the resource class to arbitrary scripts for execution.

This wrapper converts the resource class to its external scripts using create_script_file(). Inputs are then converted to JSON and passed as in FileTestResourceWrapper.

Tip

This is best to use if you want easier testing, but are concerned about the conversion of resource class to external scripts.

Parameters:
fetch_new_versions(previous_version=None)[source]

Fetch new versions of the resource.

Converts the version (if it exists) to JSON, exports the script to an external file, and then invokes fetch_new_versions(). The response is then converted back to Version instances.

Caution

The external script file only exists for the duration of this method and is not cached.

Parameters:

previous_version (Optional[Version]) – The most recent version of the resource. This will be set to None if the resource has never been run before.

Return type:

list[Version]

Returns:

A list of new versions.

download_version(version, **params)[source]

Download a version and place its files within the resource directory in your pipeline.

Converts the version to JSON, exports the script to an external file, and then invokes download_version() with the additional params as a dict. The returned version configuration is then converted back to a Version instance, and the metadata pairs converted to a standard dict.

Caution

The external script file only exists for the duration of this method and is not cached.

Parameters:

  • version (Version) – The version to be downloaded.

  • params (object) – Additional keyword parameters passed to the inner resource.

Return type:

tuple[Version, dict[str, str]]

Returns:

The version (most likely unchanged), and a dictionary of metadata.

publish_new_version(**params)[source]

Update a resource by publishing a new version.

Exports the script to an external file and then invokes publish_new_version() with the additional params as a dict. The returned version configuration is then converted to a Version instance, and the metadata pairs converted to a standard dict.

Caution

The external script file only exists for the duration of this method and is not cached.

Parameters:

params (object) – Additional keyword parameters passed to the inner resource.

Return type:

tuple[Version, dict[str, str]]

Returns:

The new version, and a dictionary of metadata.

class concoursetools.testing.DockerTestResourceWrapper(inner_resource_config, image, directory_dict=None, one_off_build=False, instance_vars=None, **env_vars)[source]

A resource wrapper which calls a Docker image.

The container only persists for the duration of the method call.

Tip

This is best to use if you want to be sure that your Docker image has been built properly, or to test resource types which have not been built with Concourse Tools.

Added in version 0.8.0.

Parameters:

  • inner_resource_config (dict[str, Any]) – The JSON configuration for the resource.

  • image (str) – The Docker image to use, which must exist in the local cache. Passed verbatim to docker run.

  • directory_dict (dict[str, Any] | None) – The initial state of the resource directory. See TemporaryDirectoryState

  • one_off_build (bool) – Set to True if you are testing a one-off build.

  • instance_vars (dict[str, str] | None) – Pass optional instance vars to emulate an instanced pipeline.

  • env_vars (str) – Pass additional environment variables, or overload the default ones.

Caution

If the image does not exist in the local cache, a RuntimeError will be raised.

Note

The working directory is explicitly set to / within the container to ensure that the resource is properly accounting for the paths it is passed.

fetch_new_versions(previous_version_config=None)[source]

Fetch new versions of the resource.

Calls the /opt/resource/check script within the Docker container using run_docker_container().

Caution

No environment variables are available to the check script.

Parameters:

previous_version_config (dict[str, str] | None) – The JSON configuration of the most recent version of the resource. This will be set to None if the resource has never been run before.

Return type:

list[dict[str, str]]

Returns:

A list of new version configurations.

download_version(version_config, params=None)[source]

Download a version and place its files within the resource directory in your pipeline.

Calls the /opt/resource/in script within the Docker container using run_docker_container().

Parameters:

  • version_config (dict[str, str]) – The JSON configuration of the version.

  • params (dict[str, Any] | None) – A mapping of additional keyword parameters passed to the inner resource.

Return type:

tuple[dict[str, str], list[MetadataPair]]

Returns:

The version configuration (most likely unchanged), and a list of metadata pairs.

publish_new_version(params=None)[source]

Update a resource by publishing a new version.

Calls the /opt/resource/out script within the Docker container using run_docker_container().

Parameters:

params (dict[str, Any] | None) – A mapping of additional keyword parameters passed to the inner resource.

Return type:

tuple[dict[str, str], list[MetadataPair]]

Returns:

The new version configuration, and a list of metadata pairs.

class concoursetools.testing.DockerConversionTestResourceWrapper(inner_resource_type, inner_resource_config, image, directory_dict=None, one_off_build=False, instance_vars=None, **env_vars)[source]

A resource wrapper based on DockerTestResourceWrapper, yet with the interface of SimpleTestResourceWrapper.

Inputs are converted to JSON and passed as in DockerTestResourceWrapper. The container only persists for the duration of the method call.

Tip

This is best to use if you want easier testing, but are concerned about the Dockerfile.

Added in version 0.8.0.

Parameters:

  • inner_resource_type (type[ConcourseResource[Version]]) – The ConcourseResource subclass corresponding to the resource.

  • inner_resource_config (dict[str, Any]) – The JSON configuration for the resource.

  • image (str) – The Docker image to use, which must exist in the local cache. Passed verbatim to docker run.

  • directory_dict (dict[str, Any] | None) – The initial state of the resource directory. See TemporaryDirectoryState

  • one_off_build (bool) – Set to True if you are testing a one-off build.

  • instance_vars (dict[str, str] | None) – Pass optional instance vars to emulate an instanced pipeline.

  • env_vars (str) – Pass additional environment variables, or overload the default ones.

fetch_new_versions(previous_version=None)[source]

Fetch new versions of the resource.

Converts the version (if it exists) to JSON and then invokes fetch_new_versions(). The response is then converted back to Version instances.

Parameters:

previous_version (Optional[Version]) – The most recent version of the resource. This will be set to None if the resource has never been run before.

Return type:

list[Version]

Returns:

A list of new versions.

download_version(version, **params)[source]

Download a version and place its files within the resource directory in your pipeline.

Converts the version to JSON and then invokes download_version() with the additional params as a dict. The returned version configuration is then converted back to a Version instance, and the metadata pairs converted to a standard dict.

Parameters:

  • version (Version) – The version to be downloaded.

  • params (object) – Additional keyword parameters passed to the inner resource.

Return type:

tuple[Version, dict[str, str]]

Returns:

The version (most likely unchanged), and a dictionary of metadata.

publish_new_version(**params)[source]

Update a resource by publishing a new version.

Converts the version to JSON and then invokes publish_new_version() with the additional params as a dict. The returned version configuration is then converted back to a Version instance, and the metadata pairs converted to a standard dict.

Parameters:

params (object) – Additional keyword parameters passed to the inner resource.

Return type:

tuple[Version, dict[str, str]]

Returns:

The new version, and a dictionary of metadata.

Running External Commands

concoursetools.testing.run_command(command, additional_args=None, env=None, cwd=None, stdin=None)[source]

Run an external command.

Parameters:

  • command (str) – The external command to be run.

  • additional_args (list[str] | None) – Additional arguments to be passed to the command.

  • env (dict[str, str] | None) – Environment variables to be made available to the command.

  • cwd (Path | None) – The working directory of the command. Defaults to current working directory.

  • stdin (str | None) – A string to be passed on stdin.

Return type:

tuple[str, str]

Returns:

The stdout and stderr of the command.

Raises:

RuntimeError – If the external command exits with a non-zero exit code.

Seealso:

This function is broadly equivalent to subprocess.run().

concoursetools.testing.run_script(script_path, additional_args=None, env=None, cwd=None, stdin=None)[source]

Run an external script.

Parameters:

  • script_path (Path) – The path to the script to be run.

  • additional_args (list[str] | None) – Additional arguments to be passed to the script.

  • env (dict[str, str] | None) – Environment variables to be made available to the script.

  • cwd (Path | None) – The working directory of the script. Defaults to current working directory.

  • stdin (str | None) – A string to be passed on stdin.

Return type:

tuple[str, str]

Returns:

The stdout and stderr of the script.

Raises:
Seealso:

This function will call run_command().

concoursetools.testing.run_docker_container(image, command, additional_args=None, env=None, cwd=None, stdin=None, rm=True, interactive=True, dir_mapping=None, hostname=None, local_only=True)[source]

Run a command within the Docker container.

Caution

Mounted directory paths are not checked for actually being directories.

Danger

Directories are not mounted in “read-only” mode.

Caution

Parameters of this function are meant to refer to the command within the Docker container, and not the external command used to run the image.

Added in version 0.8.0.

Parameters:

  • image (str) – The Docker image to use for the container, which must exist in the local cache. Passed verbatim to docker run.

  • command (str) – The command to be passed to docker run. Can also be a path to a script within the container.

  • additional_args (list[str] | None) – Additional arguments to pass to the command.

  • env (dict[str, str] | None) – Environment variables to be made available to the script.

  • cwd (Path | None) – Pass a path within the container to set the working directory, or else use the image default.

  • stdin (str | None) – A string to be passed on stdin.

  • rm (bool) – Set to True to automatically remove the container when it exits. Equivalent to passing --rm.

  • interactive (bool) – Set to True to keep stdin open even if not attached. Equivalent to passing -i or --interactive.

  • dir_mapping (dict[Path, Path | str] | None) – A mapping of directories to paths to mount within the container. Values can be paths or strings.

  • hostname (str | None) – Specify a hostname inside the container. Defaults to the container ID.

  • local_only (bool) – When set to True (default), only locally cached images can be used.

Return type:

tuple[str, str]

Returns:

The stdout and stderr of the script.

Raises:

RuntimeError – If the external script exits with a non-zero exit code.

Seealso:

This function will call run_command().