conu - Container Utilities for image testing

conu gathers utilities that come in handy when creating tests, provides nice logging for troubleshooting and is easily extensible.

Installation

There are multiple ways of installing conu.

Fedora

dnf install python{2,3}-conu

From PyPI

pip install --user conu

From git

Clone the upstream git repository:

git clone https://github.com/user-cont/conu

If using Fedora, use the provided helper script to install dependencies:

sh conu/requirements.sh

When not using Fedora, you should figure out the dependencies on your own.

And finally, install conu into python sitelib:

pip3 install --user ./conu

Examples

Run a container

For running a container, you need to have image initialized. Image holds information of repository and tag and provides other methods. To run image using docker binary (as user would) use conu.apidefs.image.Image.run_via_binary() method with conu.backend.docker.container.DockerRunBuilder as parameter.

Wait for service to be ready

conu.backend.docker.container.DockerContainer.wait_for_port() tries to reach 8080 till it’s opened. You can use your own timeout to limit time spent on waiting.

Extend image using source-to-image

Extends acts as s2i binary. It extends builder image in form of conu.backend.docker.image.S2IDockerImage using provided source and desired name of resulting image.

    container.delete()

Run image in pod

Run image inside k8s conu.backend.k8s.pod.Pod

    namespace = k8s_backend.create_namespace()

    with DockerBackend(logging_level=logging.DEBUG) as backend:
        image = backend.ImageClass("openshift/hello-openshift")

        pod = image.run_in_pod(namespace=namespace)

        try:
            pod.wait(200)
            assert pod.is_ready()
            assert pod.get_phase() == PodPhase.RUNNING
        finally:
            pod.delete()
            assert pod.get_phase() == PodPhase.TERMINATING
            k8s_backend.delete_namespace(namespace)

Deploy new application in OpenShift using remote source

Build and deploy new application in OpenShift using centos/python-36-centos7 image and remote source.

    python_image = openshift_backend.import_image("python-36-centos7",
                                                  "docker.io/centos/python-36-centos7")

    # create new app from remote source in OpenShift cluster
    app_name = openshift_backend.create_new_app_from_source(
        python_image,
        source="https://github.com/openshift/django-ex.git",
        project='myproject')

    try:
        # wait until service is ready to accept requests

        openshift_backend.wait_for_service(
            app_name=app_name,
            port=8080,
            expected_output='Welcome to your Django application on OpenShift',
            timeout=300)
    finally:
        openshift_backend.get_logs(app_name)
        openshift_backend.clean_project(app_name)

conu API Reference

API definition

Image

Image represents abstract API for image managers. It is now implemented by conu.DockerImage

class conu.apidefs.image.Image(image_reference, tag=None)

A class which represents an arbitrary container image. It contains utility methods to manipulate it.

__init__(image_reference, tag=None)

Parameters:

• image_reference – str, the reference to this image (usually name)
• tag – str, tag of the image, when not specified, “latest” is implied

create_container(container_params)

create a container using this image

Parameters:container_params – instance of ContainerParameters Returns:instance of Container

directory_is_present(directory_path)

check if directory specified via ‘directory_path’ is present inside the image; this method raises ConuException if the path exists but is not a directory

Parameters:directory_path – str, directory to check Returns:True if directory exists, False if directory does not exist

file_is_present(file_path)

check if file specified via ‘file_path’ is present in the image

Parameters:file_path – str, path to the file Returns:True if file exists, False if the file is missing

get_full_name()

provide full, complete image name

Returns:str

get_id()

get unique identifier of this image

Returns:str

get_metadata()

return general metadata for image

Returns:ImageMetadata

get_selinux_context(file_path)

return a permissions for ‘file_path’

Parameters:file_path – str, path to the file Returns:str

inspect(refresh=False)

return cached metadata by default

Parameters:refresh – bool, update the metadata with up to date content Returns:dict

classmethod load_from_file(file_path)

load Image from provided file

Parameters:file_path – str, path to the file Returns:Image instance

mount(mount_point=None)

mount image filesystem

Parameters:mount_point – str, directory where the filesystem will be mounted Returns:instance of Filesystem

mount_image(mount_point=None)

mount an image to host system :param mount_point: str, mount_point on host system :return: mount_point

pull()

pull this image

Returns:None

rmi(force=False, via_name=False)

remove selected image

Parameters:

• image – str, image name, example: “fedora:latest”
• force – bool, force removal of the image
• via_name – bool, refer to the image via name, if false, refer via ID

Returns:

None

run_in_pod(namespace='default')

run image inside Kubernetes Pod :param namespace: str, name of namespace where pod will be created :return: Pod instance

run_via_api(container_params)

create a container using this image and run it in the background

Parameters:container_params – instance of ContainerParameters Returns:instance of Container

run_via_binary(*args, **kwargs)

create a container using this image and run it in the background; this method is useful to test real user scenarios when users invoke containers using binary and not an API

Parameters:image – instance of Image Returns:instance of Container

Image represents abstract API for s2i images. It is now implemented by conu.S2IDockerImage

class conu.apidefs.image.S2Image

Additional functionality related to s2i-enabled container images

extend(source, new_image_name, s2i_args=None)

extend this s2i-enabled image using provided source, raises ConuException if s2i build fails

Parameters:

• source – str, source used to extend the image, can be path or url
• new_image_name – str, name of the new, extended image
• s2i_args – list of str, additional options and arguments provided to s2i build

Returns:

S2Image instance

usage()

Provide output of s2i usage

Returns:str

Container

Container represents abstract API for container managers. It is now implemented by conu.backend.docker.image.DockerContainer

class conu.apidefs.container.Container(image, container_id, name)

Container class definition which contains abstract methods. The instances should call the constructor

__init__(image, container_id, name)

Parameters:

• image – Image instance
• container_id – str, unique identifier of this container
• container_id – str, pretty container name

copy_from(src, dest)

copy a file or a directory from container to host system

Parameters:

• src – str, path to a file or a directory within container or image
• dest – str, path to a file or a directory on host system

Returns:

None

copy_to(src, dest)

copy a file or a directory from host system to a container

Parameters:

• src – str, path to a file or a directory on host system
• dest – str, path to a file or a directory within container

Returns:

None

delete(force=False, **kwargs)

remove this container; kwargs indicate that some container runtimes might accept more parameters

Parameters:force – bool, if container engine supports this, force the functionality Returns:None

execute(command, **kwargs)

execute a command in this container

Parameters:

• command – list of str, command to execute in the container
• kwargs – specific parameters for container engines exec methods

Returns:

str (output) or iterator

exit_code()

get exit code of container. Return value is 0 for running and created containers

Returns:int

get_IPv4s()

Return all known IPv4 addresses of this container. It may be possible that the container has disabled networking: in that case, the list is empty

Returns:list of str

get_IPv6s()

Return all known IPv6 addresses of this container. It may be possible that the container has disabled networking: in that case, the list is empty

Returns:list of str

get_id()

get unique identifier of this container

Returns:str

get_image_name()

return name of the container image

Returns:str

get_metadata()

return general metadata for container

Returns:ContainerMetadata

get_pid()

get process identifier of the root process in the container

Returns:int

get_ports()

get ports specified in container metadata

Returns:list of str

get_status()

Get status of container

Returns:Status of container

http_client(host=None, port=None)

allow requests in context – e.g.:

with container.http_client(port="80", ...) as c:
    assert c.get("/api/...")

Parameters:

• host – str, if None, set self.get_IPv4s()[0]
• port – str or int, if None, set to self.get_ports()[0]

Returns:

instance of conu.utils.http_client.HttpClient

http_request(path='/', method='GET', host=None, port=None, json=False, data=None)

perform a HTTP request

Parameters:

• path – str, path within the reqest, e.g. “/api/version”
• method – str, HTTP method
• host – str, if None, set self.get_IPv4s()[0]
• port – str or int, if None, set to self.get_ports()[0]
• json – bool, should we expect json?
• data – data to send (can be dict, list, str)

Returns:

dict

inspect(refresh=False)

return cached metadata by default

Parameters:refresh – bool, returns up to date metadata if set to True Returns:dict

is_port_open(port, timeout=10)

check if given port is open and receiving connections

Parameters:

• port – int
• timeout – int, how many seconds to wait for connection; defaults to 2

Returns:

True if the connection has been established inside timeout, False otherwise

is_running()

returns True if the container is running, this method should always ask the API and should not use a cached value

Returns:bool

kill(signal=None)

send a signal to this container (bear in mind that the process won’t have time to shutdown properly and your service may end up in an inconsistent state)

Parameters:signal – str or int, signal to use for killing the container (SIGKILL by default) Returns:None

logs(follow=False)

Get logs from this container.

Parameters:follow – bool, provide new logs as they come Returns:iterator

mount(mount_point=None)

mount container filesystem

Parameters:mount_point – str, directory where the filesystem will be mounted Returns:instance of Filesystem

name()

Return name of this container.

Returns:str

open_connection(port=None)

open a TCP connection to service running in the container, if port is None and container exposes only a single port, connect to it, otherwise raise an exception

Parameters:port – int or None Returns:socket

start()

start current container - the container has to be created

Returns:None

status()

Provide current, up-to-date status of this container. This method should not use cached value. Implementation of this method should clearly state list of possible values to get from this method

Returns:str

stop()

stop this container

Returns:None

wait(timeout)

Block until the container stops, then return its exit code.

Parameters:timeout – int, Request timeout Returns:int, exit code

Filesystem

Filesystem represents abstract API definition. It is now implemented by conu.DockerContainerViaExportFS

class conu.apidefs.filesystem.Filesystem(object_instance, mount_point=None)

Utility methods used to access filesystem of containers and images.

Implementations should probably be done using context managers.

__init__(object_instance, mount_point=None)

Parameters:

• object_instance – instance of the container or image
• mount_point – str, directory where the filesystem will be mounted

copy_from(src, dest)

copy a file or a directory from container or image to host system. If you are copying directories, the target directory must not exist (this function is using shutil.copytree to copy directories and that’s a requirement of the function). In case the directory exists, OSError on python 2 or FileExistsError on python 3 are raised.

Parameters:

• src – str, path to a file or a directory within container or image
• dest – str, path to a file or a directory on host system

Returns:

None

copy_to(src, dest)

copy a file or a directory from host system to a container – don’t implement for images, those are immutable

Parameters:

• src – str, path to a file or a directory on host system
• dest – str, path to a file or a directory within container

Returns:

None

directory_is_present(directory_path)

check if directory ‘directory_path’ is present, raise IOError if it’s not a directory

Parameters:directory_path – str, directory to check Returns:True if directory exists, False if directory does not exist

file_is_present(file_path)

check if file ‘file_path’ is present, raises IOError if file_path is not a file

Parameters:file_path – str, path to the file Returns:True if file exists, False if file does not exist

get_file(file_path, mode='r')

provide File object specified via ‘file_path’

Parameters:

• file_path – str, path to the file
• mode – str, mode used when opening the file

Returns:

File instance

get_selinux_context(file_path)

Get SELinux file context of the selected file.

Parameters:file_path – str, path to the file Returns:str, name of the SELinux file context

p(path)

provide absolute path within the container

Parameters:path – path with container Returns:str

read_file(file_path)

read file specified via ‘file_path’ and return its content - raises an ConuException if there is an issue accessing the file

Parameters:file_path – str, path to the file to read Returns:str (not bytes), content of the file

Metadata

Metadata represents common metadata of containers and images.

class conu.apidefs.metadata.Metadata(name=None, identifier=None, labels=None, command=None, creation_timestamp=None, env_variables=None)

Common metadata for container and image

__init__(name=None, identifier=None, labels=None, command=None, creation_timestamp=None, env_variables=None)

Parameters:

• name – str, name of object
• identifier – str, id of object
• labels – dict, {key: value}
• command – list of str, command to run in the container, example: - [“psql”, “–version”] - [“python3”, “-m”, “http.server”, “–bind”, “0.0.0.0”, “8080”]
• creation_timestamp – str, creation time of object instance
• env_variables – dict, {name: value}

Container specific metadata.

class conu.apidefs.metadata.ContainerMetadata(name=None, identifier=None, labels=None, command=None, creation_timestamp=None, env_variables=None, image=None, exposed_ports=None, port_mappings=None, hostname=None, ipv4_addresses=None, ipv6_addresses=None, status=None)

Specific metadata for container

__init__(name=None, identifier=None, labels=None, command=None, creation_timestamp=None, env_variables=None, image=None, exposed_ports=None, port_mappings=None, hostname=None, ipv4_addresses=None, ipv6_addresses=None, status=None)

Parameters:

• name – str, name of container
• identifier – str, id of container
• labels – dict, {key: value}
• command – list of str, command to run in the container, example: - [“psql”, “–version”] - [“python3”, “-m”, “http.server”, “–bind”, “0.0.0.0”, “8080”]
• creation_timestamp – str, creation time of container
• env_variables – dict, {name: value}
• image – Image, reference to Image instance
• exposed_ports – list, list of exposed ports
• port_mappings –

  dict, dictionary of port mappings {“container_port”: [host_port1]},

  Example:

  • {“1111/tcp”:[1234, 4567]} bind host ports 1234 and 4567

    to a single container port 1111/tcp

• hostname – str, hostname
• ipv4_addresses – dict, {address: port}
• ipv6_addresses – dict, {address: port}
• status – ContainerStatus, container status value, example: - [ContainerStatus.RUNNING] - [ContainerStatus.EXITED]

Image specific metadata.

class conu.apidefs.metadata.ImageMetadata(name=None, identifier=None, labels=None, command=None, creation_timestamp=None, env_variables=None, exposed_ports=None, image_names=None, digest=None, repo_digests=None)

Specific metadata for image

__init__(name=None, identifier=None, labels=None, command=None, creation_timestamp=None, env_variables=None, exposed_ports=None, image_names=None, digest=None, repo_digests=None)

Parameters:

• name – str, name of image
• identifier – str, id of image
• labels – dict, {key: value} example: - {“io.k8s.display-name”: “nginx”}
• command – list of str, command to run in the container, example: - [“psql”, “–version”] - [“python3”, “-m”, “http.server”, “–bind”, “0.0.0.0”, “8080”]
• creation_timestamp – str, creation time of image
• env_variables – dict, {name: value}, example: - {“MYSQL_PASSWORD”: “password”}
• exposed_ports – list, list of exposed ports
• image_names – list of str, image names, examples: - [fedora, docker.io/library/fedora:latest]
• digest – str, hash of the image
• repo_digests – str, unique pull specification based on digest

class conu.apidefs.backend.Backend(logging_level=20, logging_kwargs=None, cleanup=None)

This class groups classes and functionality related to a specific backend.

We strongly advise you to use backend as a context manager:

with SomeBackend() as backend:
    image = backend.ImageClass(...)

When entering the context manager, the backend will create a new temporary directory. You can use it if you want, the path is stored in attribute tmpdir of the backend instance. Some backend implementations use this temporary directory to store some short-lived runtime files (e.g. container-id file in case of docker). Once the context manager goes out of scope, this temporary directory is removed. If you don’t use the backend class as a context manager, the temporary directory isn’t removed and therefore lingers.

ContainerClass

alias of conu.apidefs.container.Container

ImageClass

alias of conu.apidefs.image.Image

__init__(logging_level=20, logging_kwargs=None, cleanup=None)

This method serves as a configuration interface for conu.

Parameters:

• logging_level – int, control logger verbosity: see logging.{DEBUG,INFO,ERROR}
• logging_kwargs – dict, additional keyword arguments for logger set up, for more info see docstring of set_logging function
• cleanup – list, list of cleanup policy values, examples: - [CleanupPolicy.EVERYTHING] - [CleanupPolicy.VOLUMES, CleanupPolicy.TMP_DIRS] - [CleanupPolicy.NOTHING]

cleanup_containers()

Remove containers associated with this backend instance

Returns:None

cleanup_images()

Remove images associated with this backend instance

Returns:None

cleanup_volumes()

Remove volumes associated with this backend instance

Returns:None

list_containers()

list all available containers for this backend

Returns:collection of instances of conu.apidefs.container.Container

list_images()

list all available images for this backend

Returns:collection of instances of conu.apidefs.image.Image

Backend - Docker

Docker Container

Aside from methods in API definition - conu.apidefs.container.Container, DockerContainer implements following methods:

class conu.DockerContainer(image, container_id, name=None, popen_instance=None)

__init__(image, container_id, name=None, popen_instance=None)

Parameters:

• image – DockerImage instance (if None, it will be found from the container itself)
• container_id – str, unique identifier of this container
• name – str, pretty container name
• popen_instance – instance of Popen (if container was created using method via_binary, this is the docker client process)

copy_from(src, dest)

copy a file or a directory from container or image to host system.

Parameters:

• src – str, path to a file or a directory within container or image
• dest – str, path to a file or a directory on host system

Returns:

None

copy_to(src, dest)

copy a file or a directory from host system to a container

Parameters:

• src – str, path to a file or a directory on host system
• dest – str, path to a file or a directory within container

Returns:

None

delete(force=False, volumes=False, **kwargs)

remove this container; kwargs indicate that some container runtimes might accept more parameters

Parameters:

• force – bool, if container engine supports this, force the functionality
• volumes – bool, remove also associated volumes

Returns:

None

execute(command, blocking=True, exec_create_kwargs=None, exec_start_kwargs=None)

Execute a command in this container – the container needs to be running.

If the command fails, a ConuException is thrown.

This is a blocking call by default and writes output of the command to logger using the INFO level – this behavior can be changed if you set the argument blocking to False.

If not blocking, you should consume the returned iterator in order to see logs or know when the command finished:

for line in container.execute(["ping", "-c", "4", "8.8.8.8"], blocking=False):
    print(line)
print("command finished")

Parameters:

• command – list of str, command to execute in the container
• blocking – bool, if True blocks until the command finishes
• exec_create_kwargs – dict, params to pass to exec_create()
• exec_start_kwargs – dict, params to pass to exec_start()

Returns:

iterator if non-blocking or list of bytes if blocking

exit_code()

get exit code of container. Return value is 0 for running and created containers

Returns:int

get_IPv4s()

Return all known IPv4 addresses of this container. It may be possible that the container has disabled networking: in that case, the list is empty

Returns:list of str

get_IPv6s()

Return all known IPv6 addresses of this container. It may be possible that the container has disabled networking: in that case, the list is empty

Returns:list of str

get_id()

get unique identifier of this container

Returns:str

get_image_name()

return name of the container image

Returns:str

get_metadata()

Convert dictionary returned after docker inspect command into instance of ContainerMetadata class :return: ContainerMetadata, container metadata instance

get_port_mappings(port=None)

Get list of port mappings between container and host. The format of dicts is:

{“HostIp”: XX, “HostPort”: YY};

When port is None - return all port mappings. The container needs to be running, otherwise this returns an empty list.

Parameters:port – int or None, container port Returns:list of dict or None; dict when port=None

get_ports()

get ports specified in container metadata

Returns:list of str

get_status()

Get status of container

Returns:one of: ‘created’, ‘restarting’, ‘running’, ‘paused’, ‘exited’, ‘dead’

inspect(refresh=True)

return cached metadata by default

Parameters:refresh – bool, returns up to date metadata if set to True Returns:dict

is_port_open(port, timeout=2)

check if given port is open and receiving connections on container ip_address

Parameters:

• port – int, container port
• timeout – int, how many seconds to wait for connection; defaults to 2

Returns:

True if the connection has been established inside timeout, False otherwise

is_running()

returns True if the container is running, this method should always ask the API and should not use a cached value

Returns:bool

kill(signal=None)

send a signal to this container (bear in mind that the process won’t have time to shutdown properly and your service may end up in an inconsistent state)

Parameters:signal – str or int, signal to use for killing the container (SIGKILL by default) Returns:None

logs(follow=False)

Get logs from this container. Every item of the iterator contains one log line terminated with a newline. The logs are encoded (they are bytes, not str).

Let’s look at an example:

image = conu.DockerImage("fedora", tag="27")
command = ["bash", "-c", "for x in `seq 1 5`; do echo $x; sleep 1; done"]
container = image.run_via_binary(command=command)
for line in container.logs(follow=True):
    print(line)

This will output

b'1\n'
b'2\n'
b'3\n'
b'4\n'
b'5\n'

Parameters:follow – bool, provide new logs as they come Returns:iterator (of bytes)

logs_in_bytes()

Get output of container in bytes.

Returns:bytes

logs_unicode()

Get output of container decoded using utf-8.

Returns:str

mount(mount_point=None)

mount container filesystem

Parameters:mount_point – str, directory where the filesystem will be mounted Returns:instance of DockerContainerViaExportFS

start()

start current container - the container has to be created

Returns:None

stop()

stop this container

Returns:None

wait(timeout=None)

Block until the container stops, then return its exit code. Similar to the docker wait command.

Parameters:timeout – int, Request timeout Returns:int, exit code

wait_for_port(port, timeout=10, **probe_kwargs)

block until specified port starts accepting connections, raises an exc ProbeTimeout if timeout is reached

Parameters:

• port – int, port number
• timeout – int or float (seconds), time to wait for establishing the connection
• probe_kwargs – arguments passed to Probe constructor

Returns:

None

write_to_stdin(message)

Write provided text to container’s standard input. In order to make this function work, there needs to be several conditions met:

• the container needs to be running
• the container needs to have stdin open
• the container has to be created using method run_via_binary_in_foreground

For more info see documentation in run_via_binary_in_foreground()

Parameters:message – str or bytes, text to be written to container standard input

class conu.DockerRunBuilder(command=None, additional_opts=None)

helper to execute docker run – users can easily change or override anything

__init__(command=None, additional_opts=None)

Build docker run command

Parameters:

• command – list of str, command to run in the container, examples: - [“ls”, “/”] - [“bash”, “-c”, “ls / | grep bin”]
• additional_opts – list of str, additional options for docker run

get_parameters()

Parse DockerRunBuilder options and create object with properties for docker-py run command :return: DockerContainerParameters

class conu.DockerContainerViaExportFS(container, mount_point=None)

__init__(container, mount_point=None)

Provide container as an archive

Parameters:

• container – instance of DockerContainer
• mount_point – str, directory where the filesystem will be made available

Docker Image

Aside from methods in API definition - conu.apidefs.image.Image, DockerImage implements following methods:

class conu.DockerImage(repository, tag='latest', identifier=None, pull_policy=<DockerImagePullPolicy.IF_NOT_PRESENT: 1>)

Utility functions for docker images.

These methods are specific to this backend:

• DockerImage.build()

These generic methods are not implemented in this backend:

• conu.apidefs.image.Image.load_from_file()

__init__(repository, tag='latest', identifier=None, pull_policy=<DockerImagePullPolicy.IF_NOT_PRESENT: 1>)

Parameters:

• repository – str, image name, examples: “fedora”, “registry.fedoraproject.org/fedora”, “tomastomecek/sen”, “docker.io/tomastomecek/sen”
• tag – str, tag of the image, when not specified, “latest” is implied
• identifier – str, unique identifier for this image
• pull_policy – enum, strategy to apply for pulling the image

classmethod build(path, tag=None, dockerfile=None)

Build the image from the provided dockerfile in path

:param path : str, path to the directory containing the Dockerfile :param tag: str, A tag to add to the final image :param dockerfile: str, path within the build context to the Dockerfile :return: instance of DockerImage

copy(repository=None, tag=None, source_transport=None, target_transport=<SkopeoTransport.DOCKER: 2>, source_path=None, target_path=None, logs=True)

Copy this image

:param repository to be copied to :param tag :param source_transport Transport :param target_transport Transport :param source_path needed to specify for dir, docker-archive or oci transport :param target_path needed to specify for dir, docker-archive or oci transport :param logs enable/disable logs :return: the new DockerImage

get_full_name()

Provide full, complete image name

Returns:str

get_id()

get unique identifier of this image

Returns:str

get_layer_ids(rev=True)

Get IDs of image layers

Parameters:rev – get layers reversed Returns:list of strings

get_metadata()

Provide metadata about this image.

Returns:ImageMetadata, Image metadata instance

static get_volume_options(volumes)

Generates volume options to run methods.

Parameters:volumes – tuple or list of tuples in form target x source,target x source,target,mode. Returns:list of the form [“-v”, “/source:/target”, “-v”, “/other/source:/destination:z”, …]

has_pkgs_signed_with(allowed_keys)

Check signature of packages installed in image. Raises exception when

• rpm binary is not installed in image
• parsing of rpm fails
• there are packages in image that are not signed with one of allowed keys

Parameters:allowed_keys – list of allowed keys Returns:bool

inspect(refresh=True)

provide metadata about the image; flip refresh=True if cached metadata are enough

Parameters:refresh – bool, update the metadata with up to date content Returns:dict

is_present()

Is this docker image present locally on the system?

Returns:bool, True if it is, False if it’s not

layers(rev=True)

Get list of DockerImage for every layer in image

Parameters:rev – get layers rev Returns:list of DockerImages

load_from(image)

Load from another DockerImage to this one

Parameters:image – Returns:

mount(mount_point=None)

Provide access to filesystem of this docker image.

Parameters:mount_point – str, directory where the filesystem will be mounted; if not provided, mkdtemp(dir=”/var/tmp”) is used Returns:instance of conu.apidefs.filesystem.Filesystem

pull()

Pull this image from registry. Raises an exception if the image is not found in the registry.

Returns:None

push(repository=None, tag=None)

Push image to registry. Raise exception when push fail. :param repository: str, see constructor :param tag: str, see constructor :return: None

rmi(force=False, via_name=False)

remove this image

Parameters:

• force – bool, force removal of the image
• via_name – bool, refer to the image via name, if false, refer via ID

Returns:

None

run_in_pod(namespace='default')

run image inside Kubernetes Pod :param namespace: str, name of namespace where pod will be created :return: Pod instance

run_via_api(container_params=None)

create a container using this image and run it in background via Docker-py API. https://docker-py.readthedocs.io/en/stable/api.html Note: If you are using Healthchecks, be aware that support of some options were introduced just with version of Docker-py API 1.29 :param container_params: DockerContainerParameters :return: instance of DockerContainer

run_via_binary(run_command_instance=None, command=None, volumes=None, additional_opts=None, **kwargs)

create a container using this image and run it in background; this method is useful to test real user scenarios when users invoke containers using binary

Parameters:

• run_command_instance – instance of DockerRunBuilder
• command – list of str, command to run in the container, examples: - [“ls”, “/”] - [“bash”, “-c”, “ls / | grep bin”]
• volumes –

  tuple or list of tuples in the form:

  • (“/path/to/directory”, )
  • (“/host/path”, “/container/path”)
  • (“/host/path”, “/container/path”, “mode”)

  • (conu.Directory(‘/host/path’), “/container/path”) (source can be also

    Directory instance)

• additional_opts – list of str, additional options for docker run

Returns:

instance of DockerContainer

run_via_binary_in_foreground(run_command_instance=None, command=None, volumes=None, additional_opts=None, popen_params=None, container_name=None)

Create a container using this image and run it in foreground; this method is useful to test real user scenarios when users invoke containers using binary and pass input into the container via STDIN. You are also responsible for:

• redirecting STDIN when intending to use container.write_to_stdin afterwards by setting

  popen_params={“stdin”: subprocess.PIPE} during run_via_binary_in_foreground

• checking whether the container exited successfully via:

  container.popen_instance.returncode

Please consult the documentation for subprocess python module for best practices on how you should work with instance of Popen

Parameters:

• run_command_instance – instance of DockerRunBuilder
• command – list of str, command to run in the container, examples: - [“ls”, “/”] - [“bash”, “-c”, “ls / | grep bin”]
• volumes –

  tuple or list of tuples in the form:

  • (“/path/to/directory”, )
  • (“/host/path”, “/container/path”)
  • (“/host/path”, “/container/path”, “mode”)

  • (conu.Directory(‘/host/path’), “/container/path”) (source can be also

    Directory instance)

• additional_opts – list of str, additional options for docker run
• popen_params – dict, keyword arguments passed to Popen constructor
• container_name – str, pretty container identifier

Returns:

instance of DockerContainer

save_to(image)

Save this image to another DockerImage

Parameters:image – DockerImage Returns:

skopeo_pull()

Pull image from Docker to local Docker daemon using skopeo

Returns:pulled image

skopeo_push(repository=None, tag=None)

Push image from Docker daemon to Docker using skopeo

Parameters:

• repository – repository to be pushed to
• tag – tag

Returns:

pushed image

tag_image(repository=None, tag=None)

Apply additional tags to the image or even add a new name

Parameters:

• repository – str, see constructor
• tag – str, see constructor

Returns:

instance of DockerImage

using_transport(transport=None, path=None, logs=True)

change used transport

Parameters:

• transport – from where will be this image copied
• path – in filesystem
• logs – enable/disable

Returns:

self

class conu.DockerImagePullPolicy

This Enum defines the policy for pulling the docker images. The pull operation happens when creating an instance of a docker image. Supported values:

• NEVER - do not pull the image
• IF_NOT_PRESENT - pull it only if the image is not present
• ALWAYS - always initiate the pull process - the image is being pulled even if it’s present locally. It means that it may be overwritten by a remote counterpart or there may be a exception being raised if no such image is present in the registry.

class conu.DockerImageViaArchiveFS(image, mount_point=None)

__init__(image, mount_point=None)

Provide image as an archive

Parameters:

• image – instance of DockerImage
• mount_point – str, directory where the filesystem will be made available

Aside from methods in API definition - conu.apidefs.image.S2Image, S2IDockerImage implements following methods:

class conu.S2IDockerImage(repository, tag='latest', identifier=None, pull_policy=<DockerImagePullPolicy.IF_NOT_PRESENT: 1>)

These methods are specific to this backend:

• S2IDockerImage.build()

These generic methods are not implemented in this backend:

• conu.apidefs.image.Image.load_from_file()

extend(source, new_image_name, s2i_args=None)

extend this s2i-enabled image using provided source, raises ConuException if s2i build fails

Parameters:

• source – str, source used to extend the image, can be path or url
• new_image_name – str, name of the new, extended image
• s2i_args – list of str, additional options and arguments provided to s2i build

Returns:

S2Image instance

usage()

Provide output of s2i usage

Returns:str

class conu.DockerBackend(logging_level=20, logging_kwargs=None, cleanup=None)

Bases: conu.apidefs.backend.Backend

For more info on using the Backend classes, see documentation of the parent conu.apidefs.backend.Backend class.

ContainerClass

alias of conu.backend.docker.container.DockerContainer

ImageClass

alias of conu.backend.docker.image.DockerImage

__init__(logging_level=20, logging_kwargs=None, cleanup=None)

This method serves as a configuration interface for conu.

Parameters:

• logging_level – int, control logger verbosity: see logging.{DEBUG,INFO,ERROR}
• logging_kwargs – dict, additional keyword arguments for logger set up, for more info see docstring of set_logging function
• cleanup – list, list of cleanup policy values, examples: - [CleanupPolicy.EVERYTHING] - [CleanupPolicy.VOLUMES, CleanupPolicy.TMP_DIRS] - [CleanupPolicy.NOTHING]

cleanup_containers()

Remove containers associated with this backend instance

Returns:None

cleanup_images()

Remove images associated with this backend instance

Returns:None

cleanup_volumes()

Remove volumes associated with this backend instance

Returns:None

list_containers()

List all available docker containers.

Container objects returned from this methods will contain a limited amount of metadata in property short_metadata. These are just a subset of .inspect(), but don’t require an API call against dockerd.

Returns:collection of instances of conu.DockerContainer

list_images()

List all available docker images.

Image objects returned from this methods will contain a limited amount of metadata in property short_metadata. These are just a subset of .inspect(), but don’t require an API call against dockerd.

Returns:collection of instances of conu.DockerImage

login(username, password=None, email=None, registry=None, reauth=False, dockercfg_path=None)

Parameters:

• username – The registry username
• password – The plaintext password
• email – The email for the registry account
• registry – URL to the registry, example: - https://index.docker.io/v1/
• reauth – Whether or not to refresh existing authentication on the Docker server.
• dockercfg_path – Use a custom path for the Docker config file

Returns:

None

Backend - Kubernetes

Kubernetes Pod

class conu.backend.k8s.pod.Pod(namespace, name=None, spec=None, from_template=None)

__init__(namespace, name=None, spec=None, from_template=None)

Utility functions for kubernetes pods.

Parameters:

• namespace – str, namespace in which is pod created
• name – name of pod
• spec –

  pod spec

  • https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1PodSpec.md
• from_template –

  str, pod template, example:

  • https://kubernetes.io/docs/concepts/workloads/pods/pod-overview/#pod-templates

static create(image_data)

Parameters:image_data – ImageMetadata Returns:V1Pod, https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1Pod.md

delete()

delete pod from the Kubernetes cluster :return: None

get_conditions()

get conditions through which the pod has passed :return: list of PodCondition enum or empty list

get_ip()

get IP address of Pod :return: str, IP address or empty string if is not allocated yet

get_logs()

print logs from pod :return: str or None

get_phase()

get phase of the pod :return: PodPhase enum

get_status()

get status of the Pod :return: V1PodStatus, https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1PodStatus.md

is_ready()

Check if pod is in READY condition :return: bool

wait(timeout=15)

block until pod is not ready, raises an exc ProbeTimeout if timeout is reached :param timeout: int or float (seconds), time to wait for pod to run :return: None

class conu.backend.k8s.pod.PodPhase

https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/#pod-phase Additional values in conu: TERMINATING - phase right after delete() method is called on pod

class conu.backend.k8s.pod.PodCondition

https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/#pod-conditions

Kubernetes Service

class conu.backend.k8s.service.Service(name, ports, namespace='default', labels=None, selector=None, create_in_cluster=False, spec=None)

__init__(name, ports, namespace='default', labels=None, selector=None, create_in_cluster=False, spec=None)

Utility functions for kubernetes services.

Parameters:

• name – str, name of the service
• namespace – str, name of the namespace
• ports – list of str, list of exposed ports, example: - [‘1234/tcp’, ‘8080/udp’]
• labels – dict, dict of labels
• selector – dict, route service traffic to pods with label keys and values matching this selector

create_in_cluster()

call Kubernetes API and create this Service in cluster, raise ConuExeption if the API call fails :return: None

delete()

delete service from the Kubernetes cluster :return: None

get_ip()

get IP adress of service :return: str, IP address

get_status()

get status of service :return: V1ServiceStatus, https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1ServiceStatus.md

Kubernetes Deployment

class conu.backend.k8s.deployment.Deployment(name=None, selector=None, labels=None, image_metadata=None, namespace='default', create_in_cluster=False, from_template=None)

__init__(name=None, selector=None, labels=None, image_metadata=None, namespace='default', create_in_cluster=False, from_template=None)

Utility functions for kubernetes deployments.

Parameters:

• name – str, name of the deployment
• selector – Label selector for pods. Existing ReplicaSets whose pods are selected by this will be the ones affected by this deployment. It must match the pod template’s labels
• labels – dict, dict of labels
• image_metadata – ImageMetadata
• namespace – str, name of the namespace
• create_in_cluster – bool, if True deployment is created in Kubernetes cluster
• from_template – str, deployment template, example: - https://kubernetes.io/docs/concepts/workloads/controllers/deployment/

all_pods_ready()

Check if number of replicas with same selector is equals to number of ready replicas :return: bool

create_in_cluster()

call Kubernetes API and create this Deployment in cluster, raise ConuException if the API call fails :return: None

delete()

delete Deployment from the Kubernetes cluster :return: None

get_status()

get status of the Deployment :return: V1DeploymentStatus, https://git.io/vhKE3

wait(timeout=15)

block until all replicas are not ready, raises an exc ProbeTimeout if timeout is reached :param timeout: int or float (seconds), time to wait for pods to run :return: None

class conu.backend.k8s.backend.K8sBackend(api_key=None, logging_level=20, logging_kwargs=None, cleanup=None)

Bases: conu.apidefs.backend.Backend

ContainerClass

alias of conu.apidefs.container.Container

ImageClass

alias of conu.apidefs.image.Image

__init__(api_key=None, logging_level=20, logging_kwargs=None, cleanup=None)

This method serves as a configuration interface for conu.

Parameters:

• api_key – str, Bearer API token
• logging_level – int, control logger verbosity: see logging.{DEBUG,INFO,ERROR}
• logging_kwargs – dict, additional keyword arguments for logger set up, for more info see docstring of set_logging function
• cleanup – list, list of k8s cleanup policy values, examples: - [CleanupPolicy.EVERYTHING] - [CleanupPolicy.PODS, CleanupPolicy.SERVICES] - [CleanupPolicy.NOTHING]

cleanup_containers()

Remove containers associated with this backend instance

Returns:None

cleanup_deployments()

Delete all deployments created in namespaces associated with this backend :return: None

cleanup_images()

Remove images associated with this backend instance

Returns:None

cleanup_namespaces()

Delete all namespaces created by this backend :return: None

cleanup_pods()

Delete all pods created in namespaces associated with this backend :return: None

cleanup_services()

Delete all services created in namespaces associated with this backend :return: None

cleanup_volumes()

Remove volumes associated with this backend instance

Returns:None

create_namespace()

Create namespace with random name :return: name of new created namespace

delete_namespace(name)

Delete namespace with specific name :param name: str, namespace to delete :return: None

list_containers()

list all available containers for this backend

Returns:collection of instances of conu.apidefs.container.Container

list_deployments(namespace=None)

List all available deployments.

Parameters:namespace – str, if not specified list deployments for all namespaces Returns:collection of instances of conu.backend.k8s.deployment.Deployment

list_images()

list all available images for this backend

Returns:collection of instances of conu.apidefs.image.Image

list_pods(namespace=None)

List all available pods.

Parameters:namespace – str, if not specified list pods for all namespaces Returns:collection of instances of conu.backend.k8s.pod.Pod

list_services(namespace=None)

List all available services.

Parameters:namespace – str, if not specified list services for all namespaces Returns:collection of instances of conu.backend.k8s.service.Service

class conu.backend.k8s.backend.K8sCleanupPolicy

This Enum defines the policy for cleanup.

• NOTHING - clean nothing

• EVERYTHING - delete just objects in all namespaces

  associated with this backend - (pods, service, deployments)

• NAMESPACES - delete all namespaces associated with this backend and

  objects in these namespaces (pods, service, deployments)

• PODS - delete all pods in namespaces associated with this backend
• SERVICES - delete all services in namespaces associated with this backend
• DEPLOYMENTS - delete all deployments in namespaces associated with this backend

Backend - OpenShift

class conu.backend.origin.backend.OpenshiftBackend(api_key=None, logging_level=20, logging_kwargs=None, project=None)

Bases: conu.backend.k8s.backend.K8sBackend

ContainerClass

alias of conu.apidefs.container.Container

ImageClass

alias of conu.apidefs.image.Image

__init__(api_key=None, logging_level=20, logging_kwargs=None, project=None)

This method serves as a configuration interface for conu.

Parameters:

• api_key – str, Bearer API token
• logging_level – int, control logger verbosity: see logging.{DEBUG,INFO,ERROR}
• logging_kwargs – dict, additional keyword arguments for logger set up, for more info see docstring of set_logging function
• project – str, project name that will be used while working with this backend. It is possible to specify it later, when deploying app. One instance of OpenshiftBackend should work with just one project at time.

all_pods_are_ready(app_name)

Check if all pods are ready for specific app :param app_name: str, name of the app :return: bool

clean_project(app_name=None, delete_all=False)

Delete objects in current project in OpenShift cluster. If both parameters are passed, delete all objects in project. :param app_name: str, name of app :param delete_all: bool, if true delete all objects in current project :return: None

cleanup_containers()

Remove containers associated with this backend instance

Returns:None

cleanup_deployments()

Delete all deployments created in namespaces associated with this backend :return: None

cleanup_images()

Remove images associated with this backend instance

Returns:None

cleanup_namespaces()

Delete all namespaces created by this backend :return: None

cleanup_pods()

Delete all pods created in namespaces associated with this backend :return: None

cleanup_services()

Delete all services created in namespaces associated with this backend :return: None

cleanup_volumes()

Remove volumes associated with this backend instance

Returns:None

create_app_from_template(image_name, name, template, name_in_template, other_images=None, oc_new_app_args=None, project=None)

Helper function to create app from template

Parameters:

• image_name – image to be used as builder image
• name – name of app from template
• template – str, url or local path to a template to use
• name_in_template – dict, {repository:tag} image name used in the template
• other_images – list of dict, some templates need other image to be pushed into the OpenShift registry, specify them in this parameter as list of dict [{<image>:<tag>}], where “<image>” is image name with tag and “<tag>” is a tag under which the image should be available in the OpenShift registry.
• oc_new_app_args – additional parameters for the oc new-app
• project – project where app should be created, default: current project

Returns:

None

create_namespace()

Create namespace with random name :return: name of new created namespace

create_new_app_from_source(image_name, project=None, source=None, oc_new_app_args=None)

Deploy app using source-to-image in OpenShift cluster using ‘oc new-app’ :param image_name: image to be used as builder image :param project: project where app should be created, default: current project :param source: source used to extend the image, can be path or url :param oc_new_app_args: additional parameters for the oc new-app :return: str, name of the app

delete_namespace(name)

Delete namespace with specific name :param name: str, namespace to delete :return: None

deploy_image(image_name, oc_new_app_args=None, project=None, name=None)

Deploy image in OpenShift cluster using ‘oc new-app’ :param image_name: image name with tag :param oc_new_app_args: additional parameters for the oc new-app, env variables etc. :param project: project where app should be created, default: current project :param name:str, name of application, if None random name is generated :return: str, name of the app

get_current_project()

Get name of current project using oc project command. Raise ConuException in case of an error. :return: str, project name

get_image_registry_url(image_name)

Helper function for obtain registry url of image from it’s name

Parameters:image_name – str, short name of an image, example: - conu:0.5.0 Returns:str, image registry url, example: - 172.30.1.1:5000/myproject/conu:0.5.0

get_logs(name)

Obtain cluster status and logs from all pods and print them using logger. This method is useful for debugging. :param name: str, name of app generated by oc new-app :return: str, cluster status and logs from all pods

get_status()

Get status of OpenShift cluster, similar to oc status :return: str

http_request(path='/', method='GET', host=None, port=None, json=False, data=None)

perform a HTTP request

Parameters:

• path – str, path within the request, e.g. “/api/version”
• method – str, HTTP method
• host – str, if None, set to 127.0.0.1
• port – str or int, if None, set to 8080
• json – bool, should we expect json?
• data – data to send (can be dict, list, str)

Returns:

dict

import_image(imported_image_name, image_name)

Import image using oc import-image command.

Parameters:

• imported_image_name – str, short name of an image in internal registry, example: - hello-openshift:latest
• image_name – full repository name, example: - docker.io/openshift/hello-openshift:latest

Returns:

str, short name in internal registry

list_containers()

list all available containers for this backend

Returns:collection of instances of conu.apidefs.container.Container

list_deployments(namespace=None)

List all available deployments.

Parameters:namespace – str, if not specified list deployments for all namespaces Returns:collection of instances of conu.backend.k8s.deployment.Deployment

list_images()

list all available images for this backend

Returns:collection of instances of conu.apidefs.image.Image

list_pods(namespace=None)

List all available pods.

Parameters:namespace – str, if not specified list pods for all namespaces Returns:collection of instances of conu.backend.k8s.pod.Pod

list_services(namespace=None)

List all available services.

Parameters:namespace – str, if not specified list services for all namespaces Returns:collection of instances of conu.backend.k8s.service.Service

request_service(app_name, port, expected_output=None)

Make request on service of app. If there is connection error function return False.

Parameters:

• app_name – str, name of the app
• expected_output – str, If not None method will check output returned from request and try to find matching string.
• port – str or int, port of the service

Returns:

bool, True if connection was established False if there was connection error

start_build(build, args=None)

Start new build, raise exception if build failed :param build: str, name of the build :param args: list of str, another args of ‘oc start-build’ commands :return: None

wait_for_service(app_name, port, expected_output=None, timeout=100)

Block until service is not ready to accept requests, raises an exc ProbeTimeout if timeout is reached

Parameters:

• app_name – str, name of the app
• port – str or int, port of the service
• expected_output – If not None method will check output returned from request and try to find matching string.
• timeout – int or float (seconds), time to wait for pod to run

Returns:

None

pytest fixtures

This submodule contains pytest fixtures which can be utilized when writing tests for your containers while using conu and pytest.

conu.fixtures.buildah_backend()

pytest fixture which mimics context manager: it provides new instance of BuildahBackend and cleans after it once it’s used; behaves the same as docker_backend fixture

Returns:instance of BuildahBackend

conu.fixtures.docker_backend()

pytest fixture which mimics context manager: it provides new instance of DockerBackend and cleans after it once it’s used; sample usage:

def test_my_container(docker_backend):
    image = docker_backend.ImageClass("fedora", tag="27")

Returns:instance of DockerBackend

conu.fixtures.podman_backend()

pytest fixture which mimics context manager: it provides new instance of PodmanBackend and cleans after it once it’s used; behaves the same as docker_backend fixture

Returns:instance of PodmanBackend

Helper functions

You can find some high-level functions defined in this submodule.

conu.helpers.get_container_output(backend, image_name, command, image_tag='latest', additional_opts=None)

Create a throw-away container based on provided image and tag, run the supplied command in it and return output. The container is stopped and removed after it exits.

Parameters:

• backend – instance of DockerBackend
• image_name – str, name of the container image
• command – list of str, command to run in the container
• image_tag – str, container image tag, defaults to “latest”
• additional_opts – list of str, by default this function creates the container using docker binary and run command; with this argument you can supply addition options to the “docker run” invocation

Returns:

str (unicode), output of the container

Utilities

Filesystem utilities

class conu.Directory(path, mode=None, user_owner=None, group_owner=None, facl_rules=None, selinux_context=None, selinux_user=None, selinux_role=None, selinux_type=None, selinux_range=None)

This class allows you to do advanced operations on filesystem directories, think of it as mkdir on steroids.

We advise you to use it as a context manager:

with Directory("/funky/path", mode=0o0700) as directory:
    path = os.path.join(directory.path, "my-dir")

The directory is being removed once leaving the context. You can also easily do it on your own:

directory = Directory("/funky/path", mode=0o0700)
try:
    directory.initialize()
finally:
    directory.clean()

This class utilizes CLI tools to perform some operations. If some of them is missing, the exception is raised.

__init__(path, mode=None, user_owner=None, group_owner=None, facl_rules=None, selinux_context=None, selinux_user=None, selinux_role=None, selinux_type=None, selinux_range=None)

For more info on SELinux, please see $ man chcon. An exception will be thrown if selinux_context is specified and at least one of other SELinux fields.

Parameters:

• path – str, path to the directory we will operate on
• mode – int, octal representation of permission bits, e.g. 0o0400
• user_owner – str or int, uid or username to own the directory
• group_owner – str or int, gid or group name to own the directory
• facl_rules – list of str, file ACLs to apply, e.g. “u:26:rwx”
• selinux_context – str, set directory to this SELinux context (this is the full context with all the field, example: “system_u:object_r:unlabeled_t:s0”)
• selinux_user – str, user in the target security context, e.g. “system_u”
• selinux_role – str, role in the target security context, e.g. “object_r”
• selinux_type – str, type in the target security context, e.g. “unlabeled_t”
• selinux_range – str, range in the target security context, e.g. “s0”

clean()

remove the directory we operated on

Returns:None

initialize()

create the directory if needed and configure it

Returns:None

Probe

class conu.Probe(timeout=1, pause=1, count=-1, expected_exceptions=(), expected_retval=True, fnc=<class 'bool'>, **kwargs)

Probe can be used for waiting on specific result of a function. Probe ends when function returns expected_retval or timeout is exceeded.

__init__(timeout=1, pause=1, count=-1, expected_exceptions=(), expected_retval=True, fnc=<class 'bool'>, **kwargs)

Parameters:

• timeout – Number of seconds spent on trying. Set timeout to -1 for infinite run.
• pause – Number of seconds waited between multiple function result checks
• count – Maximum number of tries, defaults to infinite, represented by -1
• expected_exceptions – When one of expected_exception is raised, probe ignores it and tries to run function again. To ignore multiple exceptions use parenthesized tuple.
• expected_retval – When expected_retval is received, probe ends successfully
• fnc – Function which run is checked by probe

Other helper functions and utilities

exception conu.utils.CommandDoesNotExistException

Requested command is not present on the system

conu.utils.are_we_root()

is uid of current process 0?

Returns:True if root, else otherwise

conu.utils.check_buildah_command_works()

Verify that buildah binary works fine by calling buildah version

Returns:bool, True if all is good, otherwise ConuException or CommandDoesNotExistException is thrown

conu.utils.check_docker_command_works()

Verify that dockerd and docker binary works fine. This is performed by calling docker version, which also checks server API version.

Returns:bool, True if all is good, otherwise ConuException or CommandDoesNotExistException is thrown

conu.utils.check_podman_command_works()

Verify that podman binary works fine by calling podman version

Returns:bool, True if all is good, otherwise ConuException or CommandDoesNotExistException is thrown

conu.utils.check_port(port, host, timeout=10)

connect to port on host and return True on success

Parameters:

• port – int, port to check
• host – string, host address
• timeout – int, number of seconds spent trying

Returns:

bool

conu.utils.command_exists(command, noop_invocation, exc_msg)

Verify that the provided command exists. Raise CommandDoesNotExistException in case of an error or if the command does not exist.

Parameters:

• command – str, command to check (python 3 only)
• noop_invocation – list of str, command to check (python 2 only)
• exc_msg – str, message of exception when command does not exist

Returns:

bool, True if everything’s all right (otherwise exception is thrown)

conu.utils.convert_kv_to_dict(data)

convert text values in format: key1=value1 key2=value2 to dict {‘key1’:’value1’, ‘key2’:’value2’}

Parameters:data – string containing lines with these values Returns:dict

conu.utils.export_docker_container_to_directory(client, container, path)

take selected docker container, create an archive out of it and unpack it to a selected location

Parameters:

• client – instance of docker.APIClient
• container – instance of DockerContainer
• path – str, path to a directory, doesn’t need to exist

Returns:

None

conu.utils.get_oc_api_token()

Get token of user logged in OpenShift cluster :return: str, API token

conu.utils.get_selinux_status()

get SELinux status of host

Returns:string, one of Enforced, Permissive, Disabled

conu.utils.graceful_get(d, *args, default=None)

Obtain values from dicts and lists gracefully. Example:

print(graceful_get({"a": [{1: 2}, {"b": "c"}]}, "a", "b"))
c

Parameters:

• d – collection (usually a dict or list)
• args – list of keys which are used as a lookup
• default – default value to return in case of an axc

Returns:

the value from your collection

conu.utils.is_oc_cluster_running()

Check status of OpenShift cluster :return: bool, True if cluster is running otherwise False

conu.utils.is_selinux_disabled()

check if SELinux is disabled

Returns:bool, True if disabled, False otherwise

conu.utils.mkdtemp()

calls tempfile.mkdtemp, the temporary directory is prefixed with ‘conu-‘

Returns:str, path to the directory

conu.utils.mkstemp(dir=None)

calls tempfile.mkstemp, the temporary file is prefixed with ‘conu-‘

Parameters:dir – str, path to dir where the temporary file should be created Returns:tuple, (fd, filename)

conu.utils.parse_reference(reference)

parse provided image reference into <image_repository>:<tag>

Parameters:reference – str, e.g. (registry.fedoraproject.org/fedora:27) Returns:collection (tuple or list), (“registry.fedoraproject.org/fedora”, “27”)

conu.utils.random_str(size=10)

create random string of selected size

Parameters:size – int, length of the string Returns:the string

conu.utils.random_tmp_filename()

generate string which can be used as a filename for temporary file

conu.utils.run_cmd(cmd, return_output=False, ignore_status=False, log_output=True, **kwargs)

run provided command on host system using the same user as you invoked this code, raises subprocess.CalledProcessError if it fails

Parameters:

• cmd – list of str
• return_output – bool, return output of the command
• ignore_status – bool, do not fail in case nonzero return code
• log_output – bool, if True, log output to debug log
• kwargs – pass keyword arguments to subprocess.check_* functions; for more info, please check help(subprocess.Popen)

Returns:

None or str

How to release conu

We are using awesome release-bot for new conu releases. If you want to make new release:

• create new issue in Github upstream repository with title x.y.z release and wait for release bot to create new PR.
• polish CHANGELOG.md and merge if tests are passing.
• Sit down, relax and watch how release bot is doing the hard work.

Contribution

In case of issues, questions and suggestions, please, file an issue or pull request on https://github.com/user-cont/conu where conu is stored and maintained. Before you open issue or pull request please read our contribution guide https://github.com/user-cont/conu/blob/master/CONTRIBUTING.md .

How to make a new gif

• Recording: asciinema rec -w 1 -c=ipython3 conu.json (needs asciinema pip3/dnf package)
• Making a gif: docker run –rm -v $PWD:/data:z asciinema/asciicast2gif -s 2 conu.json conu.gif

(Cut the size with -w <columns> -h <rows>.)

Scenario:

from conu import DockerBackend
backend = DockerBackend()


image = backend.ImageClass('docker.io/library/nginx')

container = image.run_via_binary()

assert container.is_running()
container.get_IPv4s()
container.get_ports()

resp = container.http_request(port=80)
assert resp.ok
resp.text

container.stop()
container.is_running()

############ clean screen ###############

from conu import DockerRunBuilder

run_params = DockerRunBuilder(additional_opts=['-e HELLO=hello'], command=['env'])
container = image.run_via_binary(run_params)
for l in container.logs():
    print(l)


############ clean screen ###############

with container.mount() as fs:
    assert fs.file_is_present('/etc/nginx/nginx.conf')