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 (Optional[Dict[str, Any]]) – 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 (Optional[Dict[str, str]]) – 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 (Any) – 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 (Any) – 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 (Optional[Dict[str, Any]]) – 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 (Optional[Dict[str, str]]) – 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 (Optional[Dict[str, str]]) – 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 versions 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 (Optional[Dict[str, Any]]) – A mapping of additional keyword parameters passed to the inner resource.

Return type:

Tuple[Dict[str, str], List[Dict[Literal['name', 'value'], str]]]

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 (Optional[Dict[str, Any]]) – A mapping of additional keyword parameters passed to the inner resource.
Return type:: Tuple[Dict[str, str], List[Dict[Literal['name', 'value'], str]]]
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 (Optional[Dict[str, Any]]) – 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 (Optional[Dict[str, str]]) – 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 (Any) – 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 (Any) – 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:

inner_resource_config (Dict[str, Any]) – The JSON configuration for the resource.
check_script (Optional[Path]) – The path to the external script for the check. Setting to None (default) means that fetch_new_versions() raises NotImplementedError.
in_script (Optional[Path]) – The path to the external script for the check. Setting to None (default) means that download_version() raises NotImplementedError.
out_script (Optional[Path]) – The path to the external script for the check. Setting to None (default) means that publish_new_version() raises NotImplementedError.
directory_dict (Optional[Dict[str, Any]]) – 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 (Optional[Dict[str, str]]) – Pass optional instance vars to emulate an instanced pipeline.
env_vars (str) – Pass additional environment variables, or overload the default ones.

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 capture_output_from_script() with the correct environment.

Caution

No environment variables are available to the check script.

Parameters:: previous_version_config (Optional[Dict[str, str]]) – 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 versions 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 capture_output_from_script() with the correct environment.

Parameters:

version_config (Dict[str, str]) – The JSON configuration of the version.
params (Optional[Dict[str, Any]]) – A mapping of additional keyword parameters passed to the inner resource.

Return type:

Tuple[Dict[str, str], List[Dict[Literal['name', 'value'], str]]]

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 capture_output_from_script() with the correct environment.

Parameters:: params (Optional[Dict[str, Any]]) – A mapping of additional keyword parameters passed to the inner resource.
Return type:: Tuple[Dict[str, str], List[Dict[Literal['name', 'value'], str]]]
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 (Optional[Dict[str, Any]]) – 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 (Optional[Dict[str, str]]) – Pass optional instance vars to emulate an instanced pipeline.
env_vars (str) – Pass additional environment variables, or overload the default ones.

Return type:

FileTestResourceWrapper

capture_output_from_script(script_path, stdin, additional_argv, env, cwd=PosixPath('.'))[source]#

Run an external script and capture the output.

The script is run using subprocess.run().

Parameters:

script_path (Path) – The location of the script to run.
stdin (str) – A string to be passed on stdin.
additional_argv (List[str]) – Additional strings to pass as sys.argv. The first argument is always the script path.
env (Dict[str, str]) – Environment variables to be made available to the script. PYTHONPATH is added by the method.
cwd (Union[Path, str]) – The working directory of the script. Defaults to current working directory.

Return type:

Tuple[str, str]

Returns:

The stdout and stderr of the script.

Raises:

FileNotFoundError – If the script path can not be resolved.
RuntimeError – If the external script exists with a non-zero exit code.

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:

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.
executable (str) – The executable to use for the script (at the top). (See create_script_file().)
permissions (int) – The (Linux) permissions the file should have. Defaults to rwxr-xr-x. (See create_script_file().)
encoding (Optional[str]) – The encoding of the file as passed to write_text(). Setting to None (default) will use the user’s default encoding. (See create_script_file().)
directory_dict (Optional[Dict[str, Any]]) – 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 (Optional[Dict[str, str]]) – 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, 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 (Any) – 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 (Any) – Additional keyword parameters passed to the inner resource.
Return type:: Tuple[Version, Dict[str, str]]
Returns:: The new version, and a dictionary of metadata.