tmt.steps.execute package

Submodules

tmt.steps.execute.internal module

class tmt.steps.execute.internal.ExecuteInternal(**kwargs: Any)

Bases: ExecutePlugin[ExecuteInternalData]

Use the internal tmt executor to execute tests

The internal tmt executor runs tests on the guest one by one, shows testing progress and supports interactive debugging as well. Test result is based on the script exit code (for shell tests) or the results file (for beakerlib tests).

Store plugin name, data and parent step

cli_invocation: 'tmt.cli.CliInvocation' | None = None

data: ExecuteInternalData

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.

execute(*, invocation: TestInvocation, extra_environment: Environment | None = None, logger: Logger) → list[Result]: Run test on the guest

go(*, guest: Guest, environment: Environment | None = None, logger: Logger) → None: Execute available tests

results() → list[Result]: Return test results

class tmt.steps.execute.internal.ExecuteInternalData(name: str, how: str, order: int = 50, summary: Optional[str] = None, where: list[str] = <factory>, duration: str = '1h', exit_first: bool = False, script: list[tmt.utils.ShellScript] = <factory>, interactive: bool = False, no_progress_bar: bool = False)

Bases: ExecuteStepData

interactive: bool = False

no_progress_bar: bool = False

script: list[ShellScript]

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

tmt.steps.execute.internal.TEST_PIDFILE_ROOT = Path('/var/tmp'): The default directory for storing test pid file.

tmt.steps.execute.internal.TEST_WRAPPER_FILENAME_TEMPLATE = 'tmt-test-wrapper.sh-{{ INVOCATION.test.pathless_safe_name }}-{{ INVOCATION.test.serial_number }}'

A template for the shell wrapper in which the test script is saved.

It is passed to tmt.utils.safe_filename(), but includes also test name and serial number to make it unique even among all test wrappers. See https://github.com/teemtee/tmt/issues/2997 for issue motivating the inclusion, it seems to be a good idea to prevent accidental reuse in general.

class tmt.steps.execute.internal.UpdatableMessage(plugin: ExecuteInternal)

Bases: UpdatableMessage

Updatable message suitable for plan progress reporting.

Based on tmt.utils.UpdatableMessage, simplifies reporting of plan progress, namely by extracting necessary setup parameters from the plugin.

Updatable message suitable for progress-bar-like reporting.

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

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

Parameters:

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

update(progress: str, test_name: str) → None

Update progress message.

Parameters:

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

tmt.steps.execute.internal.effective_pidfile_root() → Path

Find out what the actual pidfile directory is.

If TMT_TEST_PIDFILE_ROOT variable is set, it is used. Otherwise, TEST_PIDFILE_ROOT is picked.

tmt.steps.execute.upgrade module

class tmt.steps.execute.upgrade.ExecuteUpgrade(**kwargs: Any)

Bases: ExecuteInternal

Perform system upgrade during testing.

The upgrade executor runs the discovered tests (using the internal executor), then performs a set of upgrade tasks from a remote repository, and finally, re-runs the tests on the upgraded guest.

The IN_PLACE_UPGRADE environment variable is set during the test execution to differentiate between the stages of the test. It is set to old during the first execution and new during the second execution. Test names are prefixed with this value to make the names unique.

The upgrade tasks performing the actual system upgrade are taken from a remote repository either based on an upgrade path (e.g. fedora35to36) or filters. The upgrade path must correspond to a plan name in the remote repository whose discover step selects tests (upgrade tasks) performing the upgrade. Currently, selection of upgrade tasks in the remote repository can be done using both fmf and shell discover method. The supported keys in discover are:

ref
filter
exclude
tests
test

The environment variables defined in the remote upgrade path plan are passed to the upgrade tasks when they are executed. An example of an upgrade path plan (in the remote repository):

discover: # Selects appropriate upgrade tasks (L1 tests)
    how: fmf
    filter: "tag:fedora"
environment: # This is passed to upgrade tasks
    SOURCE: 35
    TARGET: 36
execute:
    how: tmt

If no upgrade path is specified in the plan, the tests (upgrade tasks) are selected based on the configuration of the upgrade plugin (e.g. based on the filter in its configuration).

If these two possible ways of specifying upgrade tasks are combined, the remote discover plan is used but its options are overridden with the values specified locally.

The same options and config keys and values can be used as in the internal executor.

Minimal execute config example with an upgrade path:

execute:
    how: upgrade
    url: https://github.com/teemtee/upgrade
    upgrade-path: /paths/fedora35to36

Execute config example without an upgrade path:

execute:
    how: upgrade
    url: https://github.com/teemtee/upgrade
    filter: "tag:fedora"

Store plugin name, data and parent step

cli_invocation: 'tmt.cli.CliInvocation' | None = None

data: ExecuteUpgradeData

property discover: Discover | DiscoverFmf: Return discover plugin instance

go(*, guest: Guest, environment: Environment | None = None, logger: Logger) → None: Execute available tests

class tmt.steps.execute.upgrade.ExecuteUpgradeData(name: str, how: str, order: int = 50, summary: Optional[str] = None, where: list[str] = <factory>, duration: str = '1h', exit_first: bool = False, script: list[tmt.utils.ShellScript] = <factory>, interactive: bool = False, no_progress_bar: bool = False, url: Optional[str] = None, upgrade_path: Optional[str] = None, ref: Optional[str] = None, test: list[str] = <factory>, filter: list[str] = <factory>, exclude: list[str] = <factory>)

Bases: ExecuteInternalData

exclude: list[str]

filter: list[str]

ref: str | None = None

test: list[str]

upgrade_path: str | None = None

url: str | None = None

Module contents

tmt.steps.execute.DEFAULT_SCRIPTS_DEST_DIR = Path('/usr/local/bin'): The default scripts destination directory

tmt.steps.execute.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

class tmt.steps.execute.Execute(*, plan: Plan, data: _RawStepData | list[_RawStepData], logger: Logger)

Bases: Step

Run tests using the specified executor

Initialize execute step data

DEFAULT_HOW: str = 'tmt'

cli_invocation: 'tmt.cli.CliInvocation' | None = None

cli_invocations: list['tmt.cli.CliInvocation'] = []

go(force: bool = False) → None: Execute tests

load() → None: Load test results

results() → list[Result]

Results from executed tests

Return a dictionary with test results according to the spec: https://tmt.readthedocs.io/en/latest/spec/plans.html#execute

results_for_tests(tests: list[Test]) → list[tuple[Result | None, Test | None]]

Collect results and corresponding tests.

Returns:: a list of result and test pairs. * if there is not test found for the result, e.g. when results were loaded from storage but tests were not, None represents the missing test: (result, None). * if there is no result for a test, e.g. when the test was not executed, None represents the missing result: (None, test).

save() → None: Save test results to the workdir

summary() → None: Give a concise summary of the execution

wake() → None: Wake up the step (process workdir and command line)

class tmt.steps.execute.ExecutePlugin(*, step: Step, data: ExecuteStepDataT, workdir: Literal[True] | Path | None = None, logger: Logger)

Bases: Plugin[ExecuteStepDataT, None]

Common parent of execute plugins

Store plugin name, data and parent step

classmethod base_command(usage: str, method_class: type[Command] | None = None) → Command: Create base click command (common for all execute plugins)

check_abort_file(invocation: TestInvocation) → bool

Check for an abort file created by tmt-abort

Returns whether an abort file is present (i.e. abort occurred).

cli_invocation: 'tmt.cli.CliInvocation' | None = None

property discover: Discover: Return discover plugin instance

discover_phase: str | None = None: If set, plugin should run tests only from this discover phase.

extract_custom_results(invocation: TestInvocation) → list[Result]: Extract results from the file generated by the test itself

extract_results(invocation: TestInvocation, logger: Logger) → list[Result]: Check the test result

extract_tmt_report_results(invocation: TestInvocation) → list[Result]: Extract results from a file generated by tmt-report-result script

extract_tmt_report_results_restraint(invocation: TestInvocation, default_log: Path) → list[Result]

Extract results from the file generated by tmt-report-result script.

Special, restraint-like handling is used to convert each recorded result into a standalone result.

go(*, guest: Guest, environment: Environment | None = None, logger: Logger) → None: Perform actions shared among plugins when beginning their tasks

how: str = 'tmt'

prepare_scripts(guest: Guest) → None: Prepare additional scripts for testing

prepare_tests(guest: Guest, logger: Logger) → list[TestInvocation]

Prepare discovered tests for testing

Check which tests have been discovered, for each test prepare the aggregated metadata in a file under the test data directory and finally return a list of discovered tests.

results() → list[Result]: Return test results

run_checks_after_test(*, invocation: TestInvocation, environment: Environment | None = None, logger: Logger) → list[CheckResult]

run_checks_before_test(*, invocation: TestInvocation, environment: Environment | None = None, logger: Logger) → list[CheckResult]

scripts: tuple[Script, ...] = ()

timeout_hint(invocation: TestInvocation) → None: Append a duration increase hint to the test output

class tmt.steps.execute.ExecuteStepData(name: str, how: str, order: int = 50, summary: Optional[str] = None, where: list[str] = <factory>, duration: str = '1h', exit_first: bool = False)

Bases: WhereableStepData, StepData

duration: str = '1h'

exit_first: bool = False

class tmt.steps.execute.ResultCollection(invocation: ~tmt.steps.execute.TestInvocation, filepaths: list[~tmt._compat.pathlib.Path], file_exists: bool = False, results: list[~typing.Any] = <factory>)

Bases: object

Collection of raw results loaded from a file

file_exists: bool = False

filepaths: list[Path]

invocation: TestInvocation

results: list[Any]

validate() → None

Validate raw collected results against the result JSON schema.

Report found errors as warnings via invocation logger.

tmt.steps.execute.SCRIPTS_DEST_DIR_VARIABLE = 'TMT_SCRIPTS_DIR': The tmt environment variable name for forcing SCRIPTS_DEST_DIR

tmt.steps.execute.SCRIPTS_SRC_DIR = Path('/home/docs/checkouts/readthedocs.org/user_builds/tmt/envs/3398/lib/python3.13/site-packages/tmt/steps/execute/scripts'): Scripts source directory

class tmt.steps.execute.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.execute.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.execute.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]

class tmt.steps.execute.TestInvocation(logger: ~tmt.log.Logger, phase: ~tmt.steps.execute.ExecutePlugin[~typing.Any], test: ~tmt.base.Test, guest: ~tmt.steps.provision.Guest, process: ~subprocess.Popen[bytes] | None = None, process_lock: ~_thread.lock = <factory>, results: list[~tmt.result.Result] = <factory>, check_results: list[~tmt.result.CheckResult] = <factory>, check_data: dict[str, ~typing.Any] = <factory>, return_code: int | None = None, start_time: str | None = None, end_time: str | None = None, real_duration: str | None = None, _restart_count: int = 0, _reboot_count: int = 0, hard_reboot_requested: bool = False)

Bases: object

A bundle describing one test invocation.

Describes a test invoked on a particular guest under the supervision of an execute plugin phase.

property abort_request_path: Path: A path to the abort request file

property abort_requested: bool: Whether a testing abort was requested

check_data: dict[str, Any]

property check_files_path: Path: Construct a directory path for check files needed by tmt

check_results: list[CheckResult]

end_time: str | None = None

guest: Guest

handle_reboot() → bool

Reboot the guest if the test requested it.

Check for presence of a file signalling reboot request and orchestrate the reboot if it was requested. Also increment the REBOOTCOUNT variable, reset it to zero if no reboot was requested (going forward to the next test).

Returns:: True when the reboot has taken place, False otherwise.

handle_restart() → bool

“Restart” the test if the test requested it.

Note

The test is not actually restarted, because running the test is managed by a plugin calling this method. Instead, the method performs all necessary steps before letting plugin know it should run the test once again.

Check whether a test restart was needed and allowed, and update the accounting info before letting the plugin know it’s time to run the test once again.

If requested by the test, the guest might be rebooted as well.

Returns:: True when the restart is to take place, False otherwise.

hard_reboot_requested: bool = False: If set, an asynchronous observer requested a reboot while the test was running.

property is_guest_healthy: bool: Whether the guest is deemed to be healthy and responsive.

Note

The answer is deduced from various flags set by execute code while observing the test, no additional checks are performed.

logger: Logger

property path: Path: Absolute path to invocation directory

phase: ExecutePlugin[Any]

process: Popen[bytes] | None = None: Process running the test. What binary it is depends on the guest implementation and the test, it may be, for example, a shell process, SSH process, or a podman process.

process_lock: lock

real_duration: str | None = None

property reboot_request_path: Path: A path to the reboot request file

property reboot_requested: bool: Whether a guest reboot has been requested while the test was running

property relative_path: Path: Invocation directory path relative to step workdir

property relative_test_data_path: Path: Test data path relative to step workdir

property restart_requested: bool: Whether a test restart has been requested

results: list[Result]

return_code: int | None = None

property soft_reboot_requested: bool: If set, test requested a reboot

start_time: str | None = None

terminate_process(signal: Signals = Signals.SIGTERM, logger: Logger | None = None) → None

Terminate the invocation process.

Warning

This method should be used carefully. Process running the invocation’s test has been started by some part of tmt code which is responsible for its well-being. Unless you have a really good reason to do so, doing things behind the tmt’s back may lead to unexpected results.

Parameters:

signal – signal to send to the invocation process.
logger – logger to use for logging.

test: Test

property test_data_path: Path: Absolute path to test data directory

tmt.steps.execute.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.