files.new_contents: discover:
      how: fmf
  provision:
      how: virtual
      memory: 4000
      disk: 40
  execute:
      how: tmt
  
  
  [... the rest is omitted ...]
files.new_path: tests/full/plan.fmf
files.old_contents: discover:
      how: fmf
  provision:
      how: virtual
      memory: 4000
  execute:
      how: tmt
  
  
  [... the rest is omitted ...]
files.old_path: tests/full/plan.fmf
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 shlex import quote
  from typing import Dict, List, Optional, Type
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.utils
  from tmt.steps import Action
  
  # Timeout in seconds of waiting for a connection after reboot
  CONNECTION_TIMEOUT = 5 * 60
  
  # 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()
  
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/__init__.py
files.old_contents: import collections
  import dataclasses
  import datetime
  import os
  import random
  import re
  import shlex
  import string
  import subprocess
  import tempfile
  import time
  from shlex import quote
  from threading import Thread
  from typing import Dict, List, Optional, Type
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.utils
  from tmt.steps import Action
  
  # Timeout in seconds of waiting for a connection after reboot
  CONNECTION_TIMEOUT = 5 * 60
  
  # 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()
  
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/__init__.py