tmt.steps package

Subpackages

Submodules

tmt.steps.scripts module

tmt.steps.scripts.DEFAULT_SCRIPTS_DEST_DIR = Path('/usr/local/bin')

The default scripts destination directory

tmt.steps.scripts.DEFAULT_SCRIPTS_DEST_DIR_OSTREE = Path('/var/lib/tmt/scripts')

The default scripts destination directory for rpm-ostree based distributions, https://github.com/teemtee/tmt/discussions/3260

tmt.steps.scripts.SCRIPTS_DEST_DIR_VARIABLE = 'TMT_SCRIPTS_DIR'

The tmt environment variable name for forcing SCRIPTS_DEST_DIR

tmt.steps.scripts.SCRIPTS_DIR_NAME = 'scripts'

The directory where helper scripts will be copied

tmt.steps.scripts.SCRIPTS_RESOURCE = 'steps/scripts'

Scripts source resource. Use tmt.utils.resource_files to access them.

class tmt.steps.scripts.Script(source_filename: str, destination_path: Path | None, aliases: list[str], related_variables: list[str], enabled: Callable[[Guest], bool])

Bases: object

Represents a script provided by the internal executor.

Must be used as a context manager. The context manager returns the source filename.

The source file is defined by the source_filename attribute and its location is relative to the directory specified via the SCRIPTS_SRC_DIR variable. All scripts must be located in this directory.

The default destination directory of the scripts is DEFAULT_SCRIPTS_DEST_DIR. On rpm-ostree distributions like Fedora CoreOS, the default destination directory is :py:data:DEFAULT_SCRIPTS_DEST_DIR_OSTREE. The destination directory of the scripts can be forced by the script using destination_path attribute.

The destination directory can be overridden using the environment variable defined by the DEFAULT_SCRIPTS_DEST_DIR_VARIABLE variable.

The enabled attribute can specify a function which is called with Guest instance to evaluate if the script is enabled. This can be useful to optionally disable a script for specific guests.

aliases: list[str]
destination_path: Path | None
enabled: Callable[[Guest], bool]
related_variables: list[str]
source_filename: str
class tmt.steps.scripts.ScriptCreatingFile(source_filename: str, destination_path: Path | None, aliases: list[str], related_variables: list[str], enabled: Callable[[Guest], bool], created_file: str)

Bases: Script

Represents a script which creates a file.

See Script for more details.

created_file: str
class tmt.steps.scripts.ScriptTemplate(source_filename: str, destination_path: Path | None, aliases: list[str], related_variables: list[str], enabled: Callable[[Guest], bool], context: dict[str, str], _rendered_script_path: Path | None = None)

Bases: Script

Represents a Jinja2 templated script.

The source filename is constructed from the name of the file specified via the source_filename attribute, with the .j2 suffix appended. The template file must be located in the directory specified via SCRIPTS_SRC_DIR variable.

context: dict[str, str]
tmt.steps.scripts.effective_scripts_dest_dir(default: Path = Path('/usr/local/bin')) → Path

Find out what the actual scripts destination directory is.

If the TMT_SCRIPTS_DIR environment variable is set, it is used as the scripts destination directory. Otherwise, the default parameter path is returned.

Module contents

Step Classes

class tmt.steps.Action(*, order: int = 50, **kwargs: Any)

Bases: Phase, MultiInvokableCommon

A special action performed during a normal step.

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: 'tmt.cli.CliInvocation' | None = None
cli_invocations: list['tmt.cli.CliInvocation'] = []
abstractmethod go() → None
classmethod phases(step: Step) → list[int]

Return list of phases enabled for given step

class tmt.steps.ActionTask(phase: Action, logger: Logger)

Bases: GuestlessTask[None]

A task to run an action

property name: str

A name of this task.

Left for child classes to implement, because the name depends on the actual task.

phase: Action
run(logger: Logger) → None

Perform the task.

Called once from go(). Subclasses of must implement their logic in this method rather than in go() which is already provided.

class tmt.steps.BasePlugin(*, step: Step, data: StepDataT, workdir: Literal[True] | Path | None = None, logger: Logger)

Bases: HasRunWorkdir, HasStepWorkdir, HasPlanWorkdir, HasPhaseWorkdir, Phase, Generic[StepDataT, PluginReturnValueT]

Common parent of all step plugins

Store plugin name, data and parent step

classmethod allowed_methods(step: Step) → list[Method]

Return all allowed methods

abstractmethod classmethod base_command(usage: str, method_class: type[Command] | None = None) → Command

Create base click command (common for all step plugins)

cli_invocation: 'tmt.cli.CliInvocation' | None = None
classmethod command() → Command

Prepare click command for all supported methods

data: StepDataT
default(option: str, default: Any | None = None) → Any

Return default data for given option

classmethod delegate(step: Step, data: StepDataT | None = None, raw_data: _RawStepData | None = None) → BasePlugin[StepDataT, PluginReturnValueT]

Return plugin instance implementing the data[‘how’] method

Supports searching by method prefix as well (e.g. ‘virtual’). The first matching method with the lowest ‘order’ wins.

property enabled_by_when: bool

Check if the plugin is enabled by ‘when’ keyword

enabled_on_guest(guest: Guest) → bool

Check if the plugin is enabled on the specific guest

essential_requires() → list[DependencySimple | DependencyFmfId | DependencyFile]

Collect all essential requirements of the plugin.

Essential requirements of a plugin are necessary for the plugin to perform its basic functionality.

Returns:

a list of requirements.

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

Get option from plugin data, user/system config or defaults

classmethod get_data_class() → type[StepDataT]

Return step data class for this plugin.

By default, _data_class is returned, but plugin may override this method to provide different class.

classmethod get_step_name() → str
go_prolog(logger: Logger) → None

Perform actions shared among plugins when beginning their tasks

how: str = 'shell'
classmethod methods() → list[Method]

Return all supported methods ordered by priority

classmethod options(how: str | None = None) → list[Callable[[Any], Any]]

Prepare command line options for given method

property phase_workdir: Path

Path to a phase workdir.

Raises:

GeneralError – when there is no phase workdir yet.

property plan_workdir: Path

Path to a plan workdir.

Raises:

GeneralError – when there is no plan workdir yet.

prune(logger: Logger) → None

Prune uninteresting files from the plugin workdir

By default we remove the whole workdir. Individual plugins can override this method to keep files and directories which are useful for inspection when the run is finished.

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.

property safe_name: str

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

Override parent implementation as we do not allow phase names to contain slash characters, /.

show(keys: list[str] | None = None) → None

Show plugin details for given or all available keys

property step_workdir: Path

Path to a step workdir.

Raises:

GeneralError – when there is no step workdir yet.

wake() → None

Wake up the plugin, process data, apply options

Check command line options corresponding to plugin keys and store their value into the ‘self.data’ dictionary if their value is True or non-empty.

By default, all supported options corresponding to common and plugin-specific keys are processed. List of key names in the ‘keys’ parameter can be used to override only selected ones.

tmt.steps.DEFAULT_SOFT_REBOOT_COMMAND = <tmt.utils.ShellScript object>

A default command to trigger a guest reboot when executed remotely.

class tmt.steps.DefaultNameGenerator(known_names: list[str])

Bases: object

Generator of names for that do not have any.

If user did not set any name for one or more phases, tmt will assign them a “dummy” name default-N. This class wraps the generator.

Generator of names for that do not have any.

Parameters:

known_names – already existing names the generator needs to avoid.

classmethod from_raw_phases(raw_data: Iterable[_RawStepData]) → DefaultNameGenerator

Create a generator based on available phase specifications.

A special constructor that extracts known_names from raw_data.

Parameters:

raw_data – phase specifications as collected from fmf nodes and CLI options.

get() → str
restart() → None

Reset the generator and start from the beginning again

class tmt.steps.GuestTopology(guest: Guest)

Bases: SerializableContainer

Describes a guest in the topology of provisioned tmt guests

hostname: str | None
name: str
role: str | None
class tmt.steps.GuestlessPlugin(*, step: Step, data: StepDataT, workdir: Literal[True] | Path | None = None, logger: Logger)

Bases: BasePlugin[StepDataT, PluginReturnValueT]

Common parent of all step plugins that do not work against a particular guest

Store plugin name, data and parent step

cli_invocation: 'tmt.cli.CliInvocation' | None = None
abstractmethod go(*, logger: Logger | None = None) → PluginReturnValueT

Perform actions shared among plugins when beginning their tasks

class tmt.steps.Login(*, step: Step, order: int, logger: Logger)

Bases: Action

Log into the guest

Initialize relations, store the login order

after_test(results: list[Result], cwd: Path | None = None, env: Environment | None = None) → None

Check and login after test execution

cli_invocation: 'tmt.cli.CliInvocation' | None = None
cli_invocations: list['tmt.cli.CliInvocation'] = []
classmethod command(method_class: Method | None = None, usage: str | None = None) → Command

Create the login command

go(force: bool = False) → None

Login to the guest(s)

classmethod plugins(step: Step) → list[Login]

Return list of login instances for given step

class tmt.steps.Method(name: str, class_: type[BasePlugin], doc: str | None = None, order: int = 50, installation_hint: str | None = None)

Bases: object

Step implementation method

Store method data

describe() → str

Format name and summary for a nice method overview

usage() → str

Prepare a detailed usage from summary and description

tmt.steps.PHASE_ORDER_PREPARE_INSTALL_ESSENTIAL_REQUIRES = 30

Installation of essential plugin and check requirements.

tmt.steps.PHASE_ORDER_PREPARE_INSTALL_RECOMMENDS = 75

Installation of packages recommended by tests.

tmt.steps.PHASE_ORDER_PREPARE_INSTALL_REQUIRES = 70

Installation of packages required by tests.

class tmt.steps.Phase(*, order: int = 50, **kwargs: Any)

Bases: Common

A phase of a step

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.

assert_feeling_safe(deprecated_in_version: str, subject: str) → None

Raises a tmt.utils.ProvisionError if feeling-safe is required, but not set. Warns when feeling-safe will be required in a future version. :param deprecated_in_version: Version from which feeling-safe is required, e.g. ‘1.38’. :param subject: Subject requiring feeling-safe, e.g. ‘Local provision plugin’.

cli_invocation: 'tmt.cli.CliInvocation' | None = None
property enabled_by_when: bool
enabled_on_guest(guest: Guest) → bool

Phases are enabled across all guests by default

property is_in_standalone_mode: bool

True if the phase is in stand-alone mode.

Stand-alone mode means that only this phase should be run as a part of the run (and not any other even if requested using –all). This is useful as some plugin options may completely change its behaviour from the regular behaviour based on options (e.g. listing images inside a provision plugin).

class tmt.steps.PhaseQueue(name: str, logger: Logger)

Bases: Queue[ActionTask | PluginTask[StepDataT, PluginReturnValueT]]

Queue class for running phases on guests

enqueue_action(*, phase: Action) → None
enqueue_plugin(*, phase: Plugin[StepDataT, PluginReturnValueT], guests: list[Guest]) → None
class tmt.steps.Plugin(*, step: Step, data: StepDataT, workdir: Literal[True] | Path | None = None, logger: Logger)

Bases: BasePlugin[StepDataT, PluginReturnValueT]

Common parent of all step plugins that do work against a particular guest

Store plugin name, data and parent step

cli_invocation: 'tmt.cli.CliInvocation' | None = None
abstractmethod go(*, guest: Guest, environment: Environment | None = None, logger: Logger) → PluginReturnValueT

Perform actions shared among plugins when beginning their tasks

write_command_report(*, path: Path, label: str, timer: Stopwatch, command: Command | ShellScript | None = None, output: CommandOutput, exc: None = None) → None
write_command_report(*, path: Path, label: str, timer: Stopwatch, command: Command | ShellScript | None = None, output: None = None, exc: Exception) → None

Create a report for a given command outcome.

Combines tmt.utils.render_report() and tmt.utils.render_command_report() to create a report for a given command, its output, and possibly its failure. The report is then appended to the given file.

Note

This method is focusing on reporting of command outcomes. It provides standardized header, fields, and formatting. For reports consisting of other topics see tmt.utils.render_report() and write_report().

Note

This method is designed for reporting an action of a single command. If you need to report an action consisting of multiple commands, build the body of the report by calling tmt.utils.render_command_report() for individual commands, and use write_report() to finish the report.

body: list[str] = []

for command in commands:
    body.extend(render_command_report(...))

self.write_report(..., body=body)
Parameters:
write_report(*, path: Path, label: str, timer: Stopwatch, body: Iterable[str] | None = None) → None

Format and write an arbitrary body of text into a report file.

Uses tmt.utils.render_report() to create a report with a given content. The report is then appended to the given file.

Note

This method is focusing on reporting of arbitrary actions. It provides standardized header, fields, and formatting. For reports of command outcomes see tmt.utils.render_command_report() and write_command_report().

Parameters:
class tmt.steps.PluginOutcome(results: list[PhaseResult] = <factory>, exceptions: list[Exception] = <factory>)

Bases: object

Bundles together results produced by a plugin, and possible error.

exceptions: list[Exception]

Exceptions the plugin would like to raise and report.

results: list[PhaseResult]

Results produced by the plugin invocation.

class tmt.steps.PluginReturnValueT

A type variable representing a return value of plugin’s go() method.

alias of TypeVar(‘PluginReturnValueT’)

class tmt.steps.PluginTask(phase: Plugin[StepDataT, PluginReturnValueT], guests: list[Guest], logger: Logger)

Bases: MultiGuestTask[PluginReturnValueT], Generic[StepDataT, PluginReturnValueT]

A task to run a phase on a given set of guests

property name: str

A name of this task.

Left for child classes to implement, because the name depends on the actual task.

phase: Plugin[StepDataT, PluginReturnValueT]
property phase_name: str
run_on_guest(guest: Guest, logger: Logger) → PluginReturnValueT

Perform the task.

Called once from go(). Subclasses of must implement their logic in this method rather than in go() which is already provided.

class tmt.steps.PullTask(guests: list[Guest], source: Path | None, logger: Logger)

Bases: MultiGuestTask[None]

Task performing a workdir pull from a guest

property name: str

A name of this task.

Left for child classes to implement, because the name depends on the actual task.

run_on_guest(guest: Guest, logger: Logger) → None

Perform the task.

Called once from go(). Subclasses of must implement their logic in this method rather than in go() which is already provided.

source: Path | None
class tmt.steps.PushTask(guests: list[Guest], logger: Logger)

Bases: MultiGuestTask[None]

Task performing a workdir push to a guest

property name: str

A name of this task.

Left for child classes to implement, because the name depends on the actual task.

run_on_guest(guest: Guest, logger: Logger) → None

Perform the task.

Called once from go(). Subclasses of must implement their logic in this method rather than in go() which is already provided.

class tmt.steps.RawWhereableStepData

Bases: TypedDict

where: str | list[str]
class tmt.steps.Reboot(*, step: Step, order: int, logger: Logger)

Bases: Action

Reboot guest

Initialize relations, store the reboot order

cli_invocation: 'tmt.cli.CliInvocation' | None = None
cli_invocations: list['tmt.cli.CliInvocation'] = []
classmethod command(method_class: Method | None = None, usage: str | None = None) → Command

Create the reboot command

go() → None

Reboot the guest(s)

classmethod plugins(step: Step) → list[Reboot]

Return list of reboot instances for given step

class tmt.steps.ResultT

A type variable representing a result type.

alias of TypeVar(‘ResultT’, bound=BaseResult)

class tmt.steps.Step(*, plan: Plan, data: _RawStepData | list[_RawStepData] | None = None, name: str | None = None, workdir: Literal[True] | Path | None = None, logger: Logger)

Bases: HasRunWorkdir, HasStepWorkdir, HasPlanWorkdir, MultiInvokableCommon, Exportable[Step]

Common parent of all test steps

Initialize and check the step data

DEFAULT_HOW: str = 'shell'
actions() → None

Run all loaded Login or Reboot action instances of the step

add_phase(phase: Phase) → None

Add a phase dynamically to the current step.

Parameters:

phase – The phase to add.

property allowed_methods_pattern: Pattern[str]

Return a pattern allowed methods must match

cli_invocation: 'tmt.cli.CliInvocation' | None = None
cli_invocations: list['tmt.cli.CliInvocation'] = []
property data: list[StepData]
property enabled: bool | None

True if the step is enabled

go(force: bool = False) → None

Execute the test step

load() → None

Load status and step data from the workdir

phases(classes: None = None) → list[Phase]
phases(classes: type[PhaseT]) → list[PhaseT]
phases(classes: tuple[type[PhaseT], ...]) → list[PhaseT]

Iterate over phases by their order

By default iterates over all available phases. Optional filter ‘classes’ can be used to iterate only over instances of given class (single class or tuple of classes).

property plan_workdir: Path

Path to a plan workdir.

Raises:

GeneralError – when there is no plan workdir yet.

property plugins_in_standalone_mode: int

The number of plugins in standalone mode.

Stand-alone mode means that only this step should be run as a part of the run (and not any other even if requested using –all). This is useful as some step options may completely change its behaviour from the regular behaviour based on options (e.g. listing images inside provision).

prune(logger: Logger) → None

Remove all uninteresting files from the step workdir

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.

save() → None

Save status and step data to the workdir

setup_actions() → None

Insert login and reboot plugins if requested

show() → None

Show step details

status(status: str | None = None) → str | None

Get and set current step status

The meaning of the status is as follows: todo … config, data and command line processed (we know what to do) done … the final result of the step stored to workdir (we are done)

property step_name: str
property step_workdir: Path

Path to a step workdir.

Raises:

GeneralError – when there is no step workdir yet.

abstractmethod summary() → None

Give a concise summary about the step result

To be implemented by each step.

suspend() → None

Suspend the step.

Perform any actions necessary before quitting the step and tmt. The step may be revisited by future tmt invocations.

classmethod usage(method_overview: str) → str

Prepare general usage message for the step

wake() → None

Wake up the step (process workdir and command line)

class tmt.steps.StepData(name: str, how: str, order: int = 50, when: list[str] = <factory>, summary: str | None = None)

Bases: SpecBasedContainer[_RawStepData, _RawStepData], NormalizeKeysMixin, SerializableContainer

Keys necessary to describe, create, save and restore a step.

Very basic set of keys shared across all steps.

Provides basic functionality for transition between “raw” step data, which consists of fmf nodes and CLI options, and this container representation with keys and types more suitable for internal use.

Implementation expects simple 1:1 relation between StepData attributes - keys - and their fmf/CLI sources, where keys replace options’ dashes (-) with underscores (_). For example, to hold value of an fmf key foo-bar - or value of a corresponding CLI option, --foo-bar, a step data class should declare key named foo_data. All StepData methods would honor this mapping.

classmethod from_spec(raw_data: _RawStepData, logger: Logger) → Self

Convert from a specification file or from a CLI option

how: str
name: str
order: int = 50
post_normalization(raw_data: _RawStepData, logger: Logger) → None

Called after normalization, useful for tweaking normalized data

classmethod pre_normalization(raw_data: _RawStepData, logger: Logger) → None

Called before normalization, useful for tweaking raw data

summary: str | None = None
to_spec() → _RawStepData

Convert to a form suitable for saving in a specification file

when: list[str]
class tmt.steps.Topology(guests: list[Guest])

Bases: SerializableContainer

Describes the topology of provisioned tmt guests

guest: GuestTopology | None
guest_names: list[str]
guests: dict[str, GuestTopology]
push(*, dirpath: Path, guest: Guest, filename_base: Path | None = None, logger: Logger) → Environment

Save and push topology to a given guest

role_names: list[str]
roles: dict[str, list[str]]
save(*, dirpath: Path, filename_base: Path | None = None) → list[Path]

Save the topology in files.

Parameters:

  • dirpath – a directory to save into.

  • filename_base – if set, it would be used as a base for filenames, correct suffixes would be added.

Returns:

list of paths to saved files.

save_bash(dirpath: Path, filename: Path | None = None) → Path

Save the topology in a Bash-sourceable file.

Parameters:

  • dirpath – a directory to save into.

  • filename – if set, it would be used, otherwise a filename is created from TEST_TOPOLOGY_FILENAME_BASE and .sh suffix.

Returns:

path to the saved file.

save_yaml(dirpath: Path, filename: Path | None = None) → Path

Save the topology in a YAML file.

Parameters:

  • dirpath – a directory to save into.

  • filename – if set, it would be used, otherwise a filename is created from TEST_TOPOLOGY_FILENAME_BASE and .yaml suffix.

Returns:

path to the saved file.

to_dict() → dict[str, Any]

Convert to a mapping.

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

class tmt.steps.WhereableStepData(where: list[str] = <factory>)

Bases: object

Keys shared by step data that may be limited to a particular guest.

To be used as a mixin class, adds necessary keys.

See [1] and [2] for specification.

  1. https://tmt.readthedocs.io/en/stable/spec/plans.html#where

  2. https://tmt.readthedocs.io/en/stable/spec/plans.html#spec-plans-prepare-where

where: list[str]
tmt.steps.is_action_name(val: str) → TypeGuard[Literal['login', 'reboot']]
tmt.steps.is_step_name(val: str) → TypeGuard[Literal['discover', 'provision', 'prepare', 'execute', 'report', 'finish', 'cleanup']]
tmt.steps.provides_method(name: str, doc: str | None = None, order: int = 50, installation_hint: str | None = None) → Callable[[type[BasePlugin]], type[BasePlugin]]

A plugin class decorator to register plugin’s method with tmt steps.

In the following example, developer marks SomePlugin as providing two discover methods, foo and bar, with bar being sorted to later position among methods:

@tmt.steps.provides_method('foo')
@tmt.steps.provides_method('bar', order=80)
class SomePlugin(tmt.steps.discover.DicoverPlugin):
  ...
Parameters:

  • name – name of the method.

  • doc – method documentation. If not specified, docstring of the decorated class is used.

  • order – order of the method among other step methods.

tmt.steps.prune_directory(path: Path, preserved_members: set[str], logger: Logger) → set[str]

Remove content of a given directory while preserving selected members.

Parameters:

  • path – a directory to prune.

  • preserved_members – names of directory entries that should not be removed.

  • logger – used for logging.

Returns:

names of members that were actually preserved, i.e. members that exist in the directory, and were not removed.

tmt.steps.safe_filename(template: str, phase: Phase, guest: Guest, **variables: Any) → Path

Construct a non-conflicting filename safe for parallel tasks.

Function adds enough uniqueness to a given template by adding a phase name and a guest name that the eventual filename would be safe against conflicting access from a phase running on multiple guests, and against reuse when created by the same plugin in different phases.

At first glance, the name might be an overkill: at one moment, there is just one phase running on the given guest, why bother? No other phase would touch the file on the guest. But:

  1. filenames are created locally. One phase, running against multiple guests, still needs the filename to be unique on the tmt runner. Otherwise, phase running in different threads would contest a single file.

  2. while the scenario is unlikely, user may eventually convince tmt to recognize two different names for the same machine, via -h connect --guest $IP. Therefore it may happen that one phase, running against two guests, would actually run on the very same HW. Therefore even the remote per-guest uniqueness is needed.

  3. the phase name is included to avoid reuse of the filename by different phases. A plugin may be invoked by multiple phases, and it might use a “constant” name for the file. That would lead to the filename being reused by different plugin executions. Adding the phase name should lower confusion: it would be immediately clear which phase used which filename, or whether a filename was or was not created by given phase.

Parameters:

  • template – filename template. It is enhanced with phase and guest safe name, but may use other variables provided in variables.

  • phase – a phase owning the resulting filename.

  • guest – a guest on which the filename would be used.

  • variables – additional variables template need when rendering the filename.

tmt.steps.sync_with_guests(step: Step, action: str, task: GuestSyncTaskT, logger: Logger) → None

Push and pull stuff from guests in a parallel manner.

Used by steps that run their plugins against one or more guest. Based on a PhaseQueue primitive, parallelized push/pull operations are needed, and this function handles the details.

Parameters:

  • step – step managing the sync operation.

  • action – push or pull, used for nicer logging.

  • task – PushTask or PullTask which represents the actual operation.

  • logger – logger to use for logging.