files: 
  """ Step Classes """
  
  import os
  import re
  
  import click
  from click import echo
  
  import tmt.utils
  from tmt.options import show_step_method_hints
  from tmt.utils import GeneralError
  
  # Supported steps and actions
  STEPS = ['discover', 'provision', 'prepare', 'execute', 'report', 'finish']
  ACTIONS = ['login', 'reboot']
  
  # Step phase order
  PHASE_START = 10
  PHASE_BASE = 50
  PHASE_END = 90
  
  
  class Step(tmt.utils.Common):
      """ Common parent of all test steps """
  
      # Default implementation for all steps is shell
      # except for provision (virtual) and report (display)
      how = 'shell'
  
      def __init__(self, data=None, plan=None, name=None):
          """ Initialize and check the step data """
          super().__init__(name=name, parent=plan)
          # Initialize data
          self.plan = plan
          self.data = data or {}
          self._status = None
          self._plugins = []
  
          # Create an empty step by default (can be updated from cli)
          if self.data is None:
              self.data = [{'name': tmt.utils.DEFAULT_NAME}]
          # Convert to list if only a single config provided
          elif isinstance(self.data, dict):
              # Give it a name unless defined
              if not self.data.get('name'):
                  self.data['name'] = tmt.utils.DEFAULT_NAME
              self.data = [self.data]
          # Shout about invalid configuration
          elif not isinstance(self.data, list):
              raise GeneralError(f"Invalid '{self}' config in '{self.plan}'.")
  
          # Add default unique names even to multiple configs so that the users
          # don't need to specify it if they don't care about the name
          for i, data in enumerate(self.data):
              if 'name' not in data:
                  data['name'] = f'{tmt.utils.DEFAULT_NAME}-{i}'
  
          # Final sanity checks
          for data in self.data:
              # Set 'how' to the default if not specified
              if data.get('how') is None:
                  data['how'] = self.how
              # Ensure that each config has a name
              if 'name' not in data and len(self.data) > 1:
                  raise GeneralError(
                      f"Missing 'name' in the {self} step config "
                      f"of the '{self.plan}' plan.")
  
      @property
      def enabled(self):
          """ True if the step is enabled """
          try:
              return self.name in self.plan.my_run._context.obj.steps
          except AttributeError:
              return None
  
      @classmethod
      def usage(cls, method_overview):
          """ Prepare general usage message for the step """
          # Main description comes from the class docstring
          usage = re.sub('\n    ', '\n', cls.__doc__)
          # Append the list of supported methods
          usage += '\n\n' + method_overview
          # Give a hint about detailed help
          name = cls.__name__.lower()
          usage += (
              f"\n\nUse 'tmt run {name} --help --how <method>' to learn more "
              f"about given {name} method and all its supported options.")
          return usage
  
      def status(self, status=None):
          """
          Get and set current step status
  
          The meaning of the status is as follows:
          todo ... config, data and command line processed (we know what to do)
          done ... the final result of the step stored to workdir (we are done)
          """
          # Update status
          if status is not None:
              # Check for valid values
              if status not in ['todo', 'done']:
                  raise GeneralError(f"Invalid status '{status}'.")
              # Show status only if changed
              elif self._status != status:
                  self._status = status
                  self.debug('status', status, color='yellow', level=2)
          # Return status
          return self._status
  
      def load(self, extra_keys=None):
          """ Load status and step data from the workdir """
          extra_keys = extra_keys or []
          try:
              content = tmt.utils.yaml_to_dict(self.read('step.yaml'))
              self.debug('Successfully loaded step data.', level=2)
              self.data = content['data']
              for key in extra_keys:
                  if key in content:
                      setattr(self, key, content[key])
              self.status(content['status'])
          except GeneralError:
              self.debug('Step data not found.', level=2)
  
      def save(self, data=None):
          """ Save status and step data to the workdir """
          data = data or {}
          content = dict(status=self.status(), data=self.data)
          content.update(data)
          self.write('step.yaml', tmt.utils.dict_to_yaml(content))
  
      def wake(self):
          """ Wake up the step (process workdir and command line) """
          # Cleanup possible old workdir if called with --force
          if self.opt('force'):
              self._workdir_cleanup()
  
          # Load stored data
          self.load()
  
          # Status 'todo' means the step has not finished successfully.
          # Probably interrupted in the middle. Clean up the work
          # directory to give it another chance with a fresh start.
          if self.status() == 'todo':
              self.debug("Step has not finished. Let's try once more!", level=2)
              self._workdir_cleanup()
  
          # Special handling for the report step to always enable force mode in
          # order to cover a very frequent use case 'tmt run --last report'
          # FIXME find a better way how to enable always-force per plugin
          if (isinstance(self, tmt.steps.report.Report) and
                  self.data[0].get('how') in ['display', 'html']):
              self.debug("Report step always force mode enabled.")
              self._workdir_cleanup()
              self.status('todo')
  
          # Nothing more to do when the step is already done
          if self.status() == 'done':
              return
  
  
  
  [... the rest is omitted ...]
files: tmt/steps/__init__.py
files: 
  """ Step Classes """
  
  import os
  import re
  
  import click
  from click import echo
  
  import tmt.utils
  from tmt.options import show_step_method_hints
  from tmt.utils import GeneralError
  
  # Supported steps and actions
  STEPS = ['discover', 'provision', 'prepare', 'execute', 'report', 'finish']
  ACTIONS = ['login', 'reboot']
  
  # Step phase order
  PHASE_START = 10
  PHASE_BASE = 50
  PHASE_END = 90
  
  
  class Step(tmt.utils.Common):
      """ Common parent of all test steps """
  
      # Default implementation for all steps is shell
      # except for provision (virtual) and report (display)
      how = 'shell'
  
      def __init__(self, data=None, plan=None, name=None):
          """ Initialize and check the step data """
          super().__init__(name=name, parent=plan)
          # Initialize data
          self.plan = plan
          self.data = data or {}
          self._status = None
          self._plugins = []
  
          # Create an empty step by default (can be updated from cli)
          if self.data is None:
              self.data = [{'name': tmt.utils.DEFAULT_NAME}]
          # Convert to list if only a single config provided
          elif isinstance(self.data, dict):
              # Give it a name unless defined
              if not self.data.get('name'):
                  self.data['name'] = tmt.utils.DEFAULT_NAME
              self.data = [self.data]
          # Shout about invalid configuration
          elif not isinstance(self.data, list):
              raise GeneralError(f"Invalid '{self}' config in '{self.plan}'.")
  
          # Add default unique names even to multiple configs so that the users
          # don't need to specify it if they don't care about the name
          for i, data in enumerate(self.data):
              if 'name' not in data:
                  data['name'] = f'{tmt.utils.DEFAULT_NAME}-{i}'
  
          # Final sanity checks
          for data in self.data:
              # Set 'how' to the default if not specified
              if data.get('how') is None:
                  data['how'] = self.how
              # Ensure that each config has a name
              if 'name' not in data and len(self.data) > 1:
                  raise GeneralError(
                      f"Missing 'name' in the {self} step config "
                      f"of the '{self.plan}' plan.")
  
      @property
      def enabled(self):
          """ True if the step is enabled """
          try:
              return self.name in self.plan.my_run._context.obj.steps
          except AttributeError:
              return None
  
      @classmethod
      def usage(cls, method_overview):
          """ Prepare general usage message for the step """
          # Main description comes from the class docstring
          usage = re.sub('\n    ', '\n', cls.__doc__)
          # Append the list of supported methods
          usage += '\n\n' + method_overview
          # Give a hint about detailed help
          name = cls.__name__.lower()
          usage += (
              f"\n\nUse 'tmt run {name} --help --how <method>' to learn more "
              f"about given {name} method and all its supported options.")
          return usage
  
      def status(self, status=None):
          """
          Get and set current step status
  
          The meaning of the status is as follows:
          todo ... config, data and command line processed (we know what to do)
          done ... the final result of the step stored to workdir (we are done)
          """
          # Update status
          if status is not None:
              # Check for valid values
              if status not in ['todo', 'done']:
                  raise GeneralError(f"Invalid status '{status}'.")
              # Show status only if changed
              elif self._status != status:
                  self._status = status
                  self.debug('status', status, color='yellow', level=2)
          # Return status
          return self._status
  
      def load(self, extra_keys=None):
          """ Load status and step data from the workdir """
          extra_keys = extra_keys or []
          try:
              content = tmt.utils.yaml_to_dict(self.read('step.yaml'))
              self.debug('Successfully loaded step data.', level=2)
              self.data = content['data']
              for key in extra_keys:
                  if key in content:
                      setattr(self, key, content[key])
              self.status(content['status'])
          except GeneralError:
              self.debug('Step data not found.', level=2)
  
      def save(self, data=None):
          """ Save status and step data to the workdir """
          data = data or {}
          content = dict(status=self.status(), data=self.data)
          content.update(data)
          self.write('step.yaml', tmt.utils.dict_to_yaml(content))
  
      def wake(self):
          """ Wake up the step (process workdir and command line) """
          # Cleanup possible old workdir if called with --force
          if self.opt('force'):
              self._workdir_cleanup()
  
          # Load stored data
          self.load()
  
          # Status 'todo' means the step has not finished successfully.
          # Probably interrupted in the middle. Clean up the work
          # directory to give it another chance with a fresh start.
          if self.status() == 'todo':
              self.debug("Step has not finished. Let's try once more!", level=2)
              self._workdir_cleanup()
  
          # Special handling for the report step to always enable force mode in
          # order to cover a very frequent use case 'tmt run --last report'
          # FIXME find a better way how to enable always-force per plugin
          if (isinstance(self, tmt.steps.report.Report) and
                  self.data[0].get('how') in ['display', 'html']):
              self.debug("Report step always force mode enabled.")
              self._workdir_cleanup()
              self.status('todo')
  
          # Nothing more to do when the step is already done
          if self.status() == 'done':
              return
  
  
  
  [... the rest is omitted ...]
files: tmt/steps/__init__.py
files: import copy
  import re
  
  import click
  import fmf
  
  import tmt
  
  
  class Finish(tmt.steps.Step):
      """
      Perform the finishing tasks and clean up provisioned guests.
  
      Additional actions to be performed after the test execution has been
      completed. Counterpart of the ``prepare`` step useful for various
      cleanup actions. Also takes care of stopping and removing guests.
  
      Note that the ``finish`` step is also run when any of the previous
      steps failed (for example when the environment preparation was not
      successful) so that provisioned systems are not kept running.
      """
  
      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 = FinishPlugin.delegate(self, data)
              plugin.wake()
              # Add plugin only if there are data
              if len(plugin.data.keys()) > 2:
                  self._plugins.append(plugin)
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Finish 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 finish details """
          for data in self.data:
              FinishPlugin.delegate(self, data).show()
  
      def summary(self):
          """ Give a concise summary """
          tasks = fmf.utils.listed(self.plugins(), 'task')
          self.info('summary', f'{tasks} completed', 'green', shift=1)
  
      def go(self):
          """ Execute finishing tasks """
          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
  
          # Go and execute each plugin on all guests
          for guest in self.plan.provision.guests():
              # Create a guest copy and change its parent so that the
              # operations inside finish plugins on the guest use the
              # finish step config rather than provision step config.
              guest_copy = copy.copy(guest)
              guest_copy.parent = self
              for plugin in self.plugins():
                  plugin.go(guest_copy)
              # Pull artifacts created in the plan data directory
              # if there was at least one plugin executed
              if self.plugins():
                  guest_copy.pull(self.plan.data_directory)
  
          # Stop and remove provisioned guests
          for guest in self.plan.provision.guests():
              guest.stop()
              guest.remove()
  
          # Give a summary, update status and save
          self.summary()
          self.status('done')
          self.save()
  
      def requires(self):
          """
          Packages required by all enabled finish plugins
  
          Return a list of packages which need to be installed on the
          provisioned guest so that the finishing tasks work well.
          Used by the prepare step.
          """
          requires = set()
          for plugin in self.plugins(classes=FinishPlugin):
              requires.update(plugin.requires())
          return list(requires)
  
  
  class FinishPlugin(tmt.steps.Plugin):
      """ Common parent of finish plugins """
  
      # List of all supported methods aggregated from all plugins
      _supported_methods = []
  
      @classmethod
      def base_command(cls, method_class=None, usage=None):
          """ Create base click command (common for all finish plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Finish.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 finishing tasks.')
          def finish(context, **kwargs):
              context.obj.steps.add('finish')
              Finish._save_context(context)
  
          return finish
  
  
  [... the rest is omitted ...]
files: tmt/steps/finish/__init__.py
files: import copy
  import re
  
  import click
  import fmf
  
  import tmt
  
  
  class Finish(tmt.steps.Step):
      """
      Perform the finishing tasks and clean up provisioned guests.
  
      Additional actions to be performed after the test execution has been
      completed. Counterpart of the ``prepare`` step useful for various
      cleanup actions. Also takes care of stopping and removing guests.
  
      Note that the ``finish`` step is also run when any of the previous
      steps failed (for example when the environment preparation was not
      successful) so that provisioned systems are not kept running.
      """
  
      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 = FinishPlugin.delegate(self, data)
              plugin.wake()
              # Add plugin only if there are data
              if len(plugin.data.keys()) > 2:
                  self._plugins.append(plugin)
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Finish 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 finish details """
          for data in self.data:
              FinishPlugin.delegate(self, data).show()
  
      def summary(self):
          """ Give a concise summary """
          tasks = fmf.utils.listed(self.plugins(), 'task')
          self.info('summary', f'{tasks} completed', 'green', shift=1)
  
      def go(self):
          """ Execute finishing tasks """
          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
  
          # Go and execute each plugin on all guests
          for guest in self.plan.provision.guests():
              # Create a guest copy and change its parent so that the
              # operations inside finish plugins on the guest use the
              # finish step config rather than provision step config.
              guest_copy = copy.copy(guest)
              guest_copy.parent = self
              for plugin in self.plugins():
                  plugin.go(guest_copy)
              # Pull artifacts created in the plan data directory
              # if there was at least one plugin executed
              if self.plugins():
                  guest_copy.pull(self.plan.data_directory)
  
          # Stop and remove provisioned guests
          for guest in self.plan.provision.guests():
              guest.stop()
              guest.remove()
  
          # Give a summary, update status and save
          self.summary()
          self.status('done')
          self.save()
  
  
  class FinishPlugin(tmt.steps.Plugin):
      """ Common parent of finish plugins """
  
      # List of all supported methods aggregated from all plugins
      _supported_methods = []
  
      @classmethod
      def base_command(cls, method_class=None, usage=None):
          """ Create base click command (common for all finish plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Finish.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 finishing tasks.')
          def finish(context, **kwargs):
              context.obj.steps.add('finish')
              Finish._save_context(context)
  
          return finish
  
  
  [... the rest is omitted ...]
files: tmt/steps/finish/__init__.py
files: import collections
  import copy
  
  import click
  import fmf
  
  import tmt
  
  
  class Prepare(tmt.steps.Step):
      """
      Prepare the environment for testing.
  
      Use the 'order' attribute to select in which order preparation
      should happen if there are multiple configs. Default order is '50'.
      Default order of required packages installation is '70', for the
      recommended packages it is '75'.
      """
  
      def __init__(self, data, plan):
          """ Initialize prepare step data """
          super().__init__(data, plan)
          self.preparations_applied = 0
  
      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 = PreparePlugin.delegate(self, data)
              plugin.wake()
              # Add plugin only if there are data
              if len(plugin.data.keys()) > 2:
                  self._plugins.append(plugin)
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Prepare 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:
              PreparePlugin.delegate(self, data).show()
  
      def summary(self):
          """ Give a concise summary of the preparation """
          preparations = fmf.utils.listed(
              self.preparations_applied, 'preparation')
          self.info('summary', f'{preparations} applied', 'green', shift=1)
  
      def _prepare_roles(self):
          """ Create a mapping of roles to guest names """
          role_mapping = collections.defaultdict(list)
          for guest in self.plan.provision.guests():
              if guest.role:
                  role_mapping[guest.role].append(guest.name)
          return role_mapping
  
      def _prepare_hosts(self):
          """ Create a mapping of guest names to IP addresses """
          host_mapping = {}
          for guest in self.plan.provision.guests():
              if hasattr(guest, 'guest') and guest.guest:
                  # FIXME: guest.guest may not be simply an IP address but also
                  #        a host name.
                  host_mapping[guest.name] = guest.guest
          return host_mapping
  
      def go(self):
          """ Prepare the 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
  
          # Required packages
          requires = set(
              self.plan.discover.requires() +
              self.plan.provision.requires() +
              self.plan.prepare.requires() +
              self.plan.execute.requires() +
              self.plan.report.requires() +
              self.plan.finish.requires()
              )
  
          if requires:
              data = dict(
                  how='install',
                  name='requires',
                  summary='Install required packages',
                  order=tmt.utils.DEFAULT_PLUGIN_ORDER_REQUIRES,
                  package=list(requires))
              self._plugins.append(PreparePlugin.delegate(self, data))
  
          # Recommended packages
          recommends = self.plan.discover.recommends()
          if recommends:
              data = dict(
                  how='install',
                  name='recommends',
                  summary='Install recommended packages',
                  order=tmt.utils.DEFAULT_PLUGIN_ORDER_RECOMMENDS,
                  package=recommends,
                  missing='skip')
              self._plugins.append(PreparePlugin.delegate(self, data))
  
          # Implicit multihost setup
          if self.plan.provision.is_multihost:
              data = dict(
                  how='multihost',
                  name='multihost',
                  summary='Setup guest for multihost testing',
                  order=tmt.utils.DEFAULT_PLUGIN_ORDER_MULTIHOST,
                  roles=self._prepare_roles(),
                  hosts=self._prepare_hosts(),
                  )
              self._plugins.append(PreparePlugin.delegate(self, data))
  
          # Prepare guests (including workdir sync)
          for guest in self.plan.provision.guests():
              guest.push()
              # Create a guest copy and change its parent so that the
              # operations inside prepare plugins on the guest use the
              # prepare step config rather than provision step config.
              guest_copy = copy.copy(guest)
              guest_copy.parent = self
              # Execute each prepare plugin
              for plugin in self.plugins():
                  if plugin.enabled_on_guest(guest_copy):
                      self.preparations_applied += 1
                      plugin.go(guest_copy)
                      self.info('')
              # Pull artifacts created in the plan data directory
              # if there was at least one plugin executed
              if self.plugins():
                  guest_copy.pull(self.plan.data_directory)
  
          # Give a summary, update status and save
          self.summary()
          self.status('done')
          self.save()
  
      def requires(self):
          """
          Packages required by all enabled prepare plugins
  
          Return a list of packages which need to be installed on the
          provisioned guest so that the preparation tasks work well.
          Used by the prepare step.
          """
          requires = set()
          for plugin in self.plugins(classes=PreparePlugin):
              requires.update(plugin.requires())
          return list(requires)
  
  
  class PreparePlugin(tmt.steps.Plugin):
      """ Common parent of prepare plugins """
  
      # List of all supported methods aggregated from all plugins
      _supported_methods = []
  
      # Common keys for all prepare step implementations
      _common_keys = ['where']
  
      @classmethod
      def base_command(cls, method_class=None, usage=None):
          """ Create base click command (common for all prepare plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Prepare.usage(method_overview=usage)
  
          # Create the command
          @click.command
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/__init__.py
files: import collections
  import copy
  
  import click
  import fmf
  
  import tmt
  
  
  class Prepare(tmt.steps.Step):
      """
      Prepare the environment for testing.
  
      Use the 'order' attribute to select in which order preparation
      should happen if there are multiple configs. Default order is '50'.
      Default order of required packages installation is '70', for the
      recommended packages it is '75'.
      """
  
      def __init__(self, data, plan):
          """ Initialize prepare step data """
          super().__init__(data, plan)
          self.preparations_applied = 0
  
      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 = PreparePlugin.delegate(self, data)
              plugin.wake()
              # Add plugin only if there are data
              if len(plugin.data.keys()) > 2:
                  self._plugins.append(plugin)
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Prepare 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:
              PreparePlugin.delegate(self, data).show()
  
      def summary(self):
          """ Give a concise summary of the preparation """
          preparations = fmf.utils.listed(
              self.preparations_applied, 'preparation')
          self.info('summary', f'{preparations} applied', 'green', shift=1)
  
      def _prepare_roles(self):
          """ Create a mapping of roles to guest names """
          role_mapping = collections.defaultdict(list)
          for guest in self.plan.provision.guests():
              if guest.role:
                  role_mapping[guest.role].append(guest.name)
          return role_mapping
  
      def _prepare_hosts(self):
          """ Create a mapping of guest names to IP addresses """
          host_mapping = {}
          for guest in self.plan.provision.guests():
              if hasattr(guest, 'guest') and guest.guest:
                  # FIXME: guest.guest may not be simply an IP address but also
                  #        a host name.
                  host_mapping[guest.name] = guest.guest
          return host_mapping
  
      def go(self):
          """ Prepare the 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
  
          # Required packages
          requires = set(
              self.plan.discover.requires() +
              self.plan.provision.requires() +
              self.plan.execute.requires()
              )
  
          if requires:
              data = dict(
                  how='install',
                  name='requires',
                  summary='Install required packages',
                  order=tmt.utils.DEFAULT_PLUGIN_ORDER_REQUIRES,
                  package=list(requires))
              self._plugins.append(PreparePlugin.delegate(self, data))
  
          # Recommended packages
          recommends = self.plan.discover.recommends()
          if recommends:
              data = dict(
                  how='install',
                  name='recommends',
                  summary='Install recommended packages',
                  order=tmt.utils.DEFAULT_PLUGIN_ORDER_RECOMMENDS,
                  package=recommends,
                  missing='skip')
              self._plugins.append(PreparePlugin.delegate(self, data))
  
          # Implicit multihost setup
          if self.plan.provision.is_multihost:
              data = dict(
                  how='multihost',
                  name='multihost',
                  summary='Setup guest for multihost testing',
                  order=tmt.utils.DEFAULT_PLUGIN_ORDER_MULTIHOST,
                  roles=self._prepare_roles(),
                  hosts=self._prepare_hosts(),
                  )
              self._plugins.append(PreparePlugin.delegate(self, data))
  
          # Prepare guests (including workdir sync)
          for guest in self.plan.provision.guests():
              guest.push()
              # Create a guest copy and change its parent so that the
              # operations inside prepare plugins on the guest use the
              # prepare step config rather than provision step config.
              guest_copy = copy.copy(guest)
              guest_copy.parent = self
              # Execute each prepare plugin
              for plugin in self.plugins():
                  if plugin.enabled_on_guest(guest_copy):
                      self.preparations_applied += 1
                      plugin.go(guest_copy)
                      self.info('')
              # Pull artifacts created in the plan data directory
              # if there was at least one plugin executed
              if self.plugins():
                  guest_copy.pull(self.plan.data_directory)
  
          # Give a summary, update status and save
          self.summary()
          self.status('done')
          self.save()
  
  
  class PreparePlugin(tmt.steps.Plugin):
      """ Common parent of prepare plugins """
  
      # List of all supported methods aggregated from all plugins
      _supported_methods = []
  
      # Common keys for all prepare step implementations
      _common_keys = ['where']
  
      @classmethod
      def base_command(cls, method_class=None, usage=None):
          """ Create base click command (common for all prepare plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Prepare.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 environment preparation.')
          def prepare(context, **kwargs):
              context.obj.steps.add('prepare')
              Prepare._save_context(context)
  
          return prepare
  
      def go(self, guest):
          """ Prepare the guest (common actions) """
          super().go()
  
          # Show guest name first in multihost scenarios
          if self.step.plan.pro
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/__init__.py
files: import tmt
  
  
  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 ansible(self, playbook, extra_args=None):
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, stderr = self.run(
              f'sudo sh -c "stty cols {tmt.utils.OUTPUT_WIDTH}; '
              f'{self._export_environment()}ansible-playbook '
              f'{self._ansible_verbosity()} '
              f'{self._ansible_extra_args(extra_args)} -c local -i localhost,'
              f' {playbook}"')
          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, **kwargs)
  
      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):
          """ Nothing to be done to pull workdir """
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/local.py
files: import tmt
  
  
  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 ansible(self, playbook, extra_args=None):
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, stderr = self.run(
              f'sudo sh -c "stty cols {tmt.utils.OUTPUT_WIDTH}; '
              f'{self._export_environment()}ansible-playbook '
              f'{self._ansible_verbosity()} '
              f'{self._ansible_extra_args(extra_args)} -c local -i localhost,'
              f' {playbook}"')
          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, **kwargs)
  
      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):
          """ Nothing to be done to pull workdir """
  
      @classmethod
      def requires(cls):
          """ No packages needed to sync workdir """
          return []
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/local.py
files: import os
  
  import click
  
  import tmt
  
  
  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'.
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["image", "container", "pull"]
  
      # 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.'),
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          # User 'fedora' as a default image
          if option == 'image':
              return 'fedora'
          # 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')
  
          # Instances variables initialized later
          self.container_id = None
  
      def save(self):
          """ Save guest data for future wake up """
          data = super().save()
          data['container'] = self.container
          data['image'] = self.image
          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._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.container_id = self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f'{workdir}:{workdir}:z',
               '-itd', self.image])[0].strip()
  
      def reboot(self, hard=False):
          """ Restart the container, return True if successful  """
          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()
  
      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(
              f'stty cols {tmt.utils.OUTPUT_WIDTH}; '
              f'{self._export_environment()}'
              f'{podman_unshare}ansible-playbook '
              f'{self._ansible_verbosity()} '
              f'{self._ansible_extra_args(extra_args)} '
              f'-c podman -i {self.container}, {playbook}',
              cwd=self.parent.plan.worktree)
          self._ansible_summary(stdout)
  
      def podman(self, command, **kwargs):
          """ Run given command via podman """
          return self.run(['podman'] + command, shell=False, **kwargs)
  
      def execute(self, command,
  
  [... the rest is omitted ...]
files: tmt/steps/provision/podman.py
files: import os
  
  import click
  
  import tmt
  
  
  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'.
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["image", "container", "pull"]
  
      # 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.'),
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          # User 'fedora' as a default image
          if option == 'image':
              return 'fedora'
          # 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')
  
          # Instances variables initialized later
          self.container_id = None
  
      def save(self):
          """ Save guest data for future wake up """
          data = super().save()
          data['container'] = self.container
          data['image'] = self.image
          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._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.container_id = self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f'{workdir}:{workdir}:z',
               '-itd', self.image])[0].strip()
  
      def reboot(self, hard=False):
          """ Restart the container, return True if successful  """
          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()
  
      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(
              f'stty cols {tmt.utils.OUTPUT_WIDTH}; '
              f'{self._export_environment()}'
              f'{podman_unshare}ansible-playbook '
              f'{self._ansible_verbosity()} '
              f'{self._ansible_extra_args(extra_args)} '
              f'-c podman -i {self.container}, {playbook}',
              cwd=self.parent.plan.worktree)
          self._ansible_summary(stdout)
  
      def podman(self, command, **kwargs):
          """ Run given command via podman """
          return self.run(['podman'] + command, shell=False, **kwargs)
  
      def execute(self, command,
  
  [... the rest is omitted ...]
files: tmt/steps/provision/podman.py
files: import re
  
  import click
  
  import tmt
  
  
  class Report(tmt.steps.Step):
      """ Provide test results overview and send reports. """
  
      # Default implementation for report is display
      how = 'display'
  
      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 = ReportPlugin.delegate(self, data)
              plugin.wake()
              self._plugins.append(plugin)
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Report 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:
              ReportPlugin.delegate(self, data).show()
  
      def summary(self):
          """ Give a concise report summary """
          summary = tmt.base.Result.summary(self.plan.execute.results())
          self.info('summary', summary, 'green', shift=1)
  
      def go(self):
          """ Report the 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
  
          # Perform the reporting
          for plugin in self.plugins():
              plugin.go()
  
          # Give a summary, update status and save
          self.summary()
          self.status('done')
          self.save()
  
      def requires(self):
          """
          Packages required by all enabled report plugins
  
          Return a list of packages which need to be installed on the
          provisioned guest so that the full report can be successfully
          generated. Used by the prepare step.
          """
          requires = set()
          for plugin in self.plugins(classes=ReportPlugin):
              requires.update(plugin.requires())
          return list(requires)
  
  
  class ReportPlugin(tmt.steps.Plugin):
      """ Common parent of report plugins """
  
      # Default implementation for report is display
      how = 'display'
  
      # List of all supported methods aggregated from all plugins
      _supported_methods = []
  
      @classmethod
      def base_command(cls, method_class=None, usage=None):
          """ Create base click command (common for all report plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Report.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 results reporting.')
          def report(context, **kwargs):
              context.obj.steps.add('report')
              Report._save_context(context)
  
          return report
  
  
  [... the rest is omitted ...]
files: tmt/steps/report/__init__.py
files: import re
  
  import click
  
  import tmt
  
  
  class Report(tmt.steps.Step):
      """ Provide test results overview and send reports. """
  
      # Default implementation for report is display
      how = 'display'
  
      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 = ReportPlugin.delegate(self, data)
              plugin.wake()
              self._plugins.append(plugin)
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Report 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:
              ReportPlugin.delegate(self, data).show()
  
      def summary(self):
          """ Give a concise report summary """
          summary = tmt.base.Result.summary(self.plan.execute.results())
          self.info('summary', summary, 'green', shift=1)
  
      def go(self):
          """ Report the 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
  
          # Perform the reporting
          for plugin in self.plugins():
              plugin.go()
  
          # Give a summary, update status and save
          self.summary()
          self.status('done')
          self.save()
  
  
  class ReportPlugin(tmt.steps.Plugin):
      """ Common parent of report plugins """
  
      # Default implementation for report is display
      how = 'display'
  
      # List of all supported methods aggregated from all plugins
      _supported_methods = []
  
      @classmethod
      def base_command(cls, method_class=None, usage=None):
          """ Create base click command (common for all report plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Report.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 results reporting.')
          def report(context, **kwargs):
              context.obj.steps.add('report')
              Report._save_context(context)
  
          return report
  
  
  [... the rest is omitted ...]
files: tmt/steps/report/__init__.py