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 withcolor
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 toold
during the first execution andnew
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
- 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_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]
- 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 particularguest
under the supervision of anexecute
pluginphase
.- 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
- 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.
- 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_data_path
Absolute path to test data directory