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