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
  
          # Override step data with command line options
          how = self.opt('how')
          if how is not None:
              for data in self.data:
                  data['how'] = how
  
      def setup_actions(self):
          """ Insert login and reboot plugins if requested """
          for plugin in Login.plugins(step=self):
              self.debug(
                  f"Insert a login plugin into the '{self}' step "
                  f"with order '{plugin.order}'.", level=2)
              self._plugins.append(plugin)
  
          for plugin in Reboot.plugins(step=self):
              self.debug(
                  f"Insert a reboot plugin into the '{self}' step "
                  f"with order '{plugin.order}'.", level=2)
              self._plugins.append(plugin)
  
      def plugins(self, classes=None):
          """
          Iterate over plugins by their order
  
          By default iterates over all available plugins. Optional filter
          'classes' can be used to iterate only over instances of given
          class (single class or tuple of classes).
          """
          return sorted(
              [plugin for plugin in self._plugins
                  if classes is None or isinstance(plugin, classes)],
              key=lambda plugin: plugin.order)
  
      def actions(self):
          """ Run all loaded Login or Reboot plugin instances of the step """
          for plugin in self.plugins():
              if isinstance(plugin, (Reboot, Login)):
                  plugin.go()
  
      def go(self):
          """ Execute the test step """
          # Show step header and how
          self.info(self.name, color='blue')
          # Show workdir in verbose mode
          self.debug('workdir', self.workdir, 'magenta')
  
  
  class Method(object):
      """ Step implementation method """
  
      def __init__(self, name, doc, order):
          """ Store method data """
          self.name = name
          self.doc = doc.strip()
          self.order = order
  
          # Parse summary and description from provided doc string
          lines = [re.sub('^    ', '', line) for line in self.doc.split('\n')]
          self.summary = lines[0].strip()
          self.description = '\n'.join(lines[1:]).lstrip()
  
      def describe(self):
          """ Format name and summary for a nice method overview """
          return f'{self.name} '.ljust(22, '.') + f' {self.summary}'
  
      def usage(self):
          """ Prepare a detailed usage from summary and description """
          if self.description:
              usage = self.summary + '\n\n' + self.description
          else:
              usage = self.summary
          # Disable wrapping for all paragraphs
          return re.sub('\n\n', '\n\n\b\n', usage)
  
  
  class PluginIndex(type):
      """ Plugin metaclass used to register all available plugins """
  
      def __init__(cls, name, bases, attributes):
          """ Store all defined methods in the parent class """
          try:
              for method in cls._methods:
                  # Store reference to the implementing class
                  method.class_ = cls
                  # Add to the list of supported methods in parent class
                  bases[0]._supported_methods.append(method)
          except AttributeError:
              pass
  
  
  class Plugin(tmt.utils.Common, metaclass=PluginIndex):
      """ Common parent of all step plugins """
  
      # Default implementation for all steps is shell
      # except for provision (virtual) and report (display)
      how = 'shell'
  
      # Common keys for all plugins of given step
      _common_keys = []
  
      # Keys specific for given plugin
      _keys = []
  
      def __init__(self, step, data):
          """ Store plugin name, data and parent step """
  
          # Ensure that plugin data contains name
          if 'name' not in data:
              raise GeneralError(
                  f"Missing 'name' in the {step} step config "
                  f"of the '{step.plan}' plan.")
  
          # Store name, data and parent step
          super().__init__(parent=step, name=data['name'])
          self.data = data
          self.step = step
  
          # Initialize plugin order
          try:
              self.order = int(self.data['order'])
          except (ValueError, KeyError):
              self.order = tmt.utils.DEFAULT_PLUGIN_ORDER
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          # Include common options supported across all plugins
          return tmt.options.verbose_debug_quiet + tmt.options.force_dry
  
      @classmethod
      def command(cls):
          """ Prepare click command for all supported methods """
          # Create one command for each supported method
          commands = {}
          method_overview = f'Supported methods ({cls.how} by default):\n\n\b'
          for method in cls.methods():
              method_overview += f'\n{method.describe()}'
              command = cls.base_command(usage=method.usage())
              # Apply plugin specific options
              for option in method.class_.options(method.name):
                  command = option(command)
              commands[method.name] = command
  
          # Create base command with common options using method class
          method_class = tmt.options.create_method_class(commands)
          command = cls.base_command(method_class, usage=method_overview)
          # Apply common options
          for option in cls.options():
              command = option(command)
          return command
  
      @classmethod
      def methods(cls):
          """ Return all supported methods ordered by priority """
          return sorted(cls._supported_methods, key=lambda method: method.order)
  
      @classmethod
      def delegate(cls, step, data):
          """
          Return plugin instance implementing the data['how'] method
  
          Supports searching by method prefix as well (e.g. 'virtual').
          The first matching method with the lowest 'order' wins.
          """
          # Filter matching methods, pick the one with the lowest order
          for method in cls.methods():
              if method.name.startswith(data['how']):
                  step.debug(
                      f"Using the '{method.class_.__name__}' plugin "
                      f"for the '{data['how']}' method.", level=2)
                  return method.class_(step, data)
  
          show_step_method_hints(step, step.name, data['how'])
          # Report invalid method
          raise tmt.utils.SpecificationError(
              f"Unsupported {step.name} method '{data['how']}' "
              f"in the '{step.plan.name}' plan.")
  
      def default(self, option, default=None):
          """ Return default data for given option """
          return default
  
      def get(self, option, default=None):
          """ Get option from plugin data, user/system config or defaults """
          # Check plugin data first
          try:
              return self.data[option]
          except KeyError:
              pass
  
          # Check user config and system config
          # TODO
  
          # Finally check plugin defaults
          return self.default(option, default)
  
      def show(self, keys=None):
          """ Show plugin details for given or all available keys """
          # Show empty config with default method only in verbose mode
          if (set(self.data.keys()) == set(['how', 'name'])
                  and not self.opt('verbose')
                  and self.data['how'] == self.how):
              return
          # Step name (and optional summary)
          echo(tmt.utils.format(
              self.step, self.get('summary', ''),
              key_color='blue', value_color='blue'))
          # Show all or requested step attributes
          base_keys = ['name', 'how']
          if keys is None:
              keys = self._common_keys + self._keys
          for key in base_keys + keys:
              # Skip showing the default name
              if key == 'name' and self.name == tmt.utils.DEFAULT_NAME:
                  continue
              # Skip showing summary again
              if key == 'summary':
                  continue
              value = self.get(key)
              if value is not None:
                  echo(tmt.utils.format(key, value))
  
      def enabled_on_guest(self, guest):
          """ Check if the plugin is enabled on the specific guest """
          where = self.get('where')
          if not where:
              return True
          return where in (guest.name, guest.role)
  
      def wake(self, keys=None):
          """
          Wake up the plugin, process data, apply options
  
          Check command line options corresponding to plugin keys
          and store their value into the 'self.data' dictionary if
          their value is True or non-empty.
  
          By default, all supported options corresponding to common
          and plugin-specific keys are processed. List of key names
          in the 'keys' parameter can be used to override only
          selected ones.
          """
          if keys is None:
              keys = self._common_keys + self._keys
          for key in keys:
              value = self.opt(key)
              if value:
                  self.data[key] = value
  
      def go(self):
          """ Go and perform the plugin task """
          # Show the method
          self.info('how', self.get('how'), 'magenta')
          # Give summary if provided
          if self.get('summary'):
              self.info('summary', self.get('summary'), 'magenta')
          # Show name only if it's not the default one
          if self.name != tmt.utils.DEFAULT_NAME:
              self.info('name', self.name, 'magenta')
          # Include order in verbose mode
          self.verbose('order', self.order, 'magenta', level=3)
  
      def requires(self):
          """ List of packages required by the plugin on the guest """
          return []
  
  
  class Action(tmt.utils.Common):
      """ A special action performed during a normal step. """
  
      # Dictionary containing list of requested phases for each enabled step
      _phases = None
  
      @classmethod
      def phases(cls, step):
          """ Return list of phases enabled for given step """
          # Build the phase list unless done before
          if cls._phases is None:
              cls._phases = cls._parse_phases(step)
          # Return enabled phases, empty list if step not found
          try:
              return cls._phases[step.name]
          except KeyError:
              return []
  
      @classmethod
      def _parse_phases(cls, step):
          """ Parse options and store phase order """
          phases = dict()
          options = cls._opt('step', default=[])
  
          # Use the end of the last enabled step if no --step given
          if not options:
              login_during = None
              # The last run may have failed before all enabled steps were
              # completed, select the last step done
              if step.plan.my_run.opt('last'):
                  steps = [s for s in step.plan.steps() if s.status() == 'done']
                  login_during = steps[-1] if steps else None
              # Default to the last enabled step if no completed step found
              if login_during is None:
                  login_during = list(step.plan.steps())[-1]
              # Only login if the error occurred after provision
              if login_during != step.plan.discover:
                  phases[login_during.name] = [PHASE_END]
  
          # Process provided options
          for option in options:
              # Parse the step:phase format
              matched = re.match(r'(\w+)(:(\w+))?', option)
              if matched:
                  step_name, _, phase = matched.groups()
              if not matched or step_name not in STEPS:
                  raise tmt.utils.GeneralError(f"Invalid step '{option}'.")
              # Check phase format, convert into int, use end by default
              try:
                  phase = int(phase)
              except TypeError:
                  phase = PHASE_END
              except ValueError:
                  # Convert 'start' and 'end' aliases
                  try:
                      phase = dict(start=PHASE_START, end=PHASE_END)[phase]
                  except KeyError:
                      raise tmt.utils.GeneralError(f"Invalid phase '{phase}'.")
              # Store the phase for given step
              try:
                  phases[step_name].append(phase)
              except KeyError:
                  phases[step_name] = [phase]
          return phases
  
  
  class Reboot(Action):
      """ Reboot guest """
  
      # True if reboot enabled
      _enabled = False
  
      def __init__(self, step, order):
          """ Initialize relations, store the reboot order """
          super().__init__(parent=step, name='reboot')
          self.order = order
  
      @classmethod
      def command(cls, method_class=None, usage=None):
          """ Create the reboot command """
          @click.command()
          @click.pass_context
          @click.option(
              '-s', '--step', metavar='STEP[:PHASE]', multiple=True,
              help='Reboot machine during given phase of selected step(s).')
          @click.option(
              '--hard', is_flag=True,
              help='Hard reboot of the machine. Unsaved data may be lost.')
          def reboot(context, **kwargs):
              """ Reboot the guest. """
              Reboot._save_context(context)
              Reboot._enabled = True
  
          return reboot
  
      @classmethod
      def plugins(cls, step):
          """ Return list of reboot instances for given step """
          if not Reboot._enabled:
              return []
          return [Reboot(step, phase) for phase in cls.phases(step)]
  
      def go(self, *args, **kwargs):
          """ Reboot the guest(s) """
          self.info('reboot', 'Rebooting guest', color='yellow')
          for guest in self.parent.plan.provision.guests():
              guest.reboot(self.opt('hard'))
          self.info('reboot', 'Reboot finished', color='yellow')
  
  
  class Login(Action):
      """ Log into the guest """
  
      # True if interactive login enabled
      _enabled = False
  
      def __init__(self, step, order):
          """ Initialize relations, store the login order """
          super().__init__(parent=step, name='login')
          self.order = order
  
      @classmethod
      def command(cls, method_class=None, usage=None):
          """ Create the login command """
          @click.command()
          @click.pass_context
          @click.option(
              '-s', '--step', metavar='STEP[:PHASE]', multiple=True,
              help='Log in during given phase of selected step(s).')
          @click.option(
              '-w', '--when', metavar='RESULT', multiple=True,
              type=click.Choice(tmt.base.Result._results),
              help='Log in if a test finished with given result(s).')
          @click.option(
              '-c', '--command', metavar='COMMAND',
              multiple=True, default=['bash'],
              help="Run given command(s). Default is 'bash'.")
          def login(context, **kwargs):
              """
              Provide user with an interactive shell on the guest.
  
              By default the shell is provided at the end of the last
              enabled step. When used together with the --last option the
              last completed step is selected. Use one or more --step
              options to select a different step instead.
  
              Optional phase can be provided to specify the exact phase of
              the step when the shell should be provided. The following
              values are supported:
  
              \b
                  start ... beginning of the step (same as '10')
                  end ..... end of the step (default, same as '90')
                  00-99 ... integer order defining the exact phase
  
              Usually the main step execution happens with order 50.
              Consult individual step documentation for more details.
  
              For the execute step and following steps it is also possible
              to conditionally enable the login feature only if some of
              the tests finished with given result (pass, info, fail,
              warn, error).
              """
              Login._save_context(context)
              Login._enabled = True
  
          return login
  
      @classmethod
      def plugins(cls, step):
          """ Return list of login instances for given step """
          if not Login._enabled:
              return []
          return [Login(step, phase) for phase in cls.phases(step)]
  
      def go(self, *args, **kwargs):
          """ Login to the guest(s) """
          # Verify possible test result condition
          count = dict()
          expected_results = self.opt('when')
          if expected_results:
              for expected_result in expected_results:
                  count[expected_result] = len([
                      result for result in self.parent.plan.execute.results()
                      if result.result == expected_result])
              if not any(count.values()):
                  self.info('Skipping interactive shell', color='yellow')
                  return
  
          # Run the interactive command
          commands = self.opt('command')
          self.info('login', 'Starting interactive shell', color='yellow')
          for guest in self.parent.plan.provision.guests():
              # Attempt to push the workdir to the guest
              try:
                  guest.push()
                  cwd = self.parent.plan.worktree
              except tmt.utils.GeneralError:
                  self.warn("Failed to push workdir to the guest.")
                  cwd = None
              # Execute all requested commands
              for command in commands:
                  self.debug(f"Run '{command}' in interactive mode.")
                  guest.execute(command, interactive=True, cwd=cwd)
          self.info('login', 'Interactive shell finished', color='yellow')
  
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
  
          # Override step data with command line options
          how = self.opt('how')
          if how is not None:
              for data in self.data:
                  data['how'] = how
  
      def setup_actions(self):
          """ Insert login and reboot plugins if requested """
          for plugin in Login.plugins(step=self):
              self.debug(
                  f"Insert a login plugin into the '{self}' step "
                  f"with order '{plugin.order}'.", level=2)
              self._plugins.append(plugin)
  
          for plugin in Reboot.plugins(step=self):
              self.debug(
                  f"Insert a reboot plugin into the '{self}' step "
                  f"with order '{plugin.order}'.", level=2)
              self._plugins.append(plugin)
  
      def plugins(self, classes=None):
          """
          Iterate over plugins by their order
  
          By default iterates over all available plugins. Optional filter
          'classes' can be used to iterate only over instances of given
          class (single class or tuple of classes).
          """
          return sorted(
              [plugin for plugin in self._plugins
                  if classes is None or isinstance(plugin, classes)],
              key=lambda plugin: plugin.order)
  
      def actions(self):
          """ Run all loaded Login or Reboot plugin instances of the step """
          for plugin in self.plugins():
              if isinstance(plugin, (Reboot, Login)):
                  plugin.go()
  
      def go(self):
          """ Execute the test step """
          # Show step header and how
          self.info(self.name, color='blue')
          # Show workdir in verbose mode
          self.debug('workdir', self.workdir, 'magenta')
  
  
  class Method(object):
      """ Step implementation method """
  
      def __init__(self, name, doc, order):
          """ Store method data """
          self.name = name
          self.doc = doc.strip()
          self.order = order
  
          # Parse summary and description from provided doc string
          lines = [re.sub('^    ', '', line) for line in self.doc.split('\n')]
          self.summary = lines[0].strip()
          self.description = '\n'.join(lines[1:]).lstrip()
  
      def describe(self):
          """ Format name and summary for a nice method overview """
          return f'{self.name} '.ljust(22, '.') + f' {self.summary}'
  
      def usage(self):
          """ Prepare a detailed usage from summary and description """
          if self.description:
              usage = self.summary + '\n\n' + self.description
          else:
              usage = self.summary
          # Disable wrapping for all paragraphs
          return re.sub('\n\n', '\n\n\b\n', usage)
  
  
  class PluginIndex(type):
      """ Plugin metaclass used to register all available plugins """
  
      def __init__(cls, name, bases, attributes):
          """ Store all defined methods in the parent class """
          try:
              for method in cls._methods:
                  # Store reference to the implementing class
                  method.class_ = cls
                  # Add to the list of supported methods in parent class
                  bases[0]._supported_methods.append(method)
          except AttributeError:
              pass
  
  
  class Plugin(tmt.utils.Common, metaclass=PluginIndex):
      """ Common parent of all step plugins """
  
      # Default implementation for all steps is shell
      # except for provision (virtual) and report (display)
      how = 'shell'
  
      # Common keys for all plugins of given step
      _common_keys = []
  
      # Keys specific for given plugin
      _keys = []
  
      def __init__(self, step, data):
          """ Store plugin name, data and parent step """
  
          # Ensure that plugin data contains name
          if 'name' not in data:
              raise GeneralError(
                  f"Missing 'name' in the {step} step config "
                  f"of the '{step.plan}' plan.")
  
          # Store name, data and parent step
          super().__init__(parent=step, name=data['name'])
          self.data = data
          self.step = step
  
          # Initialize plugin order
          try:
              self.order = int(self.data['order'])
          except (ValueError, KeyError):
              self.order = tmt.utils.DEFAULT_PLUGIN_ORDER
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          # Include common options supported across all plugins
          return tmt.options.verbose_debug_quiet + tmt.options.force_dry
  
      @classmethod
      def command(cls):
          """ Prepare click command for all supported methods """
          # Create one command for each supported method
          commands = {}
          method_overview = f'Supported methods ({cls.how} by default):\n\n\b'
          for method in cls.methods():
              method_overview += f'\n{method.describe()}'
              command = cls.base_command(usage=method.usage())
              # Apply plugin specific options
              for option in method.class_.options(method.name):
                  command = option(command)
              commands[method.name] = command
  
          # Create base command with common options using method class
          method_class = tmt.options.create_method_class(commands)
          command = cls.base_command(method_class, usage=method_overview)
          # Apply common options
          for option in cls.options():
              command = option(command)
          return command
  
      @classmethod
      def methods(cls):
          """ Return all supported methods ordered by priority """
          return sorted(cls._supported_methods, key=lambda method: method.order)
  
      @classmethod
      def delegate(cls, step, data):
          """
          Return plugin instance implementing the data['how'] method
  
          Supports searching by method prefix as well (e.g. 'virtual').
          The first matching method with the lowest 'order' wins.
          """
          # Filter matching methods, pick the one with the lowest order
          for method in cls.methods():
              if method.name.startswith(data['how']):
                  step.debug(
                      f"Using the '{method.class_.__name__}' plugin "
                      f"for the '{data['how']}' method.", level=2)
                  return method.class_(step, data)
  
          show_step_method_hints(step, step.name, data['how'])
          # Report invalid method
          raise tmt.utils.SpecificationError(
              f"Unsupported {step.name} method '{data['how']}' "
              f"in the '{step.plan.name}' plan.")
  
      def default(self, option, default=None):
          """ Return default data for given option """
          return default
  
      def get(self, option, default=None):
          """ Get option from plugin data, user/system config or defaults """
          # Check plugin data first
          try:
              return self.data[option]
          except KeyError:
              pass
  
          # Check user config and system config
          # TODO
  
          # Finally check plugin defaults
          return self.default(option, default)
  
      def show(self, keys=None):
          """ Show plugin details for given or all available keys """
          # Show empty config with default method only in verbose mode
          if (set(self.data.keys()) == set(['how', 'name'])
                  and not self.opt('verbose')
                  and self.data['how'] == self.how):
              return
          # Step name (and optional summary)
          echo(tmt.utils.format(
              self.step, self.get('summary', ''),
              key_color='blue', value_color='blue'))
          # Show all or requested step attributes
          base_keys = ['name', 'how']
          if keys is None:
              keys = self._common_keys + self._keys
          for key in base_keys + keys:
              # Skip showing the default name
              if key == 'name' and self.name == tmt.utils.DEFAULT_NAME:
                  continue
              # Skip showing summary again
              if key == 'summary':
                  continue
              value = self.get(key)
              if value is not None:
                  echo(tmt.utils.format(key, value))
  
      def enabled_on_guest(self, guest):
          """ Check if the plugin is enabled on the specific guest """
          where = self.get('where')
          if not where:
              return True
          return where in (guest.name, guest.role)
  
      def wake(self, keys=None):
          """
          Wake up the plugin, process data, apply options
  
          Check command line options corresponding to plugin keys
          and store their value into the 'self.data' dictionary if
          their value is True or non-empty.
  
          By default, all supported options corresponding to common
          and plugin-specific keys are processed. List of key names
          in the 'keys' parameter can be used to override only
          selected ones.
          """
          if keys is None:
              keys = self._common_keys + self._keys
          for key in keys:
              value = self.opt(key)
              if value:
                  self.data[key] = value
  
      def go(self):
          """ Go and perform the plugin task """
          # Show the method
          self.info('how', self.get('how'), 'magenta')
          # Give summary if provided
          if self.get('summary'):
              self.info('summary', self.get('summary'), 'magenta')
          # Show name only if it's not the default one
          if self.name != tmt.utils.DEFAULT_NAME:
              self.info('name', self.name, 'magenta')
          # Include order in verbose mode
          self.verbose('order', self.order, 'magenta', level=3)
  
  
  class Action(tmt.utils.Common):
      """ A special action performed during a normal step. """
  
      # Dictionary containing list of requested phases for each enabled step
      _phases = None
  
      @classmethod
      def phases(cls, step):
          """ Return list of phases enabled for given step """
          # Build the phase list unless done before
          if cls._phases is None:
              cls._phases = cls._parse_phases(step)
          # Return enabled phases, empty list if step not found
          try:
              return cls._phases[step.name]
          except KeyError:
              return []
  
      @classmethod
      def _parse_phases(cls, step):
          """ Parse options and store phase order """
          phases = dict()
          options = cls._opt('step', default=[])
  
          # Use the end of the last enabled step if no --step given
          if not options:
              login_during = None
              # The last run may have failed before all enabled steps were
              # completed, select the last step done
              if step.plan.my_run.opt('last'):
                  steps = [s for s in step.plan.steps() if s.status() == 'done']
                  login_during = steps[-1] if steps else None
              # Default to the last enabled step if no completed step found
              if login_during is None:
                  login_during = list(step.plan.steps())[-1]
              # Only login if the error occurred after provision
              if login_during != step.plan.discover:
                  phases[login_during.name] = [PHASE_END]
  
          # Process provided options
          for option in options:
              # Parse the step:phase format
              matched = re.match(r'(\w+)(:(\w+))?', option)
              if matched:
                  step_name, _, phase = matched.groups()
              if not matched or step_name not in STEPS:
                  raise tmt.utils.GeneralError(f"Invalid step '{option}'.")
              # Check phase format, convert into int, use end by default
              try:
                  phase = int(phase)
              except TypeError:
                  phase = PHASE_END
              except ValueError:
                  # Convert 'start' and 'end' aliases
                  try:
                      phase = dict(start=PHASE_START, end=PHASE_END)[phase]
                  except KeyError:
                      raise tmt.utils.GeneralError(f"Invalid phase '{phase}'.")
              # Store the phase for given step
              try:
                  phases[step_name].append(phase)
              except KeyError:
                  phases[step_name] = [phase]
          return phases
  
  
  class Reboot(Action):
      """ Reboot guest """
  
      # True if reboot enabled
      _enabled = False
  
      def __init__(self, step, order):
          """ Initialize relations, store the reboot order """
          super().__init__(parent=step, name='reboot')
          self.order = order
  
      @classmethod
      def command(cls, method_class=None, usage=None):
          """ Create the reboot command """
          @click.command()
          @click.pass_context
          @click.option(
              '-s', '--step', metavar='STEP[:PHASE]', multiple=True,
              help='Reboot machine during given phase of selected step(s).')
          @click.option(
              '--hard', is_flag=True,
              help='Hard reboot of the machine. Unsaved data may be lost.')
          def reboot(context, **kwargs):
              """ Reboot the guest. """
              Reboot._save_context(context)
              Reboot._enabled = True
  
          return reboot
  
      @classmethod
      def plugins(cls, step):
          """ Return list of reboot instances for given step """
          if not Reboot._enabled:
              return []
          return [Reboot(step, phase) for phase in cls.phases(step)]
  
      def go(self, *args, **kwargs):
          """ Reboot the guest(s) """
          self.info('reboot', 'Rebooting guest', color='yellow')
          for guest in self.parent.plan.provision.guests():
              guest.reboot(self.opt('hard'))
          self.info('reboot', 'Reboot finished', color='yellow')
  
  
  class Login(Action):
      """ Log into the guest """
  
      # True if interactive login enabled
      _enabled = False
  
      def __init__(self, step, order):
          """ Initialize relations, store the login order """
          super().__init__(parent=step, name='login')
          self.order = order
  
      @classmethod
      def command(cls, method_class=None, usage=None):
          """ Create the login command """
          @click.command()
          @click.pass_context
          @click.option(
              '-s', '--step', metavar='STEP[:PHASE]', multiple=True,
              help='Log in during given phase of selected step(s).')
          @click.option(
              '-w', '--when', metavar='RESULT', multiple=True,
              type=click.Choice(tmt.base.Result._results),
              help='Log in if a test finished with given result(s).')
          @click.option(
              '-c', '--command', metavar='COMMAND',
              multiple=True, default=['bash'],
              help="Run given command(s). Default is 'bash'.")
          def login(context, **kwargs):
              """
              Provide user with an interactive shell on the guest.
  
              By default the shell is provided at the end of the last
              enabled step. When used together with the --last option the
              last completed step is selected. Use one or more --step
              options to select a different step instead.
  
              Optional phase can be provided to specify the exact phase of
              the step when the shell should be provided. The following
              values are supported:
  
              \b
                  start ... beginning of the step (same as '10')
                  end ..... end of the step (default, same as '90')
                  00-99 ... integer order defining the exact phase
  
              Usually the main step execution happens with order 50.
              Consult individual step documentation for more details.
  
              For the execute step and following steps it is also possible
              to conditionally enable the login feature only if some of
              the tests finished with given result (pass, info, fail,
              warn, error).
              """
              Login._save_context(context)
              Login._enabled = True
  
          return login
  
      @classmethod
      def plugins(cls, step):
          """ Return list of login instances for given step """
          if not Login._enabled:
              return []
          return [Login(step, phase) for phase in cls.phases(step)]
  
      def go(self, *args, **kwargs):
          """ Login to the guest(s) """
          # Verify possible test result condition
          count = dict()
          expected_results = self.opt('when')
          if expected_results:
              for expected_result in expected_results:
                  count[expected_result] = len([
                      result for result in self.parent.plan.execute.results()
                      if result.result == expected_result])
              if not any(count.values()):
                  self.info('Skipping interactive shell', color='yellow')
                  return
  
          # Run the interactive command
          commands = self.opt('command')
          self.info('login', 'Starting interactive shell', color='yellow')
          for guest in self.parent.plan.provision.guests():
              # Attempt to push the workdir to the guest
              try:
                  guest.push()
                  cwd = self.parent.plan.worktree
              except tmt.utils.GeneralError:
                  self.warn("Failed to push workdir to the guest.")
                  cwd = None
              # Execute all requested commands
              for command in commands:
                  self.debug(f"Run '{command}' in interactive mode.")
                  guest.execute(command, interactive=True, cwd=cwd)
          self.info('login', 'Interactive shell finished', color='yellow')
  
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
  
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
  
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(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.provision.is_multihost:
              self.info('guest', guest.name, 'green')
  
          # Show requested role if defined
          where = self.get('where')
          if where:
              self.info('where', where, 'green')
  
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.provision.is_multihost:
              self.info('guest', guest.name, 'green')
  
          # Show requested role if defined
          where = self.get('where')
          if where:
              self.info('where', where, 'green')
  
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 """
  
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 []
  
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, **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(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)
  
      def pull(self, source=None, destination=None, 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: 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, **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(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)
  
      def pull(self, source=None, destination=None, 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')
  
      @classmethod
      def requires(cls):
          """ No packages needed to sync workdir to the container """
          return []
  
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
  
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
  
files: tmt/steps/report/__init__.py