files.new_contents: #!/bin/bash
  # vim: dict+=/usr/share/beakerlib/dictionary.vim cpt=.,w,b,u,t,i,k
  . /usr/share/beakerlib/beakerlib.sh || exit 1
  
  rlJournalStart
      rlPhaseStartSetup
          rlRun "run=\$(mktemp -d)" 0 "Create run directory"
          rlRun "pushd data"
          rlRun "set -o pipefail"
      rlPhaseEnd
  
      opt="--scratch -i $run"
      rlPhaseStartTest "Wrong provision"
          rlRun -s "tmt run $opt plan -n wrong provision finish" \
              2 "Names not unique"
          rlAssertGrep "must be unique" $rlRun_LOG
      rlPhaseEnd
  
      rlPhaseStartTest "Correct provision without roles"
          rlRun -s "tmt run $opt plan -n noroles provision finish" 0
          rlAssertGrep "2 guests provisioned" $rlRun_LOG
          # how: container should be there twice
          rlRun "grep 'how: container' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "2" lines
  
          # 2 guests without role are saved
          guests="$run/noroles/provision/guests.yaml"
          rlAssertNotGrep "role: [a-z]" $guests "File '$guests' should leave role unspecified"
          rlAssertGrep "client" $guests
          rlAssertGrep "server" $guests
  
          # 2 guests were removed
          rlRun "grep 'container: removed' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "2" lines
  
          rlRun "rm lines"
      rlPhaseEnd
  
      rlPhaseStartTest "Correct provision with roles"
          rlRun -s "tmt run $opt plan -n /roles provision finish" 0
          rlAssertGrep "4 guests provisioned" $rlRun_LOG
          # how: container should be there 4 times
          rlRun "grep 'how: container' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "4" lines
  
          guests="$run/roles/provision/guests.yaml"
          rlAssertGrep "client-one" $guests
          rlAssertGrep "client-two" $guests
          rlAssertGrep "server-one" $guests
          rlAssertGrep "server-two" $guests
  
          # Each guest has a role
          rlRun "grep 'role: ' $guests | wc -l > lines"
          rlAssertGrep "4" "lines"
  
          # 4 guests were removed
          rlRun "grep 'container: removed' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "4" lines
  
          rlRun "rm lines"
      rlPhaseEnd
  
      rlPhaseStartTest "Full plan without roles"
          rlRun -s "tmt run $opt plan -n noroles"
      rlPhaseEnd
  
      rlPhaseStartCleanup
          rlRun "popd"
          rlRun "rm -r $run" 0 "Remove run directory"
      rlPhaseEnd
  rlJournalEnd
  
files.new_path: tests/multihost/provision/test.sh
files.old_contents: #!/bin/bash
  # vim: dict+=/usr/share/beakerlib/dictionary.vim cpt=.,w,b,u,t,i,k
  . /usr/share/beakerlib/beakerlib.sh || exit 1
  
  rlJournalStart
      rlPhaseStartSetup
          rlRun "run=\$(mktemp -d)" 0 "Create run directory"
          rlRun "pushd data"
          rlRun "set -o pipefail"
      rlPhaseEnd
  
      opt="--scratch -i $run"
      rlPhaseStartTest "Wrong provision"
          rlRun -s "tmt run $opt plan -n wrong provision finish" \
              2 "Names not unique"
          rlAssertGrep "must be unique" $rlRun_LOG
      rlPhaseEnd
  
      rlPhaseStartTest "Correct provision without roles"
          rlRun -s "tmt run $opt plan -n noroles provision finish" 0
          rlAssertGrep "2 guests provisioned" $rlRun_LOG
          # how: container should be there twice
          rlRun "grep 'how: container' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "2" lines
  
          # 2 guests without role are saved
          guests="$run/noroles/provision/guests.yaml"
          rlAssertNotGrep "role" $guests
          rlAssertGrep "client" $guests
          rlAssertGrep "server" $guests
  
          # 2 guests were removed
          rlRun "grep 'container: removed' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "2" lines
  
          rlRun "rm lines"
      rlPhaseEnd
  
      rlPhaseStartTest "Correct provision with roles"
          rlRun -s "tmt run $opt plan -n /roles provision finish" 0
          rlAssertGrep "4 guests provisioned" $rlRun_LOG
          # how: container should be there 4 times
          rlRun "grep 'how: container' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "4" lines
  
          guests="$run/roles/provision/guests.yaml"
          rlAssertGrep "client-one" $guests
          rlAssertGrep "client-two" $guests
          rlAssertGrep "server-one" $guests
          rlAssertGrep "server-two" $guests
  
          # Each guest has a role
          rlRun "grep 'role: ' $guests | wc -l > lines"
          rlAssertGrep "4" "lines"
  
          # 4 guests were removed
          rlRun "grep 'container: removed' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "4" lines
  
          rlRun "rm lines"
      rlPhaseEnd
  
      rlPhaseStartTest "Full plan without roles"
          rlRun -s "tmt run $opt plan -n noroles"
      rlPhaseEnd
  
      rlPhaseStartCleanup
          rlRun "popd"
          rlRun "rm -r $run" 0 "Remove run directory"
      rlPhaseEnd
  rlJournalEnd
  
files.old_path: tests/multihost/provision/test.sh
files.new_contents: import collections
  import dataclasses
  import datetime
  import os
  import random
  import re
  import shlex
  import string
  import subprocess
  import tempfile
  import time
  from typing import Dict, List, Optional, Type
  
  import click
  import fmf
  
  import tmt
  import tmt.plugins
  import tmt.utils
  
  # Timeout in seconds of waiting for a connection after reboot
  # (long enough to allow operations such as system upgrade)
  # FIXME: Shorten once "tmt-reboot -t <timeout>" is implemented
  CONNECTION_TIMEOUT = 60 * 60
  
  # Wait time when reboot happens in seconds
  RECONNECT_INITIAL_WAIT_TIME = 5
  
  # Default rsync options
  DEFAULT_RSYNC_OPTIONS = [
      "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  
  class Provision(tmt.steps.Step):
      """ Provision an environment for testing or use localhost. """
  
      # Default implementation for provision is a virtual machine
      how = 'virtual'
  
      def __init__(self, plan, data):
          """ Initialize provision step data """
          super().__init__(plan=plan, data=data)
          # Check that the names are unique
          names = [data.get('name') for data in self.data]
          names_count = collections.defaultdict(int)
          for name in names:
              names_count[name] += 1
          if len(self.data) > 1 and len(self.data) != len(names_count):
              duplicate = [k for k, v in names_count.items() if v > 1]
              duplicate_string = ', '.join(duplicate)
              raise tmt.utils.GeneralError(
                  f"Provision step names must be unique for multihost testing. "
                  f"Duplicate names: {duplicate_string} in plan '{plan.name}'.")
          # List of provisioned guests and loaded guest data
          self._guests: List['Guest'] = []
          self._guest_data: Dict[str, 'GuestData'] = {}
          self.is_multihost = False
  
      def load(self, extra_keys=None):
          """ Load guest data from the workdir """
          extra_keys = extra_keys or []
          super().load(extra_keys)
          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, data=None):
          """ Save guest data to the workdir """
          data = data or {}
          super().save(data)
          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):
          """ Wake up the step (process workdir and command line) """
          super().wake()
  
          # Choose the right plugin and wake it up
          for data in self.data:
              plugin = ProvisionPlugin.delegate(self, data)
              self._phases.append(plugin)
              # If guest data loaded, perform a complete wake up
              plugin.wake(data=self._guest_data.get(plugin.name))
              if plugin.guest():
                  self._guests.append(plugin.guest())
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Provision wake up complete (already done before).', level=2)
          # Save status and step data (now we know what to do)
          else:
              self.status('todo')
              self.save()
  
      def show(self):
          """ Show discover details """
          for data in self.data:
              ProvisionPlugin.delegate(self, data).show()
  
      def summary(self):
          """ 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 guest.name != tmt.utils.DEFAULT_NAME:
                  self.verbose(guest.name, color='red', shift=2)
  
      def go(self):
          """ 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():
                  try:
                      phase.go()
                      if isinstance(phase, ProvisionPlugin):
                          phase.guest().details()
                      if self.is_multihost:
                          self.info('')
                  finally:
                      if isinstance(phase, ProvisionPlugin):
                          if phase.guest():
                              self._guests.append(phase.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):
          """
          Packages required by all enabled provision plugins
  
          Return a list of packages which need to be installed on the
          provisioned guest so that the workdir can be synced to it.
          Used by the prepare step.
          """
          requires = set()
          for plugin in self.phases(classes=ProvisionPlugin):
              requires.update(plugin.requires())
          return list(requires)
  
  
  class ProvisionPlugin(tmt.steps.Plugin):
      """ Common parent of provision plugins """
  
      # Default implementation for provision is a virtual machine
      how = 'virtual'
  
      # List of all supported methods aggregated from all plugins
      _supported_methods = []
  
      # Common keys for all provision step implementations
      _common_keys = ['role']
  
      @classmethod
      def base_command(
              cls,
              usage: str,
              method_class: Optional[Type[click.Command]] = None) -> click.Command:
          """ Create base click command (common for all provision plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Provision.usage(method_overview=usage)
  
          # Create the command
          @click.command(cls=method_class, help=usage)
          @click.pass_context
          @click.option(
              '-h', '--how', metavar='METHOD',
              help='Use specified method for provisioning.')
          def provision(context, **kwargs):
              context.obj.steps.add('provision')
              Provision._save_context(context)
  
          return provision
  
      def wake(self, keys=None, data=None):
          """
          Wake up the plugin
  
          Override data with command line options.
          Wake up the guest based on provided guest data.
          """
          super().wake(keys=keys)
  
      def guest(self):
          """
          Return provisioned guest
  
          Each ProvisionPlugin has to implement this method.
          Should return a provisioned Guest() instance.
          """
          raise NotImplementedError()
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return Guest.requires()
  
      @classmethod
      def clean_images(cls, clean, dry):
          """ Remove the images of one particular plugin """
  
  
  @dataclasses.dataclass
  class GuestData(tmt.utils.SerializableContainer):
      """
      Keys necessary to describe, create, save and restore a guest.
  
      Very basic set of keys shared across all known guest classes.
      """
  
      # guest role in the multihost scenario
      role: Optional[str] = None
      # hostname or ip address
      guest: Optional[str] = None
  
  
  class Guest(tmt.utils.Common):
      """
      Guest provisioned for test execution
  
      A base class for guest-like classes. Provides some of the basic methods
      and functionality, but note some of the methods are left intentionally
      empty. These do not have valid implementation on this level, and it's up
      to Guest subclasses to provide one working in their respective
      infrastructure.
  
      The following keys are expected in the 'data' container::
  
          role ....... guest role in the multihost scenario
          guest ...... name, hostname or ip address
  
      These are by default imported into instance attributes.
      """
  
      # Used by save() to construct the correct container for keys.
      _data_class = GuestData
  
      role: Optional[str]
      guest: 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=None, parent=None):
          """ Initialize guest data """
          super().__init__(parent, name)
  
          self.load(data)
  
      def _random_name(self, prefix='', length=16):
          """ 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):
          """ Generate a name prefixed with tmt run id """
          _, run_id = os.path.split(self.parent.plan.my_run.workdir)
          return self._random_name(prefix="tmt-{0}-".format(run_id[-3:]))
  
      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):
          """
          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):
          """
          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):
          """ Show guest details such as distro and kernel """
          # Skip distro & kernel check in dry mode
          if self.opt('dry'):
              return
  
          # Distro (check os-release first)
          try:
              distro = self.execute('cat /etc/os-release')[0].strip()
              distro = re.search('PRETTY_NAME="(.*)"', distro).group(1)
          except (tmt.utils.RunError, AttributeError):
              # Check for lsb-release
              try:
                  distro = self.execute('cat /etc/lsb-release')[0].strip()
                  distro = re.search(
                      'DISTRIB_DESCRIPTION="(.*)"', distro).group(1)
              except (tmt.utils.RunError, AttributeError):
                  # Check for redhat-release
                  try:
                      distro = self.execute('cat /etc/redhat-release')[0].strip()
                  except (tmt.utils.RunError, AttributeError):
                      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 = self.execute('uname -r')[0].strip()
          self.verbose('kernel', kernel, 'green')
  
      def _ansible_verbosity(self):
          """ 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):
          """ Prepare extra arguments for ansible-playbook"""
          if extra_args is None:
              return []
          else:
              return shlex.split(str(extra_args))
  
      def _ansible_summary(self, output):
          """ 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):
          """ 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}'.")
          playbook = os.path.join(self.parent.plan.my_run.tree.root, playbook)
          self.debug(f"Playbook full path: '{playbook}'", level=2)
          return playbook
  
      def _prepare_environment(self, execute_environment=None):
          """ Prepare dict of environment variables """
          # Prepare environment variables so they can be correctly passed
          # to shell. Create a copy to prevent modifying source.
          environment = dict()
          environment.update(execute_environment or dict())
          # Plan environment and variables provided on the command line
          # override environment provided to execute().
          environment.update(self.parent.plan.environment)
          return environment
  
      @staticmethod
      def _export_environment(environment):
          """ Prepare shell export of environment variables """
          if not environment:
              return ""
          return f'export {" ".join(tmt.utils.shell_variables(environment))}; '
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare guest using ansible playbook """
  
          raise NotImplementedError()
  
      def execute(self, command, **kwargs):
          """
          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=None, destination=None, options=None):
          """
          Push files to the guest
          """
  
          raise NotImplementedError()
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """
          Pull files from the guest
          """
  
          raise NotImplementedError()
  
      def stop(self):
          """
          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=False):
          """
          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.
          """
  
          raise NotImplementedError()
  
      def reconnect(self, timeout=CONNECTION_TIMEOUT):
          """
          Ensure the connection to the guest is working after reboot
  
          The default timeout is 1h to allow long operations such as
          system upgrade. Custom number of seconds can be provided in the
          `timeout` parameter.
          """
          # Try to wait for machine to really shutdown
          time.sleep(RECONNECT_INITIAL_WAIT_TIME)
          self.debug("Wait for a connection to the guest.")
  
          # A small shortcut... `now` or `utcnow`, should not matter, becase we
          # need a difference between two values. As long as we use the same
          # function for both sides of the equation, we should be fine.
          now = datetime.datetime.utcnow
          deadline = now() + datetime.timedelta(seconds=timeout)
          while now() < deadline:
              try:
                  self.execute('whoami')
                  break
              except tmt.utils.RunError:
                  self.debug('Failed to connect to the guest, retrying.')
                  time.sleep(1)
          else:
              self.debug("Connection to guest failed after reboot.")
              return False
          return True
  
      def remove(self):
          """
          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):
          """
          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 "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)
  
      @classmethod
      def requires(cls):
          """ 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
  
  
  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 = GuestSshData
  
      port: Optional[int]
      user: Optional[str]
      key: List[str]
      password: Optional[str]
  
      # Master ssh connection process and socket path
      _ssh_master_process = None
      _ssh_socket_path = None
  
      def _ssh_guest(self):
          """ Return user@guest """
          return f'{self.user}@{self.guest}'
  
      def _ssh_socket(self):
          """ 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=False):
          """ Return common ssh options (list or joined) """
          options = [
              '-oForwardX11=no',
              '-oStrictHostKeyChecking=no',
              '-oUserKnownHostsFile=/dev/null',
              ]
          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:
              keys = self.key if isinstance(self.key, list) else [self.key]
              for key in keys:
                  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()}')
  
          return ' '.join(options) if join else options
  
      def _ssh_master_connection(self, command):
          """ Check/create the master ssh connection """
          if self._ssh_master_process:
              return
          command = command + self._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=False):
          """ 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:
              return " ".join(command) + " " + self._ssh_options(join=True)
          else:
              return command + self._ssh_options()
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare guest using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, stderr = self.run(
              ['ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              [f'--ssh-common-args={self._ssh_options(join=True)}'] +
              ['-i', f'{self._ssh_guest()},', playbook],
              cwd=self.parent.plan.worktree,
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def execute(self, command, **kwargs):
          """
          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.
          """
  
          # 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 '{directory}'; "
  
          # Run in interactive mode if requested
          interactive = ['-t'] if kwargs.get('interactive') else []
  
          # Prepare command and run it
          if isinstance(command, (list, tuple)):
              command = ' '.join(command)
          self.debug(f"Execute command '{command}' on guest '{self.guest}'.")
          command = (
              self._ssh_command() + interactive + [self._ssh_guest()] +
              [f'{environment}{directory}{command}'])
          return self.run(command, **kwargs)
  
      def push(self, source=None, destination=None, options=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'.
          """
          # Prepare options and the push command
          if options is None:
              options = DEFAULT_RSYNC_OPTIONS
          if destination is None:
              destination = "/"
          if source is None:
              source = self.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():
              """ Run the rsync command """
              self.run(
                  ["rsync"] + options
                  + ["-e", self._ssh_command(join=True)]
                  + [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() == "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=None,
              destination=None,
              options=None,
              extend_options=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 '-Rrz --links --safe-links --protect-args'
          and 'extend_options' to extend them (e.g. by exclude).
          """
          # Prepare options and the pull command
          if options is None:
              options = "-Rrz --links --safe-links --protect-args".split()
          if extend_options is not None:
              options.extend(extend_options)
          if destination is None:
              destination = "/"
          if source is None:
              source = self.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():
              """ Run the rsync command """
              self.run(
                  ["rsync"] + options
                  + ["-e", self._ssh_command(join=True)]
                  + [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() == "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):
          """
          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=False, command=None):
          """
          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'.
          """
  
          if hard:
              raise tmt.utils.ProvisionError(
                  "Method does not support hard reboot.")
  
          try:
              command = command or "reboot"
              self.debug(f"Reboot using the command '{command}'.")
              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
          return self.reconnect()
  
      def remove(self):
          """
          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):
          """
          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 "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)
  
      @classmethod
      def requires(cls):
          """ No extra requires needed """
          return []
  
files.new_path: tmt/steps/provision/__init__.py
files.old_contents: import collections
  import datetime
  import os
  import random
  import re
  import shlex
  import string
  import subprocess
  import tempfile
  import time
  from typing import Optional, Type
  
  import click
  import fmf
  
  import tmt
  import tmt.utils
  
  # Timeout in seconds of waiting for a connection after reboot
  # (long enough to allow operations such as system upgrade)
  # FIXME: Shorten once "tmt-reboot -t <timeout>" is implemented
  CONNECTION_TIMEOUT = 60 * 60
  
  # Wait time when reboot happens in seconds
  RECONNECT_INITIAL_WAIT_TIME = 5
  
  # Default rsync options
  DEFAULT_RSYNC_OPTIONS = [
      "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  
  class Provision(tmt.steps.Step):
      """ Provision an environment for testing or use localhost. """
  
      # Default implementation for provision is a virtual machine
      how = 'virtual'
  
      def __init__(self, plan, data):
          """ Initialize provision step data """
          super().__init__(plan=plan, data=data)
          # Check that the names are unique
          names = [data.get('name') for data in self.data]
          names_count = collections.defaultdict(int)
          for name in names:
              names_count[name] += 1
          if len(self.data) > 1 and len(self.data) != len(names_count):
              duplicate = [k for k, v in names_count.items() if v > 1]
              duplicate_string = ', '.join(duplicate)
              raise tmt.utils.GeneralError(
                  f"Provision step names must be unique for multihost testing. "
                  f"Duplicate names: {duplicate_string} in plan '{plan.name}'.")
          # List of provisioned guests and loaded guest data
          self._guests = []
          self._guest_data = {}
          self.is_multihost = False
  
      def load(self, extra_keys=None):
          """ Load guest data from the workdir """
          extra_keys = extra_keys or []
          super().load(extra_keys)
          try:
              self._guest_data = tmt.utils.yaml_to_dict(self.read('guests.yaml'))
          except tmt.utils.FileError:
              self.debug('Provisioned guests not found.', level=2)
  
      def save(self, data=None):
          """ Save guest data to the workdir """
          data = data or {}
          super().save(data)
          try:
              guests = dict(
                  [(guest.name, guest.save()) for guest in self.guests()])
              self.write('guests.yaml', tmt.utils.dict_to_yaml(guests))
          except tmt.utils.FileError:
              self.debug('Failed to save provisioned guests.')
  
      def wake(self):
          """ Wake up the step (process workdir and command line) """
          super().wake()
  
          # Choose the right plugin and wake it up
          for data in self.data:
              plugin = ProvisionPlugin.delegate(self, data)
              self._phases.append(plugin)
              # If guest data loaded, perform a complete wake up
              plugin.wake(data=self._guest_data.get(plugin.name))
              if plugin.guest():
                  self._guests.append(plugin.guest())
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Provision wake up complete (already done before).', level=2)
          # Save status and step data (now we know what to do)
          else:
              self.status('todo')
              self.save()
  
      def show(self):
          """ Show discover details """
          for data in self.data:
              ProvisionPlugin.delegate(self, data).show()
  
      def summary(self):
          """ 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 guest.name != tmt.utils.DEFAULT_NAME:
                  self.verbose(guest.name, color='red', shift=2)
  
      def go(self):
          """ 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():
                  try:
                      phase.go()
                      if isinstance(phase, ProvisionPlugin):
                          phase.guest().details()
                      if self.is_multihost:
                          self.info('')
                  finally:
                      if isinstance(phase, ProvisionPlugin):
                          if phase.guest():
                              self._guests.append(phase.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):
          """ Return the list of all provisioned guests """
          return self._guests
  
      def requires(self):
          """
          Packages required by all enabled provision plugins
  
          Return a list of packages which need to be installed on the
          provisioned guest so that the workdir can be synced to it.
          Used by the prepare step.
          """
          requires = set()
          for plugin in self.phases(classes=ProvisionPlugin):
              requires.update(plugin.requires())
          return list(requires)
  
  
  class ProvisionPlugin(tmt.steps.Plugin):
      """ Common parent of provision plugins """
  
      # Default implementation for provision is a virtual machine
      how = 'virtual'
  
      # List of all supported methods aggregated from all plugins
      _supported_methods = []
  
      # Common keys for all provision step implementations
      _common_keys = ['role']
  
      @classmethod
      def base_command(
              cls,
              usage: str,
              method_class: Optional[Type[click.Command]] = None) -> click.Command:
          """ Create base click command (common for all provision plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Provision.usage(method_overview=usage)
  
          # Create the command
          @click.command(cls=method_class, help=usage)
          @click.pass_context
          @click.option(
              '-h', '--how', metavar='METHOD',
              help='Use specified method for provisioning.')
          def provision(context, **kwargs):
              context.obj.steps.add('provision')
              Provision._save_context(context)
  
          return provision
  
      def wake(self, keys=None, data=None):
          """
          Wake up the plugin
  
          Override data with command line options.
          Wake up the guest based on provided guest data.
          """
          super().wake(keys=keys)
  
      def guest(self):
          """
          Return provisioned guest
  
          Each ProvisionPlugin has to implement this method.
          Should return a provisioned Guest() instance.
          """
          raise NotImplementedError()
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return Guest.requires()
  
      @classmethod
      def clean_images(cls, clean, dry):
          """ Remove the images of one particular plugin """
  
  
  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' dictionary::
  
          role ....... guest role in the multihost scenario
          guest ...... name, hostname or ip address
  
      These are by default imported into instance attributes (see the
      class attribute '_keys' below).
      """
  
      # List of supported keys
      # (used for import/export to/from attributes during load and save)
      _keys = ['role', 'guest']
  
      def __init__(self, data, name=None, parent=None):
          """ Initialize guest data """
          super().__init__(parent, name)
          # Initialize role, it will be overridden by load() if specified
          self.role = None
          # Flag to indicate localhost guest, requires special handling
          self.localhost = False
          self.load(data)
  
      def _random_name(self, prefix='', length=16):
          """ 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):
          """ Generate a name prefixed with tmt run id """
          _, run_id = os.path.split(self.parent.plan.my_run.workdir)
          return self._random_name(prefix="tmt-{0}-".format(run_id[-3:]))
  
      def load(self, data):
          """
          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.
          """
          for key in self._keys:
              setattr(self, key, data.get(key))
  
      def save(self):
          """
          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.
          """
          data = dict()
          for key in self._keys:
              value = getattr(self, key)
              if value is not None:
                  data[key] = value
          return data
  
      def wake(self):
          """
          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):
          """
          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):
          """ Show guest details such as distro and kernel """
          # Skip distro & kernel check in dry mode
          if self.opt('dry'):
              return
  
          # Distro (check os-release first)
          try:
              distro = self.execute('cat /etc/os-release')[0].strip()
              distro = re.search('PRETTY_NAME="(.*)"', distro).group(1)
          except (tmt.utils.RunError, AttributeError):
              # Check for lsb-release
              try:
                  distro = self.execute('cat /etc/lsb-release')[0].strip()
                  distro = re.search(
                      'DISTRIB_DESCRIPTION="(.*)"', distro).group(1)
              except (tmt.utils.RunError, AttributeError):
                  # Check for redhat-release
                  try:
                      distro = self.execute('cat /etc/redhat-release')[0].strip()
                  except (tmt.utils.RunError, AttributeError):
                      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 = self.execute('uname -r')[0].strip()
          self.verbose('kernel', kernel, 'green')
  
      def _ansible_verbosity(self):
          """ 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):
          """ Prepare extra arguments for ansible-playbook"""
          if extra_args is None:
              return []
          else:
              return shlex.split(str(extra_args))
  
      def _ansible_summary(self, output):
          """ 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):
          """ 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}'.")
          playbook = os.path.join(self.parent.plan.my_run.tree.root, playbook)
          self.debug(f"Playbook full path: '{playbook}'", level=2)
          return playbook
  
      def _prepare_environment(self, execute_environment=None):
          """ Prepare dict of environment variables """
          # Prepare environment variables so they can be correctly passed
          # to shell. Create a copy to prevent modifying source.
          environment = dict()
          environment.update(execute_environment or dict())
          # Plan environment and variables provided on the command line
          # override environment provided to execute().
          environment.update(self.parent.plan.environment)
          return environment
  
      @staticmethod
      def _export_environment(environment):
          """ Prepare shell export of environment variables """
          if not environment:
              return ""
          return f'export {" ".join(tmt.utils.shell_variables(environment))}; '
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare guest using ansible playbook """
  
          raise NotImplementedError()
  
      def execute(self, command, **kwargs):
          """
          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=None, destination=None, options=None):
          """
          Push files to the guest
          """
  
          raise NotImplementedError()
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """
          Pull files from the guest
          """
  
          raise NotImplementedError()
  
      def stop(self):
          """
          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=False):
          """
          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.
          """
  
          raise NotImplementedError()
  
      def reconnect(self, timeout=CONNECTION_TIMEOUT):
          """
          Ensure the connection to the guest is working after reboot
  
          The default timeout is 1h to allow long operations such as
          system upgrade. Custom number of seconds can be provided in the
          `timeout` parameter.
          """
          # Try to wait for machine to really shutdown
          time.sleep(RECONNECT_INITIAL_WAIT_TIME)
          self.debug("Wait for a connection to the guest.")
  
          # A small shortcut... `now` or `utcnow`, should not matter, becase we
          # need a difference between two values. As long as we use the same
          # function for both sides of the equation, we should be fine.
          now = datetime.datetime.utcnow
          deadline = now() + datetime.timedelta(seconds=timeout)
          while now() < deadline:
              try:
                  self.execute('whoami')
                  break
              except tmt.utils.RunError:
                  self.debug('Failed to connect to the guest, retrying.')
                  time.sleep(1)
          else:
              self.debug("Connection to guest failed after reboot.")
              return False
          return True
  
      def remove(self):
          """
          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):
          """
          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 "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)
  
      @classmethod
      def requires(cls):
          """ No extra requires needed """
          return []
  
  
  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 (see the
      class attribute '_keys' below).
      """
  
      # List of supported keys
      # (used for import/export to/from attributes during load and save)
      _keys = Guest._keys + ['port', 'user', 'key', 'password']
  
      # Master ssh connection process and socket path
      _ssh_master_process = None
      _ssh_socket_path = None
  
      def _ssh_guest(self):
          """ Return user@guest """
          return f'{self.user}@{self.guest}'
  
      def _ssh_socket(self):
          """ 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=False):
          """ Return common ssh options (list or joined) """
          options = [
              '-oForwardX11=no',
              '-oStrictHostKeyChecking=no',
              '-oUserKnownHostsFile=/dev/null',
              ]
          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:
              keys = self.key if isinstance(self.key, list) else [self.key]
              for key in keys:
                  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()}')
  
          return ' '.join(options) if join else options
  
      def _ssh_master_connection(self, command):
          """ Check/create the master ssh connection """
          if self._ssh_master_process:
              return
          command = command + self._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=False):
          """ 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:
              return " ".join(command) + " " + self._ssh_options(join=True)
          else:
              return command + self._ssh_options()
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare guest using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, stderr = self.run(
              ['ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              [f'--ssh-common-args={self._ssh_options(join=True)}'] +
              ['-i', f'{self._ssh_guest()},', playbook],
              cwd=self.parent.plan.worktree,
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def execute(self, command, **kwargs):
          """
          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.
          """
  
          # 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 '{directory}'; "
  
          # Run in interactive mode if requested
          interactive = ['-t'] if kwargs.get('interactive') else []
  
          # Prepare command and run it
          if isinstance(command, (list, tuple)):
              command = ' '.join(command)
          self.debug(f"Execute command '{command}' on guest '{self.guest}'.")
          command = (
              self._ssh_command() + interactive + [self._ssh_guest()] +
              [f'{environment}{directory}{command}'])
          return self.run(command, **kwargs)
  
      def push(self, source=None, destination=None, options=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'.
          """
          # Prepare options and the push command
          if options is None:
              options = DEFAULT_RSYNC_OPTIONS
          if destination is None:
              destination = "/"
          if source is None:
              source = self.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():
              """ Run the rsync command """
              self.run(
                  ["rsync"] + options
                  + ["-e", self._ssh_command(join=True)]
                  + [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() == "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=None,
              destination=None,
              options=None,
              extend_options=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 '-Rrz --links --safe-links --protect-args'
          and 'extend_options' to extend them (e.g. by exclude).
          """
          # Prepare options and the pull command
          if options is None:
              options = "-Rrz --links --safe-links --protect-args".split()
          if extend_options is not None:
              options.extend(extend_options)
          if destination is None:
              destination = "/"
          if source is None:
              source = self.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():
              """ Run the rsync command """
              self.run(
                  ["rsync"] + options
                  + ["-e", self._ssh_command(join=True)]
                  + [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() == "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):
          """
          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=False, command=None):
          """
          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'.
          """
  
          if hard:
              raise tmt.utils.ProvisionError(
                  "Method does not support hard reboot.")
  
          try:
              command = command or "reboot"
              self.debug(f"Reboot using the command '{command}'.")
              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
          return self.reconnect()
  
      def remove(self):
          """
          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):
          """
          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 "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)
  
      @classmethod
      def requires(cls):
          """ No extra requires needed """
          return []
  
files.old_path: tmt/steps/provision/__init__.py
files.new_contents: import dataclasses
  import datetime
  import sys
  import time
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import requests
  
  import tmt
  import tmt.steps.provision
  from tmt.utils import ProvisionError, retry_session, updatable_message
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  
  # List of Artemis API versions supported and understood by this plugin.
  # Since API gains support for new features over time, it is important to
  # know when particular feature became available, and avoid using it with
  # older APIs.
  SUPPORTED_API_VERSIONS = (
      # NEW: added hostname HW constraint
      '0.0.38',
      # NEW: virtualization HW constraint
      '0.0.37',
      # NEW: boot.method HW constraint
      '0.0.32',
      # NEW: network HW constraint
      '0.0.28'
      )
  
  
  # The default Artemis API version - the most recent supported versions
  # should be perfectly fine.
  DEFAULT_API_VERSION = SUPPORTED_API_VERSIONS[0]
  
  DEFAULT_API_URL = 'http://127.0.0.1:8001'
  DEFAULT_USER = 'root'
  DEFAULT_ARCH = 'x86_64'
  DEFAULT_PRIORITY_GROUP = 'default-priority'
  DEFAULT_KEYNAME = 'default'
  DEFAULT_PROVISION_TIMEOUT = 600
  DEFAULT_PROVISION_TICK = 60
  DEFAULT_API_TIMEOUT = 10
  DEFAULT_API_RETRIES = 10
  # Should lead to delays of 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256 seconds
  DEFAULT_RETRY_BACKOFF_FACTOR = 1
  
  # Type annotation for "data" package describing a guest instance. Passed
  # between load() and save() calls
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class ArtemisGuestData(tmt.steps.provision.GuestSshData):  # type: ignore[misc]
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      # API
      api_url: str = DEFAULT_API_URL
      api_version: str = DEFAULT_API_VERSION
  
      # Guest request properties
      arch: str = DEFAULT_ARCH
      image: Optional[str] = None
      hardware: Optional[Any] = None
      pool: Optional[str] = None
      priority_group: str = DEFAULT_PRIORITY_GROUP
      keyname: str = DEFAULT_KEYNAME
      user_data: Dict[str, str] = dataclasses.field(default_factory=dict)
  
      # Provided by Artemis response
      guestname: Optional[str] = None
  
      # Timeouts and deadlines
      provision_timeout: int = DEFAULT_PROVISION_TIMEOUT
      provision_tick: int = DEFAULT_PROVISION_TICK
      api_timeout: int = DEFAULT_API_TIMEOUT
      api_retries: int = DEFAULT_API_RETRIES
      api_retry_backoff_factor: int = DEFAULT_RETRY_BACKOFF_FACTOR
  
  
  GUEST_STATE_COLOR_DEFAULT = 'green'
  
  GUEST_STATE_COLORS = {
      'routing': 'yellow',
      'provisioning': 'magenta',
      'promised': 'blue',
      'preparing': 'cyan',
      'cancelled': 'red',
      'error': 'red'
  }
  
  
  # Type annotation for Artemis API `GET /guests/$guestname` response.
  # Partial, not all fields necessary since plugin ignores most of them.
  GuestInspectType = TypedDict(
      'GuestInspectType', {
          'state': str,
          'address': Optional[str]
          }
      )
  
  
  class ArtemisAPI:
      def __init__(self, guest: 'GuestArtemis') -> None:
          self._guest = guest
  
          self.http_session = retry_session.create(
              retries=guest.api_retries,
              backoff_factor=guest.api_retry_backoff_factor,
              allowed_methods=('HEAD', 'GET', 'POST', 'DELETE', 'PUT'),
              status_forcelist=(
                  429,  # Too Many Requests
                  500,  # Internal Server Error
                  502,  # Bad Gateway
                  503,  # Service Unavailable
                  504   # Gateway Timeout
                  ),
              timeout=guest.api_timeout
              )
  
      def query(
              self,
              path: str,
              method: str = 'get',
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Base helper for Artemis API queries.
  
          Trivial dispatcher per method, returning retrieved response.
  
          :param path: API path to contact.
          :param method: HTTP method to use.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          url = f'{self._guest.api_url}{path}'
  
          if method == 'get':
              return self.http_session.get(url, **request_kwargs)
  
          if method == 'post':
              return self.http_session.post(url, **request_kwargs)
  
          if method == 'delete':
              return self.http_session.delete(url, **request_kwargs)
  
          if method == 'put':
              return self.http_session.put(url, **request_kwargs)
  
          raise tmt.utils.GeneralError(
              f'Unsupported Artemis API method.\n{method}')
  
      def create(
              self,
              path: str,
              data: Dict[str, Any],
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Create - or request creation of - a resource.
  
          :param path: API path to contact.
          :param data: optional key/value data to send with the request.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
          request_kwargs['json'] = data
  
          return self.query(path, method='post', request_kwargs=request_kwargs)
  
      def inspect(
              self,
              path: str,
              params: Optional[Dict[str, Any]] = None,
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Inspect a resource.
  
          :param path: API path to contact.
          :param params: optional key/value query parameters.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          if params:
              request_kwargs['params'] = params
  
          return self.query(path, request_kwargs=request_kwargs)
  
      def delete(
              self,
              path: str,
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Delete - or request removal of - a resource.
  
          :param path: API path to contact.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          return self.query(path, method='delete', request_kwargs=request_kwargs)
  
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  class ProvisionArtemis(
          tmt.steps.provision.ProvisionPlugin):  # type: ignore[misc]
      """
      Provision guest using Artemis backend
  
      Minimal configuration could look like this:
  
          provision:
              how: artemis
              image: Fedora
              api-url: https://your-artemis.com/
  
      Note that the actual value of "image" depends on what images - or
      "composes" as Artemis calls them - supports and can deliver.
  
      Note that "api-url" can be also given via ARTEMIS_API_URL
      environment variable.
  
      Full configuration example:
  
          provision:
              how: artemis
  
              # Artemis API
              api-url: https://your-artemis.com/
              api-version: 0.0.32
  
              # Mandatory environment properties
              image: Fedora
  
              # Optional environment properties
              arch: aarch64
              pool: optional-pool-name
  
              # Provisioning process control (optional)
              priority-group: custom-priority-group
              keyname: custom-SSH-key-name
  
              # Labels to be attached to guest request (optional)
              user-data:
                  foo: bar
  
              # Timeouts and deadlines (optional)
              provision-timeout: 3600
              provision-tick: 10
              api-timeout: 600
              api-retries: 5
              api-retry-backoff-factor: 1
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [
          tmt.steps.Method(name='artemis', doc=__doc__, order=50),
          ]
  
      _keys = [
          'api-url',
          'api-version',
          'arch',
          'image',
          'hardware',
          'pool',
          'priority-group',
          'keyname',
          'user-data',
          'provision-timeout',
          'provision-tick',
          'api-timeout',
          'api-retries',
          'api-retry-backoff-factor']
  
      # TODO: fix types once superclass gains its annotations
      @classmethod
      def options(cls, how: Any = None) -> List[click.Option]:
          """ Prepare command line options for Artemis """
          return cast(List[click.Option], [
              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.',
                  ),
              ]) + cast(List[click.Option], super().options(how))
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
  
          return getattr(ArtemisGuestData(), option.replace('-', '_'), default)
  
      # TODO: use better types once superclass gains its annotations
      def wake(self, keys: Optional[List[str]] = None,
               data: Optional[ArtemisGuestData] = None) -> None:
          """ Wake up the plugin, process data, apply options """
  
          super().wake(keys=keys, data=data)
  
          if data:
              self._guest = GuestArtemis(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')
              )
  
          self._guest = GuestArtemis(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self) -> Optional['GuestArtemis']:
          """ Return the provisioned guest """
          return self._guest
  
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  class GuestArtemis(tmt.GuestSsh):  # type: ignore[misc]
      """
      Artemis guest instance
  
      The following keys are expected in the 'data' dictionary:
      """
  
      # 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
  
      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')
  
          deadline = datetime.datetime.utcnow(
              ) + datetime.timedelta(seconds=self.provision_timeout)
  
          with updatable_message(
                  'state', indent_level=self._level()) as progress_message:
              while deadline > datetime.datetime.utcnow():
                  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':
                      break
  
                  time.sleep(self.provision_tick)
  
              else:
                  raise ProvisionError(
                      f'Failed to provision in the given amount '
                      f'of time (--provision-timeout={self.provision_timeout}).')
  
          self.guest = current['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}'.")
  
files.new_path: tmt/steps/provision/artemis.py
files.old_contents: import datetime
  import sys
  import time
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import requests
  
  import tmt
  import tmt.steps.provision
  from tmt.utils import ProvisionError, retry_session, updatable_message
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  
  # List of Artemis API versions supported and understood by this plugin.
  # Since API gains support for new features over time, it is important to
  # know when particular feature became available, and avoid using it with
  # older APIs.
  SUPPORTED_API_VERSIONS = (
      # NEW: added hostname HW constraint
      '0.0.38',
      # NEW: virtualization HW constraint
      '0.0.37',
      # NEW: boot.method HW constraint
      '0.0.32',
      # NEW: network HW constraint
      '0.0.28'
      )
  
  
  # The default Artemis API version - the most recent supported versions
  # should be perfectly fine.
  DEFAULT_API_VERSION = SUPPORTED_API_VERSIONS[0]
  
  # Type annotation for "data" package describing a guest instance. Passed
  # between load() and save() calls.
  StepStateType = TypedDict(
      'StepStateType',
      {
          # API
          'api-url': str,
          'api-version': str,
  
          # Guest request properties
          'arch': str,
          'image': str,
          'hardware': Any,
          'pool': Optional[str],
          'priority-group': str,
          'keyname': str,
          'user-data': Dict[str, str],
  
          # Provided by Artemis response
          'guestname': Optional[str],
          'guest': Optional[str],
          'user': str,
  
          # Timeouts and deadlines
          'provision-timeout': int,
          'provision-tick': int,
          'api-timeout': int,
          'api-retries': int,
          'api-retry-backoff-factor': int
          }
      )
  
  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
  
  DEFAULT_GUEST_DATA = cast(
      StepStateType, {
          'api-version': DEFAULT_API_VERSION,
          'arch': DEFAULT_ARCH,
          'priority-group': DEFAULT_PRIORITY_GROUP,
          'keyname': DEFAULT_KEYNAME,
          'user-data': {},
          'user': DEFAULT_USER,
          'provision-timeout': DEFAULT_PROVISION_TIMEOUT,
          'provision-tick': DEFAULT_PROVISION_TICK,
          'api-timeout': DEFAULT_API_TIMEOUT,
          'api-retries': DEFAULT_API_RETRIES,
          'api-retry-backoff-factor': DEFAULT_RETRY_BACKOFF_FACTOR
          }
      )
  
  GUEST_STATE_COLOR_DEFAULT = 'green'
  
  GUEST_STATE_COLORS = {
      'routing': 'yellow',
      'provisioning': 'magenta',
      'promised': 'blue',
      'preparing': 'cyan',
      'cancelled': 'red',
      'error': 'red'
  }
  
  
  # Type annotation for Artemis API `GET /guests/$guestname` response.
  # Partial, not all fields necessary since plugin ignores most of them.
  GuestInspectType = TypedDict(
      'GuestInspectType', {
          'state': str,
          'address': Optional[str]
          }
      )
  
  
  class ArtemisAPI:
      def __init__(self, guest: 'GuestArtemis') -> None:
          self._guest = guest
  
          self.http_session = retry_session.create(
              retries=guest.api_retries,
              backoff_factor=guest.api_retry_backoff_factor,
              allowed_methods=('HEAD', 'GET', 'POST', 'DELETE', 'PUT'),
              status_forcelist=(
                  429,  # Too Many Requests
                  500,  # Internal Server Error
                  502,  # Bad Gateway
                  503,  # Service Unavailable
                  504   # Gateway Timeout
                  ),
              timeout=guest.api_timeout
              )
  
      def query(
              self,
              path: str,
              method: str = 'get',
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Base helper for Artemis API queries.
  
          Trivial dispatcher per method, returning retrieved response.
  
          :param path: API path to contact.
          :param method: HTTP method to use.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          url = f'{self._guest.api_url}{path}'
  
          if method == 'get':
              return self.http_session.get(url, **request_kwargs)
  
          if method == 'post':
              return self.http_session.post(url, **request_kwargs)
  
          if method == 'delete':
              return self.http_session.delete(url, **request_kwargs)
  
          if method == 'put':
              return self.http_session.put(url, **request_kwargs)
  
          raise tmt.utils.GeneralError(
              f'Unsupported Artemis API method.\n{method}')
  
      def create(
              self,
              path: str,
              data: Dict[str, Any],
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Create - or request creation of - a resource.
  
          :param path: API path to contact.
          :param data: optional key/value data to send with the request.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
          request_kwargs['json'] = data
  
          return self.query(path, method='post', request_kwargs=request_kwargs)
  
      def inspect(
              self,
              path: str,
              params: Optional[Dict[str, Any]] = None,
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Inspect a resource.
  
          :param path: API path to contact.
          :param params: optional key/value query parameters.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          if params:
              request_kwargs['params'] = params
  
          return self.query(path, request_kwargs=request_kwargs)
  
      def delete(
              self,
              path: str,
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Delete - or request removal of - a resource.
  
          :param path: API path to contact.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          return self.query(path, method='delete', request_kwargs=request_kwargs)
  
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  class ProvisionArtemis(
          tmt.steps.provision.ProvisionPlugin):  # type: ignore[misc]
      """
      Provision guest using Artemis backend
  
      Minimal configuration could look like this:
  
          provision:
              how: artemis
              image: Fedora
              api-url: https://your-artemis.com/
  
      Note that the actual value of "image" depends on what images - or
      "composes" as Artemis calls them - supports and can deliver.
  
      Note that "api-url" can be also given via ARTEMIS_API_URL
      environment variable.
  
      Full configuration example:
  
          provision:
              how: artemis
  
              # Artemis API
              api-url: https://your-artemis.com/
              api-version: 0.0.32
  
              # Mandatory environment properties
              image: Fedora
  
              # Optional environment properties
              arch: aarch64
              pool: optional-pool-name
  
              # Provisioning process control (optional)
              priority-group: custom-priority-group
              keyname: custom-SSH-key-name
  
              # Labels to be attached to guest request (optional)
              user-data:
                  foo: bar
  
              # Timeouts and deadlines (optional)
              provision-timeout: 3600
              provision-tick: 10
              api-timeout: 600
              api-retries: 5
              api-retry-backoff-factor: 1
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [
          tmt.steps.Method(name='artemis', doc=__doc__, order=50),
          ]
  
      _keys = [
          'api-url',
          'api-version',
          'arch',
          'image',
          'hardware',
          'pool',
          'priority-group',
          'keyname',
          'user-data',
          'provision-timeout',
          'provision-tick',
          'api-timeout',
          'api-retries',
          'api-retry-backoff-factor']
  
      # TODO: fix types once superclass gains its annotations
      @classmethod
      def options(cls, how: Any = None) -> List[click.Option]:
          """ Prepare command line options for Artemis """
          return cast(List[click.Option], [
              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.',
                  ),
              ]) + cast(List[click.Option], super().options(how))
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
  
          return DEFAULT_GUEST_DATA.get(option, default)
  
      # TODO: use better types once superclass gains its annotations
      def wake(self, keys: Optional[List[str]] = None,
               data: Optional[StepStateType] = None) -> None:
          """ Wake up the plugin, process data, apply options """
  
          super().wake(keys=keys, data=data)
  
          if data:
              self._guest = GuestArtemis(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: StepStateType = {
              '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,
              'guestname': None,
              'guest': None,
              'user': DEFAULT_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')
              }
  
          self._guest = GuestArtemis(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self) -> Optional['GuestArtemis']:
          """ Return the provisioned guest """
          return self._guest
  
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  class GuestArtemis(tmt.GuestSsh):  # type: ignore[misc]
      """
      Artemis guest instance
  
      The following keys are expected in the 'data' dictionary:
      """
  
      _api: Optional[ArtemisAPI] = None
  
      @property
      def api(self) -> ArtemisAPI:
          if self._api is None:
              self._api = ArtemisAPI(self)
  
          return self._api
  
      def load(self, data: StepStateType) -> None:
          super().load(data)
  
          self.api_url = data['api-url']
          self.api_version = data['api-version']
          self.arch = data['arch']
          self.image = data['image']
          self.hardware = data['hardware']
          self.pool = data['pool']
          self.priority_group = data['priority-group']
          self.keyname = data['keyname']
          self.user_data = data['user-data']
          self.guestname = data['guestname']
          self.guest = data['guest']
          self.user = data['user']
          self.provision_timeout = data['provision-timeout']
          self.provision_tick = data['provision-tick']
          self.api_timeout = data['api-timeout']
          self.api_retries = data['api-retries']
          self.api_retry_backoff_factor = data['api-retry-backoff-factor']
  
      def save(self) -> StepStateType:
          data = cast(StepStateType, super().save())
  
          data['api-url'] = self.api_url
          data['api-version'] = self.api_version
          data['arch'] = self.arch
          data['image'] = self.image
          data['hardware'] = self.hardware
          data['pool'] = self.pool
          data['priority-group'] = self.priority_group
          data['keyname'] = self.keyname
          data['user-data'] = self.user_data
          data['guestname'] = self.guestname
          data['guest'] = self.guest
          data['user'] = self.user
          data['provision-timeout'] = self.provision_timeout
          data['provision-tick'] = self.provision_tick
          data['api-timeout'] = self.api_timeout
          data['api-retries'] = self.api_retries
          data['api-retry-backoff-factor'] = self.api_retry_backoff_factor
  
          return data
  
      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')
  
          deadline = datetime.datetime.utcnow(
              ) + datetime.timedelta(seconds=self.provision_timeout)
  
          with updatable_message(
                  'state', indent_level=self._level()) as progress_message:
              while deadline > datetime.datetime.utcnow():
                  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':
                      break
  
                  time.sleep(self.provision_tick)
  
              else:
                  raise ProvisionError(
                      f'Failed to provision in the given amount '
                      f'of time (--provision-timeout={self.provision_timeout}).')
  
          self.guest = current['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}'.")
  
files.old_path: tmt/steps/provision/artemis.py
files.new_contents: import click
  
  import tmt
  import tmt.steps.provision
  
  
  class ProvisionConnect(tmt.steps.provision.ProvisionPlugin):
      """
      Connect to a provisioned guest using ssh
  
      Private key authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              key: /home/psss/.ssh/example_rsa
  
      Password authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              password: secret
  
      User defaults to 'root', so if you have private key correctly set
      the minimal configuration can look like this:
  
          provision:
              how: connect
              guest: host.example.org
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [tmt.steps.Method(name='connect', doc=__doc__, order=50)]
  
      # Supported keys
      _keys = ["guest", "key", "user", "password", "port"]
  
      @classmethod
      def options(cls, how=None):
          """ 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 default(self, option, default=None):
          """ Return default data for given option """
          # User root as the default user
          if option == 'user':
              return 'root'
          # No other defaults available
          return default
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          if data:
              self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def go(self):
          """ 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 = tmt.steps.provision.GuestSshData(
              role=self.get('role'),
              guest=guest,
              user=user
              )
          self.info('guest', guest, 'green')
          self.info('user', user, 'green')
          if port:
              self.info('port', port, 'green')
              data.port = port
  
          # Use provided password for authentication
          if password:
              self.info('password', password, 'green')
              self.debug('Using password authentication.')
              data.password = password
          # Default to using a private key (user can have configured one)
          else:
              self.info('key', key or 'not provided', 'green')
              self.debug('Using private key authentication.')
              data.key = key
  
          # And finally create the guest
          self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
files.new_path: tmt/steps/provision/connect.py
files.old_contents: import click
  
  import tmt
  import tmt.steps.provision
  
  
  class ProvisionConnect(tmt.steps.provision.ProvisionPlugin):
      """
      Connect to a provisioned guest using ssh
  
      Private key authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              key: /home/psss/.ssh/example_rsa
  
      Password authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              password: secret
  
      User defaults to 'root', so if you have private key correctly set
      the minimal configuration can look like this:
  
          provision:
              how: connect
              guest: host.example.org
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [tmt.steps.Method(name='connect', doc=__doc__, order=50)]
  
      # Supported keys
      _keys = ["guest", "key", "user", "password", "port"]
  
      @classmethod
      def options(cls, how=None):
          """ 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 default(self, option, default=None):
          """ Return default data for given option """
          # User root as the default user
          if option == 'user':
              return 'root'
          # No other defaults available
          return default
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          if data:
              self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def go(self):
          """ 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 = dict(guest=guest, user=user, role=self.get('role'))
          self.info('guest', guest, 'green')
          self.info('user', user, 'green')
          if port:
              self.info('port', port, 'green')
              data['port'] = port
  
          # Use provided password for authentication
          if password:
              self.info('password', password, 'green')
              self.debug('Using password authentication.')
              data['password'] = password
          # Default to using a private key (user can have configured one)
          else:
              self.info('key', key or 'not provided', 'green')
              self.debug('Using private key authentication.')
              data['key'] = key
  
          # And finally create the guest
          self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
files.old_path: tmt/steps/provision/connect.py
files.new_contents: import tmt
  import tmt.steps.provision
  
  
  class ProvisionLocal(tmt.steps.provision.ProvisionPlugin):
      """
      Use local host for test execution
  
      In general it is not recommended to run tests on your local machine
      as there might be security risks. Run only those tests which you
      know are safe so that you don't destroy your laptop ;-)
  
      Example config:
  
          provision:
              how: local
  
      Note that 'tmt run' is expected to be executed under a regular user.
      If there are admin rights required (for example in the prepare step)
      you might be asked for a sudo password.
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [tmt.steps.Method(name='local', doc=__doc__, order=50)]
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          if data:
              self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Create a GuestLocal instance
          data = tmt.steps.provision.GuestSshData(
              guest='localhost',
              role=self.get('role')
              )
          self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestLocal.requires()
  
  
  class GuestLocal(tmt.Guest):
      """ Local Host """
  
      localhost = True
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, stderr = self.run(
              ['sudo', '-E', 'ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'local', '-i', 'localhost,', playbook],
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def execute(self, command, **kwargs):
          """ Execute command on localhost """
          # Prepare the environment (plan/cli variables override)
          environment = dict()
          environment.update(kwargs.pop('env', dict()))
          environment.update(self.parent.plan.environment)
          # Run the command under the prepared environment
          return self.run(command, env=environment, shell=True, **kwargs)
  
      def stop(self):
          """ Stop the guest """
  
          self.debug(f"Doing nothing to stop guest '{self.guest}'.")
  
      def reboot(self, hard=False):
          """ Reboot the guest, return True if successful """
  
          self.debug(f"Doing nothing to reboot guest '{self.guest}'.")
  
          return False
  
      def push(self, source=None, destination=None, options=None):
          """ Nothing to be done to push workdir """
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """ Nothing to be done to pull workdir """
  
files.new_path: tmt/steps/provision/local.py
files.old_contents: import tmt
  import tmt.steps.provision
  
  
  class ProvisionLocal(tmt.steps.provision.ProvisionPlugin):
      """
      Use local host for test execution
  
      In general it is not recommended to run tests on your local machine
      as there might be security risks. Run only those tests which you
      know are safe so that you don't destroy your laptop ;-)
  
      Example config:
  
          provision:
              how: local
  
      Note that 'tmt run' is expected to be executed under a regular user.
      If there are admin rights required (for example in the prepare step)
      you might be asked for a sudo password.
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [tmt.steps.Method(name='local', doc=__doc__, order=50)]
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          if data:
              self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Create a GuestLocal instance
          data = {'guest': 'localhost', 'role': self.get('role')}
          self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestLocal.requires()
  
  
  class GuestLocal(tmt.Guest):
      """ Local Host """
  
      def __init__(self, data, name=None, parent=None):
          """ Initialize guest data """
          super().__init__(data, name, parent)
          self.localhost = True
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, stderr = self.run(
              ['sudo', '-E', 'ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'local', '-i', 'localhost,', playbook],
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def execute(self, command, **kwargs):
          """ Execute command on localhost """
          # Prepare the environment (plan/cli variables override)
          environment = dict()
          environment.update(kwargs.pop('env', dict()))
          environment.update(self.parent.plan.environment)
          # Run the command under the prepared environment
          return self.run(command, env=environment, shell=True, **kwargs)
  
      def stop(self):
          """ Stop the guest """
  
          self.debug(f"Doing nothing to stop guest '{self.guest}'.")
  
      def reboot(self, hard=False):
          """ Reboot the guest, return True if successful """
  
          self.debug(f"Doing nothing to reboot guest '{self.guest}'.")
  
          return False
  
      def push(self, source=None, destination=None, options=None):
          """ Nothing to be done to push workdir """
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """ Nothing to be done to pull workdir """
  
files.old_path: tmt/steps/provision/local.py
files.new_contents: import dataclasses
  import os
  from typing import Optional
  
  import click
  
  import tmt
  import tmt.steps.provision
  
  # Timeout in seconds of waiting for a connection
  CONNECTION_TIMEOUT = 60
  
  # Defaults
  DEFAULT_IMAGE = "fedora"
  DEFAULT_USER = "root"
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class PodmanGuestData(tmt.steps.provision.GuestData):  # type: ignore[misc]
      image: str = DEFAULT_IMAGE
      user: str = DEFAULT_USER
      force_pull: bool = False
  
      container: Optional[str] = None
  
  
  class ProvisionPodman(tmt.steps.provision.ProvisionPlugin):
      """
      Create a new container using podman
  
      Example config:
  
          provision:
              how: container
              image: fedora:latest
  
      In order to always pull the fresh container image use 'pull: true'.
  
      In order to run the container with different user as the default 'root',
      use 'user: USER'.
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["image", "container", "pull", "user"]
  
      # Supported methods
      _methods = [tmt.steps.Method(name='container', doc=__doc__, order=50)]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for connect """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to use. Short name or complete url.'),
              click.option(
                  '-c', '--container', metavar='NAME',
                  help='Name or id of an existing container to be used.'),
              click.option(
                  '-p', '--pull', is_flag=True,
                  help='Force pulling a fresh container image.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='User to use for all container operations.')
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          if option == 'pull':
              return PodmanGuestData().force_pull
  
          return getattr(PodmanGuestData(), option.replace('-', '_'), default)
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          # Wake up podman instance
          if data:
              guest = GuestContainer(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Show which image we are using
          pull = ' (force pull)' if self.get('pull') else ''
          self.info('image', f"{self.get('image')}{pull}", 'green')
  
          # Prepare data for the guest instance
          data_from_options = {
              key: self.get(key)
              for key in PodmanGuestData.keys()
              if key != 'force_pull'
              }
  
          data_from_options['force_pull'] = self.get('pull')
  
          data = PodmanGuestData(**data_from_options)
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestContainer(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestContainer.requires()
  
  
  class GuestContainer(tmt.Guest):
      """ Container Instance """
  
      _data_class = PodmanGuestData
  
      image: Optional[str]
      container: Optional[str]
      user: str
      force_pull: bool
  
      def wake(self):
          """ Wake up the guest """
          self.debug(
              f"Waking up container '{self.container}'.", level=2, shift=0)
  
      def start(self):
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Check if the image is available
          image_id = self.podman(
              ['images', '-q', self.image],
              message=f"Check for container image '{self.image}'.")[0].strip()
  
          # Pull image if not available or pull forced
          if not image_id or self.force_pull:
              self.podman(
                  ['pull', '-q', self.image],
                  message=f"Pull image '{self.image}'.")
  
          # Mount the whole plan directory in the container
          workdir = self.parent.plan.workdir
  
          self.container = self.guest = self._tmt_name()
          self.verbose('name', self.container, 'green')
  
          # FIXME: Workaround for BZ#1900021 (f34 container on centos-8)
          workaround = ['--security-opt', 'seccomp=unconfined']
  
          # Run the container
          self.debug(f"Start container '{self.image}'.")
          self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f'{workdir}:{workdir}:z',
               '-itd', '--user', self.user, self.image])
  
      def reboot(self, hard=False, command=None):
          """ Restart the container, return True if successful  """
          if command:
              raise tmt.utils.ProvisionError(
                  "Custom reboot command not supported in podman provision.")
          if not hard:
              raise tmt.utils.ProvisionError(
                  "Containers do not support soft reboot, they can only be "
                  "stopped and started again (hard reboot).")
          self.podman(['container', 'restart', self.container])
          return self.reconnect(timeout=CONNECTION_TIMEOUT)
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare container using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          # As non-root we must run with podman unshare
          podman_unshare = ['podman', 'unshare'] if os.geteuid() != 0 else []
          stdout, stderr = self.run(
              podman_unshare + ['ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'podman', '-i', f'{self.container},', playbook],
              cwd=self.parent.plan.worktree,
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def podman(self, command, **kwargs):
          """ Run given command via podman """
          return self.run(['podman'] + command, **kwargs)
  
      def execute(self, command, **kwargs):
          """ Execute given commands in podman via shell """
          if not self.container and not self.opt('dry'):
              raise tmt.utils.ProvisionError(
                  'Could not execute without provisioned container.')
  
          # Change to given directory on guest if cwd provided
          directory = kwargs.get('cwd', '')
          if directory:
              directory = f"cd '{directory}'; "
  
          # Prepare the environment variables export
          environment = self._export_environment(
              self._prepare_environment(kwargs.get('env', dict())))
  
          # Run in interactive mode if requested
          interactive = ['-it'] if kwargs.get('interactive') else []
  
          # Note that we MUST run commands via bash, so variables
          # work as expected
          if isinstance(command, list):
              command = ' '.join(command)
          command = directory + environment + command
          return self.podman(
              ['exec'] + interactive +
              [self.container or 'dry', 'bash', '-c', command], **kwargs)
  
      def push(self, source=None, destination=None, options=None):
          """ Make sure that the workdir has a correct selinux context """
          self.debug("Update selinux context of the run workdir.", level=3)
          self.run(
              ["chcon", "--recursive", "--type=container_file_t",
               self.parent.plan.workdir], shell=False)
          # In case explicit destination is given, use `podman cp` to copy data
          # to the container
          if destination:
              self.podman(["cp", source, f"{self.container}:{destination}"])
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """ Nothing to be done to pull workdir """
  
      def stop(self):
          """ Stop provisioned guest """
          if self.container:
              self.podman(['container', 'stop', self.container])
              self.info('container', 'stopped', 'green')
  
      def remove(self):
          """ Remove the container """
          if self.container:
              self.podman(['container', 'rm', '-f', self.container])
              self.info('container', 'removed', 'green')
  
files.new_path: tmt/steps/provision/podman.py
files.old_contents: import os
  
  import click
  
  import tmt
  import tmt.steps.provision
  
  # Timeout in seconds of waiting for a connection
  CONNECTION_TIMEOUT = 60
  
  # Defaults
  DEFAULT_IMAGE = "fedora"
  DEFAULT_USER = "root"
  
  
  class ProvisionPodman(tmt.steps.provision.ProvisionPlugin):
      """
      Create a new container using podman
  
      Example config:
  
          provision:
              how: container
              image: fedora:latest
  
      In order to always pull the fresh container image use 'pull: true'.
  
      In order to run the container with different user as the default 'root',
      use 'user: USER'.
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["image", "container", "pull", "user"]
  
      # Supported methods
      _methods = [tmt.steps.Method(name='container', doc=__doc__, order=50)]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for connect """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to use. Short name or complete url.'),
              click.option(
                  '-c', '--container', metavar='NAME',
                  help='Name or id of an existing container to be used.'),
              click.option(
                  '-p', '--pull', is_flag=True,
                  help='Force pulling a fresh container image.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='User to use for all container operations.')
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          # Use 'fedora' as a default image
          if option == 'image':
              return DEFAULT_IMAGE
          # Use 'root' as a default user
          if option == 'user':
              return DEFAULT_USER
          # No other defaults available
          return default
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          # Wake up podman instance
          if data:
              guest = GuestContainer(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Show which image we are using
          pull = ' (force pull)' if self.get('pull') else ''
          self.info('image', f"{self.get('image')}{pull}", 'green')
  
          # Prepare data for the guest instance
          data = dict()
          for key in self._keys + self._common_keys:
              data[key] = self.get(key)
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestContainer(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestContainer.requires()
  
  
  class GuestContainer(tmt.Guest):
      """ Container Instance """
  
      def load(self, data):
          """ Load guest data and initialize attributes """
          super().load(data)
  
          # Load basic data
          self.image = data.get('image')
          self.force_pull = data.get('pull')
          self.container = data.get('container')
          self.user = data.get('user')
  
      def save(self):
          """ Save guest data for future wake up """
          data = super().save()
          data['container'] = self.container
          data['image'] = self.image
          data['user'] = self.user
          return data
  
      def wake(self):
          """ Wake up the guest """
          self.debug(
              f"Waking up container '{self.container}'.", level=2, shift=0)
  
      def start(self):
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Check if the image is available
          image_id = self.podman(
              ['images', '-q', self.image],
              message=f"Check for container image '{self.image}'.")[0].strip()
  
          # Pull image if not available or pull forced
          if not image_id or self.force_pull:
              self.podman(
                  ['pull', '-q', self.image],
                  message=f"Pull image '{self.image}'.")
  
          # Mount the whole plan directory in the container
          workdir = self.parent.plan.workdir
  
          self.container = self.guest = self._tmt_name()
          self.verbose('name', self.container, 'green')
  
          # FIXME: Workaround for BZ#1900021 (f34 container on centos-8)
          workaround = ['--security-opt', 'seccomp=unconfined']
  
          # Run the container
          self.debug(f"Start container '{self.image}'.")
          self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f'{workdir}:{workdir}:z',
               '-itd', '--user', self.user, self.image])
  
      def reboot(self, hard=False, command=None):
          """ Restart the container, return True if successful  """
          if command:
              raise tmt.utils.ProvisionError(
                  "Custom reboot command not supported in podman provision.")
          if not hard:
              raise tmt.utils.ProvisionError(
                  "Containers do not support soft reboot, they can only be "
                  "stopped and started again (hard reboot).")
          self.podman(['container', 'restart', self.container])
          return self.reconnect(timeout=CONNECTION_TIMEOUT)
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare container using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          # As non-root we must run with podman unshare
          podman_unshare = ['podman', 'unshare'] if os.geteuid() != 0 else []
          stdout, stderr = self.run(
              podman_unshare + ['ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'podman', '-i', f'{self.container},', playbook],
              cwd=self.parent.plan.worktree,
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def podman(self, command, **kwargs):
          """ Run given command via podman """
          return self.run(['podman'] + command, **kwargs)
  
      def execute(self, command, **kwargs):
          """ Execute given commands in podman via shell """
          if not self.container and not self.opt('dry'):
              raise tmt.utils.ProvisionError(
                  'Could not execute without provisioned container.')
  
          # Change to given directory on guest if cwd provided
          directory = kwargs.get('cwd', '')
          if directory:
              directory = f"cd '{directory}'; "
  
          # Prepare the environment variables export
          environment = self._export_environment(
              self._prepare_environment(kwargs.get('env', dict())))
  
          # Run in interactive mode if requested
          interactive = ['-it'] if kwargs.get('interactive') else []
  
          # Note that we MUST run commands via bash, so variables
          # work as expected
          if isinstance(command, list):
              command = ' '.join(command)
          command = directory + environment + command
          return self.podman(
              ['exec'] + interactive +
              [self.container or 'dry', 'bash', '-c', command], **kwargs)
  
      def push(self, source=None, destination=None, options=None):
          """ Make sure that the workdir has a correct selinux context """
          self.debug("Update selinux context of the run workdir.", level=3)
          self.run(
              ["chcon", "--recursive", "--type=container_file_t",
               self.parent.plan.workdir], shell=False)
          # In case explicit destination is given, use `podman cp` to copy data
          # to the container
          if destination:
              self.podman(["cp", source, f"{self.container}:{destination}"])
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """ Nothing to be done to pull workdir """
  
      def stop(self):
          """ Stop provisioned guest """
          if self.container:
              self.podman(['container', 'stop', self.container])
              self.info('container', 'stopped', 'green')
  
      def remove(self):
          """ Remove the container """
          if self.container:
              self.podman(['container', 'rm', '-f', self.container])
              self.info('container', 'removed', 'green')
  
files.old_path: tmt/steps/provision/podman.py
files.new_contents: # coding: utf-8
  
  import dataclasses
  import os
  import platform
  import re
  import time
  from typing import Optional
  
  import click
  import fmf
  import requests
  
  import tmt
  import tmt.steps.provision
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  
  def import_testcloud():
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
    - {public_key}
  ssh_pwauth: true
  disable_root: false
  runcmd:
    - sed -i -e '/^.*PermitRootLogin/s/^.*$/PermitRootLogin yes/'
      -e '/^.*UseDNS/s/^.*$/UseDNS no/'
      -e '/^.*GSSAPIAuthentication/s/^.*$/GSSAPIAuthentication no/'
      /etc/ssh/sshd_config
    - systemctl reload sshd
    - [sh, -c, 'if [ ! -f /etc/systemd/network/20-tc-usernet.network ] &&
    systemctl status systemd-networkd | grep -q "enabled;\\svendor\\spreset:\\senabled";
    then mkdir -p /etc/systemd/network/ &&
    echo "[Match]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "Name=en*" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "[Network]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "DHCP=yes" >> /etc/systemd/network/20-tc-usernet.network; fi']
    - [sh, -c, 'if systemctl status systemd-networkd |
    grep -q "enabled;\\svendor\\spreset:\\senabled"; then
    systemctl restart systemd-networkd; fi']
    - [sh, -c, 'if cat /etc/os-release |
    grep -q platform:el8; then systemctl restart sshd; fi']
  """
  
  COREOS_DATA = """variant: fcos
  version: 1.4.0
  passwd:
    users:
      - name: {user_name}
        ssh_authorized_keys:
          - {public_key}
  """
  
  # Libvirt domain XML template related variables
  DOMAIN_TEMPLATE_NAME = 'domain-template.jinja'
  DOMAIN_TEMPLATE_FILE = os.path.join(TESTCLOUD_DATA, DOMAIN_TEMPLATE_NAME)
  DOMAIN_TEMPLATE = """<domain type='{{ virt_type }}' xmlns:qemu='http://libvirt.org/schemas/domain/qemu/1.0'>
    <name>{{ domain_name }}</name>
    <uuid>{{ uuid }}</uuid>
    <memory unit='KiB'>{{ memory }}</memory>
    <currentMemory unit='KiB'>{{ memory }}</currentMemory>
    <vcpu placement='static'>2</vcpu>
    <os>
      <type arch='{{ arch }}' machine='{{ model }}'>hvm</type>
      {{ uefi_loader }}
      <boot dev='hd'/>
    </os>
    {{ cpu }}
    {{ extra_specs }}
    <clock offset='utc'>
      <timer name='rtc' tickpolicy='catchup'/>
      <timer name='pit' tickpolicy='delay'/>
      <timer name='hpet' present='no'/>
    </clock>
    <on_poweroff>destroy</on_poweroff>
    <on_reboot>restart</on_reboot>
    <on_crash>restart</on_crash>
    <devices>
      <emulator>{{ emulator_path }}</emulator>
      <disk type='file' device='disk'>
        <driver name='qemu' type='qcow2' cache='unsafe'/>
        <source file="{{ disk }}"/>
        <target dev='vda' bus='virtio'/>
        {{ boot_drive_address }}
      </disk>
      <disk type='file' device='disk'>
        <driver name='qemu' type='raw'/>
        <source file="{{ seed }}"/>
        <target dev='vdb' bus='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x08' function='0x0'/>
      </disk>
      <interface type='{{ network_type }}'>
        <mac address="{{ mac_address }}"/>
        {{ network_source }}
        {{ ip_setup }}
        <model type='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>
      </interface>
      <serial type='pty'>
        <target port='0'/>
      </serial>
      <console type='pty'>
        <target type='serial' port='0'/>
      </console>
      <input type="keyboard" bus="virtio"/>
      <rng model='virtio'>
        <backend model='random'>/dev/urandom</backend>
      </rng>
    </devices>
    {{ qemu_args }}
  </domain>
  """  # noqa: E501
  
  # VM defaults
  DEFAULT_BOOT_TIMEOUT = 60      # seconds
  DEFAULT_CONNECT_TIMEOUT = 60   # seconds
  NON_KVM_ADDITIONAL_WAIT = 10       # seconds
  NON_KVM_TIMEOUT_COEF = 10          # times
  
  # SSH key type, set None for ssh-keygen default one
  SSH_KEYGEN_TYPE = "ecdsa"
  
  DEFAULT_USER = 'root'
  DEFAULT_MEMORY = 2048
  DEFAULT_DISK = 10
  DEFAULT_IMAGE = 'fedora'
  DEFAULT_CONNECTION = 'session'
  DEFAULT_ARCH = platform.machine()
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class TestcloudGuestData(
          tmt.steps.provision.GuestSshData):  # type: ignore[misc]
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      image: str = DEFAULT_IMAGE
      memory: int = DEFAULT_MEMORY
      disk: int = DEFAULT_DISK
      connection: str = DEFAULT_CONNECTION
      arch: str = DEFAULT_ARCH
  
      image_url: Optional[str] = None
      instance_name: Optional[str] = None
  
  
  class ProvisionTestcloud(tmt.steps.provision.ProvisionPlugin):
      """
      Local virtual machine using testcloud
  
      Minimal config which uses the latest fedora image:
  
          provision:
              how: virtual
  
      Here's a full config example:
  
          provision:
              how: virtual
              image: fedora
              user: root
              memory: 2048
  
      As the image use 'fedora' for the latest released Fedora compose,
      'rawhide' for the latest Rawhide compose, short aliases such as
      'fedora-32', 'f-32' or 'f32' for specific release or a full url to
      the qcow2 image for example from:
  
          https://kojipkgs.fedoraproject.org/compose/
  
      Short names are also provided for 'centos', 'centos-stream',
      'debian' and 'ubuntu' (e.g. 'centos-8' or 'c8').
  
      Supported Fedora CoreOS images are:
  
          fedora-coreos
          fedora-coreos-stable
          fedora-coreos-testing
          fedora-coreos-next
  
      Use the full path for images stored on local disk, for example:
  
          /var/tmp/images/Fedora-Cloud-Base-31-1.9.x86_64.qcow2
  
      In addition to the qcow2 format, vagrant boxes can be used as well,
      testcloud will take care of unpacking the image for you.
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [
          tmt.steps.Method(name='virtual.testcloud', doc=__doc__, order=50),
          ]
  
      # Supported keys
      _keys = ["image", "user", "memory", "disk", "connection", "arch"]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for testcloud """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to be used. Provide a short name, '
                       'full path to a local file or a complete url.'),
              click.option(
                  '-m', '--memory', metavar='MEMORY',
                  help='Set available memory in MB, 2048 MB by default.'),
              click.option(
                  '-D', '--disk', metavar='MEMORY',
                  help='Specify disk size in GB, 10 GB by default.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-c', '--connection',
                  type=click.Choice(['session', 'system']),
                  help="What session type to use, 'session' by default."),
              click.option(
                  '-a', '--arch',
                  type=click.Choice(['x86_64', 'aarch64', 's390x', 'ppc64le']),
                  help="What architecture to virtualize, host arch by default."),
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          return getattr(TestcloudGuestData(), option, default)
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
  
          # Convert memory and disk to integers
          # TODO: can they ever *not* be integers at this point?
          # for key in ['memory', 'disk']:
          #     if isinstance(self.get(key), str):
          #         self.data[key] = int(self.data[key])
  
          # Wake up testcloud instance
          if data:
              guest = GuestTestcloud(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self):
          """ Provision the testcloud instance """
          super().go()
  
          # Give info about provided data
          data = TestcloudGuestData(**{
              key: self.get(key)
              for key in TestcloudGuestData.keys()
              })
          for key, value in data.to_dict().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 value is not None:
                  self.info(key, value, 'green')
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestTestcloud(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      @classmethod
      def clean_images(cls, clean, dry):
          """ 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
          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)
                  os.remove(image)
  
  
  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.
      _image: Optional['testcloud.image.Image'] = None
      _instance: Optional['testcloud.instance.Instance'] = None
  
      def _get_url(self, url, message):
          """ Get url, retry when fails, return response """
          timeout = DEFAULT_CONNECT_TIMEOUT
          wait = 1
          while True:
              try:
                  with retry_session() as session:
                      response = session.get(url)
                  if response.ok:
                      return response
              except requests.RequestException:
                  pass
              if timeout < 0:
                  raise ProvisionError(
                      f'Failed to {message} in {DEFAULT_CONNECT_TIMEOUT}s.')
              self.debug(
                  f'Unable to {message} ({url}), retrying, '
                  f'{fmf.utils.listed(timeout, "second")} left.')
              time.sleep(wait)
              wait += 1
              timeout -= wait
  
      def _guess_image_url(self, name):
          """ 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 = 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)
  
          # Plain name match means we want the latest release
          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", 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), 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)
  
          if not url:
              raise ProvisionError(f"Could not map '{name}' to compose.")
          return url
  
      @staticmethod
      def _create_template():
          """ 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):
          """ Wake up the guest """
          self.debug(
              f"Waking up testcloud instance '{self.instance_name}'.",
              level=2, shift=0)
          self.prepare_config()
          self._image = testcloud.image.Image(self.image_url)
          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=None):
          """ Prepare ssh key for authentication """
          # Create ssh key paths
          key_name = "id_{}".format(key_type if key_type is not None else 'rsa')
          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, "-N", ""]
          if key_type is not None:
              command.extend(["-t", key_type])
          self.run(command)
          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):
          """ Prepare common configuration """
          import_testcloud()
  
          # Get configuration
          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):
          """ 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
          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.", original=error)
          except (testcloud.exceptions.TestcloudPermissionsError,
                  PermissionError) as error:
              raise ProvisionError(
                  f"Failed to prepare the image. Check the '{TESTCLOUD_IMAGES}' "
                  f"directory permissions.", original=error)
  
          # Create instance
          self.instance_name = self._tmt_name()
          self._instance = testcloud.instance.Instance(
              name=self.instance_name, 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
              self._instance.ssh_path = self.key
          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
          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 = self._instance.get_instance_port()
          self.verbose('ip', self.guest, 'green')
          self.verbose('port', self.port, 'green')
          self._instance.create_ip_file(self.guest)
  
          # Wait a bit until the box is up
          timeout = DEFAULT_CONNECT_TIMEOUT * time_coeff
          wait = 1
          while True:
              start_time = time.time()
              try:
                  self.execute('whoami')
                  break
              except tmt.utils.RunError:
                  if timeout < 0:
                      raise ProvisionError(
                          f"Failed to connect in "
                          f"{DEFAULT_CONNECT_TIMEOUT * time_coeff}s.")
                  self.debug(
                      f'Failed to connect to machine, retrying, '
                      f'{fmf.utils.listed(timeout, "second")} left.')
              attempt_duration = round(time.time() - start_time)
              time.sleep(wait)
              wait += 1
              timeout -= wait + attempt_duration
  
          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):
          """ 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}'.")
              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):
          """ 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=False, command=None):
          """ 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()
  
files.new_path: tmt/steps/provision/testcloud.py
files.old_contents: # coding: utf-8
  
  import os
  import platform
  import re
  import time
  from typing import Optional
  
  import click
  import fmf
  import requests
  
  import tmt
  import tmt.steps.provision
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  
  def import_testcloud():
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
    - {public_key}
  ssh_pwauth: true
  disable_root: false
  runcmd:
    - sed -i -e '/^.*PermitRootLogin/s/^.*$/PermitRootLogin yes/'
      -e '/^.*UseDNS/s/^.*$/UseDNS no/'
      -e '/^.*GSSAPIAuthentication/s/^.*$/GSSAPIAuthentication no/'
      /etc/ssh/sshd_config
    - systemctl reload sshd
    - [sh, -c, 'if [ ! -f /etc/systemd/network/20-tc-usernet.network ] &&
    systemctl status systemd-networkd | grep -q "enabled;\\svendor\\spreset:\\senabled";
    then mkdir -p /etc/systemd/network/ &&
    echo "[Match]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "Name=en*" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "[Network]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "DHCP=yes" >> /etc/systemd/network/20-tc-usernet.network; fi']
    - [sh, -c, 'if systemctl status systemd-networkd |
    grep -q "enabled;\\svendor\\spreset:\\senabled"; then
    systemctl restart systemd-networkd; fi']
    - [sh, -c, 'if cat /etc/os-release |
    grep -q platform:el8; then systemctl restart sshd; fi']
  """
  
  COREOS_DATA = """variant: fcos
  version: 1.4.0
  passwd:
    users:
      - name: {user_name}
        ssh_authorized_keys:
          - {public_key}
  """
  
  # Libvirt domain XML template related variables
  DOMAIN_TEMPLATE_NAME = 'domain-template.jinja'
  DOMAIN_TEMPLATE_FILE = os.path.join(TESTCLOUD_DATA, DOMAIN_TEMPLATE_NAME)
  DOMAIN_TEMPLATE = """<domain type='{{ virt_type }}' xmlns:qemu='http://libvirt.org/schemas/domain/qemu/1.0'>
    <name>{{ domain_name }}</name>
    <uuid>{{ uuid }}</uuid>
    <memory unit='KiB'>{{ memory }}</memory>
    <currentMemory unit='KiB'>{{ memory }}</currentMemory>
    <vcpu placement='static'>2</vcpu>
    <os>
      <type arch='{{ arch }}' machine='{{ model }}'>hvm</type>
      {{ uefi_loader }}
      <boot dev='hd'/>
    </os>
    {{ cpu }}
    {{ extra_specs }}
    <clock offset='utc'>
      <timer name='rtc' tickpolicy='catchup'/>
      <timer name='pit' tickpolicy='delay'/>
      <timer name='hpet' present='no'/>
    </clock>
    <on_poweroff>destroy</on_poweroff>
    <on_reboot>restart</on_reboot>
    <on_crash>restart</on_crash>
    <devices>
      <emulator>{{ emulator_path }}</emulator>
      <disk type='file' device='disk'>
        <driver name='qemu' type='qcow2' cache='unsafe'/>
        <source file="{{ disk }}"/>
        <target dev='vda' bus='virtio'/>
        {{ boot_drive_address }}
      </disk>
      <disk type='file' device='disk'>
        <driver name='qemu' type='raw'/>
        <source file="{{ seed }}"/>
        <target dev='vdb' bus='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x08' function='0x0'/>
      </disk>
      <interface type='{{ network_type }}'>
        <mac address="{{ mac_address }}"/>
        {{ network_source }}
        {{ ip_setup }}
        <model type='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>
      </interface>
      <serial type='pty'>
        <target port='0'/>
      </serial>
      <console type='pty'>
        <target type='serial' port='0'/>
      </console>
      <input type="keyboard" bus="virtio"/>
      <rng model='virtio'>
        <backend model='random'>/dev/urandom</backend>
      </rng>
    </devices>
    {{ qemu_args }}
  </domain>
  """  # noqa: E501
  
  # VM defaults
  DEFAULT_BOOT_TIMEOUT = 60      # seconds
  DEFAULT_CONNECT_TIMEOUT = 60   # seconds
  NON_KVM_ADDITIONAL_WAIT = 10       # seconds
  NON_KVM_TIMEOUT_COEF = 10          # times
  
  # SSH key type, set None for ssh-keygen default one
  SSH_KEYGEN_TYPE = "ecdsa"
  
  
  class ProvisionTestcloud(tmt.steps.provision.ProvisionPlugin):
      """
      Local virtual machine using testcloud
  
      Minimal config which uses the latest fedora image:
  
          provision:
              how: virtual
  
      Here's a full config example:
  
          provision:
              how: virtual
              image: fedora
              user: root
              memory: 2048
  
      As the image use 'fedora' for the latest released Fedora compose,
      'rawhide' for the latest Rawhide compose, short aliases such as
      'fedora-32', 'f-32' or 'f32' for specific release or a full url to
      the qcow2 image for example from:
  
          https://kojipkgs.fedoraproject.org/compose/
  
      Short names are also provided for 'centos', 'centos-stream',
      'debian' and 'ubuntu' (e.g. 'centos-8' or 'c8').
  
      Supported Fedora CoreOS images are:
  
          fedora-coreos
          fedora-coreos-stable
          fedora-coreos-testing
          fedora-coreos-next
  
      Use the full path for images stored on local disk, for example:
  
          /var/tmp/images/Fedora-Cloud-Base-31-1.9.x86_64.qcow2
  
      In addition to the qcow2 format, vagrant boxes can be used as well,
      testcloud will take care of unpacking the image for you.
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [
          tmt.steps.Method(name='virtual.testcloud', doc=__doc__, order=50),
          ]
  
      # Supported keys
      _keys = ["image", "user", "memory", "disk", "connection", "arch"]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for testcloud """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to be used. Provide a short name, '
                       'full path to a local file or a complete url.'),
              click.option(
                  '-m', '--memory', metavar='MEMORY',
                  help='Set available memory in MB, 2048 MB by default.'),
              click.option(
                  '-D', '--disk', metavar='MEMORY',
                  help='Specify disk size in GB, 10 GB by default.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-c', '--connection',
                  type=click.Choice(['session', 'system']),
                  help="What session type to use, 'session' by default."),
              click.option(
                  '-a', '--arch',
                  type=click.Choice(['x86_64', 'aarch64', 's390x', 'ppc64le']),
                  help="What architecture to virtualize, host arch by default."),
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          defaults = {
              'user': 'root',
              'memory': 2048,
              'disk': 10,
              'image': 'fedora',
              'connection': 'session',
              'arch': platform.machine()
              }
          if option in defaults:
              return defaults[option]
          return default
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
  
          # Convert memory and disk to integers
          for key in ['memory', 'disk']:
              if isinstance(self.get(key), str):
                  self.data[key] = int(self.data[key])
  
          # Wake up testcloud instance
          if data:
              guest = GuestTestcloud(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self):
          """ Provision the testcloud instance """
          super().go()
  
          # Give info about provided data
          data = dict()
          for key in self._keys + self._common_keys:
              data[key] = self.get(key)
              if key == 'memory':
                  self.info('memory', f"{self.get('memory')} MB", 'green')
              elif key == 'disk':
                  self.info('disk', f"{self.get('disk')} GB", 'green')
              elif key == 'connection':
                  self.verbose(key, data[key], 'green')
              else:
                  if data[key] is not None:
                      self.info(key, data[key], 'green')
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestTestcloud(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      @classmethod
      def clean_images(cls, clean, dry):
          """ 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
          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)
                  os.remove(image)
  
  
  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
      """
  
      # Not to be saved, recreated from image_url/instance_name/... every
      # time guest is instantiated.
      _image: Optional['testcloud.image.Image'] = None
      _instance: Optional['testcloud.instance.Instance'] = None
  
      def _get_url(self, url, message):
          """ Get url, retry when fails, return response """
          timeout = DEFAULT_CONNECT_TIMEOUT
          wait = 1
          while True:
              try:
                  with retry_session() as session:
                      response = session.get(url)
                  if response.ok:
                      return response
              except requests.RequestException:
                  pass
              if timeout < 0:
                  raise ProvisionError(
                      f'Failed to {message} in {DEFAULT_CONNECT_TIMEOUT}s.')
              self.debug(
                  f'Unable to {message} ({url}), retrying, '
                  f'{fmf.utils.listed(timeout, "second")} left.')
              time.sleep(wait)
              wait += 1
              timeout -= wait
  
      def _guess_image_url(self, name):
          """ 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 = 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)
  
          # Plain name match means we want the latest release
          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", 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), 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)
  
          if not url:
              raise ProvisionError(f"Could not map '{name}' to compose.")
          return url
  
      @staticmethod
      def _create_template():
          """ Create libvirt domain template """
          # Write always to ovewrite possible outdated version
          with open(DOMAIN_TEMPLATE_FILE, 'w') as template:
              template.write(DOMAIN_TEMPLATE)
  
      def load(self, data):
          """ Load guest data and initialize attributes """
          super().load(data)
          self.image = data.get('image')
          self.image_url = data.get('image_url')
          self.instance_name = data.get('instance_name')
          self.memory = data.get('memory')
          self.disk = data.get('disk')
          self.connection = data.get('connection')
          self.arch = data.get('arch')
  
      def save(self):
          """ Save guest data for future wake up """
          data = super().save()
          data['instance_name'] = self.instance_name
          data['image'] = self.image
          data['image_url'] = self.image_url
          data['connection'] = self.connection
          data['arch'] = self.arch
          return data
  
      def wake(self):
          """ Wake up the guest """
          self.debug(
              f"Waking up testcloud instance '{self.instance_name}'.",
              level=2, shift=0)
          self.prepare_config()
          self._image = testcloud.image.Image(self.image_url)
          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=None):
          """ Prepare ssh key for authentication """
          # Create ssh key paths
          key_name = "id_{}".format(key_type if key_type is not None else 'rsa')
          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, "-N", ""]
          if key_type is not None:
              command.extend(["-t", key_type])
          self.run(command)
          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):
          """ Prepare common configuration """
          import_testcloud()
  
          # Get configuration
          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):
          """ 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
          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.", original=error)
          except (testcloud.exceptions.TestcloudPermissionsError,
                  PermissionError) as error:
              raise ProvisionError(
                  f"Failed to prepare the image. Check the '{TESTCLOUD_IMAGES}' "
                  f"directory permissions.", original=error)
  
          # Create instance
          self.instance_name = self._tmt_name()
          self._instance = testcloud.instance.Instance(
              name=self.instance_name, 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
              self._instance.ssh_path = self.key
          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
          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 = self._instance.get_instance_port()
          self.verbose('ip', self.guest, 'green')
          self.verbose('port', self.port, 'green')
          self._instance.create_ip_file(self.guest)
  
          # Wait a bit until the box is up
          timeout = DEFAULT_CONNECT_TIMEOUT * time_coeff
          wait = 1
          while True:
              start_time = time.time()
              try:
                  self.execute('whoami')
                  break
              except tmt.utils.RunError:
                  if timeout < 0:
                      raise ProvisionError(
                          f"Failed to connect in "
                          f"{DEFAULT_CONNECT_TIMEOUT * time_coeff}s.")
                  self.debug(
                      f'Failed to connect to machine, retrying, '
                      f'{fmf.utils.listed(timeout, "second")} left.')
              attempt_duration = round(time.time() - start_time)
              time.sleep(wait)
              wait += 1
              timeout -= wait + attempt_duration
  
          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):
          """ 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}'.")
              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):
          """ 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=False, command=None):
          """ 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()
  
files.old_path: tmt/steps/provision/testcloud.py