files: """ Test Management Tool """
  
  from tmt.base import Tree, Test, Plan, Story, Run, Result
  from tmt.steps.provision import Guest
  
  __all__ = ['Tree', 'Test', 'Plan', 'Story', 'Run', 'Guest', 'Result']
  
  
  [... the rest is omitted ...]
files: tmt/__init__.py
files: """ Test Management Tool """
  
  from tmt.base import Tree, Test, Plan, Story, Run, Guest, Result
  
  __all__ = ['Tree', 'Test', 'Plan', 'Story', 'Run', 'Guest', 'Result']
  
  
  [... the rest is omitted ...]
files: tmt/__init__.py
files: 
  """ Base Metadata Classes """
  
  import os
  import re
  import fmf
  import yaml
  import click
  import pprint
  import subprocess
  
  import tmt.steps
  import tmt.utils
  import tmt.export
  import tmt.templates
  
  import tmt.steps.discover
  import tmt.steps.provision
  import tmt.steps.prepare
  import tmt.steps.execute
  import tmt.steps.report
  import tmt.steps.finish
  
  from tmt.utils import verdict
  from fmf.utils import listed
  from click import echo, style
  
  DEFAULT_TEST_DURATION = '5m'
  
  class Node(tmt.utils.Common):
      """
      General node object
  
      Corresponds to given fmf.Tree node.
      Implements common Test, Plan and Story methods.
      """
  
      # Supported attributes
      _keys = ['name']
  
      def __init__(self, node, parent=None):
          """ Initialize the node """
          super(Node, self).__init__(parent=parent, name=node.name)
          self.node = node
  
      def __str__(self):
          """ Node name """
          return self.name
  
      def _sources(self):
          """ Show source files """
          echo(tmt.utils.format(
              'sources', self.node.sources, key_color='magenta'))
  
      def _check(self, key, expected, default=None, listify=False):
          """
          Check that the key is of expected type
  
          Handle default and convert into a list if requested.
          """
          value = getattr(self, key)
          # Handle default
          if value is None:
              setattr(self, key, default)
              return
          # Check for correct type
          if not isinstance(value, expected):
              expected = tmt.utils.listify(expected)
              expected_names = fmf.utils.listed(
                  [type_.__name__ for type_ in expected], join='or')
              class_name = self.__class__.__name__.lower()
              raise tmt.utils.SpecificationError(
                  f"Invalid '{key}' in {class_name} '{self.name}' (should be "
                  f"a '{expected_names}', got a '{type(value).__name__}').")
          # Convert into a list if requested
          if listify:
              setattr(self, key, tmt.utils.listify(value))
  
      def _fmf_id(self):
          """ Show fmf identifier """
          echo(tmt.utils.format('fmf-id', self.fmf_id, key_color='magenta'))
  
      @property
      def fmf_id(self):
          """ Return full fmf identifier of the node """
  
          def run(command):
              """ Run command, return output """
              result = subprocess.run(command.split(), capture_output=True)
              return result.stdout.strip().decode("utf-8")
  
          fmf_id = {'name': self.name}
  
          # Prepare url (for now handle just the most common schemas)
          origin = run('git config --get remote.origin.url')
          fmf_id['url'] = tmt.utils.public_git_url(origin)
  
          # Get the ref (skip for master as it is the default)
          ref = run('git rev-parse --abbrev-ref HEAD')
          if ref != 'master':
              fmf_id['ref'] = ref
  
          # Construct path (if different from git root)
          git_root = run('git rev-parse --show-toplevel')
          fmf_root = self.node.root
          if git_root != fmf_root:
              fmf_id['path'] = os.path.join(
                  '/', os.path.relpath(fmf_root, git_root))
  
          return fmf_id
  
      @classmethod
      def _save_context(cls, context):
          """ Save provided command line context for future use """
          super(Node, cls)._save_context(context)
  
          # Handle '.' as an alias for the current working directory
          names = cls._opt('names')
          if names is not None and '.' in names:
              root = context.obj.tree.root
              current = os.getcwd()
              # Handle special case when directly in the metadata root
              if current == root:
                  path = '/'
              # Prepare path from the tree root to the current directory
              else:
                  path = os.path.join('/', os.path.relpath(current, root))
              cls._context.params['names'] = (
                  path if name == '.' else name for name in names)
  
      def name_and_summary(self):
          """ Node name and optional summary """
          if self.summary:
              return '{0} ({1})'.format(self.name, self.summary)
          else:
              return self.name
  
      def ls(self, summary=False):
          """ List
  
  [... the rest is omitted ...]
files: tmt/base.py
files: 
  """ Base Metadata Classes """
  
  import os
  import re
  import fmf
  import yaml
  import click
  import pprint
  import random
  import string
  import subprocess
  
  import tmt.steps
  import tmt.utils
  import tmt.export
  import tmt.templates
  
  import tmt.steps.discover
  import tmt.steps.provision
  import tmt.steps.prepare
  import tmt.steps.execute
  import tmt.steps.report
  import tmt.steps.finish
  
  from tmt.utils import verdict
  from fmf.utils import listed
  from click import echo, style
  
  DEFAULT_TEST_DURATION = '5m'
  
  class Node(tmt.utils.Common):
      """
      General node object
  
      Corresponds to given fmf.Tree node.
      Implements common Test, Plan and Story methods.
      """
  
      # Supported attributes
      _keys = ['name']
  
      def __init__(self, node, parent=None):
          """ Initialize the node """
          super(Node, self).__init__(parent=parent, name=node.name)
          self.node = node
  
      def __str__(self):
          """ Node name """
          return self.name
  
      def _sources(self):
          """ Show source files """
          echo(tmt.utils.format(
              'sources', self.node.sources, key_color='magenta'))
  
      def _check(self, key, expected, default=None, listify=False):
          """
          Check that the key is of expected type
  
          Handle default and convert into a list if requested.
          """
          value = getattr(self, key)
          # Handle default
          if value is None:
              setattr(self, key, default)
              return
          # Check for correct type
          if not isinstance(value, expected):
              expected = tmt.utils.listify(expected)
              expected_names = fmf.utils.listed(
                  [type_.__name__ for type_ in expected], join='or')
              class_name = self.__class__.__name__.lower()
              raise tmt.utils.SpecificationError(
                  f"Invalid '{key}' in {class_name} '{self.name}' (should be "
                  f"a '{expected_names}', got a '{type(value).__name__}').")
          # Convert into a list if requested
          if listify:
              setattr(self, key, tmt.utils.listify(value))
  
      def _fmf_id(self):
          """ Show fmf identifier """
          echo(tmt.utils.format('fmf-id', self.fmf_id, key_color='magenta'))
  
      @property
      def fmf_id(self):
          """ Return full fmf identifier of the node """
  
          def run(command):
              """ Run command, return output """
              result = subprocess.run(command.split(), capture_output=True)
              return result.stdout.strip().decode("utf-8")
  
          fmf_id = {'name': self.name}
  
          # Prepare url (for now handle just the most common schemas)
          origin = run('git config --get remote.origin.url')
          fmf_id['url'] = tmt.utils.public_git_url(origin)
  
          # Get the ref (skip for master as it is the default)
          ref = run('git rev-parse --abbrev-ref HEAD')
          if ref != 'master':
              fmf_id['ref'] = ref
  
          # Construct path (if different from git root)
          git_root = run('git rev-parse --show-toplevel')
          fmf_root = self.node.root
          if git_root != fmf_root:
              fmf_id['path'] = os.path.join(
                  '/', os.path.relpath(fmf_root, git_root))
  
          return fmf_id
  
      @classmethod
      def _save_context(cls, context):
          """ Save provided command line context for future use """
          super(Node, cls)._save_context(context)
  
          # Handle '.' as an alias for the current working directory
          names = cls._opt('names')
          if names is not None and '.' in names:
              root = context.obj.tree.root
              current = os.getcwd()
              # Handle special case when directly in the metadata root
              if current == root:
                  path = '/'
              # Prepare path from the tree root to the current directory
              else:
                  path = os.path.join('/', os.path.relpath(current, root))
              cls._context.params['names'] = (
                  path if name == '.' else name for name in names)
  
      def name_and_summary(self):
          """ Node name and optional summary """
          if self.summary:
              return '{0} ({1})'.format(self.name, self.summary)
          else:
              return self.name
  
      def ls(self,
  
  [... the rest is omitted ...]
files: tmt/base.py
files: import os
  import re
  import click
  import random
  import string
  
  import fmf
  import tmt
  
  
  class Provision(tmt.steps.Step):
      """ Provision an environment for testing or use localhost """
  
      # Default implementation for provision is a virtual machine
      how = 'virtual'
  
      def __init__(self, data, plan):
          """ Initialize provision step data """
          super().__init__(data, plan)
          # List of provisioned guests and loaded guest data
          self._guests = []
          self._guest_data = {}
  
      def load(self):
          """ Load guest data from the workdir """
          super().load()
          try:
              self._guest_data = tmt.utils.yaml_to_dict(self.read('guests.yaml'))
          except tmt.utils.FileError:
              self.debug('Provisioned guests not found.', level=2)
  
      def save(self):
          """ Save guest data to the workdir """
          super().save()
          try:
              guests = dict(
                  [(guest.name, guest.save()) for guest in self.guests()])
              self.write('guests.yaml', tmt.utils.dict_to_yaml(guests))
          except tmt.utils.FileError:
              self.debug('Failed to save provisioned guests.')
  
      def wake(self):
          """ Wake up the step (process workdir and command line) """
          super().wake()
  
          # Choose the right plugin and wake it up
          for data in self.data:
              plugin = ProvisionPlugin.delegate(self, data)
              self._plugins.append(plugin)
              # If guest data loaded, perform a complete wake up
              plugin.wake(data=self._guest_data.get(plugin.name))
              if plugin.guest():
                  self._guests.append(plugin.guest())
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Provision wake up complete (already done before).', level=2)
          # Save status and step data (now we know what to do)
          else:
              self.status('todo')
              self.save()
  
      def show(self):
          """ Show discover details """
          for data in self.data:
              ProvisionPlugin.delegate(self, data).show()
  
      def summary(self):
          """ Give a concise summary of the provisioning """
          # Summary of provisioned guests
          guests = fmf.utils.listed(self.guests(), 'guest')
          self.info('summary', f'{guests} provisioned', 'green', shift=1)
          # Guest list in verbose mode
          for guest in self.guests():
              if guest.name != tmt.utils.DEFAULT_NAME:
                  self.verbose(guest.name, color='red', shift=2)
  
      def go(self):
          """ Provision all guests"""
          super().go()
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.info('status', 'done', 'green', shift=1)
              self.summary()
              return
  
          # Provision guests
          self._guests = []
          for plugin in self.plugins():
              plugin.go()
              if isinstance(plugin, ProvisionPlugin):
                  plugin.guest().details()
                  self._guests.append(plugin.guest())
  
          # Give a summary, update status and save
          self.summary()
          self.status('done')
          self.save()
  
      def guests(self):
          """ Return the list of all provisioned guests """
          return self._guests
  
  
  class ProvisionPlugin(tmt.steps.Plugin):
      """ Common parent of provision plugins """
  
      # Default implementation for provision is a virtual machine
      how = 'virtual'
  
      # List of all supported methods aggregated from all plugins
      _supported_methods = []
  
      @classmethod
      def base_command(cls, method_class=None, usage=None):
          """ Create base click command (common for all provision plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Provision.usage(method_overview=usage)
  
          # Create the command
          @click.command(cls=method_class, help=usage)
          @click.pass_context
          @click.option(
              '-h', '--how', metavar='METHOD',
              help='Use specified method for provisioning.')
          def provision(context, **kwargs):
              context.obj.steps.add('provision')
              Provision._save_context(context)
  
          return provision
  
      def wake(self, options=None, data=None):
          """
          Wake up the plugin
  
          Override data with command line options.
          Wake up the guest based on provided guest data.
          """
          super().wake(options)
  
      def guest(self):
          """
          Return provisioned guest
  
          Each ProvisionPlugin has to
  
  [... the rest is omitted ...]
files: tmt/steps/provision/__init__.py
files: import os
  import click
  
  import fmf
  import tmt
  
  class Provision(tmt.steps.Step):
      """ Provision an environment for testing or use localhost """
  
      # Default implementation for provision is a virtual machine
      how = 'virtual'
  
      def __init__(self, data, plan):
          """ Initialize provision step data """
          super().__init__(data, plan)
          # List of provisioned guests and loaded guest data
          self._guests = []
          self._guest_data = {}
  
      def load(self):
          """ Load guest data from the workdir """
          super().load()
          try:
              self._guest_data = tmt.utils.yaml_to_dict(self.read('guests.yaml'))
          except tmt.utils.FileError:
              self.debug('Provisioned guests not found.', level=2)
  
      def save(self):
          """ Save guest data to the workdir """
          super().save()
          try:
              guests = dict(
                  [(guest.name, guest.save()) for guest in self.guests()])
              self.write('guests.yaml', tmt.utils.dict_to_yaml(guests))
          except tmt.utils.FileError:
              self.debug('Failed to save provisioned guests.')
  
      def wake(self):
          """ Wake up the step (process workdir and command line) """
          super().wake()
  
          # Choose the right plugin and wake it up
          for data in self.data:
              plugin = ProvisionPlugin.delegate(self, data)
              self._plugins.append(plugin)
              # If guest data loaded, perform a complete wake up
              plugin.wake(data=self._guest_data.get(plugin.name))
              if plugin.guest():
                  self._guests.append(plugin.guest())
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Provision wake up complete (already done before).', level=2)
          # Save status and step data (now we know what to do)
          else:
              self.status('todo')
              self.save()
  
      def show(self):
          """ Show discover details """
          for data in self.data:
              ProvisionPlugin.delegate(self, data).show()
  
      def summary(self):
          """ Give a concise summary of the provisioning """
          # Summary of provisioned guests
          guests = fmf.utils.listed(self.guests(), 'guest')
          self.info('summary', f'{guests} provisioned', 'green', shift=1)
          # Guest list in verbose mode
          for guest in self.guests():
              if guest.name != tmt.utils.DEFAULT_NAME:
                  self.verbose(guest.name, color='red', shift=2)
  
      def go(self):
          """ Provision all guests"""
          super().go()
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.info('status', 'done', 'green', shift=1)
              self.summary()
              return
  
          # Provision guests
          self._guests = []
          for plugin in self.plugins():
              plugin.go()
              if isinstance(plugin, ProvisionPlugin):
                  plugin.guest().details()
                  self._guests.append(plugin.guest())
  
          # Give a summary, update status and save
          self.summary()
          self.status('done')
          self.save()
  
      def guests(self):
          """ Return the list of all provisioned guests """
          return self._guests
  
  
  class ProvisionPlugin(tmt.steps.Plugin):
      """ Common parent of provision plugins """
  
      # Default implementation for provision is a virtual machine
      how = 'virtual'
  
      # List of all supported methods aggregated from all plugins
      _supported_methods = []
  
      @classmethod
      def base_command(cls, method_class=None, usage=None):
          """ Create base click command (common for all provision plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Provision.usage(method_overview=usage)
  
          # Create the command
          @click.command(cls=method_class, help=usage)
          @click.pass_context
          @click.option(
              '-h', '--how', metavar='METHOD',
              help='Use specified method for provisioning.')
          def provision(context, **kwargs):
              context.obj.steps.add('provision')
              Provision._save_context(context)
  
          return provision
  
      def wake(self, options=None, data=None):
          """
          Wake up the plugin
  
          Override data with command line options.
          Wake up the guest based on provided guest data.
          """
          super().wake(options)
  
      def guest(self):
          """
          Return provisioned guest
  
          Each ProvisionPlugin has to implement this method.
          Should return a provision
  
  [... the rest is omitted ...]
files: tmt/steps/provision/__init__.py
files: import tmt
  import click
  
  class ProvisionConnect(tmt.steps.provision.ProvisionPlugin):
      """
      Connect to a provisioned guest using ssh
  
      Private key authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              key: /home/psss/.ssh/example_rsa
  
      Password authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              password: secret
  
      User defaults to 'root', so if you have private key correctly set
      the minimal configuration can look like this:
  
          provision:
              how: connect
              guest: host.example.org
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [tmt.steps.Method(name='connect', doc=__doc__, order=50)]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for connect """
          return [
              click.option(
                  '-g', '--guest', metavar='GUEST',
                  help='Select remote host to connect to (hostname or ip).'),
              click.option(
                  '-k', '--key', metavar='PRIVATE_KEY',
                  help='Private key for login into the guest system.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-p', '--password', metavar='PASSWORD',
                  help='Password for login into the guest system.'),
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          # User root as the default user
          if option == 'user':
              return 'root'
          # No other defaults available
          return default
  
      def show(self):
          """ Show provision details """
          super().show(['guest', 'key', 'user', 'password'])
  
      def wake(self, data=None):
          """ Override options and wake up the guest """
          super().wake(['guest', 'key', 'user', 'password'])
          if data:
              self._guest = tmt.Guest(data, name=self.name, parent=self.step)
  
      def go(self):
          """ Prepare the connection """
          super().go()
  
          # Check connection details
          guest = self.get('guest')
          user = self.get('user')
          key = self.get('key')
          password = self.get('password')
  
          # Check guest and auth info
          if not guest:
              raise tmt.utils.SpecificationError(
                  'Provide a host name or an ip address to connect.')
          data = dict(guest=guest, user=user)
          self.info('guest', guest, 'green')
          self.info('user', user, 'green')
  
          # Use provided password for authentication
          if password:
              self.info('password', password, 'green')
              self.debug('Using password authentication.')
              data['password'] = password
          # Default to using a private key (user can have configured one)
          else:
              self.info('key', key or 'not provided', 'green')
              self.debug('Using private key authentication.')
              data['key'] = key
  
          # And finally create the guest
          self._guest = tmt.Guest(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/connect.py
files: import tmt
  import click
  
  class ProvisionConnect(tmt.steps.provision.ProvisionPlugin):
      """
      Connect to a provisioned guest using ssh
  
      Private key authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              key: /home/psss/.ssh/example_rsa
  
      Password authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              password: secret
  
      User defaults to 'root', so if you have private key correctly set
      the minimal configuration can look like this:
  
          provision:
              how: connect
              guest: host.example.org
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [tmt.steps.Method(name='connect', doc=__doc__, order=50)]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for connect """
          return [
              click.option(
                  '-g', '--guest', metavar='GUEST',
                  help='Select remote host to connect to (hostname or ip).'),
              click.option(
                  '-k', '--key', metavar='PRIVATE_KEY',
                  help='Private key for login into the guest system.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-p', '--password', metavar='PASSWORD',
                  help='Password for login into the guest system.'),
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          # User root as the default user
          if option == 'user':
              return 'root'
          # No other defaults available
          return default
  
      def show(self):
          """ Show provision details """
          super().show(['guest', 'key', 'user', 'password'])
  
      def wake(self, data=None):
          """ Override options and wake up the guest """
          super().wake(['guest', 'key', 'user', 'password'])
          if data:
              self._guest = tmt.Guest(data, name=self.name, parent=self.step)
  
      def go(self):
          """ Prepare the connection """
          super().go()
  
          # Check connection details
          guest = self.get('guest')
          user = self.get('user')
          key = self.get('key')
          password = self.get('password')
  
          # Check guest and auth info
          if not guest:
              raise tmt.utils.SpecificationError(
                  'Provide a host name or an ip address to connect.')
          data = dict(guest=guest, user=user)
          self.info('guest', guest, 'green')
          self.info('user', user, 'green')
  
          # Use provided password for authentication
          if password:
              self.info('password', password, 'green')
              self.debug('Using password authentication.')
              data['password'] = password
          # Default to using a private key (user can have configured one)
          else:
              self.info('key', key or 'not provided', 'green')
              self.debug('Using private key authentication.')
              data['key'] = key
  
          # And finally create the guest
          self._guest = tmt.base.Guest(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/connect.py