Archive

Ar

Module to read UNIX ar files

autils.archive.ar.MAGIC = b'!<arch>\n': The first eight bytes of all AR archives

autils.archive.ar.FILE_HEADER_FMT = '16s12s6s6s8s10s2c': The header for each file in the archive

class autils.archive.ar.ArMember(identifier, size, offset)[source]

Bases: object

A member of an UNIX ar archive.

class autils.archive.ar.Ar(path)[source]

Bases: object

An UNIX ar archive iterator.

It reads ar archive file and iterates over the members in form of autils.archive.ar.ArMember

is_valid()[source]

Checks if a file looks like an AR archive.

Returns:: If file looks like an AR archive
Return type:: bool

list()[source]

List the members in the archive.

Returns:: List the names of the mebers in the archive
Return type:: list

read_member(identifier)[source]

Reads the data for the given member identifier.

Parameters:: identifier (str) – Archive member name.
Returns:: data for the given member
Return type:: bytes or None

Data Structures

This module contains handy classes that can be used inside your code or plugins.

exception autils.data_structures.InvalidDataSize[source]

Bases: ValueError

Signals that the value given to DataSize is not valid.

This exception is raised when an invalid data size string is provided to the DataSize class constructor.

autils.data_structures.ordered_list_unique(object_list)[source]

Returns an unique list of objects, with their original order preserved.

This function removes duplicates from a list while maintaining the original order of the first occurrence of each element.

Parameters:: object_list (list) – List of objects that may contain duplicates
Returns:: List with duplicates removed, order preserved
Return type:: list

Example:

>>> ordered_list_unique([1, 2, 2, 3, 1, 4])
[1, 2, 3, 4]

autils.data_structures.geometric_mean(values)[source]

Evaluates the geometric mean for a list of numeric values.

This implementation is slower but allows unlimited number of values. The geometric mean is calculated as the nth root of the product of n numbers.

Parameters:: values (list) – List with numeric values
Returns:: Single value representing the geometric mean for the list values, or None if the list is empty
Return type:: float or None
Raises:: ValueError – If any value in the list cannot be converted to int
See:: http://en.wikipedia.org/wiki/Geometric_mean

Example:

>>> geometric_mean([1, 2, 4, 8])
2.8284271247461903

autils.data_structures.compare_matrices(matrix1, matrix2, threshold=0.05)[source]

Compare 2 matrices nxm and return a matrix nxm with comparison data and stats.

When the first columns match, they are considered as header and included in the results intact. This function is useful for comparing performance data between different test runs.

Parameters:

matrix1 (list of lists) – Reference Matrix of floats; first column could be header
matrix2 (list of lists) – Matrix that will be compared; first column could be header
threshold (float) – Any difference greater than this percent threshold will be reported (default: 0.05 = 5%)

Returns:

Tuple containing: - Matrix with the difference in comparison - Number of improvements - Number of regressions - Total number of comparisons

Return type:

tuple(list, int, int, int)

Example:

>>> matrix1 = [['test1', 10.0, 20.0]]
>>> matrix2 = [['test1', 12.0, 18.0]]
>>> result = compare_matrices(matrix1, matrix2)
>>> # Returns comparison matrix and statistics

autils.data_structures.comma_separated_ranges_to_list(string)[source]

Provides a list from comma separated ranges.

Converts a string containing comma-separated ranges into a list of integers. Ranges can be specified as single numbers or as ranges using hyphens.

Parameters:: string (str) – String of comma separated range (e.g., “1,3-5,7”)
Returns:: List of integer values in comma separated range
Return type:: list of int
Raises:: ValueError – If the string contains invalid range format

Example:

>>> comma_separated_ranges_to_list("1,3-5,7")
[1, 3, 4, 5, 7]
>>> comma_separated_ranges_to_list("10-12")
[10, 11, 12]

autils.data_structures.recursive_compare_dict(dict1, dict2, level='DictKey', diff_btw_dict=None)[source]

Finds difference between two dictionaries.

Recursively compares two dictionaries and returns a list of differences. The function handles nested structures and provides detailed difference information.

Parameters:

dict1 (dict, list, or any) – First dictionary to compare
dict2 (dict, list, or any) – Second dictionary to compare
level (str) – Current level identifier for nested comparison
diff_btw_dict (list or None) – List to store differences (used internally for recursion)

Returns:

List of differences between the two dictionaries, or None for recursive calls

Return type:

list or None

Example:

>>> dict1 = {'a': 1, 'b': {'c': 2}}
>>> dict2 = {'a': 2, 'b': {'c': 2}}
>>> differences = recursive_compare_dict(dict1, dict2)
>>> # Returns list of differences

class autils.data_structures.Borg[source]

Bases: object

Multiple instances of this class will share the same state.

This is considered a better design pattern in Python than more popular patterns, such as the Singleton. The Borg pattern allows multiple instances to exist but they all share the same state, making them effectively equivalent. Inspired by Alex Martelli’s article mentioned below.

All instances of this class will have the same __dict__, so any changes to instance variables will be reflected across all instances.

See:: http://www.aleax.it/5ep.html

Example:

>>> b1 = Borg()
>>> b2 = Borg()
>>> b1.value = 42
>>> b2.value  # Will be 42, state is shared
42

class autils.data_structures.LazyProperty(f_get)[source]

Bases: object

Lazily instantiated property.

Use this decorator when you want to set a property that will only be evaluated the first time it’s accessed. This is useful for expensive computations that should be deferred until actually needed.

Once computed, the value is stored as a regular attribute on the instance, avoiding repeated computation. Inspired by the discussion in the Stack Overflow thread below.

See:: http://stackoverflow.com/questions/15226721/

class autils.data_structures.CallbackRegister(name, log)[source]

Bases: object

Registers pickable functions to be executed later.

This class maintains a registry of functions with their arguments that can be called at a later time, typically for cleanup operations. All registered functions must be pickable (serializable).

register(func, args, kwargs, once=False)[source]

Register function/args to be called on self.run().

Parameters:

func (callable) – Pickable function to be called later
args (tuple) – Pickable positional arguments for the function
kwargs (dict) – Pickable keyword arguments for the function
once (bool) – Add unique (func,args,kwargs) combination only once

unregister(func, args, kwargs)[source]

Unregister (func,args,kwargs) combination.

Parameters:

func (callable) – Pickable function to unregister
args (tuple) – Pickable positional arguments
kwargs (dict) – Pickable keyword arguments

run()[source]

Call all registered functions.

Executes all registered functions with their associated arguments. If any function raises an exception, it is logged and execution continues with the remaining functions. Functions are called in LIFO order (last registered, first executed).

autils.data_structures.time_to_seconds(time)[source]

Convert time in minutes, hours and days to seconds.

Converts a time string with optional unit suffix to seconds. Supported units are ‘s’ (seconds), ‘m’ (minutes), ‘h’ (hours), and ‘d’ (days). If no unit is specified, the value is assumed to be in seconds.

Parameters:: time (str, int, or None) – Time, optionally including the unit (i.e. ‘10d’, ‘5m’, ‘30’)
Returns:: Time converted to seconds
Return type:: int
Raises:: ValueError – If the time format is invalid

Example:

>>> time_to_seconds('10m')
600
>>> time_to_seconds('2h')
7200
>>> time_to_seconds('30')
30
>>> time_to_seconds(None)
0

class autils.data_structures.DataSize(data)[source]

Bases: object

Data Size object with builtin unit-converted attributes.

Represents a data size with automatic unit conversion capabilities. Supports bytes (b), kibibytes (k), mebibytes (m), gibibytes (g), and tebibytes (t). All conversions use binary multipliers (1024-based).

Parameters:: data (str, int, or float) – Data size plus optional unit string. i.e. ‘10m’. No unit string means the data size is in bytes.
Raises:: InvalidDataSize – If the data format is invalid

Example:

>>> size = DataSize('10m')
>>> size.b  # bytes
10485760
>>> size.k  # kibibytes
10240
>>> size.g  # gibibytes
0

MULTIPLIERS = {'b': 1, 'g': 1073741824, 'k': 1024, 'm': 1048576, 't': 1099511627776}

property value

The numeric value of the data size.

Returns:: The original numeric value without unit conversion
Return type:: float

property unit

The unit of the data size.

Returns:: Single character representing the unit (‘b’, ‘k’, ‘m’, ‘g’, ‘t’)
Return type:: str

property b

Data size in bytes.

Returns:: Size converted to bytes
Return type:: float

property k

Data size in kibibytes.

Returns:: Size converted to kibibytes (truncated to integer)
Return type:: int

property m

Data size in mebibytes.

Returns:: Size converted to mebibytes (truncated to integer)
Return type:: int

property g

Data size in gibibytes.

Returns:: Size converted to gibibytes (truncated to integer)
Return type:: int

property t

Data size in tebibytes.

Returns:: Size converted to tebibytes (truncated to integer)
Return type:: int

Network

Ports

Module with network related utility functions.

autils.network.ports.FAMILIES = (AddressFamily.AF_INET, AddressFamily.AF_INET6): Families taken into account in this class

autils.network.ports.PROTOCOLS = (SocketKind.SOCK_STREAM, SocketKind.SOCK_DGRAM): Protocols taken into account in this class

autils.network.ports.is_port_available(port, address, family=AddressFamily.AF_INET, protocol=SocketKind.SOCK_STREAM)[source]

Return True if the given port is available for use.

Parameters:

port (int) – Port value to check.
address (str) – Address to use this port.
family (socket.AddressFamily.AF_*) – Default is socket.AF_INET. Accepted values are: socket.AF_INET or socket.AF_INET6.
protocol (socket.AddressFamily.SOCK_*) – Protocol type. Default is socket.SOCK_STREAM (TCP). Accepted values are: socket.SOCK_STREAM or socket.SOCK_DGRAM.

Returns:

True if the port is available, False otherwise.

Return type:

bool

autils.network.ports.is_port_free(port, address)[source]

This method is deprecated. Please use is_port_available().

Parameters:

port (int) – Port value to check.
address (str) – Address to use this port.

Returns:

True if the port is available, False otherwise.

Return type:

bool

autils.network.ports.find_free_port(start_port=1024, end_port=65535, address='localhost', sequent=False, family=AddressFamily.AF_INET, protocol=SocketKind.SOCK_STREAM)[source]

Return a host free port in the range [start_port, end_port].

Parameters:

start_port (int) – header of candidate port range, defaults to 1024
end_port (int) – ender of candidate port range, defaults to 65535
address (str) – Socket address to bind or connect
sequent (bool) – Find port sequentially, random order if it’s False
family (socket.AddressFamily.AF_*) – Default is socket.AF_INET. Accepted values are: socket.AF_INET or socket.AF_INET6.
protocol (socket.AddressFamily.SOCK_*) – Protocol type. Default is socket.SOCK_STREAM (TCP). Accepted values are: socket.SOCK_STREAM or socket.SOCK_DGRAM.

Returns:

A free port number, or None if no free port is found.

Return type:

int or None if no free port found

autils.network.ports.find_free_ports(start_port, end_port, count, address='localhost', sequent=False, family=AddressFamily.AF_INET, protocol=SocketKind.SOCK_STREAM)[source]

Return a number of host free ports in the range [start_port, end_port].

Parameters:

start_port (int) – header of candidate port range
end_port (int) – ender of candidate port range
count (int) – Initial number of ports known to be free in the range.
address (str) – Socket address to bind or connect
sequent (bool) – Find port sequentially, random order if it’s False
family (socket.AddressFamily.AF_*) – Default is socket.AF_INET. Accepted values are: socket.AF_INET or socket.AF_INET6.
protocol (socket.AddressFamily.SOCK_*) – Protocol type. Default is socket.SOCK_STREAM (TCP). Accepted values are: socket.SOCK_STREAM or socket.SOCK_DGRAM.

Returns:

A list of free port numbers.

Return type:

list[int]

class autils.network.ports.PortTracker[source]

Bases: Borg

Tracks ports used in the host machine.

register_port(port)[source]

Registers a port as being in use.

Parameters:: port (int) – The port number to register.
Returns:: The registered port number if successful.
Return type:: int
Raises:: ValueError – If the port is already in use or cannot be registered.

find_free_port(start_port=None)[source]

Finds and registers a free port.

It starts searching from start_port if provided, otherwise it uses the default self.start_port.

Parameters:: start_port (int or None) – The port number to start searching from.
Returns:: The first free port number found.
Return type:: int

release_port(port)[source]

Releases a previously registered port.

Parameters:: port (int) – The port number to release.

System

CPU

Utilities for querying and managing CPU information on the current machine.

This module provides functions to read CPU details from /proc/cpuinfo and sysfs, including architecture, vendor, version, online/offline status, NUMA topology, and frequency governor settings. It supports x86_64, i386, powerpc, s390, aarch64, and other architectures.

autils.system.cpu.VENDORS_MAP = {'amd': (b'AMD',), 'ibm': (b'POWER\\d', b'IBM/S390', b'Power\\d'), 'intel': (b'GenuineIntel',)}: Map vendor’s name with expected string in /proc/cpuinfo.

exception autils.system.cpu.FamilyException[source]

Bases: Exception

Raised when CPU family cannot be determined for the current architecture.

autils.system.cpu.cpu_has_flags(flags)[source]

Check if a list of flags are available on current CPU info.

Parameters:: flags (list of str) – A list of cpu flags that must exists on the current CPU.
Returns:: True if all the flags were found or False if not
Return type:: bool

autils.system.cpu.get_version()[source]

Get cpu version.

Returns:: cpu version of given machine e.g.:- ‘i5-5300U’ for Intel and ‘POWER9’ for IBM machines in case of unknown/unsupported machines, return an empty string.
Return type:: str

autils.system.cpu.get_vendor()[source]

Get the current cpu vendor name.

Returns:: a key of VENDORS_MAP (e.g. ‘intel’) depending on the current CPU architecture. Return None if it was unable to determine the vendor name.
Return type:: str or None

autils.system.cpu.get_revision()[source]

Get revision from /proc/cpuinfo.

Returns:: Revision entry from /proc/cpuinfo (e.g. ‘0080’ for IBM POWER10), or None if no revision line is found.
Return type:: str or None

autils.system.cpu.get_va_bits()[source]

Get virtual address bit size from /proc/cpuinfo (x86).

Returns:: VA address bit size as string (e.g. ‘48’), or empty string if not found or on non-x86 architectures.
Return type:: str

autils.system.cpu.get_arch()[source]

Detect the CPU architecture of the current machine.

Returns:: Architecture string (e.g. ‘x86_64’, ‘powerpc’, ‘s390’, ‘aarch64’).
Return type:: str

autils.system.cpu.get_family()[source]

Get CPU family or microarchitecture name.

Returns:

Family string (e.g. ‘broadwell’, ‘haswell’, ‘power8’, ‘power9’, ‘z15’) depending on architecture.

Return type:

str

Raises:

FamilyException – When family cannot be determined.
NotImplementedError – On unsupported architectures.

autils.system.cpu.get_model()[source]

Get CPU model number (x86 only).

Returns:: Model integer from /proc/cpuinfo, or None if not found.
Return type:: int or None
Raises:: NotImplementedError – On non-x86 architectures.

autils.system.cpu.get_x86_amd_zen(family=None, model=None)[source]

Get the AMD Zen architecture version for x86 AMD CPUs.

Parameters:

family (int or None) – AMD CPU family (default: from get_family()).
model (int or None) – AMD CPU model (default: from get_model()).

Returns:

Zen generation (1-6), or None if not an AMD Zen CPU.

Return type:

int or None

autils.system.cpu.online_list()[source]

Report a list of indexes of the online CPUs.

Returns:: List of online CPU indices.
Return type:: list of int

autils.system.cpu.total_count()[source]

Return number of total CPUs in the system including offline CPUs.

Returns:: Total CPU count.
Return type:: int

autils.system.cpu.online_count()[source]

Return number of online CPUs in the system.

Returns:: Online CPU count.
Return type:: int

autils.system.cpu.is_hotpluggable(cpu)[source]

Check whether a CPU can be hot-plugged (offlined/onlined).

Parameters:: cpu (int) – CPU index to check.
Returns:: True if the CPU has an ‘online’ sysfs interface.
Return type:: bool

autils.system.cpu.online(cpu)[source]

Bring a CPU online.

Parameters:: cpu (int) – CPU index to bring online.
Returns:: 1 on success, 0 on failure (requires root).
Return type:: int

autils.system.cpu.offline(cpu)[source]

Take a CPU offline.

Parameters:: cpu (int) – CPU index to take offline.
Returns:: 0 on success, 1 on failure (requires root).
Return type:: int

autils.system.cpu.get_idle_state()[source]

Get current CPU idle state values.

Returns:: Dict of cpuidle state values for all CPUs
Return type:: dict

autils.system.cpu.set_idle_state(state_number='all', disable=True, setstate=None)[source]

Set or reset CPU idle states for all CPUs.

Parameters:

state_number (str) – cpuidle state number, default: all all states
disable (bool) – whether to disable/enable given cpu idle state, default is to disable.
setstate (dict) – cpuidle state value, output of get_idle_state()

autils.system.cpu.set_freq_governor(governor='random')[source]

Change the CPU frequency governor for all CPUs.

Parameters:: governor (str) – Governor name (e.g. ‘performance’, ‘powersave’), or ‘random’ to pick one randomly from available governors.
Returns:: True on success, False on failure.
Return type:: bool

autils.system.cpu.get_freq_governor()[source]

Get the current CPU frequency governor.

Returns:: Governor name (e.g. ‘performance’), or empty string on error.
Return type:: str

autils.system.cpu.get_pid_cpus(pid)[source]

Get CPU indices used by a process (from /proc/<pid>/task/*/stat).

Parameters:: pid (int or str) – Process ID.
Returns:: List of CPU index strings the process threads are running on.
Return type:: list of str

autils.system.cpu.get_numa_node_has_cpus()[source]

Get NUMA node numbers that have CPUs assigned.

Returns:: List of NUMA node identifiers that have CPUs.
Return type:: list of str

autils.system.cpu.numa_nodes_with_assigned_cpus()[source]

Get NUMA nodes with their associated CPU indices.

Returns:: Dict mapping NUMA node ID to sorted list of CPU indices.
Return type:: dict

autils.system.cpu.lscpu()[source]

Get CPU topology by executing the ‘lscpu’ command.

Returns:: Dict with keys such as ‘cores_per_chip’, ‘physical_sockets’, ‘physical_chips’, ‘threads_per_core’, ‘sockets’, ‘chips’, ‘physical_cores’ (depending on lscpu output).
Return type:: dict

autils.system.cpu.total_cpus_count(*args, **kwargs)

Wrapper that emits deprecation warning and delegates to newfunc.

Parameters:

args – Positional arguments passed through to newfunc.
kwargs – Keyword arguments passed through to newfunc.

Returns:

Result of newfunc(*args, **kwargs).

Return type:

any

autils.system.cpu.get_cpu_vendor_name(*args, **kwargs)

Wrapper that emits deprecation warning and delegates to newfunc.

Parameters:

args – Positional arguments passed through to newfunc.
kwargs – Keyword arguments passed through to newfunc.

Returns:

Result of newfunc(*args, **kwargs).

Return type:

any

autils.system.cpu.get_cpu_arch(*args, **kwargs)

Wrapper that emits deprecation warning and delegates to newfunc.

Parameters:

args – Positional arguments passed through to newfunc.
kwargs – Keyword arguments passed through to newfunc.

Returns:

Result of newfunc(*args, **kwargs).

Return type:

any

autils.system.cpu.cpu_online_list(*args, **kwargs)

Wrapper that emits deprecation warning and delegates to newfunc.

Parameters:

args – Positional arguments passed through to newfunc.
kwargs – Keyword arguments passed through to newfunc.

Returns:

Result of newfunc(*args, **kwargs).

Return type:

any

autils.system.cpu.online_cpus_count(*args, **kwargs)

Wrapper that emits deprecation warning and delegates to newfunc.

Parameters:

args – Positional arguments passed through to newfunc.
kwargs – Keyword arguments passed through to newfunc.

Returns:

Result of newfunc(*args, **kwargs).

Return type:

any

autils.system.cpu.get_cpuidle_state(*args, **kwargs)

Wrapper that emits deprecation warning and delegates to newfunc.

Parameters:

args – Positional arguments passed through to newfunc.
kwargs – Keyword arguments passed through to newfunc.

Returns:

Result of newfunc(*args, **kwargs).

Return type:

any

autils.system.cpu.set_cpuidle_state(*args, **kwargs)

Wrapper that emits deprecation warning and delegates to newfunc.

Parameters:

args – Positional arguments passed through to newfunc.
kwargs – Keyword arguments passed through to newfunc.

Returns:

Result of newfunc(*args, **kwargs).

Return type:

any

autils.system.cpu.set_cpufreq_governor(*args, **kwargs)

Wrapper that emits deprecation warning and delegates to newfunc.

Parameters:

args – Positional arguments passed through to newfunc.
kwargs – Keyword arguments passed through to newfunc.

Returns:

Result of newfunc(*args, **kwargs).

Return type:

any

autils.system.cpu.get_cpufreq_governor(*args, **kwargs)

Wrapper that emits deprecation warning and delegates to newfunc.

Parameters:

args – Positional arguments passed through to newfunc.
kwargs – Keyword arguments passed through to newfunc.

Returns:

Result of newfunc(*args, **kwargs).

Return type:

any

File

Crypto

Cryptographic hash utilities for file verification.

autils.file.crypto.hash_file(filename, size=None, algorithm='md5')[source]

Calculate the hash value of a file.

Computes a cryptographic hash of the specified file using the given algorithm. Optionally limits hashing to the first N bytes of the file, which is useful for verifying partial downloads or large files.

Parameters:

filename (str) – Path of the file that will have its hash calculated.
size (int or None) – If provided, hash only the first size bytes of the file. If None or 0, the entire file is hashed. If size exceeds the file size, the entire file is hashed.
algorithm (str) – Hash algorithm to use. Supported algorithms include md5, sha1, sha256, sha512, blake2b, and others available in hashlib.

Returns:

Hexadecimal digest string of the computed hash. Returns None if an invalid algorithm is specified.

Return type:

str or None

Raises:

FileNotFoundError – When the specified file does not exist.
PermissionError – When the file cannot be read due to permissions.

Example:

>>> hash_file('/path/to/file')
'abc123...'
>>> hash_file('/path/to/file', algorithm='sha256')
'e3b0c44298fc1c149afbf4c8996fb924...'
>>> hash_file('/path/to/large_file', size=1024)
'abc123...'

Genio

Generic IO related functions for file operations.

exception autils.file.genio.GenIOError[source]

Bases: Exception

Base Exception Class for all IO exceptions.

autils.file.genio.ask(question, auto=False)[source]

Prompt the user with a (y/n) question.

Parameters:

question (str) – Question to be asked.
auto (bool) – Whether to return “y” instead of asking the question.

Returns:

User answer.

Return type:

str

Example:

>>> ask("Do you want to continue?", auto=True)
'y'

autils.file.genio.read_file(filename)[source]

Read the entire contents of a file.

Parameters:

filename (str) – Path to the file.

Returns:

File contents.

Return type:

str

Raises:

FileNotFoundError – When the file does not exist.
PermissionError – When the file cannot be read due to permissions.

Example:

>>> read_file("/etc/hostname")  
'myhost\n'

autils.file.genio.read_one_line(filename)[source]

Read the first line of a file.

The returned line has the trailing newline character stripped.

Parameters:

filename (str) – Path to the file.

Returns:

First line contents with newline stripped.

Return type:

str

Raises:

FileNotFoundError – When the file does not exist.
PermissionError – When the file cannot be read due to permissions.

Example:

>>> read_one_line("/etc/hostname")  
'myhost'

autils.file.genio.read_all_lines(filename)[source]

Return all lines of a given file.

This utility method returns an empty list in any error scenario, that is, it doesn’t attempt to identify error paths and raise appropriate exceptions. It does exactly the opposite to that.

This should be used when it’s fine or desirable to have an empty set of lines if a file is missing or is unreadable.

Parameters:: filename (str) – Path to the file.
Returns:: All lines of the file as a list with newlines stripped.
Return type:: list

Example:

>>> read_all_lines("/etc/hosts")  
['127.0.0.1 localhost', '::1 localhost']
>>> read_all_lines("/nonexistent/file.txt")
[]

autils.file.genio.read_line_with_matching_pattern(filename, pattern)[source]

Return lines from a file that contain a given pattern.

This method returns all lines where the pattern substring is found. The search uses simple substring matching (not regex).

Parameters:

filename (str) – Path to the file to be read.
pattern (str) – Pattern substring to search for in each line.

Returns:

All lines from the file that contain the pattern, with newlines stripped.

Return type:

list

Raises:

FileNotFoundError – When the file does not exist.
PermissionError – When the file cannot be read due to permissions.

Example:

>>> read_line_with_matching_pattern("/etc/passwd", "root")  
['root:x:0:0:root:/root:/bin/bash']

autils.file.genio.write_file(filename, data)[source]

Write data to a file.

This will overwrite any existing content in the file. If the file does not exist, it will be created.

Parameters:

filename (str) – Path to the file.
data (str) – Data to be written to the file.

Raises:

FileNotFoundError – When the parent directory does not exist.
PermissionError – When the file cannot be written due to permissions.

Example:

>>> write_file("/tmp/test.txt", "Hello World")  

autils.file.genio.write_one_line(filename, line)[source]

Write one line of text to a file.

A newline character is automatically appended. Any existing trailing newline in the input line is stripped before adding the newline.

Parameters:

filename (str) – Path to the file.
line (str) – Line to be written.

Raises:

FileNotFoundError – When the parent directory does not exist.
PermissionError – When the file cannot be written due to permissions.

Example:

>>> write_one_line("/tmp/test.txt", "Hello World")  

autils.file.genio.write_file_or_fail(filename, data)[source]

Write to a file and raise GenIOError on write failure.

Unlike write_file(), this function catches OSError exceptions and re-raises them as GenIOError with a descriptive message.

Parameters:

filename (str) – Path to the file.
data (str) – Data to be written to the file.

Raises:

GenIOError – When the write operation fails for any reason.

Example:

>>> write_file_or_fail("/tmp/test.txt", "Hello World")  

autils.file.genio.append_file(filename, data)[source]

Append data to a file.

If the file does not exist, it will be created.

Parameters:

filename (str) – Path to the file.
data (str) – Data to be appended to the file.

Raises:

FileNotFoundError – When the parent directory does not exist.
PermissionError – When the file cannot be written due to permissions.

Example:

>>> append_file("/tmp/log.txt", "New log entry\n")  

autils.file.genio.append_one_line(filename, line)[source]

Append one line of text to a file.

A newline character is automatically appended. Any existing trailing newline in the input line is stripped before adding the newline. If the file does not exist, it will be created.

Parameters:

filename (str) – Path to the file.
line (str) – Line to be appended.

Raises:

FileNotFoundError – When the parent directory does not exist.
PermissionError – When the file cannot be written due to permissions.

Example:

>>> append_one_line("/tmp/log.txt", "Log entry 1")  
>>> append_one_line("/tmp/log.txt", "Log entry 2")  

autils.file.genio.is_pattern_in_file(filename, pattern)[source]

Check if a regex pattern matches anywhere in a file.

The pattern is matched using Python’s re.search with MULTILINE mode, allowing patterns like ^ and $ to match at line boundaries.

Parameters:

filename (str) – Path to the file.
pattern (str) – Regular expression pattern to search for.

Returns:

True if pattern matches anywhere in the file, False otherwise.

Return type:

bool

Raises:

GenIOError – When filename is not a regular file (e.g., directory).

Example:

>>> is_pattern_in_file("/etc/passwd", r"^root:")  
True
>>> is_pattern_in_file("/etc/passwd", r"nonexistent")  
False

autils.file.genio.are_files_equal(filename, other)[source]

Compare two files for equality using cryptographic hashing.

This function computes the hash of both files and compares them, which is efficient for large files. Files are considered equal if they have identical content.

Parameters:

filename (str) – Path to the first file.
other (str) – Path to the second file.

Returns:

True if files have identical content, False otherwise.

Return type:

bool

Example:

>>> are_files_equal("/tmp/file1.txt", "/tmp/file2.txt")  
True

Path

Module to handle file and directory paths. It provides functions to manipulate file and directory paths, check permissions, and inspect files.

autils.file.path.SHEBANG = '#!': The indicator for a script file, usually the first line of the file.

exception autils.file.path.CmdNotFoundError(cmd, paths)[source]

Bases: Exception

Indicates that the command was not found in the system after a search.

Parameters:

cmd – String with the command.
paths – List of paths where we looked after.

autils.file.path.get_path(base_path, user_path)[source]

Translate a user specified path to a real path.

If user_path is relative, append it to base_path. If user_path is absolute, return it as is.

Parameters:

base_path (type) – The base path of relative user specified paths.
base_path – str
user_path (str) – The user specified path.

Returns:

The resolved path.

Return type:

str

autils.file.path.init_dir(*args)[source]

Wrapper around os.path.join that creates dirs based on the final path.

Parameters:: args – List of dir arguments that will be os.path.joined.
Returns:: directory.
Return type:: str

autils.file.path.find_command(cmd, default=None, check_exec=True)[source]

Try to find a command in the PATH, paranoid version.

Parameters:

cmd (str) – Command to be found.
default (str or None) – Command path to use as a fallback if not found in the standard directories.
check_exec (bool) – if a check for permissions that render the command executable by the current user should be performed.

Raises:

avocado.utils.path.CmdNotFoundError – in case the command was not found and no default was given.

Returns:

Returns an absolute path to the command or the default value if the command is not found

Return type:

str

class autils.file.path.PathInspector(path)[source]

Bases: object

Inspects paths to provide information about them.

Parameters:: path (str) – The path to inspect.

get_first_line()[source]

Reads and returns the first line of the file from path.

Returns:: The first line of the file or an empty string if the file does not exist or is empty.
Return type:: str

has_exec_permission()[source]

Checks if the file from path has execute permissions for the user.

Returns:: True if the file has execute permissions, False otherwise.
Return type:: bool

is_empty()[source]

Checks if the file in path is empty.

Returns:: True if the file is empty, False otherwise.
Return type:: bool

is_script(language=None)[source]

Checks if the file in the path is a script, optionally checking for a specific language.

Parameters:: language (str, optional) – The scripting language to check for (e.g., “python”). If None, checks for any shebang.
Returns:: True if the file is a script (and matches the language, if provided), False otherwise.
Return type:: bool

is_python()[source]

Checks if the file in path is a Python script.

Returns:: True if the file is a Python script, False otherwise.
Return type:: bool

autils.file.path.usable_rw_dir(directory, create=True)[source]

Verify whether we can use this dir (read/write).

Checks for appropriate permissions, and creates missing dirs as needed.

Parameters:

directory (str) – Directory to check.
create (bool) – whether to create the directory if it does not exist.

Returns:

True if the directory is usable for rw, False otherwise.

Return type:

bool

autils.file.path.usable_ro_dir(directory)[source]

Verify whether dir exists and we can access its contents.

Check if a usable RO directory is there.

Parameters:: directory (str) – Directory to check.
Returns:: True if the directory is accessible, False otherwise.
Return type:: bool

autils.file.path.check_readable(path)[source]

Verify that the given path exists and is readable.

This should be used where an assertion makes sense, and is useful because it can provide a better message in the exception it raises.

Parameters:: path (str) – the path to test
Raises:: OSError – path does not exist or path could not be read

autils.file.path.get_path_mount_point(path)[source]

Returns the mount point for a given file path.

Parameters:: path (str) – the complete filename path. if a non-absolute path is given, it’s transformed into an absolute path first.
Returns:: the mount point for a given file path
Return type:: str

autils.file.path.get_max_file_name_length(path)[source]

Returns the maximum length of a file name in the underlying file system.

Parameters:: path (str) – the complete filename path. if a non-absolute path is given, it’s transformed into an absolute path first.
Returns:: the maximum length of a file name
Return type:: int

Devel

astring

Operations with strings (conversion and sanitation).

The unusual name aims to avoid causing name clashes with the stdlib module string. Even with the dot notation, people may try to do things like

import string … from avocado.utils import string

And not notice until their code starts failing.

autils.devel.astring.ENCODING = 'UTF-8': On import evaluated value representing the system encoding based on system locales using locale.getpreferredencoding(). Use this value wisely as some files are dumped in different encoding.

autils.devel.astring.FS_UNSAFE_CHARS = '<>:"/\\|?*;': String containing all fs-unfriendly chars (Windows-fat/Linux-ext3)

autils.devel.astring.bitlist_to_string(data)[source]

Transform from bit list to ASCII string.

Converts a list of bits to an ASCII string representation. Only complete bytes (8 bits) are processed; partial bytes are ignored.

Parameters:: data (list[int]) – List of integers representing bits to be transformed
Returns:: ASCII string representation of the bit list
Return type:: str
Raises:: UnicodeDecodeError – If the resulting byte values are not valid ASCII

Note

Only processes complete bytes. If the bit list length is not a multiple of 8, the remaining bits are ignored.

Example

>>> bitlist_to_string([0, 1, 0, 0, 0, 0, 0, 1])  # 'A' = 65
'A'
>>> bitlist_to_string([1, 0, 0, 0])  # Incomplete byte
''

autils.devel.astring.string_to_bitlist(data)[source]

Transform from ASCII string to bit list.

Converts each character in the string to its 8-bit binary representation and returns a flat list of all bits.

Parameters:: data (str) – ASCII string to be transformed to bit list
Returns:: List of integers representing the bits of each character
Return type:: list[int]

Note

Each character produces exactly 8 bits, with the most significant bit first.

Example

>>> string_to_bitlist('A')  # 'A' = 65 = 01000001
[0, 1, 0, 0, 0, 0, 0, 1]
>>> string_to_bitlist('AB')
[0, 1, 0, 0, 0, 0, 0, 1, 0, 1, 0, 0, 0, 0, 1, 0]

autils.devel.astring.shell_escape(command)[source]

Escape special characters from a command so that it can be passed as a double quoted (” “) string in a (ba)sh command.

Escapes backslashes, dollar signs, double quotes, and backticks that have special meaning in bash when inside double quotes.

Parameters:: command (str) – The command string to escape
Returns:: The escaped command string safe for shell execution
Return type:: str

autils.devel.astring.strip_console_codes(output, custom_codes=None)[source]

Remove Linux console escape and control sequences from console output.

Removes ANSI escape sequences and other console control codes to make the output readable and suitable for result checking. Handles common codes used during system boot and terminal color formatting.

Parameters:

output (str) – The console output string containing escape sequences
custom_codes (str or None) – Additional regex patterns for codes not covered by the default patterns. Will be added to the built-in console codes regex.

Returns:

Clean string with all console escape sequences removed

Return type:

str

Raises:

ValueError – If unknown console codes are encountered that don’t match the known patterns

Note

If the output doesn’t contain \x1b (ESC character), the original string is returned unchanged for performance.

Supported Console Codes

ANSI color codes: \x1b[31m, \x1b[0m, etc.
Cursor positioning: \x1b[H, \x1b[2J, etc.
Character set selection: \x1b(B, \x1b(0, etc.
Custom codes via the custom_codes parameter

Example

>>> strip_console_codes('\x1b[31mRed Text\x1b[0m')
'Red Text'
>>> strip_console_codes('Normal text')
'Normal text'

autils.devel.astring.iter_tabular_output(matrix, header=None, strip=False)[source]

Generator for a pretty, aligned string representation of a nxm matrix.

This representation can be used to print any tabular data, such as database results. It works by scanning the lengths of each element in each column, and determining the format string dynamically.

Parameters:

matrix (list) – Matrix representation (list with n rows of m elements).
header (tuple or list or None) – Optional tuple or list with header elements to be displayed.
strip (bool) – Optionally remove trailing whitespace from each row.

Returns:

Generator yielding each formatted row of the tabular output

Return type:

generator of str

autils.devel.astring.tabular_output(matrix, header=None, strip=False)[source]

Pretty, aligned string representation of a matrix.

Creates a single formatted string with column-aligned tabular data, suitable for printing or logging. This is a convenience wrapper around iter_tabular_output() that joins all rows with newlines.

Parameters:

matrix (list[list]) – Matrix representation as list of rows, where each row is a list of column elements. Rows may have different lengths.
header (list or tuple or None) – Optional header row elements to be displayed at the top. If provided, will be formatted with the same column alignment.
strip (bool) – If True, removes trailing whitespace from each output row

Returns:

Complete formatted table as a single string with newline separators

Return type:

str

Example

>>> matrix = [['Alice', '25', 'Engineer'], ['Bob', '30', 'Designer']]
>>> print(tabular_output(matrix, header=['Name', 'Age', 'Role']))
Name  Age Role
Alice 25  Engineer
Bob   30  Designer

GDB

GDB Communication and Debugging Utilities

This module provides comprehensive functionality for interacting with the GNU Debugger (GDB) through multiple interfaces and protocols. It supports both local debugging sessions and remote debugging scenarios.

Key Features:

GDB/MI (Machine Interface) communication for programmatic control
GDB Server management for remote debugging sessions
GDB Remote Protocol client implementation
Command execution with structured result parsing
Breakpoint management and program flow control
Support for both CLI and MI command interfaces

Main Classes:

GDB: Wraps a local GDB subprocess with MI interface communication GDBServer: Manages a gdbserver instance for remote debugging GDBRemote: Implements GDB remote protocol client for direct communication CommandResult: Encapsulates command execution results and metadata

Common Usage Patterns:

Automated debugging workflows in test environments
Remote debugging of embedded or containerized applications
Programmatic analysis of application crashes and core dumps
Integration with continuous integration and testing frameworks

The module handles low-level protocol details, message parsing, and connection management, providing a high-level Python interface for GDB operations.

class autils.devel.gdb.GDB(path=None, *extra_args)[source]

Bases: object

Wraps a GDB subprocess for easier manipulation

REQUIRED_ARGS = ['--interpreter=mi', '--quiet']

DEFAULT_BREAK = 'main'

read_gdb_response(timeout=0.01, max_tries=100)[source]

Read raw responses from GDB

Parameters:

timeout (float) – the amount of time to way between read attempts
max_tries (int) – the maximum number of cycles to try to read until a response is obtained

Returns:

a string containing a raw response from GDB

Return type:

str

Raises:

ValueError – if can’t read GDB response

read_until_break(max_lines=100)[source]

Read lines from GDB until a break condition is reached

Parameters:: max_lines (int) – the maximum number of lines to read
Returns:: a list of messages read
Return type:: list of str

send_gdb_command(command)[source]

Send a raw command to the GNU debugger input

Parameters:: command (str) – the GDB command, hopefully in MI language

cmd(command)[source]

Sends a command and parses all lines until prompt is received

Parameters:: command (str) – the GDB command, hopefully in MI language
Returns:: a CommandResult instance
Return type:: CommandResult
Raises:: gdbmi_parser.GdbMiError – if there are many result responses to a single cmd

cli_cmd(command)[source]

Sends a cli command encoded as an MI command

Parameters:: command (str) – a regular GDB cli command
Returns:: a CommandResult instance
Return type:: CommandResult

cmd_exists(command)[source]

Checks if a given command exists

Parameters:: command (str) – a GDB MI command, including the dash (-) prefix
Returns:: either True or False
Return type:: bool

set_file(path)[source]

Sets the file that will be executed

Parameters:: path (str) – the path of the binary that will be executed
Returns:: a CommandResult instance
Return type:: CommandResult
Raises:: UnexpectedResponseError – if response is unexpected

set_break(location, ignore_error=False)[source]

Sets a new breakpoint on the binary currently being debugged

Parameters:

location (str) – a breakpoint location expression as accepted by GDB
ignore_error (bool) – if set, won’t raise exceptions

Returns:

a CommandResult instance

Return type:

CommandResult

Raises:

UnexpectedResponseError – if response is unexpected

del_break(number)[source]

Deletes a breakpoint by its number

Parameters:: number (int) – the breakpoint number
Returns:: a CommandResult instance
Return type:: CommandResult
Raises:: UnexpectedResponseError – if response is unexpected

run(args=None)[source]

Runs the application inside the debugger

Parameters:: args (builtin.list) – the arguments to be passed to the binary as command line arguments
Returns:: a CommandResult instance
Return type:: CommandResult
Raises:: UnexpectedResponseError – if response is unexpected

connect(port)[source]

Connects to a remote debugger (a gdbserver) at the given TCP port

This uses the “extended-remote” target type only

Parameters:: port (int) – the TCP port number
Returns:: a CommandResult instance
Return type:: CommandResult
Raises:: UnexpectedResponseError – if response is unexpected

disconnect()[source]

Disconnects from a remote debugger

Returns:: a CommandResult instance
Return type:: CommandResult
Raises:: UnexpectedResponseError – if response is unexpected

exit()[source]

Exits the GDB application gracefully

Returns:: the result of subprocess.POpen.wait(), that is, a subprocess.POpen.returncode
Return type:: int or None

class autils.devel.gdb.GDBServer(path=None, port=None, wait_until_running=True, *extra_args)[source]

Bases: object

Wraps a gdbserver instance

REQUIRED_ARGS = ['--multi']: The default arguments used when starting the GDB server process

PORT_RANGE = (20000, 20999): The range from which a port to GDB server will try to be allocated from

INIT_TIMEOUT = 5.0: The time to optionally wait for the server to initialize itself and be ready to accept new connections

exit(force=True)[source]

Quits the gdb_server process

Most correct way of quitting the GDB server is by sending it a command. If no GDB client is connected, then we can try to connect to it and send a quit command. If this is not possible, we send it a signal and wait for it to finish.

Parameters:: force (bool) – if a forced exit (sending SIGTERM) should be attempted

class autils.devel.gdb.GDBRemote(host, port, no_ack_mode=True, extended_mode=True)[source]

Bases: object

A GDBRemote acts like a client that speaks the GDB remote protocol, documented at:

https://sourceware.org/gdb/current/onlinedocs/gdb/Remote-Protocol.html

Caveat: we currently do not support communicating with devices, only with TCP sockets. This limitation is basically due to the lack of use cases that justify an implementation, but not due to any technical shortcoming.

static checksum(input_message)[source]

Calculates a remote message checksum.

More details are available at: https://sourceware.org/gdb/current/onlinedocs/gdb/Overview.html

Parameters:: input_message (bytes) – the message input payload, without the start and end markers
Returns:: two byte checksum
Return type:: bytes

static encode(data)[source]

Encodes a command.

That is, add prefix, suffix and checksum.

More details are available at: https://sourceware.org/gdb/current/onlinedocs/gdb/Overview.html

Parameters:: data (bytes) – the command data payload
Returns:: the encoded command, ready to be sent to a remote GDB
Return type:: bytes

static decode(data)[source]

Decodes a packet and returns its payload.

More details are available at: https://sourceware.org/gdb/current/onlinedocs/gdb/Overview.html

Parameters:: data (bytes) – the command data payload
Returns:: the encoded command, ready to be sent to a remote GDB
Return type:: bytes
Raises:: InvalidPacketError – if the packet is not well constructed, like in checksum mismatches

cmd(command_data, expected_response=None)[source]

Sends a command data to a remote gdb server

Limitations: the current version does not deal with retransmissions.

Parameters:

command_data (str) – the remote command to send the the remote stub
expected_response (str) – the (optional) response that is expected as a response for the command sent

Returns:

raw data read from from the remote server

Return type:

str

Raises:

NotConnectedError – if the socket is not initialized
RetransmissionRequestedError – if there was a failure while reading the result of the command
UnexpectedResponseError – if response is unexpected

set_extended_mode()[source]

Enable extended mode. In extended mode, the remote server is made persistent. The ‘R’ packet is used to restart the program being debugged. Original documentation at:

https://sourceware.org/gdb/current/onlinedocs/gdb/Packets.html#extended-mode

start_no_ack_mode()[source]

Request that the remote stub disable the normal +/- protocol acknowledgments. Original documentation at:

https://sourceware.org/gdb/current/onlinedocs/gdb/General-Query-Packets.html#QStartNoAckMode

connect()[source]: Connects to the remote target and initializes the chosen modes

Output

Utility functions for user friendly display of information.

autils.devel.output.display_data_size(size)[source]

Display data size in human readable units (SI).

Parameters:: size (int) – Data size, in Bytes.
Returns:: Human readable string with data size, using SI prefixes.
Return type:: str

class autils.devel.output.ProgressBar(minimum=0, maximum=100, width=75, title='')[source]

Bases: object

Displays interactively the progress of a given task.

Inspired/adapted from https://gist.github.com/t0xicCode/3306295

append_amount(amount)[source]

Increments the current amount value by the specified amount.

Parameters:: amount (int or float) – The value to add to the current amount

update_percentage(percentage)[source]

Updates the progress bar to the specified percentage value.

Parameters:: percentage (int or float) – The percentage value to set (0-100)

update_amount(amount)[source]

Performs sanity checks and updates the current amount value.

The amount is clamped between the minimum and maximum values set during initialization. After updating the amount, the progress bar is refreshed and redrawn.

Parameters:: amount (int or float) – The new amount value to set

draw()[source]

Prints the updated progress bar text to the screen.

Only prints when the progress bar has changed from the previous state to avoid unnecessary screen updates. Uses carriage return to overwrite the previous progress bar on the same line.

Process

Functions dedicated to find and run external commands.

exception autils.devel.process.CmdError(command=None, result=None, additional_text=None)[source]

Bases: Exception

Exception raised when a command fails execution.

exception autils.devel.process.CmdInputError[source]

Bases: Exception

Raised when the command given is invalid, such as an empty command.

autils.devel.process.can_sudo(cmd=None)[source]

Check whether sudo is available or if running as root.

This function checks if the current process has the ability to run commands with elevated privileges. It first checks if the process is running as root (UID 0), then checks if sudo is installed and functional.

Parameters:: cmd (str or None) – Optional command to test sudo capabilities with. If provided, tests whether this specific command can be run with sudo. If not provided, tests basic sudo functionality.
Returns:: True if sudo is available or running as root, False otherwise.
Return type:: bool

Example:

>>> can_sudo()
True
>>> can_sudo("ls /root")
True

autils.devel.process.get_capabilities(pid=None)[source]

Gets a list of all capabilities for a process.

In case the getpcaps command is not available, and empty list will be returned.

It supports getpcaps’ two different formats, the current and the so called legacy/ugly.

Parameters:: pid (int) – the process ID (PID), if one is not given, the current PID is used (given by os.getpid())
Returns:: all capabilities
Return type:: list

autils.devel.process.has_capability(capability, pid=None)[source]

Check if a process has a given Linux capability.

This is a simple wrapper around getpcaps, part of the libcap package. In case the getpcaps command is not available, the capability will be considered not to be available.

Parameters:

capability (str) – The name of the capability (e.g., “cap_sys_admin”). Refer to capabilities(7) man page for more information. Note: capability names are UPPERCASE in capabilities(7) (e.g., CAP_SYS_ADMIN) but must be lowercase in Python (e.g., “cap_sys_admin”).
pid (int or None) – The process ID to check. If None, checks the current process.

Returns:

True if the capability is available, False otherwise.

Return type:

bool

Example:

>>> has_capability("cap_chown")
True
>>> has_capability("cap_sys_admin", pid=1234)
False

autils.devel.process.pid_exists(pid)[source]

Check if a process with the given PID exists.

This function uses os.kill with signal 0 to check if a process exists without actually sending a signal to it.

Parameters:: pid (int) – The process ID number to check.
Returns:: True if the process exists, False otherwise.
Return type:: bool

Example:

>>> pid_exists(1)
True
>>> pid_exists(999999)
False

autils.devel.process.safe_kill(pid, signal)[source]

Attempt to send a signal to a process that may or may not exist.

This function safely sends a signal to a process, handling cases where the process might not exist or require elevated privileges.

Parameters:

pid (int) – The process ID to send the signal to.
signal (int) – The signal number to send (e.g., signal.SIGTERM).

Returns:

True if signal was sent successfully, False otherwise.

Return type:

bool

Example:

>>> safe_kill(1234, signal.SIGTERM)
True

autils.devel.process.get_parent_pid(pid)[source]

Get the parent process ID for a given process.

This function reads the /proc filesystem to determine the parent PID.

Note

This is currently Linux specific.

Parameters:: pid (int) – The PID of the child process.
Returns:: The parent process ID.
Return type:: int
Raises:: IOError – If the /proc entry cannot be read.

Example:

>>> get_parent_pid(1234)
1

autils.devel.process.get_children_pids(parent_pid, recursive=False)[source]

Get the list of child process IDs for a given parent process.

This function scans the /proc filesystem to find all child processes of the specified parent PID.

Note

This is currently Linux specific.

Parameters:

parent_pid (int) – The PID of the parent process.
recursive (bool) – If True, also returns grandchildren and all descendants. If False, only returns direct children.

Returns:

List of child process IDs.

Return type:

list of int

Example:

>>> get_children_pids(1)
[234, 456, 789]
>>> get_children_pids(1, recursive=True)
[234, 456, 789, 1011, 1213]

autils.devel.process.kill_process_tree(pid, sig=None, send_sigcont=True, timeout=0)[source]

Signal a process and all of its children.

If the process does not exist – return.

Parameters:

pid (int) – The pid of the process to signal.
sig (int or None) – The signal to send to the processes, defaults to signal.SIGKILL
send_sigcont (bool) – Send SIGCONT to allow killing stopped processes.
timeout (int or float) – How long to wait for the pid(s) to die (negative=infinity, 0=don’t wait, positive=number_of_seconds).

Returns:

List of all PIDs we sent signal to.

Return type:

list

Raises:

RuntimeError – If timeout is reached waiting for processes to die.

autils.devel.process.kill_process_by_pattern(pattern)[source]

Send SIGTERM signal to processes matching a pattern.

This function uses the pkill command to terminate processes whose command line matches the given pattern.

Parameters:: pattern (str) – Pattern to match against process command lines. This is matched using pkill’s -f flag, which matches against the full command line.

Example:

>>> kill_process_by_pattern("firefox")
>>> kill_process_by_pattern("python.*test_script")

autils.devel.process.process_in_ptree_is_defunct(ppid)[source]

Verify if any processes deriving from PPID are in the defunct state.

Attempt to verify if parent process and any children from PPID is defunct (zombie) or not.

This relies on the GNU version of “ps” and is not guaranteed to work in MacOS.

Parameters:: ppid (int) – The parent PID of the process to verify.
Returns:: True if any process in the tree is defunct, False otherwise.
Return type:: bool

autils.devel.process.binary_from_shell_cmd(cmd)[source]

Extract the first binary path from a shell-like command string.

This function parses a shell command and returns the first binary/executable found, skipping environment variable assignments.

Note

This is a naive implementation that handles common patterns like environment variable assignments before the binary name.

Parameters:: cmd (str) – A shell-like command string to parse.
Returns:: The first binary/executable found in the command.
Return type:: str
Raises:: ValueError – If no binary can be extracted from the command.

Example:

>>> binary_from_shell_cmd("binary")
'binary'
>>> binary_from_shell_cmd("VAR=VAL binary -args")
'binary'
>>> binary_from_shell_cmd("FOO=bar ./script.py")
'./script.py'

autils.devel.process.cmd_split(s, comments=False, posix=True): This is kept for compatibility purposes, but is now deprecated and will be removed in later versions. Please use shlex.split() instead.

class autils.devel.process.CmdResult(command='', stdout=b'', stderr=b'', exit_status=None, duration=0, pid=None, encoding=None)[source]

Bases: object

Command execution result.

Parameters:

command (str) – the command line itself
exit_status (int) – exit code of the process
stdout (bytes) – content of the process stdout
stderr (bytes) – content of the process stderr
duration (float) – elapsed wall clock time running the process
pid (int) – ID of the process
encoding (str) – the encoding to use for the text version of stdout and stderr, by default autils.devel.astring.ENCODING

stdout: The raw stdout (bytes)

stderr: The raw stderr (bytes)

property stdout_text

Return stdout decoded as text.

Returns:: The stdout content as a string.
Return type:: str
Raises:: TypeError – If stdout cannot be decoded.

property stderr_text

Return stderr decoded as text.

Returns:: The stderr content as a string.
Return type:: str
Raises:: TypeError – If stderr cannot be decoded.

class autils.devel.process.FDDrainer(fd, result, name=None, logger=None, logger_prefix='%s', stream_logger=None, ignore_bg_processes=False, verbose=False)[source]

Bases: object

Reads data from a file descriptor in a thread, storing locally.

start()[source]: Start the drainer thread to read from the file descriptor.

flush()[source]: Wait for drainer thread to complete and flush stream handlers.

class autils.devel.process.SubProcess(cmd, verbose=True, shell=False, env=None, sudo=False, ignore_bg_processes=False, encoding=None, logger=None)[source]

Bases: object

Run a subprocess in the background, collecting stdout/stderr streams.

start()[source]

Start running the subprocess.

This method is particularly useful for background processes, since you can start the subprocess and not block your test flow.

Returns:: Subprocess PID.
Return type:: int

get_stdout()[source]

Get the full stdout of the subprocess so far.

Returns:: Standard output of the process.
Return type:: bytes

get_stderr()[source]

Get the full stderr of the subprocess so far.

Returns:: Standard error of the process.
Return type:: bytes

terminate()[source]

Send a signal.SIGTERM to the process.

Please consider using stop() instead if you want to do all that’s possible to finalize the process and wait for it to finish.

kill()[source]

Send a signal.SIGKILL to the process.

Please consider using stop() instead if you want to do all that’s possible to finalize the process and wait for it to finish.

send_signal(sig)[source]

Send the specified signal to the process.

Parameters:: sig (int) – Signal to send.

poll()[source]

Call the subprocess poll() method, fill results if rc is not None.

Returns:: Return code if process has finished, None otherwise.
Return type:: int or None

wait(timeout=None, sig=Signals.SIGTERM)[source]

Wait for subprocess to complete, fill results when done.

Parameters:

timeout (float or None) – Time (seconds) we’ll wait until the process is finished. If it’s not, we’ll try to terminate it and it’s children using sig and get a status. When the process refuses to die within 1s we use SIGKILL and report the status (be it exit_code or zombie).
sig (int) – Signal to send to the process in case it did not end after the specified timeout.

Returns:

Exit status of the process.

Return type:

int

Raises:

AssertionError – If the process becomes a zombie.

stop(timeout=None)[source]

Stop background subprocess.

Call this method to terminate the background subprocess and wait for its results.

Parameters:: timeout (float or None) – Time (seconds) we’ll wait until the process is finished. If it’s not, we’ll try to terminate it and its children using sig and get a status. When the process refuses to die within 1s we use SIGKILL and report the status (be it exit_code or zombie).
Returns:: Exit status of the process.
Return type:: int

get_pid()[source]

Report PID of this process.

Returns:: Process ID.
Return type:: int

get_user_id()[source]

Report user id of this process.

Returns:: User ID of the process owner.
Return type:: int or None

is_sudo_enabled()[source]

Return whether the subprocess is running with sudo enabled.

Returns:: True if running as root (UID 0), False otherwise.
Return type:: bool

run(timeout=None, sig=Signals.SIGTERM)[source]

Start a process and wait for it to end, returning the result attr.

If the process was already started using .start(), this will simply wait for it to end.

Parameters:

timeout (float or None) – Time (seconds) we’ll wait until the process is finished. If it’s not, we’ll try to terminate it and its children using sig and get a status. When the process refuses to die within 1s we use SIGKILL and report the status (be it exit_code or zombie).
sig (int) – Signal to send to the process in case it did not end after the specified timeout.

Returns:

The command result object.

Return type:

CmdResult

autils.devel.process.run(cmd, timeout=None, verbose=True, ignore_status=False, shell=False, env=None, sudo=False, ignore_bg_processes=False, encoding=None, logger=None)[source]

Run a subprocess, returning a CmdResult object.

Parameters:

cmd (str) – Command line to run.
timeout (float or None) – Time limit in seconds before attempting to kill the running process. This function will take a few seconds longer than ‘timeout’ to complete if it has to kill the process.
verbose (bool) – Whether to log the command run and stdout/stderr.
ignore_status (bool) – Whether to raise an exception when command returns =! 0 (False), or not (True).
shell (bool) – Whether to run the command on a subshell.
env (dict) – Use extra environment variables.
sudo (bool) – Whether the command requires admin privileges to run, so that sudo will be prepended to the command. The assumption here is that the user running the command has a sudo configuration such that a password won’t be prompted. If that’s not the case, the command will straight out fail.
ignore_bg_processes (bool) – Whether to ignore background processes.
encoding (str) – the encoding to use for the text representation of the command result stdout and stderr, by default autils.devel.astring.ENCODING
logger (logging.Logger) – User’s custom logger, which will be logging the subprocess outputs. When this parameter is not set, the autils.devel.process logger will be used.

Returns:

A CmdResult object.

Return type:

CmdResult

Raises:

CmdInputError – If the command is empty.
CmdError – If ignore_status=False and command fails.

autils.devel.process.system(cmd, timeout=None, verbose=True, ignore_status=False, shell=False, env=None, sudo=False, ignore_bg_processes=False, encoding=None, logger=None)[source]

Run a subprocess, returning its exit code.

Parameters:

cmd (str) – Command line to run.
timeout (float or None) – Time limit in seconds before attempting to kill the running process. This function will take a few seconds longer than ‘timeout’ to complete if it has to kill the process.
verbose (bool) – Whether to log the command run and stdout/stderr.
ignore_status (bool) – Whether to raise an exception when command returns =! 0 (False), or not (True).
shell (bool) – Whether to run the command on a subshell.
env (dict) – Use extra environment variables.
sudo (bool) – Whether the command requires admin privileges to run, so that sudo will be prepended to the command. The assumption here is that the user running the command has a sudo configuration such that a password won’t be prompted. If that’s not the case, the command will straight out fail.
ignore_bg_processes (bool) – Whether to ignore background processes.
encoding (str) – the encoding to use for the text representation of the command result stdout and stderr, by default autils.devel.astring.ENCODING
logger (logging.Logger) – User’s custom logger, which will be logging the subprocess outputs. When this parameter is not set, the autils.devel.process logger will be used.

Returns:

Exit code.

Return type:

int

Raises:

CmdError – If ignore_status=False and command fails.

autils.devel.process.system_output(cmd, timeout=None, verbose=True, ignore_status=False, shell=False, env=None, sudo=False, ignore_bg_processes=False, strip_trail_nl=True, encoding=None, logger=None)[source]

Run a subprocess, returning its output.

Parameters:

cmd (str) – Command line to run.
timeout (float or None) – Time limit in seconds before attempting to kill the running process. This function will take a few seconds longer than ‘timeout’ to complete if it has to kill the process.
verbose (bool) – Whether to log the command run and stdout/stderr.
ignore_status (bool) – Whether to raise an exception when command returns =! 0 (False), or not (True).
shell (bool) – Whether to run the command on a subshell.
env (dict) – Use extra environment variables.
sudo (bool) – Whether the command requires admin privileges to run, so that sudo will be prepended to the command. The assumption here is that the user running the command has a sudo configuration such that a password won’t be prompted. If that’s not the case, the command will straight out fail.
ignore_bg_processes (bool) – Whether to ignore background processes.
strip_trail_nl (bool) – Whether to strip the trailing newline.
encoding (str) – the encoding to use for the text representation of the command result stdout and stderr, by default autils.devel.astring.ENCODING
logger (logging.Logger) – User’s custom logger, which will be logging the subprocess outputs. When this parameter is not set, the autils.devel.process logger will be used.

Returns:

Command output.

Return type:

bytes

Raises:

CmdError – If ignore_status=False and command fails.

autils.devel.process.getoutput(cmd, timeout=None, verbose=False, ignore_status=True, shell=True, env=None, sudo=False, ignore_bg_processes=False, logger=None)[source]

Return output (stdout or stderr) of executing cmd in a shell.

Because commands module is removed in Python3 and it redirect stderr to stdout, we port commands.getoutput to make code compatible.

Parameters:

cmd (str) – Command line to run.
timeout (float or None) – Time limit in seconds before attempting to kill the running process. This function will take a few seconds longer than ‘timeout’ to complete if it has to kill the process.
verbose (bool) – Whether to log the command run and stdout/stderr.
ignore_status (bool) – Whether to raise an exception when command returns =! 0 (False), or not (True).
shell (bool) – Whether to run the command on a subshell.
env (dict) – Use extra environment variables.
sudo (bool) – Whether the command requires admin privileges to run, so that sudo will be prepended to the command. The assumption here is that the user running the command has a sudo configuration such that a password won’t be prompted. If that’s not the case, the command will straight out fail.
ignore_bg_processes (bool) – Whether to ignore background processes.
logger (logging.Logger) – User’s custom logger, which will be logging the subprocess outputs. When this parameter is not set, the autils.devel.process logger will be used.

Returns:

Command output (stdout or stderr).

Return type:

str

autils.devel.process.getstatusoutput(cmd, timeout=None, verbose=False, ignore_status=True, shell=True, env=None, sudo=False, ignore_bg_processes=False, logger=None)[source]

Return (status, output) of executing cmd in a shell.

Because commands module is removed in Python3 and it redirect stderr to stdout, we port commands.getstatusoutput to make code compatible.

Parameters:

cmd (str) – Command line to run.
timeout (float or None) – Time limit in seconds before attempting to kill the running process. This function will take a few seconds longer than ‘timeout’ to complete if it has to kill the process.
verbose (bool) – Whether to log the command run and stdout/stderr.
ignore_status (bool) – Whether to raise an exception when command returns =! 0 (False), or not (True).
shell (bool) – Whether to run the command on a subshell.
env (dict) – Use extra environment variables.
sudo (bool) – Whether the command requires admin privileges to run, so that sudo will be prepended to the command. The assumption here is that the user running the command has a sudo configuration such that a password won’t be prompted. If that’s not the case, the command will straight out fail.
ignore_bg_processes (bool) – Whether to ignore background processes.
logger (logging.Logger) – User’s custom logger, which will be logging the subprocess outputs. When this parameter is not set, the autils.devel.process logger will be used.

Returns:

Exit status and command output (stdout and stderr).

Return type:

tuple

autils.devel.process.get_owner_id(pid)[source]

Get the user ID of the process owner.

This function reads the /proc filesystem to determine the user ID that owns the specified process.

Note

This is currently Linux specific.

Parameters:: pid (int) – The process ID to query.
Returns:: The user ID of the process owner, or None if not found.
Return type:: int or None

Example:

>>> get_owner_id(1)
0
>>> get_owner_id(999999)
None

autils.devel.process.get_command_output_matching(command, pattern)[source]

Run a command and return lines matching a pattern.

This function executes a command and searches its output for lines containing the specified pattern, returning all matching lines.

Parameters:

command (str) – The command to execute.
pattern (str) – Pattern to search for in the output. Matching is done on a line-by-line basis using substring matching.

Returns:

List of lines from the command output that contain the pattern.

Return type:

list of str

Example:

>>> get_command_output_matching("ls -la", "txt")
['file1.txt', 'file2.txt']

Script

Module to handle script file creation and management.

This module provides utilities for creating executable script files, both permanent and temporary, with proper file permissions and automatic cleanup capabilities.

autils.devel.script.DEFAULT_MODE = 509: What is commonly known as “0775” or “u=rwx,g=rwx,o=rx”

autils.devel.script.READ_ONLY_MODE = 292: What is commonly known as “0444” or “u=r,g=r,o=r”

class autils.devel.script.Script(path, content, mode=509, open_mode='w')[source]

Bases: object

Class that represents a script file.

This class provides methods to create, save, and remove script files with configurable permissions. It supports context manager protocol for automatic cleanup.

save()[source]

Store script content to the file system.

Creates parent directories if they do not exist and sets the specified file permissions.

Returns:: True if script has been stored successfully.
Return type:: bool

remove()[source]

Remove script file from the file system.

Returns:: True if the script file was removed, False if it did not exist.
Return type:: bool

class autils.devel.script.TemporaryScript(name, content, prefix='avocado_script', mode=509, open_mode='w')[source]

Bases: Script

Class that represents a temporary script in an auto-managed directory.

The script is created in a temporary directory that is automatically cleaned up when the instance is garbage collected or when used as a context manager.

remove()[source]

Remove script file from the file system.

Returns:: True if the script file was removed, False if it did not exist.
Return type:: bool

autils.devel.script.make_script(path, content, mode=509)[source]

Creates and saves a new script file to the file system.

This is a convenience function that creates a Script instance, saves it, and returns the path.

Parameters:

path (str) – The path where the script file will be created.
content (str or bytes) – The content to write to the script file.
mode (int) – File permissions mode. Defaults to 0775 (rwxrwxr-x).

Returns:

The path to the created script file.

Return type:

str

autils.devel.script.make_temp_script(name, content, prefix='avocado_script', mode=509)[source]

Creates and saves a new temporary script in a temporary directory.

This is a convenience function that creates a TemporaryScript instance and saves it. Note that the script’s temporary directory will be removed when the TemporaryScript object is garbage collected.

Parameters:

name (str) – The script filename (not the full path).
content (str or bytes) – The content to write to the script file.
prefix (str) – Prefix for the temporary directory name.
mode (int) – File permissions mode. Defaults to 0775 (rwxrwxr-x).

Returns:

The full path to the created script file.

Return type:

str

Wait

Utilities for waiting for conditions to be met.

This module provides utilities for polling functions until they return a truthy value or a timeout expires, useful for testing and development scenarios where you need to wait for system state changes.

autils.devel.wait.wait_for(func, timeout, first=0.0, step=1.0, text=None, args=None, kwargs=None)[source]

Wait until a function returns a truthy value or timeout expires.

This function repeatedly calls a given function with optional arguments until it returns a truthy value (anything that evaluates to True in a boolean context) or until the specified timeout expires. It provides configurable delays before the first attempt and between subsequent attempts, making it useful for polling operations in testing and development scenarios.

The function uses time.monotonic() for reliable timeout calculation that is not affected by system clock adjustments. Note that the step sleep duration is not interrupted when timeout expires, so actual elapsed time may exceed the specified timeout by up to one step duration.

Parameters:

func (callable) – Callable to be executed repeatedly until it returns a truthy value. Can be any callable object (function, lambda, method, callable class instance).
timeout (float or int) – Maximum time in seconds to wait for func to return a truthy value. Must be a non-negative number. If timeout expires before func returns truthy, None is returned.
first (float or int) – Time in seconds to sleep before the first attempt to call func. Useful when you know the condition won’t be met immediately. Defaults to 0.0 (no initial delay).
step (float or int) – Time in seconds to sleep between successive calls to func. The actual sleep happens after each failed attempt. Defaults to 1.0 second. Note that this sleep is not interrupted when timeout expires.
text (str or None) – Optional debug message to log before each attempt. When provided, logs at DEBUG level with elapsed time since start. If None, no logging occurs. Useful for debugging wait operations.
args (list, tuple, or None) – Optional list or tuple of positional arguments to pass to func on each call. If None, defaults to empty list.
kwargs (dict or None) – Optional dictionary of keyword arguments to pass to func on each call. If None, defaults to empty dict.

Returns:

The truthy return value from func if it succeeds within timeout, or None if timeout expires without func returning a truthy value. The actual return value from func is preserved (e.g., strings, numbers, lists, objects).

Return type:

Any (return type of func) or None

Raises:

Any exception raised by func will be propagated to the caller. No exception handling is performed on func calls.

Example:

>>> import os
>>> # Wait for a file to exist
>>> wait_for(lambda: os.path.exists("/tmp/myfile"), timeout=30, step=1)
True
>>> # Wait for a counter to reach threshold
>>> counter = [0]
>>> def check(): counter[0] += 1; return counter[0] >= 5
>>> wait_for(check, timeout=10, step=0.5)
True
>>> # Wait with custom function and arguments
>>> def check_value(expected, current):
...     return current >= expected
>>> wait_for(check_value, timeout=5, step=0.1, args=[10, 15])
True
>>> # Wait with debug logging
>>> wait_for(lambda: False, timeout=2, step=0.5, text="Waiting for condition")
None