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_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. This is performed by calling podman version, which also checks server API 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)

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

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