files: 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 TYPE_CHECKING, 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
  from tmt.utils import BaseLoggerFnType
  
  if TYPE_CHECKING:
      import tmt.cli
  
  
  # 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 = [
      "-s", "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  DEFAULT_RSYNC_PUSH_OPTIONS = ["-s", "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  DEFAULT_RSYNC_PULL_OPTIONS = ["-s", "-R", "-r", "-z", "--links", "--safe-links", "--protect-args"]
  
  
  class CheckRsyncOutcome(enum.Enum):
      ALREADY_INSTALLED = 'already-installed'
      INSTALLED = 'installed'
  
  
  @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: Type[GuestData] = GuestData
  
      role: Optional[str]
      guest: Optional[str]
  
      # Flag to indicate localhost guest, requires special handling
      localhost = False
  
      # TODO: do we need this list? Can whatever code is using it use _data_class directly?
      # List of supported keys
      # (used for import/export to/from attributes during load and save)
      @property
      def _keys(self) -> List[str]:
          return list(self._data_class.keys())
  
      def __init__(self,
                   *,
                   data: GuestData,
                   name: Optional[str] = None,
                   parent: Optional[tmt.utils.Common] = None) -> None:
          """ Initialize guest data """
          super().__init__(parent=parent, name=name)
  
          self.load(data)
  
      def _random_name(self, prefix: str = '', length: int = 16) -> str:
          """ Generate a random name """
          # Append at least 5 random characters
          min_random_part = max(5, length - len(prefix))
          name = prefix + ''.join(
              random.choices(string.ascii_letters, k=min_random_part))
          # Return tail (containing random characters) of name
          return name[-length:]
  
      def _tmt_name(self) -> str:
          """ Generate a name prefixed with tmt run id """
          # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
          parent = cast(Provision, self.parent)
  
          assert parent.plan.my_run is not None  # narrow type
          assert parent.plan.my_run.workdir is not None  # narrow type
          _, run_id = os.path.split(parent.plan.my_run.workdir)
          return self._random_name(prefix="tmt-{0}-".format(run_id[-3:]))
  
      @property
      def is_ready(self) -> bool:
          """ Detect guest is ready or not """
  
          raise NotImplementedError()
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options related to guests """
          return list()
  
      def load(self, data: GuestData) -> None:
          """
          Load guest data into object attributes for easy access
  
          Called during guest object initialization. Takes care of storing
          all supported keys (see class attribute _keys for the list) from
          provided data to the guest object attributes. Child classes can
          extend it to make additional guest attributes easily available.
  
          Data dictionary can contain guest information from both command
          line options / L2 metadata / user configuration and wake up data
          stored by the save() method below.
          """
          data.inject_to(self)
  
      def save(self) -> GuestData:
          """
          Save guest data for future wake up
  
          Export all essential guest data into a dictionary which will be
          stored in the `guests.yaml` file for possible future wake up of
          the guest. Everything needed to attach to a running instance
          should be added into the data dictionary by child classes.
          """
          return self._data_class.extract_from(self)
  
      def wake(self) -> None:
          """
          Wake up the guest
  
          Perform any actions necessary after step wake up to be able to
          attach to a running guest instance and execute commands. Called
          after load() is completed so all guest data should be prepared.
          """
          self.debug(f"Doing nothing to wake up guest '{self.guest}'.")
  
      def start(self) -> None:
          """
          Start the guest
  
          Get a new guest instance running. This should include preparing
          any configuration necessary to get it started. Called after
          load() is completed so all guest data should be available.
          """
          self.debug(f"Doing nothing to start guest '{self.guest}'.")
  
      def details(self) -> None:
          """ Show guest details such as distro and kernel """
          # Skip distro & kernel check in dry mode
          if self.opt('dry'):
              return
  
          # A small helper to make the repeated run & extract combo easier on eyes.
          def _fetch_detail(command: str, pattern: str) -> str:
              output = self.execute(command, silent=True)
  
              if not output.stdout:
                  raise tmt.utils.RunError(
                      'command produced no usable output',
                      command,
                      0,
                      output.stdout,
                      output.stderr)
  
              match = re.search(pattern, output.stdout)
  
              if not match:
                  raise tmt.utils.RunError(
                      'command produced no usable output',
                      command,
                      0,
                      output.stdout,
                      output.stderr)
  
              return match.group(1)
  
          # Distro
          distro_commands = [
              # Check os-release first
              ('cat /etc/os-release', r'PRETTY_NAME="(.*)"'),
              # Check for lsb-release
              ('cat /etc/lsb-release', r'DISTRIB_DESCRIPTION="(.*)"'),
              # Check for redhat-release
              ('cat /etc/redhat-release', r'(.*)')
              ]
  
          for command, pattern in distro_commands:
              try:
                  distro: Optional[str] = _fetch_detail(command, pattern)
                  break
              except tmt.utils.RunError:
                  continue
          else:
              distro = None
  
          # Handle standard cloud images message when connecting
          if distro is not None and 'Please login as the user' in distro:
              raise tmt.utils.GeneralError(
                  f'Login to the guest failed.\n{distro}')
  
          if distro:
              self.info('distro', distro, 'green')
  
          # Kernel
          kernel = _fetch_detail('uname -r', r'(.+)')
          self.verbose('kernel', kernel, 'green')
  
      def _ansible_verbosity(self) -> List[str]:
          """ Prepare verbose level based on the --debug option count """
          if self.opt('debug') < 3:
              return []
          else:
              return ['-' + (self.opt('debug') - 2) * 'v']
  
      @staticmethod
      def _ansible_extra_args(extra_args: Optional[str]) -> List[str]:
          """ Prepare extra arguments for ansible-playbook"""
          if extra_args is None:
              return []
          else:
              return shlex.split(str(extra_args))
  
      def _ansible_summary(self, output: Optional[str]) -> None:
          """ Check the output for ansible result summary numbers """
          if not output:
              return
          keys = 'ok changed unreachable failed skipped rescued ignored'.split()
          for key in keys:
              matched = re.search(rf'^.*\s:\s.*{key}=(\d+).*$', output, re.M)
              if matched and int(matched.group(1)) > 0:
                  tasks = fmf.utils.listed(matched.group(1), 'task')
                  self.verbose(key, tasks, 'green')
  
      def _ansible_playbook_path(self, playbook: str) -> str:
          """ Prepare full ansible playbook path """
          # Playbook paths should be relative to the metadata tree root
          self.debug(f"Applying playbook '{playbook}' on guest '{self.guest}'.")
          # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
          parent = cast(Provision, self.parent)
          assert parent.plan.my_run is not None  # narrow type
          assert parent.plan.my_run.tree is not None  # narrow type
          assert parent.plan.my_run.tree.root is not None  # narrow type
          playbook = os.path.join(parent.plan.my_run.tree.root, playbook)
          self.debug(f"Playbook full path: '{playbook}'", level=2)
          return playbook
  
      def _prepare_environment(
          self,
          execute_environment: Optional[tmt.utils.EnvironmentType] = None
              ) -> tmt.utils.EnvironmentType:
          """ Prepare dict of environment variables """
          # Prepare environment variables so they can be correctly passed
          # to shell. Create a copy to prevent modifying source.
          environment: tmt.utils.EnvironmentType = dict()
          environment.update(execute_environment or dict())
          # Plan environment and variables provided on the command line
          # override environment provided to execute().
          # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
          parent = cast(Provision, self.parent)
          environment.update(parent.plan.environment)
          return environment
  
      @staticmethod
      def _export_environment(environment: tmt.utils.EnvironmentType) -> str:
          """ Prepare shell export of environment variables """
          if not environment:
              return ""
          return f'export {" ".join(tmt.utils.shell_variables(environment))}; '
  
      def ansible(self, playbook: str, extra_args: Optional[str] = None) -> None:
          """ Prepare guest using ansible playbook """
  
          raise NotImplementedError()
  
      def execute(self,
                  command: Union[str, List[str]],
                  friendly_command: Optional[str] = None,
                  test_session: bool = False,
                  silent: bool = False,
                  log: Optional[BaseLoggerFnType] = None,
                  **kwargs: Any) -> tmt.utils.CommandOutput:
          """
          Execute command on the guest
  
          command ... string or list of command arguments (required)
          env ....... dictionary with environment variables
          cwd ....... working directory to be entered before execution
  
          If the command is provided as a list, it will be space-joined.
          If necessary, quote escaping has to be handled by the caller.
          """
  
          raise NotImplementedError()
  
      def push(self,
               source: Optional[str] = None,
               destination: Optional[str] = None,
               options: Optional[List[str]] = None) -> None:
          """
          Push files to the guest
          """
  
          raise NotImplementedError()
  
      def pull(self,
               source: Optional[str] = None,
               destination: Optional[str] = None,
               options: Optional[List[str]] = None,
               extend_options: Optional[List[str]] = None) -> None:
          """
          Pull files from the guest
          """
  
          raise NotImplementedError()
  
      def stop(self) -> None:
          """
          Stop the guest
  
          Shut down a running guest instance so that it does not consume
          any memory or cpu resources. If needed, perform any actions
          necessary to store the instance status to disk.
          """
  
          raise NotImplementedError()
  
      def reboot(
              self,
              hard: bool = False,
              command: Optional[str] = None,
              timeout: Optional[int] = None) -> bool:
          """
          Reboot the guest, return True if successful
  
          Parameter 'hard' set to True means that guest should be
          rebooted by way which is not clean in sense that data can be
          lost. When set to False reboot should be done gracefully.
  
          Use the 'command' parameter to specify a custom reboot command
          instead of the default 'reboot'.
  
          Parameter 'timeout' can be used to specify time (in seconds) to
          wait for the guest to come back up after rebooting.
          """
  
          raise NotImplementedError()
  
      def reconnect(
              self,
              timeout: Optional[int] = None,
              tick: float = RECONNECT_WAIT_TICK,
              tick_increase: float = RECONNECT_WAIT_TICK_INCREASE
              ) -> bool:
          """
          Ensure the connection to the guest is working
  
          The default timeout is 5 minutes. Custom number of seconds can be
          provided in the `timeout` parameter. This may be useful when long
          operations (such as system upgrade) are performed.
          """
          # The default is handled here rather than in the argument so that
          # the caller can pass in None as an argument (i.e. don't care value)
          timeout = timeout or CONNECTION_TIMEOUT
          self.debug("Wait for a connection to the guest.")
  
          def try_whoami() -> None:
              try:
                  self.execute('whoami', silent=True)
  
              except tmt.utils.RunError:
                  raise tmt.utils.WaitingIncomplete()
  
          try:
              tmt.utils.wait(
                  self,
                  try_whoami,
                  datetime.timedelta(seconds=timeout),
                  tick=tick,
                  tick_increase=tick_increase)
  
          except tmt.utils.WaitingTimedOutError:
              self.debug("Connection to guest failed after reboot.")
              return False
  
          return True
  
      def remove(self) -> None:
          """
          Remove the guest
  
          Completely remove all guest instance data so that it does not
          consume any disk resources.
          """
          self.debug(f"Doing nothing to remove guest '{self.guest}'.")
  
      def _check_rsync(self) -> CheckRsyncOutcome:
          """
          Make sure that rsync is installed on the guest
  
          On read-only distros install it under the '/root/pkg' directory.
          Returns 'already installed' when rsync is already present.
          """
  
          # Check for rsync (nothing to do if already installed)
          self.debug("Ensure that rsync is installed on the guest.")
          try:
              self.execute("rsync --version")
              return CheckRsyncOutcome.ALREADY_INSTALLED
          except tmt.utils.RunError:
              pass
  
          # Check the package manager
          self.debug("Check the package manager.")
          try:
              self.execute("dnf --version")
              package_manager = "dnf"
          except tmt.utils.RunError:
              package_manager = "yum"
  
          # Install under '/root/pkg' for read-only distros
          # (for now the check is based on 'rpm-ostree' presence)
          # FIXME: Find a better way how to detect read-only distros
          self.debug("Check for a read-only distro.")
          try:
              self.execute("rpm-ostree --version")
              readonly = (
                  " --installroot=/root/pkg --releasever / "
                  "&& ln -sf /root/pkg/bin/rsync /usr/local/bin/rsync")
          except tmt.utils.RunError:
              readonly = ""
  
          # Install the rsync
          self.execute(f"{package_manager} install -y rsync" + readonly)
  
          return CheckRsyncOutcome.INSTALLED
  
      @classmethod
      def requires(cls) -> List[str]:
          """ No extra requires needed """
          return []
  
  
  @dataclasses.dataclass
  class GuestSshData(GuestData):
      """
      Keys necessary to describe, create, save and restore a guest with SSH
      capability.
  
      Derived from GuestData, this class adds keys relevant for guests that can be
      reached over SSH.
      """
  
      # port to connect to
      port: Optional[int] = None
      # user name to log in
      user: Optional[str] = None
      # path to the private key
      key: List[str] = dataclasses.field(default_factory=list)
      # password
      password: Optional[str] = None
      ssh_option: List[str] = dataclasses.field(default_factory=list)
  
  
  class GuestSsh(Guest):
      """
      Guest provisioned for test execution, capable of accepting SSH connections
  
      The following keys are expected in the 'data' dictionary::
  
          role ....... guest role in the multihost scenario (inherited)
          guest ...... hostname or ip address (inherited)
          port ....... port to connect to
          user ....... user name to log in
          key ........ path to the private key (str or list)
          password ... password
  
      These are by default imported into instance attributes.
      """
  
      _data_class: Type[GuestData] = GuestSshData
  
      port: Optional[int]
      user: Optional[str]
      key: List[str]
      password: Optional[str]
      ssh_option: List[str]
  
      # Master ssh connection process and socket path
      _ssh_master_process: Optional['subprocess.Popen[bytes]'] = None
      _ssh_socket_path: Optional[str] = None
  
      def _ssh_guest(self) -> str:
          """ Return user@guest """
          return f'{self.user}@{self.guest}'
  
      def _ssh_socket(self) -> str:
          """ Prepare path to the master connection socket """
          if not self._ssh_socket_path:
              # Use '/run/user/uid' if it exists, '/tmp' otherwise
              run_dir = f"/run/user/{os.getuid()}"
              if os.path.isdir(run_dir):
                  socket_dir = os.path.join(run_dir, "tmt")
              else:
                  socket_dir = "/tmp"
              os.makedirs(socket_dir, exist_ok=True)
              self._ssh_socket_path = tempfile.mktemp(dir=socket_dir)
          return self._ssh_socket_path
  
      def _ssh_options(self, join: bool = False) -> Union[str, List[str]]:
          """ Return common ssh options (list or joined) """
          options = [
              '-oForwardX11=no',
              '-oStrictHostKeyChecking=no',
              '-oUserKnownHostsFile=/dev/null',
              # Prevent ssh from disconnecting if no data has been
              # received from the server for a long time (#868).
              '-oServerAliveInterval=60',
              '-oServerAliveCountMax=5',
              ]
          if self.key or self.password:
              # Skip ssh-agent (it adds additional identities)
              options.append('-oIdentitiesOnly=yes')
          if self.port:
              options.append(f'-p{self.port}')
          if self.key:
              for key in self.key:
                  options.append(f'-i{shlex.quote(key) if join else key}')
          if self.password:
              options.extend(['-oPasswordAuthentication=yes'])
  
          # Use the shared master connection
          options.append(f'-S{self._ssh_socket()}')
  
          options.extend([f'-o{option}' for option in self.ssh_option])
  
          return ' '.join(options) if join else options
  
      def _ssh_master_connection(self, command: List[str]) -> None:
          """ Check/create the master ssh connection """
          if self._ssh_master_process:
              return
  
          # TODO: _ssh_options() returns two incompatible types, we need to chose the right one.
          ssh_options = self._ssh_options()
          assert isinstance(ssh_options, list)
  
          command = command + ssh_options + ["-MNnT", self._ssh_guest()]
          self.debug(f"Create the master ssh connection: {' '.join(command)}")
          self._ssh_master_process = subprocess.Popen(
              command,
              stdin=subprocess.DEVNULL,
              stdout=subprocess.DEVNULL,
              stderr=subprocess.DEVNULL)
  
      def _ssh_command(self, join: bool = False) -> Union[str, List[str]]:
          """ Prepare an ssh command line for execution (list or joined) """
          command = []
          if self.password:
              password = shlex.quote(self.password) if join else self.password
              command.extend(["sshpass", "-p", password])
          command.append("ssh")
  
          # Check the master connection
          self._ssh_master_connection(command)
  
          if join:
              # TODO: _ssh_options() returns two incompatible types, we need to chose the right one.
              joined_ssh_options = self._ssh_options(join=True)
              assert isinstance(joined_ssh_options, str)
  
              return " ".join(command) + " " + joined_ssh_options
          else:
              # TODO: _ssh_options() returns two incompatible types, we need to chose the right one.
              ssh_options = self._ssh_options()
              assert isinstance(ssh_options, list)
  
              return command + ssh_options
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options related to SSH-capable guests """
          return super().options(how=how) + [
              click.option('--ssh-option', metavar="OPTION", multiple=True, default=[],
                           help="Specify additional SSH option. "
                                "Value is passed to SSH's -o option, see ssh_config(5) for "
                                "supported options. Can be specified multiple times.")]
  
      def ansible(self, playbook: str, extra_args: Optional[str] = None) -> None:
          """ Prepare guest using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
  
          # TODO: _ssh_options() returns two incompatible types, we need to chose the right one.
          joined_ssh_options = self._ssh_options(join=True)
          assert isinstance(joined_ssh_options, str)
  
          ansible_command = [
              'ansible-playbook'
              ] + self._ansible_verbosity()
  
          if extra_args:
              ansible_command += self._ansible_extra_args(extra_args)
  
          ansible_command += [
              f'--ssh-common-args={joined_ssh_options}',
              '-i', f'{self._ssh_guest()},',
              playbook
              ]
  
          # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
          parent = cast(Provision, self.parent)
  
          stdout, _ = self.run(
              ansible_command,
              cwd=parent.plan.worktree,
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      @property
      def is_ready(self) -> bool:
          """ Detect guest is ready or not """
  
          # Enough for now, ssh connection can be created later
          return self.guest is not None
  
      def execute(self,
                  command: Union[str, List[str]],
                  friendly_command: Optional[str] = None,
                  test_session: bool = False,
                  silent: bool = False,
                  log: Optional[BaseLoggerFnType] = None,
                  **kwargs: Any) -> tmt.utils.CommandOutput:
          """
          Execute command on the guest
  
          command ... string or list of command arguments (required)
          env ....... dictionary with environment variables
          cwd ....... working directory to be entered before execution
  
          If the command is provided as a list, it will be space-joined.
          If necessary, quote escaping has to be handled by the caller.
          """
          # Abort if guest is unavailable
          if self.guest is None:
              if not self.opt('dry'):
                  raise tmt.utils.GeneralError('The guest is not available.')
  
          # Prepare the export of environment variables
          environment = self._export_environment(
              self._prepare_environment(kwargs.get('env', dict())))
  
          # Change to given directory on guest if cwd provided
          directory = kwargs.get('cwd') or ''
          if directory:
              directory = f"cd {quote(directory)}; "
  
          # Run in interactive mode if requested
          interactive = ['-t'] if kwargs.get('interactive') else []
  
          # Force ssh to allocate pseudo-terminal if requested. Without a pseudo-terminal,
          # remote processes spawned by SSH would keep running after SSH process death, e.g.
          # in the case of a timeout.
          #
          # Note that polite request, `-t`, is not enough since `ssh` itself has no pseudo-terminal,
          # and a single `-t` wouldn't have the necessary effect.
          ptty = ['-tt'] if test_session else []
  
          # Prepare command and run it
          if isinstance(command, (list, tuple)):
              command = ' '.join(command)
          self.debug(f"Execute command '{command}' on guest '{self.guest}'.")
          friendly_command = friendly_command or command
  
          # TODO: _ssh_command() returns two incompatible types, we need to chose the right one.
          ssh_command = self._ssh_command()
          assert isinstance(ssh_command, list)
  
          command = (
              ssh_command + interactive + ptty + [self._ssh_guest()] +
              [f'{environment}{directory}{command}'])
          return self.run(command,
                          log=log if log else self._command_verbose_logger,
                          friendly_command=friendly_command,
                          silent=silent,
                          **kwargs)
  
      def push(self,
               source: Optional[str] = None,
               destination: Optional[str] = None,
               options: Optional[List[str]] = None) -> None:
          """
          Push files to the guest
  
          By default the whole plan workdir is synced to the same location
          on the guest. Use the 'source' and 'destination' to sync custom
          location and the 'options' parametr to modify default options
          which are '-Rrz --links --safe-links --delete'.
          """
          # Abort if guest is unavailable
          if self.guest is None:
              if not self.opt('dry'):
                  raise tmt.utils.GeneralError('The guest is not available.')
  
          # Prepare options and the push command
          options = options or DEFAULT_RSYNC_PUSH_OPTIONS
          if destination is None:
              destination = "/"
          if source is None:
              # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
              parent = cast(Provision, self.parent)
  
              assert parent.plan.workdir is not None
  
              source = parent.plan.workdir
              self.debug(f"Push workdir to guest '{self.guest}'.")
          else:
              self.debug(f"Copy '{source}' to '{destination}' on the guest.")
  
          def rsync() -> None:
              """ Run the rsync command """
              # In closure, mypy has hard times to reason about the state of used variables.
              assert options
              assert source
              assert destination
  
              # TODO: _ssh_command() returns two incompatible types, we need to chose the right one.
              joined_ssh_command = self._ssh_command(join=True)
              assert isinstance(joined_ssh_command, str)
  
              self.run(
                  ["rsync"] + options
                  + ["-e", joined_ssh_command]
                  + [source, f"{self._ssh_guest()}:{destination}"])
  
          # Try to push twice, check for rsync after the first failure
          try:
              rsync()
          except tmt.utils.RunError:
              try:
                  if self._check_rsync() == CheckRsyncOutcome.ALREADY_INSTALLED:
                      raise
                  rsync()
              except tmt.utils.RunError:
                  # Provide a reasonable error to the user
                  self.fail(
                      f"Failed to push workdir to the guest. This usually means "
                      f"that login as '{self.user}' to the guest does not work.")
                  raise
  
      def pull(self,
               source: Optional[str] = None,
               destination: Optional[str] = None,
               options: Optional[List[str]] = None,
               extend_options: Optional[List[str]] = None) -> None:
          """
          Pull files from the guest
  
          By default the whole plan workdir is synced from the same
          location on the guest. Use the 'source' and 'destination' to
          sync custom location, the 'options' parameter to modify
          default options :py:data:`DEFAULT_RSYNC_PULL_OPTIONS`
          and 'extend_options' to extend them (e.g. by exclude).
          """
          # Abort if guest is unavailable
          if self.guest is None:
              if not self.opt('dry'):
                  raise tmt.utils.GeneralError('The guest is not available.')
  
          # Prepare options and the pull command
          options = options or DEFAULT_RSYNC_PULL_OPTIONS
          if extend_options is not None:
              options.extend(extend_options)
          if destination is None:
              destination = "/"
          if source is None:
              # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
              parent = cast(Provision, self.parent)
  
              assert parent.plan.workdir is not None
  
              source = parent.plan.workdir
              self.debug(f"Pull workdir from guest '{self.guest}'.")
          else:
              self.debug(f"Copy '{source}' from the guest to '{destination}'.")
  
          def rsync() -> None:
              """ Run the rsync command """
              # In closure, mypy has hard times to reason about the state of used variables.
              assert options
              assert source
              assert destination
  
              # TODO: _ssh_command() returns two incompatible types, we need to chose the right one.
              joined_ssh_command = self._ssh_command(join=True)
              assert isinstance(joined_ssh_command, str)
  
              self.run(
                  ["rsync"] + options
                  + ["-e", joined_ssh_command]
                  + [f"{self._ssh_guest()}:{source}", destination])
  
          # Try to pull twice, check for rsync after the first failure
          try:
              rsync()
          except tmt.utils.RunError:
              try:
                  if self._check_rsync() == CheckRsyncOutcome.ALREADY_INSTALLED:
                      raise
                  rsync()
              except tmt.utils.RunError:
                  # Provide a reasonable error to the user
                  self.fail(
                      f"Failed to pull workdir from the guest. "
                      f"This usually means that login as '{self.user}' "
                      f"to the guest does not work.")
                  raise
  
      def stop(self) -> None:
          """
          Stop the guest
  
          Shut down a running guest instance so that it does not consume
          any memory or cpu resources. If needed, perform any actions
          necessary to store the instance status to disk.
          """
  
          # Close the master ssh connection
          if self._ssh_master_process:
              self.debug("Close the master ssh connection.", level=3)
              try:
                  self._ssh_master_process.terminate()
                  self._ssh_master_process.wait(timeout=3)
              except subprocess.TimeoutExpired:
                  pass
  
          # Remove the ssh socket
          if self._ssh_socket_path and os.path.exists(self._ssh_socket_path):
              self.debug(
                  f"Remove ssh socket '{self._ssh_socket_path}'.", level=3)
              try:
                  os.unlink(self._ssh_socket_path)
              except OSError as error:
                  self.debug(f"Failed to remove the socket: {error}", level=3)
  
      def reboot(
              self,
              hard: bool = False,
              command: Optional[str] = None,
              timeout: Optional[int] = None,
              tick: float = tmt.utils.DEFAULT_WAIT_TICK,
              tick_increase: float = tmt.utils.DEFAULT_WAIT_TICK_INCREASE) -> bool:
          """
          Reboot the guest, return True if reconnect was successful
  
          Parameter 'hard' set to True means that guest should be
          rebooted by way which is not clean in sense that data can be
          lost. When set to False reboot should be done gracefully.
  
          Use the 'command' parameter to specify a custom reboot command
          instead of the default 'reboot'.
          """
  
          if hard:
              raise tmt.utils.ProvisionError(
                  "Method does not support hard reboot.")
  
          command = command or "reboot"
          self.debug(f"Reboot using the command '{command}'.")
  
          re_boot_time = re.compile(r'btime\s+(\d+)')
  
          def get_boot_time() -> int:
              """ Reads btime from /proc/stat """
              stdout = self.execute(["cat", "/proc/stat"]).stdout
              assert stdout
  
              match = re_boot_time.search(stdout)
  
              if match is None:
                  raise tmt.utils.ProvisionError('Failed to retrieve boot time from guest')
  
              return int(match.group(1))
  
          current_boot_time = get_boot_time()
  
          try:
              self.execute(command)
          except tmt.utils.RunError as error:
              # Connection can be closed by the remote host even before the
              # reboot command is completed. Let's ignore such errors.
              if error.returncode == 255:
                  self.debug(
                      "Seems the connection was closed too fast, ignoring.")
              else:
                  raise
  
          # Wait until we get new boot time, connection will drop and will be
          # unreachable for some time
          def check_boot_time() -> None:
              try:
                  new_boot_time = get_boot_time()
  
                  if new_boot_time != current_boot_time:
                      # Different boot time and we are reconnected
                      return
  
                  # Same boot time, reboot didn't happen yet, retrying
                  raise tmt.utils.WaitingIncomplete()
  
              except tmt.utils.RunError:
                  self.debug('Failed to connect to the guest.')
                  raise tmt.utils.WaitingIncomplete()
  
          timeout = timeout or CONNECTION_TIMEOUT
  
          try:
              tmt.utils.wait(
                  self,
                  check_boot_time,
                  datetime.timedelta(seconds=timeout),
                  tick=tick,
                  tick_increase=tick_increase)
  
          except tmt.utils.WaitingTimedOutError:
              self.debug("Connection to guest failed after reboot.")
              return False
  
          self.debug("Connection to guest succeeded after reboot.")
          return True
  
      def remove(self) -> None:
          """
          Remove the guest
  
          Completely remove all guest instance data so that it does not
          consume any disk resources.
          """
          self.debug(f"Doing nothing to remove guest '{self.guest}'.")
  
      def _check_rsync(self) -> CheckRsyncOutcome:
          """
          Make sure that rsync is installed on the guest
  
          On read-only distros install it under the '/root/pkg' directory.
          Returns 'already installed' when rsync is already present.
          """
  
          # Check for rsync (nothing to do if already installed)
          self.debug("Ensure that rsync is installed on the guest.")
          try:
              self.execute("rsync --version")
              return CheckRsyncOutcome.ALREADY_INSTALLED
          except tmt.utils.RunError:
              pass
  
          # Check the package manager
          self.debug("Check the package manager.")
          try:
              self.execute("dnf --version")
              package_manager = "dnf"
          except tmt.utils.RunError:
              package_manager = "yum"
  
          # Install under '/root/pkg' for read-only distros
          # (for now the check is based on 'rpm-ostree' presence)
          # FIXME: Find a better way how to detect read-only distros
          self.debug("Check for a read-only distro.")
          try:
              self.execute("rpm-ostree --version")
              readonly = (
                  " --installroot=/root/pkg --releasever / "
                  "&& ln -sf /root/pkg/bin/rsync /usr/local/bin/rsync")
          except tmt.utils.RunError:
              readonly = ""
  
          # Install the rsync
          self.execute(f"{package_manager} install -y rsync" + readonly)
  
          return CheckRsyncOutcome.INSTALLED
  
      @classmethod
      def requires(cls) -> List[str]:
          """ No extra requires needed """
          return []
  
  
  @dataclasses.dataclass
  class ProvisionStepData(tmt.steps.StepData):
      pass
  
  
  class ProvisionPlugin(tmt.steps.GuestlessPlugin):
      """ Common parent of provision plugins """
  
      _data_class = ProvisionStepData
      _guest_class = Guest
  
      # 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] = []
  
      # TODO: Generics would provide a better type, https://github.com/teemtee/tmt/issues/1437
      _guest: Optional[Guest] = None
  
      @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: 'tmt.cli.Context', **kwargs: Any) -> None:
              context.obj.steps.add('provision')
              Provision._save_context(context)
  
          return provision
  
      # TODO: this might be needed until https://github.com/teemtee/tmt/issues/1696 is resolved
      def opt(self, option: str, default: Optional[Any] = None) -> Any:
          """ Get an option from the command line options """
  
          if option == 'ssh-option':
              value = super().opt(option, default=default)
  
              if isinstance(value, tuple):
                  return list(value)
  
              return value
  
          return super().opt(option, default=default)
  
      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()
  
          if data is not None:
              guest = self._guest_class(data=data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      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 options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Return list of options. """
          return super().options(how) + cls._guest_class.options(how)
  
      @classmethod
      def clean_images(cls, clean: 'tmt.base.Clean', dry: bool) -> bool:
          """ Remove the images of one particular plugin """
          return True
  
  
  class Provision(tmt.steps.Step):
      """ Provision an environment for testing or use localhost. """
  
      # Default implementation for provision is a virtual machine
      DEFAULT_HOW = 'virtual'
  
      _plugin_base_class = ProvisionPlugin
  
      _preserved_files = ['step.yaml', 'guests.yaml']
  
      def __init__(self, *, plan: 'tmt.Plan', data: tmt.steps.RawStepDataArgument) -> None:
          """ Initialize provision step data """
          super().__init__(plan=plan, data=data)
  
          # List of provisioned guests and loaded guest data
          self._guests: List[Guest] = []
          self._guest_data: Dict[str, GuestData] = {}
          self.is_multihost = False
  
      def _normalize_data(self, raw_data: List[tmt.steps._RawStepData]) -> List[tmt.steps.StepData]:
          data = super()._normalize_data(raw_data)
  
          # Check that the names are unique
          names = [datum.name for datum in data]
          names_count = collections.Counter(names)
          if len(data) > 1 and names_count.most_common(1)[0][1] > 1:
              duplicate_names = [name for name in names_count.elements() if names_count[name] > 1]
              duplicate_string = ', '.join(duplicate_names)
              raise tmt.utils.GeneralError(
                  f"Provision step names must be unique for multihost testing. "
                  f"Duplicate names: '{duplicate_string}' in plan '{self.plan.name}'.")
  
          return data
  
      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:
              # FIXME: cast() - see https://github.com/teemtee/tmt/issues/1599
              plugin = cast(ProvisionPlugin, ProvisionPlugin.delegate(self, data=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 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('')
                  except (tmt.utils.RunError, tmt.utils.ProvisionError) as error:
                      self.fail(str(error))
                      raise
                  finally:
                      if isinstance(phase, ProvisionPlugin):
                          guest = phase.guest()
                          if guest and (guest.is_ready or self.opt('dry')):
                              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)
  
files: tmt/steps/provision/__init__.py
files: 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 TYPE_CHECKING, 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
  from tmt.utils import BaseLoggerFnType
  
  if TYPE_CHECKING:
      import tmt.cli
  
  
  # 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 = [
      "-s", "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  DEFAULT_RSYNC_PUSH_OPTIONS = ["-s", "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  DEFAULT_RSYNC_PULL_OPTIONS = ["-s", "-R", "-r", "-z", "--links", "--safe-links", "--protect-args"]
  
  
  class CheckRsyncOutcome(enum.Enum):
      ALREADY_INSTALLED = 'already-installed'
      INSTALLED = 'installed'
  
  
  @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: Type[GuestData] = GuestData
  
      role: Optional[str]
      guest: Optional[str]
  
      # Flag to indicate localhost guest, requires special handling
      localhost = False
  
      # TODO: do we need this list? Can whatever code is using it use _data_class directly?
      # List of supported keys
      # (used for import/export to/from attributes during load and save)
      @property
      def _keys(self) -> List[str]:
          return list(self._data_class.keys())
  
      def __init__(self,
                   *,
                   data: GuestData,
                   name: Optional[str] = None,
                   parent: Optional[tmt.utils.Common] = None) -> None:
          """ Initialize guest data """
          super().__init__(parent=parent, name=name)
  
          self.load(data)
  
      def _random_name(self, prefix: str = '', length: int = 16) -> str:
          """ Generate a random name """
          # Append at least 5 random characters
          min_random_part = max(5, length - len(prefix))
          name = prefix + ''.join(
              random.choices(string.ascii_letters, k=min_random_part))
          # Return tail (containing random characters) of name
          return name[-length:]
  
      def _tmt_name(self) -> str:
          """ Generate a name prefixed with tmt run id """
          # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
          parent = cast(Provision, self.parent)
  
          assert parent.plan.my_run is not None  # narrow type
          assert parent.plan.my_run.workdir is not None  # narrow type
          _, run_id = os.path.split(parent.plan.my_run.workdir)
          return self._random_name(prefix="tmt-{0}-".format(run_id[-3:]))
  
      @property
      def is_ready(self) -> bool:
          """ Detect guest is ready or not """
  
          raise NotImplementedError()
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options related to guests """
          return list()
  
      def load(self, data: GuestData) -> None:
          """
          Load guest data into object attributes for easy access
  
          Called during guest object initialization. Takes care of storing
          all supported keys (see class attribute _keys for the list) from
          provided data to the guest object attributes. Child classes can
          extend it to make additional guest attributes easily available.
  
          Data dictionary can contain guest information from both command
          line options / L2 metadata / user configuration and wake up data
          stored by the save() method below.
          """
          data.inject_to(self)
  
      def save(self) -> GuestData:
          """
          Save guest data for future wake up
  
          Export all essential guest data into a dictionary which will be
          stored in the `guests.yaml` file for possible future wake up of
          the guest. Everything needed to attach to a running instance
          should be added into the data dictionary by child classes.
          """
          return self._data_class.extract_from(self)
  
      def wake(self) -> None:
          """
          Wake up the guest
  
          Perform any actions necessary after step wake up to be able to
          attach to a running guest instance and execute commands. Called
          after load() is completed so all guest data should be prepared.
          """
          self.debug(f"Doing nothing to wake up guest '{self.guest}'.")
  
      def start(self) -> None:
          """
          Start the guest
  
          Get a new guest instance running. This should include preparing
          any configuration necessary to get it started. Called after
          load() is completed so all guest data should be available.
          """
          self.debug(f"Doing nothing to start guest '{self.guest}'.")
  
      def details(self) -> None:
          """ Show guest details such as distro and kernel """
          # Skip distro & kernel check in dry mode
          if self.opt('dry'):
              return
  
          # A small helper to make the repeated run & extract combo easier on eyes.
          def _fetch_detail(command: str, pattern: str) -> str:
              output = self.execute(command, silent=True)
  
              if not output.stdout:
                  raise tmt.utils.RunError(
                      'command produced no usable output',
                      command,
                      0,
                      output.stdout,
                      output.stderr)
  
              match = re.search(pattern, output.stdout)
  
              if not match:
                  raise tmt.utils.RunError(
                      'command produced no usable output',
                      command,
                      0,
                      output.stdout,
                      output.stderr)
  
              return match.group(1)
  
          # Distro
          distro_commands = [
              # Check os-release first
              ('cat /etc/os-release', r'PRETTY_NAME="(.*)"'),
              # Check for lsb-release
              ('cat /etc/lsb-release', r'DISTRIB_DESCRIPTION="(.*)"'),
              # Check for redhat-release
              ('cat /etc/redhat-release', r'(.*)')
              ]
  
          for command, pattern in distro_commands:
              try:
                  distro: Optional[str] = _fetch_detail(command, pattern)
                  break
              except tmt.utils.RunError:
                  continue
          else:
              distro = None
  
          # Handle standard cloud images message when connecting
          if distro is not None and 'Please login as the user' in distro:
              raise tmt.utils.GeneralError(
                  f'Login to the guest failed.\n{distro}')
  
          if distro:
              self.info('distro', distro, 'green')
  
          # Kernel
          kernel = _fetch_detail('uname -r', r'(.+)')
          self.verbose('kernel', kernel, 'green')
  
      def _ansible_verbosity(self) -> List[str]:
          """ Prepare verbose level based on the --debug option count """
          if self.opt('debug') < 3:
              return []
          else:
              return ['-' + (self.opt('debug') - 2) * 'v']
  
      @staticmethod
      def _ansible_extra_args(extra_args: Optional[str]) -> List[str]:
          """ Prepare extra arguments for ansible-playbook"""
          if extra_args is None:
              return []
          else:
              return shlex.split(str(extra_args))
  
      def _ansible_summary(self, output: Optional[str]) -> None:
          """ Check the output for ansible result summary numbers """
          if not output:
              return
          keys = 'ok changed unreachable failed skipped rescued ignored'.split()
          for key in keys:
              matched = re.search(rf'^.*\s:\s.*{key}=(\d+).*$', output, re.M)
              if matched and int(matched.group(1)) > 0:
                  tasks = fmf.utils.listed(matched.group(1), 'task')
                  self.verbose(key, tasks, 'green')
  
      def _ansible_playbook_path(self, playbook: str) -> str:
          """ Prepare full ansible playbook path """
          # Playbook paths should be relative to the metadata tree root
          self.debug(f"Applying playbook '{playbook}' on guest '{self.guest}'.")
          # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
          parent = cast(Provision, self.parent)
          assert parent.plan.my_run is not None  # narrow type
          assert parent.plan.my_run.tree is not None  # narrow type
          assert parent.plan.my_run.tree.root is not None  # narrow type
          playbook = os.path.join(parent.plan.my_run.tree.root, playbook)
          self.debug(f"Playbook full path: '{playbook}'", level=2)
          return playbook
  
      def _prepare_environment(
          self,
          execute_environment: Optional[tmt.utils.EnvironmentType] = None
              ) -> tmt.utils.EnvironmentType:
          """ Prepare dict of environment variables """
          # Prepare environment variables so they can be correctly passed
          # to shell. Create a copy to prevent modifying source.
          environment: tmt.utils.EnvironmentType = dict()
          environment.update(execute_environment or dict())
          # Plan environment and variables provided on the command line
          # override environment provided to execute().
          # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
          parent = cast(Provision, self.parent)
          environment.update(parent.plan.environment)
          return environment
  
      @staticmethod
      def _export_environment(environment: tmt.utils.EnvironmentType) -> str:
          """ Prepare shell export of environment variables """
          if not environment:
              return ""
          return f'export {" ".join(tmt.utils.shell_variables(environment))}; '
  
      def ansible(self, playbook: str, extra_args: Optional[str] = None) -> None:
          """ Prepare guest using ansible playbook """
  
          raise NotImplementedError()
  
      def execute(self,
                  command: Union[str, List[str]],
                  friendly_command: Optional[str] = None,
                  test_session: bool = False,
                  silent: bool = False,
                  log: Optional[BaseLoggerFnType] = None,
                  **kwargs: Any) -> tmt.utils.CommandOutput:
          """
          Execute command on the guest
  
          command ... string or list of command arguments (required)
          env ....... dictionary with environment variables
          cwd ....... working directory to be entered before execution
  
          If the command is provided as a list, it will be space-joined.
          If necessary, quote escaping has to be handled by the caller.
          """
  
          raise NotImplementedError()
  
      def push(self,
               source: Optional[str] = None,
               destination: Optional[str] = None,
               options: Optional[List[str]] = None) -> None:
          """
          Push files to the guest
          """
  
          raise NotImplementedError()
  
      def pull(self,
               source: Optional[str] = None,
               destination: Optional[str] = None,
               options: Optional[List[str]] = None,
               extend_options: Optional[List[str]] = None) -> None:
          """
          Pull files from the guest
          """
  
          raise NotImplementedError()
  
      def stop(self) -> None:
          """
          Stop the guest
  
          Shut down a running guest instance so that it does not consume
          any memory or cpu resources. If needed, perform any actions
          necessary to store the instance status to disk.
          """
  
          raise NotImplementedError()
  
      def reboot(
              self,
              hard: bool = False,
              command: Optional[str] = None,
              timeout: Optional[int] = None) -> bool:
          """
          Reboot the guest, return True if successful
  
          Parameter 'hard' set to True means that guest should be
          rebooted by way which is not clean in sense that data can be
          lost. When set to False reboot should be done gracefully.
  
          Use the 'command' parameter to specify a custom reboot command
          instead of the default 'reboot'.
  
          Parameter 'timeout' can be used to specify time (in seconds) to
          wait for the guest to come back up after rebooting.
          """
  
          raise NotImplementedError()
  
      def reconnect(
              self,
              timeout: Optional[int] = None,
              tick: float = RECONNECT_WAIT_TICK,
              tick_increase: float = RECONNECT_WAIT_TICK_INCREASE
              ) -> bool:
          """
          Ensure the connection to the guest is working
  
          The default timeout is 5 minutes. Custom number of seconds can be
          provided in the `timeout` parameter. This may be useful when long
          operations (such as system upgrade) are performed.
          """
          # The default is handled here rather than in the argument so that
          # the caller can pass in None as an argument (i.e. don't care value)
          timeout = timeout or CONNECTION_TIMEOUT
          self.debug("Wait for a connection to the guest.")
  
          def try_whoami() -> None:
              try:
                  self.execute('whoami', silent=True)
  
              except tmt.utils.RunError:
                  raise tmt.utils.WaitingIncomplete()
  
          try:
              tmt.utils.wait(
                  self,
                  try_whoami,
                  datetime.timedelta(seconds=timeout),
                  tick=tick,
                  tick_increase=tick_increase)
  
          except tmt.utils.WaitingTimedOutError:
              self.debug("Connection to guest failed after reboot.")
              return False
  
          return True
  
      def remove(self) -> None:
          """
          Remove the guest
  
          Completely remove all guest instance data so that it does not
          consume any disk resources.
          """
          self.debug(f"Doing nothing to remove guest '{self.guest}'.")
  
      def _check_rsync(self) -> CheckRsyncOutcome:
          """
          Make sure that rsync is installed on the guest
  
          On read-only distros install it under the '/root/pkg' directory.
          Returns 'already installed' when rsync is already present.
          """
  
          # Check for rsync (nothing to do if already installed)
          self.debug("Ensure that rsync is installed on the guest.")
          try:
              self.execute("rsync --version")
              return CheckRsyncOutcome.ALREADY_INSTALLED
          except tmt.utils.RunError:
              pass
  
          # Check the package manager
          self.debug("Check the package manager.")
          try:
              self.execute("dnf --version")
              package_manager = "dnf"
          except tmt.utils.RunError:
              package_manager = "yum"
  
          # Install under '/root/pkg' for read-only distros
          # (for now the check is based on 'rpm-ostree' presence)
          # FIXME: Find a better way how to detect read-only distros
          self.debug("Check for a read-only distro.")
          try:
              self.execute("rpm-ostree --version")
              readonly = (
                  " --installroot=/root/pkg --releasever / "
                  "&& ln -sf /root/pkg/bin/rsync /usr/local/bin/rsync")
          except tmt.utils.RunError:
              readonly = ""
  
          # Install the rsync
          self.execute(f"{package_manager} install -y rsync" + readonly)
  
          return CheckRsyncOutcome.INSTALLED
  
      @classmethod
      def requires(cls) -> List[str]:
          """ No extra requires needed """
          return []
  
  
  @dataclasses.dataclass
  class GuestSshData(GuestData):
      """
      Keys necessary to describe, create, save and restore a guest with SSH
      capability.
  
      Derived from GuestData, this class adds keys relevant for guests that can be
      reached over SSH.
      """
  
      # port to connect to
      port: Optional[int] = None
      # user name to log in
      user: Optional[str] = None
      # path to the private key
      key: List[str] = dataclasses.field(default_factory=list)
      # password
      password: Optional[str] = None
      ssh_option: List[str] = dataclasses.field(default_factory=list)
  
  
  class GuestSsh(Guest):
      """
      Guest provisioned for test execution, capable of accepting SSH connections
  
      The following keys are expected in the 'data' dictionary::
  
          role ....... guest role in the multihost scenario (inherited)
          guest ...... hostname or ip address (inherited)
          port ....... port to connect to
          user ....... user name to log in
          key ........ path to the private key (str or list)
          password ... password
  
      These are by default imported into instance attributes.
      """
  
      _data_class: Type[GuestData] = GuestSshData
  
      port: Optional[int]
      user: Optional[str]
      key: List[str]
      password: Optional[str]
      ssh_option: List[str]
  
      # Master ssh connection process and socket path
      _ssh_master_process: Optional['subprocess.Popen[bytes]'] = None
      _ssh_socket_path: Optional[str] = None
  
      def _ssh_guest(self) -> str:
          """ Return user@guest """
          return f'{self.user}@{self.guest}'
  
      def _ssh_socket(self) -> str:
          """ Prepare path to the master connection socket """
          if not self._ssh_socket_path:
              # Use '/run/user/uid' if it exists, '/tmp' otherwise
              run_dir = f"/run/user/{os.getuid()}"
              if os.path.isdir(run_dir):
                  socket_dir = os.path.join(run_dir, "tmt")
              else:
                  socket_dir = "/tmp"
              os.makedirs(socket_dir, exist_ok=True)
              self._ssh_socket_path = tempfile.mktemp(dir=socket_dir)
          return self._ssh_socket_path
  
      def _ssh_options(self, join: bool = False) -> Union[str, List[str]]:
          """ Return common ssh options (list or joined) """
          options = [
              '-oForwardX11=no',
              '-oStrictHostKeyChecking=no',
              '-oUserKnownHostsFile=/dev/null',
              # Prevent ssh from disconnecting if no data has been
              # received from the server for a long time (#868).
              '-oServerAliveInterval=60',
              '-oServerAliveCountMax=5',
              ]
          if self.key or self.password:
              # Skip ssh-agent (it adds additional identities)
              options.append('-oIdentitiesOnly=yes')
          if self.port:
              options.append(f'-p{self.port}')
          if self.key:
              for key in self.key:
                  options.append(f'-i{shlex.quote(key) if join else key}')
          if self.password:
              options.extend(['-oPasswordAuthentication=yes'])
  
          # Use the shared master connection
          options.append(f'-S{self._ssh_socket()}')
  
          options.extend([f'-o{option}' for option in self.ssh_option])
  
          return ' '.join(options) if join else options
  
      def _ssh_master_connection(self, command: List[str]) -> None:
          """ Check/create the master ssh connection """
          if self._ssh_master_process:
              return
  
          # TODO: _ssh_options() returns two incompatible types, we need to chose the right one.
          ssh_options = self._ssh_options()
          assert isinstance(ssh_options, list)
  
          command = command + ssh_options + ["-MNnT", self._ssh_guest()]
          self.debug(f"Create the master ssh connection: {' '.join(command)}")
          self._ssh_master_process = subprocess.Popen(
              command,
              stdin=subprocess.DEVNULL,
              stdout=subprocess.DEVNULL,
              stderr=subprocess.DEVNULL)
  
      def _ssh_command(self, join: bool = False) -> Union[str, List[str]]:
          """ Prepare an ssh command line for execution (list or joined) """
          command = []
          if self.password:
              password = shlex.quote(self.password) if join else self.password
              command.extend(["sshpass", "-p", password])
          command.append("ssh")
  
          # Check the master connection
          self._ssh_master_connection(command)
  
          if join:
              # TODO: _ssh_options() returns two incompatible types, we need to chose the right one.
              joined_ssh_options = self._ssh_options(join=True)
              assert isinstance(joined_ssh_options, str)
  
              return " ".join(command) + " " + joined_ssh_options
          else:
              # TODO: _ssh_options() returns two incompatible types, we need to chose the right one.
              ssh_options = self._ssh_options()
              assert isinstance(ssh_options, list)
  
              return command + ssh_options
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options related to SSH-capable guests """
          return super().options(how=how) + [
              click.option('--ssh-option', metavar="OPTION", multiple=True, default=[],
                           help="Specify additional SSH option. "
                                "Value is passed to SSH's -o option, see ssh_config(5) for "
                                "supported options. Can be specified multiple times.")]
  
      def ansible(self, playbook: str, extra_args: Optional[str] = None) -> None:
          """ Prepare guest using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
  
          # TODO: _ssh_options() returns two incompatible types, we need to chose the right one.
          joined_ssh_options = self._ssh_options(join=True)
          assert isinstance(joined_ssh_options, str)
  
          ansible_command = [
              'ansible-playbook'
              ] + self._ansible_verbosity()
  
          if extra_args:
              ansible_command += self._ansible_extra_args(extra_args)
  
          ansible_command += [
              f'--ssh-common-args={joined_ssh_options}',
              '-i', f'{self._ssh_guest()},',
              playbook
              ]
  
          # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
          parent = cast(Provision, self.parent)
  
          stdout, _ = self.run(
              ansible_command,
              cwd=parent.plan.worktree,
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      @property
      def is_ready(self) -> bool:
          """ Detect guest is ready or not """
  
          # Enough for now, ssh connection can be created later
          return self.guest is not None
  
      def execute(self,
                  command: Union[str, List[str]],
                  friendly_command: Optional[str] = None,
                  test_session: bool = False,
                  silent: bool = False,
                  log: Optional[BaseLoggerFnType] = None,
                  **kwargs: Any) -> tmt.utils.CommandOutput:
          """
          Execute command on the guest
  
          command ... string or list of command arguments (required)
          env ....... dictionary with environment variables
          cwd ....... working directory to be entered before execution
  
          If the command is provided as a list, it will be space-joined.
          If necessary, quote escaping has to be handled by the caller.
          """
          # Abort if guest is unavailable
          if self.guest is None:
              if not self.opt('dry'):
                  raise tmt.utils.GeneralError('The guest is not available.')
  
          # Prepare the export of environment variables
          environment = self._export_environment(
              self._prepare_environment(kwargs.get('env', dict())))
  
          # Change to given directory on guest if cwd provided
          directory = kwargs.get('cwd') or ''
          if directory:
              directory = f"cd {quote(directory)}; "
  
          # Run in interactive mode if requested
          interactive = ['-t'] if kwargs.get('interactive') else []
  
          # Force ssh to allocate pseudo-terminal if requested. Without a pseudo-terminal,
          # remote processes spawned by SSH would keep running after SSH process death, e.g.
          # in the case of a timeout.
          #
          # Note that polite request, `-t`, is not enough since `ssh` itself has no pseudo-terminal,
          # and a single `-t` wouldn't have the necessary effect.
          ptty = ['-tt'] if test_session else []
  
          # Prepare command and run it
          if isinstance(command, (list, tuple)):
              command = ' '.join(command)
          self.debug(f"Execute command '{command}' on guest '{self.guest}'.")
          friendly_command = friendly_command or command
  
          # TODO: _ssh_command() returns two incompatible types, we need to chose the right one.
          ssh_command = self._ssh_command()
          assert isinstance(ssh_command, list)
  
          command = (
              ssh_command + interactive + ptty + [self._ssh_guest()] +
              [f'{environment}{directory}{command}'])
          return self.run(command,
                          log=log if log else self._command_verbose_logger,
                          friendly_command=friendly_command,
                          silent=silent,
                          **kwargs)
  
      def push(self,
               source: Optional[str] = None,
               destination: Optional[str] = None,
               options: Optional[List[str]] = None) -> None:
          """
          Push files to the guest
  
          By default the whole plan workdir is synced to the same location
          on the guest. Use the 'source' and 'destination' to sync custom
          location and the 'options' parametr to modify default options
          which are '-Rrz --links --safe-links --delete'.
          """
          # Abort if guest is unavailable
          if self.guest is None:
              if not self.opt('dry'):
                  raise tmt.utils.GeneralError('The guest is not available.')
  
          # Prepare options and the push command
          options = options or DEFAULT_RSYNC_PUSH_OPTIONS
          if destination is None:
              destination = "/"
          if source is None:
              # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
              parent = cast(Provision, self.parent)
  
              assert parent.plan.workdir is not None
  
              source = parent.plan.workdir
              self.debug(f"Push workdir to guest '{self.guest}'.")
          else:
              self.debug(f"Copy '{source}' to '{destination}' on the guest.")
  
          def rsync() -> None:
              """ Run the rsync command """
              # In closure, mypy has hard times to reason about the state of used variables.
              assert options
              assert source
              assert destination
  
              # TODO: _ssh_command() returns two incompatible types, we need to chose the right one.
              joined_ssh_command = self._ssh_command(join=True)
              assert isinstance(joined_ssh_command, str)
  
              self.run(
                  ["rsync"] + options
                  + ["-e", joined_ssh_command]
                  + [source, f"{self._ssh_guest()}:{destination}"])
  
          # Try to push twice, check for rsync after the first failure
          try:
              rsync()
          except tmt.utils.RunError:
              try:
                  if self._check_rsync() == CheckRsyncOutcome.ALREADY_INSTALLED:
                      raise
                  rsync()
              except tmt.utils.RunError:
                  # Provide a reasonable error to the user
                  self.fail(
                      f"Failed to push workdir to the guest. This usually means "
                      f"that login as '{self.user}' to the guest does not work.")
                  raise
  
      def pull(self,
               source: Optional[str] = None,
               destination: Optional[str] = None,
               options: Optional[List[str]] = None,
               extend_options: Optional[List[str]] = None) -> None:
          """
          Pull files from the guest
  
          By default the whole plan workdir is synced from the same
          location on the guest. Use the 'source' and 'destination' to
          sync custom location, the 'options' parameter to modify
          default options :py:data:`DEFAULT_RSYNC_PULL_OPTIONS`
          and 'extend_options' to extend them (e.g. by exclude).
          """
          # Abort if guest is unavailable
          if self.guest is None:
              if not self.opt('dry'):
                  raise tmt.utils.GeneralError('The guest is not available.')
  
          # Prepare options and the pull command
          options = options or DEFAULT_RSYNC_PULL_OPTIONS
          if extend_options is not None:
              options.extend(extend_options)
          if destination is None:
              destination = "/"
          if source is None:
              # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
              parent = cast(Provision, self.parent)
  
              assert parent.plan.workdir is not None
  
              source = parent.plan.workdir
              self.debug(f"Pull workdir from guest '{self.guest}'.")
          else:
              self.debug(f"Copy '{source}' from the guest to '{destination}'.")
  
          def rsync() -> None:
              """ Run the rsync command """
              # In closure, mypy has hard times to reason about the state of used variables.
              assert options
              assert source
              assert destination
  
              # TODO: _ssh_command() returns two incompatible types, we need to chose the right one.
              joined_ssh_command = self._ssh_command(join=True)
              assert isinstance(joined_ssh_command, str)
  
              self.run(
                  ["rsync"] + options
                  + ["-e", joined_ssh_command]
                  + [f"{self._ssh_guest()}:{source}", destination])
  
          # Try to pull twice, check for rsync after the first failure
          try:
              rsync()
          except tmt.utils.RunError:
              try:
                  if self._check_rsync() == CheckRsyncOutcome.ALREADY_INSTALLED:
                      raise
                  rsync()
              except tmt.utils.RunError:
                  # Provide a reasonable error to the user
                  self.fail(
                      f"Failed to pull workdir from the guest. "
                      f"This usually means that login as '{self.user}' "
                      f"to the guest does not work.")
                  raise
  
      def stop(self) -> None:
          """
          Stop the guest
  
          Shut down a running guest instance so that it does not consume
          any memory or cpu resources. If needed, perform any actions
          necessary to store the instance status to disk.
          """
  
          # Close the master ssh connection
          if self._ssh_master_process:
              self.debug("Close the master ssh connection.", level=3)
              try:
                  self._ssh_master_process.terminate()
                  self._ssh_master_process.wait(timeout=3)
              except subprocess.TimeoutExpired:
                  pass
  
          # Remove the ssh socket
          if self._ssh_socket_path and os.path.exists(self._ssh_socket_path):
              self.debug(
                  f"Remove ssh socket '{self._ssh_socket_path}'.", level=3)
              try:
                  os.unlink(self._ssh_socket_path)
              except OSError as error:
                  self.debug(f"Failed to remove the socket: {error}", level=3)
  
      def reboot(
              self,
              hard: bool = False,
              command: Optional[str] = None,
              timeout: Optional[int] = None,
              tick: float = tmt.utils.DEFAULT_WAIT_TICK,
              tick_increase: float = tmt.utils.DEFAULT_WAIT_TICK_INCREASE) -> bool:
          """
          Reboot the guest, return True if reconnect was successful
  
          Parameter 'hard' set to True means that guest should be
          rebooted by way which is not clean in sense that data can be
          lost. When set to False reboot should be done gracefully.
  
          Use the 'command' parameter to specify a custom reboot command
          instead of the default 'reboot'.
          """
  
          if hard:
              raise tmt.utils.ProvisionError(
                  "Method does not support hard reboot.")
  
          command = command or "reboot"
          self.debug(f"Reboot using the command '{command}'.")
  
          re_boot_time = re.compile(r'btime\s+(\d+)')
  
          def get_boot_time() -> int:
              """ Reads btime from /proc/stat """
              stdout = self.execute(["cat", "/proc/stat"]).stdout
              assert stdout
  
              match = re_boot_time.search(stdout)
  
              if match is None:
                  raise tmt.utils.ProvisionError('Failed to retrieve boot time from guest')
  
              return int(match.group(1))
  
          current_boot_time = get_boot_time()
  
          try:
              self.execute(command)
          except tmt.utils.RunError as error:
              # Connection can be closed by the remote host even before the
              # reboot command is completed. Let's ignore such errors.
              if error.returncode == 255:
                  self.debug(
                      "Seems the connection was closed too fast, ignoring.")
              else:
                  raise
  
          # Wait until we get new boot time, connection will drop and will be
          # unreachable for some time
          def check_boot_time() -> None:
              try:
                  new_boot_time = get_boot_time()
  
                  if new_boot_time != current_boot_time:
                      # Different boot time and we are reconnected
                      return
  
                  # Same boot time, reboot didn't happen yet, retrying
                  raise tmt.utils.WaitingIncomplete()
  
              except tmt.utils.RunError:
                  self.debug('Failed to connect to the guest.')
                  raise tmt.utils.WaitingIncomplete()
  
          timeout = timeout or CONNECTION_TIMEOUT
  
          try:
              tmt.utils.wait(
                  self,
                  check_boot_time,
                  datetime.timedelta(seconds=timeout),
                  tick=tick,
                  tick_increase=tick_increase)
  
          except tmt.utils.WaitingTimedOutError:
              self.debug("Connection to guest failed after reboot.")
              return False
  
          self.debug("Connection to guest succeeded after reboot.")
          return True
  
      def remove(self) -> None:
          """
          Remove the guest
  
          Completely remove all guest instance data so that it does not
          consume any disk resources.
          """
          self.debug(f"Doing nothing to remove guest '{self.guest}'.")
  
      def _check_rsync(self) -> CheckRsyncOutcome:
          """
          Make sure that rsync is installed on the guest
  
          On read-only distros install it under the '/root/pkg' directory.
          Returns 'already installed' when rsync is already present.
          """
  
          # Check for rsync (nothing to do if already installed)
          self.debug("Ensure that rsync is installed on the guest.")
          try:
              self.execute("rsync --version")
              return CheckRsyncOutcome.ALREADY_INSTALLED
          except tmt.utils.RunError:
              pass
  
          # Check the package manager
          self.debug("Check the package manager.")
          try:
              self.execute("dnf --version")
              package_manager = "dnf"
          except tmt.utils.RunError:
              package_manager = "yum"
  
          # Install under '/root/pkg' for read-only distros
          # (for now the check is based on 'rpm-ostree' presence)
          # FIXME: Find a better way how to detect read-only distros
          self.debug("Check for a read-only distro.")
          try:
              self.execute("rpm-ostree --version")
              readonly = (
                  " --installroot=/root/pkg --releasever / "
                  "&& ln -sf /root/pkg/bin/rsync /usr/local/bin/rsync")
          except tmt.utils.RunError:
              readonly = ""
  
          # Install the rsync
          self.execute(f"{package_manager} install -y rsync" + readonly)
  
          return CheckRsyncOutcome.INSTALLED
  
      @classmethod
      def requires(cls) -> List[str]:
          """ No extra requires needed """
          return []
  
  
  @dataclasses.dataclass
  class ProvisionStepData(tmt.steps.StepData):
      pass
  
  
  class ProvisionPlugin(tmt.steps.GuestlessPlugin):
      """ Common parent of provision plugins """
  
      _data_class = ProvisionStepData
      _guest_class = Guest
  
      # 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] = []
  
      @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: 'tmt.cli.Context', **kwargs: Any) -> None:
              context.obj.steps.add('provision')
              Provision._save_context(context)
  
          return provision
  
      # TODO: this might be needed until https://github.com/teemtee/tmt/issues/1696 is resolved
      def opt(self, option: str, default: Optional[Any] = None) -> Any:
          """ Get an option from the command line options """
  
          if option == 'ssh-option':
              value = super().opt(option, default=default)
  
              if isinstance(value, tuple):
                  return list(value)
  
              return value
  
          return super().opt(option, default=default)
  
      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 options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Return list of options. """
          return super().options(how) + cls._guest_class.options(how)
  
      @classmethod
      def clean_images(cls, clean: 'tmt.base.Clean', dry: bool) -> bool:
          """ Remove the images of one particular plugin """
          return True
  
  
  class Provision(tmt.steps.Step):
      """ Provision an environment for testing or use localhost. """
  
      # Default implementation for provision is a virtual machine
      DEFAULT_HOW = 'virtual'
  
      _plugin_base_class = ProvisionPlugin
  
      _preserved_files = ['step.yaml', 'guests.yaml']
  
      def __init__(self, *, plan: 'tmt.Plan', data: tmt.steps.RawStepDataArgument) -> None:
          """ Initialize provision step data """
          super().__init__(plan=plan, data=data)
  
          # List of provisioned guests and loaded guest data
          self._guests: List[Guest] = []
          self._guest_data: Dict[str, GuestData] = {}
          self.is_multihost = False
  
      def _normalize_data(self, raw_data: List[tmt.steps._RawStepData]) -> List[tmt.steps.StepData]:
          data = super()._normalize_data(raw_data)
  
          # Check that the names are unique
          names = [datum.name for datum in data]
          names_count = collections.Counter(names)
          if len(data) > 1 and names_count.most_common(1)[0][1] > 1:
              duplicate_names = [name for name in names_count.elements() if names_count[name] > 1]
              duplicate_string = ', '.join(duplicate_names)
              raise tmt.utils.GeneralError(
                  f"Provision step names must be unique for multihost testing. "
                  f"Duplicate names: '{duplicate_string}' in plan '{self.plan.name}'.")
  
          return data
  
      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:
              # FIXME: cast() - see https://github.com/teemtee/tmt/issues/1599
              plugin = cast(ProvisionPlugin, ProvisionPlugin.delegate(self, data=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 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('')
                  except (tmt.utils.RunError, tmt.utils.ProvisionError) as error:
                      self.fail(str(error))
                      raise
                  finally:
                      if isinstance(phase, ProvisionPlugin):
                          guest = phase.guest()
                          if guest and (guest.is_ready or self.opt('dry')):
                              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)
  
files: tmt/steps/provision/__init__.py
files: import dataclasses
  import datetime
  import sys
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import fmf.utils
  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 missing cpu.processors constraint
      '0.0.47',
      # NEW: added new CPU constraints
      '0.0.46',
      # 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
  
  
  @dataclasses.dataclass
  class ProvisionArtemisData(ArtemisGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  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)
  
  
  class GuestArtemis(tmt.GuestSsh):
      """
      Artemis guest instance
  
      The following keys are expected in the 'data' dictionary:
      """
  
      _data_class = ArtemisGuestData
  
      # API
      api_url: str
      api_version: str
  
      # Guest request properties
      arch: str
      image: str
      hardware: Optional[Any]
      pool: Optional[str]
      priority_group: str
      keyname: str
      user_data: Dict[str, str]
  
      # Provided by Artemis response
      guestname: Optional[str]
  
      # Timeouts and deadlines
      provision_timeout: int
      provision_tick: int
      api_timeout: int
      api_retries: int
      api_retry_backoff_factor: int
  
      _api: Optional[ArtemisAPI] = None
  
      @property
      def api(self) -> ArtemisAPI:
          if self._api is None:
              self._api = ArtemisAPI(self)
  
          return self._api
  
      @property
      def is_ready(self) -> bool:
          """ Detect the guest is ready or not """
  
          # FIXME: A more robust solution should be provided. Currently just
          #        return True if self.guest is not None
          return self.guest is not None
  
      def _create(self) -> None:
          environment: Dict[str, Any] = {
              'hw': {
                  'arch': self.arch
                  },
              'os': {
                  'compose': self.image
                  }
              }
  
          data: Dict[str, Any] = {
              'environment': environment,
              'keyname': self.keyname,
              'priority_group': self.priority_group,
              'user_data': self.user_data
              }
  
          if self.pool:
              environment['pool'] = self.pool
  
          if self.hardware is not None:
              assert isinstance(self.hardware, dict)
  
              environment['hw']['constraints'] = self.hardware
  
          # TODO: snapshots
          # TODO: spot instance
          # TODO: post-install script
          # TODO: log types
  
          response = self.api.create('/guests/', data)
  
          if response.status_code == 201:
              self.info('guest', 'has been requested', 'green')
  
          else:
              raise ProvisionError(
                  f"Failed to create, "
                  f"unhandled API response '{response.status_code}'.")
  
          self.guestname = response.json()['guestname']
          self.info('guestname', self.guestname, 'green')
  
          with updatable_message(
                  'state', indent_level=self._level()) as progress_message:
  
              def get_new_state() -> GuestInspectType:
                  response = self.api.inspect(f'/guests/{self.guestname}')
  
                  if response.status_code != 200:
                      raise ProvisionError(
                          f"Failed to create, "
                          f"unhandled API response '{response.status_code}'.")
  
                  current = cast(GuestInspectType, response.json())
                  state = current['state']
                  state_color = GUEST_STATE_COLORS.get(
                      state, GUEST_STATE_COLOR_DEFAULT)
  
                  progress_message.update(state, color=state_color)
  
                  if state == 'error':
                      raise ProvisionError(
                          'Failed to create, provisioning failed.')
  
                  if state == 'ready':
                      return current
  
                  raise tmt.utils.WaitingIncomplete()
  
              try:
                  guest_info = tmt.utils.wait(
                      self, get_new_state, datetime.timedelta(
                          seconds=self.provision_timeout), tick=self.provision_tick)
  
              except tmt.utils.WaitingTimedOutError:
                  raise ProvisionError(
                      f'Failed to provision in the given amount '
                      f'of time (--provision-timeout={self.provision_timeout}).')
  
          self.guest = guest_info['address']
          self.info('address', self.guest, 'green')
  
      def start(self) -> None:
          """
          Start the guest
  
          Get a new guest instance running. This should include preparing
          any configuration necessary to get it started. Called after
          load() is completed so all guest data should be available.
          """
  
          if self.guestname is None or self.guest is None:
              self._create()
  
      def remove(self) -> None:
          """ Remove the guest """
  
          if self.guestname is None:
              return
  
          response = self.api.delete(f'/guests/{self.guestname}')
  
          if response.status_code == 404:
              self.info('guest', 'no longer exists', 'red')
  
          elif response.status_code == 409:
              self.info('guest', 'has existing snapshots', 'red')
  
          elif response.ok:
              self.info('guest', 'has been removed', 'green')
  
          else:
              self.info(
                  'guest',
                  f"Failed to remove, "
                  f"unhandled API response '{response.status_code}'.")
  
  
  @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
      """
  
      _data_class = ProvisionArtemisData
      _guest_class = GuestArtemis
  
      # Guest instance
      _guest = None
  
      # TODO: fix types once superclass gains its annotations
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for Artemis """
          return [
              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',
                  help='Pool to enforce.'
                  ),
              click.option(
                  '--priority-group', metavar='NAME',
                  help='Provisioning priority group.'
                  ),
              click.option(
                  '--keyname', metavar='NAME',
                  help='SSH key name.'
                  ),
              click.option(
                  '--user-data', metavar='KEY=VALUE',
                  help='Optional data to attach to guest.',
                  multiple=True,
                  default=[]
                  ),
              click.option(
                  '--provision-timeout', metavar='SECONDS',
                  help=f'How long to wait for provisioning to complete, '
                       f'{DEFAULT_PROVISION_TIMEOUT} seconds by default.'
                  ),
              click.option(
                  '--provision-tick', metavar='SECONDS',
                  help=f'How often check Artemis API for provisioning status, '
                       f'{DEFAULT_PROVISION_TICK} seconds by default.',
                  ),
              click.option(
                  '--api-timeout', metavar='SECONDS',
                  help=f'How long to wait for API operations to complete, '
                       f'{DEFAULT_API_TIMEOUT} seconds by default.',
                  ),
              click.option(
                  '--api-retries', metavar='COUNT',
                  help=f'How many attempts to use when talking to API, '
                       f'{DEFAULT_API_RETRIES} by default.',
                  ),
              click.option(
                  '--api-retry-backoff-factor', metavar='COUNT',
                  help=f'A factor for exponential API retry backoff, '
                       f'{DEFAULT_RETRY_BACKOFF_FACTOR} by default.',
                  ),
              ] + super().options(how)
  
      def go(self) -> None:
          """ Provision the guest """
          super().go()
  
          api_version = self.get('api-version')
  
          if api_version not in SUPPORTED_API_VERSIONS:
              raise ProvisionError(f"API version '{api_version}' not supported.")
  
          try:
              user_data = {
                  key.strip(): value.strip()
                  for key, value in (
                      pair.split('=', 1)
                      for pair in self.get('user-data')
                      )
                  }
  
          except ValueError:
              raise ProvisionError('Cannot parse user-data.')
  
          data = ArtemisGuestData(
              api_url=self.get('api-url'),
              api_version=api_version,
              arch=self.get('arch'),
              image=self.get('image'),
              hardware=self.get('hardware'),
              pool=self.get('pool'),
              priority_group=self.get('priority-group'),
              keyname=self.get('keyname'),
              user_data=user_data,
              user=self.get('user'),
              provision_timeout=self.get('provision-timeout'),
              provision_tick=self.get('provision-tick'),
              api_timeout=self.get('api-timeout'),
              api_retries=self.get('api-retries'),
              api_retry_backoff_factor=self.get('api-retry-backoff-factor')
              )
  
          # FIXME: cast() - typeless "dispatcher" method
          data.ssh_option = cast(List[str], self.get('ssh-option'))
          if data.ssh_option:
              self.info('ssh options', fmf.utils.listed(data.ssh_option), 'green')
  
          self._guest = GuestArtemis(data=data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self) -> Optional[GuestArtemis]:
          """ Return the provisioned guest """
          return self._guest
  
files: tmt/steps/provision/artemis.py
files: import dataclasses
  import datetime
  import sys
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import fmf.utils
  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 missing cpu.processors constraint
      '0.0.47',
      # NEW: added new CPU constraints
      '0.0.46',
      # 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
  
  
  @dataclasses.dataclass
  class ProvisionArtemisData(ArtemisGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  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)
  
  
  class GuestArtemis(tmt.GuestSsh):
      """
      Artemis guest instance
  
      The following keys are expected in the 'data' dictionary:
      """
  
      _data_class = ArtemisGuestData
  
      # API
      api_url: str
      api_version: str
  
      # Guest request properties
      arch: str
      image: str
      hardware: Optional[Any]
      pool: Optional[str]
      priority_group: str
      keyname: str
      user_data: Dict[str, str]
  
      # Provided by Artemis response
      guestname: Optional[str]
  
      # Timeouts and deadlines
      provision_timeout: int
      provision_tick: int
      api_timeout: int
      api_retries: int
      api_retry_backoff_factor: int
  
      _api: Optional[ArtemisAPI] = None
  
      @property
      def api(self) -> ArtemisAPI:
          if self._api is None:
              self._api = ArtemisAPI(self)
  
          return self._api
  
      @property
      def is_ready(self) -> bool:
          """ Detect the guest is ready or not """
  
          # FIXME: A more robust solution should be provided. Currently just
          #        return True if self.guest is not None
          return self.guest is not None
  
      def _create(self) -> None:
          environment: Dict[str, Any] = {
              'hw': {
                  'arch': self.arch
                  },
              'os': {
                  'compose': self.image
                  }
              }
  
          data: Dict[str, Any] = {
              'environment': environment,
              'keyname': self.keyname,
              'priority_group': self.priority_group,
              'user_data': self.user_data
              }
  
          if self.pool:
              environment['pool'] = self.pool
  
          if self.hardware is not None:
              assert isinstance(self.hardware, dict)
  
              environment['hw']['constraints'] = self.hardware
  
          # TODO: snapshots
          # TODO: spot instance
          # TODO: post-install script
          # TODO: log types
  
          response = self.api.create('/guests/', data)
  
          if response.status_code == 201:
              self.info('guest', 'has been requested', 'green')
  
          else:
              raise ProvisionError(
                  f"Failed to create, "
                  f"unhandled API response '{response.status_code}'.")
  
          self.guestname = response.json()['guestname']
          self.info('guestname', self.guestname, 'green')
  
          with updatable_message(
                  'state', indent_level=self._level()) as progress_message:
  
              def get_new_state() -> GuestInspectType:
                  response = self.api.inspect(f'/guests/{self.guestname}')
  
                  if response.status_code != 200:
                      raise ProvisionError(
                          f"Failed to create, "
                          f"unhandled API response '{response.status_code}'.")
  
                  current = cast(GuestInspectType, response.json())
                  state = current['state']
                  state_color = GUEST_STATE_COLORS.get(
                      state, GUEST_STATE_COLOR_DEFAULT)
  
                  progress_message.update(state, color=state_color)
  
                  if state == 'error':
                      raise ProvisionError(
                          'Failed to create, provisioning failed.')
  
                  if state == 'ready':
                      return current
  
                  raise tmt.utils.WaitingIncomplete()
  
              try:
                  guest_info = tmt.utils.wait(
                      self, get_new_state, datetime.timedelta(
                          seconds=self.provision_timeout), tick=self.provision_tick)
  
              except tmt.utils.WaitingTimedOutError:
                  raise ProvisionError(
                      f'Failed to provision in the given amount '
                      f'of time (--provision-timeout={self.provision_timeout}).')
  
          self.guest = guest_info['address']
          self.info('address', self.guest, 'green')
  
      def start(self) -> None:
          """
          Start the guest
  
          Get a new guest instance running. This should include preparing
          any configuration necessary to get it started. Called after
          load() is completed so all guest data should be available.
          """
  
          if self.guestname is None or self.guest is None:
              self._create()
  
      def remove(self) -> None:
          """ Remove the guest """
  
          if self.guestname is None:
              return
  
          response = self.api.delete(f'/guests/{self.guestname}')
  
          if response.status_code == 404:
              self.info('guest', 'no longer exists', 'red')
  
          elif response.status_code == 409:
              self.info('guest', 'has existing snapshots', 'red')
  
          elif response.ok:
              self.info('guest', 'has been removed', 'green')
  
          else:
              self.info(
                  'guest',
                  f"Failed to remove, "
                  f"unhandled API response '{response.status_code}'.")
  
  
  @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
      """
  
      _data_class = ProvisionArtemisData
      _guest_class = GuestArtemis
  
      # Guest instance
      _guest = None
  
      # TODO: fix types once superclass gains its annotations
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for Artemis """
          return [
              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',
                  help='Pool to enforce.'
                  ),
              click.option(
                  '--priority-group', metavar='NAME',
                  help='Provisioning priority group.'
                  ),
              click.option(
                  '--keyname', metavar='NAME',
                  help='SSH key name.'
                  ),
              click.option(
                  '--user-data', metavar='KEY=VALUE',
                  help='Optional data to attach to guest.',
                  multiple=True,
                  default=[]
                  ),
              click.option(
                  '--provision-timeout', metavar='SECONDS',
                  help=f'How long to wait for provisioning to complete, '
                       f'{DEFAULT_PROVISION_TIMEOUT} seconds by default.'
                  ),
              click.option(
                  '--provision-tick', metavar='SECONDS',
                  help=f'How often check Artemis API for provisioning status, '
                       f'{DEFAULT_PROVISION_TICK} seconds by default.',
                  ),
              click.option(
                  '--api-timeout', metavar='SECONDS',
                  help=f'How long to wait for API operations to complete, '
                       f'{DEFAULT_API_TIMEOUT} seconds by default.',
                  ),
              click.option(
                  '--api-retries', metavar='COUNT',
                  help=f'How many attempts to use when talking to API, '
                       f'{DEFAULT_API_RETRIES} by default.',
                  ),
              click.option(
                  '--api-retry-backoff-factor', metavar='COUNT',
                  help=f'A factor for exponential API retry backoff, '
                       f'{DEFAULT_RETRY_BACKOFF_FACTOR} by default.',
                  ),
              ] + super().options(how)
  
      # FIXME: ignore - https://github.com/teemtee/tmt/issues/1437
      def wake(self, data: Optional[ArtemisGuestData] = None) -> None:  # type: ignore[override]
          """ Wake up the plugin, process data, apply options """
  
          super().wake(data=data)
  
          if data:
              self._guest = GuestArtemis(data=data, name=self.name, parent=self.step)
  
      def go(self) -> None:
          """ Provision the guest """
          super().go()
  
          api_version = self.get('api-version')
  
          if api_version not in SUPPORTED_API_VERSIONS:
              raise ProvisionError(f"API version '{api_version}' not supported.")
  
          try:
              user_data = {
                  key.strip(): value.strip()
                  for key, value in (
                      pair.split('=', 1)
                      for pair in self.get('user-data')
                      )
                  }
  
          except ValueError:
              raise ProvisionError('Cannot parse user-data.')
  
          data = ArtemisGuestData(
              api_url=self.get('api-url'),
              api_version=api_version,
              arch=self.get('arch'),
              image=self.get('image'),
              hardware=self.get('hardware'),
              pool=self.get('pool'),
              priority_group=self.get('priority-group'),
              keyname=self.get('keyname'),
              user_data=user_data,
              user=self.get('user'),
              provision_timeout=self.get('provision-timeout'),
              provision_tick=self.get('provision-tick'),
              api_timeout=self.get('api-timeout'),
              api_retries=self.get('api-retries'),
              api_retry_backoff_factor=self.get('api-retry-backoff-factor')
              )
  
          # FIXME: cast() - typeless "dispatcher" method
          data.ssh_option = cast(List[str], self.get('ssh-option'))
          if data.ssh_option:
              self.info('ssh options', fmf.utils.listed(data.ssh_option), 'green')
  
          self._guest = GuestArtemis(data=data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self) -> Optional[GuestArtemis]:
          """ Return the provisioned guest """
          return self._guest
  
files: tmt/steps/provision/artemis.py
files: import dataclasses
  from typing import List, Optional, cast
  
  import click
  import fmf.utils
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  
  DEFAULT_USER = "root"
  
  
  @dataclasses.dataclass
  class ConnectGuestData(tmt.steps.provision.GuestSshData):
      user: str = DEFAULT_USER
  
  
  @dataclasses.dataclass
  class ProvisionConnectData(ConnectGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  @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
      """
  
      _data_class = ProvisionConnectData
      _guest_class = tmt.steps.provision.GuestSsh
  
      # Guest instance
      _guest = None
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for connect """
          return [
              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.'),
              ] + super().options(how)
  
      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 = ConnectGuestData(
              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.')
              # TODO: something to fix in the future, multiple --key options would help.
              # Right now, the default value is List[str], while the option is just str.
              if isinstance(key, list):
                  data.key = key
              else:
                  data.key = [key]
  
          # FIXME: cast() - typeless "dispatcher" method
          data.ssh_option = cast(List[str], self.get('ssh-option'))
          if data.ssh_option:
              self.info('ssh options', fmf.utils.listed(data.ssh_option), 'green')
  
          # And finally create the guest
          self._guest = tmt.GuestSsh(data=data, name=self.name, parent=self.step)
  
      def guest(self) -> Optional[tmt.GuestSsh]:
          """ Return the provisioned guest """
          return self._guest
  
files: tmt/steps/provision/connect.py
files: import dataclasses
  from typing import List, Optional, cast
  
  import click
  import fmf.utils
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.steps.provision import GuestSshData
  
  DEFAULT_USER = "root"
  
  
  @dataclasses.dataclass
  class ConnectGuestData(tmt.steps.provision.GuestSshData):
      user: str = DEFAULT_USER
  
  
  @dataclasses.dataclass
  class ProvisionConnectData(ConnectGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  @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
      """
  
      _data_class = ProvisionConnectData
      _guest_class = tmt.steps.provision.GuestSsh
  
      # Guest instance
      _guest = None
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for connect """
          return [
              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.'),
              ] + super().options(how)
  
      # FIXME: ignore - https://github.com/teemtee/tmt/issues/1437
      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=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 = ConnectGuestData(
              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.')
              # TODO: something to fix in the future, multiple --key options would help.
              # Right now, the default value is List[str], while the option is just str.
              if isinstance(key, list):
                  data.key = key
              else:
                  data.key = [key]
  
          # FIXME: cast() - typeless "dispatcher" method
          data.ssh_option = cast(List[str], self.get('ssh-option'))
          if data.ssh_option:
              self.info('ssh options', fmf.utils.listed(data.ssh_option), 'green')
  
          # And finally create the guest
          self._guest = tmt.GuestSsh(data=data, name=self.name, parent=self.step)
  
      def guest(self) -> Optional[tmt.GuestSsh]:
          """ Return the provisioned guest """
          return self._guest
  
files: tmt/steps/provision/connect.py
files: import dataclasses
  import shlex
  from typing import Any, List, Optional, Union
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import BaseLoggerFnType
  
  
  @dataclasses.dataclass
  class ProvisionLocalData(tmt.steps.provision.GuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  class GuestLocal(tmt.Guest):
      """ Local Host """
  
      localhost = True
      parent: tmt.steps.Step
  
      @property
      def is_ready(self) -> bool:
          """ Local is always ready """
          return True
  
      def ansible(self, playbook: str, extra_args: Optional[str] = None) -> None:
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, _ = 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: Union[List[str], str],
                  friendly_command: Optional[str] = None,
                  test_session: bool = False,
                  silent: bool = False,
                  log: Optional[BaseLoggerFnType] = None,
                  **kwargs: Any) -> tmt.utils.CommandOutput:
          """ Execute command on localhost """
          # Prepare the environment (plan/cli variables override)
          environment: tmt.utils.EnvironmentType = dict()
          # Beware of using dict() as default: `env` is `Optional[EnvironmentType]`, and may
          # be set by a caller by perfectly acceptable `None`. In such a case, `pop()` with
          # a default would return not the default, but existing key, i.e. `None`.
          environment.update(kwargs.pop('env', None) or {})
          environment.update(self.parent.plan.environment)
          if friendly_command is None:
              if isinstance(command, (list, tuple)):
                  friendly_command = ' '.join(shlex.quote(s) for s in command)
              else:
                  friendly_command = command
          # Run the command under the prepared environment
          return self.run(command,
                          env=environment,
                          shell=True,
                          log=log if log else self._command_verbose_logger,
                          friendly_command=friendly_command,
                          silent=silent,
                          **kwargs)
  
      def stop(self) -> None:
          """ Stop the guest """
  
          self.debug(f"Doing nothing to stop guest '{self.guest}'.")
  
      def reboot(self,
                 hard: bool = False,
                 command: Optional[str] = None,
                 timeout: Optional[int] = None) -> bool:
          """ Reboot the guest, return True if successful """
  
          self.debug(f"Doing nothing to reboot guest '{self.guest}'.")
  
          return False
  
      def push(
              self,
              source: Optional[str] = None,
              destination: Optional[str] = None,
              options: Optional[List[str]] = None) -> None:
          """ Nothing to be done to push workdir """
  
      def pull(
              self,
              source: Optional[str] = None,
              destination: Optional[str] = None,
              options: Optional[List[str]] = None,
              extend_options: Optional[List[str]] = None) -> None:
          """ Nothing to be done to pull workdir """
  
  
  @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.
      """
  
      _data_class = ProvisionLocalData
      _guest_class = GuestLocal
  
      # Guest instance
      _guest = None
  
      def go(self) -> None:
          """ Provision the container """
          super().go()
  
          # Create a GuestLocal instance
          data = tmt.steps.provision.GuestData(
              guest='localhost',
              role=self.get('role')
              )
          self._guest = GuestLocal(data=data, name=self.name, parent=self.step)
  
      def guest(self) -> Optional[GuestLocal]:
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self) -> List[str]:
          """ List of required packages needed for workdir sync """
          return GuestLocal.requires()
  
files: tmt/steps/provision/local.py
files: import dataclasses
  import shlex
  from typing import Any, List, Optional, Union
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import BaseLoggerFnType
  
  
  @dataclasses.dataclass
  class ProvisionLocalData(tmt.steps.provision.GuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  class GuestLocal(tmt.Guest):
      """ Local Host """
  
      localhost = True
      parent: tmt.steps.Step
  
      @property
      def is_ready(self) -> bool:
          """ Local is always ready """
          return True
  
      def ansible(self, playbook: str, extra_args: Optional[str] = None) -> None:
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, _ = 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: Union[List[str], str],
                  friendly_command: Optional[str] = None,
                  test_session: bool = False,
                  silent: bool = False,
                  log: Optional[BaseLoggerFnType] = None,
                  **kwargs: Any) -> tmt.utils.CommandOutput:
          """ Execute command on localhost """
          # Prepare the environment (plan/cli variables override)
          environment: tmt.utils.EnvironmentType = dict()
          # Beware of using dict() as default: `env` is `Optional[EnvironmentType]`, and may
          # be set by a caller by perfectly acceptable `None`. In such a case, `pop()` with
          # a default would return not the default, but existing key, i.e. `None`.
          environment.update(kwargs.pop('env', None) or {})
          environment.update(self.parent.plan.environment)
          if friendly_command is None:
              if isinstance(command, (list, tuple)):
                  friendly_command = ' '.join(shlex.quote(s) for s in command)
              else:
                  friendly_command = command
          # Run the command under the prepared environment
          return self.run(command,
                          env=environment,
                          shell=True,
                          log=log if log else self._command_verbose_logger,
                          friendly_command=friendly_command,
                          silent=silent,
                          **kwargs)
  
      def stop(self) -> None:
          """ Stop the guest """
  
          self.debug(f"Doing nothing to stop guest '{self.guest}'.")
  
      def reboot(self,
                 hard: bool = False,
                 command: Optional[str] = None,
                 timeout: Optional[int] = None) -> bool:
          """ Reboot the guest, return True if successful """
  
          self.debug(f"Doing nothing to reboot guest '{self.guest}'.")
  
          return False
  
      def push(
              self,
              source: Optional[str] = None,
              destination: Optional[str] = None,
              options: Optional[List[str]] = None) -> None:
          """ Nothing to be done to push workdir """
  
      def pull(
              self,
              source: Optional[str] = None,
              destination: Optional[str] = None,
              options: Optional[List[str]] = None,
              extend_options: Optional[List[str]] = None) -> None:
          """ Nothing to be done to pull workdir """
  
  
  @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.
      """
  
      _data_class = ProvisionLocalData
      _guest_class = GuestLocal
  
      # Guest instance
      _guest = None
  
      def wake(self, data: Optional[tmt.steps.provision.GuestData] = None) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
          if data:
              self._guest = GuestLocal(data=data, name=self.name, parent=self.step)
  
      def go(self) -> None:
          """ Provision the container """
          super().go()
  
          # Create a GuestLocal instance
          data = tmt.steps.provision.GuestData(
              guest='localhost',
              role=self.get('role')
              )
          self._guest = GuestLocal(data=data, name=self.name, parent=self.step)
  
      def guest(self) -> Optional[GuestLocal]:
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self) -> List[str]:
          """ List of required packages needed for workdir sync """
          return GuestLocal.requires()
  
files: tmt/steps/provision/local.py
files: import dataclasses
  import os
  from shlex import quote
  from typing import Any, List, Optional, Tuple, Union
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  from tmt.utils import BaseLoggerFnType
  
  # Timeout in seconds of waiting for a connection
  CONNECTION_TIMEOUT = 60
  
  # Defaults
  DEFAULT_IMAGE = "fedora"
  DEFAULT_USER = "root"
  
  
  @dataclasses.dataclass
  class PodmanGuestData(tmt.steps.provision.GuestData):
      image: str = DEFAULT_IMAGE
      user: str = DEFAULT_USER
      force_pull: bool = False
  
      container: Optional[str] = None
  
  
  @dataclasses.dataclass
  class ProvisionPodmanData(PodmanGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  class GuestContainer(tmt.Guest):
      """ Container Instance """
  
      _data_class = PodmanGuestData
  
      image: Optional[str]
      container: Optional[str]
      user: str
      force_pull: bool
      parent: tmt.steps.Step
  
      @property
      def is_ready(self) -> bool:
          """ Detect the guest is ready or not """
          # Check the container is running or not
          if self.container is None:
              return False
          cmd_output = self.podman(['container', 'inspect', '--format',
                                    '{{json .State.Running}}', self.container])
          cmd_stdout, cmd_stderr = cmd_output
          return str(cmd_stdout).strip() == 'true'
  
      def wake(self) -> None:
          """ Wake up the guest """
          self.debug(
              f"Waking up container '{self.container}'.", level=2, shift=0)
  
      def start(self) -> None:
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Check if the image is available
          assert self.image is not None
          command = ['image', 'exists', self.image]
          try:
              self.podman(command, message=f"Check for container image '{self.image}'.")
              needs_pull = False
          except tmt.utils.RunError:
              needs_pull = True
  
          # Pull image if not available or pull forced
          if needs_pull 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}'.")
          assert self.container is not None
          self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f'{workdir}:{workdir}:z',
               '-itd', '--user', self.user, self.image])
  
      def reboot(self, hard: bool = False,
                 command: Optional[Union[str, List[str], Tuple[str, ...]]] = None,
                 timeout: Optional[int] = None) -> bool:
          """ 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).")
          assert self.container is not None
          self.podman(['container', 'restart', self.container])
          return self.reconnect(timeout=timeout or CONNECTION_TIMEOUT)
  
      def ansible(self, playbook: str, extra_args: Optional[str] = None) -> 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, _ = 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: List[str], **kwargs: Any) -> tmt.utils.CommandOutput:
          """ Run given command via podman """
          return self.run(['podman'] + command, **kwargs)
  
      def execute(self,
                  command: Union[List[str], str],
                  friendly_command: Optional[str] = None,
                  test_session: bool = False,
                  silent: bool = False,
                  log: Optional[BaseLoggerFnType] = None,
                  **kwargs: Any) -> tmt.utils.CommandOutput:
          """ 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
          assert isinstance(command, str)
          return self.podman(
              ['exec'] + interactive +
              [self.container or 'dry', 'bash', '-c', command],
              log=log if log else self._command_verbose_logger,
              friendly_command=friendly_command or command,
              silent=silent,
              **kwargs)
  
      def push(
              self,
              source: Optional[str] = None,
              destination: Optional[str] = None,
              options: Optional[List[str]] = None) -> None:
          """ Make sure that the workdir has a correct selinux context """
          if not self.is_ready:
              return
  
          self.debug("Update selinux context of the run workdir.", level=3)
          assert self.parent.plan.workdir is not None  # narrow type
          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 source and destination:
              self.podman(["cp", source, f"{self.container}:{destination}"])
  
      def pull(
              self,
              source: Optional[str] = None,
              destination: Optional[str] = None,
              options: Optional[List[str]] = None,
              extend_options: Optional[List[str]] = None) -> None:
          """ Nothing to be done to pull workdir """
          if not self.is_ready:
              return
  
      def stop(self) -> None:
          """ Stop provisioned guest """
          if self.container:
              self.podman(['container', 'stop', self.container])
              self.info('container', 'stopped', 'green')
  
      def remove(self) -> None:
          """ Remove the container """
          if self.container:
              self.podman(['container', 'rm', '-f', self.container])
              self.info('container', 'removed', 'green')
  
  
  @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'.
      """
  
      _data_class = ProvisionPodmanData
      _guest_class = GuestContainer
  
      # Guest instance
      _guest = None
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ 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', 'force_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: str, default: Any = None) -> Any:
          """ Return default data for given option """
          if option == 'pull':
              return self.get('force-pull', default=default)
  
          return super().default(option, default=default)
  
      def go(self) -> None:
          """ Provision the container """
          super().go()
  
          # Show which image we are using
          pull = ' (force pull)' if self.get('force_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()
              }
  
          data = PodmanGuestData(**data_from_options)
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestContainer(data=data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self) -> Optional[GuestContainer]:
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self) -> List[str]:
          """ List of required packages needed for workdir sync """
          return GuestContainer.requires()
  
files: tmt/steps/provision/podman.py
files: import dataclasses
  import os
  from shlex import quote
  from typing import Any, List, Optional, Tuple, Union
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  from tmt.utils import BaseLoggerFnType
  
  # Timeout in seconds of waiting for a connection
  CONNECTION_TIMEOUT = 60
  
  # Defaults
  DEFAULT_IMAGE = "fedora"
  DEFAULT_USER = "root"
  
  
  @dataclasses.dataclass
  class PodmanGuestData(tmt.steps.provision.GuestData):
      image: str = DEFAULT_IMAGE
      user: str = DEFAULT_USER
      force_pull: bool = False
  
      container: Optional[str] = None
  
  
  @dataclasses.dataclass
  class ProvisionPodmanData(PodmanGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  class GuestContainer(tmt.Guest):
      """ Container Instance """
  
      _data_class = PodmanGuestData
  
      image: Optional[str]
      container: Optional[str]
      user: str
      force_pull: bool
      parent: tmt.steps.Step
  
      @property
      def is_ready(self) -> bool:
          """ Detect the guest is ready or not """
          # Check the container is running or not
          if self.container is None:
              return False
          cmd_output = self.podman(['container', 'inspect', '--format',
                                    '{{json .State.Running}}', self.container])
          cmd_stdout, cmd_stderr = cmd_output
          return str(cmd_stdout).strip() == 'true'
  
      def wake(self) -> None:
          """ Wake up the guest """
          self.debug(
              f"Waking up container '{self.container}'.", level=2, shift=0)
  
      def start(self) -> None:
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Check if the image is available
          assert self.image is not None
          command = ['image', 'exists', self.image]
          try:
              self.podman(command, message=f"Check for container image '{self.image}'.")
              needs_pull = False
          except tmt.utils.RunError:
              needs_pull = True
  
          # Pull image if not available or pull forced
          if needs_pull 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}'.")
          assert self.container is not None
          self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f'{workdir}:{workdir}:z',
               '-itd', '--user', self.user, self.image])
  
      def reboot(self, hard: bool = False,
                 command: Optional[Union[str, List[str], Tuple[str, ...]]] = None,
                 timeout: Optional[int] = None) -> bool:
          """ 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).")
          assert self.container is not None
          self.podman(['container', 'restart', self.container])
          return self.reconnect(timeout=timeout or CONNECTION_TIMEOUT)
  
      def ansible(self, playbook: str, extra_args: Optional[str] = None) -> 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, _ = 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: List[str], **kwargs: Any) -> tmt.utils.CommandOutput:
          """ Run given command via podman """
          return self.run(['podman'] + command, **kwargs)
  
      def execute(self,
                  command: Union[List[str], str],
                  friendly_command: Optional[str] = None,
                  test_session: bool = False,
                  silent: bool = False,
                  log: Optional[BaseLoggerFnType] = None,
                  **kwargs: Any) -> tmt.utils.CommandOutput:
          """ 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
          assert isinstance(command, str)
          return self.podman(
              ['exec'] + interactive +
              [self.container or 'dry', 'bash', '-c', command],
              log=log if log else self._command_verbose_logger,
              friendly_command=friendly_command or command,
              silent=silent,
              **kwargs)
  
      def push(
              self,
              source: Optional[str] = None,
              destination: Optional[str] = None,
              options: Optional[List[str]] = None) -> None:
          """ Make sure that the workdir has a correct selinux context """
          if not self.is_ready:
              return
  
          self.debug("Update selinux context of the run workdir.", level=3)
          assert self.parent.plan.workdir is not None  # narrow type
          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 source and destination:
              self.podman(["cp", source, f"{self.container}:{destination}"])
  
      def pull(
              self,
              source: Optional[str] = None,
              destination: Optional[str] = None,
              options: Optional[List[str]] = None,
              extend_options: Optional[List[str]] = None) -> None:
          """ Nothing to be done to pull workdir """
          if not self.is_ready:
              return
  
      def stop(self) -> None:
          """ Stop provisioned guest """
          if self.container:
              self.podman(['container', 'stop', self.container])
              self.info('container', 'stopped', 'green')
  
      def remove(self) -> None:
          """ Remove the container """
          if self.container:
              self.podman(['container', 'rm', '-f', self.container])
              self.info('container', 'removed', 'green')
  
  
  @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'.
      """
  
      _data_class = ProvisionPodmanData
      _guest_class = GuestContainer
  
      # Guest instance
      _guest = None
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ 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', 'force_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: str, default: Any = None) -> Any:
          """ Return default data for given option """
          if option == 'pull':
              return self.get('force-pull', default=default)
  
          return super().default(option, default=default)
  
      def wake(self, data: Optional[tmt.steps.provision.GuestData] = None) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
          # Wake up podman instance
          if data:
              guest = GuestContainer(data=data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self) -> None:
          """ Provision the container """
          super().go()
  
          # Show which image we are using
          pull = ' (force pull)' if self.get('force_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()
              }
  
          data = PodmanGuestData(**data_from_options)
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestContainer(data=data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self) -> Optional[GuestContainer]:
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self) -> List[str]:
          """ List of required packages needed for workdir sync """
          return GuestContainer.requires()
  
files: tmt/steps/provision/podman.py
files: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  import types
  from typing import TYPE_CHECKING, List, Optional
  
  import click
  import fmf
  import fmf.utils
  import requests
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  libvirt: Optional[types.ModuleType] = None
  testcloud: Optional[types.ModuleType] = None
  
  
  def import_testcloud() -> None:
      """
      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()
  
  
  @dataclasses.dataclass
  class TestcloudGuestData(tmt.steps.provision.GuestSshData):
      # 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
  
  
  @dataclasses.dataclass
  class ProvisionTestcloudData(TestcloudGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  class GuestTestcloud(tmt.GuestSsh):
      """
      Testcloud Instance
  
      The following keys are expected in the 'data' dictionary::
  
          image ...... qcov image name or url
          user ....... user name to log in
          memory ..... memory size for vm
          disk ....... disk size for vm
          connection . either session (default) or system, to be passed to qemu
          arch ....... architecture for the VM, host arch is the default
      """
  
      _data_class = TestcloudGuestData
  
      image: str
      image_url: Optional[str]
      instance_name: Optional[str]
      memory: int
      disk: str
      connection: str
      arch: str
  
      # Not to be saved, recreated from image_url/instance_name/... every
      # time guest is instantiated.
      # FIXME: ignore[name-defined]: https://github.com/teemtee/tmt/issues/1616
      _image: Optional['testcloud.image.Image'] = None  # type: ignore[name-defined]
      _instance: Optional['testcloud.instance.Instance'] = None  # type: ignore[name-defined]
  
      @property
      def is_ready(self) -> bool:
          if self._instance is None:
              return False
  
          assert testcloud is not None
          assert libvirt is not None
          try:
              state = testcloud.instance._find_domain(self._instance.name, self._instance.connection)
              # Note the type of variable 'state' is 'Any'. Hence, we don't use:
              #     return state == 'running'
              # to avoid error from type checking.
              return True if state == 'running' else False
          except libvirt.libvirtError:
              return False
  
      def _get_url(self, url: str, message: str) -> requests.Response:
          """ Get url, retry when fails, return response """
  
          def try_get_url() -> requests.Response:
              try:
                  with retry_session() as session:
                      response = session.get(url)
  
                  if response.ok:
                      return response
  
              except requests.RequestException:
                  pass
              finally:
                  raise tmt.utils.WaitingIncomplete()
  
          try:
              return tmt.utils.wait(
                  self, try_get_url, datetime.timedelta(
                      seconds=DEFAULT_CONNECT_TIMEOUT), tick=1)
  
          except tmt.utils.WaitingTimedOutError:
              raise ProvisionError(
                  f'Failed to {message} in {DEFAULT_CONNECT_TIMEOUT}s.')
  
      def _guess_image_url(self, name: str) -> str:
          """ Guess image url for given name """
  
          # Try to check if given url is a local file
          if os.path.isabs(name) and os.path.isfile(name):
              return f'file://{name}'
  
          name = name.lower().strip()
          url: Optional[str] = None
  
          # Map fedora aliases (e.g. rawhide, fedora, fedora-32, f-32, f32)
          matched_fedora = re.match(r'^f(edora)?-?(\d+)$', name)
          # Map fedora coreos aliases (e.g. stable, next or testing)
          matched_fedora_coreos = re.match(r'^f(edora-coreos)?-?(stable|testing|next)$', name)
          # Map centos aliases (e.g. centos:X, centos, centos-stream:X)
          matched_centos = [re.match(r'^c(entos)?-?(\d+)$', name),
                            re.match(r'^c(entos-stream)?-?(\d+)$', name)]
          matched_ubuntu = re.match(r'^u(buntu)?-?(\w+)$', name)
          matched_debian = re.match(r'^d(ebian)?-?(\w+)$', name)
  
          assert testcloud is not None
          # Plain name match means we want the latest release
          try:
              if name == 'fedora':
                  url = testcloud.util.get_fedora_image_url("latest", self.arch)
              elif name == 'fedora-coreos':
                  url = testcloud.util.get_fedora_image_url("stable", self.arch)
              elif name == 'centos':
                  url = testcloud.util.get_centos_image_url("latest", stream=False, arch=self.arch)
              elif name == 'centos-stream':
                  url = testcloud.util.get_centos_image_url("latest", stream=True, arch=self.arch)
              elif name == 'ubuntu':
                  url = testcloud.util.get_ubuntu_image_url("latest", self.arch)
              elif name == 'debian':
                  url = testcloud.util.get_debian_image_url("latest", self.arch)
  
              elif matched_fedora:
                  url = testcloud.util.get_fedora_image_url(
                      matched_fedora.group(2), self.arch)
              elif matched_fedora_coreos:
                  url = testcloud.util.get_fedora_image_url(
                      matched_fedora_coreos.group(2), self.arch)
              elif matched_centos[0]:
                  url = testcloud.util.get_centos_image_url(
                      matched_centos[0].group(2), stream=False, arch=self.arch)
              elif matched_centos[1]:
                  url = testcloud.util.get_centos_image_url(
                      matched_centos[1].group(2), stream=True, arch=self.arch)
              elif matched_ubuntu:
                  url = testcloud.util.get_ubuntu_image_url(
                      matched_ubuntu.group(2), self.arch)
              elif matched_debian:
                  url = testcloud.util.get_debian_image_url(
                      matched_debian.group(2), self.arch)
              elif 'rawhide' in name:
                  url = testcloud.util.get_fedora_image_url("rawhide", self.arch)
          except Exception as error:
              raise ProvisionError("Could not get image url.") from error
  
          if not url:
              raise ProvisionError(f"Could not map '{name}' to compose.")
          return url
  
      @staticmethod
      def _create_template() -> None:
          """ Create libvirt domain template """
          # Write always to ovewrite possible outdated version
          with open(DOMAIN_TEMPLATE_FILE, 'w') as template:
              template.write(DOMAIN_TEMPLATE)
  
      def wake(self) -> None:
          """ Wake up the guest """
          self.debug(
              f"Waking up testcloud instance '{self.instance_name}'.",
              level=2, shift=0)
          self.prepare_config()
          assert testcloud is not None
          self._image = testcloud.image.Image(self.image_url)
          if self.instance_name is None:
              raise ProvisionError(f"The instance name '{self.instance_name}' is invalid.")
          self._instance = testcloud.instance.Instance(
              self.instance_name, image=self._image,
              connection=f"qemu:///{self.connection}", desired_arch=self.arch)
  
      def prepare_ssh_key(self, key_type: Optional[str] = None) -> None:
          """ Prepare ssh key for authentication """
          # Create ssh key paths
          key_name = "id_{}".format(key_type if key_type is not None else 'rsa')
          assert self.workdir is not None
          self.key = [os.path.join(self.workdir, key_name)]
          self.pubkey = os.path.join(self.workdir, f'{key_name}.pub')
  
          # Generate ssh key
          self.debug('Generating an ssh key.')
          command = ["ssh-keygen", "-f", self.key[0], "-N", ""]
          if key_type is not None:
              command.extend(["-t", key_type])
          self.run(command)
          self.verbose('key', self.key[0], 'green')
          with open(self.pubkey, 'r') as pubkey:
              self.config.USER_DATA = USER_DATA.format(
                  user_name=self.user, public_key=pubkey.read())
          with open(self.pubkey, 'r') as pubkey:
              self.config.COREOS_DATA = COREOS_DATA.format(
                  user_name=self.user, public_key=pubkey.read())
  
      def prepare_config(self) -> None:
          """ Prepare common configuration """
          import_testcloud()
  
          # Get configuration
          assert testcloud is not None
          self.config = testcloud.config.get_config()
  
          # Make sure download progress is disabled unless in debug mode,
          # so it does not spoil our logging
          self.config.DOWNLOAD_PROGRESS = self.opt('debug') > 2
          self.config.DOWNLOAD_PROGRESS_VERBOSE = False
  
          # Configure to tmt's storage directories
          self.config.DATA_DIR = TESTCLOUD_DATA
          self.config.STORE_DIR = TESTCLOUD_IMAGES
  
      def start(self) -> None:
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Make sure required directories exist
          os.makedirs(TESTCLOUD_DATA, exist_ok=True)
          os.makedirs(TESTCLOUD_IMAGES, exist_ok=True)
  
          # Make sure libvirt domain template exists
          GuestTestcloud._create_template()
  
          # Prepare config
          self.prepare_config()
  
          # Kick off image URL with the given image
          self.image_url = self.image
  
          # If image does not start with http/https/file, consider it a
          # mapping value and try to guess the URL
          if not re.match(r'^(?:https?|file)://.*', self.image_url):
              self.image_url = self._guess_image_url(self.image_url)
              self.debug(f"Guessed image url: '{self.image_url}'", level=3)
  
          # Initialize and prepare testcloud image
          assert testcloud is not None
          self._image = testcloud.image.Image(self.image_url)
          self.verbose('qcow', self._image.name, 'green')
          if not os.path.exists(self._image.local_path):
              self.info('progress', 'downloading...', 'cyan')
          try:
              self._image.prepare()
          except FileNotFoundError as error:
              raise ProvisionError(
                  f"Image '{self._image.local_path}' not found.") from error
          except (testcloud.exceptions.TestcloudPermissionsError,
                  PermissionError) as error:
              raise ProvisionError(
                  f"Failed to prepare the image. Check the '{TESTCLOUD_IMAGES}' "
                  f"directory permissions.") from error
          except KeyError as error:
              raise ProvisionError(
                  f"Failed to prepare image '{self.image_url}'.") from error
  
          # Prepare hostname (get rid of possible unwanted characters)
          hostname = re.sub(r"[^a-zA-Z0-9\-]+", "-", self.name.lower()).strip("-")
  
          # Create instance
          self.instance_name = self._tmt_name()
          self._instance = testcloud.instance.Instance(
              name=self.instance_name,
              hostname=hostname,
              image=self._image,
              connection=f"qemu:///{self.connection}",
              desired_arch=self.arch)
          self.verbose('name', self.instance_name, 'green')
  
          # Decide if we want to multiply timeouts when emulating an architecture
          time_coeff = NON_KVM_TIMEOUT_COEF if not self._instance.kvm else 1
  
          # Decide which networking setup to use
          # Autodetect works with libguestfs python bindings
          # We fall back to basic heuristics based on file name
          # without that installed (eg. from pypi).
          # https://bugzilla.redhat.com/show_bug.cgi?id=1075594
          try:
              import guestfs  # noqa: F401
          except ImportError:
              match_legacy = re.search(
                  r'(rhel|centos).*-7', self.image_url.lower())
              if match_legacy:
                  self._instance.pci_net = "e1000"
              else:
                  self._instance.pci_net = "virtio-net-pci"
  
          # Prepare ssh key
          # TODO: Maybe... some better way to do this?
          if "coreos" in self.image.lower():
              self._instance.coreos = True
              # prepare_ssh_key() writes key directly to COREOS_DATA
              self._instance.ssh_path = []
          self.prepare_ssh_key(SSH_KEYGEN_TYPE)
  
          # Boot the virtual machine
          self.info('progress', 'booting...', 'cyan')
          self._instance.ram = self.memory
          self._instance.disk_size = self.disk
          assert libvirt is not None
          try:
              self._instance.prepare()
              self._instance.spawn_vm()
              self._instance.start(DEFAULT_BOOT_TIMEOUT * time_coeff)
          except (testcloud.exceptions.TestcloudInstanceError,
                  libvirt.libvirtError) as error:
              raise ProvisionError(
                  f'Failed to boot testcloud instance ({error}).')
          self.guest = self._instance.get_ip()
          self.port = int(self._instance.get_instance_port())
          self.verbose('ip', self.guest, 'green')
          self.verbose('port', str(self.port), 'green')
          self._instance.create_ip_file(self.guest)
  
          # Wait a bit until the box is up
          if not self.reconnect(
                  timeout=DEFAULT_CONNECT_TIMEOUT *
                  time_coeff,
                  tick=1):
              raise ProvisionError(
                  f"Failed to connect in {DEFAULT_CONNECT_TIMEOUT * time_coeff}s.")
  
          if not self._instance.kvm:
              self.debug(
                  f"Waiting {NON_KVM_ADDITIONAL_WAIT} seconds "
                  f"for non-kvm instance...")
              time.sleep(NON_KVM_ADDITIONAL_WAIT)
  
      def stop(self) -> None:
          """ Stop provisioned guest """
          super().stop()
          # Stop only if the instance successfully booted
          if self._instance and self.guest:
              self.debug(f"Stopping testcloud instance '{self.instance_name}'.")
              assert testcloud is not None
              try:
                  self._instance.stop()
              except testcloud.exceptions.TestcloudInstanceError as error:
                  raise tmt.utils.ProvisionError(
                      f"Failed to stop testcloud instance: {error}")
              self.info('guest', 'stopped', 'green')
  
      def remove(self) -> None:
          """ Remove the guest (disk cleanup) """
          if self._instance:
              self.debug(f"Removing testcloud instance '{self.instance_name}'.")
              try:
                  self._instance.remove(autostop=True)
              except FileNotFoundError as error:
                  raise tmt.utils.ProvisionError(
                      f"Failed to remove testcloud instance: {error}")
              self.info('guest', 'removed', 'green')
  
      def reboot(self,
                 hard: bool = False,
                 command: Optional[str] = None,
                 timeout: Optional[int] = None,
                 tick: float = tmt.utils.DEFAULT_WAIT_TICK,
                 tick_increase: float = tmt.utils.DEFAULT_WAIT_TICK_INCREASE) -> bool:
          """ Reboot the guest, return True if successful """
          # Use custom reboot command if provided
          if command:
              return super().reboot(hard=hard, command=command)
          if not self._instance:
              raise tmt.utils.ProvisionError("No instance initialized.")
          self._instance.reboot(soft=not hard)
          return self.reconnect(timeout=timeout)
  
  
  @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.
      """
  
      _data_class = ProvisionTestcloudData
      _guest_class = GuestTestcloud
  
      # Guest instance
      _guest = None
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ 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 go(self) -> None:
          """ Provision the testcloud instance """
          super().go()
  
          # Give info about provided data
          data = TestcloudGuestData(**{
              key: self.get(key)
              for key in TestcloudGuestData.keys()
              })
  
          # Once plan schema is enforced this won't be necessary
          # click enforces int for cmdline and schema validation
          # will make sure 'int' gets from plan data.
          # Another key is 'port' however that is not exposed to the cli
          for int_key in ["memory", "disk", "port"]:
              value = getattr(data, int_key)
              if value is not None:
                  try:
                      setattr(data, int_key, int(value))
                  except ValueError:
                      raise tmt.utils.SpecificationError(
                          f"Value '{value}' cannot be converted to int for '{int_key}' attribute.")
  
          for key, value in data.items():
              if key == 'memory':
                  self.info('memory', f"{value} MB", 'green')
              elif key == 'disk':
                  self.info('disk', f"{value} GB", 'green')
              elif key == 'connection':
                  self.verbose('connection', value, 'green')
              elif key == 'key':
                  if value:
                      self.info('key', fmf.utils.listed(value), 'green')
              elif key == 'ssh_option':
                  if value:
                      self.info('ssh options', fmf.utils.listed(value), 'green')
              elif value is not None:
                  self.info(key, value, 'green')
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestTestcloud(data=data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self) -> Optional[tmt.Guest]:
          """ Return the provisioned guest """
          return self._guest
  
      @classmethod
      def clean_images(cls, clean: 'tmt.base.Clean', dry: bool) -> bool:
          """ Remove the testcloud images """
          clean.info('testcloud', shift=1, color='green')
          if not os.path.exists(TESTCLOUD_IMAGES):
              clean.warn(
                  f"Directory '{TESTCLOUD_IMAGES}' does not exist.", shift=2)
              return True
          successful = True
          for image in os.listdir(TESTCLOUD_IMAGES):
              image = os.path.join(TESTCLOUD_IMAGES, image)
              if dry:
                  clean.verbose(f"Would remove '{image}'.", shift=2)
              else:
                  clean.verbose(f"Removing '{image}'.", shift=2)
                  try:
                      os.remove(image)
                  except OSError:
                      clean.fail(f"Failed to remove '{image}'.", shift=2)
                      successful = False
          return successful
  
files: tmt/steps/provision/testcloud.py
files: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  import types
  from typing import TYPE_CHECKING, List, Optional
  
  import click
  import fmf
  import fmf.utils
  import requests
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  libvirt: Optional[types.ModuleType] = None
  testcloud: Optional[types.ModuleType] = None
  
  
  def import_testcloud() -> None:
      """
      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()
  
  
  @dataclasses.dataclass
  class TestcloudGuestData(tmt.steps.provision.GuestSshData):
      # 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
  
  
  @dataclasses.dataclass
  class ProvisionTestcloudData(TestcloudGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  class GuestTestcloud(tmt.GuestSsh):
      """
      Testcloud Instance
  
      The following keys are expected in the 'data' dictionary::
  
          image ...... qcov image name or url
          user ....... user name to log in
          memory ..... memory size for vm
          disk ....... disk size for vm
          connection . either session (default) or system, to be passed to qemu
          arch ....... architecture for the VM, host arch is the default
      """
  
      _data_class = TestcloudGuestData
  
      image: str
      image_url: Optional[str]
      instance_name: Optional[str]
      memory: int
      disk: str
      connection: str
      arch: str
  
      # Not to be saved, recreated from image_url/instance_name/... every
      # time guest is instantiated.
      # FIXME: ignore[name-defined]: https://github.com/teemtee/tmt/issues/1616
      _image: Optional['testcloud.image.Image'] = None  # type: ignore[name-defined]
      _instance: Optional['testcloud.instance.Instance'] = None  # type: ignore[name-defined]
  
      @property
      def is_ready(self) -> bool:
          if self._instance is None:
              return False
  
          assert testcloud is not None
          assert libvirt is not None
          try:
              state = testcloud.instance._find_domain(self._instance.name, self._instance.connection)
              # Note the type of variable 'state' is 'Any'. Hence, we don't use:
              #     return state == 'running'
              # to avoid error from type checking.
              return True if state == 'running' else False
          except libvirt.libvirtError:
              return False
  
      def _get_url(self, url: str, message: str) -> requests.Response:
          """ Get url, retry when fails, return response """
  
          def try_get_url() -> requests.Response:
              try:
                  with retry_session() as session:
                      response = session.get(url)
  
                  if response.ok:
                      return response
  
              except requests.RequestException:
                  pass
              finally:
                  raise tmt.utils.WaitingIncomplete()
  
          try:
              return tmt.utils.wait(
                  self, try_get_url, datetime.timedelta(
                      seconds=DEFAULT_CONNECT_TIMEOUT), tick=1)
  
          except tmt.utils.WaitingTimedOutError:
              raise ProvisionError(
                  f'Failed to {message} in {DEFAULT_CONNECT_TIMEOUT}s.')
  
      def _guess_image_url(self, name: str) -> str:
          """ Guess image url for given name """
  
          # Try to check if given url is a local file
          if os.path.isabs(name) and os.path.isfile(name):
              return f'file://{name}'
  
          name = name.lower().strip()
          url: Optional[str] = None
  
          # Map fedora aliases (e.g. rawhide, fedora, fedora-32, f-32, f32)
          matched_fedora = re.match(r'^f(edora)?-?(\d+)$', name)
          # Map fedora coreos aliases (e.g. stable, next or testing)
          matched_fedora_coreos = re.match(r'^f(edora-coreos)?-?(stable|testing|next)$', name)
          # Map centos aliases (e.g. centos:X, centos, centos-stream:X)
          matched_centos = [re.match(r'^c(entos)?-?(\d+)$', name),
                            re.match(r'^c(entos-stream)?-?(\d+)$', name)]
          matched_ubuntu = re.match(r'^u(buntu)?-?(\w+)$', name)
          matched_debian = re.match(r'^d(ebian)?-?(\w+)$', name)
  
          assert testcloud is not None
          # Plain name match means we want the latest release
          try:
              if name == 'fedora':
                  url = testcloud.util.get_fedora_image_url("latest", self.arch)
              elif name == 'fedora-coreos':
                  url = testcloud.util.get_fedora_image_url("stable", self.arch)
              elif name == 'centos':
                  url = testcloud.util.get_centos_image_url("latest", stream=False, arch=self.arch)
              elif name == 'centos-stream':
                  url = testcloud.util.get_centos_image_url("latest", stream=True, arch=self.arch)
              elif name == 'ubuntu':
                  url = testcloud.util.get_ubuntu_image_url("latest", self.arch)
              elif name == 'debian':
                  url = testcloud.util.get_debian_image_url("latest", self.arch)
  
              elif matched_fedora:
                  url = testcloud.util.get_fedora_image_url(
                      matched_fedora.group(2), self.arch)
              elif matched_fedora_coreos:
                  url = testcloud.util.get_fedora_image_url(
                      matched_fedora_coreos.group(2), self.arch)
              elif matched_centos[0]:
                  url = testcloud.util.get_centos_image_url(
                      matched_centos[0].group(2), stream=False, arch=self.arch)
              elif matched_centos[1]:
                  url = testcloud.util.get_centos_image_url(
                      matched_centos[1].group(2), stream=True, arch=self.arch)
              elif matched_ubuntu:
                  url = testcloud.util.get_ubuntu_image_url(
                      matched_ubuntu.group(2), self.arch)
              elif matched_debian:
                  url = testcloud.util.get_debian_image_url(
                      matched_debian.group(2), self.arch)
              elif 'rawhide' in name:
                  url = testcloud.util.get_fedora_image_url("rawhide", self.arch)
          except Exception as error:
              raise ProvisionError("Could not get image url.") from error
  
          if not url:
              raise ProvisionError(f"Could not map '{name}' to compose.")
          return url
  
      @staticmethod
      def _create_template() -> None:
          """ Create libvirt domain template """
          # Write always to ovewrite possible outdated version
          with open(DOMAIN_TEMPLATE_FILE, 'w') as template:
              template.write(DOMAIN_TEMPLATE)
  
      def wake(self) -> None:
          """ Wake up the guest """
          self.debug(
              f"Waking up testcloud instance '{self.instance_name}'.",
              level=2, shift=0)
          self.prepare_config()
          assert testcloud is not None
          self._image = testcloud.image.Image(self.image_url)
          if self.instance_name is None:
              raise ProvisionError(f"The instance name '{self.instance_name}' is invalid.")
          self._instance = testcloud.instance.Instance(
              self.instance_name, image=self._image,
              connection=f"qemu:///{self.connection}", desired_arch=self.arch)
  
      def prepare_ssh_key(self, key_type: Optional[str] = None) -> None:
          """ Prepare ssh key for authentication """
          # Create ssh key paths
          key_name = "id_{}".format(key_type if key_type is not None else 'rsa')
          assert self.workdir is not None
          self.key = [os.path.join(self.workdir, key_name)]
          self.pubkey = os.path.join(self.workdir, f'{key_name}.pub')
  
          # Generate ssh key
          self.debug('Generating an ssh key.')
          command = ["ssh-keygen", "-f", self.key[0], "-N", ""]
          if key_type is not None:
              command.extend(["-t", key_type])
          self.run(command)
          self.verbose('key', self.key[0], 'green')
          with open(self.pubkey, 'r') as pubkey:
              self.config.USER_DATA = USER_DATA.format(
                  user_name=self.user, public_key=pubkey.read())
          with open(self.pubkey, 'r') as pubkey:
              self.config.COREOS_DATA = COREOS_DATA.format(
                  user_name=self.user, public_key=pubkey.read())
  
      def prepare_config(self) -> None:
          """ Prepare common configuration """
          import_testcloud()
  
          # Get configuration
          assert testcloud is not None
          self.config = testcloud.config.get_config()
  
          # Make sure download progress is disabled unless in debug mode,
          # so it does not spoil our logging
          self.config.DOWNLOAD_PROGRESS = self.opt('debug') > 2
          self.config.DOWNLOAD_PROGRESS_VERBOSE = False
  
          # Configure to tmt's storage directories
          self.config.DATA_DIR = TESTCLOUD_DATA
          self.config.STORE_DIR = TESTCLOUD_IMAGES
  
      def start(self) -> None:
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Make sure required directories exist
          os.makedirs(TESTCLOUD_DATA, exist_ok=True)
          os.makedirs(TESTCLOUD_IMAGES, exist_ok=True)
  
          # Make sure libvirt domain template exists
          GuestTestcloud._create_template()
  
          # Prepare config
          self.prepare_config()
  
          # Kick off image URL with the given image
          self.image_url = self.image
  
          # If image does not start with http/https/file, consider it a
          # mapping value and try to guess the URL
          if not re.match(r'^(?:https?|file)://.*', self.image_url):
              self.image_url = self._guess_image_url(self.image_url)
              self.debug(f"Guessed image url: '{self.image_url}'", level=3)
  
          # Initialize and prepare testcloud image
          assert testcloud is not None
          self._image = testcloud.image.Image(self.image_url)
          self.verbose('qcow', self._image.name, 'green')
          if not os.path.exists(self._image.local_path):
              self.info('progress', 'downloading...', 'cyan')
          try:
              self._image.prepare()
          except FileNotFoundError as error:
              raise ProvisionError(
                  f"Image '{self._image.local_path}' not found.") from error
          except (testcloud.exceptions.TestcloudPermissionsError,
                  PermissionError) as error:
              raise ProvisionError(
                  f"Failed to prepare the image. Check the '{TESTCLOUD_IMAGES}' "
                  f"directory permissions.") from error
          except KeyError as error:
              raise ProvisionError(
                  f"Failed to prepare image '{self.image_url}'.") from error
  
          # Prepare hostname (get rid of possible unwanted characters)
          hostname = re.sub(r"[^a-zA-Z0-9\-]+", "-", self.name.lower()).strip("-")
  
          # Create instance
          self.instance_name = self._tmt_name()
          self._instance = testcloud.instance.Instance(
              name=self.instance_name,
              hostname=hostname,
              image=self._image,
              connection=f"qemu:///{self.connection}",
              desired_arch=self.arch)
          self.verbose('name', self.instance_name, 'green')
  
          # Decide if we want to multiply timeouts when emulating an architecture
          time_coeff = NON_KVM_TIMEOUT_COEF if not self._instance.kvm else 1
  
          # Decide which networking setup to use
          # Autodetect works with libguestfs python bindings
          # We fall back to basic heuristics based on file name
          # without that installed (eg. from pypi).
          # https://bugzilla.redhat.com/show_bug.cgi?id=1075594
          try:
              import guestfs  # noqa: F401
          except ImportError:
              match_legacy = re.search(
                  r'(rhel|centos).*-7', self.image_url.lower())
              if match_legacy:
                  self._instance.pci_net = "e1000"
              else:
                  self._instance.pci_net = "virtio-net-pci"
  
          # Prepare ssh key
          # TODO: Maybe... some better way to do this?
          if "coreos" in self.image.lower():
              self._instance.coreos = True
              # prepare_ssh_key() writes key directly to COREOS_DATA
              self._instance.ssh_path = []
          self.prepare_ssh_key(SSH_KEYGEN_TYPE)
  
          # Boot the virtual machine
          self.info('progress', 'booting...', 'cyan')
          self._instance.ram = self.memory
          self._instance.disk_size = self.disk
          assert libvirt is not None
          try:
              self._instance.prepare()
              self._instance.spawn_vm()
              self._instance.start(DEFAULT_BOOT_TIMEOUT * time_coeff)
          except (testcloud.exceptions.TestcloudInstanceError,
                  libvirt.libvirtError) as error:
              raise ProvisionError(
                  f'Failed to boot testcloud instance ({error}).')
          self.guest = self._instance.get_ip()
          self.port = int(self._instance.get_instance_port())
          self.verbose('ip', self.guest, 'green')
          self.verbose('port', str(self.port), 'green')
          self._instance.create_ip_file(self.guest)
  
          # Wait a bit until the box is up
          if not self.reconnect(
                  timeout=DEFAULT_CONNECT_TIMEOUT *
                  time_coeff,
                  tick=1):
              raise ProvisionError(
                  f"Failed to connect in {DEFAULT_CONNECT_TIMEOUT * time_coeff}s.")
  
          if not self._instance.kvm:
              self.debug(
                  f"Waiting {NON_KVM_ADDITIONAL_WAIT} seconds "
                  f"for non-kvm instance...")
              time.sleep(NON_KVM_ADDITIONAL_WAIT)
  
      def stop(self) -> None:
          """ Stop provisioned guest """
          super().stop()
          # Stop only if the instance successfully booted
          if self._instance and self.guest:
              self.debug(f"Stopping testcloud instance '{self.instance_name}'.")
              assert testcloud is not None
              try:
                  self._instance.stop()
              except testcloud.exceptions.TestcloudInstanceError as error:
                  raise tmt.utils.ProvisionError(
                      f"Failed to stop testcloud instance: {error}")
              self.info('guest', 'stopped', 'green')
  
      def remove(self) -> None:
          """ Remove the guest (disk cleanup) """
          if self._instance:
              self.debug(f"Removing testcloud instance '{self.instance_name}'.")
              try:
                  self._instance.remove(autostop=True)
              except FileNotFoundError as error:
                  raise tmt.utils.ProvisionError(
                      f"Failed to remove testcloud instance: {error}")
              self.info('guest', 'removed', 'green')
  
      def reboot(self,
                 hard: bool = False,
                 command: Optional[str] = None,
                 timeout: Optional[int] = None,
                 tick: float = tmt.utils.DEFAULT_WAIT_TICK,
                 tick_increase: float = tmt.utils.DEFAULT_WAIT_TICK_INCREASE) -> bool:
          """ Reboot the guest, return True if successful """
          # Use custom reboot command if provided
          if command:
              return super().reboot(hard=hard, command=command)
          if not self._instance:
              raise tmt.utils.ProvisionError("No instance initialized.")
          self._instance.reboot(soft=not hard)
          return self.reconnect(timeout=timeout)
  
  
  @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.
      """
  
      _data_class = ProvisionTestcloudData
      _guest_class = GuestTestcloud
  
      # Guest instance
      _guest = None
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ 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)
  
      # FIXME: ignore - https://github.com/teemtee/tmt/issues/1437
      def wake(self, data: Optional[TestcloudGuestData] = None) -> None:  # type: ignore[override]
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
  
          # Wake up testcloud instance
          if data:
              guest = GuestTestcloud(data=data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self) -> None:
          """ Provision the testcloud instance """
          super().go()
  
          # Give info about provided data
          data = TestcloudGuestData(**{
              key: self.get(key)
              for key in TestcloudGuestData.keys()
              })
  
          # Once plan schema is enforced this won't be necessary
          # click enforces int for cmdline and schema validation
          # will make sure 'int' gets from plan data.
          # Another key is 'port' however that is not exposed to the cli
          for int_key in ["memory", "disk", "port"]:
              value = getattr(data, int_key)
              if value is not None:
                  try:
                      setattr(data, int_key, int(value))
                  except ValueError:
                      raise tmt.utils.SpecificationError(
                          f"Value '{value}' cannot be converted to int for '{int_key}' attribute.")
  
          for key, value in data.items():
              if key == 'memory':
                  self.info('memory', f"{value} MB", 'green')
              elif key == 'disk':
                  self.info('disk', f"{value} GB", 'green')
              elif key == 'connection':
                  self.verbose('connection', value, 'green')
              elif key == 'key':
                  if value:
                      self.info('key', fmf.utils.listed(value), 'green')
              elif key == 'ssh_option':
                  if value:
                      self.info('ssh options', fmf.utils.listed(value), 'green')
              elif value is not None:
                  self.info(key, value, 'green')
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestTestcloud(data=data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self) -> Optional[tmt.Guest]:
          """ Return the provisioned guest """
          return self._guest
  
      @classmethod
      def clean_images(cls, clean: 'tmt.base.Clean', dry: bool) -> bool:
          """ Remove the testcloud images """
          clean.info('testcloud', shift=1, color='green')
          if not os.path.exists(TESTCLOUD_IMAGES):
              clean.warn(
                  f"Directory '{TESTCLOUD_IMAGES}' does not exist.", shift=2)
              return True
          successful = True
          for image in os.listdir(TESTCLOUD_IMAGES):
              image = os.path.join(TESTCLOUD_IMAGES, image)
              if dry:
                  clean.verbose(f"Would remove '{image}'.", shift=2)
              else:
                  clean.verbose(f"Removing '{image}'.", shift=2)
                  try:
                      os.remove(image)
                  except OSError:
                      clean.fail(f"Failed to remove '{image}'.", shift=2)
                      successful = False
          return successful
  
files: tmt/steps/provision/testcloud.py