tmt.utils package

Submodules

tmt.utils.git module

Test Metadata Utilities

class tmt.utils.git.CentOSDistGit

Bases: DistGitHandler

CentOS Handler

lookaside_server: str = 'https://sources.stream.centos.org/sources'

re_source: Pattern[str] = re.compile('^(\\w+) \$([^)]+)\$ = ([0-9a-fA-F]+)$')

remote_substring: Pattern[str] = re.compile('redhat/centos')

usage_name: str = 'centos'

class tmt.utils.git.DistGitHandler

Bases: object

Common functionality for DistGit handlers

its_me(remotes: list[str]) → bool: True if self can work with remotes

lookaside_server: str

re_source: Pattern[str]

re_supported_extensions: Pattern[str] = re.compile('\\.((tar\\.(gz|Z|bz2|lz|lzma|lzo|xz|zst))|tgz|taz|taZ|tz2|tbz2|tbz|tlz|tzst)$')

remote_substring: Pattern[str]

sources_file_name = 'sources'

uri = '/rpms/{name}/{filename}/{hashtype}/{hash}/{filename}'

url_and_name(cwd: Path | None = None) → list[tuple[str, str]]

Return list of urls and basenames of the used source

The ‘cwd’ parameter has to be a DistGit directory.

usage_name: str

class tmt.utils.git.FedoraDistGit

Bases: DistGitHandler

Fedora Handler

lookaside_server: str = 'https://src.fedoraproject.org/repo/pkgs'

re_source: Pattern[str] = re.compile('^(\\w+) \$([^)]+)\$ = ([0-9a-fA-F]+)$')

remote_substring: Pattern[str] = re.compile('fedoraproject\\.org')

usage_name: str = 'fedora'

class tmt.utils.git.GitInfo(git_root: Path, ref: str, remote: str, default_branch: str | None, url: str | None)

Bases: object

Data container for commonly queried git data.

default_branch: str | None: Default branch of the remote.

classmethod from_fmf_root(*, fmf_root: Path, logger: Logger) → GitInfo | None

Get the current git info of an fmf tree.

Parameters:

fmf_root – Root path of the fmf tree
logger – Current tmt logger

Returns:

Git info container or None if git metadata could not be resolved

git_root: Path: Path to the git root.

ref: str: Most human-readable git ref.

remote: str: Git remote linked to the current git ref.

url: str | None: Public url of the remote.

class tmt.utils.git.RedHatGitlab

Bases: DistGitHandler

Red Hat on Gitlab

lookaside_server: str = 'http://pkgs.devel.redhat.com/repo'

re_source: Pattern[str] = re.compile('^(\\w+) \$([^)]+)\$ = ([0-9a-fA-F]+)$')

remote_substring: Pattern[str] = re.compile('redhat/rhel/')

usage_name: str = 'redhatgitlab'

tmt.utils.git.check_git_url(url: str, logger: Logger) → str: Check that a remote git url is accessible

tmt.utils.git.clonable_git_url(url: str) → str: Modify the git repo url so it can be cloned

tmt.utils.git.default_branch(*, repository: Path, remote: str = 'origin', logger: Logger) → str | None: Detect default branch from given local git repository

tmt.utils.git.distgit_download(*, distgit_dir: Path, target_dir: Path, handler_name: str | None = None, caller: Common | None = None, logger: Logger) → None

Download sources to the target_dir

distgit_dir is path to the DistGit repository

tmt.utils.git.get_distgit_handler(remotes: list[str] | None = None, usage_name: str | None = None) → DistGitHandler

Return the right DistGitHandler

Pick the DistGitHandler class which understands specified remotes or by usage_name.

tmt.utils.git.get_distgit_handler_names() → list[str]: All known distgit handlers

tmt.utils.git.git_add(*, path: Path, logger: Logger) → None

Add path to the git index.

Parameters:

path – path to add to the git index.
logger – used for logging.

tmt.utils.git.git_clone(*, url: str, destination: Path, shallow: bool = False, can_change: bool = True, env: Environment | None = None, attempts: int | None = None, interval: int | None = None, timeout: int | None = None, logger: Logger) → CommandOutput

Clone git repository from provided url to the destination directory

Parameters:

url – Source URL of the git repository.
destination – Full path to the destination directory.
shallow – For shallow=True first try to clone repository using --depth=1 option. If not successful clone repo with the whole history.
can_change – URL can be modified with hardcoded rules. Use can_change=False to disable rewrite rules.
env – Environment provided to the git clone process.
attempts – Number of tries to call the function.
interval – Amount of seconds to wait before a new try.
timeout – Overall maximum time in seconds to clone the repo.
logger – A Logger instance to be used for logging.

Returns:

Command output, bundled in a CommandOutput tuple.

tmt.utils.git.git_hash(*, directory: Path, logger: Logger) → str | None

Return short hash of current HEAD in the git repo in directory.

Parameters:

directory – path to a local git repository.
logger – used for logging.

Returns:

short hash as string

tmt.utils.git.git_ignore(*, root: Path, logger: Logger) → list[Path]

Collect effective paths ignored by git.

Parameters:

root – path to the root of git repository.
logger – used for logging.

Returns:

list of actual paths that would be ignored by git based on its .gitignore files. If a whole directory is to be ignored, it is listed as a directory path, not listing its content.

tmt.utils.git.git_root(*, fmf_root: Path, logger: Logger) → Path | None

Find a path to the root of git repository containing an fmf root.

Parameters:

fmf_root – path to an fmf root that is supposedly in a git repository.
logger – used for logging.

Returns:

path to the git repository root, if fmf root lies in one, or None.

tmt.utils.git.inject_auth_git_url(url: str) → str

Inject username or token to the git url

Parameters:: url – original git repo url
Returns:: URL with injected authentication based on pattern from the environment or unmodified URL

tmt.utils.git.public_git_url(url: str) → str

Convert a git url into a public format.

Parameters:: url – an URL to convert.
Returns:: URL that is publicly accessible without authentication, or the original URL if no applicable conversion was found.

tmt.utils.git.rewrite_git_url(url: str, patterns: list[tuple[str, str]]) → str

Rewrite git url based on supplied patterns

Parameters:

url – an URL to modify
patterns – List of patterns to try in order

Returns:

Modified url or the original one if no pattern was be applied.

tmt.utils.git.validate_git_status(test: Test) → tuple[bool, str]

Validate that test has current metadata on fmf_id

Return a tuple (boolean, message) as the result of validation.

Checks that sources: - all local changes are committed - up to date on remote repository - .fmf/version marking fmf root is committed as well

When all checks pass returns (True, ‘’).

tmt.utils.git.web_git_url(url: str, ref: str, path: Path | None = None) → str

Convert a public git url into a clickable web url format

Compose a clickable link from git url, ref and path to file for the most common git servers.

tmt.utils.jira module

class tmt.utils.jira.JiraInstance(issue_tracker: IssueTracker, logger: Logger)

Bases: object

A Jira instance configured with url and token

Initialize Jira instance from the issue tracker config

add_link_to_issue(issue_url: str, tmt_objects: Sequence[Test | Plan | Story]) → None: Link one or more tmt objects to the given Jira issue

classmethod from_issue_url(issue_url: str, logger: Logger) → JiraInstance | None: Search configured issues trackers for matching Jira instance

tmt.utils.jira.link(*, tmt_objects: Sequence[Test | Plan | Story], links: Links, separate: bool = False, logger: Logger) → None

Link provided tmt object(s) with related Jira issue(s)

The link is added to the following two locations:

test, plan or story metadata on disk (always)

tmt web link added to the Jira issue (if configured)

Parameters:

tmt_objects – list of tmt tests, plan or stories to be linked
links – target jira issues to be linked
separate – by default a single link is created for all provided tmt objects (e.g. test + plan covering an issue), if True, separate links will be created for each tmt object
logger – a logger instance for logging

tmt.utils.jira.prepare_url_params(tmt_object: Core) → dict[str, str]

Prepare url parameters prefixed with tmt object type

This is the format in which the tmt web API accepts the specification of the objects to be displayed to the user.

tmt.utils.jira.save_link_to_metadata(tmt_object: Test | Plan | Story, link: Link, logger: Logger) → None: Store the link into the object metadata on disk

tmt.utils.rest module

ReST rendering.

Package provides primitives for ReST rendering used mainly for CLI help texts.

tmt.utils.rest.NL = '': Special string representing a new-line in the stack of rendered paragraphs.

class tmt.utils.rest.RestVisitor(document: document, logger: Logger)

Bases: NodeVisitor

Custom renderer of docutils nodes.

See docutils.nodes.NodeVisitor for details, but the functionality is fairly simple: for each node type, a pair of methods is expected, visit_$NODE_TYPE and depart_$NODE_TYPE. As the visitor class iterates over nodes in the document, corresponding methods are called. These methods render the given node, filling “rendered paragraphs” list with rendered strings.

depart_Text(node: Node) → None

depart_bullet_list(node: bullet_list) → None

depart_document(node: document) → None

depart_inline(node: Node) → None

depart_list_item(node: list_item) → None

depart_literal(node: Node) → None

depart_literal_block(node: literal_block) → None

depart_note(node: note) → None

depart_paragraph(node: paragraph) → None

depart_reference(node: Node) → None

depart_warning(node: warning) → None

flush() → None: Finalize rendering of the current paragraph

nl() → None: Render a new, empty line

property rendered: str: Return the rendered document as a single string

unknown_departure(node: Node) → None

Called before exiting unknown Node types.

Raise exception unless overridden.

unknown_visit(node: Node) → None

Called when entering unknown Node types.

Raise an exception unless overridden.

visit_Text(node: Text) → None

visit_bullet_list(node: bullet_list) → None

visit_document(node: Node) → None

visit_inline(node: Node) → None

visit_list_item(node: list_item) → None

visit_literal(node: literal) → None

visit_literal_block(node: literal_block) → None

visit_note(node: note) → None

visit_paragraph(node: paragraph) → None

visit_reference(node: Node) → None

visit_warning(node: warning) → None

tmt.utils.rest.parse_rst(text: str) → document: Parse a ReST document into docutils tree of nodes

tmt.utils.rest.render_rst(text: str, logger: Logger) → str: Render a ReST document

tmt.utils.structured_field module

class tmt.utils.structured_field.StructuredField(text: str | None = None, version: int = 1, multi: bool = False)

Bases: object

Handling multiple text data in a single text field

The StructuredField allows you to easily store and extract several sections of text data to/from a single text field. The sections are separated by section names in square brackets and can be hosted in other text as well.

The section names have to be provided on a separate line and there must be no leading/trailing white space before/after the brackets. The StructuredField supports two versions of the format:

Version 0: Simple, concise, useful when neither the surrounding text or the section data can contain lines which could resemble section names. Here’s an example of a simple StructuredField:

Note written by human.

[section-one]
Section one content.

[section-two]
Section two content.

[section-three]
Section three content.

[end]

Another note written by human.

Version 1: Includes unique header to prevent collisions with the surrounding text and escapes any section-like lines in the content:

Note written by human.

[structured-field-start]
This is StructuredField version 1. Please, edit with care.

[section-one]
Section one content.

[section-two]
Section two content.
[structured-field-escape][something-resembling-section-name]

[section-three]
Section three content.

[structured-field-end]

Another note written by human.

Note that an additional empty line is added at the end of each section to improve the readability. This line is not considered to be part of the section content.

Besides handling the whole section content it’s also possible to store several key-value pairs in a single section, similarly as in the ini config format:

[section]
key1 = value1
key2 = value2
key3 = value3

Provide the key name as the optional argument ‘item’ when accessing these single-line items. Note that the section cannot contain both plain text data and key-value pairs.

field = qe.StructuredField()
field.set("project", "Project Name")
field.set("details", "somebody", "owner")
field.set("details", "2013-05-27", "started")
field.set("description", "This is a description.\n"
        "It spans across multiple lines.\n")
print field.save()

    [structured-field-start]
    This is StructuredField version 1. Please, edit with care.

    [project]
    Project Name

    [details]
    owner = somebody
    started = 2013-05-27

    [description]
    This is a description.
    It spans across multiple lines.

    [structured-field-end]

field.version(0)
print field.save()

    [project]
    Project Name

    [details]
    owner = somebody
    started = 2013-05-27

    [description]
    This is a description.
    It spans across multiple lines.

    [end]

Multiple values for the same key are supported as well. Enable this feature with ‘multi=True’ when initializing the structured field. If multiple values are present their list will be returned instead of a single string. Similarly use list for setting multiple values:

field = qe.StructuredField(multi=True)
requirements = ['hypervisor=', 'labcontroller=lab.example.com']
field.set("hardware", requirements, "hostrequire")
print field.save()

    [structured-field-start]
    This is StructuredField version 1. Please, edit with care.

    [hardware]
    hostrequire = hypervisor=
    hostrequire = labcontroller=lab.example.com

    [structured-field-end]

print field.get("hardware", "hostrequire")

    ['hypervisor=', 'labcontroller=lab.example.com']

Initialize the structured field

footer(content: str | None = None) → str: Get or set the footer content

get(section: str, item: str | None = None) → str | list[str]: Return content of given section or section item

header(content: str | None = None) → str: Get or set the header content

iterate() → Iterator[tuple[str, str]]: Return (section, content) tuples for all sections

load(text: str, version: int | None = None) → None: Load the StructuredField from a string

remove(section: str, item: str | None = None) → None: Remove given section or section item

save() → str: Convert the StructuredField into a string

sections() → list[str]: Get the list of available sections

set(section: str, content: Any, item: str | None = None) → None: Update content of given section or section item

version(version: int | None = None) → int: Get or set the StructuredField version

tmt.utils.templates module

Template rendering.

Package provides primitives for Jinja2 template rendering, plus our custom filters.

tmt.utils.templates.default_template_environment() → Environment

Create a Jinja2 environment with default settings.

Adds common filters, and enables block trimming and left strip.

tmt.utils.templates.render_template(template: str, template_filepath: Path | None = None, environment: Environment | None = None, **variables: Any) → str

Render a template.

Parameters:

template – template to render.
template_filepath – path to the template file, if any.
environment – Jinja2 environment to use.
variables – variables to pass to the template.

tmt.utils.templates.render_template_file(template_filepath: Path, environment: Environment | None = None, **variables: Any) → str

Render a template from a file.

Parameters:

template_filepath – path to the template file.
environment – Jinja2 environment to use.
variables – variables to pass to the template.

Module contents

Test Metadata Utilities

exception tmt.utils.BackwardIncompatibleDataError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: GeneralError

A backward incompatible data cannot be processed

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

class tmt.utils.CentOSDistGit

Bases: DistGitHandler

CentOS Handler

lookaside_server: str = 'https://sources.stream.centos.org/sources'

re_source: Pattern[str] = re.compile('^(\\w+) \$([^)]+)\$ = ([0-9a-fA-F]+)$')

remote_substring: Pattern[str] = re.compile('redhat/centos')

usage_name: str = 'centos'

class tmt.utils.Command(*elements: str | Path)

Bases: object

A command with its arguments.

run(*, cwd: Path | None, shell: bool = False, env: Environment | None = None, dry: bool = False, join: bool = False, interactive: bool = False, timeout: int | None = None, on_process_start: Callable[[Command, Popen[bytes], Logger], None] | None = None, message: str | None = None, friendly_command: str | None = None, log: LoggingFunction | None = None, silent: bool = False, stream_output: bool = True, caller: Common | None = None, logger: Logger) → CommandOutput

Run command, give message, handle errors.

Parameters:

cwd – if set, command would be executed in the given directory, otherwise the current working directory is used.
shell – if set, the command would be executed in a shell.
env – environment variables to combine with the current environment before running the command.
dry – if set, the command would not be actually executed.
join – if set, stdout and stderr of the command would be merged into a single output text.
interactive – if set, the command would be executed in an interactive manner, i.e. with stdout and stdout connected to terminal for live interaction with user.
timeout – if set, command would be interrupted, if still running, after this many seconds.
on_process_start – if set, this callable would be called after the command process started.
message – if set, it would be logged for more friendly logging.
friendly_command – if set, it would be logged instead of the command itself, to improve visibility of the command in logging output.
log – a logging function to use for logging of command output. By default, logger.debug is used.
silent – if set, logging of steps taken by this function would be reduced.
stream_output – if set, command output would be streamed live into the log. When unset, the output would be logged only when the command fails.
caller – optional “parent” of the command execution, used for better linked exceptions.
logger – logger to use for logging.

Returns:

command output, bundled in a CommandOutput tuple.

to_element() → str

Convert a command to a shell command line element.

Use when a command or just a list of command options should become a part of another command. Common examples of such “higher level” commands would be would be rsync -e or ansible-playbook --ssh-common-args.

to_popen() → list[str]: Convert a command to form accepted by subprocess.Popen

to_script() → ShellScript

Convert a command to a shell script.

Use when a command is supposed to become a part of a shell script.

class tmt.utils.CommandOutput(stdout: str | None, stderr: str | None)

Bases: object

stderr: str | None

stdout: str | None

Bases: _CommonBase

Common shared stuff

Takes care of command line context, options and workdir handling. Provides logging functions info(), verbose() and debug(). Implements read() and write() for comfortable file access. Provides the run() method for easy command execution.

Initialize name and relation with the parent object

Prepare the workdir for provided id / directory path or generate a new workdir name if workdir=True given. Store command line context and options for future use if context is provided.

cli_invocation: CliInvocation | None = None

property clone_dirpath: Path

Path for cloning into

Used internally for picking specific libraries (or anything else) from cloned repos for filtering purposes, it is removed at the end of relevant step.

Show message if in requested debug mode level

In quiet mode debug messages are not displayed.

property debug_level: int: The current debug level applied to this object

fail(message: str, shift: int = 0) → None: Show a red failure message on info level, send to stderr

ignore_class_options: bool = False

info(key: str, value: str | dict[str, Any] | int | bool | float | Environment | FmfContext | Path | Command | ShellScript | None = None, color: str | None = None, shift: int = 0) → None: Show a message unless in quiet mode

inject_logger(logger: Logger) → None

property is_dry_run: bool: Whether the current run is a dry-run

property is_feeling_safe: bool: Whether the current run is allowed to run unsafe actions

property is_forced_run: bool: Whether the current run is allowed to overwrite files and data

property name: str

opt(option: str, default: Any | None = None) → Any

Get an option from the command line options

Checks also parent options. For flags (boolean values) parent’s True wins over child’s False (e.g. run –quiet enables quiet mode for all included plans and steps).

For options that can be used multiple times, the child overrides the parent if it was defined (e.g. run -av provision -vvv runs all steps except for provision in mildly verbose mode, provision is run with the most verbosity).

Environment variables override command line options.

parent: Common | None = None

property pathless_safe_name: str

A safe variant of the name which does not contain any special characters.

Unlike safe_name, this property removes even slashes, /.

print(text: str, color: str | None = None, shift: int = 0) → None

Print out an output.

This method is supposed to be used for emitting a command output. Not to be mistaken with logging - errors, warnings, general command progress, and so on.

print() emits even when --quiet is used, as the option suppresses logging but not the actual command output.

property quietness: bool: The current quietness level applied to this object

read(path: Path, level: int = 2) → str: Read a file from the workdir

run(command: Command, friendly_command: str | None = None, silent: bool = False, message: str | None = None, cwd: Path | None = None, ignore_dry: bool = False, shell: bool = False, env: Environment | None = None, interactive: bool = False, join: bool = False, log: LoggingFunction | None = None, timeout: int | None = None, on_process_start: Callable[[Command, Popen[bytes], Logger], None] | None = None) → CommandOutput

Run command, give message, handle errors

Command is run in the workdir be default. In dry mode commands are not executed unless ignore_dry=True. Environment is updated with variables from the ‘env’ dictionary.

Output is logged using self.debug() or custom ‘log’ function. A user friendly command string ‘friendly_command’ will be shown, if provided, at the beginning of the command output.

Returns named tuple CommandOutput.

property safe_name: str

A safe variant of the name which does not contain special characters.

Spaces and other special characters are removed to prevent problems with tools which do not expect them (e.g. in directory names).

Unlike pathless_safe_name(), this property preserves slashes, /.

property should_run_again: bool: Whether selected step or the whole run should be run again

classmethod store_cli_invocation(context: Context | None, options: dict[str, Any] | None = None) → CliInvocation

Record a CLI invocation and options it carries for later use.

Warning

The given context is saved into a class variable, therefore it will function as a “default” context for instances on which store_cli_invocation() has not been called.

Parameters:

context – CLI context representing the invocation.
options – Optional dictionary with custom options. If provided, context is ignored.

Raises:

GeneralError – when there was a previously saved invocation already. Multiple invocations are not allowed.

Show message if in requested verbose mode level

In quiet mode verbose messages are not displayed.

property verbosity_level: int: The current verbosity level applied to this object

warn(message: str, shift: int = 0) → None: Show a yellow warning message on info level, send to stderr

property workdir: Path | None: Get the workdir, create if does not exist

property workdir_root: Path

write(path: Path, data: str, mode: Literal['w', 'a'] = 'w', level: int = 2) → None: Write a file to the workdir

tmt.utils.ContainerClass

Types for generic “data container” classes and instances. In tmt code, this reduces to data classes and data class instances. Our DataContainer are perfectly compatible data classes, but some helper methods may be used on raw data classes, not just on DataContainer instances.

alias of type[DataclassInstance]

exception tmt.utils.ConvertError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: MetadataError

Metadata conversion error

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

tmt.utils.DEFAULT_OUTPUT_WIDTH: int = 79: How wide should the output be at maximum. This is the default value tmt would use unless told otherwise.

class tmt.utils.DataContainer

Bases: object

A base class for objects that have keys and values

property is_bare: bool

Check whether all keys are either unset or have their default value.

Returns:: True if all keys either hold their default value or are not set at all, False otherwise.

items() → Iterator[tuple[str, Any]]: Iterate over key/value pairs

classmethod keys() → Iterator[str]: Iterate over key names

to_dict() → dict[str, Any]

Convert to a mapping.

See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions for more details.

to_minimal_dict() → dict[str, Any]

Convert to a mapping with unset keys omitted.

See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions for more details.

values() → Iterator[Any]: Iterate over key values

exception tmt.utils.DiscoverError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: GeneralError

Discover step error

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

class tmt.utils.DistGitHandler

Bases: object

Common functionality for DistGit handlers

its_me(remotes: list[str]) → bool: True if self can work with remotes

lookaside_server: str

re_source: Pattern[str]

re_supported_extensions: Pattern[str] = re.compile('\\.((tar\\.(gz|Z|bz2|lz|lzma|lzo|xz|zst))|tgz|taz|taZ|tz2|tbz2|tbz|tlz|tzst)$')

remote_substring: Pattern[str]

sources_file_name = 'sources'

uri = '/rpms/{name}/{filename}/{hashtype}/{hash}/{filename}'

url_and_name(cwd: Path | None = None) → list[tuple[str, str]]

Return list of urls and basenames of the used source

The ‘cwd’ parameter has to be a DistGit directory.

usage_name: str

tmt.utils.EnvVarName: A type of environment variable name.

class tmt.utils.EnvVarValue(raw_value: Any)

Bases: str

A type of environment variable value

class tmt.utils.Environment(data: dict[str, EnvVarValue] | None = None)

Bases: dict[str, EnvVarValue]

Represents a set of environment variables.

See https://tmt.readthedocs.io/en/latest/spec/tests.html#environment, https://tmt.readthedocs.io/en/latest/spec/plans.html#environment and https://tmt.readthedocs.io/en/latest/spec/plans.html#environment-file.

as_environ() → Iterator[None]

A context manager replacing os.environ with this environment.

When left, the original content of os.environ is restored.

Warning

This method is not thread safe! Beware of using it in code that runs in multiple threads, e.g. from provision/prepare/execute/finish phases.

copy() → Environment: Return a shallow copy of the dict.

classmethod from_dict(data: dict[str, Any] | None = None) → Environment: Create environment variables from a dictionary

classmethod from_dotenv(content: str) → Environment

Construct environment from a .env format.

Parameters:: content – string containing variables defined in the “dotenv” format, https://hexdocs.pm/dotenvy/dotenv-file-format.html.

classmethod from_environ() → Environment: Extract environment variables from the live environment

classmethod from_file(*, filename: str, root: Path | None = None, logger: Logger) → Environment

Construct environment from a file.

YAML files - recognized by .yaml or .yml suffixes - or .env-like files are supported.

dotenv file example

A=B
C=D

YAML file example

A: B
C: D

Note

For loading environment variables from multiple files, see Environment.from_files().

classmethod from_files(*, filenames: Iterable[str], root: Path | None = None, logger: Logger) → Environment

Read environment variables from the given list of files.

Files should be in YAML format (.yaml or .yml suffixes), or in dotenv format.

dotenv file example

A=B
C=D

YAML file example

A: B
C: D

Path to each file should be relative to the metadata tree root.

Note

For loading environment variables from a single file, see Environment.from_file(), which is a method called for each file, accumulating data from all input files.

classmethod from_fmf_context(fmf_context: FmfContext) → Environment: Create environment variables from an fmf context

classmethod from_fmf_spec(data: dict[str, Any] | None = None) → Environment: Create environment from an fmf specification

classmethod from_inputs(*, raw_fmf_environment: Any = None, raw_fmf_environment_files: Any = None, raw_cli_environment: Any = None, raw_cli_environment_files: Any = None, file_root: Path | None = None, key_address: str | None = None, logger: Logger) → Environment

Extract environment variables from various sources.

Combines various raw sources into a set of environment variables. Calls necessary functions to process environment files, dictionaries and CLI inputs.

All inputs are optional, and there is a clear order of preference, which is, from the most preferred:

--environment CLI option (raw_cli_environment)
--environment-file CLI option (raw_cli_environment_files)
environment fmf key (raw_fmf_environment)
environment-file fmf key (raw_fmf_environment_files)

Parameters:

raw_fmf_environment – content of environment fmf key. None and a dictionary are accepted.
raw_fmf_environment_files – content of environment-file fmf key. None and a list of paths are accepted.
raw_cli_environment – content of --environment CLI option. None, a tuple or a list are accepted.
raw_cli_environment_files – content of –environment-file` CLI option. None, a tuple or a list are accepted.

Raises:

NormalizationError – when an input is of a type which is not allowed for that particular source.

classmethod from_sequence(variables: str | list[str], logger: Logger) → Environment

Construct environment from a sequence of variables.

Variables may be specified in two ways:

NAME=VALUE pairs, or
@foo.yaml signaling variables to be read from a file.

If a “variable” starts with @, it is treated as a path to a YAML file that contains key/value pairs which are then transparently loaded and added to the final environment.

Parameters:

variables –

string or a sequence of strings containing variables. The acceptable formats are:

'X=1'
'X=1 Y=2 Z=3'
['X=1', 'Y=2', 'Z=3']
['X=1 Y=2 Z=3', 'A=1 B=2 C=3']
'TXT="Some text with spaces in it"'
@foo.yaml
@../../bar.yaml

classmethod from_yaml(content: str) → Environment

Construct environment from a YAML format.

Parameters:: content – string containing variables defined in a YAML dictionary, i.e. key: value entries.

classmethod from_yaml_file(filepath: Path, logger: Logger) → Environment

Construct environment from a YAML file.

File is expected to contain variables in a YAML dictionary, i.e. key: value entries. Only primitive types - strings, numbers, booleans - are allowed as values.

Parameters:

path – path to the file with variables.
logger – used for logging.

classmethod normalize(key_address: str, value: Any, logger: Logger) → Environment: Normalize value of environment key

to_environ() → dict[str, str]: Convert to a form compatible with os.environ

to_fmf_spec() → dict[str, str]: Convert to an fmf specification

to_popen() → dict[str, str]: Convert to a form accepted by subprocess.Popen

exception tmt.utils.ExecuteError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: GeneralError

Execute step error

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

class tmt.utils.FedoraDistGit

Bases: DistGitHandler

Fedora Handler

lookaside_server: str = 'https://src.fedoraproject.org/repo/pkgs'

re_source: Pattern[str] = re.compile('^(\\w+) \$([^)]+)\$ = ([0-9a-fA-F]+)$')

remote_substring: Pattern[str] = re.compile('fedoraproject\\.org')

usage_name: str = 'fedora'

tmt.utils.FieldCLIOption

Type of field’s CLI option specification.

alias of str | Sequence[str]

tmt.utils.FieldExporter

Type of field’s exporter callback.

alias of Callable[[T], Any]

class tmt.utils.FieldMetadata(internal: bool = False, help: str | None = None, _metavar: str | None = None, default: ~tmt.utils.T | None = None, default_factory: ~typing.Callable[[], ~tmt.utils.T] | None = None, is_flag: bool = False, multiple: bool = False, show_default: bool = False, _choices: None | ~collections.abc.Sequence[str] | ~typing.Callable[[], ~collections.abc.Sequence[str]] = None, envvar: str | None = None, deprecated: ~tmt.options.Deprecated | None = None, cli_option: str | ~collections.abc.Sequence[str] | None = None, normalize_callback: ~typing.Callable[[str, ~typing.Any, ~tmt.log.Logger], ~tmt.utils.T] | None = None, serialize_callback: ~typing.Callable[[~tmt.utils.T], ~typing.Any] | None = None, unserialize_callback: ~typing.Callable[[~tmt.utils.T], ~typing.Any] | None = None, export_callback: ~typing.Callable[[~tmt.utils.T], ~typing.Any] | None = None, _option_args: str | ~collections.abc.Sequence[str] | None = None, _option_kwargs: dict[str, ~typing.Any] = <factory>, _option: ~typing.Callable[[~typing.Any], ~typing.Any] | None = None)

Bases: Generic[T]

A dataclass metadata container used by our custom dataclass field management.

Attached to fields defined with field()

property choices: Sequence[str] | None: A list of allowed values the field can take

cli_option: str | Sequence[str] | None = None: One or more command-line option names.

default: T | None = None: The default value for the field.

default_factory: Callable[[], T] | None = None: A zero-argument callable that will be called when a default value is needed for the field.

deprecated: Deprecated | None = None: Mark the option as deprecated. Instance of Deprecated describes the version in which the field was deprecated plus an optional hint with the recommended alternative. Documentation and help texts would contain this info.

envvar: str | None = None: Environment variable providing value for the field.

export_callback: Callable[[T], Any] | None = None: An export callback to call when exporting the field (performed by tmt.export.Exportable).

property has_default: bool: Whether the field has a default value

help: str | None = None: Help text documenting the field.

internal: bool = False

is_flag: bool = False: Marks the fields as a flag.

property materialized_default: T | None: Returns the actual default value of the field

property metavar: str | None: Placeholder for field’s value in documentation and help

multiple: bool = False: Marks the field as accepting multiple values. When used on command line, the option could be used multiple times, accumulating values.

normalize_callback: Callable[[str, Any, Logger], T] | None = None: A normalization callback to call when loading the value from key source (performed by NormalizeKeysMixin).

property option: Callable[[Any], Any] | None

serialize_callback: Callable[[T], Any] | None = None

show_default: bool = False: If set, show the default value in command line help.

unserialize_callback: Callable[[T], Any] | None = None

exception tmt.utils.FileError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: GeneralError

File operation error

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

exception tmt.utils.FinishError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: GeneralError

Finish step error

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

class tmt.utils.FmfContext(data: dict[str, list[str]] | None = None)

Bases: dict[str, list[str]]

Represents an fmf context.

See https://tmt.readthedocs.io/en/latest/spec/context.html and https://fmf.readthedocs.io/en/latest/context.html.

classmethod from_spec(key_address: str, spec: Any, logger: Logger) → FmfContext

Convert from a specification file or from a CLI option.

See https://tmt.readthedocs.io/en/stable/spec/context.html for details on context.

to_spec() → dict[str, Any]: Convert to a form suitable for saving in a specification file

exception tmt.utils.GeneralError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: Exception

General error

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

exception tmt.utils.GitUrlError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: GeneralError

Remote git url is not reachable

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

tmt.utils.KeySource

A type representing compatible sources of keys and values.

alias of dict[str, Any] | Tree

class tmt.utils.ListFormat(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

How to format lists

LISTED = 1: Use fmf.utils.listed().

LONG = 3: One list item per line.

SHORT = 2: Produce comma-separated list.

class tmt.utils.LoadFmfKeysMixin(*, node: Tree, logger: Logger, **kwargs: Any): Bases: NormalizeKeysMixin

exception tmt.utils.MetadataError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: GeneralError

General metadata error

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

class tmt.utils.MultiInvokableCommon(**kwargs: Any)

Bases: Common

Initialize name and relation with the parent object

Prepare the workdir for provided id / directory path or generate a new workdir name if workdir=True given. Store command line context and options for future use if context is provided.

cli_invocation: CliInvocation | None = None

cli_invocations: list[CliInvocation] = []

classmethod store_cli_invocation(context: Context | None, options: dict[str, Any] | None = None) → CliInvocation

Save a CLI context and options it carries for later use.

Warning

The given context is saved into a class variable, therefore it will function as a “default” context for instances on which _save_cli_context_to_instance() has not been called.

Warning

The given context will overwrite any previously saved context.

Parameters:

context – CLI context to save.
options – Optional dictionary with custom options. If provided, context is ignored.

exception tmt.utils.NormalizationError(key_address: str, raw_value: Any, expected_type: str, *args: Any, **kwargs: Any)

Bases: SpecificationError

Raised when a key normalization fails

Raised when a key normalization fails.

A subclass of SpecificationError, but describing errors that appear in a very specific point of key loading in a unified manner.

Parameters:

key_address – the key in question, preferably with detailed location, e.g. /plans/foo:discover[0].tests.
raw_value – input value, the one that failed the normalization.
expected_type – string description of expected, allowed types, as a hint in the error message.

tmt.utils.NormalizeCallback

Type of field’s normalization callback.

alias of Callable[[str, Any, Logger], T]

class tmt.utils.NormalizeKeysMixin(**kwargs: Any)

Bases: _CommonBase

Mixin adding support for loading fmf keys into object attributes.

When invoked, annotated class-level variables are searched for in a given source container - a mapping, an fmf node, etc. - and if the key of the same name as the variable exists, its value is “promoted” to instance variable.

If a method named _normalize_<variable name> exists, it is called with the fmf key value as its single argument, and its return value is assigned to instance variable. This gives class chance to modify or transform the original value when needed, e.g. to convert the original value to a type more suitable for internal processing.

items() → Iterator[tuple[str, Any]]

Iterate over keys and their values.

Keys are yielded in the order: keys declared by parent classes first, then keys declared by the class itself, all following the order in which keys were defined in their respective classes.

Yields:: pairs of key name and its value.

classmethod keys() → Iterator[str]

Iterate over key names.

Keys are yielded in the order: keys declared by parent classes first, then keys declared by the class itself, all following the order in which keys were defined in their respective classes.

Yields:: key names.

tmt.utils.OUTPUT_WIDTH: int = 79: How wide should the output be at maximum. This is the effective value, combining the default and optional envvar, TMT_OUTPUT_WIDTH.

tmt.utils.OnProcessStartCallback

Type of a callable to be called by Command.run() after starting the child process.

alias of Callable[[Command, Popen[bytes], Logger], None]

exception tmt.utils.PrepareError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: GeneralError

Prepare step error

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

class tmt.utils.ProcessExitCodes(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: IntEnum

FAILURE = 1: Unsuccessful run.

NOT_FOUND = 127: Command not found, or PATH error.

PERMISSION_DENIED = 126: Permission denied (or) unable to execute.

SIGINT = 130: Terminated by either Ctrl+C combo or SIGINT signal.

SIGTERM = 143: Terminated by a SIGTERM signal.

SUCCESS = 0: Successful run.

TEST_PIDFILE_LOCK_FAILED = 122: tmt pidfile lock operation failed.

TEST_PIDFILE_UNLOCK_FAILED = 123: tmt pidfile unlock operation failed.

TIMEOUT = 124: Command was terminated because of a timeout.

classmethod format(exit_code: int) → str | None: Format a given exit code for nicer logging

classmethod is_pidfile(exit_code: int | None) → bool

exception tmt.utils.ProvisionError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: GeneralError

Provision step error

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

tmt.utils.RawCommand

A raw command line form, a list of elements.

alias of list[str | Path]

tmt.utils.RawCommandElement

A single element of raw command line in its list form.

alias of str | Path

exception tmt.utils.RebootTimeoutError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: ExecuteError

Reboot failed due to a timeout

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

class tmt.utils.RedHatGitlab

Bases: DistGitHandler

Red Hat on Gitlab

lookaside_server: str = 'http://pkgs.devel.redhat.com/repo'

re_source: Pattern[str] = re.compile('^(\\w+) \$([^)]+)\$ = ([0-9a-fA-F]+)$')

remote_substring: Pattern[str] = re.compile('redhat/rhel/')

usage_name: str = 'redhatgitlab'

exception tmt.utils.ReportError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: GeneralError

Report step error

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

exception tmt.utils.RetryError(label: str, causes: list[Exception])

Bases: GeneralError

Retries unsuccessful

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

class tmt.utils.RetryStrategy(*args: Any, logger: Logger | None = None, **kwargs: Any)

Bases: Retry

increment(*args: Any, **kwargs: Any) → Retry

Return a new Retry object with incremented retry counters.

Parameters:

response (BaseHTTPResponse) – A response object, or None, if the server did not return a response.
error (Exception) – An error encountered during the request, or None if the response was received successfully.

Returns:

A new Retry object.

new(**kw: Any) → Self

exception tmt.utils.RunError(message: str, command: Command, returncode: int, stdout: str | None = None, stderr: str | None = None, caller: Common | None = None, *args: Any, **kwargs: Any)

Bases: GeneralError

Command execution error

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

property output: CommandOutput: Captured output of the command.

Note

This field contains basically the same info as stdout and stderr, but it’s bundled into a single object. This is how command output is passed between functions, therefore the exception should offer it as well.

class tmt.utils.SerializableContainer

Bases: DataContainer

A mixin class for saving and loading objects

classmethod default(key: str, default: Any = None) → Any

classmethod extract_from(obj: Any) → SerializableContainerDerivedType: Extract keys from given object, and save them in a container

classmethod from_serialized(serialized: dict[str, Any]) → SerializableContainerDerivedType

Convert from a serialized form loaded from a file.

See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions for more details.

See to_serialized() for its counterpart.

inject_to(obj: Any) → None: Inject keys from this container into attributes of a given object

to_serialized() → dict[str, Any]

Convert to a form suitable for saving in a file.

See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions for more details.

See from_serialized() for its counterpart.

static unserialize(serialized: dict[str, Any], logger: Logger) → SerializableContainerDerivedType

Convert from a serialized form loaded from a file.

Similar to from_serialized(), but this method knows nothing about container’s class, and will locate the correct module and class by inspecting serialized data. Discovered class’ from_serialized() is then used to create the container.

Used to transform data read from a YAML file into original containers when their classes are not know to the code. Restoring such containers requires inspection of serialized data and dynamic imports of modules as needed.

See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions for more details.

See to_serialized() for its counterpart.

tmt.utils.SerializeCallback

Type of field’s serialization callback.

alias of Callable[[T], Any]

class tmt.utils.ShellScript(script: str)

Bases: object

A shell script, a free-form blob of text understood by a shell.

Parameters:: script – the actual script to be encapsulated by ShellScript wrapper.

classmethod from_scripts(scripts: list[ShellScript]) → ShellScript

Create a single script from many shorter ones.

Scripts are merged into a single ShellScript instance, joined together with ; character.

Parameters:: scripts – scripts to merge into one.

to_element() → str: Convert a shell script to a command element

to_shell_command() → Command

Convert a shell script into a shell-driven command.

Turns a shell script into a full-fledged command one might pass to the OS. Basically what would run(script, shell=True) do.

class tmt.utils.SpecBasedContainer

Bases: Generic[SpecInT, SpecOutT], DataContainer

classmethod from_spec(spec: SpecInT) → SpecBasedContainerT

Convert from a specification file or from a CLI option

See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions for more details.

See to_spec() for its counterpart.

to_minimal_spec() → SpecOutT

Convert to specification, skip default values

See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions for more details.

See from_spec() for its counterpart.

to_spec() → SpecOutT

Convert to a form suitable for saving in a specification file

See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions for more details.

See from_spec() for its counterpart.

class tmt.utils.SpecBasedContainerT

A typevar bound to spec-based container base class. A stand-in for all classes derived from SpecBasedContainer.

alias of TypeVar(‘SpecBasedContainerT’, bound=SpecBasedContainer)

class tmt.utils.SpecInT

A typevar representing an input specification consumed by SpecBasedContainer.

alias of TypeVar(‘SpecInT’)

class tmt.utils.SpecOutT

A typevar representing an output specification produced by SpecBasedContainer.

alias of TypeVar(‘SpecOutT’)

exception tmt.utils.SpecificationError(message: str, validation_errors: list[tuple[ValidationError, str]] | None = None, *args: Any, **kwargs: Any)

Bases: MetadataError

Metadata specification error

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

class tmt.utils.Stopwatch

Bases: AbstractContextManager[Stopwatch]

property duration: timedelta

end_time: datetime

start_time: datetime

class tmt.utils.StreamLogger(log_header: str, *, stream: IO[bytes] | None = None, logger: LoggingFunction | None = None, click_context: Context | None = None, stream_output: bool = True)

Bases: Thread

Reading pipes of running process in threads.

Code based on: https://github.com/packit/packit/blob/main/packit/utils/logging.py#L10

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is a list or tuple of arguments for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

get_output() → str | None

run() → None

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

exception tmt.utils.StructuredFieldError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: GeneralError

StructuredField parsing error

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

class tmt.utils.TimeoutHTTPAdapter(*args: Any, **kwargs: Any)

Bases: HTTPAdapter

Spice up request’s session with custom timeout

send(request: PreparedRequest, **kwargs: Any) → Response

Send request.

All arguments are passed to superclass after enforcing the timeout.

Parameters:: request – the request to send.

class tmt.utils.TracebackVerbosity(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Levels of logged traveback verbosity

DEFAULT = '0': Render only exception and its causes.

FULL = 'full': Render also call stack and local variables for exception and each of its causes.

VERBOSE = '1': Render also call stack for exception and each of its causes.

classmethod from_env() → TracebackVerbosity

classmethod from_spec(spec: str) → TracebackVerbosity

tmt.utils.UnserializeCallback

Type of field’s unserialization callback.

alias of Callable[[Any], T]

class tmt.utils.UnusedStreamLogger(log_header: str)

Bases: StreamLogger

Special variant of StreamLogger that records no data.

It is designed to make the implementation of merged streams easier in Command.run(). Instance of this class is created to log stderr when, in fact, stderr is merged into stdout. This class returns values compatible with CommandOutput notion of “no output”.

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is a list or tuple of arguments for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

get_output() → str | None

run() → None

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

class tmt.utils.UpdatableMessage(key: str, enabled: bool = True, indent_level: int = 0, key_color: str | None = None, default_value_color: str | None = None, clear_on_exit: bool = False)

Bases: AbstractContextManager

Updatable message suitable for progress-bar-like reporting

Updatable message suitable for progress-bar-like reporting.

with UpdatableMessage('foo') as message:
    while ...:
        ...

        # check state of remote request, and update message
        state = remote_api.check()
        message.update(state)

Parameters:

key – a string to use as the left-hand part of logged message.
enabled – if unset, no output would be performed.
indent_level – desired indentation level.
key_color – optional color to apply to key.
default_color – optional color to apply to value when update() is called with color left out.
clear_on_exit – if set, the message area would be cleared when leaving the progress bar when used as a context manager.

clear() → None: Clear the message area

update(value: str, color: str | None = None) → None

Update progress message.

Parameters:

value – new message to update message area with.
color – optional message color.

class tmt.utils.ValidateFmfMixin(*, node: Tree, skip_validation: bool = False, raise_on_validation_error: bool = False, logger: Logger, **kwargs: Any)

Bases: _CommonBase

Mixin adding validation of an fmf node.

Loads a schema whose name is derived from class name, and uses fmf’s validate() method to perform the validation.

tmt.utils.ValueFormatter

A type describing a per-type formatting helper.

alias of Callable[[Any, int | None, str | None, ListFormat, Literal[True, False, ‘auto’]], Iterator[str]]

exception tmt.utils.WaitingIncompleteError

Bases: GeneralError

Waiting incomplete

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

exception tmt.utils.WaitingTimedOutError(check: Callable[[], T], timeout: timedelta, check_success: bool = False)

Bases: GeneralError

Waiting ran out of time

General error.

Parameters:

message – error message.
causes – optional list of exceptions that caused this one. Since raise ... from ... allows only for a single cause, and some of our workflows may raise exceptions triggered by more than one exception, we need a mechanism for storing them. Our reporting will honor this field, and report causes the same way as __cause__.

tmt.utils.assert_window_size(window_size: int | None) → None

Raise an exception if window size is zero or a negative integer.

Protects possible underflows in formatters employed by format_value().

tmt.utils.catch_warnings_safe(action: ~typing.Literal['default', 'error', 'ignore', 'always', 'module', 'once'], category: type[Warning] = <class 'Warning'>) → Iterator[None]

Optionally catch the given warning category.

Using this context manager you can catch/suppress given warnings category. These warnings gets re-enabled/reset with an exit from this context manager.

This function uses a reentrant lock for thread synchronization to be a thread-safe. That’s why it’s wrapping warnings.catch_warnings() instead of using it directly.

The example can be suppressing of the urllib insecure request warning:

with catch_warnings_safe('ignore', urllib3.exceptions.InsecureRequestWarning):
    ...

tmt.utils.configure_constant(default: int, envvar: str) → int

Deduce the actual value of global constant.

Parameters:

default – the default value of the constant.
envvar – name of the optional environment variable which would override the default value.

Returns:

value extracted from the environment variable, or the given default value if the variable did not exist.

tmt.utils.configure_optional_constant(default: int | None, envvar: str) → int | None

Deduce the actual value of a global constant which may be left unset.

Parameters:

default – the default value of the constant.
envvar – name of the optional environment variable which would override the default value.

Returns:

value extracted from the environment variable, or the given default value if the variable did not exist.

tmt.utils.container_field(container: type[DataclassInstance] | DataclassInstance, key: str) → tuple[str, str, Any, Field[Any], FieldMetadata[Any]]

Return a dataclass/data container field info by the field’s name.

Surprisingly, dataclasses package does not have a helper for this. One can iterate over fields, but there’s no public API for retrieving a field when one knows its name.

Parameters:

cls – a dataclass/data container class whose fields to search.
key – field name to retrieve.

Raises:

GeneralError – when the field does not exist.

tmt.utils.container_fields(container: type[DataclassInstance] | DataclassInstance) → Iterator[Field[Any]]

tmt.utils.container_items(container: DataclassInstance) → Iterator[tuple[str, Any]]: Iterate over key/value pairs in a container

tmt.utils.container_keys(container: type[DataclassInstance] | DataclassInstance) → Iterator[str]: Iterate over key names in a container

tmt.utils.container_values(container: DataclassInstance) → Iterator[Any]: Iterate over values in a container

tmt.utils.create_directory(*, path: Path, name: str, dry: bool = False, quiet: bool = False, logger: Logger) → None

Create a new directory.

Before creating the directory, function checks whether it exists already - the existing directory is not removed and re-created.

The outcome of the operation will be logged in a debug log, but may also be sent to console with quiet=False.

Parameters:

path – a path to be created.
name – a “label” of the path, used for logging.
dry – if set, directory would not be created. Still, the existence check will happen.
quiet – if set, an outcome of the operation would not be logged to console.
logger – logger to use for logging.

Raises:

FileError – when function tried to create the directory, but failed.

tmt.utils.create_file(*, path: Path, content: str, name: str, dry: bool = False, force: bool = False, mode: int = 436, quiet: bool = False, logger: Logger) → None

Create a new file.

Before creating the file, function checks whether it exists already - the existing file is not removed and re-created, unless force is set.

The outcome of the operation will be logged in a debug log, but may also be sent to console with quiet=False.

Parameters:

path – a path to be created.
content – content to save into the file
name – a “label” of the path, used for logging.
dry – if set, the file would not be created or overwritten. Still, the existence check will happen.
force – if set, the file would be overwritten if it already exists.
mode – permissions to set for the file.
quiet – if set, an outcome of the operation would not be logged to console.
logger – logger to use for logging.

Raises:

FileError – when function tried to create the file, but failed.

tmt.utils.dataclass_normalize_field(container: Any, key_address: str, keyname: str, raw_value: Any, logger: Logger) → Any

Normalize and assign a value to container field.

If there is a normalization callback defined for the field via normalize parameter of field(), the callback is called to coerce raw_value, and the return value is assigned to container field instead of value.

tmt.utils.dict_to_yaml(data: dict[str, Any] | list[Any] | _RawFmfId, width: int | None = None, sort: bool = False, start: bool = False) → str: Convert dictionary into yaml

tmt.utils.distgit_download(*, distgit_dir: Path, target_dir: Path, handler_name: str | None = None, caller: Common | None = None, logger: Logger) → None

Download sources to the target_dir

distgit_dir is path to the DistGit repository

tmt.utils.duplicates(values: Iterable[T | None]) → Iterator[T]: Iterate over all duplicate values in values

tmt.utils.duration_to_seconds(duration: str, injected_default: str | None = None) → int

Convert extended sleep time format into seconds

Optional ‘injected_default’ argument to evaluate ‘duration’ when it contains only multiplication.

tmt.utils.effective_workdir_root(workdir_root_option: Path | None = None) → Path

Find out what the actual workdir root is.

If the workdir-root cli option is set, it is used as the workdir root. Otherwise, the TMT_WORKDIR_ROOT environment variable is used if set. If neither is specified, the default value of :py:data:WORKDIR_ROOT is used.

tmt.utils.field(*, default: bool, option: FieldCLIOption | None = None, is_flag: bool = True, choices: None | Sequence[str] | Callable[[], Sequence[str]] = None, multiple: bool = False, metavar: str | None = None, envvar: str | None = None, deprecated: 'tmt.options.Deprecated' | None = None, help: str | None = None, show_default: bool = False, internal: bool = False) → bool

tmt.utils.field(*, default: T, option: FieldCLIOption | None = None, is_flag: bool = False, choices: None | Sequence[str] | Callable[[], Sequence[str]] = None, multiple: bool = False, metavar: str | None = None, envvar: str | None = None, deprecated: 'tmt.options.Deprecated' | None = None, help: str | None = None, show_default: bool = False, internal: bool = False, normalize: NormalizeCallback[T] | None = None, serialize: SerializeCallback[T] | None = None, unserialize: UnserializeCallback[T] | None = None, exporter: FieldExporter[T] | None = None) → T

tmt.utils.field(*, default_factory: Callable[[], T], option: FieldCLIOption | None = None, is_flag: bool = False, choices: None | Sequence[str] | Callable[[], Sequence[str]] = None, multiple: bool = False, metavar: str | None = None, envvar: str | None = None, deprecated: 'tmt.options.Deprecated' | None = None, help: str | None = None, show_default: bool = False, internal: bool = False, normalize: NormalizeCallback[T] | None = None, serialize: SerializeCallback[T] | None = None, unserialize: UnserializeCallback[T] | None = None, exporter: FieldExporter[T] | None = None) → T

tmt.utils.field(*, option: FieldCLIOption | None = None, is_flag: bool = False, choices: None | Sequence[str] | Callable[[], Sequence[str]] = None, multiple: bool = False, metavar: str | None = None, envvar: str | None = None, deprecated: 'tmt.options.Deprecated' | None = None, help: str | None = None, show_default: bool = False, internal: bool = False, normalize: NormalizeCallback[T] | None = None, serialize: SerializeCallback[T] | None = None, unserialize: UnserializeCallback[T] | None = None, exporter: FieldExporter[T] | None = None) → T

Define a DataContainer field.

Effectively a fancy wrapper over dataclasses.field(), tailored for tmt code needs and simplification of various common tasks.

Parameters:

default – if provided, this will be the default value for this field. Passed directly to dataclass.field(). It is an error to specify both default and default_factory.
default_factory – if provided, it must be a zero-argument callable that will be called when a default value is needed for this field. Passed directly to dataclass.field(). It is an error to specify both default and default_factory.
option – one or more command-line option names. Passed directly to click.option().
is_flag – marks this option as a flag. Passed directly to click.option().
choices – if provided, the command-line option would accept only the listed input values. Passed to click.option() as a click.Choice instance.
multiple – accept multiple arguments of the same name. Passed directly to click.option().
metavar – how the input value is represented in the help page. Passed directly to click.option().
envvar – environment variable used for this option. Passed directly to click.option().
deprecated – mark the option as deprecated Provide an instance of Deprecated() with version in which the option was obsoleted and an optional hint with the recommended alternative. A warning message will be added to the option help.
help – the help string for the command-line option. Multiline strings can be used, textwrap.dedent() is applied before passing help to click.option().
show_default – show default value Passed directly to click.option().
internal – if set, the field is treated as internal-only, and will not appear when showing objects via show() method, or in export created by Core._export().
normalize – a callback for normalizing the input value. Consumed by NormalizeKeysMixin.
serialize – a callback for custom serialization of the field value. Consumed by SerializableKeysMixin.
unserialize – a callback for custom unserialization of the field value. Consumed by SerializableKeysMixin.
exporter – a callback for custom export of the field value. Consumed by tmt.export.Exportable.

tmt.utils.filter_paths(directory: Path, searching: list[str], files_only: bool = False) → list[Path]

Filter files for specific paths we are searching for inside a directory

Returns list of matching paths.

tmt.utils.find_fmf_root(path: Path, ignore_paths: list[Path] | None = None) → list[Path]

Search through path and return all fmf roots that exist there

Returned list is ordered by path length, shortest one first.

Raise MetadataError if no fmf root is found.

tmt.utils.flatten(lists: Iterable[list[T]], unique: bool = False) → list[T]

“Flatten” a list of lists into a single-level list.

Parameters:

lists – an iterable of lists to flatten.
unique – if set, duplicate items would be removed, leaving only a single instance in the final list.

Returns:

list of items from all given lists.

tmt.utils.fmf_id(*, name: str, fmf_root: Path, logger: Logger) → FmfId: Return full fmf identifier of the node

Nicely format and indent a key-value pair

Parameters:

key – a key introducing the value.
value – an object to format.
indent – the key would be right-justified to this column.
window_size – rendering will try to fit produce lines whose length would exceed window_size. A window not wide enough may result into not using fmf.utils.listed() for lists, or wrapping lines in a text paragraph.
wrap – if set to True, always reformat text and wrap long lines; if set to False, preserve text formatting and make no changes; the default, auto, tries to rewrap lines as needed to obey window_size.
key_color – if set, dictionary keys would be colorized by this color.
list_format – preferred list formatting. It may be ignored if window_size is set and not wide enough to hold the desired formatting; ListFormat.LONG would be the fallback choice.

Returns:

a formatted string representation of value.

tmt.utils.format_duration(duration: timedelta) → str: Convert duration to a human readable format

tmt.utils.format_timestamp(timestamp: datetime) → str: Convert timestamp to a human readable format

tmt.utils.format_value(value: Any, window_size: int | None = None, key_color: str | None = None, list_format: ListFormat = ListFormat.LISTED, wrap: Literal[True, False, 'auto'] = 'auto') → str

Render a nicely-formatted string representation of a value.

Parameters:

value – an object to format.
window_size – if set, rendering will try to produce lines whose length would not exceed window_size. A window not wide enough may result into not using fmf.utils.listed(), or wrapping lines in a text paragraph.
key_color – if set, dictionary keys would be colorized by this color.
list_format – preferred list formatting. It may be ignored if window_size is set and not wide enough to hold the desired formatting; ListFormat.LONG would be the fallback choice.

Returns:

a formatted string representation of value.

tmt.utils.generate_runs(path: Path, id_: tuple[str, ...]) → Iterator[Path]: Generate absolute paths to runs from path

tmt.utils.get_distgit_handler(remotes: list[str] | None = None, usage_name: str | None = None) → DistGitHandler

Return the right DistGitHandler

Pick the DistGitHandler class which understands specified remotes or by usage_name.

tmt.utils.get_distgit_handler_names() → list[str]: All known distgit handlers

tmt.utils.get_full_metadata(fmf_tree_path: Path, node_path: str) → Any

Get full metadata for a node in any fmf tree

Go through fmf tree nodes using given relative node path and return full data as dictionary.

tmt.utils.get_url_content(url: str) → str: Get content of a given URL as a string

tmt.utils.is_key_origin(node: Tree, key: str) → bool

Find out whether the given key is defined in the given node.

Parameters:

node – node to check.
key – key to check.

Returns:

True if the key is defined in node, not by one of its parents, False otherwise.

tmt.utils.is_url(url: str) → bool: Check if the given string is a valid URL

tmt.utils.json_to_list(data: Any) → list[Any]: Convert json into list

tmt.utils.key_to_option(key: str) → str: Convert a key name to corresponding option name

tmt.utils.load_run(run: Run) → tuple[bool, Exception | None]: Load a run and its steps from the workdir

tmt.utils.load_schema(schema_filepath: Path) → dict[str, Any]

Load a JSON schema from a given filepath.

Recommended for general use, the method may apply some post-loading touches to the given schema, and unless caller is interested in the raw content of the file, this functions should be used instead of the real workhorse of schema loading, _load_schema().

tmt.utils.load_schema_store() → dict[str, dict[str, Any]]

Load all available JSON schemas, and put them into a “store”.

Schema store is a simple mapping between schema IDs and schemas.

tmt.utils.locate_key_origin(node: Tree, key: str) → Tree | None

Find an fmf node where the given key is defined.

Parameters:

node – node to begin with.
key – key to look for.

Returns:

first node in which the key is defined, None if node nor any of its parents define it.

tmt.utils.markdown_to_html(filename: Path) → str

Convert markdown to html

Expects: Markdown document as a file. Returns: An HTML document as a string.

tmt.utils.normalize_adjust(key_address: str, raw_value: Any, logger: Logger) → list[_RawAdjustRule] | None

tmt.utils.normalize_data_amount(key_address: str, raw_value: Any, logger: Logger) → Size

tmt.utils.normalize_int(key_address: str, value: Any, logger: Logger) → int

Normalize an integer.

For a field that takes an integer input. The field might be also left out, but it does have a default value.

tmt.utils.normalize_integer_list(key_address: str, value: Any, logger: Logger) → list[int]

Normalize an integer-or-list-of-integers input value.

foo: 11

foo:
  - 11
  - 79

Parameters:: value – input value from key source.

tmt.utils.normalize_optional_int(key_address: str, value: Any, logger: Logger) → int | None

Normalize an integer that may be unset as well.

For a field that takes an integer input, but might be also left out, and has no default value.

tmt.utils.normalize_path(key_address: str, value: Any, logger: Logger) → Path | None: Normalize content of the test path key

tmt.utils.normalize_path_list(key_address: str, value: None | str | list[str], logger: Logger) → list[Path]

Normalize a path-or-list-of-paths input value.

This is a fairly common input format present mostly in fmf nodes where tmt, to make things easier for humans, allows this:

foo: /foo/bar

foo:
  - /foo/bar
  - /baz

Internally, we should stick to one type only, and make sure whatever we get on the input, a list of strings would be the output.

Parameters:: value – input value from key source.

tmt.utils.normalize_pattern_list(key_address: str, value: Any, logger: Logger) → list[Pattern[str]]

Normalize a pattern-or-list-of-patterns input value.

foo: 'bar.*'

foo:
  - 'bar.*'
  - '(?i)BaZ+'

tmt.utils.normalize_shell_script(key_address: str, value: None | str, logger: Logger) → ShellScript | None

Normalize a single shell script input that may be unset.

Parameters:: value – input value from key source.

tmt.utils.normalize_shell_script_list(key_address: str, value: None | str | list[str], logger: Logger) → list[ShellScript]

Normalize a string-or-list-of-strings input value.

This is a fairly common input format present mostly in fmf nodes where tmt, to make things easier for humans, allows this:

foo: bar

foo:
  - bar
  - baz

Internally, we should stick to one type only, and make sure whatever we get on the input, a list of strings would be the output.

Parameters:: value – input value from key source.

tmt.utils.normalize_storage_size(key_address: str, value: Any, logger: Logger) → int

Normalize a storage size.

As of now, it’s just a simple integer with units interpreted by the owning plugin. In the future, we want this function to switch to proper units and return pint.Quantity instead.

tmt.utils.normalize_string_dict(key_address: str, raw_value: Any, logger: Logger) → dict[str, str]

Normalize a key/value dictionary.

The input value could be specified in two ways:

a dictionary, or
a list of KEY=VALUE strings.

For example, the following are acceptable inputs:

{'foo': 'bar', 'qux': 'quux'}

['foo=bar', 'qux=quux']

Parameters:: value – input value from key source.

tmt.utils.normalize_string_list(key_address: str, value: Any, logger: Logger) → list[str]

Normalize a string-or-list-of-strings input value.

This is a fairly common input format present mostly in fmf nodes where tmt, to make things easier for humans, allows this:

foo: bar

foo:
  - bar
  - baz

Internally, we should stick to one type only, and make sure whatever we get on the input, a list of strings would be the output.

Parameters:: value – input value from key source.

tmt.utils.option_to_key(option: str) → str: Convert an option name to corresponding key name

tmt.utils.preformat_jsonschema_validation_errors(raw_errors: list[ValidationError], prefix: str | None = None) → list[tuple[ValidationError, str]]

A helper to preformat JSON schema validation errors.

Raw errors can be converted to strings with a simple str() call, but resulting string is very JSON-ish. This helper provides simplified string representation consisting of error message and element path.

Parameters:

raw_error – raw validation errors as provided by jsonschema.
prefix – if specified, it is added at the beginning of each stringified error.

Returns:

a list of two-item tuples, the first item being the original validation error, the second item being its simplified string rendering.

tmt.utils.pure_ascii(text: Any) → bytes: Transliterate special unicode characters into pure ascii

tmt.utils.quote(string: str) → str: Surround a string with double quotes

tmt.utils.remove_color(text: str) → str: Remove ansi color sequences from the string

tmt.utils.render_command_report(*, label: str, command: ShellScript | Command | None = None, output: CommandOutput, exc: None = None) → Iterator[str]

tmt.utils.render_command_report(*, label: str, command: ShellScript | Command | None = None, output: None = None, exc: RunError) → Iterator[str]

Format a command output for a report file.

To provide unified look of various files reporting command outputs, this helper would combine its arguments and emit lines the caller may then write to a file. The following template is used:

## ${label}

# ${command}

# stdout (N lines)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
...
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

# stderr (N lines)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
...
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Parameters:

label – a string describing the intent of the command. It is useful for user who reads the report file eventually.
command – command that was executed.
output – if set, it contains output of the command. It has higher priority than exc.
exc – if set, it represents a failed command, and input stored in it is rendered.
comment_sign – a character to mark lines with comments that document the report.

tmt.utils.render_exception(exception: BaseException, traceback_verbosity: TracebackVerbosity = TracebackVerbosity.DEFAULT) → Iterator[str]: Render the exception and its causes for printing

tmt.utils.render_exception_stack(exception: BaseException, traceback_verbosity: TracebackVerbosity = TracebackVerbosity.DEFAULT) → Iterator[str]: Render traceback of the given exception

tmt.utils.render_run_exception(exception: RunError) → Iterator[str]: Render detailed output upon command execution errors for printing

tmt.utils.render_run_exception_streams(output: CommandOutput, verbose: int = 0, comment_sign: str = '#') → Iterator[str]: Render run exception output streams for printing

tmt.utils.resource_files(path: str | Path, package: str | ModuleType = 'tmt') → Path

Helper function to get path of package file or directory.

A thin wrapper for importlib.resources.files(): files() returns Traversable object, though in our use-case it should always produce a pathlib.PosixPath object. Converting it to tmt.utils.Path instance should be safe and stick to the “Path only!” rule in tmt’s code base.

Parameters:

path – file or directory path to retrieve, relative to the package root.
package – package in which to search for the file/directory.

Returns:

an absolute path to the requested file or directory.

tmt.utils.retry(func: Callable[[...], T], attempts: int, interval: int, label: str, logger: Logger, *args: Any, **kwargs: Any) → T

Retry functionality to be used elsewhere in the code.

Parameters:

func – function to be called with all unclaimed positional and keyword arguments.
attempts – number of tries to call the function
interval – amount of seconds to wait before a new try
label – action to retry

Returns:

propagates return value of func.

class tmt.utils.retry_session(retries: int = 3, backoff_factor: float = 0.1, allowed_methods: tuple[str, ...] | None = None, status_forcelist: tuple[int, ...] | None = None, timeout: int | None = None, logger: Logger | None = None)

Bases: AbstractContextManager

Context manager for requests.Session with retries and timeout

static create(retries: int = 3, backoff_factor: float = 0.1, allowed_methods: tuple[str, ...] | None = None, status_forcelist: tuple[int, ...] | None = None, timeout: int | None = None, logger: Logger | None = None) → Session

tmt.utils.sanitize_name(name: str, allow_slash: bool = True) → str

Create a safe variant of a name that does not contain special characters.

Spaces and other special characters are removed to prevent problems with tools which do not expect them (e.g. in directory names).

Parameters:

name – a name to sanitize.
allow_slash – if set, even a slash character, /, would be replaced.

tmt.utils.shell_variables(data: list[str] | tuple[str, ...] | dict[str, Any]) → list[str]

Prepare variables to be consumed by shell

Convert dictionary or list/tuple of key=value pairs to list of key=value pairs where value is quoted with shlex.quote().

tmt.utils.show_exception(exception: BaseException, include_logfiles: bool = True) → None

Display the exception and its causes.

Parameters:

exception – exception to log.
include_logfiles – if set, exception will be logged into known logfiles as well as to standard error output.

tmt.utils.uniq(values: list[T]) → list[T]: Return a list of all unique items from values

tmt.utils.validate_fmf_node(node: Tree, schema_name: str, logger: Logger) → list[tuple[ValidationError, str]]: Validate a given fmf node

tmt.utils.verdict(decision: bool, comment: str | None = None, good: str = 'pass', bad: str = 'fail', problem: str = 'warn', **kwargs: Any) → bool

tmt.utils.verdict(decision: None, comment: str | None = None, good: str = 'pass', bad: str = 'fail', problem: str = 'warn', **kwargs: Any) → None

Print verdict in green, red or yellow based on the decision

The supported decision values are:

True …. good (green) False … bad (red) None …. problem (yellow)

Anything else raises an exception. Additional arguments are passed to the echo function. Returns back the decision.

tmt.utils.wait(parent: Common, check: Callable[[], T], timeout: timedelta, tick: float = 30.0, tick_increase: float = 1.0) → T

Wait for a condition to become true.

To test the condition state, a check callback is called every tick seconds until check reports a success. The callback may:

decide the condition has been fulfilled. This is a successful outcome, check shall then simply return, and waiting ends. Or,
decide more time is needed. This is not a successful outcome, check shall then raise WaitingIncomplete exception, and wait() will try again later.

Parameters:

parent – “owner” of the wait process. Used for its logging capability.
check – a callable responsible for testing the condition. Accepts no arguments. To indicate more time and attempts are needed, the callable shall raise WaitingIncomplete, otherwise it shall return without exception. Its return value will be propagated by wait() up to wait()’s. All other exceptions raised by check will propagate to wait()’s caller as well, terminating the wait.
timeout – amount of time wait() is allowed to spend waiting for successful outcome of check call.
tick – how many seconds to wait between two consecutive calls of check.
tick_increase – a multiplier applied to tick after every attempt.

Returns:

value returned by check reporting success.

Raises:

GeneralError – when tick is not a positive integer.
WaitingTimedOutError – when time quota has been consumed.

tmt.utils.yaml_to_dict(data: Any, yaml_type: Literal['rt', 'safe', 'unsafe', 'base'] | None = None) → dict[Any, Any]: Convert yaml into dictionary

tmt.utils.yaml_to_list(data: Any, yaml_type: Literal['rt', 'safe', 'unsafe', 'base'] | None = 'safe') → list[Any]: Convert yaml into list