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