tmt.utils package

Submodules

tmt.utils.filesystem module

Utility functions for filesystem operations.

tmt.utils.filesystem.copy_tree(src: Path, dst: Path, logger: Logger) → None

Copy directory efficiently, trying different strategies.

Attempts strategies in order:

cp -a --reflink=auto (copy-on-write, with cp’s own fallback).
- Reflinks provide fast, space-efficient copies that behave like normal copies
- They don’t use additional storage space unless the file is modified
- Supported on btrfs (Fedora default since F33) and XFS (CentOS Stream 8+)
- Using --reflink=auto means cp automatically falls back to standard copy if reflink isn’t supported by the filesystem
shutil.copytree() as a final fallback.
- Used if the cp command fails for any reason
- Maintains symlinks (symlinks=True)
- Merges with existing destination directories (dirs_exist_ok=True)

Symlinks are always preserved. The destination directory dst and its parents will be created if they do not exist. File permissions and timestamps are preserved in all copy strategies.

Example usage:

# Copy a directory tree with all its content
copy_tree(Path("/path/to/source"), Path("/path/to/destination"), logger)

# Copy with relative paths
copy_tree(workdir / "original", workdir / "backup", logger)

Parameters:

src – Source directory path. Must exist and be a directory.
dst – Destination directory path.
logger – Logger to use for debug messages.

Raises:

GeneralError – when copying fails using all strategies, or if src does not exist or is not a directory.

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.LocalDistGit

Bases: DistGitHandler

Local source files

its_me(remotes: list[str]) → bool: Has to be an explicit request for this type

lookaside_server: str = 'file://'

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

uri = '{cwd}/{filename}'

usage_name: str = 'local'

class tmt.utils.git.RedHatDistGit

Bases: DistGitHandler

Red Hat Handler

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/|pkgs\\.devel\\.redhat\\.com')

usage_name: str = 'redhat'

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.hints module

Hints for users when facing installation-related issues.

Plugins, steps, and tmt code in general can register hints to be shown to user when an important (or optional, but interesting) package is not available.

Hints are shown when importing plugins fails, and rendered as part of both their CLI help and HTML documentation.

Hints are registered, each having its own ID. A hint may take advantage of ReST to format its content for better readability.

Hints should follow the rules of docstrings, i.e. provide a short summary on the first line, followed by more details in the rest of the text.

class tmt.utils.hints.Hint(hint_id: str, text: str, cli_output_patterns: list[str] | None = None)

Bases: object

Initialize hint id, text and search patterns

cli_output_patterns: list[Pattern[str]]: Regular expression patterns for searching the command line output

id: str: Unique hint id

print(logger: Logger) → None: Print hint to the user

property ref: str

render(logger: Logger) → str

render_summary(logger: Logger) → str

render_summary_ref(logger: Logger) → str

search_cli_patterns(*outputs: str | None) → bool: Check provided command line outputs for known error patterns

property summary: str

property summary_ref: str

text: str: Detailed text with hint information

tmt.utils.hints.get_hint(hint_id: str, ignore_missing: Literal[True]) → Hint | None

tmt.utils.hints.get_hint(hint_id: str, ignore_missing: Literal[False]) → Hint

Return hint for the provided identifier

Parameters:

hint_id – Hint identifier.
ignore_missing – If not set, non-existent hint will raise an exception. Otherwise, non-existent hint will be skipped.

Returns:

Hint if found, None otherwise.

tmt.utils.hints.get_hints(*ids: str, ignore_missing: bool = False) → list[Hint]

Find given hints.

Parameters:

ids – ids of hints to retrieve.
ignore_missing – if not set, non-existent hints will raise an exception. Otherwise, non-existent hints will be skipped.

Returns:

found hints.

tmt.utils.hints.hints_as_notes(*ids: str) → list[str]

Format hints as a list of Result notes.

Hint.summary_ref of each hint is added as a distinct note.

tmt.utils.hints.print_hints(*ids: str, ignore_missing: bool = False, logger: Logger) → None

Display given hints by printing them as info-level messages.

Parameters:

ids – ids of hints to render.
ignore_missing – if not set, non-existent hints will raise an exception.
logger – to use for logging.

tmt.utils.hints.register_hint(hint_id: str, hint: str) → None

Register a hint for users.

Parameters:

hint_id – step name for step-specific hints, <step name>/<plugin name> for plugin-specific hints, or an arbitrary string.
hint – a hint to register.

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(link: Link, 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.

class tmt.utils.rest.ANSIString: Bases: str

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_definition(node: definition) → None

depart_definition_list(node: Node) → None

depart_definition_list_item(node: Node) → None

depart_document(node: document) → None

depart_emphasis(node: Node) → 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_section(node: Node) → None

depart_strong(node: Node) → None

depart_target(node: Node) → None

depart_term(node: Node) → None

depart_title(node: Node) → None

depart_transition(node: Node) → None

depart_warning(node: warning) → None

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

log_departure(node: Node | Body) → None

log_visit(node: Node | Body) → None

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_definition(node: definition) → None

visit_definition_list(node: Node) → None

visit_definition_list_item(node: Node) → None

visit_document(node: Node) → None

visit_emphasis(node: emphasis) → 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_section(node: Node) → None

visit_strong(node: strong) → None

visit_target(node: Node) → None

visit_term(node: term) → None

visit_title(node: title) → None

visit_transition(node: transition) → None

visit_warning(node: warning) → None

class tmt.utils.rest.TextWrapper(width=70, initial_indent='', subsequent_indent='', expand_tabs=True, replace_whitespace=True, fix_sentence_endings=False, break_long_words=True, drop_whitespace=True, break_on_hyphens=True, tabsize=8, *, max_lines=None, placeholder=' [...]'): Bases: TextWrapper

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.rest.role_ref(name: str, rawtext: str, text: str, lineno: int, inliner: Inliner, options: Mapping[str, Any] | None = None, content: Sequence[str] | None = None) → tuple[Sequence[reference], Sequence[reference]]

A handler for :ref: role.

Returns:: a simple docutils.nodes.Text node with text of the “link”: foo for both :ref:`foo` and :ref:`foo</bar>`.

tmt.utils.signals module

Signal handling in tmt.

Provides custom signal handler for SIGINT (aka Ctrl+C) and SIGTERM, and delayed actions for these signals.

Some of the factors tmt need to take into account when dealing with signals:

tmt runs a lot of things in dedicated threads,
sometimes tmt cannot let signals interrupt whatever is being done at the moment, namely saving API responses from provisioning services,
signals in Python are always executed in the context of the main thread,
signals are most often delivered to the main thread because that is what Ctrl+C will deliver to,
masking signals in a thread will not affect other threads.

To support uninterruptible blocks, signal handler and PreventSignals use events to record whether a signal was delivered and needs handling:

if signals are not blocked, signal handler would raise the KeyboardInterrupt exception right away,
if signals are blocked, signal handler would record its delivery, and PreventSignals would raise the KeyboardInterrupt when leaving the uninterruptible block.

tmt.utils.signals.INTERRUPT_PENDING = <threading.Event at 0x7b30cee7c2d0: unset>: When set, interrupt was delivered to tmt, and tmt should react to it.

exception tmt.utils.signals.Interrupted(*args: Any, **kwargs: Any)

Bases: GeneralError

Raised by code that interrupted its work because of tmt shutdown.

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.signals.PreventSignals(logger: Logger)

Bases: AbstractContextManager[PreventSignals]

For the duration of this context manager, interrupt signals are postponed.

If, while the context was active, signals were delivered, KeyboardInterrupt exception would be raised when leaving the context.

logger: Logger

tmt.utils.signals.add_callback(fn: ~typing.Callable[[~P], None], *args: ~typing.~P, **kwargs: ~typing.~P) → int

Register a callback to be executed when tmt is interrupted.

Parameters:

fn – callable to execute:
args – position arguments to pass to fn.
kwargs – keyword arguments to pass to fn.

Returns:

unique token to use when unregistering the callback via remove_callback().

tmt.utils.signals.install_handlers() → None: Install tmt’s signal handlers

tmt.utils.signals.remove_callback(token: int) → None

Unregister a on-interrupt callback.

Parameters:: token – identifies the callback registered by add_callback().

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.

tmt.utils.templates.render_template_file_into_file(input_filepath: Path, output_filepath: Path, environment: Environment | None = None, **variables: Any) → None

Render a template from a file, and write the result into a file.

Combines render_template_file() and Path.write_text(), but makes sure the file ends with an empty line.

Parameters:

input_filepath – path to the template file.
output_filepath – path to the file in which the rendering should be saved.
environment – Jinja2 environment to use.
variables – variables to pass to the template.

tmt.utils.themes module

tmt.utils.themes.Style = None | str | ForwardRef('ThemeStyle'): A style to apply to a string.

Note

Eventually, this would be ThemeStyle. For now, we need to allow the Colorama color specification, strings.

Apply a style to a string.

style is the most preferred argument, and, if set, no other arguments would be used. If style is not given, remaining keyword arguments are passed directly to click.style().

Parameters:

s – string to colorize.
style – style to apply. If set, it will be preferred over any other arguments.
fg – foreground color.
bold – whether the text should be using bold font.
underline – whether the text should be underlined.

Returns:

colorized string.

tmt.utils.url module

Url handling helpers.

tmt.utils.url.download(url: str, destination: Path, *, logger: Logger) → Path

tmt.utils.wait module

class tmt.utils.wait.Deadline(timeout: timedelta)

Bases: object

A point in time when something should end.

Instead of raw timeouts that represent remaining time, we deal more with points when things should stop. Timeouts must be updated regularly, decreased as time progresses, while deadlines are set and can be easily tested.

classmethod from_delta(timeout: timedelta) → Deadline: Create a deadline from a delta.

classmethod from_seconds(timeout: float) → Deadline: Create a deadline from the number of seconds of a timeout.

property is_due: bool: True when the deadline has been reached, False otherwise.

original_timeout: timedelta: The original timeout that populated this deadline.

property time_left: timedelta: The remaining time left.

Note

The value will be negative when the deadline has been reached already.

property time_over: timedelta: The time past the deadline.

Note

The value will be negative when the deadline has not been reached yet.

class tmt.utils.wait.Waiting(deadline: Deadline, tick: float = 30.0, tick_increase: float = 1.0)

Bases: object

Context describing how to wait for a condition with limited deadline.

deadline: Deadline: The deadline that limits the waiting.

tick: float = 30.0: How many seconds to wait between two consecutive checks whether the condition is satisfied.

tick_increase: float = 1.0: A multiplier applied to tick after every attempt.

wait(check: Callable[[], T], logger: Logger) → 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.

wait() will also stop and quit if tmt has been interrupted.

Parameters:

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 returned by wait() itself. All other exceptions raised by check will be propagated upstream, terminating the wait.

Returns:

value returned by check reporting success.

Raises:

WaitingTimedOutError – when time quota has been consumed.
Interrupted – when tmt has been interrupted.

exception tmt.utils.wait.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.wait.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__.

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.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, on_process_end: Callable[[Command, Popen[bytes], CommandOutput, 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.
on_process_end – if set, this callable would be called after the command process finishes.
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

property fmf_context: FmfContext: An fmf context set for this object.

ignore_class_options: bool = False

property import_before_name_filter: bool: Whether to import all external plans before applying name-based filters

info(key: str, value: str | dict[str, Any] | int | bool | float | tmt.utils.Environment | tmt.utils.FmfContext | tmt.utils.Path | tmt.utils.Command | tmt.utils.ShellScript | None = None, color: tmt.utils.themes.Style = None, shift: int = 0, topic: Topic | None = None) → 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: tmt.utils.themes.Style = None) → 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, on_process_end: Callable[[Command, Popen[bytes], CommandOutput, 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

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.

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__.

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.

A=B
C=D

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.

A=B
C=D

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 or DOTENV 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
@foo.env

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

to_shell() → list[str]

Convert to a form accepted by shell commands.

>>> Environment({'FOO': EnvVarValue('bar'), 'BAZ': EnvVarValue('qu ux')}).to_shell()
['FOO=bar', "BAZ='qu ux'"]

to_shell_exports() → list[ShellScript]

Convert to a sequence of export shell commands.

>>> Environment({'FOO': EnvVarValue('bar'), 'BAZ': EnvVarValue('qu ux')}).to_shell_exports()
[ShellScript("export FOO=bar"), ShellScript("export BAZ='qu ux'")]

update([E, ]**F) → None. Update D from mapping/iterable E and F.: If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

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.FieldValueSource(*values)

Bases: Enum

Indicates source of metadata field value.

CLI = 'cli': The value was provided by CLI option.

DEFAULT = 'default': The value is the default value defined for the field.

FMF = 'fmf': The value was provided by fmf node key.

POLICY = 'policy'

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__.

class tmt.utils.HasEnvironment

Bases: ABC

A class that provides environment attribute.

abstract property environment: Environment: Environment variables this object wants to expose to user commands.

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

Bases: ABC

Provides assured access to a guest workdir.

abstract property guest_workdir: Path

Path to a guest workdir.

Raises:: GeneralError – when there is no guest workdir yet.

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

Bases: ABC

Provides assured access to a phase workdir.

abstract property phase_workdir: Path

Path to a phase workdir.

Raises:: GeneralError – when there is no phase workdir yet.

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

Bases: ABC

Provides assured access to a plan workdir.

abstract property plan_workdir: Path

Path to a plan workdir.

Raises:: GeneralError – when there is no plan workdir yet.

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

Bases: ABC

Provides assured access to a run workdir.

abstract property run_workdir: Path

Path to a run workdir.

Raises:: GeneralError – when there is no current run, or the run does not have a workdir yet.

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

Bases: ABC

Provides assured access to a step workdir.

abstract property step_workdir: Path

Path to a step workdir.

Raises:: GeneralError – when there is no step workdir yet.

class tmt.utils.ListFormat(*values)

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.

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.OnProcessEndCallback

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

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

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(*values)

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 = str | tmt._compat.pathlib.Path: A single element of raw command line in its list form.

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__.

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

Bases: ExecuteError

Failed to reconnect to the guest 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__.

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.RestartMaxAttemptsError(message: str, causes: list[Exception] | None = None, *args: Any, **kwargs: Any)

Bases: ExecuteError

Test restart failed due to maximum attempts reached.

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(*, logger: Logger, **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.

is_retry(method: str, status_code: int, has_retry_after: bool = False) → bool: Is this method/status code retryable? (Based on allowlists and control variables such as the number of total retries to allow, whether to respect the Retry-After header, whether this header is present, and whether the returned status code is on the list of status codes to be retried upon on the presence of the aforementioned header)

logger: Logger

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.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.

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

property end_time_formatted: str

classmethod measure(fn: ~typing.Callable[[~P], ~tmt.utils.T], *args: ~typing.~P, **kwargs: ~typing.~P) → tuple[T | None, Exception | None, Stopwatch]

Run a function while the stopwatch is running.

Parameters:: fn – a function to call. It will be invoked with all positional and keyword arguments given to measure().
Returns:: a tuple of three items: the return value of fn, None, and the stopwatch instance on success, or None, the raised exception, and the stopwatch instance when fn raised an exception.

start_time: datetime

property start_time_formatted: str

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 {}.

context is the contextvars.Context value to use for the thread. The default value is None, which means to check sys.flags.thread_inherit_context. If that flag is true, use a copy of the context of the caller. If false, use an empty context. To explicitly start with an empty context, pass a new instance of contextvars.Context(). To explicitly start with a copy of the current context, pass the value from contextvars.copy_context().

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__.

tmt.utils.TRACEBACK_LOCALS_TRIM = 1024: How many leading characters to display in tracebacks with TMT_SHOW_TRACEBACK=2.

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(*values)

Bases: Enum

Levels of logged traveback verbosity

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

FULL = 'full': Render everything that can be shown: all causes, their call stacks, all frames and all locals in their completeness.

LOCALS = '2': Render also call stack for exception and each of its causes, plus all local variables in each frame, trimmed to first 1024 characters of their values.

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

classmethod from_env() → TracebackVerbosity

classmethod from_spec(spec: str) → TracebackVerbosity

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”.