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[Union[tmt.base.DependencySimple, tmt.base.DependencyFmfId, tmt.base.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[tmt.result.Result]: Run test on the guest

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

results() → list[tmt.result.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[tmt.utils.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.

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

class tmt.steps.execute.Execute(*, plan: Plan, data: _RawStepData | list[tmt.steps._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[tmt.result.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

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]

Common parent of execute plugins

Store plugin name, data and parent step

classmethod base_command(usage: str, method_class: type[click.core.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_results(invocation: TestInvocation, logger: Logger) → list[tmt.result.Result]: Check the test result

static format_duration(duration: timedelta) → str: Convert duration to a human readable format

static format_timestamp(timestamp: datetime) → str: Convert timestamp to a human readable format

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

how: str = 'tmt'

load_custom_results(invocation: TestInvocation) → list[tmt.result.Result]: Process custom results.yaml file created by the test itself.

load_tmt_report_results(invocation: TestInvocation) → list[tmt.result.Result]

Load results from a file created by tmt-report-result script.

Returns:: list of tmt.Result instances loaded from the file, or an empty list if the file does not exist.

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

prepare_tests(guest: Guest, logger: Logger) → list[tmt.steps.execute.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[tmt.result.Result]: Return test results

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

run_checks_before_test(*, invocation: TestInvocation, environment: Environment | None = None, logger: Logger) → list[tmt.result.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.Script(path: Path, aliases: list[tmt.utils.Path], related_variables: list[str])

Bases: object

Represents a script provided by the internal executor

aliases: list[tmt.utils.Path]

path: Path

related_variables: list[str]

class tmt.steps.execute.ScriptCreatingFile(path: Path, aliases: list[tmt.utils.Path], related_variables: list[str], created_file: str)

Bases: Script

Represents a script which creates a file

created_file: 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.allocate_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, _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.

check_data: dict[str, Any]

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

check_results: list[tmt.result.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.

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

logger: Logger

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: allocate_lock

real_duration: str | None = None

reboot_request_path: A path to the reboot request file

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

relative_path: Invocation directory path relative to step workdir

relative_test_data_path: Test data path relative to step workdir

results: list[tmt.result.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

test_data_path: Absolute path to test data directory