taucmdr.util module

Utility functions.

Handles system manipulation and status tasks, e.g. subprocess management or file creation.

taucmdr.util._ARCHIVE_ERRORS = (<class 'tarfile.TarError'>, <class 'EOFError'>, <class 'zlib.error'>, <class '_lzma.LZMAError'>)

Exceptions raised while reading a corrupt or truncated archive, such as an interrupted download.

Type:

tuple

taucmdr.util._cleanup_dtemp()

At exit, keep copies of failed builds’ temporary directories, then delete all temporary directories.

taucmdr.util._create_dl_subprocess(abs_cmd, src, dest, timeout)

taucmdr.util._extract_all(archive, dest, members=None)

Extract members of the open tar file archive into dest.

Parameters:

• archive (tarfile.TarFile) – Open tar file.

• dest (str) – Destination folder.

• members – Members to extract, or None for all of them.

taucmdr.util._null_context(label)

taucmdr.util._rmtree(path, ignore_errors, onexc)

Call shutil.rmtree with an onexc handler on any Python version.

taucmdr.util._show_extract_progress(members)

taucmdr.util.add_error_stack(path)

taucmdr.util.archive_toplevel(archive)

Returns the name of the top-level directory in an archive.

Assumes that the archive file is rooted in a single top-level directory::

foo

/bar /baz

The top-level directory here is “foo” This routine will return stupid results for archives with multiple top-level elements.

Parameters:

archive (str) – Path to archive file.

Raises:

IOError – archive could not be read.

Returns:

Directory name.

Return type:

str

taucmdr.util.calculate_uid(parts)

Create a new unique identifier.

Parameters:

parts (list) – Ordered list of strings to include in the UID calculation.

Returns:

A string of hexadecimal digits uniquely calculated from parts.

Return type:

str

taucmdr.util.camelcase(name)

Converts a string to CamelCase.

Parameters:

name (str) – String to convert.

Returns:

name in CamelCase.

Return type:

str

taucmdr.util.cleanup_dtemp(path)

Delete a temporary directory now instead of when the program exits.

Parameters:

path (tempfile.TemporaryDirectory) – A directory from mkdtemp or tempfile.TemporaryDirectory.

taucmdr.util.color_text(text, *args, **kwargs)

Use termcolor.colored to colorize text.

Parameters:

• text (str) – Text to colorize.

• *args – Positional arguments to pass to termcolor.colored.

• **kwargs – Keyword arguments to pass to termcolor.colored.

Returns:

The colorized text.

Return type:

str

taucmdr.util.copy_file(src, dest, show_progress=True)

Works just like shutil.copy except with progress bars.

taucmdr.util.create_archive(fmt, dest, items, cwd=None, show_progress=True)

Creates a new archive file in the specified format.

Parameters:

• fmt (str) – Archive fmt, e.g. ‘zip’ or ‘tgz’.

• dest (str) – Path to the archive file that will be created.

• items (list) – Items (i.e. files or folders) to add to the archive.

• cwd (str) – Current working directory while creating the archive.

taucmdr.util.create_subprocess(cmd, cwd=None, env=None, stdout=True, log=True, show_progress=False, error_buf=50, record_output=False)

Create a subprocess.

See subprocess.Popen.

Parameters:

• cmd (list) – Command and its command line arguments.

• cwd (str) – If not None, change directory to cwd before creating the subprocess.

• env (dict) – Environment variables to set or unset before launching cmd.

• stdout (bool) – If True send subprocess stdout and stderr to this processes’ stdout.

• log (bool) – If True send subprocess stdout and stderr to the debug log.

• error_buf (int) – If non-zero, stdout is not already being sent, and return value is non-zero then send last error_buf lines of subprocess stdout and stderr to this processes’ stdout.

• record_output (bool) – If True return output.

Returns:

Subprocess return code.

Return type:

int

taucmdr.util.download(src, dest, timeout=8)

Downloads or copies files.

src may be a file path or URL. The destination folder will be created if it doesn’t exist. Download is via curl, wget, or Python’s urllib as appropriate.

Parameters:

• src (str) – Path or URL to source file.

• dest (str) – Path to file copy or download destination.

• timeout (int) – Maximum time in seconds for the connection to the server. 0 for no timeout.

Raises:

IOError – File copy or download failed.

taucmdr.util.extract_archive(archive, dest, show_progress=True)

Extracts archive file to dest.

Supports compressed and uncompressed tar archives. Destination folder will be created if it doesn’t exist.

Parameters:

• archive (str) – Path to archive file to extract.

• dest (str) – Destination folder.

Returns:

Full path to extracted files.

Return type:

str

Raises:

OSError – Failed to extract archive, or the archive has members that would be unsafe to extract.

taucmdr.util.get_binary_linkage(cmd)

taucmdr.util.get_command_output(cmd)

Return the possibly cached output of a command.

Just subprocess.check_output with a cache. Subprocess stderr is always sent to subprocess stdout.

Parameters:

cmd (list) – Command and its command line arguments.

Raises:

subprocess.CalledProcessError – return code was non-zero.

Returns:

Subprocess output.

Return type:

str

taucmdr.util.hline(title, *args, **kwargs)

Build a colorful horizontal rule for console output.

Uses logger.LINE_WIDTH to generate a string of ‘=’ characters as wide as the terminal. title is included in the string near the left of the horizontal line.

Parameters:

• title (str) – Text to put on the horizontal rule.

• *args – Positional arguments to pass to termcolor.colored.

• **kwargs – Keyword arguments to pass to termcolor.colored.

Returns:

The horizontal rule.

Return type:

str

taucmdr.util.human_size(num, suffix='B')

Converts a byte count to human readable units.

Parameters:

• num (int) – Number to convert.

• suffix (str) – Unit suffix, e.g. ‘B’ for bytes.

Returns:

num as a human readable string.

Return type:

str

taucmdr.util.is_clean_container(obj)

Recursively checks a container for bytes, bytearray and memory view objects in keys and values. Care must be taken to avoid infinite loops caused by cycles in a dictionary with cycles.

Parameters:

obj (any) – Container to be checked for disallowed binary data

Returns:

Whether or not bad binary entries were found

Return type:

bool

taucmdr.util.is_url(url)

Check if url is a URL.

Parameters:

url (str) – String to check

Returns:

True if url is a URL, False otherwise.

Return type:

bool

taucmdr.util.mkdirp(*args)

Creates a directory and all its parents.

Works just like mkdir -p.

Parameters:

*args – Paths to create.

taucmdr.util.mkdtemp(*args, **kwargs)

Like tempfile.mkdtemp but directory will be recursively deleted when program exits.

Returns:

The new directory. Pass it to cleanup_dtemp to delete it sooner.

Return type:

tempfile.TemporaryDirectory

taucmdr.util.page_output(output_string: str)

Pipe string to a pager.

If PAGER is an environment then use that as pager, otherwise use less.

Parameters:

output_string (str) – String to put output.

taucmdr.util.parse_bool(value, additional_true=None, additional_false=None)

Parses a value to a boolean value.

If value is a string try to interpret it as a bool: * [‘1’, ‘t’, ‘y’, ‘true’, ‘yes’, ‘on’] ==> True * [‘0’, ‘f’, ‘n’, ‘false’, ‘no’, ‘off’] ==> False Otherwise raise TypeError.

Parameters:

• value – value to parse to a boolean.

• additional_true (list) – optional additional string values that stand for True.

• additional_false (list) – optional additional string values that stand for False.

Returns:

True if value is true, False if value is false.

Return type:

bool

Raises:

ValueError – value does not parse.

taucmdr.util.path_accessible(path, mode='r')

Check if a file or directory exists and is accessible.

Files are checked by attempting to open them with the given mode. Directories are checked by testing their access bits only, which may fail for some filesystems which may have permissions semantics beyond the usual POSIX permission-bit model. We’ll fix this if it becomes a problem.

Parameters:

• path (str) – Path to file or directory to check.

• mode (str) – File access mode to test, e.g. ‘r’ or ‘rw’

Returns:

True if the file exists and can be opened in the specified mode, False otherwise.

taucmdr.util.rmtree(path, ignore_errors=False, onexc=None, attempts=5)

Wrapper around shutil.rmtree to work around stale or slow NFS directories.

Tries repeatedly to recursively remove path and sleeps between attempts.

Parameters:

• path (str) – A directory but not a symbolic link to a directory.

• ignore_errors (bool) – If True then errors resulting from failed removals will be ignored. If False or omitted, such errors are handled by calling a handler specified by onexc or, if that is omitted, they raise an exception.

• onexc – Callable that accepts function, path, and the exception. See shutil.rmtree.

• attempts (int) – Number of times to repeat shutil.rmtree before giving up.

taucmdr.util.tmpfs_prefix()

Path to a uniquely named directory in a temporary filesystem, ideally a ramdisk.

/dev/shm is the preferred tmpfs, but if it’s unavailable or mounted with noexec then fall back to tempfile.gettempdir(), which is usually /tmp. If that filesystem is also unavailable then use the filesystem prefix of the highest writable storage container.

An attempt is made to ensure that there is at least 2GiB of free space in the selected filesystem.

Returns:

A uniquely-named directory in the temporary filesystem. The directory

and all its contents will be deleted when the program exits if it installs correctly.

Return type:

tempfile.TemporaryDirectory

taucmdr.util.umask(new_mask)

Context manager to temporarily set the process umask.

Parameters:

new_mask – The argument to os.umask.

taucmdr.util.uncolor_text(text)

Remove color control chars from a string.

Parameters:

text (str) – Text to colorize.

Returns:

The text without control chars.

Return type:

str

taucmdr.util.verify_archive(archive)

Check that an archive can be read to its end.

Reading only an archive’s first members doesn’t catch an archive that was cut short, such as by an interrupted copy, so this reads the whole file.

Parameters:

archive (str) – Path to archive file.

Raises:

IOError – archive could not be read.

taucmdr.util.which(program, use_cached=True)

Returns the full path to a program command.

Program must exist and be executable. Searches the system PATH and the current directory. Caches the result.

Parameters:

• program (str) – program to find.

• use_cached (bool) – If False then don’t use cached results.

Returns:

Full path to program or None if program can’t be found.

Return type:

str