files.new_contents: 
  """ Step Classes """
  
  import re
  import sys
  import textwrap
  import warnings
  from typing import (TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Type,
                      TypeVar, Union, cast, overload)
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  import click
  from click import echo
  
  import tmt.options
  import tmt.utils
  from tmt.options import show_step_method_hints
  
  if TYPE_CHECKING:
      import tmt.base
      from tmt.base import Plan
      from tmt.steps.provision import Guest
  
  
  DEFAULT_PLUGIN_METHOD_ORDER: int = 50
  
  
  # Supported steps and actions
  STEPS = ['discover', 'provision', 'prepare', 'execute', 'report', 'finish']
  ACTIONS = ['login', 'reboot']
  
  # Step phase order
  PHASE_START = 10
  PHASE_BASE = 50
  PHASE_END = 90
  
  
  class Phase(tmt.utils.Common):
      """ A phase of a step """
  
      def __init__(
              self,
              order: int = tmt.utils.DEFAULT_PLUGIN_ORDER,
              *args: Any,
              **kwargs: Any):
          super().__init__(*args, **kwargs)
          self.order: int = order
  
      def enabled_on_guest(self, guest: 'Guest') -> bool:
          """ Phases are enabled across all guests by default """
          return True
  
      @property
      def is_in_standalone_mode(self) -> 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).
          """
          return False
  
  
  # A variable used to describe a generic type for all classes derived from Phase
  PhaseT = TypeVar('PhaseT', bound=Phase)
  
  
  class StepData(TypedDict, total=False):
      """
      Step data structure
      """
      name: str
      how: str
      order: Optional[int]
      tests: Optional[List['tmt.base.Test']]
  
  
  class Step(tmt.utils.Common):
      """ Common parent of all test steps """
  
      # Default implementation for all steps is "shell", but some
      # steps like provision may have better defaults for their
      # area of expertise.
      DEFAULT_HOW: str = 'shell'
  
      def __init__(
              self,
              plan: 'Plan',
              data: Optional[Union[StepData, List[StepData]]] = None,
              name: Optional[str] = None,
              workdir: tmt.utils.WorkdirArgumentType = None) -> None:
          """ Initialize and check the step data """
          super().__init__(name=name, parent=plan, workdir=workdir)
          # Initialize data
          self.plan: 'Plan' = plan
          self._status: Optional[str] = None
          self._phases: List[Phase] = []
  
          # Normalize `data` attribute to carry a list of step configuration data, one item per
          # distinct step configuration. Make sure all items have `name`` and `how` keys.
          self.data: List[StepData] = []
  
          # Create an empty step by default (can be updated from cli)
          if data is None:
              self.data = [{}]
  
          # Convert to list if only a single config provided
          elif isinstance(data, dict):
              self.data = [data]
  
          # List is as good as it gets
          elif isinstance(data, list):
              self.data = data
  
          # Shout about invalid configuration
          else:
              raise tmt.utils.GeneralError(
                  f"Invalid '{self}' config in '{self.plan}'.")
  
          # Final sanity checks
          for i, single_config_data in enumerate(self.data):
              # Add default unique names even to multiple configs so that the users
              # don't need to specify it if they don't care about the name
              if 'name' not in single_config_data:
                  single_config_data['name'] = f'{tmt.utils.DEFAULT_NAME}-{i}'
  
              # Set 'how' to the default if not specified
              if single_config_data.get('how') is None:
                  single_config_data['how'] = self.DEFAULT_HOW
  
      @property
      def enabled(self) -> Optional[bool]:
          """ True if the step is enabled """
          try:
              return self.name in self.plan.my_run._context.obj.steps
          except AttributeError:
              return None
  
      @property
      def plugins_in_standalone_mode(self) -> 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).
          """
          return sum(phase.is_in_standalone_mode for phase in self.phases())
  
      @classmethod
      def usage(cls, method_overview: str) -> str:
          """ Prepare general usage message for the step """
          # Main description comes from the class docstring
          if cls.__name__ is None:
              raise tmt.utils.GeneralError("Missing name of the step.")
  
          if cls.__doc__ is None:
              raise tmt.utils.GeneralError(
                  f"Missing docstring of the step {cls.__name__.lower()}.")
  
          usage = textwrap.dedent(cls.__doc__)
          # Append the list of supported methods
          usage += '\n\n' + method_overview
          # Give a hint about detailed help
          name = cls.__name__.lower()
          usage += (
              f"\n\nUse 'tmt run {name} --help --how <method>' to learn more "
              f"about given {name} method and all its supported options.")
          return usage
  
      def status(self, status: Optional[str] = None) -> Optional[str]:
          """
          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)
          """
          # Update status
          if status is not None:
              # Check for valid values
              if status not in ['todo', 'done']:
                  raise tmt.utils.GeneralError(f"Invalid status '{status}'.")
              # Show status only if changed
              elif self._status != status:
                  self._status = status
                  self.debug('status', status, color='yellow', level=2)
          # Return status
          return self._status
  
      def load(self) -> None:
          """ Load status and step data from the workdir """
          try:
              content: Dict[Any, Any] = tmt.utils.yaml_to_dict(
                  self.read('step.yaml'))
              self.debug('Successfully loaded step data.', level=2)
              self.data = content['data']
              self.status(content['status'])
          except tmt.utils.GeneralError:
              self.debug('Step data not found.', level=2)
  
      def save(self) -> None:
          """ Save status and step data to the workdir """
          content: Dict[str, Any] = {
              'status': self.status(),
              'data': self.data
              }
          self.write('step.yaml', tmt.utils.dict_to_yaml(content))
  
      def wake(self) -> None:
          """ Wake up the step (process workdir and command line) """
          # Cleanup possible old workdir if called with --force
          if self.opt('force'):
              self._workdir_cleanup()
  
          # Load stored data
          self.load()
  
          # Status 'todo' means the step has not finished successfully.
          # Probably interrupted in the middle. Clean up the work
          # directory to give it another chance with a fresh start.
          if self.status() == 'todo':
              self.debug("Step has not finished. Let's try once more!", level=2)
              self._workdir_cleanup()
  
          # Importing here to avoid circular imports
          import tmt.steps.report
  
          # Special handling for the report step to always enable force mode in
          # order to cover a very frequent use case 'tmt run --last report'
          # FIXME find a better way how to enable always-force per plugin
          if (isinstance(self, tmt.steps.report.Report) and
                  self.data[0].get('how') in ['display', 'html']):
              self.debug("Report step always force mode enabled.")
              self._workdir_cleanup()
              self.status('todo')
  
          # Nothing more to do when the step is already done
          if self.status() == 'done':
              return
  
          # Override step data with command line options
          how: str = self.opt('how')
          if how is not None:
              for data in self.data:
                  assert isinstance(data, dict)
                  data['how'] = how
  
      def setup_actions(self) -> None:
          """ Insert login and reboot plugins if requested """
          for login_plugin in Login.plugins(step=self):
              self.debug(
                  f"Insert a login plugin into the '{self}' step "
                  f"with order '{login_plugin.order}'.", level=2)
              self._phases.append(login_plugin)
  
          for reboot_plugin in Reboot.plugins(step=self):
              self.debug(
                  f"Insert a reboot plugin into the '{self}' step "
                  f"with order '{reboot_plugin.order}'.", level=2)
              self._phases.append(reboot_plugin)
  
      @overload
      def phases(self, classes: None = None) -> List[Phase]:
          pass
  
      @overload
      def phases(self, classes: Type[PhaseT]) -> List[PhaseT]:
          pass
  
      @overload
      def phases(self, classes: Tuple[Type[PhaseT], ...]) -> List[PhaseT]:
          pass
  
      def phases(self, classes: Optional[Union[Type[PhaseT],
                 Tuple[Type[PhaseT], ...]]] = None) -> List[PhaseT]:
          """
  
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/__init__.py
files.old_contents: 
  """ Step Classes """
  
  import re
  import sys
  import textwrap
  import warnings
  from typing import (TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Type,
                      TypeVar, Union, cast, overload)
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  import click
  from click import echo
  
  import tmt.options
  import tmt.utils
  from tmt.options import show_step_method_hints
  
  if TYPE_CHECKING:
      import tmt.base
      from tmt.base import Plan
      from tmt.steps.provision import Guest
  
  
  DEFAULT_PLUGIN_METHOD_ORDER: int = 50
  
  
  # Supported steps and actions
  STEPS = ['discover', 'provision', 'prepare', 'execute', 'report', 'finish']
  ACTIONS = ['login', 'reboot']
  
  # Step phase order
  PHASE_START = 10
  PHASE_BASE = 50
  PHASE_END = 90
  
  
  class Phase(tmt.utils.Common):
      """ A phase of a step """
  
      def __init__(
              self,
              order: int = tmt.utils.DEFAULT_PLUGIN_ORDER,
              *args: Any,
              **kwargs: Any):
          super().__init__(*args, **kwargs)
          self.order: int = order
  
      def enabled_on_guest(self, guest: 'Guest') -> bool:
          """ Phases are enabled across all guests by default """
          return True
  
      @property
      def is_in_standalone_mode(self) -> 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).
          """
          return False
  
  
  # A variable used to describe a generic type for all classes derived from Phase
  PhaseT = TypeVar('PhaseT', bound=Phase)
  
  
  class StepData(TypedDict, total=False):
      """
      Step data structure
      """
      name: str
      how: str
      order: Optional[int]
      tests: Optional[List['tmt.base.Test']]
  
  
  class Step(tmt.utils.Common):
      """ Common parent of all test steps """
  
      # Default implementation for all steps is "shell", but some
      # steps like provision may have better defaults for their
      # area of expertise.
      DEFAULT_HOW: str = 'shell'
  
      def __init__(
              self,
              plan: 'Plan',
              data: Optional[Union[StepData, List[StepData]]] = None,
              name: Optional[str] = None,
              workdir: tmt.utils.WorkdirArgumentType = None) -> None:
          """ Initialize and check the step data """
          super().__init__(name=name, parent=plan, workdir=workdir)
          # Initialize data
          self.plan: 'Plan' = plan
          self._status: Optional[str] = None
          self._phases: List[Phase] = []
  
          # Normalize `data` attribute to carry a list of step configuration data, one item per
          # distinct step configuration. Make sure all items have `name`` and `how` keys.
          self.data: List[StepData] = []
  
          # Create an empty step by default (can be updated from cli)
          if data is None:
              self.data = [{}]
  
          # Convert to list if only a single config provided
          elif isinstance(data, dict):
              self.data = [data]
  
          # List is as good as it gets
          elif isinstance(data, list):
              self.data = data
  
          # Shout about invalid configuration
          else:
              raise tmt.utils.GeneralError(
                  f"Invalid '{self}' config in '{self.plan}'.")
  
          # Final sanity checks
          for i, single_config_data in enumerate(self.data):
              # Add default unique names even to multiple configs so that the users
              # don't need to specify it if they don't care about the name
              if 'name' not in single_config_data:
                  single_config_data['name'] = f'{tmt.utils.DEFAULT_NAME}-{i}'
  
              # Set 'how' to the default if not specified
              if single_config_data.get('how') is None:
                  single_config_data['how'] = self.DEFAULT_HOW
  
      @property
      def enabled(self) -> Optional[bool]:
          """ True if the step is enabled """
          try:
              return self.name in self.plan.my_run._context.obj.steps
          except AttributeError:
              return None
  
      @property
      def plugins_in_standalone_mode(self) -> 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).
          """
          return sum(phase.is_in_standalone_mode for phase in self.phases())
  
      @classmethod
      def usage(cls, method_overview: str) -> str:
          """ Prepare general usage message for the step """
          # Main description comes from the class docstring
          if cls.__name__ is None:
              raise tmt.utils.GeneralError("Missing name of the step.")
  
          if cls.__doc__ is None:
              raise tmt.utils.GeneralError(
                  f"Missing docstring of the step {cls.__name__.lower()}.")
  
          usage = textwrap.dedent(cls.__doc__)
          # Append the list of supported methods
          usage += '\n\n' + method_overview
          # Give a hint about detailed help
          name = cls.__name__.lower()
          usage += (
              f"\n\nUse 'tmt run {name} --help --how <method>' to learn more "
              f"about given {name} method and all its supported options.")
          return usage
  
      def status(self, status: Optional[str] = None) -> Optional[str]:
          """
          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)
          """
          # Update status
          if status is not None:
              # Check for valid values
              if status not in ['todo', 'done']:
                  raise tmt.utils.GeneralError(f"Invalid status '{status}'.")
              # Show status only if changed
              elif self._status != status:
                  self._status = status
                  self.debug('status', status, color='yellow', level=2)
          # Return status
          return self._status
  
      def load(self) -> None:
          """ Load status and step data from the workdir """
          try:
              content: Dict[Any, Any] = tmt.utils.yaml_to_dict(
                  self.read('step.yaml'))
              self.debug('Successfully loaded step data.', level=2)
              self.data = content['data']
              self.status(content['status'])
          except tmt.utils.GeneralError:
              self.debug('Step data not found.', level=2)
  
      def save(self) -> None:
          """ Save status and step data to the workdir """
          content: Dict[str, Any] = {
              'status': self.status(),
              'data': self.data
              }
          self.write('step.yaml', tmt.utils.dict_to_yaml(content))
  
      def wake(self) -> None:
          """ Wake up the step (process workdir and command line) """
          # Cleanup possible old workdir if called with --force
          if self.opt('force'):
              self._workdir_cleanup()
  
          # Load stored data
          self.load()
  
          # Status 'todo' means the step has not finished successfully.
          # Probably interrupted in the middle. Clean up the work
          # directory to give it another chance with a fresh start.
          if self.status() == 'todo':
              self.debug("Step has not finished. Let's try once more!", level=2)
              self._workdir_cleanup()
  
          # Importing here to avoid circular imports
          import tmt.steps.report
  
          # Special handling for the report step to always enable force mode in
          # order to cover a very frequent use case 'tmt run --last report'
          # FIXME find a better way how to enable always-force per plugin
          if (isinstance(self, tmt.steps.report.Report) and
                  self.data[0].get('how') in ['display', 'html']):
              self.debug("Report step always force mode enabled.")
              self._workdir_cleanup()
              self.status('todo')
  
          # Nothing more to do when the step is already done
          if self.status() == 'done':
              return
  
          # Override step data with command line options
          how: str = self.opt('how')
          if how is not None:
              for data in self.data:
                  assert isinstance(data, dict)
                  data['how'] = how
  
      def setup_actions(self) -> None:
          """ Insert login and reboot plugins if requested """
          for login_plugin in Login.plugins(step=self):
              self.debug(
                  f"Insert a login plugin into the '{self}' step "
                  f"with order '{login_plugin.order}'.", level=2)
              self._phases.append(login_plugin)
  
          for reboot_plugin in Reboot.plugins(step=self):
              self.debug(
                  f"Insert a reboot plugin into the '{self}' step "
                  f"with order '{reboot_plugin.order}'.", level=2)
              self._phases.append(reboot_plugin)
  
      @overload
      def phases(self, classes: None = None) -> List[Phase]:
          pass
  
      @overload
      def phases(self, classes: Type[PhaseT]) -> List[PhaseT]:
          pass
  
      @overload
      def phases(self, classes: Tuple[Type[PhaseT], ...]) -> List[PhaseT]:
          pass
  
      def phases(self, classes: Optional[Union[Type[PhaseT],
                 Tuple[Type[PhaseT], ...]]] = None) -> List[PhaseT]:
          """
  
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/__init__.py
files.new_contents: import glob
  import os
  import re
  import shutil
  import subprocess
  
  import click
  import fmf
  
  import tmt
  import tmt.beakerlib
  import tmt.steps
  import tmt.steps.discover
  from tmt.utils import git_clone
  
  
  @tmt.steps.provides_method('fmf')
  class DiscoverFmf(tmt.steps.discover.DiscoverPlugin):
      """
      Discover available tests from fmf metadata
  
      By default all available tests from the current repository are used
      so the minimal configuration looks like this:
  
          discover:
              how: fmf
  
      Full config example:
  
          discover:
              how: fmf
              url: https://github.com/teemtee/tmt
              ref: main
              path: /fmf/root
              test: /tests/basic
              filter: 'tier: 1'
  
      If no 'ref' is provided, the default branch from the origin is used.
  
      For DistGit repo one can extract source tarball first and discover
      tests from it by using 'distgit-source: true'. It can be used
      together with 'ref', 'path' and 'url', However 'ref' is not possible
      without using 'url'.
  
          discover:
              how: fmf
              dist-git-source: true
  
      Related config options (all optional):
      * dist-git-merge - set to True if you want to copy in extracted
        sources to the local repo
      * dist-git-init - set to True and 'fmf init' will be called inside
        extracted sources (at dist-git-extract or top directory)
      * dist-git-extract - directory (glob supported) to copy from
        extracted sources (defaults to inner fmf-root)
      * dist-git-remove-fmf-root - set to True to remove fmf root from
        extracted sources
  
      Selecting tests containing specified link is possible using 'link'
      option accepting RELATION:TARGET format of values. Regular
      expressions are supported for both relation and target part of the
      value. Relation can be omitted to target match any relation.
  
          discover:
              how: fmf
              link: verifies:.*issue/850$
  
      It is also possible to limit tests only to those that have changed
      in git since a given revision. This can be particularly useful when
      testing changes to tests themselves (e.g. in a pull request CI).
  
      Related config options (all optional):
      * modified-only - set to True if you want to filter modified tests
      * modified-url - fetched as "reference" remote in the test dir
      * modified-ref - the ref to compare against
  
      Example to compare local repo against upstream 'main' branch:
  
          discover:
              how: fmf
              modified-only: True
              modified-url: https://github.com/teemtee/tmt
              modified-ref: reference/main
  
      Note that internally the modified tests are appended to the list
      specified via 'test', so those tests will also be selected even if
      not modified.
      """
  
      # Supported keys
      _keys = [
          "url", "ref", "path", "test", "link", "filter",
          "modified-only", "modified-url", "modified-ref",
          "dist-git-init", "dist-git-remove-fmf-root", "dist-git-merge",
          "dist-git-extract",
          "fmf-id", "exclude"]
  
      REF_OPTION = click.option(
          '-r', '--ref', metavar='REVISION',
          help='Branch, tag or commit specifying the git revision.')
      TEST_OPTION = click.option(
          '-t', '--test', metavar='NAMES', multiple=True,
          help='Select tests by name.')
      FILTER_OPTION = click.option(
          '-F', '--filter', metavar='FILTERS', multiple=True,
          help='Include only tests matching the filter.')
      EXCLUDE_OPTION = click.option(
          '-x', '--exclude', metavar='[REGEXP]', multiple=True,
          help="Exclude a regular expression from search result.")
  
      # Options which require .git to be present for their functionality
      _REQUIRES_GIT = (
          "ref",
          "modified-url",
          "modified-only",
          "fmf-id",
          )
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          return [
              click.option(
                  '-u', '--url', metavar='REPOSITORY',
                  help='URL of the git repository with fmf metadata.'),
              cls.REF_OPTION,
              click.option(
                  '--modified-url', metavar='REPOSITORY',
                  help='URL of the reference git repository with fmf metadata.'),
              click.option(
                  '--modified-ref', metavar='REVISION',
                  help='Branch, tag or commit specifying the reference git '
                  'revision (if not provided, the default branch is used).'),
              click.option(
                  '-m', '--modified-only', is_flag=True,
                  help='If set, select only tests modified '
                  'since reference revision.'),
              click.option(
                  '-p', '--path', metavar='ROOT',
                  help='Path to the metadata tree root.'),
              cls.TEST_OPTION,
              click.option(
                  '--link', metavar="RELATION:TARGET", multiple=True,
                  help="Filter by linked objects (regular expressions are "
                       "supported for both relation and target)."),
              cls.FILTER_OPTION,
              cls.EXCLUDE_OPTION,
              click.option(
                  '--fmf-id', default=False, is_flag=True,
                  help='Show fmf identifiers for tests discovered in plan.'),
              click.option(
                  '--sync-repo', default=False, is_flag=True,
                  help='Force the sync of the whole git repo. By default, the '
                       'repo is copied only if the used options require it.'),
              click.option(
                  '--dist-git-init', is_flag=True,
                  help='Initialize fmf root inside extracted sources '
                       '(at dist-git-extract or top directory).'),
              click.option(
                  '--dist-git-remove-fmf-root', is_flag=True,
                  help='Remove fmf root from extracted source '
                       '(top one or selected by copy-path, '
                       'happens before dist-git-extract.'),
              click.option(
                  '--dist-git-merge', is_flag=True,
                  help='Merge copied sources and plan fmf root.'),
              click.option(
                  '--dist-git-extract',
                  help='What to copy from extracted sources, globbing is '
                       'supported. Defaults to the top fmf root if it is '
                       'present, otherwise top directory (shortcut "/").'
                  ),
              ] + super().options(how)
  
      def wake(self):
          """ Wake up the plugin, process data, apply options """
          # Handle backward-compatible stuff
          if 'repository' in self.data:
              self.data['url'] = self.data.pop('repository')
          if 'revision' in self.data:
              self.data['ref'] = self.data.pop('revision')
  
          # Make sure that 'exclude', 'filter' and 'test' keys are lists
          tmt.utils.listify(self.data, keys=["exclude", "filter", "test"])
  
          # Process command line options, apply defaults
          super().wake()
  
      @property
      def is_in_standalone_mode(self):
          """ Enable standalone mode when listing fmf ids """
          if self.opt('fmf_id'):
              return True
          return super().is_in_standalone_mode
  
      def go(self):
          """ Discover available tests """
          super(DiscoverFmf, self).go()
  
          # Check url and path, prepare test directory
          url = self.get('url')
          path = self.get('path')
          # Save the test directory so that others can reference it
          ref = self.get('ref')
          self.testdir = os.path.join(self.workdir, 'tests')
          sourcedir = os.path.join(self.workdir, 'source')
          dist_git_source = self.get('dist-git-source', False)
          dist_git_merge = self.get('dist-git-merge', False)
          dist_git_init = self.get('dist-git-init', False)
          dist_git_extract = self.get('dist-git-extract', None)
          dist_git_remove_fmf_root = self.get('dist-git-remove-fmf-root', False)
  
          # Self checks
          if dist_git_source and not dist_git_merge and (ref or url):
              raise tmt.utils.DiscoverError(
                  "Cannot manipulate with dist-git without "
                  "the `--dist-git-merge` option.")
  
          def get_git_root(dir):
              return self.run(
                  ["git", "rev-parse", "--show-toplevel"],
                  cwd=dir, dry=True)[0].strip('\n')
  
          # Raise an exception if --fmf-id uses w/o url and git root
          # doesn't exist for discovered plan
          if self.opt('fmf_id'):
              def assert_git_url(plan_name=None):
                  try:
                      subprocess.run(
                          'git rev-parse --show-toplevel'.split(),
                          stdout=subprocess.PIPE,
                          stderr=subprocess.DEVNULL,
                          check=True)
                  except subprocess.CalledProcessError:
                      raise tmt.utils.DiscoverError(
                          f"`tmt run discover --fmf-id` without `url` option in "
                          f"plan `{plan_name}` can be used only within"
                          f" git repo.")
              # It covers only one case, when there is:
              # 1) no --url on CLI
              # 2) plan w/o url exists in test run
              if not self.opt('url'):
                  try:
                      fmf_tree = fmf.Tree(os.getcwd())
                  except fmf.utils.RootError:
                      raise tmt.utils.DiscoverError(
                          "No metadata found in the current directory. "
                          "Use 'tmt init' to get started.")
                  for i, attr in enumerate(fmf_tree.climb()):
                      try:
                          plan_url = attr.data.get('discover').get('url')
                          plan_name = attr.name
                          if not plan_url:
                              assert_git_url
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/discover/fmf.py
files.old_contents: import glob
  import os
  import re
  import shutil
  import subprocess
  
  import click
  import fmf
  
  import tmt
  import tmt.beakerlib
  import tmt.steps
  import tmt.steps.discover
  from tmt.utils import git_clone
  
  
  @tmt.steps.provides_method('fmf')
  class DiscoverFmf(tmt.steps.discover.DiscoverPlugin):
      """
      Discover available tests from fmf metadata
  
      By default all available tests from the current repository are used
      so the minimal configuration looks like this:
  
          discover:
              how: fmf
  
      Full config example:
  
          discover:
              how: fmf
              url: https://github.com/teemtee/tmt
              ref: main
              path: /fmf/root
              test: /tests/basic
              filter: 'tier: 1'
  
      If no 'ref' is provided, the default branch from the origin is used.
  
      For DistGit repo one can extract source tarball first and discover
      tests from it by using 'distgit-source: true'. It can be used
      together with 'ref', 'path' and 'url', However 'ref' is not possible
      without using 'url'.
  
          discover:
              how: fmf
              dist-git-source: true
  
      Related config options (all optional):
      * dist-git-merge - set to True if you want to copy in extracted
        sources to the local repo
      * dist-git-init - set to True and 'fmf init' will be called inside
        extracted sources (at dist-git-extract or top directory)
      * dist-git-extract - directory (glob supported) to copy from
        extracted sources (defaults to inner fmf-root)
      * dist-git-remove-fmf-root - set to True to remove fmf root from
        extracted sources
  
      Selecting tests containing specified link is possible using 'link'
      option accepting RELATION:TARGET format of values. Regular
      expressions are supported for both relation and target part of the
      value. Relation can be omitted to target match any relation.
  
          discover:
              how: fmf
              link: verifies:.*issue/850$
  
      It is also possible to limit tests only to those that have changed
      in git since a given revision. This can be particularly useful when
      testing changes to tests themselves (e.g. in a pull request CI).
  
      Related config options (all optional):
      * modified-only - set to True if you want to filter modified tests
      * modified-url - fetched as "reference" remote in the test dir
      * modified-ref - the ref to compare against
  
      Example to compare local repo against upstream 'main' branch:
  
          discover:
              how: fmf
              modified-only: True
              modified-url: https://github.com/teemtee/tmt
              modified-ref: reference/main
  
      Note that internally the modified tests are appended to the list
      specified via 'test', so those tests will also be selected even if
      not modified.
      """
  
      # Supported keys
      _keys = [
          "url", "ref", "path", "test", "link", "filter",
          "modified-only", "modified-url", "modified-ref",
          "dist-git-init", "dist-git-remove-fmf-root", "dist-git-merge",
          "dist-git-extract",
          "fmf-id", "exclude"]
  
      REF_OPTION = click.option(
          '-r', '--ref', metavar='REVISION',
          help='Branch, tag or commit specifying the git revision.')
      TEST_OPTION = click.option(
          '-t', '--test', metavar='NAMES', multiple=True,
          help='Select tests by name.')
      FILTER_OPTION = click.option(
          '-F', '--filter', metavar='FILTERS', multiple=True,
          help='Include only tests matching the filter.')
      EXCLUDE_OPTION = click.option(
          '-x', '--exclude', metavar='[REGEXP]', multiple=True,
          help="Exclude a regular expression from search result.")
  
      # Options which require .git to be present for their functionality
      _REQUIRES_GIT = (
          "ref",
          "modified-url",
          "modified-only",
          "fmf-id",
          )
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          return [
              click.option(
                  '-u', '--url', metavar='REPOSITORY',
                  help='URL of the git repository with fmf metadata.'),
              cls.REF_OPTION,
              click.option(
                  '--modified-url', metavar='REPOSITORY',
                  help='URL of the reference git repository with fmf metadata.'),
              click.option(
                  '--modified-ref', metavar='REVISION',
                  help='Branch, tag or commit specifying the reference git '
                  'revision (if not provided, the default branch is used).'),
              click.option(
                  '-m', '--modified-only', is_flag=True,
                  help='If set, select only tests modified '
                  'since reference revision.'),
              click.option(
                  '-p', '--path', metavar='ROOT',
                  help='Path to the metadata tree root.'),
              cls.TEST_OPTION,
              click.option(
                  '--link', metavar="RELATION:TARGET", multiple=True,
                  help="Filter by linked objects (regular expressions are "
                       "supported for both relation and target)."),
              cls.FILTER_OPTION,
              cls.EXCLUDE_OPTION,
              click.option(
                  '--fmf-id', default=False, is_flag=True,
                  help='Show fmf identifiers for tests discovered in plan.'),
              click.option(
                  '--sync-repo', default=False, is_flag=True,
                  help='Force the sync of the whole git repo. By default, the '
                       'repo is copied only if the used options require it.'),
              click.option(
                  '--dist-git-init', is_flag=True,
                  help='Initialize fmf root inside extracted sources '
                       '(at dist-git-extract or top directory).'),
              click.option(
                  '--dist-git-remove-fmf-root', is_flag=True,
                  help='Remove fmf root from extracted source '
                       '(top one or selected by copy-path, '
                       'happens before dist-git-extract.'),
              click.option(
                  '--dist-git-merge', is_flag=True,
                  help='Merge copied sources and plan fmf root.'),
              click.option(
                  '--dist-git-extract',
                  help='What to copy from extracted sources, globbing is '
                       'supported. Defaults to the top fmf root if it is '
                       'present, otherwise top directory (shortcut "/").'
                  ),
              ] + super().options(how)
  
      def wake(self, keys=None):
          """ Wake up the plugin, process data, apply options """
          # Handle backward-compatible stuff
          if 'repository' in self.data:
              self.data['url'] = self.data.pop('repository')
          if 'revision' in self.data:
              self.data['ref'] = self.data.pop('revision')
  
          # Make sure that 'exclude', 'filter' and 'test' keys are lists
          tmt.utils.listify(self.data, keys=["exclude", "filter", "test"])
  
          # Process command line options, apply defaults
          super().wake(keys=keys)
  
      @property
      def is_in_standalone_mode(self):
          """ Enable standalone mode when listing fmf ids """
          if self.opt('fmf_id'):
              return True
          return super().is_in_standalone_mode
  
      def go(self):
          """ Discover available tests """
          super(DiscoverFmf, self).go()
  
          # Check url and path, prepare test directory
          url = self.get('url')
          path = self.get('path')
          # Save the test directory so that others can reference it
          ref = self.get('ref')
          self.testdir = os.path.join(self.workdir, 'tests')
          sourcedir = os.path.join(self.workdir, 'source')
          dist_git_source = self.get('dist-git-source', False)
          dist_git_merge = self.get('dist-git-merge', False)
          dist_git_init = self.get('dist-git-init', False)
          dist_git_extract = self.get('dist-git-extract', None)
          dist_git_remove_fmf_root = self.get('dist-git-remove-fmf-root', False)
  
          # Self checks
          if dist_git_source and not dist_git_merge and (ref or url):
              raise tmt.utils.DiscoverError(
                  "Cannot manipulate with dist-git without "
                  "the `--dist-git-merge` option.")
  
          def get_git_root(dir):
              return self.run(
                  ["git", "rev-parse", "--show-toplevel"],
                  cwd=dir, dry=True)[0].strip('\n')
  
          # Raise an exception if --fmf-id uses w/o url and git root
          # doesn't exist for discovered plan
          if self.opt('fmf_id'):
              def assert_git_url(plan_name=None):
                  try:
                      subprocess.run(
                          'git rev-parse --show-toplevel'.split(),
                          stdout=subprocess.PIPE,
                          stderr=subprocess.DEVNULL,
                          check=True)
                  except subprocess.CalledProcessError:
                      raise tmt.utils.DiscoverError(
                          f"`tmt run discover --fmf-id` without `url` option in "
                          f"plan `{plan_name}` can be used only within"
                          f" git repo.")
              # It covers only one case, when there is:
              # 1) no --url on CLI
              # 2) plan w/o url exists in test run
              if not self.opt('url'):
                  try:
                      fmf_tree = fmf.Tree(os.getcwd())
                  except fmf.utils.RootError:
                      raise tmt.utils.DiscoverError(
                          "No metadata found in the current directory. "
                          "Use 'tmt init' to get started.")
                  for i, attr in enumerate(fmf_tree.climb()):
                      try:
                          plan_url = attr.data.get('discover').get('url')
                          plan_name = attr.name
                          if not plan
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/discover/fmf.py
files.new_contents: import copy
  import os
  from typing import TYPE_CHECKING, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.discover
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  @tmt.steps.provides_method('shell')
  class DiscoverShell(tmt.steps.discover.DiscoverPlugin):
      """
      Use provided list of shell script tests
  
      List of test cases to be executed can be defined manually directly
      in the plan as a list of dictionaries containing test name, actual
      test script and optionally a path to the test. Example config:
  
      discover:
          how: shell
          tests:
          - name: /help/main
            test: tmt --help
          - name: /help/test
            test: tmt test --help
          - name: /help/smoke
            test: ./smoke.sh
            path: /tests/shell
  
      For DistGit repo one can extract source tarball and use its code.
      It is extracted to TMT_SOURCE_DIR however no patches are applied
      (only source tarball is extracted).
  
      discover:
          how: shell
          dist-git-source: true
          tests:
          - name: /upstream
            test: cd $TMT_SOURCE_DIR/*/tests && make test
      """
  
      def show(self, keys: Optional[List[str]] = None) -> None:
          """ Show config details """
          super().show([])
          tests = self.get('tests')
          if tests:
              test_names = [test['name'] for test in tests]
              click.echo(tmt.utils.format('tests', test_names))
  
      def wake(self) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake()
          # Check provided tests, default to an empty list
          if 'tests' not in self.data:
              self.data['tests'] = []
          self._tests: List[tmt.base.Test] = []
  
      def go(self) -> None:
          """ Discover available tests """
          super(DiscoverShell, self).go()
          tests = fmf.Tree(dict(summary='tests'))
  
          # dist-git related
          assert self.workdir is not None
          sourcedir = os.path.join(self.workdir, 'source')
          dist_git_source = self.get('dist-git-source', False)
  
          # Check and process each defined shell test
          assert self.data['tests'] is not None
          for data in self.data['tests']:
              # Create data copy (we want to keep original data for save()
              data = copy.deepcopy(data)
              # Extract name, make sure it is present
              try:
                  name = data.pop('name')
              except KeyError:
                  raise tmt.utils.SpecificationError(
                      f"Missing test name in '{self.step.plan.name}'.")
              # Make sure that the test script is defined
              if 'test' not in data:
                  raise tmt.utils.SpecificationError(
                      f"Missing test script in '{self.step.plan.name}'.")
              # Prepare path to the test working directory (tree root by default)
              try:
                  data['path'] = f"/tests{data['path']}"
              except KeyError:
                  data['path'] = "/tests"
              # Apply default test duration unless provided
              if 'duration' not in data:
                  data['duration'] = tmt.base.DEFAULT_TEST_DURATION_L2
              # Add source dir path variable
              if dist_git_source:
                  data.setdefault('environment', {})[
                      'TMT_SOURCE_DIR'] = sourcedir
              # Create a simple fmf node, adjust its name
              tests.child(name, data)
  
          # Symlink tests directory to the plan work tree
          testdir = os.path.join(self.workdir, "tests")
          relative_path = os.path.relpath(self.step.plan.worktree, self.workdir)
          os.symlink(relative_path, testdir)
  
          if dist_git_source:
              try:
                  run_result = self.run(
                      ["git", "rev-parse", "--show-toplevel"],
                      cwd=self.step.plan.my_run.tree.root,
                      dry=True)[0]
                  assert run_result is not None
                  git_root = run_result.strip('\n')
              except tmt.utils.RunError:
                  raise tmt.utils.DiscoverError(
                      f"Directory '{self.step.plan.my_run.tree.root}' "
                      f"is not a git repository.")
              try:
                  self.extract_distgit_source(
                      git_root, sourcedir, self.get('dist-git-type'))
              except Exception as error:
                  raise tmt.utils.DiscoverError(
                      "Failed to process 'dist-git-source'.", original=error)
  
          # Use a tmt.Tree to apply possible command line filters
          tests = tmt.Tree(tree=tests).tests(conditions=["manual is False"])
          self._tests = tests
  
      def tests(self) -> List[tmt.base.Test]:
          return self._tests
  
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/discover/shell.py
files.old_contents: import copy
  import os
  from typing import TYPE_CHECKING, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.discover
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  @tmt.steps.provides_method('shell')
  class DiscoverShell(tmt.steps.discover.DiscoverPlugin):
      """
      Use provided list of shell script tests
  
      List of test cases to be executed can be defined manually directly
      in the plan as a list of dictionaries containing test name, actual
      test script and optionally a path to the test. Example config:
  
      discover:
          how: shell
          tests:
          - name: /help/main
            test: tmt --help
          - name: /help/test
            test: tmt test --help
          - name: /help/smoke
            test: ./smoke.sh
            path: /tests/shell
  
      For DistGit repo one can extract source tarball and use its code.
      It is extracted to TMT_SOURCE_DIR however no patches are applied
      (only source tarball is extracted).
  
      discover:
          how: shell
          dist-git-source: true
          tests:
          - name: /upstream
            test: cd $TMT_SOURCE_DIR/*/tests && make test
      """
  
      def show(self, keys: Optional[List[str]] = None) -> None:
          """ Show config details """
          super().show([])
          tests = self.get('tests')
          if tests:
              test_names = [test['name'] for test in tests]
              click.echo(tmt.utils.format('tests', test_names))
  
      def wake(self, keys: Optional[List[str]] = None) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys)
          # Check provided tests, default to an empty list
          if 'tests' not in self.data:
              self.data['tests'] = []
          self._tests: List[tmt.base.Test] = []
  
      def go(self) -> None:
          """ Discover available tests """
          super(DiscoverShell, self).go()
          tests = fmf.Tree(dict(summary='tests'))
  
          # dist-git related
          assert self.workdir is not None
          sourcedir = os.path.join(self.workdir, 'source')
          dist_git_source = self.get('dist-git-source', False)
  
          # Check and process each defined shell test
          assert self.data['tests'] is not None
          for data in self.data['tests']:
              # Create data copy (we want to keep original data for save()
              data = copy.deepcopy(data)
              # Extract name, make sure it is present
              try:
                  name = data.pop('name')
              except KeyError:
                  raise tmt.utils.SpecificationError(
                      f"Missing test name in '{self.step.plan.name}'.")
              # Make sure that the test script is defined
              if 'test' not in data:
                  raise tmt.utils.SpecificationError(
                      f"Missing test script in '{self.step.plan.name}'.")
              # Prepare path to the test working directory (tree root by default)
              try:
                  data['path'] = f"/tests{data['path']}"
              except KeyError:
                  data['path'] = "/tests"
              # Apply default test duration unless provided
              if 'duration' not in data:
                  data['duration'] = tmt.base.DEFAULT_TEST_DURATION_L2
              # Add source dir path variable
              if dist_git_source:
                  data.setdefault('environment', {})[
                      'TMT_SOURCE_DIR'] = sourcedir
              # Create a simple fmf node, adjust its name
              tests.child(name, data)
  
          # Symlink tests directory to the plan work tree
          testdir = os.path.join(self.workdir, "tests")
          relative_path = os.path.relpath(self.step.plan.worktree, self.workdir)
          os.symlink(relative_path, testdir)
  
          if dist_git_source:
              try:
                  run_result = self.run(
                      ["git", "rev-parse", "--show-toplevel"],
                      cwd=self.step.plan.my_run.tree.root,
                      dry=True)[0]
                  assert run_result is not None
                  git_root = run_result.strip('\n')
              except tmt.utils.RunError:
                  raise tmt.utils.DiscoverError(
                      f"Directory '{self.step.plan.my_run.tree.root}' "
                      f"is not a git repository.")
              try:
                  self.extract_distgit_source(
                      git_root, sourcedir, self.get('dist-git-type'))
              except Exception as error:
                  raise tmt.utils.DiscoverError(
                      "Failed to process 'dist-git-source'.", original=error)
  
          # Use a tmt.Tree to apply possible command line filters
          tests = tmt.Tree(tree=tests).tests(conditions=["manual is False"])
          self._tests = tests
  
      def tests(self) -> List[tmt.base.Test]:
          return self._tests
  
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/discover/shell.py
files.new_contents: import json
  import os
  import sys
  import time
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.execute
  import tmt.utils
  from tmt.steps.execute import TEST_OUTPUT_FILENAME, Script
  
  # Script handling reboots, in restraint compatible fashion
  TMT_REBOOT_SCRIPT = Script("/usr/local/bin/tmt-reboot",
                             aliases=[
                                 "/usr/local/bin/rstrnt-reboot",
                                 "/usr/local/bin/rhts-reboot"],
                             related_variables=[
                                 "TMT_REBOOT_COUNT",
                                 "REBOOTCOUNT",
                                 "RSTRNT_REBOOTCOUNT"]
                             )
  
  # Script handling result reporting, in restraint compatible fashion
  TMT_REPORT_RESULT_SCRIPT = Script("/usr/local/bin/tmt-report-result",
                                    aliases=[
                                        "/usr/local/bin/rstrnt-report-result",
                                        "/usr/local/bin/rhts-report-result"],
                                    related_variables=[]
                                    )
  
  # Script for archiving a file, usable for BEAKERLIB_COMMAND_SUBMIT_LOG
  TMT_FILE_SUBMIT_SCRIPT = Script("/usr/local/bin/tmt-file-submit",
                                  aliases=[
                                      "/usr/local/bin/rstrnt-report-log",
                                      "/usr/local/bin/rhts-submit-log",
                                      "/usr/local/bin/rhts_submit_log"],
                                  related_variables=[]
                                  )
  
  # Script handling text execution abortion, in restraint compatible fashion
  TMT_ABORT_SCRIPT = Script("/usr/local/bin/tmt-abort",
                            aliases=[
                                "/usr/local/bin/rstrnt-abort",
                                "/usr/local/bin/rhts-abort"],
                            related_variables=[]
                            )
  
  # File for requesting reboot
  REBOOT_REQUEST_FILENAME = "reboot_request"
  
  # List of all available scripts
  SCRIPTS = (TMT_FILE_SUBMIT_SCRIPT,
             TMT_REBOOT_SCRIPT,
             TMT_REPORT_RESULT_SCRIPT,
             TMT_ABORT_SCRIPT
             )
  
  
  @tmt.steps.provides_method('tmt')
  @tmt.steps.provides_method('shell.tmt', order=80)
  @tmt.steps.provides_method('beakerlib.tmt', order=80)
  class ExecuteInternal(tmt.steps.execute.ExecutePlugin):
      """
      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).
      """
  
      # Supported keys
      _keys = ["script", "interactive"]
  
      def __init__(self, *args, **kwargs):
          super().__init__(*args, **kwargs)
          self._previous_progress_message = ""
          self.scripts = SCRIPTS
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          options = []
          # Shell script as a test
          options.append(click.option(
              '-s', '--script', metavar='SCRIPT', multiple=True,
              help='Shell script to be executed as a test.'))
          # Interactive mode
          options.append(click.option(
              '-i', '--interactive', is_flag=True,
              help='Run in interactive mode, do not capture output.'))
          # Disable interactive progress bar
          options.append(click.option(
              '--no-progress-bar', is_flag=True,
              help='Disable interactive progress bar showing the current test.'))
          return options + super().options(how)
  
      def wake(self):
          """ Wake up the plugin, process data, apply options """
          super().wake()
          # Make sure that script is a list
          tmt.utils.listify(self.data, keys=['script'])
  
      # TODO: consider switching to utils.updatable_message() - might need more
      # work, since use of _show_progress is split over several methods.
      def _show_progress(self, progress, test_name, finish=False):
          """
          Show an interactive progress bar in non-verbose mode.
  
          If the output is not an interactive terminal, or progress bar is
          disabled using an option, just output the message as info without
          utilising \r. If finish is True, overwrite the previous progress bar.
          """
          # Verbose mode outputs other information, using \r to
          # create a status bar wouldn't work.
          if self.opt('verbose'):
              return
  
          # No progress if terminal not attached or explicitly disabled
          if not sys.stdout.isatty() or self.opt('no-progress-bar'):
              return
  
          # For debug mode show just an info message (unless finishing)
          message = f"{test_name} [{progress}]" if not finish else ""
          if self.opt('debug'):
              if not finish:
                  self.info(message, shift=1)
              return
  
          # Show progress bar in an interactive shell.
          # We need to completely override the previous message, add
          # spaces if necessary.
          message = message.ljust(len(self._previous_progress_message))
          self._previous_progress_message = message
          message = self._indent('progress', message, color='cyan')
          sys.stdout.write(f"\r{message}")
          if finish:
              # The progress has been overwritten, return back to the start
              sys.stdout.write("\r")
              self._previous_progress_message = ""
          sys.stdout.flush()
  
      def _test_environment(self, test, extra_environment):
          """ Return test environment """
          data_directory = self.data_path(test, full=True, create=True)
  
          environment = extra_environment.copy()
          environment.update(test.environment)
          environment["TMT_TREE"] = self.parent.plan.worktree
          environment["TMT_TEST_DATA"] = os.path.join(
              data_directory, tmt.steps.execute.TEST_DATA)
          environment["TMT_REBOOT_REQUEST"] = os.path.join(
              data_directory,
              tmt.steps.execute.TEST_DATA,
              REBOOT_REQUEST_FILENAME)
          # Set all supported reboot variables
          for reboot_variable in TMT_REBOOT_SCRIPT.related_variables:
              environment[reboot_variable] = str(test._reboot_count)
          # Variables related to beakerlib tests
          if test.framework == 'beakerlib':
              environment['BEAKERLIB_DIR'] = data_directory
              environment['BEAKERLIB_COMMAND_SUBMIT_LOG'] = (
                  f"bash {TMT_FILE_SUBMIT_SCRIPT.path}")
  
          return environment
  
      def execute(self, test, guest, extra_environment):
          """ Run test on the guest """
          self.debug(f"Execute '{test.name}' as a '{test.framework}' test.")
  
          # Test will be executed in it's own directory, relative to the workdir
          workdir = os.path.join(self.discover.workdir, test.path.lstrip('/'))
          self.debug(f"Use workdir '{workdir}'.", level=3)
  
          # Create data directory, prepare test environment
          environment = self._test_environment(test, extra_environment)
  
          # Prepare the test command (use default options for shell tests)
          if test.framework == "shell":
              command = f"{tmt.utils.SHELL_OPTIONS}; {test.test}"
          else:
              command = test.test
  
          # Prepare custom function to log output in verbose mode
          def log(key, value=None, color=None, shift=1, level=1):
              self.verbose(key, value, color, shift=2, level=3)
  
          # Execute the test, save the output and return code
          start = time.time()
          try:
              stdout, stderr = guest.execute(
                  command, cwd=workdir, env=environment,
                  join=True, interactive=self.get('interactive'), log=log,
                  timeout=tmt.utils.duration_to_seconds(test.duration))
              test.returncode = 0
          except tmt.utils.RunError as error:
              stdout = error.stdout
              test.returncode = error.returncode
              if test.returncode == tmt.utils.PROCESS_TIMEOUT:
                  self.debug(f"Test duration '{test.duration}' exceeded.")
          end = time.time()
          self.write(
              self.data_path(test, TEST_OUTPUT_FILENAME, full=True),
              stdout or '', mode='a', level=3)
          test.real_duration = self.test_duration(start, end)
  
      def check(self, test):
          """ Check the test result """
          self.debug(f"Check result of '{test.name}'.")
          if test.framework == 'beakerlib':
              return self.check_beakerlib(test)
          else:
              try:
                  return self.check_result_file(test)
              except tmt.utils.FileError:
                  return self.check_shell(test)
  
      def _will_reboot(self, test):
          """ True if reboot is requested """
          return os.path.exists(self._reboot_request_path(test))
  
      def _reboot_request_path(self, test):
          """ Return reboot_request """
          reboot_request_path = os.path.join(
              self.data_path(test, full=True),
              tmt.steps.execute.TEST_DATA,
              REBOOT_REQUEST_FILENAME)
          return reboot_request_path
  
      def _handle_reboot(self, test, guest):
          """
          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
          REBOOTCOUNT variable, reset it to 0 if no reboot was requested
          (going forward to the next test). Return whether reboot was done.
          """
          if self._will_reboot(test):
              test._reboot_count += 1
              self.debug(f"Reboot during test '{test}' "
                         f"with reboot count {test._reboot_count}.")
              reboot_request_path = self._reboot_request_path(test)
              test_data = os.path.join(
                  self.data_path(test, full=True),
                  tmt.steps.execute.TEST_DATA)
              with open(reboot_request_path, 'r') as reboot_file:
                  reboot_data = json.loads(reboot_file.read())
              reboot_command = reboot_data.get('command')
              try:
                  timeout = int(reboot_data.get('timeout'))
              except ValueError:
                  timeout = None
              # Reset the file
              os.remove(reboot_request_path)
              guest.push(test_data)
              try:
                  guest.reboot(command=reboot_command,
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/execute/internal.py
files.old_contents: import json
  import os
  import sys
  import time
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.execute
  import tmt.utils
  from tmt.steps.execute import TEST_OUTPUT_FILENAME, Script
  
  # Script handling reboots, in restraint compatible fashion
  TMT_REBOOT_SCRIPT = Script("/usr/local/bin/tmt-reboot",
                             aliases=[
                                 "/usr/local/bin/rstrnt-reboot",
                                 "/usr/local/bin/rhts-reboot"],
                             related_variables=[
                                 "TMT_REBOOT_COUNT",
                                 "REBOOTCOUNT",
                                 "RSTRNT_REBOOTCOUNT"]
                             )
  
  # Script handling result reporting, in restraint compatible fashion
  TMT_REPORT_RESULT_SCRIPT = Script("/usr/local/bin/tmt-report-result",
                                    aliases=[
                                        "/usr/local/bin/rstrnt-report-result",
                                        "/usr/local/bin/rhts-report-result"],
                                    related_variables=[]
                                    )
  
  # Script for archiving a file, usable for BEAKERLIB_COMMAND_SUBMIT_LOG
  TMT_FILE_SUBMIT_SCRIPT = Script("/usr/local/bin/tmt-file-submit",
                                  aliases=[
                                      "/usr/local/bin/rstrnt-report-log",
                                      "/usr/local/bin/rhts-submit-log",
                                      "/usr/local/bin/rhts_submit_log"],
                                  related_variables=[]
                                  )
  
  # Script handling text execution abortion, in restraint compatible fashion
  TMT_ABORT_SCRIPT = Script("/usr/local/bin/tmt-abort",
                            aliases=[
                                "/usr/local/bin/rstrnt-abort",
                                "/usr/local/bin/rhts-abort"],
                            related_variables=[]
                            )
  
  # File for requesting reboot
  REBOOT_REQUEST_FILENAME = "reboot_request"
  
  # List of all available scripts
  SCRIPTS = (TMT_FILE_SUBMIT_SCRIPT,
             TMT_REBOOT_SCRIPT,
             TMT_REPORT_RESULT_SCRIPT,
             TMT_ABORT_SCRIPT
             )
  
  
  @tmt.steps.provides_method('tmt')
  @tmt.steps.provides_method('shell.tmt', order=80)
  @tmt.steps.provides_method('beakerlib.tmt', order=80)
  class ExecuteInternal(tmt.steps.execute.ExecutePlugin):
      """
      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).
      """
  
      # Supported keys
      _keys = ["script", "interactive"]
  
      def __init__(self, *args, **kwargs):
          super().__init__(*args, **kwargs)
          self._previous_progress_message = ""
          self.scripts = SCRIPTS
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          options = []
          # Shell script as a test
          options.append(click.option(
              '-s', '--script', metavar='SCRIPT', multiple=True,
              help='Shell script to be executed as a test.'))
          # Interactive mode
          options.append(click.option(
              '-i', '--interactive', is_flag=True,
              help='Run in interactive mode, do not capture output.'))
          # Disable interactive progress bar
          options.append(click.option(
              '--no-progress-bar', is_flag=True,
              help='Disable interactive progress bar showing the current test.'))
          return options + super().options(how)
  
      def wake(self, keys=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys)
          # Make sure that script is a list
          tmt.utils.listify(self.data, keys=['script'])
  
      # TODO: consider switching to utils.updatable_message() - might need more
      # work, since use of _show_progress is split over several methods.
      def _show_progress(self, progress, test_name, finish=False):
          """
          Show an interactive progress bar in non-verbose mode.
  
          If the output is not an interactive terminal, or progress bar is
          disabled using an option, just output the message as info without
          utilising \r. If finish is True, overwrite the previous progress bar.
          """
          # Verbose mode outputs other information, using \r to
          # create a status bar wouldn't work.
          if self.opt('verbose'):
              return
  
          # No progress if terminal not attached or explicitly disabled
          if not sys.stdout.isatty() or self.opt('no-progress-bar'):
              return
  
          # For debug mode show just an info message (unless finishing)
          message = f"{test_name} [{progress}]" if not finish else ""
          if self.opt('debug'):
              if not finish:
                  self.info(message, shift=1)
              return
  
          # Show progress bar in an interactive shell.
          # We need to completely override the previous message, add
          # spaces if necessary.
          message = message.ljust(len(self._previous_progress_message))
          self._previous_progress_message = message
          message = self._indent('progress', message, color='cyan')
          sys.stdout.write(f"\r{message}")
          if finish:
              # The progress has been overwritten, return back to the start
              sys.stdout.write("\r")
              self._previous_progress_message = ""
          sys.stdout.flush()
  
      def _test_environment(self, test, extra_environment):
          """ Return test environment """
          data_directory = self.data_path(test, full=True, create=True)
  
          environment = extra_environment.copy()
          environment.update(test.environment)
          environment["TMT_TREE"] = self.parent.plan.worktree
          environment["TMT_TEST_DATA"] = os.path.join(
              data_directory, tmt.steps.execute.TEST_DATA)
          environment["TMT_REBOOT_REQUEST"] = os.path.join(
              data_directory,
              tmt.steps.execute.TEST_DATA,
              REBOOT_REQUEST_FILENAME)
          # Set all supported reboot variables
          for reboot_variable in TMT_REBOOT_SCRIPT.related_variables:
              environment[reboot_variable] = str(test._reboot_count)
          # Variables related to beakerlib tests
          if test.framework == 'beakerlib':
              environment['BEAKERLIB_DIR'] = data_directory
              environment['BEAKERLIB_COMMAND_SUBMIT_LOG'] = (
                  f"bash {TMT_FILE_SUBMIT_SCRIPT.path}")
  
          return environment
  
      def execute(self, test, guest, extra_environment):
          """ Run test on the guest """
          self.debug(f"Execute '{test.name}' as a '{test.framework}' test.")
  
          # Test will be executed in it's own directory, relative to the workdir
          workdir = os.path.join(self.discover.workdir, test.path.lstrip('/'))
          self.debug(f"Use workdir '{workdir}'.", level=3)
  
          # Create data directory, prepare test environment
          environment = self._test_environment(test, extra_environment)
  
          # Prepare the test command (use default options for shell tests)
          if test.framework == "shell":
              command = f"{tmt.utils.SHELL_OPTIONS}; {test.test}"
          else:
              command = test.test
  
          # Prepare custom function to log output in verbose mode
          def log(key, value=None, color=None, shift=1, level=1):
              self.verbose(key, value, color, shift=2, level=3)
  
          # Execute the test, save the output and return code
          start = time.time()
          try:
              stdout, stderr = guest.execute(
                  command, cwd=workdir, env=environment,
                  join=True, interactive=self.get('interactive'), log=log,
                  timeout=tmt.utils.duration_to_seconds(test.duration))
              test.returncode = 0
          except tmt.utils.RunError as error:
              stdout = error.stdout
              test.returncode = error.returncode
              if test.returncode == tmt.utils.PROCESS_TIMEOUT:
                  self.debug(f"Test duration '{test.duration}' exceeded.")
          end = time.time()
          self.write(
              self.data_path(test, TEST_OUTPUT_FILENAME, full=True),
              stdout or '', mode='a', level=3)
          test.real_duration = self.test_duration(start, end)
  
      def check(self, test):
          """ Check the test result """
          self.debug(f"Check result of '{test.name}'.")
          if test.framework == 'beakerlib':
              return self.check_beakerlib(test)
          else:
              try:
                  return self.check_result_file(test)
              except tmt.utils.FileError:
                  return self.check_shell(test)
  
      def _will_reboot(self, test):
          """ True if reboot is requested """
          return os.path.exists(self._reboot_request_path(test))
  
      def _reboot_request_path(self, test):
          """ Return reboot_request """
          reboot_request_path = os.path.join(
              self.data_path(test, full=True),
              tmt.steps.execute.TEST_DATA,
              REBOOT_REQUEST_FILENAME)
          return reboot_request_path
  
      def _handle_reboot(self, test, guest):
          """
          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
          REBOOTCOUNT variable, reset it to 0 if no reboot was requested
          (going forward to the next test). Return whether reboot was done.
          """
          if self._will_reboot(test):
              test._reboot_count += 1
              self.debug(f"Reboot during test '{test}' "
                         f"with reboot count {test._reboot_count}.")
              reboot_request_path = self._reboot_request_path(test)
              test_data = os.path.join(
                  self.data_path(test, full=True),
                  tmt.steps.execute.TEST_DATA)
              with open(reboot_request_path, 'r') as reboot_file:
                  reboot_data = json.loads(reboot_file.read())
              reboot_command = reboot_data.get('command')
              try:
                  timeout = int(reboot_data.get('timeout'))
              except ValueError:
                  timeout = None
              # Reset the file
              os.remove(reboot_request_path)
              guest.push(test_data)
              try:
                  guest.re
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/execute/internal.py
files.new_contents: from typing import Any, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.finish
  from tmt.steps.provision import Guest
  
  
  @tmt.steps.provides_method('shell')
  class FinishShell(tmt.steps.finish.FinishPlugin):
      """
      Perform finishing tasks using shell scripts
  
      Example config:
  
      finish:
          how: shell
          script:
              - upload-logs.sh || true
              - rm -rf /tmp/temporary-files
  
      Use the 'order' attribute to select in which order finishing tasks
      should happen if there are multiple configs. Default order is '50'.
      """
  
      # Supported keys
      _keys = ["script"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Finish command line options """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '-s', '--script', metavar='SCRIPT',
                  help='Shell script to be executed.')
              ]
          return options
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'script':
              return []
          return default
  
      # TODO: use better types once superclass gains its annotations
      def wake(self) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake()
  
          # Convert to list if single script provided
          tmt.utils.listify(self.data, keys=['script'])
  
      def go(self, guest: Guest) -> None:
          """ Perform finishing tasks on given guest """
          super().go(guest)
  
          # Give a short summary
          scripts = self.get('script')
          overview = fmf.utils.listed(scripts, 'script')
          self.info('overview', f'{overview} found', 'green')
  
          # Execute each script on the guest
          for script in scripts:
              self.verbose('script', script, 'green')
              guest.execute(script, cwd=self.step.plan.worktree)
  
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/finish/shell.py
files.old_contents: from typing import Any, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.finish
  from tmt.steps.provision import Guest
  
  
  @tmt.steps.provides_method('shell')
  class FinishShell(tmt.steps.finish.FinishPlugin):
      """
      Perform finishing tasks using shell scripts
  
      Example config:
  
      finish:
          how: shell
          script:
              - upload-logs.sh || true
              - rm -rf /tmp/temporary-files
  
      Use the 'order' attribute to select in which order finishing tasks
      should happen if there are multiple configs. Default order is '50'.
      """
  
      # Supported keys
      _keys = ["script"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Finish command line options """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '-s', '--script', metavar='SCRIPT',
                  help='Shell script to be executed.')
              ]
          return options
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'script':
              return []
          return default
  
      # TODO: use better types once superclass gains its annotations
      def wake(self, keys: Optional[List[str]] = None) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys)
  
          # Convert to list if single script provided
          tmt.utils.listify(self.data, keys=['script'])
  
      def go(self, guest: Guest) -> None:
          """ Perform finishing tasks on given guest """
          super().go(guest)
  
          # Give a short summary
          scripts = self.get('script')
          overview = fmf.utils.listed(scripts, 'script')
          self.info('overview', f'{overview} found', 'green')
  
          # Execute each script on the guest
          for script in scripts:
              self.verbose('script', script, 'green')
              guest.execute(script, cwd=self.step.plan.worktree)
  
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/finish/shell.py
files.new_contents: import sys
  from typing import Any, List, Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  import tmt.utils
  from tmt.steps import Step
  from tmt.steps.provision import Guest
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  StepDataType = TypedDict(
      'StepDataType',
      {
          'playbook': List[str],
          'playbooks': List[str]
          }
      )
  
  
  @tmt.steps.provides_method('ansible')
  class PrepareAnsible(tmt.steps.prepare.PreparePlugin):  # type: ignore[misc]
      """
      Prepare guest using ansible
  
      Single playbook config:
  
          prepare:
              how: ansible
              playbook: ansible/packages.yml
  
      Multiple playbooks config:
  
          prepare:
              how: ansible
              playbook:
                - playbook/one.yml
                - playbook/two.yml
                - playbook/three.yml
              extra-args: '-vvv'
  
      The playbook path should be relative to the metadata tree root.
      Use 'order' attribute to select in which order preparation should
      happen if there are multiple configs. Default order is '50'.
      Default order of required packages installation is '70'.
      """
  
      # Supported keys
      _keys = ["playbook", "extra-args"]
  
      def __init__(self, step: Step, data: StepDataType) -> None:
          """ Store plugin name, data and parent step """
          super().__init__(step, data)
          # Rename plural playbooks to singular
          if 'playbooks' in self.data:
              self.data['playbook'] = self.data.pop('playbooks')
  
      # TODO: fix types once superclass gains its annotations
      @classmethod
      def options(cls, how: Optional[str] = None) -> Any:
          """ Prepare command line options """
          return [
              click.option(
                  '-p', '--playbook', metavar='PLAYBOOK', multiple=True,
                  help='Path to an ansible playbook to run.'),
              click.option(
                  '--extra-args', metavar='EXTRA-ARGS',
                  help='Optional arguments for ansible-playbook.')
              ] + super().options(how)
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'playbook':
              return []
          return default
  
      # TODO: use better types once superclass gains its annotations
      def wake(self) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake()
  
          # Convert to list if necessary
          tmt.utils.listify(self.data, keys=['playbook'])
  
      def go(self, guest: Guest) -> None:
          """ Prepare the guests """
          super().go(guest)
  
          # Apply each playbook on the guest
          for playbook in self.get('playbook'):
              self.info('playbook', playbook, 'green')
              guest.ansible(playbook, self.get('extra-args'))
  
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/prepare/ansible.py
files.old_contents: import sys
  from typing import Any, List, Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  import tmt.utils
  from tmt.steps import Step
  from tmt.steps.provision import Guest
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  StepDataType = TypedDict(
      'StepDataType',
      {
          'playbook': List[str],
          'playbooks': List[str]
          }
      )
  
  
  @tmt.steps.provides_method('ansible')
  class PrepareAnsible(tmt.steps.prepare.PreparePlugin):  # type: ignore[misc]
      """
      Prepare guest using ansible
  
      Single playbook config:
  
          prepare:
              how: ansible
              playbook: ansible/packages.yml
  
      Multiple playbooks config:
  
          prepare:
              how: ansible
              playbook:
                - playbook/one.yml
                - playbook/two.yml
                - playbook/three.yml
              extra-args: '-vvv'
  
      The playbook path should be relative to the metadata tree root.
      Use 'order' attribute to select in which order preparation should
      happen if there are multiple configs. Default order is '50'.
      Default order of required packages installation is '70'.
      """
  
      # Supported keys
      _keys = ["playbook", "extra-args"]
  
      def __init__(self, step: Step, data: StepDataType) -> None:
          """ Store plugin name, data and parent step """
          super().__init__(step, data)
          # Rename plural playbooks to singular
          if 'playbooks' in self.data:
              self.data['playbook'] = self.data.pop('playbooks')
  
      # TODO: fix types once superclass gains its annotations
      @classmethod
      def options(cls, how: Optional[str] = None) -> Any:
          """ Prepare command line options """
          return [
              click.option(
                  '-p', '--playbook', metavar='PLAYBOOK', multiple=True,
                  help='Path to an ansible playbook to run.'),
              click.option(
                  '--extra-args', metavar='EXTRA-ARGS',
                  help='Optional arguments for ansible-playbook.')
              ] + super().options(how)
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'playbook':
              return []
          return default
  
      # TODO: use better types once superclass gains its annotations
      def wake(self, keys: Optional[List[str]] = None) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys)
  
          # Convert to list if necessary
          tmt.utils.listify(self.data, keys=['playbook'])
  
      def go(self, guest: Guest) -> None:
          """ Prepare the guests """
          super().go(guest)
  
          # Apply each playbook on the guest
          for playbook in self.get('playbook'):
              self.info('playbook', playbook, 'green')
              guest.ansible(playbook, self.get('extra-args'))
  
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/prepare/ansible.py
files.new_contents: import os
  import re
  import shutil
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  
  COPR_URL = 'https://copr.fedorainfracloud.org/coprs'
  
  
  class InstallBase(tmt.utils.Common):
      """ Base class for installation implementations """
  
      # Each installer knows its package manager and copr plugin
      package_manager = None
      copr_plugin = None
  
      def __init__(self, parent, guest):
          """ Initialize installation data """
          super().__init__(parent=parent, relative_indent=0)
          self.guest = guest
  
          # Get package related data from the plugin
          self.packages = self.parent.get("package", [])
          self.directories = self.parent.get("directory", [])
          self.exclude = self.parent.get("exclude", [])
          if not self.packages and not self.directories:
              self.debug("No packages for installation found.", level=3)
  
          # Prepare package lists and installation command
          self.prepare_packages()
          self.prepare_sudo()
          self.prepare_command()
  
      def prepare_packages(self):
          """ Process package names and directories """
          self.local_packages = []
          self.debuginfo_packages = []
          self.repository_packages = []
  
          # Detect local, debuginfo and repository packages
          for package in self.packages:
              if package.endswith('.rpm'):
                  self.local_packages.append(package)
              elif re.search(r"-debug(info|source)(\.|$)", package):
                  # Strip the '-debuginfo' string from package name
                  # (installing with it doesn't work on RHEL7)
                  package = re.sub(r"-debuginfo((?=\.)|$)", "", package)
                  self.debuginfo_packages.append(package)
              else:
                  self.repository_packages.append(package)
  
          # Check rpm packages in local directories
          for directory in self.directories:
              self.info('directory', directory, 'green')
              if not os.path.isdir(directory):
                  raise tmt.utils.PrepareError(f"Packages directory '{directory}' not found.")
              for filename in os.listdir(directory):
                  if filename.endswith('.rpm'):
                      self.debug(f"Found rpm '{filename}'.", level=3)
                      self.local_packages.append(os.path.join(directory, filename))
  
      def prepare_sudo(self):
          """ Check if sudo is needed for installation """
          self.debug('Check if sudo is necessary.', level=2)
          user = self.guest.execute('whoami')[0].strip()
          self.sudo = '' if user == 'root' else 'sudo '
  
      def prepare_command(self):
          """ Prepare installation command"""
          raise NotImplementedError
  
      def prepare_repository(self, **kwargs):
          """ Configure additional repository """
          raise NotImplementedError
  
      def list_packages(self, package_list, title):
          """ Show package info and return quoted package names """
          # Show a brief summary by default
          if not self.opt('verbose'):
              summary = fmf.utils.listed(package_list, max=3)
              self.info(title, summary, 'green')
          # Provide a full list of packages in verbose mode
          else:
              summary = fmf.utils.listed(package_list, 'package')
              self.info(title, summary + ' requested', 'green')
              for package in sorted(package_list):
                  self.verbose(package, shift=1)
          # Return quoted package names
          return " ".join([tmt.utils.quote(package) for package in package_list])
  
      def enable_copr_epel6(self, copr):
          """ Manually enable copr repositories for epel6 """
          # Parse the copr repo name
          matched = re.match("^(@)?([^/]+)/([^/]+)$", copr)
          if not matched:
              raise tmt.utils.PrepareError(f"Invalid copr repository '{copr}'.")
          group, name, project = matched.groups()
          group = 'group_' if group else ''
          # Prepare the repo file url
          parts = [COPR_URL] + (['g'] if group else [])
          parts += [name, project, 'repo', 'epel-6']
          parts += [f"{group}{name}-{project}-epel-6.repo"]
          url = '/'.join(parts)
          # Download the repo file on guest
          try:
              self.guest.execute(f'curl -LOf {url}', cwd='/etc/yum.repos.d')
          except tmt.utils.RunError as error:
              if 'not found' in error.stderr.lower():
                  raise tmt.utils.PrepareError(
                      f"Copr repository '{copr}' not found.")
              raise
  
      def enable_copr(self):
          """ Enable requested copr repositories """
          coprs = self.parent.get('copr')
          if not coprs:
              return
          # Try to install copr plugin
          self.debug('Make sure the copr plugin is available.')
          try:
              self.guest.execute(
                  f'rpm -q {self.copr_plugin} || {self.command} install -y {self.copr_plugin}')
          # Enable repositories manually for epel6
          except tmt.utils.RunError:
              for copr in coprs:
                  self.info('copr', copr, 'green')
                  self.enable_copr_epel6(copr)
          # Enable repositories using copr plugin
          else:
              for copr in coprs:
                  self.info('copr', copr, 'green')
                  self.guest.execute(f'{self.command} copr enable -y {copr}')
  
      def prepare_install_local(self):
          """ Copy packages to the test system """
          self.rpms_directory = os.path.join(self.parent.step.workdir, 'rpms')
          os.makedirs(self.rpms_directory)
  
          # Copy local packages into workdir, push to guests
          for package in self.local_packages:
              self.verbose(os.path.basename(package), shift=1)
              self.debug(f"Copy '{package}' to '{self.rpms_directory}'.", level=3)
              shutil.copy(package, self.rpms_directory)
          self.guest.push()
  
      def install_from_repository(self):
          """ Default base install method for packages from repositories """
          pass
  
      def install_local(self):
          """ Default base install method for local packages """
          pass
  
      def install_debuginfo(self):
          """ Default base install method for debuginfo packages """
          pass
  
      def install(self):
          """ Perform the actual installation """
          if self.local_packages:
              self.prepare_install_local()
              self.install_local()
          if self.repository_packages:
              self.install_from_repository()
          if self.debuginfo_packages:
              self.install_debuginfo()
  
  
  class InstallDnf(InstallBase):
      """ Install packages using dnf """
  
      package_manager = "dnf"
      copr_plugin = "dnf-plugins-core"
  
      def prepare_command(self):
          """ Prepare installation command """
          self.options = '-y'
          self.skip = ' --skip-broken' if self.parent.get('missing') == 'skip' else ''
          for package in self.exclude:
              self.options += " --exclude " + tmt.utils.quote(package)
  
          self.command = f"{self.sudo}{self.package_manager}{self.skip}"
          self.debug(f"Using '{self.command}' for all package operations.")
          self.debug(f"Options for package operations are '{self.options}'")
  
      def install_local(self):
          """ Install copied local packages """
          # Use both dnf install/reinstall to get all packages refreshed
          # FIXME Simplify this once BZ#1831022 is fixed/implemeted.
          self.guest.execute(f"{self.command} install {self.options} {self.rpms_directory}/*")
          self.guest.execute(f"{self.command} reinstall {self.options} {self.rpms_directory}/*")
          summary = fmf.utils.listed(self.local_packages, 'local package')
          self.info('total', f"{summary} installed", 'green')
  
      def install_from_repository(self):
          """ Install packages from the repository """
          packages = self.list_packages(self.repository_packages, title="package")
          check = f'rpm -q --whatprovides {packages}'
          # Check and install
          self.guest.execute(f"{check} || {self.command} install {self.options} {packages}")
  
      def install_debuginfo(self):
          """ Install debuginfo packages """
          packages = self.list_packages(self.debuginfo_packages, title="debuginfo")
          # Make sure debuginfo-install is present on the target system
          self.guest.execute(f"{self.command} install -y /usr/bin/debuginfo-install")
          self.guest.execute(f"debuginfo-install -y {packages}")
  
  
  class InstallYum(InstallDnf):
      """ Install packages using yum """
  
      package_manager = "yum"
      copr_plugin = "yum-plugin-copr"
  
      def install_from_repository(self):
          """ Install packages from the repository """
          packages = self.list_packages(self.repository_packages, title="package")
          # Extra ignore/check for yum to workaround BZ#1920176
          check = f'rpm -q --whatprovides {packages}'
          final_check = " || true" if self.skip else f" && {check}"
          # Check and install
          self.guest.execute(
              f"{check} || {self.command} install {self.options} {packages}{final_check}")
  
  
  class InstallRpmOstree(InstallBase):
      """ Install packages using rpm-ostree """
  
      package_manager = "rpm-ostree"
      copr_plugin = "dnf-plugins-core"
  
      def sort_packages(self):
          """ Identify required and recommended packages """
          self.recommended_packages = []
          self.required_packages = []
          for package in self.repository_packages:
              try:
                  output = self.guest.execute(f"rpm -q --whatprovides '{package}'")
                  self.debug(f"Package '{output.stdout.strip()}' already installed.")
              except tmt.utils.RunError:
                  if self.skip:
                      self.recommended_packages.append(package)
                  else:
                      self.required_packages.append(package)
  
      def prepare_command(self):
          """ Prepare installation command
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/prepare/install.py
files.old_contents: import os
  import re
  import shutil
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  
  COPR_URL = 'https://copr.fedorainfracloud.org/coprs'
  
  
  class InstallBase(tmt.utils.Common):
      """ Base class for installation implementations """
  
      # Each installer knows its package manager and copr plugin
      package_manager = None
      copr_plugin = None
  
      def __init__(self, parent, guest):
          """ Initialize installation data """
          super().__init__(parent=parent, relative_indent=0)
          self.guest = guest
  
          # Get package related data from the plugin
          self.packages = self.parent.get("package", [])
          self.directories = self.parent.get("directory", [])
          self.exclude = self.parent.get("exclude", [])
          if not self.packages and not self.directories:
              self.debug("No packages for installation found.", level=3)
  
          # Prepare package lists and installation command
          self.prepare_packages()
          self.prepare_sudo()
          self.prepare_command()
  
      def prepare_packages(self):
          """ Process package names and directories """
          self.local_packages = []
          self.debuginfo_packages = []
          self.repository_packages = []
  
          # Detect local, debuginfo and repository packages
          for package in self.packages:
              if package.endswith('.rpm'):
                  self.local_packages.append(package)
              elif re.search(r"-debug(info|source)(\.|$)", package):
                  # Strip the '-debuginfo' string from package name
                  # (installing with it doesn't work on RHEL7)
                  package = re.sub(r"-debuginfo((?=\.)|$)", "", package)
                  self.debuginfo_packages.append(package)
              else:
                  self.repository_packages.append(package)
  
          # Check rpm packages in local directories
          for directory in self.directories:
              self.info('directory', directory, 'green')
              if not os.path.isdir(directory):
                  raise tmt.utils.PrepareError(f"Packages directory '{directory}' not found.")
              for filename in os.listdir(directory):
                  if filename.endswith('.rpm'):
                      self.debug(f"Found rpm '{filename}'.", level=3)
                      self.local_packages.append(os.path.join(directory, filename))
  
      def prepare_sudo(self):
          """ Check if sudo is needed for installation """
          self.debug('Check if sudo is necessary.', level=2)
          user = self.guest.execute('whoami')[0].strip()
          self.sudo = '' if user == 'root' else 'sudo '
  
      def prepare_command(self):
          """ Prepare installation command"""
          raise NotImplementedError
  
      def prepare_repository(self, **kwargs):
          """ Configure additional repository """
          raise NotImplementedError
  
      def list_packages(self, package_list, title):
          """ Show package info and return quoted package names """
          # Show a brief summary by default
          if not self.opt('verbose'):
              summary = fmf.utils.listed(package_list, max=3)
              self.info(title, summary, 'green')
          # Provide a full list of packages in verbose mode
          else:
              summary = fmf.utils.listed(package_list, 'package')
              self.info(title, summary + ' requested', 'green')
              for package in sorted(package_list):
                  self.verbose(package, shift=1)
          # Return quoted package names
          return " ".join([tmt.utils.quote(package) for package in package_list])
  
      def enable_copr_epel6(self, copr):
          """ Manually enable copr repositories for epel6 """
          # Parse the copr repo name
          matched = re.match("^(@)?([^/]+)/([^/]+)$", copr)
          if not matched:
              raise tmt.utils.PrepareError(f"Invalid copr repository '{copr}'.")
          group, name, project = matched.groups()
          group = 'group_' if group else ''
          # Prepare the repo file url
          parts = [COPR_URL] + (['g'] if group else [])
          parts += [name, project, 'repo', 'epel-6']
          parts += [f"{group}{name}-{project}-epel-6.repo"]
          url = '/'.join(parts)
          # Download the repo file on guest
          try:
              self.guest.execute(f'curl -LOf {url}', cwd='/etc/yum.repos.d')
          except tmt.utils.RunError as error:
              if 'not found' in error.stderr.lower():
                  raise tmt.utils.PrepareError(
                      f"Copr repository '{copr}' not found.")
              raise
  
      def enable_copr(self):
          """ Enable requested copr repositories """
          coprs = self.parent.get('copr')
          if not coprs:
              return
          # Try to install copr plugin
          self.debug('Make sure the copr plugin is available.')
          try:
              self.guest.execute(
                  f'rpm -q {self.copr_plugin} || {self.command} install -y {self.copr_plugin}')
          # Enable repositories manually for epel6
          except tmt.utils.RunError:
              for copr in coprs:
                  self.info('copr', copr, 'green')
                  self.enable_copr_epel6(copr)
          # Enable repositories using copr plugin
          else:
              for copr in coprs:
                  self.info('copr', copr, 'green')
                  self.guest.execute(f'{self.command} copr enable -y {copr}')
  
      def prepare_install_local(self):
          """ Copy packages to the test system """
          self.rpms_directory = os.path.join(self.parent.step.workdir, 'rpms')
          os.makedirs(self.rpms_directory)
  
          # Copy local packages into workdir, push to guests
          for package in self.local_packages:
              self.verbose(os.path.basename(package), shift=1)
              self.debug(f"Copy '{package}' to '{self.rpms_directory}'.", level=3)
              shutil.copy(package, self.rpms_directory)
          self.guest.push()
  
      def install_from_repository(self):
          """ Default base install method for packages from repositories """
          pass
  
      def install_local(self):
          """ Default base install method for local packages """
          pass
  
      def install_debuginfo(self):
          """ Default base install method for debuginfo packages """
          pass
  
      def install(self):
          """ Perform the actual installation """
          if self.local_packages:
              self.prepare_install_local()
              self.install_local()
          if self.repository_packages:
              self.install_from_repository()
          if self.debuginfo_packages:
              self.install_debuginfo()
  
  
  class InstallDnf(InstallBase):
      """ Install packages using dnf """
  
      package_manager = "dnf"
      copr_plugin = "dnf-plugins-core"
  
      def prepare_command(self):
          """ Prepare installation command """
          self.options = '-y'
          self.skip = ' --skip-broken' if self.parent.get('missing') == 'skip' else ''
          for package in self.exclude:
              self.options += " --exclude " + tmt.utils.quote(package)
  
          self.command = f"{self.sudo}{self.package_manager}{self.skip}"
          self.debug(f"Using '{self.command}' for all package operations.")
          self.debug(f"Options for package operations are '{self.options}'")
  
      def install_local(self):
          """ Install copied local packages """
          # Use both dnf install/reinstall to get all packages refreshed
          # FIXME Simplify this once BZ#1831022 is fixed/implemeted.
          self.guest.execute(f"{self.command} install {self.options} {self.rpms_directory}/*")
          self.guest.execute(f"{self.command} reinstall {self.options} {self.rpms_directory}/*")
          summary = fmf.utils.listed(self.local_packages, 'local package')
          self.info('total', f"{summary} installed", 'green')
  
      def install_from_repository(self):
          """ Install packages from the repository """
          packages = self.list_packages(self.repository_packages, title="package")
          check = f'rpm -q --whatprovides {packages}'
          # Check and install
          self.guest.execute(f"{check} || {self.command} install {self.options} {packages}")
  
      def install_debuginfo(self):
          """ Install debuginfo packages """
          packages = self.list_packages(self.debuginfo_packages, title="debuginfo")
          # Make sure debuginfo-install is present on the target system
          self.guest.execute(f"{self.command} install -y /usr/bin/debuginfo-install")
          self.guest.execute(f"debuginfo-install -y {packages}")
  
  
  class InstallYum(InstallDnf):
      """ Install packages using yum """
  
      package_manager = "yum"
      copr_plugin = "yum-plugin-copr"
  
      def install_from_repository(self):
          """ Install packages from the repository """
          packages = self.list_packages(self.repository_packages, title="package")
          # Extra ignore/check for yum to workaround BZ#1920176
          check = f'rpm -q --whatprovides {packages}'
          final_check = " || true" if self.skip else f" && {check}"
          # Check and install
          self.guest.execute(
              f"{check} || {self.command} install {self.options} {packages}{final_check}")
  
  
  class InstallRpmOstree(InstallBase):
      """ Install packages using rpm-ostree """
  
      package_manager = "rpm-ostree"
      copr_plugin = "dnf-plugins-core"
  
      def sort_packages(self):
          """ Identify required and recommended packages """
          self.recommended_packages = []
          self.required_packages = []
          for package in self.repository_packages:
              try:
                  output = self.guest.execute(f"rpm -q --whatprovides '{package}'")
                  self.debug(f"Package '{output.stdout.strip()}' already installed.")
              except tmt.utils.RunError:
                  if self.skip:
                      self.recommended_packages.append(package)
                  else:
                      self.required_packages.append(package)
  
      def prepare_command(self):
          """ Prepare installation command
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/prepare/install.py
files.new_contents: from typing import Any, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  import tmt.utils
  from tmt.steps.provision import Guest
  
  
  # TODO: drop ignore once type annotations between modules enabled
  @tmt.steps.provides_method('shell')
  class PrepareShell(tmt.steps.prepare.PreparePlugin):  # type: ignore[misc]
      """
      Prepare guest using shell scripts
  
      Example config:
  
      prepare:
          how: shell
          script:
          - sudo dnf install -y 'dnf-command(copr)'
          - sudo dnf copr enable -y psss/tmt
          - sudo dnf install -y tmt
  
      Use 'order' attribute to select in which order preparation should
      happen if there are multiple configs. Default order is '50'.
      Default order of required packages installation is '70'.
      """
  
      # Supported keys
      _keys = ["script"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> Any:
          """ Prepare command line options """
          return [
              click.option(
                  '-s', '--script', metavar='SCRIPT',
                  help='Shell script to be executed.')
              ] + super().options(how)
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'script':
              return []
          return default
  
      def wake(self) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake()
  
          # Convert to list if single script provided
          tmt.utils.listify(self.data, keys=['script'])
  
      def go(self, guest: Guest) -> None:
          """ Prepare the guests """
          super().go(guest)
  
          # Give a short summary
          scripts = self.get('script')
          overview = fmf.utils.listed(scripts, 'script')
          self.info('overview', f'{overview} found', 'green')
  
          # Execute each script on the guest (with default shell options)
          for script in scripts:
              self.verbose('script', script, 'green')
              script_with_options = f'{tmt.utils.SHELL_OPTIONS}; {script}'
              guest.execute(script_with_options, cwd=self.step.plan.worktree)
  
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/prepare/shell.py
files.old_contents: from typing import Any, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  import tmt.utils
  from tmt.steps.provision import Guest
  
  
  # TODO: drop ignore once type annotations between modules enabled
  @tmt.steps.provides_method('shell')
  class PrepareShell(tmt.steps.prepare.PreparePlugin):  # type: ignore[misc]
      """
      Prepare guest using shell scripts
  
      Example config:
  
      prepare:
          how: shell
          script:
          - sudo dnf install -y 'dnf-command(copr)'
          - sudo dnf copr enable -y psss/tmt
          - sudo dnf install -y tmt
  
      Use 'order' attribute to select in which order preparation should
      happen if there are multiple configs. Default order is '50'.
      Default order of required packages installation is '70'.
      """
  
      # Supported keys
      _keys = ["script"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> Any:
          """ Prepare command line options """
          return [
              click.option(
                  '-s', '--script', metavar='SCRIPT',
                  help='Shell script to be executed.')
              ] + super().options(how)
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'script':
              return []
          return default
  
      def wake(self, keys: Optional[List[str]] = None) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys)
  
          # Convert to list if single script provided
          tmt.utils.listify(self.data, keys=['script'])
  
      def go(self, guest: Guest) -> None:
          """ Prepare the guests """
          super().go(guest)
  
          # Give a short summary
          scripts = self.get('script')
          overview = fmf.utils.listed(scripts, 'script')
          self.info('overview', f'{overview} found', 'green')
  
          # Execute each script on the guest (with default shell options)
          for script in scripts:
              self.verbose('script', script, 'green')
              script_with_options = f'{tmt.utils.SHELL_OPTIONS}; {script}'
              guest.execute(script_with_options, cwd=self.step.plan.worktree)
  
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/prepare/shell.py
files.new_contents: import collections
  import dataclasses
  import datetime
  import enum
  import os
  import random
  import re
  import shlex
  import string
  import subprocess
  import tempfile
  from shlex import quote
  from typing import Any, Dict, List, Optional, Type, Union, cast
  
  import click
  import fmf
  
  import tmt
  import tmt.plugins
  import tmt.steps
  import tmt.utils
  from tmt.steps import Action
  
  # Timeout in seconds of waiting for a connection after reboot
  CONNECTION_TIMEOUT = 5 * 60
  
  # When waiting for guest to recover from reboot, try re-connecting every
  # this many seconds.
  RECONNECT_WAIT_TICK = 5
  RECONNECT_WAIT_TICK_INCREASE = 1.0
  
  # Default rsync options
  DEFAULT_RSYNC_OPTIONS = [
      "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  DEFAULT_RSYNC_PUSH_OPTIONS = ["-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  DEFAULT_RSYNC_PULL_OPTIONS = ["-R", "-r", "-z", "--links", "--safe-links", "--protect-args"]
  
  
  class CheckRsyncOutcome(enum.Enum):
      ALREADY_INSTALLED = 'already-installed'
      INSTALLED = 'installed'
  
  
  class Provision(tmt.steps.Step):
      """ Provision an environment for testing or use localhost. """
  
      # Default implementation for provision is a virtual machine
      DEFAULT_HOW = 'virtual'
  
      def __init__(self, plan: 'tmt.Plan', data: tmt.steps.StepData) -> None:
          """ Initialize provision step data """
          super().__init__(plan=plan, data=data)
          # Check that the names are unique
          names = [data['name'] for data in self.data]
          names_count: Dict[str, int] = collections.defaultdict(int)
          for name in names:
              names_count[name] += 1
          if len(self.data) > 1 and len(self.data) != len(names_count):
              duplicate = [k for k, v in names_count.items() if v > 1]
              duplicate_string = ', '.join(duplicate)
              raise tmt.utils.GeneralError(
                  f"Provision step names must be unique for multihost testing. "
                  f"Duplicate names: {duplicate_string} in plan '{plan.name}'.")
          # List of provisioned guests and loaded guest data
          self._guests: List['Guest'] = []
          self._guest_data: Dict[str, 'GuestData'] = {}
          self.is_multihost = False
  
      def load(self) -> None:
          """ Load guest data from the workdir """
          super().load()
          try:
              raw_guest_data = tmt.utils.yaml_to_dict(self.read('guests.yaml'))
  
              self._guest_data = {
                  name: tmt.utils.SerializableContainer.unserialize(guest_data)
                  for name, guest_data in raw_guest_data.items()
                  }
  
          except tmt.utils.FileError:
              self.debug('Provisioned guests not found.', level=2)
  
      def save(self) -> None:
          """ Save guest data to the workdir """
          super().save()
          try:
              raw_guest_data = {guest.name: guest.save().to_serialized()
                                for guest in self.guests()}
  
              self.write('guests.yaml', tmt.utils.dict_to_yaml(raw_guest_data))
          except tmt.utils.FileError:
              self.debug('Failed to save provisioned guests.')
  
      def wake(self) -> None:
          """ Wake up the step (process workdir and command line) """
          super().wake()
  
          # Choose the right plugin and wake it up
          for data in self.data:
              # TODO: with generic BasePlugin, delegate() should return more fitting type,
              # not the base class.
              plugin = cast(ProvisionPlugin, ProvisionPlugin.delegate(self, data))
              self._phases.append(plugin)
              # If guest data loaded, perform a complete wake up
              plugin.wake(data=self._guest_data.get(plugin.name))
  
              guest = plugin.guest()
              if guest:
                  self._guests.append(guest)
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Provision wake up complete (already done before).', level=2)
          # Save status and step data (now we know what to do)
          else:
              self.status('todo')
              self.save()
  
      def show(self) -> None:
          """ Show discover details """
          for data in self.data:
              ProvisionPlugin.delegate(self, data).show()
  
      def summary(self) -> None:
          """ Give a concise summary of the provisioning """
          # Summary of provisioned guests
          guests = fmf.utils.listed(self.guests(), 'guest')
          self.info('summary', f'{guests} provisioned', 'green', shift=1)
          # Guest list in verbose mode
          for guest in self.guests():
              if not guest.name.startswith(tmt.utils.DEFAULT_NAME):
                  self.verbose(guest.name, color='red', shift=2)
  
      def go(self) -> None:
          """ Provision all guests"""
          super().go()
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.info('status', 'done', 'green', shift=1)
              self.summary()
              self.actions()
              return
  
          # Provision guests
          self._guests = []
          save = True
          self.is_multihost = sum([isinstance(phase, ProvisionPlugin)
                                  for phase in self.phases()]) > 1
          try:
              for phase in self.phases(classes=(Action, ProvisionPlugin)):
                  try:
                      if isinstance(phase, Action):
                          phase.go()
  
                      elif isinstance(phase, ProvisionPlugin):
                          phase.go()
  
                          guest = phase.guest()
                          if guest:
                              guest.details()
  
                      if self.is_multihost:
                          self.info('')
                  finally:
                      if isinstance(phase, ProvisionPlugin):
                          guest = phase.guest()
                          if guest:
                              self._guests.append(guest)
  
              # Give a summary, update status and save
              self.summary()
              self.status('done')
          except (SystemExit, tmt.utils.SpecificationError) as error:
              # A plugin will only raise SystemExit if the exit is really desired
              # and no other actions should be done. An example of this is
              # listing available images. In such case, the workdir is deleted
              # as it's redundant and save() would throw an error.
              save = False
              raise error
          finally:
              if save:
                  self.save()
  
      def guests(self) -> List['Guest']:
          """ Return the list of all provisioned guests """
          return self._guests
  
      def requires(self) -> List[str]:
          """
          Packages required by all enabled provision plugins
  
          Return a list of packages which need to be installed on the
          provisioned guest so that the workdir can be synced to it.
          Used by the prepare step.
          """
          requires = set()
          for plugin in self.phases(classes=ProvisionPlugin):
              requires.update(plugin.requires())
          return list(requires)
  
  
  class ProvisionPlugin(tmt.steps.GuestlessPlugin):
      """ Common parent of provision plugins """
  
      # Default implementation for provision is a virtual machine
      how = 'virtual'
  
      # List of all supported methods aggregated from all plugins of the same step.
      _supported_methods: List[tmt.steps.Method] = []
  
      # Common keys for all provision step implementations
      _common_keys = ['role']
  
      @classmethod
      def base_command(
              cls,
              usage: str,
              method_class: Optional[Type[click.Command]] = None) -> click.Command:
          """ Create base click command (common for all provision plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Provision.usage(method_overview=usage)
  
          # Create the command
          @click.command(cls=method_class, help=usage)
          @click.pass_context
          @click.option(
              '-h', '--how', metavar='METHOD',
              help='Use specified method for provisioning.')
          def provision(context: click.Context, **kwargs: Any) -> None:
              context.obj.steps.add('provision')
              Provision._save_context(context)
  
          return provision
  
      def wake(self, data: Optional['GuestData'] = None) -> None:
          """
          Wake up the plugin
  
          Override data with command line options.
          Wake up the guest based on provided guest data.
          """
          super().wake()
  
      def guest(self) -> Optional['Guest']:
          """
          Return provisioned guest
  
          Each ProvisionPlugin has to implement this method.
          Should return a provisioned Guest() instance.
          """
          raise NotImplementedError()
  
      def requires(self) -> List[str]:
          """ List of required packages needed for workdir sync """
          return Guest.requires()
  
      @classmethod
      def clean_images(cls, clean: 'tmt.base.Clean', dry: bool) -> None:
          """ Remove the images of one particular plugin """
  
  
  @dataclasses.dataclass
  class GuestData(tmt.utils.SerializableContainer):
      """
      Keys necessary to describe, create, save and restore a guest.
  
      Very basic set of keys shared across all known guest classes.
      """
  
      # guest role in the multihost scenario
      role: Optional[str] = None
      # hostname or ip address
      guest: Optional[str] = None
  
  
  class Guest(tmt.utils.Common):
      """
      Guest provisioned for test execution
  
      A base class for guest-like classes. Provides some of the basic methods
      and functionality, but note some of the methods are left intentionally
      empty. These do not have valid implementation on this level, and it's up
      to Guest subclasses to provide one working in their respective
      infrastructure.
  
      The following keys are expected in the 'data' container::
  
          role ....... guest role in the multihost scenario
          guest ...... name, hostname or ip address
  
      These are by default imported into instance attributes.
      """
  
      # Used by save() to construct the correct container for keys.
      _data_class = GuestData
  
      role: Optional[str]
      guest:
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/__init__.py
files.old_contents: import collections
  import dataclasses
  import datetime
  import enum
  import os
  import random
  import re
  import shlex
  import string
  import subprocess
  import tempfile
  from shlex import quote
  from typing import Any, Dict, List, Optional, Type, Union, cast
  
  import click
  import fmf
  
  import tmt
  import tmt.plugins
  import tmt.steps
  import tmt.utils
  from tmt.steps import Action
  
  # Timeout in seconds of waiting for a connection after reboot
  CONNECTION_TIMEOUT = 5 * 60
  
  # When waiting for guest to recover from reboot, try re-connecting every
  # this many seconds.
  RECONNECT_WAIT_TICK = 5
  RECONNECT_WAIT_TICK_INCREASE = 1.0
  
  # Default rsync options
  DEFAULT_RSYNC_OPTIONS = [
      "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  DEFAULT_RSYNC_PUSH_OPTIONS = ["-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  DEFAULT_RSYNC_PULL_OPTIONS = ["-R", "-r", "-z", "--links", "--safe-links", "--protect-args"]
  
  
  class CheckRsyncOutcome(enum.Enum):
      ALREADY_INSTALLED = 'already-installed'
      INSTALLED = 'installed'
  
  
  class Provision(tmt.steps.Step):
      """ Provision an environment for testing or use localhost. """
  
      # Default implementation for provision is a virtual machine
      DEFAULT_HOW = 'virtual'
  
      def __init__(self, plan: 'tmt.Plan', data: tmt.steps.StepData) -> None:
          """ Initialize provision step data """
          super().__init__(plan=plan, data=data)
          # Check that the names are unique
          names = [data['name'] for data in self.data]
          names_count: Dict[str, int] = collections.defaultdict(int)
          for name in names:
              names_count[name] += 1
          if len(self.data) > 1 and len(self.data) != len(names_count):
              duplicate = [k for k, v in names_count.items() if v > 1]
              duplicate_string = ', '.join(duplicate)
              raise tmt.utils.GeneralError(
                  f"Provision step names must be unique for multihost testing. "
                  f"Duplicate names: {duplicate_string} in plan '{plan.name}'.")
          # List of provisioned guests and loaded guest data
          self._guests: List['Guest'] = []
          self._guest_data: Dict[str, 'GuestData'] = {}
          self.is_multihost = False
  
      def load(self) -> None:
          """ Load guest data from the workdir """
          super().load()
          try:
              raw_guest_data = tmt.utils.yaml_to_dict(self.read('guests.yaml'))
  
              self._guest_data = {
                  name: tmt.utils.SerializableContainer.unserialize(guest_data)
                  for name, guest_data in raw_guest_data.items()
                  }
  
          except tmt.utils.FileError:
              self.debug('Provisioned guests not found.', level=2)
  
      def save(self) -> None:
          """ Save guest data to the workdir """
          super().save()
          try:
              raw_guest_data = {guest.name: guest.save().to_serialized()
                                for guest in self.guests()}
  
              self.write('guests.yaml', tmt.utils.dict_to_yaml(raw_guest_data))
          except tmt.utils.FileError:
              self.debug('Failed to save provisioned guests.')
  
      def wake(self) -> None:
          """ Wake up the step (process workdir and command line) """
          super().wake()
  
          # Choose the right plugin and wake it up
          for data in self.data:
              # TODO: with generic BasePlugin, delegate() should return more fitting type,
              # not the base class.
              plugin = cast(ProvisionPlugin, ProvisionPlugin.delegate(self, data))
              self._phases.append(plugin)
              # If guest data loaded, perform a complete wake up
              plugin.wake(data=self._guest_data.get(plugin.name))
  
              guest = plugin.guest()
              if guest:
                  self._guests.append(guest)
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Provision wake up complete (already done before).', level=2)
          # Save status and step data (now we know what to do)
          else:
              self.status('todo')
              self.save()
  
      def show(self) -> None:
          """ Show discover details """
          for data in self.data:
              ProvisionPlugin.delegate(self, data).show()
  
      def summary(self) -> None:
          """ Give a concise summary of the provisioning """
          # Summary of provisioned guests
          guests = fmf.utils.listed(self.guests(), 'guest')
          self.info('summary', f'{guests} provisioned', 'green', shift=1)
          # Guest list in verbose mode
          for guest in self.guests():
              if not guest.name.startswith(tmt.utils.DEFAULT_NAME):
                  self.verbose(guest.name, color='red', shift=2)
  
      def go(self) -> None:
          """ Provision all guests"""
          super().go()
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.info('status', 'done', 'green', shift=1)
              self.summary()
              self.actions()
              return
  
          # Provision guests
          self._guests = []
          save = True
          self.is_multihost = sum([isinstance(phase, ProvisionPlugin)
                                  for phase in self.phases()]) > 1
          try:
              for phase in self.phases(classes=(Action, ProvisionPlugin)):
                  try:
                      if isinstance(phase, Action):
                          phase.go()
  
                      elif isinstance(phase, ProvisionPlugin):
                          phase.go()
  
                          guest = phase.guest()
                          if guest:
                              guest.details()
  
                      if self.is_multihost:
                          self.info('')
                  finally:
                      if isinstance(phase, ProvisionPlugin):
                          guest = phase.guest()
                          if guest:
                              self._guests.append(guest)
  
              # Give a summary, update status and save
              self.summary()
              self.status('done')
          except (SystemExit, tmt.utils.SpecificationError) as error:
              # A plugin will only raise SystemExit if the exit is really desired
              # and no other actions should be done. An example of this is
              # listing available images. In such case, the workdir is deleted
              # as it's redundant and save() would throw an error.
              save = False
              raise error
          finally:
              if save:
                  self.save()
  
      def guests(self) -> List['Guest']:
          """ Return the list of all provisioned guests """
          return self._guests
  
      def requires(self) -> List[str]:
          """
          Packages required by all enabled provision plugins
  
          Return a list of packages which need to be installed on the
          provisioned guest so that the workdir can be synced to it.
          Used by the prepare step.
          """
          requires = set()
          for plugin in self.phases(classes=ProvisionPlugin):
              requires.update(plugin.requires())
          return list(requires)
  
  
  class ProvisionPlugin(tmt.steps.GuestlessPlugin):
      """ Common parent of provision plugins """
  
      # Default implementation for provision is a virtual machine
      how = 'virtual'
  
      # List of all supported methods aggregated from all plugins of the same step.
      _supported_methods: List[tmt.steps.Method] = []
  
      # Common keys for all provision step implementations
      _common_keys = ['role']
  
      @classmethod
      def base_command(
              cls,
              usage: str,
              method_class: Optional[Type[click.Command]] = None) -> click.Command:
          """ Create base click command (common for all provision plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Provision.usage(method_overview=usage)
  
          # Create the command
          @click.command(cls=method_class, help=usage)
          @click.pass_context
          @click.option(
              '-h', '--how', metavar='METHOD',
              help='Use specified method for provisioning.')
          def provision(context: click.Context, **kwargs: Any) -> None:
              context.obj.steps.add('provision')
              Provision._save_context(context)
  
          return provision
  
      def wake(self, keys: Optional[List[str]] = None, data: Optional['GuestData'] = None) -> None:
          """
          Wake up the plugin
  
          Override data with command line options.
          Wake up the guest based on provided guest data.
          """
          super().wake(keys=keys)
  
      def guest(self) -> Optional['Guest']:
          """
          Return provisioned guest
  
          Each ProvisionPlugin has to implement this method.
          Should return a provisioned Guest() instance.
          """
          raise NotImplementedError()
  
      def requires(self) -> List[str]:
          """ List of required packages needed for workdir sync """
          return Guest.requires()
  
      @classmethod
      def clean_images(cls, clean: 'tmt.base.Clean', dry: bool) -> None:
          """ Remove the images of one particular plugin """
  
  
  @dataclasses.dataclass
  class GuestData(tmt.utils.SerializableContainer):
      """
      Keys necessary to describe, create, save and restore a guest.
  
      Very basic set of keys shared across all known guest classes.
      """
  
      # guest role in the multihost scenario
      role: Optional[str] = None
      # hostname or ip address
      guest: Optional[str] = None
  
  
  class Guest(tmt.utils.Common):
      """
      Guest provisioned for test execution
  
      A base class for guest-like classes. Provides some of the basic methods
      and functionality, but note some of the methods are left intentionally
      empty. These do not have valid implementation on this level, and it's up
      to Guest subclasses to provide one working in their respective
      infrastructure.
  
      The following keys are expected in the 'data' container::
  
          role ....... guest role in the multihost scenario
          guest ...... name, hostname or ip address
  
      These are by default imported into instance attributes.
      """
  
      # Used by save() to construct the correct container for keys.
      _data_class
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/__init__.py
files.new_contents: import dataclasses
  import datetime
  import sys
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import requests
  
  import tmt
  import tmt.options
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import ProvisionError, retry_session, updatable_message
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  
  # List of Artemis API versions supported and understood by this plugin.
  # Since API gains support for new features over time, it is important to
  # know when particular feature became available, and avoid using it with
  # older APIs.
  SUPPORTED_API_VERSIONS = (
      # NEW: added hostname HW constraint
      '0.0.38',
      # NEW: virtualization HW constraint
      '0.0.37',
      # NEW: boot.method HW constraint
      '0.0.32',
      # NEW: network HW constraint
      '0.0.28'
      )
  
  
  # The default Artemis API version - the most recent supported versions
  # should be perfectly fine.
  DEFAULT_API_VERSION = SUPPORTED_API_VERSIONS[0]
  
  DEFAULT_API_URL = 'http://127.0.0.1:8001'
  DEFAULT_USER = 'root'
  DEFAULT_ARCH = 'x86_64'
  DEFAULT_PRIORITY_GROUP = 'default-priority'
  DEFAULT_KEYNAME = 'default'
  DEFAULT_PROVISION_TIMEOUT = 600
  DEFAULT_PROVISION_TICK = 60
  DEFAULT_API_TIMEOUT = 10
  DEFAULT_API_RETRIES = 10
  # Should lead to delays of 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256 seconds
  DEFAULT_RETRY_BACKOFF_FACTOR = 1
  
  # Type annotation for "data" package describing a guest instance. Passed
  # between load() and save() calls
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class ArtemisGuestData(tmt.steps.provision.GuestSshData):
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      # API
      api_url: str = DEFAULT_API_URL
      api_version: str = DEFAULT_API_VERSION
  
      # Guest request properties
      arch: str = DEFAULT_ARCH
      image: Optional[str] = None
      hardware: Optional[Any] = None
      pool: Optional[str] = None
      priority_group: str = DEFAULT_PRIORITY_GROUP
      keyname: str = DEFAULT_KEYNAME
      user_data: Dict[str, str] = dataclasses.field(default_factory=dict)
  
      # Provided by Artemis response
      guestname: Optional[str] = None
  
      # Timeouts and deadlines
      provision_timeout: int = DEFAULT_PROVISION_TIMEOUT
      provision_tick: int = DEFAULT_PROVISION_TICK
      api_timeout: int = DEFAULT_API_TIMEOUT
      api_retries: int = DEFAULT_API_RETRIES
      api_retry_backoff_factor: int = DEFAULT_RETRY_BACKOFF_FACTOR
  
  
  GUEST_STATE_COLOR_DEFAULT = 'green'
  
  GUEST_STATE_COLORS = {
      'routing': 'yellow',
      'provisioning': 'magenta',
      'promised': 'blue',
      'preparing': 'cyan',
      'cancelled': 'red',
      'error': 'red'
  }
  
  
  # Type annotation for Artemis API `GET /guests/$guestname` response.
  # Partial, not all fields necessary since plugin ignores most of them.
  GuestInspectType = TypedDict(
      'GuestInspectType', {
          'state': str,
          'address': Optional[str]
          }
      )
  
  
  class ArtemisAPI:
      def __init__(self, guest: 'GuestArtemis') -> None:
          self._guest = guest
  
          self.http_session = retry_session.create(
              retries=guest.api_retries,
              backoff_factor=guest.api_retry_backoff_factor,
              allowed_methods=('HEAD', 'GET', 'POST', 'DELETE', 'PUT'),
              status_forcelist=(
                  429,  # Too Many Requests
                  500,  # Internal Server Error
                  502,  # Bad Gateway
                  503,  # Service Unavailable
                  504   # Gateway Timeout
                  ),
              timeout=guest.api_timeout
              )
  
      def query(
              self,
              path: str,
              method: str = 'get',
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Base helper for Artemis API queries.
  
          Trivial dispatcher per method, returning retrieved response.
  
          :param path: API path to contact.
          :param method: HTTP method to use.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          url = f'{self._guest.api_url}{path}'
  
          if method == 'get':
              return self.http_session.get(url, **request_kwargs)
  
          if method == 'post':
              return self.http_session.post(url, **request_kwargs)
  
          if method == 'delete':
              return self.http_session.delete(url, **request_kwargs)
  
          if method == 'put':
              return self.http_session.put(url, **request_kwargs)
  
          raise tmt.utils.GeneralError(
              f'Unsupported Artemis API method.\n{method}')
  
      def create(
              self,
              path: str,
              data: Dict[str, Any],
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Create - or request creation of - a resource.
  
          :param path: API path to contact.
          :param data: optional key/value data to send with the request.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
          request_kwargs['json'] = data
  
          return self.query(path, method='post', request_kwargs=request_kwargs)
  
      def inspect(
              self,
              path: str,
              params: Optional[Dict[str, Any]] = None,
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Inspect a resource.
  
          :param path: API path to contact.
          :param params: optional key/value query parameters.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          if params:
              request_kwargs['params'] = params
  
          return self.query(path, request_kwargs=request_kwargs)
  
      def delete(
              self,
              path: str,
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Delete - or request removal of - a resource.
  
          :param path: API path to contact.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          return self.query(path, method='delete', request_kwargs=request_kwargs)
  
  
  @tmt.steps.provides_method('artemis')
  class ProvisionArtemis(tmt.steps.provision.ProvisionPlugin):
      """
      Provision guest using Artemis backend
  
      Minimal configuration could look like this:
  
          provision:
              how: artemis
              image: Fedora
              api-url: https://your-artemis.com/
  
      Note that the actual value of "image" depends on what images - or
      "composes" as Artemis calls them - supports and can deliver.
  
      Note that "api-url" can be also given via ARTEMIS_API_URL
      environment variable.
  
      Full configuration example:
  
          provision:
              how: artemis
  
              # Artemis API
              api-url: https://your-artemis.com/
              api-version: 0.0.32
  
              # Mandatory environment properties
              image: Fedora
  
              # Optional environment properties
              arch: aarch64
              pool: optional-pool-name
  
              # Provisioning process control (optional)
              priority-group: custom-priority-group
              keyname: custom-SSH-key-name
  
              # Labels to be attached to guest request (optional)
              user-data:
                  foo: bar
  
              # Timeouts and deadlines (optional)
              provision-timeout: 3600
              provision-tick: 10
              api-timeout: 600
              api-retries: 5
              api-retry-backoff-factor: 1
      """
  
      # Guest instance
      _guest = None
  
      _keys = [
          'api-url',
          'api-version',
          'arch',
          'image',
          'hardware',
          'pool',
          'priority-group',
          'keyname',
          'user-data',
          'provision-timeout',
          'provision-tick',
          'api-timeout',
          'api-retries',
          'api-retry-backoff-factor']
  
      # TODO: fix types once superclass gains its annotations
      @classmethod
      def options(cls, how: Any = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for Artemis """
          return cast(List[tmt.options.ClickOptionDecoratorType], [
              click.option(
                  '--api-url', metavar='URL',
                  help="Artemis API URL.",
                  envvar='ARTEMIS_API_URL'
                  ),
              click.option(
                  '--api-version', metavar='x.y.z',
                  help="Artemis API version to use.",
                  type=click.Choice(SUPPORTED_API_VERSIONS),
                  envvar='ARTEMIS_API_VERSION'
                  ),
              click.option(
                  '--arch', metavar='ARCH',
                  help='Architecture to provision.'
                  ),
              click.option(
                  '--image', metavar='COMPOSE',
                  help='Image (or "compose" in Artemis terminology) '
                       'to provision.'
                  ),
              click.option(
                  '--pool', metavar='NAME',
  
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/artemis.py
files.old_contents: import dataclasses
  import datetime
  import sys
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import requests
  
  import tmt
  import tmt.options
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import ProvisionError, retry_session, updatable_message
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  
  # List of Artemis API versions supported and understood by this plugin.
  # Since API gains support for new features over time, it is important to
  # know when particular feature became available, and avoid using it with
  # older APIs.
  SUPPORTED_API_VERSIONS = (
      # NEW: added hostname HW constraint
      '0.0.38',
      # NEW: virtualization HW constraint
      '0.0.37',
      # NEW: boot.method HW constraint
      '0.0.32',
      # NEW: network HW constraint
      '0.0.28'
      )
  
  
  # The default Artemis API version - the most recent supported versions
  # should be perfectly fine.
  DEFAULT_API_VERSION = SUPPORTED_API_VERSIONS[0]
  
  DEFAULT_API_URL = 'http://127.0.0.1:8001'
  DEFAULT_USER = 'root'
  DEFAULT_ARCH = 'x86_64'
  DEFAULT_PRIORITY_GROUP = 'default-priority'
  DEFAULT_KEYNAME = 'default'
  DEFAULT_PROVISION_TIMEOUT = 600
  DEFAULT_PROVISION_TICK = 60
  DEFAULT_API_TIMEOUT = 10
  DEFAULT_API_RETRIES = 10
  # Should lead to delays of 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256 seconds
  DEFAULT_RETRY_BACKOFF_FACTOR = 1
  
  # Type annotation for "data" package describing a guest instance. Passed
  # between load() and save() calls
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class ArtemisGuestData(tmt.steps.provision.GuestSshData):
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      # API
      api_url: str = DEFAULT_API_URL
      api_version: str = DEFAULT_API_VERSION
  
      # Guest request properties
      arch: str = DEFAULT_ARCH
      image: Optional[str] = None
      hardware: Optional[Any] = None
      pool: Optional[str] = None
      priority_group: str = DEFAULT_PRIORITY_GROUP
      keyname: str = DEFAULT_KEYNAME
      user_data: Dict[str, str] = dataclasses.field(default_factory=dict)
  
      # Provided by Artemis response
      guestname: Optional[str] = None
  
      # Timeouts and deadlines
      provision_timeout: int = DEFAULT_PROVISION_TIMEOUT
      provision_tick: int = DEFAULT_PROVISION_TICK
      api_timeout: int = DEFAULT_API_TIMEOUT
      api_retries: int = DEFAULT_API_RETRIES
      api_retry_backoff_factor: int = DEFAULT_RETRY_BACKOFF_FACTOR
  
  
  GUEST_STATE_COLOR_DEFAULT = 'green'
  
  GUEST_STATE_COLORS = {
      'routing': 'yellow',
      'provisioning': 'magenta',
      'promised': 'blue',
      'preparing': 'cyan',
      'cancelled': 'red',
      'error': 'red'
  }
  
  
  # Type annotation for Artemis API `GET /guests/$guestname` response.
  # Partial, not all fields necessary since plugin ignores most of them.
  GuestInspectType = TypedDict(
      'GuestInspectType', {
          'state': str,
          'address': Optional[str]
          }
      )
  
  
  class ArtemisAPI:
      def __init__(self, guest: 'GuestArtemis') -> None:
          self._guest = guest
  
          self.http_session = retry_session.create(
              retries=guest.api_retries,
              backoff_factor=guest.api_retry_backoff_factor,
              allowed_methods=('HEAD', 'GET', 'POST', 'DELETE', 'PUT'),
              status_forcelist=(
                  429,  # Too Many Requests
                  500,  # Internal Server Error
                  502,  # Bad Gateway
                  503,  # Service Unavailable
                  504   # Gateway Timeout
                  ),
              timeout=guest.api_timeout
              )
  
      def query(
              self,
              path: str,
              method: str = 'get',
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Base helper for Artemis API queries.
  
          Trivial dispatcher per method, returning retrieved response.
  
          :param path: API path to contact.
          :param method: HTTP method to use.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          url = f'{self._guest.api_url}{path}'
  
          if method == 'get':
              return self.http_session.get(url, **request_kwargs)
  
          if method == 'post':
              return self.http_session.post(url, **request_kwargs)
  
          if method == 'delete':
              return self.http_session.delete(url, **request_kwargs)
  
          if method == 'put':
              return self.http_session.put(url, **request_kwargs)
  
          raise tmt.utils.GeneralError(
              f'Unsupported Artemis API method.\n{method}')
  
      def create(
              self,
              path: str,
              data: Dict[str, Any],
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Create - or request creation of - a resource.
  
          :param path: API path to contact.
          :param data: optional key/value data to send with the request.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
          request_kwargs['json'] = data
  
          return self.query(path, method='post', request_kwargs=request_kwargs)
  
      def inspect(
              self,
              path: str,
              params: Optional[Dict[str, Any]] = None,
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Inspect a resource.
  
          :param path: API path to contact.
          :param params: optional key/value query parameters.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          if params:
              request_kwargs['params'] = params
  
          return self.query(path, request_kwargs=request_kwargs)
  
      def delete(
              self,
              path: str,
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Delete - or request removal of - a resource.
  
          :param path: API path to contact.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          return self.query(path, method='delete', request_kwargs=request_kwargs)
  
  
  @tmt.steps.provides_method('artemis')
  class ProvisionArtemis(tmt.steps.provision.ProvisionPlugin):
      """
      Provision guest using Artemis backend
  
      Minimal configuration could look like this:
  
          provision:
              how: artemis
              image: Fedora
              api-url: https://your-artemis.com/
  
      Note that the actual value of "image" depends on what images - or
      "composes" as Artemis calls them - supports and can deliver.
  
      Note that "api-url" can be also given via ARTEMIS_API_URL
      environment variable.
  
      Full configuration example:
  
          provision:
              how: artemis
  
              # Artemis API
              api-url: https://your-artemis.com/
              api-version: 0.0.32
  
              # Mandatory environment properties
              image: Fedora
  
              # Optional environment properties
              arch: aarch64
              pool: optional-pool-name
  
              # Provisioning process control (optional)
              priority-group: custom-priority-group
              keyname: custom-SSH-key-name
  
              # Labels to be attached to guest request (optional)
              user-data:
                  foo: bar
  
              # Timeouts and deadlines (optional)
              provision-timeout: 3600
              provision-tick: 10
              api-timeout: 600
              api-retries: 5
              api-retry-backoff-factor: 1
      """
  
      # Guest instance
      _guest = None
  
      _keys = [
          'api-url',
          'api-version',
          'arch',
          'image',
          'hardware',
          'pool',
          'priority-group',
          'keyname',
          'user-data',
          'provision-timeout',
          'provision-tick',
          'api-timeout',
          'api-retries',
          'api-retry-backoff-factor']
  
      # TODO: fix types once superclass gains its annotations
      @classmethod
      def options(cls, how: Any = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for Artemis """
          return cast(List[tmt.options.ClickOptionDecoratorType], [
              click.option(
                  '--api-url', metavar='URL',
                  help="Artemis API URL.",
                  envvar='ARTEMIS_API_URL'
                  ),
              click.option(
                  '--api-version', metavar='x.y.z',
                  help="Artemis API version to use.",
                  type=click.Choice(SUPPORTED_API_VERSIONS),
                  envvar='ARTEMIS_API_VERSION'
                  ),
              click.option(
                  '--arch', metavar='ARCH',
                  help='Architecture to provision.'
                  ),
              click.option(
                  '--image', metavar='COMPOSE',
                  help='Image (or "compose" in Artemis terminology) '
                       'to provision.'
                  ),
              click.option(
                  '--pool', metavar='NAME',
  
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/artemis.py
files.new_contents: from typing import Any, List, Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.steps.provision import GuestSshData
  
  
  @tmt.steps.provides_method('connect')
  class ProvisionConnect(tmt.steps.provision.ProvisionPlugin):
      """
      Connect to a provisioned guest using ssh
  
      Private key authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              key: /home/psss/.ssh/example_rsa
  
      Password authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              password: secret
  
      User defaults to 'root', so if you have private key correctly set
      the minimal configuration can look like this:
  
          provision:
              how: connect
              guest: host.example.org
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["guest", "key", "user", "password", "port"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for connect """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '-g', '--guest', metavar='GUEST',
                  help='Select remote host to connect to (hostname or ip).'),
              click.option(
                  '-P', '--port', metavar='PORT',
                  help='Use specific port to connect to.'),
              click.option(
                  '-k', '--key', metavar='PRIVATE_KEY',
                  help='Private key for login into the guest system.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-p', '--password', metavar='PASSWORD',
                  help='Password for login into the guest system.'),
              ]
          return options
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          # User root as the default user
          if option == 'user':
              return 'root'
          # No other defaults available
          return default
  
      def wake(self, data: Optional[GuestSshData] = None) -> None:  # type: ignore[override]
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
          if data:
              self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def go(self) -> None:
          """ Prepare the connection """
          super().go()
  
          # Check connection details
          guest = self.get('guest')
          user = self.get('user')
          key = self.get('key')
          password = self.get('password')
          port = self.get('port')
  
          # Check guest and auth info
          if not guest:
              raise tmt.utils.SpecificationError(
                  'Provide a host name or an ip address to connect.')
          data = GuestSshData(
              role=self.get('role'),
              guest=guest,
              user=user
              )
          self.info('guest', guest, 'green')
          self.info('user', user, 'green')
          if port:
              self.info('port', port, 'green')
              data.port = port
  
          # Use provided password for authentication
          if password:
              self.info('password', password, 'green')
              self.debug('Using password authentication.')
              data.password = password
          # Default to using a private key (user can have configured one)
          else:
              self.info('key', key or 'not provided', 'green')
              self.debug('Using private key authentication.')
              data.key = [key]
  
          # And finally create the guest
          self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def guest(self) -> Optional[tmt.GuestSsh]:
          """ Return the provisioned guest """
          return self._guest
  
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/connect.py
files.old_contents: from typing import Any, List, Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.steps.provision import GuestSshData
  
  
  @tmt.steps.provides_method('connect')
  class ProvisionConnect(tmt.steps.provision.ProvisionPlugin):
      """
      Connect to a provisioned guest using ssh
  
      Private key authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              key: /home/psss/.ssh/example_rsa
  
      Password authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              password: secret
  
      User defaults to 'root', so if you have private key correctly set
      the minimal configuration can look like this:
  
          provision:
              how: connect
              guest: host.example.org
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["guest", "key", "user", "password", "port"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for connect """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '-g', '--guest', metavar='GUEST',
                  help='Select remote host to connect to (hostname or ip).'),
              click.option(
                  '-P', '--port', metavar='PORT',
                  help='Use specific port to connect to.'),
              click.option(
                  '-k', '--key', metavar='PRIVATE_KEY',
                  help='Private key for login into the guest system.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-p', '--password', metavar='PASSWORD',
                  help='Password for login into the guest system.'),
              ]
          return options
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          # User root as the default user
          if option == 'user':
              return 'root'
          # No other defaults available
          return default
  
      # More specific type is a violation of Liskov substitution principle, and mypy
      # complains about it - rightfully so. Ignoring the issue which should be resolved
      # with https://github.com/teemtee/tmt/pull/1439.
      def wake(self, keys: Optional[List[str]] = None,  # type: ignore[override]
               data: Optional[GuestSshData] = None) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          if data:
              self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def go(self) -> None:
          """ Prepare the connection """
          super().go()
  
          # Check connection details
          guest = self.get('guest')
          user = self.get('user')
          key = self.get('key')
          password = self.get('password')
          port = self.get('port')
  
          # Check guest and auth info
          if not guest:
              raise tmt.utils.SpecificationError(
                  'Provide a host name or an ip address to connect.')
          data = GuestSshData(
              role=self.get('role'),
              guest=guest,
              user=user
              )
          self.info('guest', guest, 'green')
          self.info('user', user, 'green')
          if port:
              self.info('port', port, 'green')
              data.port = port
  
          # Use provided password for authentication
          if password:
              self.info('password', password, 'green')
              self.debug('Using password authentication.')
              data.password = password
          # Default to using a private key (user can have configured one)
          else:
              self.info('key', key or 'not provided', 'green')
              self.debug('Using private key authentication.')
              data.key = [key]
  
          # And finally create the guest
          self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def guest(self) -> Optional[tmt.GuestSsh]:
          """ Return the provisioned guest """
          return self._guest
  
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/connect.py
files.new_contents: import tmt
  import tmt.steps
  import tmt.steps.provision
  
  
  @tmt.steps.provides_method('local')
  class ProvisionLocal(tmt.steps.provision.ProvisionPlugin):
      """
      Use local host for test execution
  
      In general it is not recommended to run tests on your local machine
      as there might be security risks. Run only those tests which you
      know are safe so that you don't destroy your laptop ;-)
  
      Example config:
  
          provision:
              how: local
  
      Note that 'tmt run' is expected to be executed under a regular user.
      If there are admin rights required (for example in the prepare step)
      you might be asked for a sudo password.
      """
  
      # Guest instance
      _guest = None
  
      def wake(self, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
          if data:
              self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Create a GuestLocal instance
          data = tmt.steps.provision.GuestSshData(
              guest='localhost',
              role=self.get('role')
              )
          self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestLocal.requires()
  
  
  class GuestLocal(tmt.Guest):
      """ Local Host """
  
      localhost = True
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, stderr = self.run(
              ['sudo', '-E', 'ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'local', '-i', 'localhost,', playbook],
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def execute(self, command, **kwargs):
          """ Execute command on localhost """
          # Prepare the environment (plan/cli variables override)
          environment = dict()
          environment.update(kwargs.pop('env', dict()))
          environment.update(self.parent.plan.environment)
          # Run the command under the prepared environment
          return self.run(command, env=environment, shell=True, **kwargs)
  
      def stop(self):
          """ Stop the guest """
  
          self.debug(f"Doing nothing to stop guest '{self.guest}'.")
  
      def reboot(self, hard=False, command=None, timeout=None):
          """ Reboot the guest, return True if successful """
  
          self.debug(f"Doing nothing to reboot guest '{self.guest}'.")
  
          return False
  
      def push(self, source=None, destination=None, options=None):
          """ Nothing to be done to push workdir """
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """ Nothing to be done to pull workdir """
  
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/local.py
files.old_contents: import tmt
  import tmt.steps
  import tmt.steps.provision
  
  
  @tmt.steps.provides_method('local')
  class ProvisionLocal(tmt.steps.provision.ProvisionPlugin):
      """
      Use local host for test execution
  
      In general it is not recommended to run tests on your local machine
      as there might be security risks. Run only those tests which you
      know are safe so that you don't destroy your laptop ;-)
  
      Example config:
  
          provision:
              how: local
  
      Note that 'tmt run' is expected to be executed under a regular user.
      If there are admin rights required (for example in the prepare step)
      you might be asked for a sudo password.
      """
  
      # Guest instance
      _guest = None
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          if data:
              self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Create a GuestLocal instance
          data = tmt.steps.provision.GuestSshData(
              guest='localhost',
              role=self.get('role')
              )
          self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestLocal.requires()
  
  
  class GuestLocal(tmt.Guest):
      """ Local Host """
  
      localhost = True
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, stderr = self.run(
              ['sudo', '-E', 'ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'local', '-i', 'localhost,', playbook],
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def execute(self, command, **kwargs):
          """ Execute command on localhost """
          # Prepare the environment (plan/cli variables override)
          environment = dict()
          environment.update(kwargs.pop('env', dict()))
          environment.update(self.parent.plan.environment)
          # Run the command under the prepared environment
          return self.run(command, env=environment, shell=True, **kwargs)
  
      def stop(self):
          """ Stop the guest """
  
          self.debug(f"Doing nothing to stop guest '{self.guest}'.")
  
      def reboot(self, hard=False, command=None, timeout=None):
          """ Reboot the guest, return True if successful """
  
          self.debug(f"Doing nothing to reboot guest '{self.guest}'.")
  
          return False
  
      def push(self, source=None, destination=None, options=None):
          """ Nothing to be done to push workdir """
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """ Nothing to be done to pull workdir """
  
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/local.py
files.new_contents: import dataclasses
  import os
  from shlex import quote
  from typing import Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  
  # Timeout in seconds of waiting for a connection
  CONNECTION_TIMEOUT = 60
  
  # Defaults
  DEFAULT_IMAGE = "fedora"
  DEFAULT_USER = "root"
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class PodmanGuestData(tmt.steps.provision.GuestData):  # type: ignore[misc]
      image: str = DEFAULT_IMAGE
      user: str = DEFAULT_USER
      force_pull: bool = False
  
      container: Optional[str] = None
  
  
  @tmt.steps.provides_method('container')
  class ProvisionPodman(tmt.steps.provision.ProvisionPlugin):
      """
      Create a new container using podman
  
      Example config:
  
          provision:
              how: container
              image: fedora:latest
  
      In order to always pull the fresh container image use 'pull: true'.
  
      In order to run the container with different user as the default 'root',
      use 'user: USER'.
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["image", "container", "pull", "user"]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for connect """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to use. Short name or complete url.'),
              click.option(
                  '-c', '--container', metavar='NAME',
                  help='Name or id of an existing container to be used.'),
              click.option(
                  '-p', '--pull', is_flag=True,
                  help='Force pulling a fresh container image.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='User to use for all container operations.')
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          if option == 'pull':
              return PodmanGuestData().force_pull
  
          return getattr(PodmanGuestData(), option.replace('-', '_'), default)
  
      def wake(self, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
          # Wake up podman instance
          if data:
              guest = GuestContainer(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Show which image we are using
          pull = ' (force pull)' if self.get('pull') else ''
          self.info('image', f"{self.get('image')}{pull}", 'green')
  
          # Prepare data for the guest instance
          data_from_options = {
              key: self.get(key)
              for key in PodmanGuestData.keys()
              if key != 'force_pull'
              }
  
          data_from_options['force_pull'] = self.get('pull')
  
          data = PodmanGuestData(**data_from_options)
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestContainer(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestContainer.requires()
  
  
  class GuestContainer(tmt.Guest):
      """ Container Instance """
  
      _data_class = PodmanGuestData
  
      image: Optional[str]
      container: Optional[str]
      user: str
      force_pull: bool
  
      def wake(self):
          """ Wake up the guest """
          self.debug(
              f"Waking up container '{self.container}'.", level=2, shift=0)
  
      def start(self):
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Check if the image is available
          image_id = self.podman(
              ['images', '-q', self.image],
              message=f"Check for container image '{self.image}'.")[0].strip()
  
          # Pull image if not available or pull forced
          if not image_id or self.force_pull:
              self.podman(
                  ['pull', '-q', self.image],
                  message=f"Pull image '{self.image}'.")
  
          # Mount the whole plan directory in the container
          workdir = self.parent.plan.workdir
  
          self.container = self.guest = self._tmt_name()
          self.verbose('name', self.container, 'green')
  
          # FIXME: Workaround for BZ#1900021 (f34 container on centos-8)
          workaround = ['--security-opt', 'seccomp=unconfined']
  
          # Run the container
          self.debug(f"Start container '{self.image}'.")
          self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f'{workdir}:{workdir}:z',
               '-itd', '--user', self.user, self.image])
  
      def reboot(self, hard=False, command=None, timeout=None):
          """ Restart the container, return True if successful  """
          if command:
              raise tmt.utils.ProvisionError(
                  "Custom reboot command not supported in podman provision.")
          if not hard:
              raise tmt.utils.ProvisionError(
                  "Containers do not support soft reboot, they can only be "
                  "stopped and started again (hard reboot).")
          self.podman(['container', 'restart', self.container])
          return self.reconnect(timeout=timeout or CONNECTION_TIMEOUT)
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare container using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          # As non-root we must run with podman unshare
          podman_unshare = ['podman', 'unshare'] if os.geteuid() != 0 else []
          stdout, stderr = self.run(
              podman_unshare + ['ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'podman', '-i', f'{self.container},', playbook],
              cwd=self.parent.plan.worktree,
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def podman(self, command, **kwargs):
          """ Run given command via podman """
          return self.run(['podman'] + command, **kwargs)
  
      def execute(self, command, **kwargs):
          """ Execute given commands in podman via shell """
          if not self.container and not self.opt('dry'):
              raise tmt.utils.ProvisionError(
                  'Could not execute without provisioned container.')
  
          # Change to given directory on guest if cwd provided
          directory = kwargs.get('cwd', '')
          if directory:
              directory = f"cd {quote(directory)}; "
  
          # Prepare the environment variables export
          environment = self._export_environment(
              self._prepare_environment(kwargs.get('env', dict())))
  
          # Run in interactive mode if requested
          interactive = ['-it'] if kwargs.get('interactive') else []
  
          # Note that we MUST run commands via bash, so variables
          # work as expected
          if isinstance(command, list):
              command = ' '.join(command)
          command = directory + environment + command
          return self.podman(
              ['exec'] + interactive +
              [self.container or 'dry', 'bash', '-c', command], **kwargs)
  
      def push(self, source=None, destination=None, options=None):
          """ Make sure that the workdir has a correct selinux context """
          self.debug("Update selinux context of the run workdir.", level=3)
          self.run(
              ["chcon", "--recursive", "--type=container_file_t",
               self.parent.plan.workdir], shell=False)
          # In case explicit destination is given, use `podman cp` to copy data
          # to the container
          if destination:
              self.podman(["cp", source, f"{self.container}:{destination}"])
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """ Nothing to be done to pull workdir """
  
      def stop(self):
          """ Stop provisioned guest """
          if self.container:
              self.podman(['container', 'stop', self.container])
              self.info('container', 'stopped', 'green')
  
      def remove(self):
          """ Remove the container """
          if self.container:
              self.podman(['container', 'rm', '-f', self.container])
              self.info('container', 'removed', 'green')
  
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/podman.py
files.old_contents: import dataclasses
  import os
  from shlex import quote
  from typing import Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  
  # Timeout in seconds of waiting for a connection
  CONNECTION_TIMEOUT = 60
  
  # Defaults
  DEFAULT_IMAGE = "fedora"
  DEFAULT_USER = "root"
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class PodmanGuestData(tmt.steps.provision.GuestData):  # type: ignore[misc]
      image: str = DEFAULT_IMAGE
      user: str = DEFAULT_USER
      force_pull: bool = False
  
      container: Optional[str] = None
  
  
  @tmt.steps.provides_method('container')
  class ProvisionPodman(tmt.steps.provision.ProvisionPlugin):
      """
      Create a new container using podman
  
      Example config:
  
          provision:
              how: container
              image: fedora:latest
  
      In order to always pull the fresh container image use 'pull: true'.
  
      In order to run the container with different user as the default 'root',
      use 'user: USER'.
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["image", "container", "pull", "user"]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for connect """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to use. Short name or complete url.'),
              click.option(
                  '-c', '--container', metavar='NAME',
                  help='Name or id of an existing container to be used.'),
              click.option(
                  '-p', '--pull', is_flag=True,
                  help='Force pulling a fresh container image.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='User to use for all container operations.')
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          if option == 'pull':
              return PodmanGuestData().force_pull
  
          return getattr(PodmanGuestData(), option.replace('-', '_'), default)
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          # Wake up podman instance
          if data:
              guest = GuestContainer(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Show which image we are using
          pull = ' (force pull)' if self.get('pull') else ''
          self.info('image', f"{self.get('image')}{pull}", 'green')
  
          # Prepare data for the guest instance
          data_from_options = {
              key: self.get(key)
              for key in PodmanGuestData.keys()
              if key != 'force_pull'
              }
  
          data_from_options['force_pull'] = self.get('pull')
  
          data = PodmanGuestData(**data_from_options)
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestContainer(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestContainer.requires()
  
  
  class GuestContainer(tmt.Guest):
      """ Container Instance """
  
      _data_class = PodmanGuestData
  
      image: Optional[str]
      container: Optional[str]
      user: str
      force_pull: bool
  
      def wake(self):
          """ Wake up the guest """
          self.debug(
              f"Waking up container '{self.container}'.", level=2, shift=0)
  
      def start(self):
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Check if the image is available
          image_id = self.podman(
              ['images', '-q', self.image],
              message=f"Check for container image '{self.image}'.")[0].strip()
  
          # Pull image if not available or pull forced
          if not image_id or self.force_pull:
              self.podman(
                  ['pull', '-q', self.image],
                  message=f"Pull image '{self.image}'.")
  
          # Mount the whole plan directory in the container
          workdir = self.parent.plan.workdir
  
          self.container = self.guest = self._tmt_name()
          self.verbose('name', self.container, 'green')
  
          # FIXME: Workaround for BZ#1900021 (f34 container on centos-8)
          workaround = ['--security-opt', 'seccomp=unconfined']
  
          # Run the container
          self.debug(f"Start container '{self.image}'.")
          self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f'{workdir}:{workdir}:z',
               '-itd', '--user', self.user, self.image])
  
      def reboot(self, hard=False, command=None, timeout=None):
          """ Restart the container, return True if successful  """
          if command:
              raise tmt.utils.ProvisionError(
                  "Custom reboot command not supported in podman provision.")
          if not hard:
              raise tmt.utils.ProvisionError(
                  "Containers do not support soft reboot, they can only be "
                  "stopped and started again (hard reboot).")
          self.podman(['container', 'restart', self.container])
          return self.reconnect(timeout=timeout or CONNECTION_TIMEOUT)
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare container using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          # As non-root we must run with podman unshare
          podman_unshare = ['podman', 'unshare'] if os.geteuid() != 0 else []
          stdout, stderr = self.run(
              podman_unshare + ['ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'podman', '-i', f'{self.container},', playbook],
              cwd=self.parent.plan.worktree,
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def podman(self, command, **kwargs):
          """ Run given command via podman """
          return self.run(['podman'] + command, **kwargs)
  
      def execute(self, command, **kwargs):
          """ Execute given commands in podman via shell """
          if not self.container and not self.opt('dry'):
              raise tmt.utils.ProvisionError(
                  'Could not execute without provisioned container.')
  
          # Change to given directory on guest if cwd provided
          directory = kwargs.get('cwd', '')
          if directory:
              directory = f"cd {quote(directory)}; "
  
          # Prepare the environment variables export
          environment = self._export_environment(
              self._prepare_environment(kwargs.get('env', dict())))
  
          # Run in interactive mode if requested
          interactive = ['-it'] if kwargs.get('interactive') else []
  
          # Note that we MUST run commands via bash, so variables
          # work as expected
          if isinstance(command, list):
              command = ' '.join(command)
          command = directory + environment + command
          return self.podman(
              ['exec'] + interactive +
              [self.container or 'dry', 'bash', '-c', command], **kwargs)
  
      def push(self, source=None, destination=None, options=None):
          """ Make sure that the workdir has a correct selinux context """
          self.debug("Update selinux context of the run workdir.", level=3)
          self.run(
              ["chcon", "--recursive", "--type=container_file_t",
               self.parent.plan.workdir], shell=False)
          # In case explicit destination is given, use `podman cp` to copy data
          # to the container
          if destination:
              self.podman(["cp", source, f"{self.container}:{destination}"])
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """ Nothing to be done to pull workdir """
  
      def stop(self):
          """ Stop provisioned guest """
          if self.container:
              self.podman(['container', 'stop', self.container])
              self.info('container', 'stopped', 'green')
  
      def remove(self):
          """ Remove the container """
          if self.container:
              self.podman(['container', 'rm', '-f', self.container])
              self.info('container', 'removed', 'green')
  
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/podman.py
files.new_contents: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  from typing import Optional
  
  import click
  import requests
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  
  def import_testcloud():
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
    - {public_key}
  ssh_pwauth: true
  disable_root: false
  runcmd:
    - sed -i -e '/^.*PermitRootLogin/s/^.*$/PermitRootLogin yes/'
      -e '/^.*UseDNS/s/^.*$/UseDNS no/'
      -e '/^.*GSSAPIAuthentication/s/^.*$/GSSAPIAuthentication no/'
      /etc/ssh/sshd_config
    - systemctl reload sshd
    - [sh, -c, 'if [ ! -f /etc/systemd/network/20-tc-usernet.network ] &&
    systemctl status systemd-networkd | grep -q "enabled;\\svendor\\spreset:\\senabled";
    then mkdir -p /etc/systemd/network/ &&
    echo "[Match]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "Name=en*" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "[Network]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "DHCP=yes" >> /etc/systemd/network/20-tc-usernet.network; fi']
    - [sh, -c, 'if systemctl status systemd-networkd |
    grep -q "enabled;\\svendor\\spreset:\\senabled"; then
    systemctl restart systemd-networkd; fi']
    - [sh, -c, 'if cat /etc/os-release |
    grep -q platform:el8; then systemctl restart sshd; fi']
  """
  
  COREOS_DATA = """variant: fcos
  version: 1.4.0
  passwd:
    users:
      - name: {user_name}
        ssh_authorized_keys:
          - {public_key}
  """
  
  # Libvirt domain XML template related variables
  DOMAIN_TEMPLATE_NAME = 'domain-template.jinja'
  DOMAIN_TEMPLATE_FILE = os.path.join(TESTCLOUD_DATA, DOMAIN_TEMPLATE_NAME)
  DOMAIN_TEMPLATE = """<domain type='{{ virt_type }}' xmlns:qemu='http://libvirt.org/schemas/domain/qemu/1.0'>
    <name>{{ domain_name }}</name>
    <uuid>{{ uuid }}</uuid>
    <memory unit='KiB'>{{ memory }}</memory>
    <currentMemory unit='KiB'>{{ memory }}</currentMemory>
    <vcpu placement='static'>2</vcpu>
    <os>
      <type arch='{{ arch }}' machine='{{ model }}'>hvm</type>
      {{ uefi_loader }}
      <boot dev='hd'/>
    </os>
    {{ cpu }}
    {{ extra_specs }}
    <clock offset='utc'>
      <timer name='rtc' tickpolicy='catchup'/>
      <timer name='pit' tickpolicy='delay'/>
      <timer name='hpet' present='no'/>
    </clock>
    <on_poweroff>destroy</on_poweroff>
    <on_reboot>restart</on_reboot>
    <on_crash>restart</on_crash>
    <devices>
      <emulator>{{ emulator_path }}</emulator>
      <disk type='file' device='disk'>
        <driver name='qemu' type='qcow2' cache='unsafe'/>
        <source file="{{ disk }}"/>
        <target dev='vda' bus='virtio'/>
        {{ boot_drive_address }}
      </disk>
      <disk type='file' device='disk'>
        <driver name='qemu' type='raw'/>
        <source file="{{ seed }}"/>
        <target dev='vdb' bus='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x08' function='0x0'/>
      </disk>
      <interface type='{{ network_type }}'>
        <mac address="{{ mac_address }}"/>
        {{ network_source }}
        {{ ip_setup }}
        <model type='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>
      </interface>
      <serial type='pty'>
        <target port='0'/>
      </serial>
      <console type='pty'>
        <target type='serial' port='0'/>
      </console>
      <input type="keyboard" bus="virtio"/>
      <rng model='virtio'>
        <backend model='random'>/dev/urandom</backend>
      </rng>
    </devices>
    {{ qemu_args }}
  </domain>
  """  # noqa: E501
  
  # VM defaults
  DEFAULT_BOOT_TIMEOUT = 60      # seconds
  DEFAULT_CONNECT_TIMEOUT = 60   # seconds
  NON_KVM_ADDITIONAL_WAIT = 10       # seconds
  NON_KVM_TIMEOUT_COEF = 10          # times
  
  # SSH key type, set None for ssh-keygen default one
  SSH_KEYGEN_TYPE = "ecdsa"
  
  DEFAULT_USER = 'root'
  DEFAULT_MEMORY = 2048
  DEFAULT_DISK = 10
  DEFAULT_IMAGE = 'fedora'
  DEFAULT_CONNECTION = 'session'
  DEFAULT_ARCH = platform.machine()
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class TestcloudGuestData(
          tmt.steps.provision.GuestSshData):  # type: ignore[misc]
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      image: str = DEFAULT_IMAGE
      memory: int = DEFAULT_MEMORY
      disk: int = DEFAULT_DISK
      connection: str = DEFAULT_CONNECTION
      arch: str = DEFAULT_ARCH
  
      image_url: Optional[str] = None
      instance_name: Optional[str] = None
  
  
  @tmt.steps.provides_method('virtual.testcloud')
  class ProvisionTestcloud(tmt.steps.provision.ProvisionPlugin):
      """
      Local virtual machine using testcloud
  
      Minimal config which uses the latest fedora image:
  
          provision:
              how: virtual
  
      Here's a full config example:
  
          provision:
              how: virtual
              image: fedora
              user: root
              memory: 2048
  
      As the image use 'fedora' for the latest released Fedora compose,
      'rawhide' for the latest Rawhide compose, short aliases such as
      'fedora-32', 'f-32' or 'f32' for specific release or a full url to
      the qcow2 image for example from:
  
          https://kojipkgs.fedoraproject.org/compose/
  
      Short names are also provided for 'centos', 'centos-stream',
      'debian' and 'ubuntu' (e.g. 'centos-8' or 'c8').
  
      Supported Fedora CoreOS images are:
  
          fedora-coreos
          fedora-coreos-stable
          fedora-coreos-testing
          fedora-coreos-next
  
      Use the full path for images stored on local disk, for example:
  
          /var/tmp/images/Fedora-Cloud-Base-31-1.9.x86_64.qcow2
  
      In addition to the qcow2 format, vagrant boxes can be used as well,
      testcloud will take care of unpacking the image for you.
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["image", "user", "memory", "disk", "connection", "arch"]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for testcloud """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to be used. Provide a short name, '
                       'full path to a local file or a complete url.'),
              click.option(
                  '-m', '--memory', metavar='MEMORY', type=int,
                  help='Set available memory in MB, 2048 MB by default.'),
              click.option(
                  '-D', '--disk', metavar='MEMORY', type=int,
                  help='Specify disk size in GB, 10 GB by default.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-c', '--connection',
                  type=click.Choice(['session', 'system']),
                  help="What session type to use, 'session' by default."),
              click.option(
                  '-a', '--arch',
                  type=click.Choice(['x86_64', 'aarch64', 's390x', 'ppc64le']),
                  help="What architecture to virtualize, host arch by default."),
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          return
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/testcloud.py
files.old_contents: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  from typing import Optional
  
  import click
  import requests
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  
  def import_testcloud():
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
    - {public_key}
  ssh_pwauth: true
  disable_root: false
  runcmd:
    - sed -i -e '/^.*PermitRootLogin/s/^.*$/PermitRootLogin yes/'
      -e '/^.*UseDNS/s/^.*$/UseDNS no/'
      -e '/^.*GSSAPIAuthentication/s/^.*$/GSSAPIAuthentication no/'
      /etc/ssh/sshd_config
    - systemctl reload sshd
    - [sh, -c, 'if [ ! -f /etc/systemd/network/20-tc-usernet.network ] &&
    systemctl status systemd-networkd | grep -q "enabled;\\svendor\\spreset:\\senabled";
    then mkdir -p /etc/systemd/network/ &&
    echo "[Match]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "Name=en*" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "[Network]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "DHCP=yes" >> /etc/systemd/network/20-tc-usernet.network; fi']
    - [sh, -c, 'if systemctl status systemd-networkd |
    grep -q "enabled;\\svendor\\spreset:\\senabled"; then
    systemctl restart systemd-networkd; fi']
    - [sh, -c, 'if cat /etc/os-release |
    grep -q platform:el8; then systemctl restart sshd; fi']
  """
  
  COREOS_DATA = """variant: fcos
  version: 1.4.0
  passwd:
    users:
      - name: {user_name}
        ssh_authorized_keys:
          - {public_key}
  """
  
  # Libvirt domain XML template related variables
  DOMAIN_TEMPLATE_NAME = 'domain-template.jinja'
  DOMAIN_TEMPLATE_FILE = os.path.join(TESTCLOUD_DATA, DOMAIN_TEMPLATE_NAME)
  DOMAIN_TEMPLATE = """<domain type='{{ virt_type }}' xmlns:qemu='http://libvirt.org/schemas/domain/qemu/1.0'>
    <name>{{ domain_name }}</name>
    <uuid>{{ uuid }}</uuid>
    <memory unit='KiB'>{{ memory }}</memory>
    <currentMemory unit='KiB'>{{ memory }}</currentMemory>
    <vcpu placement='static'>2</vcpu>
    <os>
      <type arch='{{ arch }}' machine='{{ model }}'>hvm</type>
      {{ uefi_loader }}
      <boot dev='hd'/>
    </os>
    {{ cpu }}
    {{ extra_specs }}
    <clock offset='utc'>
      <timer name='rtc' tickpolicy='catchup'/>
      <timer name='pit' tickpolicy='delay'/>
      <timer name='hpet' present='no'/>
    </clock>
    <on_poweroff>destroy</on_poweroff>
    <on_reboot>restart</on_reboot>
    <on_crash>restart</on_crash>
    <devices>
      <emulator>{{ emulator_path }}</emulator>
      <disk type='file' device='disk'>
        <driver name='qemu' type='qcow2' cache='unsafe'/>
        <source file="{{ disk }}"/>
        <target dev='vda' bus='virtio'/>
        {{ boot_drive_address }}
      </disk>
      <disk type='file' device='disk'>
        <driver name='qemu' type='raw'/>
        <source file="{{ seed }}"/>
        <target dev='vdb' bus='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x08' function='0x0'/>
      </disk>
      <interface type='{{ network_type }}'>
        <mac address="{{ mac_address }}"/>
        {{ network_source }}
        {{ ip_setup }}
        <model type='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>
      </interface>
      <serial type='pty'>
        <target port='0'/>
      </serial>
      <console type='pty'>
        <target type='serial' port='0'/>
      </console>
      <input type="keyboard" bus="virtio"/>
      <rng model='virtio'>
        <backend model='random'>/dev/urandom</backend>
      </rng>
    </devices>
    {{ qemu_args }}
  </domain>
  """  # noqa: E501
  
  # VM defaults
  DEFAULT_BOOT_TIMEOUT = 60      # seconds
  DEFAULT_CONNECT_TIMEOUT = 60   # seconds
  NON_KVM_ADDITIONAL_WAIT = 10       # seconds
  NON_KVM_TIMEOUT_COEF = 10          # times
  
  # SSH key type, set None for ssh-keygen default one
  SSH_KEYGEN_TYPE = "ecdsa"
  
  DEFAULT_USER = 'root'
  DEFAULT_MEMORY = 2048
  DEFAULT_DISK = 10
  DEFAULT_IMAGE = 'fedora'
  DEFAULT_CONNECTION = 'session'
  DEFAULT_ARCH = platform.machine()
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class TestcloudGuestData(
          tmt.steps.provision.GuestSshData):  # type: ignore[misc]
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      image: str = DEFAULT_IMAGE
      memory: int = DEFAULT_MEMORY
      disk: int = DEFAULT_DISK
      connection: str = DEFAULT_CONNECTION
      arch: str = DEFAULT_ARCH
  
      image_url: Optional[str] = None
      instance_name: Optional[str] = None
  
  
  @tmt.steps.provides_method('virtual.testcloud')
  class ProvisionTestcloud(tmt.steps.provision.ProvisionPlugin):
      """
      Local virtual machine using testcloud
  
      Minimal config which uses the latest fedora image:
  
          provision:
              how: virtual
  
      Here's a full config example:
  
          provision:
              how: virtual
              image: fedora
              user: root
              memory: 2048
  
      As the image use 'fedora' for the latest released Fedora compose,
      'rawhide' for the latest Rawhide compose, short aliases such as
      'fedora-32', 'f-32' or 'f32' for specific release or a full url to
      the qcow2 image for example from:
  
          https://kojipkgs.fedoraproject.org/compose/
  
      Short names are also provided for 'centos', 'centos-stream',
      'debian' and 'ubuntu' (e.g. 'centos-8' or 'c8').
  
      Supported Fedora CoreOS images are:
  
          fedora-coreos
          fedora-coreos-stable
          fedora-coreos-testing
          fedora-coreos-next
  
      Use the full path for images stored on local disk, for example:
  
          /var/tmp/images/Fedora-Cloud-Base-31-1.9.x86_64.qcow2
  
      In addition to the qcow2 format, vagrant boxes can be used as well,
      testcloud will take care of unpacking the image for you.
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["image", "user", "memory", "disk", "connection", "arch"]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for testcloud """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to be used. Provide a short name, '
                       'full path to a local file or a complete url.'),
              click.option(
                  '-m', '--memory', metavar='MEMORY', type=int,
                  help='Set available memory in MB, 2048 MB by default.'),
              click.option(
                  '-D', '--disk', metavar='MEMORY', type=int,
                  help='Specify disk size in GB, 10 GB by default.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-c', '--connection',
                  type=click.Choice(['session', 'system']),
                  help="What session type to use, 'session' by default."),
              click.option(
                  '-a', '--arch',
                  type=click.Choice(['x86_64', 'aarch64', 's390x', 'ppc64le']),
                  help="What architecture to virtualize, host arch by default."),
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          return
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/testcloud.py