files: import os
  
  import click
  from fmf.utils import listed
  
  import tmt
  import tmt.utils
  
  
  class Discover(tmt.steps.Step):
      """ Gather information about test cases to be executed. """
  
      def __init__(self, plan, data):
          """ Store supported attributes, check for sanity """
          super().__init__(plan=plan, data=data)
  
          # List of Test() objects representing discovered tests
          self._tests = []
  
      def load(self, extra_keys=None):
          """ Load step data from the workdir """
          extra_keys = extra_keys or []
          super().load(extra_keys)
          try:
              tests = tmt.utils.yaml_to_dict(self.read('tests.yaml'))
              self._tests = [
                  tmt.Test(data, name) for name, data in tests.items()]
          except tmt.utils.FileError:
              self.debug('Discovered tests not found.', level=2)
  
      def save(self, data=None):
          """ Save step data to the workdir """
          data = data or {}
          super().save(data)
  
          # Create tests.yaml with the full test data
          tests = dict([
              (test.name, test.export(format_='dict'))
              for test in self.tests()])
          self.write('tests.yaml', tmt.utils.dict_to_yaml(tests))
  
          # Create 'run.yaml' with the list of tests for the executor
          if not self.tests():
              return
          tests = dict([
              (test.name, test.export(format_='execute'))
              for test in self.tests()])
          self.write('run.yaml', tmt.utils.dict_to_yaml(tests, width=1000000))
  
      def _discover_from_execute(self):
          """ Check the execute step for possible shell script tests """
  
          # Check scripts for command line and data, convert to list if needed
          scripts = self.plan.execute.opt('script')
          if not scripts:
              scripts = self.plan.execute.data[0].get('script')
          if not scripts:
              return
          if isinstance(scripts, str):
              scripts = [scripts]
  
          # Give a warning when discover step defined as well
          default_data = [{'name': 'default', 'how': 'shell'}]
          if self.data and self.data != default_data:
              raise tmt.utils.DiscoverError(
                  "Use either 'discover' or 'execute' step "
                  "to define tests, but not both.")
  
          # Check the execute step for possible custom duration limit
          duration = self.plan.execute.data[0].get(
              'duration', tmt.base.DEFAULT_TEST_DURATION_L2)
  
          # Prepare the list of tests
          tests = []
          for index, script in enumerate(scripts):
              name = f'script-{str(index).zfill(2)}'
              tests.append(dict(name=name, test=script, duration=duration))
  
          # Append new data if tests already defined
          if self.data[0].get('tests'):
              self.data.append(
                  dict(how='shell', tests=tests, name='execute'))
          # Otherwise override current empty definition
          else:
              self.data[0]['tests'] = tests
  
      def wake(self):
          """ Wake up the step (process workdir and command line) """
          super().wake()
  
          # Check execute step for possible tests (unless already done)
          if self.status() is None:
              self._discover_from_execute()
  
          # Choose the right plugin and wake it up
          for data in self.data:
              plugin = DiscoverPlugin.delegate(self, data)
              self._phases.append(plugin)
              plugin.wake()
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Discover 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:
              DiscoverPlugin.delegate(self, data).show()
  
      def summary(self):
          """ Give a concise summary of the discovery """
          # Summary of selected tests
          text = listed(len(self.tests()), 'test') + ' selected'
          self.info('summary', text, 'green', shift=1)
          # Test list in verbose mode
          for test in self.tests():
              self.verbose(test.name, color='red', shift=2)
  
      def go(self):
          """ Execute all steps """
          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 test discovery, gather discovered tests
          self._tests = []
          for phase in self.phases():
              # Go and discover tests
              phase.go()
              # Nothing more to be done for other plugins
              if not isinstance(phase, DiscoverPlugin):
                  continue
              # Prefix test name only if multiple plugins configured
              prefix = f'/{phase.name}' if len(self.phases()) > 1 else ''
              # Check discovered tests, modify test name/path
              for test in phase.tests():
                  test.name = f"{prefix}{test.name}"
                  test.path = f"/{phase.name}{test.path}"
                  # Use the default test framework if not defined in L1
                  # FIXME remove when we drop the old execution methods
                  if not test.framework:
                      test.framework = self.plan.execute._framework
                  # Update test environment with plan environment
                  test.environment.update(self.plan.environment)
                  self._tests.append(test)
  
          # Show fmf identifiers for tests discovered in plan
          # TODO: This part should go into the 'fmf.py' module
          if self.opt('fmf_id'):
              if self.tests():
                  fmf_id_list = [tmt.utils.dict_to_yaml(test.fmf_id, start=True)
                                 for test in self.tests()
                                 if 'url' in test.fmf_id]
                  click.echo(''.join(fmf_id_list), nl=False)
              return
  
          # Give a summary, update status and save
          self.summary()
          self.status('done')
          self.save()
  
      def tests(self):
          """ Return the list of all enabled tests """
          return [test for test in self._tests if test.enabled]
  
      def requires(self):
          """ Return all tests' requires """
          requires = set()
          for test in self.tests():
              for value in getattr(test, 'require', []):
                  requires.add(value)
          return list(requires)
  
      def recommends(self):
          """ Return all packages recommended by tests """
          recommends = set()
          for test in self.tests():
              for value in getattr(test, 'recommend', []):
                  recommends.add(value)
          return list(recommends)
  
  
  class DiscoverPlugin(tmt.steps.Plugin):
      """ Common parent of discover plugins """
  
      # List of all supported methods aggregated from all plugins
      _supported_methods = []
  
      # Common keys for all discover step implementations
      _common_keys = [
          "dist-git-source",
          "dist-git-type",
          ]
  
      @classmethod
      def base_command(cls, method_class=None, usage=None):
          """ Create base click command (common for all discover plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Discover.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 to discover tests.')
          def discover(context, **kwargs):
              # TODO: This part should go into the 'fmf.py' module
              if kwargs.get('fmf_id'):
                  # Set quiet, disable debug and verbose to avoid logging
                  # to terminal with discover --fmf-id
                  context.parent.params['quiet'] = True
                  context.parent.params['debug'] = 0
                  context.parent.params['verbose'] = 0
              context.obj.steps.add('discover')
              Discover._save_context(context)
  
          return discover
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          return [
              click.option(
                  '--dist-git-source',
                  is_flag=True,
                  help='Extract DistGit sources.'),
              click.option(
                  '--dist-git-type',
                  type=click.Choice(tmt.utils.get_distgit_handler_names()),
                  help='Use the provided DistGit handler '
                       'instead of the auto detection.'),
              ] + super().options(how)
  
      def tests(self):
          """
          Return discovered tests
  
          Each DiscoverPlugin has to implement this method.
          Should return a list of Test() objects.
          """
          raise NotImplementedError
  
      def extract_distgit_source(
              self, distgit_dir, target_dir, handler_name=None):
          """
          Extract source tarball into target_dir
  
          distgit_dir is path to the DistGit repository.
          Source tarball is discovered from the 'sources' file content.
          """
          if handler_name is None:
              stdout, _ = self.run(
                  ["git", "config", "--get-regexp", '^remote\\..*.url'],
                  cwd=distgit_dir)
              remotes = stdout.split('\n')
              handler = tmt.utils.get_distgit_handler(remotes=remotes)
          else:
              handler = tmt.utils.get_distgit_handler(usage_name=handler_name)
          for url, source_name in handler.url_and_name(distgit_dir):
              if handler.re_ignore_extensions.search(source_name):
                  continue
              self.debug(f"Download sources from '{url}'.")
              with tmt.utils.retry_session() as session:
                  response = session.get(url)
              response.raise_for_status()
              os.makedirs(target_dir, exist_ok=True)
              with open(os.path.join(target_dir, source_name), 'wb') as tarball:
                  tarball.write(response.content)
              self.run(
                  ["tar", "--auto-compress", "--extract", "-f", source_name],
                  cwd=target_dir)
  
  
  [... the rest is omitted ...]
files: tmt/steps/discover/__init__.py
files: import os
  
  import click
  from fmf.utils import listed
  
  import tmt
  import tmt.utils
  
  
  class Discover(tmt.steps.Step):
      """ Gather information about test cases to be executed. """
  
      def __init__(self, plan, data):
          """ Store supported attributes, check for sanity """
          super().__init__(plan=plan, data=data)
  
          # List of Test() objects representing discovered tests
          self._tests = []
  
      def load(self, extra_keys=None):
          """ Load step data from the workdir """
          extra_keys = extra_keys or []
          super().load(extra_keys)
          try:
              tests = tmt.utils.yaml_to_dict(self.read('tests.yaml'))
              self._tests = [
                  tmt.Test(data, name) for name, data in tests.items()]
          except tmt.utils.FileError:
              self.debug('Discovered tests not found.', level=2)
  
      def save(self, data=None):
          """ Save step data to the workdir """
          data = data or {}
          super().save(data)
  
          # Create tests.yaml with the full test data
          tests = dict([
              (test.name, test.export(format_='dict'))
              for test in self.tests()])
          self.write('tests.yaml', tmt.utils.dict_to_yaml(tests))
  
          # Create 'run.yaml' with the list of tests for the executor
          if not self.tests():
              return
          tests = dict([
              (test.name, test.export(format_='execute'))
              for test in self.tests()])
          self.write('run.yaml', tmt.utils.dict_to_yaml(tests, width=1000000))
  
      def _discover_from_execute(self):
          """ Check the execute step for possible shell script tests """
  
          # Check scripts for command line and data, convert to list if needed
          scripts = self.plan.execute.opt('script')
          if not scripts:
              scripts = self.plan.execute.data[0].get('script')
          if not scripts:
              return
          if isinstance(scripts, str):
              scripts = [scripts]
  
          # Give a warning when discover step defined as well
          default_data = [{'name': 'default', 'how': 'shell'}]
          if self.data and self.data != default_data:
              raise tmt.utils.DiscoverError(
                  "Use either 'discover' or 'execute' step "
                  "to define tests, but not both.")
  
          # Check the execute step for possible custom duration limit
          duration = self.plan.execute.data[0].get(
              'duration', tmt.base.DEFAULT_TEST_DURATION_L2)
  
          # Prepare the list of tests
          tests = []
          for index, script in enumerate(scripts):
              name = f'script-{str(index).zfill(2)}'
              tests.append(dict(name=name, test=script, duration=duration))
  
          # Append new data if tests already defined
          if self.data[0].get('tests'):
              self.data.append(
                  dict(how='shell', tests=tests, name='execute'))
          # Otherwise override current empty definition
          else:
              self.data[0]['tests'] = tests
  
      def wake(self):
          """ Wake up the step (process workdir and command line) """
          super().wake()
  
          # Check execute step for possible tests (unless already done)
          if self.status() is None:
              self._discover_from_execute()
  
          # Choose the right plugin and wake it up
          for data in self.data:
              plugin = DiscoverPlugin.delegate(self, data)
              self._phases.append(plugin)
              plugin.wake()
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Discover 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:
              DiscoverPlugin.delegate(self, data).show()
  
      def summary(self):
          """ Give a concise summary of the discovery """
          # Summary of selected tests
          text = listed(len(self.tests()), 'test') + ' selected'
          self.info('summary', text, 'green', shift=1)
          # Test list in verbose mode
          for test in self.tests():
              self.verbose(test.name, color='red', shift=2)
  
      def go(self):
          """ Execute all steps """
          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 test discovery, gather discovered tests
          self._tests = []
          for phase in self.phases():
              # Go and discover tests
              phase.go()
              # Nothing more to be done for other plugins
              if not isinstance(phase, DiscoverPlugin):
                  continue
              # Prefix test name only if multiple plugins configured
              prefix = f'/{phase.name}' if len(self.phases()) > 1 else ''
              # Check discovered tests, modify test name/path
              for test in phase.tests():
                  test.name = f"{prefix}{test.name}"
                  test.path = f"/{phase.name}{test.path}"
                  # Use the default test framework if not defined in L1
                  # FIXME remove when we drop the old execution methods
                  if not test.framework:
                      test.framework = self.plan.execute._framework
                  # Update test environment with plan environment
                  test.environment.update(self.plan.environment)
                  self._tests.append(test)
  
          # Show fmf identifiers for tests discovered in plan
          # TODO: This part should go into the 'fmf.py' module
          if self.opt('fmf_id'):
              if self.tests():
                  fmf_id_list = [tmt.utils.dict_to_yaml(test.fmf_id, start=True)
                                 for test in self.tests()
                                 if 'url' in test.fmf_id]
                  click.echo(''.join(fmf_id_list), nl=False)
              return
  
          # Give a summary, update status and save
          self.summary()
          self.status('done')
          self.save()
  
      def tests(self):
          """ Return the list of all enabled tests """
          return [test for test in self._tests if test.enabled]
  
      def requires(self):
          """ Return all tests' requires """
          requires = set()
          for test in self.tests():
              for value in getattr(test, 'require', []):
                  requires.add(value)
          return list(requires)
  
      def recommends(self):
          """ Return all packages recommended by tests """
          recommends = set()
          for test in self.tests():
              for value in getattr(test, 'recommend', []):
                  recommends.add(value)
          return list(recommends)
  
  
  class DiscoverPlugin(tmt.steps.Plugin):
      """ Common parent of discover plugins """
  
      # List of all supported methods aggregated from all plugins
      _supported_methods = []
  
      # Common keys for all discover step implementations
      _common_keys = [
          "dist-git-source",
          "dist-git-type",
          ]
  
      @classmethod
      def base_command(cls, method_class=None, usage=None):
          """ Create base click command (common for all discover plugins) """
  
          # Prepare general usage message for the step
          if method_class:
              usage = Discover.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 to discover tests.')
          def discover(context, **kwargs):
              # TODO: This part should go into the 'fmf.py' module
              if kwargs.get('fmf_id'):
                  # Set quiet, disable debug and verbose to avoid logging
                  # to terminal with discover --fmf-id
                  context.parent.params['quiet'] = True
                  context.parent.params['debug'] = 0
                  context.parent.params['verbose'] = 0
              context.obj.steps.add('discover')
              Discover._save_context(context)
  
          return discover
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          return [
              click.option(
                  '--dist-git-source',
                  is_flag=True,
                  help='Extract DistGit sources.'),
              click.option(
                  '--dist-git-type',
                  type=click.Choice(tmt.utils.get_distgit_handler_names()),
                  help='Use the provided DistGit handler '
                       'instead of the auto detection.'),
              ] + super().options(how)
  
      def tests(self):
          """
          Return discovered tests
  
          Each DiscoverPlugin has to implement this method.
          Should return a list of Test() objects.
          """
          raise NotImplementedError
  
      def extract_distgit_source(
              self, distgit_dir, target_dir, handler_name=None):
          """
          Extract source tarball into target_dir
  
          distgit_dir is path to the DistGit repository.
          Source tarball is discovered from the 'sources' file content.
          """
          if handler_name is None:
              stdout, _ = self.run(
                  ["git", "config", "--get-regexp", '^remote\\..*.url'],
                  cwd=distgit_dir)
              remotes = stdout.split('\n')
              handler = tmt.utils.get_distgit_handler(remotes=remotes)
          else:
              handler = tmt.utils.get_distgit_handler(usage_name=handler_name)
          for url, source_name in handler.url_and_name(distgit_dir):
              if handler.re_ignore_extensions.search(source_name):
                  continue
              self.debug(f"Download sources from '{url}'.")
              session = tmt.utils.retry_session()
              response = session.get(url)
              response.raise_for_status()
              os.makedirs(target_dir, exist_ok=True)
              with open(os.path.join(target_dir, source_name), 'wb') as tarball:
                  tarball.write(response.content)
              self.run(
                  ["tar", "--auto-compress", "--extract", "-f", source_name],
                  cwd=target_dir)
  
  
  [... the rest is omitted ...]
files: tmt/steps/discover/__init__.py
files: import datetime
  import sys
  import time
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import requests
  
  import tmt
  import tmt.steps.provision
  from tmt.utils import ProvisionError, retry_session, updatable_message
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  
  # List of Artemis API versions supported and understood by this plugin.
  # Since API gains support for new features over time, it is important to
  # know when particular feature became available, and avoid using it with
  # older APIs.
  SUPPORTED_API_VERSIONS = (
      # NEW: added hostname HW constraint
      'v0.0.38',
      # NEW: virtualization HW constraint
      'v0.0.37',
      # NEW: boot.method HW constraint
      'v0.0.32',
      # NEW: network HW constraint
      'v0.0.28'
      )
  
  
  # The default Artemis API version - the most recent supported versions
  # should be perfectly fine.
  DEFAULT_API_VERSION = SUPPORTED_API_VERSIONS[0]
  
  # Type annotation for "data" package describing a guest instance. Passed
  # between load() and save() calls.
  StepStateType = TypedDict(
      'StepStateType',
      {
          # API
          'api-url': str,
          'api-version': str,
  
          # Guest request properties
          'arch': str,
          'image': str,
          'hardware': Any,
          'pool': Optional[str],
          'priority-group': str,
          'keyname': str,
          'user-data': Dict[str, str],
  
          # Provided by Artemis response
          'guestname': Optional[str],
          'guest': Optional[str],
          'user': str,
  
          # Timeouts and deadlines
          'provision-timeout': int,
          'provision-tick': int,
          'api-timeout': int,
          'api-retries': int,
          'api-retry-backoff-factor': int
          }
      )
  
  DEFAULT_USER = 'root'
  DEFAULT_ARCH = 'x86_64'
  DEFAULT_PRIORITY_GROUP = 'default-priority'
  DEFAULT_KEYNAME = 'default'
  DEFAULT_PROVISION_TIMEOUT = 600
  DEFAULT_PROVISION_TICK = 60
  DEFAULT_API_TIMEOUT = 10
  DEFAULT_API_RETRIES = 10
  # Should lead to delays of 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256 seconds
  DEFAULT_RETRY_BACKOFF_FACTOR = 1
  
  DEFAULT_GUEST_DATA = cast(
      StepStateType, {
          'api-version': DEFAULT_API_VERSION,
          'arch': DEFAULT_ARCH,
          'priority-group': DEFAULT_PRIORITY_GROUP,
          'keyname': DEFAULT_KEYNAME,
          'user-data': {},
          'user': DEFAULT_USER,
          'provision-timeout': DEFAULT_PROVISION_TIMEOUT,
          'provision-tick': DEFAULT_PROVISION_TICK,
          'api-timeout': DEFAULT_API_TIMEOUT,
          'api-retries': DEFAULT_API_RETRIES,
          'api-retry-backoff-factor': DEFAULT_RETRY_BACKOFF_FACTOR
          }
      )
  
  GUEST_STATE_COLOR_DEFAULT = 'green'
  
  GUEST_STATE_COLORS = {
      'routing': 'yellow',
      'provisioning': 'magenta',
      'promised': 'blue',
      'preparing': 'cyan',
      'cancelled': 'red',
      'error': 'red'
  }
  
  
  # Type annotation for Artemis API `GET /guests/$guestname` response.
  # Partial, not all fields necessary since plugin ignores most of them.
  GuestInspectType = TypedDict(
      'GuestInspectType', {
          'state': str,
          'address': Optional[str]
          }
      )
  
  
  class ArtemisAPI:
      def __init__(self, guest: 'GuestArtemis') -> None:
          self._guest = guest
  
          self.http_session = retry_session.create(
              retries=guest.api_retries,
              backoff_factor=guest.api_retry_backoff_factor,
              allowed_methods=('HEAD', 'GET', 'POST', 'DELETE', 'PUT'),
              status_forcelist=(
                  429,  # Too Many Requests
                  500,  # Internal Server Error
                  502,  # Bad Gateway
                  503,  # Service Unavailable
                  504   # Gateway Timeout
                  ),
              timeout=guest.api_timeout
              )
  
      def query(
              self,
              path: str,
              method: str = 'get',
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Base helper for Artemis API queries.
  
          Trivial dispatcher per method, returning retrieved response.
  
          :param path: API path to contact.
          :param method: HTTP method to use.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          url = f'{self._guest.api_url}{path}'
  
          if method == 'get':
              return self.http_session.get(url, **request_kwargs)
  
          if method == 'post':
              return self.http_session.post(url, **request_kwargs)
  
          if method == 'delete':
              return self.http_session.delete(url, **request_kwargs)
  
          if method == 'put':
              return self.http_session.put(url, **request_kwargs)
  
          raise tmt.utils.GeneralError(
              f'Unsupported Artemis API method.\n{method}')
  
      def create(
              self,
              path: str,
              data: Dict[str, Any],
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Create - or request creation of - a resource.
  
          :param path: API path to contact.
          :param data: optional key/value data to send with the request.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
          request_kwargs['json'] = data
  
          return self.query(path, method='post', request_kwargs=request_kwargs)
  
      def inspect(
              self,
              path: str,
              params: Optional[Dict[str, Any]] = None,
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Inspect a resource.
  
          :param path: API path to contact.
          :param params: optional key/value query parameters.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          if params:
              request_kwargs['params'] = params
  
          return self.query(path, request_kwargs=request_kwargs)
  
      def delete(
              self,
              path: str,
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Delete - or request removal of - a resource.
  
          :param path: API path to contact.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          return self.query(path, method='delete', request_kwargs=request_kwargs)
  
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  class ProvisionArtemis(
          tmt.steps.provision.ProvisionPlugin):  # type: ignore[misc]
      """
      Provision guest using Artemis backend
  
      Minimal configuration could look like this:
  
          provision:
              how: artemis
              image: Fedora
              api-url: https://your-artemis.com/
  
      Note that the actual value of "image" depends on what images - or
      "composes" as Artemis calls them - supports and can deliver.
  
      Note that "api-url" can be also given via ARTEMIS_API_URL
      environment variable.
  
      Full configuration example:
  
          provision:
              how: artemis
  
              # Artemis API
              api-url: https://your-artemis.com/
              api-version: 0.0.32
  
              # Mandatory environment properties
              image: Fedora
  
              # Optional environment properties
              arch: aarch64
              pool: optional-pool-name
  
              # Provisioning process control (optional)
              priority-group: custom-priority-group
              keyname: custom-SSH-key-name
  
              # Labels to be attached to guest request (optional)
              user-data:
                  foo: bar
  
              # Timeouts and deadlines (optional)
              provision-timeout: 3600
              provision-tick: 10
              api-timeout: 600
              api-retries: 5
              api-retry-backoff-factor: 1
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [
          tmt.steps.Method(name='artemis', doc=__doc__, order=50),
          ]
  
      _keys = [
          'api-url',
          'api-version',
          'arch',
          'image',
          'hardware',
          'pool',
          'priority-group',
          'keyname',
          'user-data',
          'provision-timeout',
          'provision-tick',
          'api-timeout',
          'api-retries',
          'api-retry-backoff-factor']
  
      # TODO: fix types once superclass gains its annotations
      @classmethod
      def options(cls, how: Any = None) -> List[click.Option]:
          """ Prepare command line options for Artemis """
          return cast(List[click.Option], [
              click.option(
                  '--api-url', metavar='URL',
                  help="Artemis API URL.",
                  envvar='ARTEMIS_API_URL'
                  ),
              click.option(
                  '--api-version', metavar='x.y.z',
                  help="Artemis API version to use.",
                  type=click.Choice(SUPPORTED_API_VERSIONS),
                  envvar='ARTEMIS_API_VERSION'
                  ),
              click.option(
                  '--arch', metavar='ARCH',
                  help='Architecture to provision.'
                  ),
              click.option(
                  '--image', metavar='COMPOSE',
                  help='Image (or "compose" in Artemis terminology) '
                       'to provision.'
                  ),
              click.option(
                  '--pool', metavar='NAME',
                  help='Pool to enforce.'
                  ),
              click.option(
                  '--priority-group', metavar='NAME',
                  help='Provisioning priority group.'
                  ),
              click.option(
                  '--keyname', metavar='NAME',
                  help='SSH key name.'
                  ),
              click.option(
                  '--user-data', metavar='KEY=VALUE',
                  help='Optional data to attach to guest.',
                  multiple=True,
                  default=[]
                  ),
              click.option(
                  '--provision-timeout', metavar='SECONDS',
                  help=f'How long to wait for provisioning to complete, '
                       f'{DEFAULT_PROVISION_TIMEOUT} seconds by default.'
                  ),
              click.option(
                  '--provision-tick', metavar='SECONDS',
                  help=f'How often check Artemis API for provisioning status, '
                       f'{DEFAULT_PROVISION_TICK} seconds by default.',
                  ),
              click.option(
                  '--api-timeout', metavar='SECONDS',
                  help=f'How long to wait for API operations to complete, '
                       f'{DEFAULT_API_TIMEOUT} seconds by default.',
                  ),
              click.option(
                  '--api-retries', metavar='COUNT',
                  help=f'How many attempts to use when talking to API, '
                       f'{DEFAULT_API_RETRIES} by default.',
                  ),
              click.option(
                  '--api-retry-backoff-factor', metavar='COUNT',
                  help=f'A factor for exponential API retry backoff, '
                       f'{DEFAULT_RETRY_BACKOFF_FACTOR} by default.',
                  ),
              ]) + cast(List[click.Option], super().options(how))
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
  
          return DEFAULT_GUEST_DATA.get(option, default)
  
      # TODO: use better types once superclass gains its annotations
      def wake(self, keys: Optional[List[str]] = None,
               data: Optional[StepStateType] = None) -> None:
          """ Wake up the plugin, process data, apply options """
  
          super().wake(keys=keys, data=data)
  
          if data:
              self._guest = GuestArtemis(data, name=self.name, parent=self.step)
  
      def go(self) -> None:
          """ Provision the guest """
          super().go()
  
          api_version = self.get('api-version')
  
          if api_version not in SUPPORTED_API_VERSIONS:
              raise ProvisionError(f"API version '{api_version}' not supported.")
  
          try:
              user_data = {
                  key.strip(): value.strip()
                  for key, value in (
                      pair.split('=', 1)
                      for pair in self.get('user-data')
                      )
                  }
  
          except ValueError:
              raise ProvisionError('Cannot parse user-data.')
  
          data: StepStateType = {
              'api-url': self.get('api-url'),
              'api-version': api_version,
              'arch': self.get('arch'),
              'image': self.get('image'),
              'hardware': self.get('hardware'),
              'pool': self.get('pool'),
              'priority-group': self.get('priority-group'),
              'keyname': self.get('keyname'),
              'user-data': user_data,
              'guestname': None,
              'guest': None,
              'user': DEFAULT_USER,
              'provision-timeout': self.get('provision-timeout'),
              'provision-tick': self.get('provision-tick'),
              'api-timeout': self.get('api-timeout'),
              'api-retries': self.get('api-retries'),
              'api-retry-backoff-factor': self.get('api-retry-backoff-factor')
              }
  
          self._guest = GuestArtemis(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self) -> Optional['GuestArtemis']:
          """ Return the provisioned guest """
          return self._guest
  
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  class GuestArtemis(tmt.GuestSsh):  # type: ignore[misc]
      """
      Artemis guest instance
  
      The following keys are expected in the 'data' dictionary:
      """
  
      _api: Optional[ArtemisAPI] = None
  
      @property
      def api(self) -> ArtemisAPI:
          if self._api is None:
              self._api = ArtemisAPI(self)
  
          return self._api
  
      def load(self, data: StepStateType) -> None:
          super().load(data)
  
          self.api_url = data['api-url']
          self.api_version = data['api-version']
          self.arch = data['arch']
          self.image = data['image']
          self.hardware = data['hardware']
          self.pool = data['pool']
          self.priority_group = data['priority-group']
          self.keyname = data['keyname']
          self.user_data = data['user-data']
          self.guestname = data['guestname']
          self.guest = data['guest']
          self.user = data['user']
          self.provision_timeout = data['provision-timeout']
          self.provision_tick = data['provision-tick']
          self.api_timeout = data['api-timeout']
          self.api_retries = data['api-retries']
          self.api_retry_backoff_factor = data['api-retry-backoff-factor']
  
      def save(self) -> StepStateType:
          data = cast(StepStateType, super().save())
  
          data['api-url'] = self.api_url
          data['api-version'] = self.api_version
          data['arch'] = self.arch
          data['image'] = self.image
          data['hardware'] = self.hardware
          data['pool'] = self.pool
          data['priority-group'] = self.priority_group
          data['keyname'] = self.keyname
          data['user-data'] = self.user_data
          data['guestname'] = self.guestname
          data['guest'] = self.guest
          data['user'] = self.user
          data['provision-timeout'] = self.provision_timeout
          data['provision-tick'] = self.provision_tick
          data['api-timeout'] = self.api_timeout
          data['api-retries'] = self.api_retries
          data['api-retry-backoff-factor'] = self.api_retry_backoff_factor
  
          return data
  
      def _create(self) -> None:
          environment: Dict[str, Any] = {
              'hw': {
                  'arch': self.arch
                  },
              'os': {
                  'compose': self.image
                  }
              }
  
          data: Dict[str, Any] = {
              'environment': environment,
              'keyname': self.keyname,
              'priority_group': self.priority_group,
              'user_data': self.user_data
              }
  
          if self.pool:
              environment['pool'] = self.pool
  
          if self.hardware is not None:
              assert isinstance(self.hardware, dict)
  
              environment['hw']['constraints'] = self.hardware
  
          # TODO: snapshots
          # TODO: spot instance
          # TODO: post-install script
          # TODO: log types
  
          response = self.api.create('/guests/', data)
  
          if response.status_code == 201:
              self.info('guest', 'has been requested', 'green')
  
          else:
              raise ProvisionError(
                  f"Failed to create, "
                  f"unhandled API response '{response.status_code}'.")
  
          self.guest
  
  [... the rest is omitted ...]
files: tmt/steps/provision/artemis.py
files: import datetime
  import sys
  import time
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import requests
  from requests.packages.urllib3.util.retry import Retry as RequestsRetry
  
  import tmt
  import tmt.steps.provision
  from tmt.utils import ProvisionError, updatable_message
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  
  # List of Artemis API versions supported and understood by this plugin.
  # Since API gains support for new features over time, it is important to
  # know when particular feature became available, and avoid using it with
  # older APIs.
  SUPPORTED_API_VERSIONS = (
      # NEW: added hostname HW constraint
      'v0.0.38',
      # NEW: virtualization HW constraint
      'v0.0.37',
      # NEW: boot.method HW constraint
      'v0.0.32',
      # NEW: network HW constraint
      'v0.0.28'
      )
  
  
  # The default Artemis API version - the most recent supported versions
  # should be perfectly fine.
  DEFAULT_API_VERSION = SUPPORTED_API_VERSIONS[0]
  
  # Type annotation for "data" package describing a guest instance. Passed
  # between load() and save() calls.
  StepStateType = TypedDict(
      'StepStateType',
      {
          # API
          'api-url': str,
          'api-version': str,
  
          # Guest request properties
          'arch': str,
          'image': str,
          'hardware': Any,
          'pool': Optional[str],
          'priority-group': str,
          'keyname': str,
          'user-data': Dict[str, str],
  
          # Provided by Artemis response
          'guestname': Optional[str],
          'guest': Optional[str],
          'user': str,
  
          # Timeouts and deadlines
          'provision-timeout': int,
          'provision-tick': int,
          'api-timeout': int,
          'api-retries': int,
          'api-retry-backoff-factor': int
          }
      )
  
  DEFAULT_USER = 'root'
  DEFAULT_ARCH = 'x86_64'
  DEFAULT_PRIORITY_GROUP = 'default-priority'
  DEFAULT_KEYNAME = 'default'
  DEFAULT_PROVISION_TIMEOUT = 600
  DEFAULT_PROVISION_TICK = 60
  DEFAULT_API_TIMEOUT = 10
  DEFAULT_API_RETRIES = 10
  # Should lead to delays of 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256 seconds
  DEFAULT_RETRY_BACKOFF_FACTOR = 1
  
  DEFAULT_GUEST_DATA = cast(
      StepStateType, {
          'api-version': DEFAULT_API_VERSION,
          'arch': DEFAULT_ARCH,
          'priority-group': DEFAULT_PRIORITY_GROUP,
          'keyname': DEFAULT_KEYNAME,
          'user-data': {},
          'user': DEFAULT_USER,
          'provision-timeout': DEFAULT_PROVISION_TIMEOUT,
          'provision-tick': DEFAULT_PROVISION_TICK,
          'api-timeout': DEFAULT_API_TIMEOUT,
          'api-retries': DEFAULT_API_RETRIES,
          'api-retry-backoff-factor': DEFAULT_RETRY_BACKOFF_FACTOR
          }
      )
  
  GUEST_STATE_COLOR_DEFAULT = 'green'
  
  GUEST_STATE_COLORS = {
      'routing': 'yellow',
      'provisioning': 'magenta',
      'promised': 'blue',
      'preparing': 'cyan',
      'cancelled': 'red',
      'error': 'red'
  }
  
  
  # Type annotation for Artemis API `GET /guests/$guestname` response.
  # Partial, not all fields necessary since plugin ignores most of them.
  GuestInspectType = TypedDict(
      'GuestInspectType', {
          'state': str,
          'address': Optional[str]
          }
      )
  
  
  class TimeoutHTTPAdapter(requests.adapters.HTTPAdapter):
      """
      Spice up request's session with custom timeout.
      """
  
      def __init__(self, *args: Any, **kwargs: Any) -> None:
          self.timeout = kwargs.pop('timeout', DEFAULT_API_TIMEOUT)
  
          super().__init__(*args, **kwargs)
  
      def send(  # type: ignore # does not match superclass type on purpose
              self,
              request: requests.PreparedRequest,
              **kwargs: Any) -> requests.Response:
          kwargs.setdefault('timeout', self.timeout)
  
          return super().send(request, **kwargs)
  
  
  class ArtemisAPI:
      def install_http_retries(
              self,
              timeout: int,
              retries: int,
              retry_backoff_factor: int
              ) -> None:
          """
          Install custom "retry strategy" and timeout to our HTTP session.
  
          Strategy and timeout work together, "consuming" the timeout as
          specified by the strategy.
          """
  
          retry_strategy = RequestsRetry(
              total=retries,
              status_forcelist=[
                  429,  # Too Many Requests
                  500,  # Internal Server Error
                  502,  # Bad Gateway
                  503,  # Service Unavailable
                  504   # Gateway Timeout
                  ],
              method_whitelist=[
                  'HEAD', 'GET', 'POST', 'DELETE', 'PUT'
                  ],
              backoff_factor=retry_backoff_factor
              )
  
          timeout_adapter = TimeoutHTTPAdapter(
              timeout=timeout,
              max_retries=retry_strategy
              )
  
          self.http_session.mount('https://', timeout_adapter)
          self.http_session.mount('http://', timeout_adapter)
  
      def __init__(self, guest: 'GuestArtemis') -> None:
          self._guest = guest
  
          self.http_session = requests.Session()
  
          self.install_http_retries(
              timeout=guest.api_timeout,
              retries=guest.api_retries,
              retry_backoff_factor=guest.api_retry_backoff_factor
              )
  
      def query(
              self,
              path: str,
              method: str = 'get',
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Base helper for Artemis API queries.
  
          Trivial dispatcher per method, returning retrieved response.
  
          :param path: API path to contact.
          :param method: HTTP method to use.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          url = f'{self._guest.api_url}{path}'
  
          if method == 'get':
              return self.http_session.get(url, **request_kwargs)
  
          if method == 'post':
              return self.http_session.post(url, **request_kwargs)
  
          if method == 'delete':
              return self.http_session.delete(url, **request_kwargs)
  
          if method == 'put':
              return self.http_session.put(url, **request_kwargs)
  
          raise tmt.utils.GeneralError(
              f'Unsupported Artemis API method.\n{method}')
  
      def create(
              self,
              path: str,
              data: Dict[str, Any],
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Create - or request creation of - a resource.
  
          :param path: API path to contact.
          :param data: optional key/value data to send with the request.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
          request_kwargs['json'] = data
  
          return self.query(path, method='post', request_kwargs=request_kwargs)
  
      def inspect(
              self,
              path: str,
              params: Optional[Dict[str, Any]] = None,
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Inspect a resource.
  
          :param path: API path to contact.
          :param params: optional key/value query parameters.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          if params:
              request_kwargs['params'] = params
  
          return self.query(path, request_kwargs=request_kwargs)
  
      def delete(
              self,
              path: str,
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Delete - or request removal of - a resource.
  
          :param path: API path to contact.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          return self.query(path, method='delete', request_kwargs=request_kwargs)
  
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  class ProvisionArtemis(
          tmt.steps.provision.ProvisionPlugin):  # type: ignore[misc]
      """
      Provision guest using Artemis backend
  
      Minimal configuration could look like this:
  
          provision:
              how: artemis
              image: Fedora
              api-url: https://your-artemis.com/
  
      Note that the actual value of "image" depends on what images - or
      "composes" as Artemis calls them - supports and can deliver.
  
      Note that "api-url" can be also given via ARTEMIS_API_URL
      environment variable.
  
      Full configuration example:
  
          provision:
              how: artemis
  
              # Artemis API
              api-url: https://your-artemis.com/
              api-version: 0.0.32
  
              # Mandatory environment properties
              image: Fedora
  
              # Optional environment properties
              arch: aarch64
              pool: optional-pool-name
  
              # Provisioning process control (optional)
              priority-group: custom-priority-group
              keyname: custom-SSH-key-name
  
              # Labels to be attached to guest request (optional)
              user-data:
                  foo: bar
  
              # Timeouts and deadlines (optional)
              provision-timeout: 3600
              provision-tick: 10
              api-timeout: 600
              api-retries: 5
              api-retry-backoff-factor: 1
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [
          tmt.steps.Method(name='artemis', doc=__doc__, order=50),
          ]
  
      _keys = [
          'api-url',
          'api-version',
          'arch',
          'image',
          'hardware',
          'pool',
          'priority-group',
          'keyname',
          'user-data',
          'provision-timeout',
          'provision-tick',
          'api-timeout',
          'api-retries',
          'api-retry-backoff-factor']
  
      # TODO: fix types once superclass gains its annotations
      @classmethod
      def options(cls, how: Any = None) -> List[click.Option]:
          """ Prepare command line options for Artemis """
          return cast(List[click.Option], [
              click.option(
                  '--api-url', metavar='URL',
                  help="Artemis API URL.",
                  envvar='ARTEMIS_API_URL'
                  ),
              click.option(
                  '--api-version', metavar='x.y.z',
                  help="Artemis API version to use.",
                  type=click.Choice(SUPPORTED_API_VERSIONS),
                  envvar='ARTEMIS_API_VERSION'
                  ),
              click.option(
                  '--arch', metavar='ARCH',
                  help='Architecture to provision.'
                  ),
              click.option(
                  '--image', metavar='COMPOSE',
                  help='Image (or "compose" in Artemis terminology) '
                       'to provision.'
                  ),
              click.option(
                  '--pool', metavar='NAME',
                  help='Pool to enforce.'
                  ),
              click.option(
                  '--priority-group', metavar='NAME',
                  help='Provisioning priority group.'
                  ),
              click.option(
                  '--keyname', metavar='NAME',
                  help='SSH key name.'
                  ),
              click.option(
                  '--user-data', metavar='KEY=VALUE',
                  help='Optional data to attach to guest.',
                  multiple=True,
                  default=[]
                  ),
              click.option(
                  '--provision-timeout', metavar='SECONDS',
                  help=f'How long to wait for provisioning to complete, '
                       f'{DEFAULT_PROVISION_TIMEOUT} seconds by default.'
                  ),
              click.option(
                  '--provision-tick', metavar='SECONDS',
                  help=f'How often check Artemis API for provisioning status, '
                       f'{DEFAULT_PROVISION_TICK} seconds by default.',
                  ),
              click.option(
                  '--api-timeout', metavar='SECONDS',
                  help=f'How long to wait for API operations to complete, '
                       f'{DEFAULT_API_TIMEOUT} seconds by default.',
                  ),
              click.option(
                  '--api-retries', metavar='COUNT',
                  help=f'How many attempts to use when talking to API, '
                       f'{DEFAULT_API_RETRIES} by default.',
                  ),
              click.option(
                  '--api-retry-backoff-factor', metavar='COUNT',
                  help=f'A factor for exponential API retry backoff, '
                       f'{DEFAULT_RETRY_BACKOFF_FACTOR} by default.',
                  ),
              ]) + cast(List[click.Option], super().options(how))
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
  
          return DEFAULT_GUEST_DATA.get(option, default)
  
      # TODO: use better types once superclass gains its annotations
      def wake(self, keys: Optional[List[str]] = None,
               data: Optional[StepStateType] = None) -> None:
          """ Wake up the plugin, process data, apply options """
  
          super().wake(keys=keys, data=data)
  
          if data:
              self._guest = GuestArtemis(data, name=self.name, parent=self.step)
  
      def go(self) -> None:
          """ Provision the guest """
          super().go()
  
          api_version = self.get('api-version')
  
          if api_version not in SUPPORTED_API_VERSIONS:
              raise ProvisionError(f"API version '{api_version}' not supported.")
  
          try:
              user_data = {
                  key.strip(): value.strip()
                  for key, value in (
                      pair.split('=', 1)
                      for pair in self.get('user-data')
                      )
                  }
  
          except ValueError:
              raise ProvisionError('Cannot parse user-data.')
  
          data: StepStateType = {
              'api-url': self.get('api-url'),
              'api-version': api_version,
              'arch': self.get('arch'),
              'image': self.get('image'),
              'hardware': self.get('hardware'),
              'pool': self.get('pool'),
              'priority-group': self.get('priority-group'),
              'keyname': self.get('keyname'),
              'user-data': user_data,
              'guestname': None,
              'guest': None,
              'user': DEFAULT_USER,
              'provision-timeout': self.get('provision-timeout'),
              'provision-tick': self.get('provision-tick'),
              'api-timeout': self.get('api-timeout'),
              'api-retries': self.get('api-retries'),
              'api-retry-backoff-factor': self.get('api-retry-backoff-factor')
              }
  
          self._guest = GuestArtemis(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self) -> Optional['GuestArtemis']:
          """ Return the provisioned guest """
          return self._guest
  
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  class GuestArtemis(tmt.GuestSsh):  # type: ignore[misc]
      """
      Artemis guest instance
  
      The following keys are expected in the 'data' dictionary:
      """
  
      _api: Optional[ArtemisAPI] = None
  
      @property
      def api(self) -> ArtemisAPI:
          if self._api is None:
              self._api = ArtemisAPI(self)
  
          return self._api
  
      def load(self, data: StepStateType) -> None:
          super().load(data)
  
          self.api_url = data['api-url']
          self.api_version = data['api-version']
          self.arch = data['arch']
          self.image = data['image']
          self.hardware = data['hardware']
          self.pool = data['pool']
          self.priority_group = data['priority-group']
          self.keyname = data['keyname']
          self.user_data = data['user-data']
          self.guestname = data['guestname']
          self.guest = data['guest']
          self.user = data['user']
          self.provision_timeout = data['provision-timeout']
          self.provision_tick = data['provision-tick']
          self.api_timeout = data['api-timeout']
          self.api_retries = data['api-retries']
          self.api_retry_backoff_factor = data['api-retry-backoff-factor']
  
      def save(self) -> StepStateType:
          data = cast(StepStateType, super().save())
  
          data['api-url'] = self.api_url
          data['api-version'] = self.api_version
          data['arch'] = self.arch
          data['image'] = self.image
          data['hardware'] = self.hardware
          data['pool'] = self.pool
          data['priority-group'] = self.priority_group
          data['keyname'] = self.keyname
          data['user-data'] = self.user_data
          data['guestname'] = self.guestname
          data['guest'] = self.guest
         
  
  [... the rest is omitted ...]
files: tmt/steps/provision/artemis.py
files: # coding: utf-8
  
  import os
  import platform
  import re
  import time
  from typing import Optional
  
  import click
  import fmf
  import requests
  
  import tmt
  import tmt.steps.provision
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  
  def import_testcloud():
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
    - {public_key}
  ssh_pwauth: true
  disable_root: false
  runcmd:
    - sed -i -e '/^.*PermitRootLogin/s/^.*$/PermitRootLogin yes/'
      -e '/^.*UseDNS/s/^.*$/UseDNS no/'
      -e '/^.*GSSAPIAuthentication/s/^.*$/GSSAPIAuthentication no/'
      /etc/ssh/sshd_config
    - systemctl reload sshd
    - [sh, -c, 'if [ ! -f /etc/systemd/network/20-tc-usernet.network ] &&
    systemctl status systemd-networkd | grep -q "enabled;\\svendor\\spreset:\\senabled";
    then mkdir -p /etc/systemd/network/ &&
    echo "[Match]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "Name=en*" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "[Network]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "DHCP=yes" >> /etc/systemd/network/20-tc-usernet.network; fi']
    - [sh, -c, 'if systemctl status systemd-networkd |
    grep -q "enabled;\\svendor\\spreset:\\senabled"; then
    systemctl restart systemd-networkd; fi']
    - [sh, -c, 'if cat /etc/os-release |
    grep -q platform:el8; then systemctl restart sshd; fi']
  """
  
  # Libvirt domain XML template related variables
  DOMAIN_TEMPLATE_NAME = 'domain-template.jinja'
  DOMAIN_TEMPLATE_FILE = os.path.join(TESTCLOUD_DATA, DOMAIN_TEMPLATE_NAME)
  DOMAIN_TEMPLATE = """<domain type='{{ virt_type }}' xmlns:qemu='http://libvirt.org/schemas/domain/qemu/1.0'>
    <name>{{ domain_name }}</name>
    <uuid>{{ uuid }}</uuid>
    <memory unit='KiB'>{{ memory }}</memory>
    <currentMemory unit='KiB'>{{ memory }}</currentMemory>
    <vcpu placement='static'>2</vcpu>
    <os>
      <type arch='{{ arch }}' machine='{{ model }}'>hvm</type>
      {{ uefi_loader }}
      <boot dev='hd'/>
    </os>
    {{ cpu }}
    {{ extra_specs }}
    <clock offset='utc'>
      <timer name='rtc' tickpolicy='catchup'/>
      <timer name='pit' tickpolicy='delay'/>
      <timer name='hpet' present='no'/>
    </clock>
    <on_poweroff>destroy</on_poweroff>
    <on_reboot>restart</on_reboot>
    <on_crash>restart</on_crash>
    <devices>
      <emulator>{{ emulator_path }}</emulator>
      <disk type='file' device='disk'>
        <driver name='qemu' type='qcow2' cache='unsafe'/>
        <source file="{{ disk }}"/>
        <target dev='vda' bus='virtio'/>
        {{ boot_drive_address }}
      </disk>
      <disk type='file' device='disk'>
        <driver name='qemu' type='raw'/>
        <source file="{{ seed }}"/>
        <target dev='vdb' bus='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x08' function='0x0'/>
      </disk>
      <interface type='{{ network_type }}'>
        <mac address="{{ mac_address }}"/>
        {{ network_source }}
        {{ ip_setup }}
        <model type='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>
      </interface>
      <serial type='pty'>
        <target port='0'/>
      </serial>
      <console type='pty'>
        <target type='serial' port='0'/>
      </console>
      <input type="keyboard" bus="virtio"/>
      <rng model='virtio'>
        <backend model='random'>/dev/urandom</backend>
      </rng>
    </devices>
    {{ qemu_args }}
  </domain>
  """  # noqa: E501
  
  # VM defaults
  DEFAULT_BOOT_TIMEOUT = 60      # seconds
  DEFAULT_CONNECT_TIMEOUT = 60   # seconds
  NON_KVM_ADDITIONAL_WAIT = 10       # seconds
  NON_KVM_TIMEOUT_COEF = 10          # times
  
  # SSH key type, set None for ssh-keygen default one
  SSH_KEYGEN_TYPE = "ecdsa"
  
  
  class ProvisionTestcloud(tmt.steps.provision.ProvisionPlugin):
      """
      Local virtual machine using testcloud
  
      Minimal config which uses the latest fedora image:
  
          provision:
              how: virtual
  
      Here's a full config example:
  
          provision:
              how: virtual
              image: fedora
              user: root
              memory: 2048
  
      As the image use 'fedora' for the latest released Fedora compose,
      'rawhide' for the latest Rawhide compose, short aliases such as
      'fedora-32', 'f-32' or 'f32' for specific release or a full url to
      the qcow2 image for example from:
  
          https://kojipkgs.fedoraproject.org/compose/
  
      Short names are also provided for 'centos', 'centos-stream',
      'debian' and 'ubuntu' (e.g. 'centos-8' or 'c8').
  
      Use the full path for images stored on local disk, for example:
  
          /var/tmp/images/Fedora-Cloud-Base-31-1.9.x86_64.qcow2
  
      In addition to the qcow2 format, vagrant boxes can be used as well,
      testcloud will take care of unpacking the image for you.
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [
          tmt.steps.Method(name='virtual.testcloud', doc=__doc__, order=50),
          ]
  
      # Supported keys
      _keys = ["image", "user", "memory", "disk", "connection", "arch"]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for testcloud """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to be used. Provide a short name, '
                       'full path to a local file or a complete url.'),
              click.option(
                  '-m', '--memory', metavar='MEMORY',
                  help='Set available memory in MB, 2048 MB by default.'),
              click.option(
                  '-D', '--disk', metavar='MEMORY',
                  help='Specify disk size in GB, 10 GB by default.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-c', '--connection',
                  type=click.Choice(['session', 'system']),
                  help="What session type to use, 'session' by default."),
              click.option(
                  '-a', '--arch',
                  type=click.Choice(['x86_64', 'aarch64', 's390x', 'ppc64le']),
                  help="What architecture to virtualize, host arch by default."),
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          defaults = {
              'user': 'root',
              'memory': 2048,
              'disk': 10,
              'image': 'fedora',
              'connection': 'session',
              'arch': platform.machine()
              }
          if option in defaults:
              return defaults[option]
          return default
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
  
          # Convert memory and disk to integers
          for key in ['memory', 'disk']:
              if isinstance(self.get(key), str):
                  self.data[key] = int(self.data[key])
  
          # Wake up testcloud instance
          if data:
              guest = GuestTestcloud(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self):
          """ Provision the testcloud instance """
          super().go()
  
          # Give info about provided data
          data = dict()
          for key in self._keys + self._common_keys:
              data[key] = self.get(key)
              if key == 'memory':
                  self.info('memory', f"{self.get('memory')} MB", 'green')
              elif key == 'disk':
                  self.info('disk', f"{self.get('disk')} GB", 'green')
              elif key == 'connection':
                  self.verbose(key, data[key], 'green')
              else:
                  if data[key] is not None:
                      self.info(key, data[key], 'green')
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestTestcloud(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      @classmethod
      def clean_images(cls, clean, dry):
          """ Remove the testcloud images """
          clean.info('testcloud', shift=1, color='green')
          if not os.path.exists(TESTCLOUD_IMAGES):
              clean.warn(
                  f"Directory '{TESTCLOUD_IMAGES}' does not exist.", shift=2)
              return
          for image in os.listdir(TESTCLOUD_IMAGES):
              image = os.path.join(TESTCLOUD_IMAGES, image)
              if dry:
                  clean.verbose(f"Would remove '{image}'.", shift=2)
              else:
                  clean.verbose(f"Removing '{image}'.", shift=2)
                  os.remove(image)
  
  
  class GuestTestcloud(tmt.GuestSsh):
      """
      Testcloud Instance
  
      The following keys are expected in the 'data' dictionary::
  
          image ...... qcov image name or url
          user ....... user name to log in
          memory ..... memory size for vm
          disk ....... disk size for vm
          connection . either session (default) or system, to be passed to qemu
          arch ....... architecture for the VM, host arch is the default
      """
  
      # Not to be saved, recreated from image_url/instance_name/... every
      # time guest is instantiated.
      _image: Optional['testcloud.image.Image'] = None
      _instance: Optional['testcloud.instance.Instance'] = None
  
      def _get_url(self, url, message):
          """ Get url, retry when fails, return response """
          timeout = DEFAULT_CONNECT_TIMEOUT
          wait = 1
          while True:
              try:
                  with retry_session() as session:
                      response = session.get(url)
                  if response.ok:
                      return response
              except requests.RequestException:
                  pass
              if timeout < 0:
                  raise ProvisionError(
                      f'Failed to {message} in {DEFAULT_CONNECT_TIMEOUT}s.')
              self.debug(
                  f'Unable to {message} ({url}), retrying, '
                  f'{fmf.utils.listed(timeout, "second")} left.')
              time.sleep(wait)
              wait += 1
              timeout -= wait
  
      def _guess_image_url(self, name):
          """ Guess image url for given name """
  
          # Try to check if given url is a local file
          if os.path.isabs(name) and os.path.isfile(name):
              return f'file://{name}'
  
          name = name.lower().strip()
          url = None
  
          # Map fedora aliases (e.g. rawhide, fedora, fedora-32, f-32, f32)
          matched_fedora = re.match(r'^f(edora)?-?(\d+)$', name)
          # Map centos aliases (e.g. centos:X, centos, centos-stream:X)
          matched_centos = [re.match(r'^c(entos)?-?(\d+)$', name),
                            re.match(r'^c(entos-stream)?-?(\d+)$', name)]
          matched_ubuntu = re.match(r'^u(buntu)?-?(\w+)$', name)
          matched_debian = re.match(r'^d(ebian)?-?(\w+)$', name)
  
          # Plain name match means we want the latest release
          if name == 'fedora':
              url = testcloud.util.get_fedora_image_url("latest", self.arch)
          elif name == 'centos':
              url = testcloud.util.get_centos_image_url("latest", self.arch)
          elif name == 'centos-stream':
              url = testcloud.util.get_centos_image_url(
                  "latest", stream=True, arch=self.arch)
          elif name == 'ubuntu':
              url = testcloud.util.get_ubuntu_image_url("latest", self.arch)
          elif name == 'debian':
              url = testcloud.util.get_debian_image_url("latest", self.arch)
  
          elif matched_fedora:
              url = testcloud.util.get_fedora_image_url(
                  matched_fedora.group(2), self.arch)
          elif matched_centos[0]:
              url = testcloud.util.get_centos_image_url(
                  matched_centos[0].group(2), self.arch)
          elif matched_centos[1]:
              url = testcloud.util.get_centos_image_url(
                  matched_centos[1].group(2), stream=True, arch=self.arch)
          elif matched_ubuntu:
              url = testcloud.util.get_ubuntu_image_url(
                  matched_ubuntu.group(2), self.arch)
          elif matched_debian:
              url = testcloud.util.get_debian_image_url(
                  matched_debian.group(2), self.arch)
          elif 'rawhide' in name:
              url = testcloud.util.get_fedora_image_url("rawhide", self.arch)
  
          if not url:
              raise ProvisionError(f"Could not map '{name}' to compose.")
          return url
  
      @staticmethod
      def _create_template():
          """ Create libvirt domain template """
          # Write always to ovewrite possible outdated version
          with open(DOMAIN_TEMPLATE_FILE, 'w') as template:
              template.write(DOMAIN_TEMPLATE)
  
      def load(self, data):
          """ Load guest data and initialize attributes """
          super().load(data)
          self.image = data.get('image')
          self.image_url = data.get('image_url')
          self.instance_name = data.get('instance_name')
          self.memory = data.get('memory')
          self.disk = data.get('disk')
          self.connection = data.get('connection')
          self.arch = data.get('arch')
  
      def save(self):
          """ Save guest data for future wake up """
          data = super().save()
          data['instance_name'] = self.instance_name
          data['image'] = self.image
          data['image_url'] = self.image_url
          data['connection'] = self.connection
          data['arch'] = self.arch
          return data
  
      def wake(self):
          """ Wake up the guest """
          self.debug(
              f"Waking up testcloud instance '{self.instance_name}'.",
              level=2, shift=0)
          self.prepare_config()
          self._image = testcloud.image.Image(self.image_url)
          self._instance = testcloud.instance.Instance(
              self.instance_name, image=self._image,
              connection=f"qemu:///{self.connection}", desired_arch=self.arch)
  
      def prepare_ssh_key(self, key_type=None):
          """ Prepare ssh key for authentication """
          # Create ssh key paths
          key_name = "id_{}".format(key_type if key_type is not None else 'rsa')
          self.key = os.path.join(self.workdir, key_name)
          self.pubkey = os.path.join(self.workdir, f'{key_name}.pub')
  
          # Generate ssh key
          self.debug('Generating an ssh key.')
          command = ["ssh-keygen", "-f", self.key, "-N", ""]
          if key_type is not None:
              command.extend(["-t", key_type])
          self.run(command)
          with open(self.pubkey, 'r') as pubkey:
              self.config.USER_DATA = USER_DATA.format(
                  user_name=self.user, public_key=pubkey.read())
  
      def prepare_config(self):
          """ Prepare
  
  [... the rest is omitted ...]
files: tmt/steps/provision/testcloud.py
files: # coding: utf-8
  
  import os
  import platform
  import re
  import time
  from typing import Optional
  
  import click
  import fmf
  import requests
  
  import tmt
  import tmt.steps.provision
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  
  def import_testcloud():
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
    - {public_key}
  ssh_pwauth: true
  disable_root: false
  runcmd:
    - sed -i -e '/^.*PermitRootLogin/s/^.*$/PermitRootLogin yes/'
      -e '/^.*UseDNS/s/^.*$/UseDNS no/'
      -e '/^.*GSSAPIAuthentication/s/^.*$/GSSAPIAuthentication no/'
      /etc/ssh/sshd_config
    - systemctl reload sshd
    - [sh, -c, 'if [ ! -f /etc/systemd/network/20-tc-usernet.network ] &&
    systemctl status systemd-networkd | grep -q "enabled;\\svendor\\spreset:\\senabled";
    then mkdir -p /etc/systemd/network/ &&
    echo "[Match]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "Name=en*" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "[Network]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "DHCP=yes" >> /etc/systemd/network/20-tc-usernet.network; fi']
    - [sh, -c, 'if systemctl status systemd-networkd |
    grep -q "enabled;\\svendor\\spreset:\\senabled"; then
    systemctl restart systemd-networkd; fi']
    - [sh, -c, 'if cat /etc/os-release |
    grep -q platform:el8; then systemctl restart sshd; fi']
  """
  
  # Libvirt domain XML template related variables
  DOMAIN_TEMPLATE_NAME = 'domain-template.jinja'
  DOMAIN_TEMPLATE_FILE = os.path.join(TESTCLOUD_DATA, DOMAIN_TEMPLATE_NAME)
  DOMAIN_TEMPLATE = """<domain type='{{ virt_type }}' xmlns:qemu='http://libvirt.org/schemas/domain/qemu/1.0'>
    <name>{{ domain_name }}</name>
    <uuid>{{ uuid }}</uuid>
    <memory unit='KiB'>{{ memory }}</memory>
    <currentMemory unit='KiB'>{{ memory }}</currentMemory>
    <vcpu placement='static'>2</vcpu>
    <os>
      <type arch='{{ arch }}' machine='{{ model }}'>hvm</type>
      {{ uefi_loader }}
      <boot dev='hd'/>
    </os>
    {{ cpu }}
    {{ extra_specs }}
    <clock offset='utc'>
      <timer name='rtc' tickpolicy='catchup'/>
      <timer name='pit' tickpolicy='delay'/>
      <timer name='hpet' present='no'/>
    </clock>
    <on_poweroff>destroy</on_poweroff>
    <on_reboot>restart</on_reboot>
    <on_crash>restart</on_crash>
    <devices>
      <emulator>{{ emulator_path }}</emulator>
      <disk type='file' device='disk'>
        <driver name='qemu' type='qcow2' cache='unsafe'/>
        <source file="{{ disk }}"/>
        <target dev='vda' bus='virtio'/>
        {{ boot_drive_address }}
      </disk>
      <disk type='file' device='disk'>
        <driver name='qemu' type='raw'/>
        <source file="{{ seed }}"/>
        <target dev='vdb' bus='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x08' function='0x0'/>
      </disk>
      <interface type='{{ network_type }}'>
        <mac address="{{ mac_address }}"/>
        {{ network_source }}
        {{ ip_setup }}
        <model type='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>
      </interface>
      <serial type='pty'>
        <target port='0'/>
      </serial>
      <console type='pty'>
        <target type='serial' port='0'/>
      </console>
      <input type="keyboard" bus="virtio"/>
      <rng model='virtio'>
        <backend model='random'>/dev/urandom</backend>
      </rng>
    </devices>
    {{ qemu_args }}
  </domain>
  """  # noqa: E501
  
  # VM defaults
  DEFAULT_BOOT_TIMEOUT = 60      # seconds
  DEFAULT_CONNECT_TIMEOUT = 60   # seconds
  NON_KVM_ADDITIONAL_WAIT = 10       # seconds
  NON_KVM_TIMEOUT_COEF = 10          # times
  
  # SSH key type, set None for ssh-keygen default one
  SSH_KEYGEN_TYPE = "ecdsa"
  
  
  class ProvisionTestcloud(tmt.steps.provision.ProvisionPlugin):
      """
      Local virtual machine using testcloud
  
      Minimal config which uses the latest fedora image:
  
          provision:
              how: virtual
  
      Here's a full config example:
  
          provision:
              how: virtual
              image: fedora
              user: root
              memory: 2048
  
      As the image use 'fedora' for the latest released Fedora compose,
      'rawhide' for the latest Rawhide compose, short aliases such as
      'fedora-32', 'f-32' or 'f32' for specific release or a full url to
      the qcow2 image for example from:
  
          https://kojipkgs.fedoraproject.org/compose/
  
      Short names are also provided for 'centos', 'centos-stream',
      'debian' and 'ubuntu' (e.g. 'centos-8' or 'c8').
  
      Use the full path for images stored on local disk, for example:
  
          /var/tmp/images/Fedora-Cloud-Base-31-1.9.x86_64.qcow2
  
      In addition to the qcow2 format, vagrant boxes can be used as well,
      testcloud will take care of unpacking the image for you.
      """
  
      # Guest instance
      _guest = None
  
      # Supported methods
      _methods = [
          tmt.steps.Method(name='virtual.testcloud', doc=__doc__, order=50),
          ]
  
      # Supported keys
      _keys = ["image", "user", "memory", "disk", "connection", "arch"]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for testcloud """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to be used. Provide a short name, '
                       'full path to a local file or a complete url.'),
              click.option(
                  '-m', '--memory', metavar='MEMORY',
                  help='Set available memory in MB, 2048 MB by default.'),
              click.option(
                  '-D', '--disk', metavar='MEMORY',
                  help='Specify disk size in GB, 10 GB by default.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-c', '--connection',
                  type=click.Choice(['session', 'system']),
                  help="What session type to use, 'session' by default."),
              click.option(
                  '-a', '--arch',
                  type=click.Choice(['x86_64', 'aarch64', 's390x', 'ppc64le']),
                  help="What architecture to virtualize, host arch by default."),
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          defaults = {
              'user': 'root',
              'memory': 2048,
              'disk': 10,
              'image': 'fedora',
              'connection': 'session',
              'arch': platform.machine()
              }
          if option in defaults:
              return defaults[option]
          return default
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
  
          # Convert memory and disk to integers
          for key in ['memory', 'disk']:
              if isinstance(self.get(key), str):
                  self.data[key] = int(self.data[key])
  
          # Wake up testcloud instance
          if data:
              guest = GuestTestcloud(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self):
          """ Provision the testcloud instance """
          super().go()
  
          # Give info about provided data
          data = dict()
          for key in self._keys + self._common_keys:
              data[key] = self.get(key)
              if key == 'memory':
                  self.info('memory', f"{self.get('memory')} MB", 'green')
              elif key == 'disk':
                  self.info('disk', f"{self.get('disk')} GB", 'green')
              elif key == 'connection':
                  self.verbose(key, data[key], 'green')
              else:
                  if data[key] is not None:
                      self.info(key, data[key], 'green')
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestTestcloud(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      @classmethod
      def clean_images(cls, clean, dry):
          """ Remove the testcloud images """
          clean.info('testcloud', shift=1, color='green')
          if not os.path.exists(TESTCLOUD_IMAGES):
              clean.warn(
                  f"Directory '{TESTCLOUD_IMAGES}' does not exist.", shift=2)
              return
          for image in os.listdir(TESTCLOUD_IMAGES):
              image = os.path.join(TESTCLOUD_IMAGES, image)
              if dry:
                  clean.verbose(f"Would remove '{image}'.", shift=2)
              else:
                  clean.verbose(f"Removing '{image}'.", shift=2)
                  os.remove(image)
  
  
  class GuestTestcloud(tmt.GuestSsh):
      """
      Testcloud Instance
  
      The following keys are expected in the 'data' dictionary::
  
          image ...... qcov image name or url
          user ....... user name to log in
          memory ..... memory size for vm
          disk ....... disk size for vm
          connection . either session (default) or system, to be passed to qemu
          arch ....... architecture for the VM, host arch is the default
      """
  
      # Not to be saved, recreated from image_url/instance_name/... every
      # time guest is instantiated.
      _image: Optional['testcloud.image.Image'] = None
      _instance: Optional['testcloud.instance.Instance'] = None
  
      def _get_url(self, url, message):
          """ Get url, retry when fails, return response """
          timeout = DEFAULT_CONNECT_TIMEOUT
          wait = 1
          while True:
              try:
                  response = retry_session().get(url)
                  if response.ok:
                      return response
              except requests.RequestException:
                  pass
              if timeout < 0:
                  raise ProvisionError(
                      f'Failed to {message} in {DEFAULT_CONNECT_TIMEOUT}s.')
              self.debug(
                  f'Unable to {message} ({url}), retrying, '
                  f'{fmf.utils.listed(timeout, "second")} left.')
              time.sleep(wait)
              wait += 1
              timeout -= wait
  
      def _guess_image_url(self, name):
          """ Guess image url for given name """
  
          # Try to check if given url is a local file
          if os.path.isabs(name) and os.path.isfile(name):
              return f'file://{name}'
  
          name = name.lower().strip()
          url = None
  
          # Map fedora aliases (e.g. rawhide, fedora, fedora-32, f-32, f32)
          matched_fedora = re.match(r'^f(edora)?-?(\d+)$', name)
          # Map centos aliases (e.g. centos:X, centos, centos-stream:X)
          matched_centos = [re.match(r'^c(entos)?-?(\d+)$', name),
                            re.match(r'^c(entos-stream)?-?(\d+)$', name)]
          matched_ubuntu = re.match(r'^u(buntu)?-?(\w+)$', name)
          matched_debian = re.match(r'^d(ebian)?-?(\w+)$', name)
  
          # Plain name match means we want the latest release
          if name == 'fedora':
              url = testcloud.util.get_fedora_image_url("latest", self.arch)
          elif name == 'centos':
              url = testcloud.util.get_centos_image_url("latest", self.arch)
          elif name == 'centos-stream':
              url = testcloud.util.get_centos_image_url(
                  "latest", stream=True, arch=self.arch)
          elif name == 'ubuntu':
              url = testcloud.util.get_ubuntu_image_url("latest", self.arch)
          elif name == 'debian':
              url = testcloud.util.get_debian_image_url("latest", self.arch)
  
          elif matched_fedora:
              url = testcloud.util.get_fedora_image_url(
                  matched_fedora.group(2), self.arch)
          elif matched_centos[0]:
              url = testcloud.util.get_centos_image_url(
                  matched_centos[0].group(2), self.arch)
          elif matched_centos[1]:
              url = testcloud.util.get_centos_image_url(
                  matched_centos[1].group(2), stream=True, arch=self.arch)
          elif matched_ubuntu:
              url = testcloud.util.get_ubuntu_image_url(
                  matched_ubuntu.group(2), self.arch)
          elif matched_debian:
              url = testcloud.util.get_debian_image_url(
                  matched_debian.group(2), self.arch)
          elif 'rawhide' in name:
              url = testcloud.util.get_fedora_image_url("rawhide", self.arch)
  
          if not url:
              raise ProvisionError(f"Could not map '{name}' to compose.")
          return url
  
      @staticmethod
      def _create_template():
          """ Create libvirt domain template """
          # Write always to ovewrite possible outdated version
          with open(DOMAIN_TEMPLATE_FILE, 'w') as template:
              template.write(DOMAIN_TEMPLATE)
  
      def load(self, data):
          """ Load guest data and initialize attributes """
          super().load(data)
          self.image = data.get('image')
          self.image_url = data.get('image_url')
          self.instance_name = data.get('instance_name')
          self.memory = data.get('memory')
          self.disk = data.get('disk')
          self.connection = data.get('connection')
          self.arch = data.get('arch')
  
      def save(self):
          """ Save guest data for future wake up """
          data = super().save()
          data['instance_name'] = self.instance_name
          data['image'] = self.image
          data['image_url'] = self.image_url
          data['connection'] = self.connection
          data['arch'] = self.arch
          return data
  
      def wake(self):
          """ Wake up the guest """
          self.debug(
              f"Waking up testcloud instance '{self.instance_name}'.",
              level=2, shift=0)
          self.prepare_config()
          self._image = testcloud.image.Image(self.image_url)
          self._instance = testcloud.instance.Instance(
              self.instance_name, image=self._image,
              connection=f"qemu:///{self.connection}", desired_arch=self.arch)
  
      def prepare_ssh_key(self, key_type=None):
          """ Prepare ssh key for authentication """
          # Create ssh key paths
          key_name = "id_{}".format(key_type if key_type is not None else 'rsa')
          self.key = os.path.join(self.workdir, key_name)
          self.pubkey = os.path.join(self.workdir, f'{key_name}.pub')
  
          # Generate ssh key
          self.debug('Generating an ssh key.')
          command = ["ssh-keygen", "-f", self.key, "-N", ""]
          if key_type is not None:
              command.extend(["-t", key_type])
          self.run(command)
          with open(self.pubkey, 'r') as pubkey:
              self.config.USER_DATA = USER_DATA.format(
                  user_name=self.user, public_key=pubkey.read())
  
      def prepare_config(self):
          """ Prepare common configuration """
          import_test
  
  [... the rest is omitted ...]
files: tmt/steps/provision/testcloud.py
files: 
  """ Test Metadata Utilities """
  
  import contextlib
  import dataclasses
  import datetime
  import glob
  import io
  import os
  import pprint
  import re
  import shlex
  import shutil
  import subprocess
  import sys
  import unicodedata
  from collections import OrderedDict
  from functools import lru_cache
  from pathlib import Path
  from threading import Thread
  from typing import (IO, TYPE_CHECKING, Any, Dict, Generator, Iterable, List,
                      NamedTuple, Optional, Pattern, Tuple, Type, TypeVar, Union,
                      cast, overload)
  
  import click
  import fmf
  import requests
  import requests.adapters
  from click import echo, style, wrap_text
  from requests.packages.urllib3.util.retry import Retry
  from ruamel.yaml import YAML, scalarstring
  from ruamel.yaml.comments import CommentedMap
  
  if sys.version_info >= (3, 8):
      from typing import Literal, Protocol
  else:
      from typing_extensions import Literal, Protocol
  
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  log = fmf.utils.Logging('tmt').logger
  
  # Default workdir root and max
  WORKDIR_ROOT = '/var/tmp/tmt'
  WORKDIR_MAX = 1000
  
  # Log in workdir
  LOG_FILENAME = 'log.txt'
  
  # Maximum number of lines of stdout/stderr to show upon errors
  OUTPUT_LINES = 100
  # Default output width
  OUTPUT_WIDTH = 79
  
  # Hierarchy indent
  INDENT = 4
  
  # Default name and order for step plugins
  DEFAULT_NAME = 'default'
  DEFAULT_PLUGIN_ORDER = 50
  DEFAULT_PLUGIN_ORDER_MULTIHOST = 10
  DEFAULT_PLUGIN_ORDER_REQUIRES = 70
  DEFAULT_PLUGIN_ORDER_RECOMMENDS = 75
  
  # Config directory
  CONFIG_PATH = '~/.config/tmt'
  
  # Special process return code
  PROCESS_TIMEOUT = 124
  
  # Default select.select(timeout) in seconds
  DEFAULT_SELECT_TIMEOUT = 5
  
  # Shell options to be set for all run shell scripts
  SHELL_OPTIONS = 'set -eo pipefail'
  
  # Defaults for HTTP/HTTPS retries and timeouts (see `retry_session()`).
  DEFAULT_RETRY_SESSION_RETRIES: int = 3
  DEFAULT_RETRY_SESSION_BACKOFF_FACTOR: float = 0.1
  
  # A stand-in variable for generic use.
  T = TypeVar('T')
  
  # A FMF context type, representing name/values context.
  FmfContextType = Dict[str, List[str]]
  
  # A "environment" type, representing name/value environment variables.
  EnvironmentType = Dict[str, str]
  
  # Workdir argument type, can be True, a string, a path or None
  WorkdirArgumentType = Union[Literal[True], str, None]
  
  # Workdir type, can be None or a string
  WorkdirType = Optional[str]
  
  # Option to skip to initialize work tree in plan
  PLAN_SKIP_WORKTREE_INIT = 'plan_skip_worktree_init'
  
  
  class BaseLoggerFnType(Protocol):
      def __call__(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          pass
  
  
  class LevelessLoggerFnType(Protocol):
      def __call__(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          pass
  
  
  class SemanticLoggerFnType(Protocol):
      def __call__(self, message: str, shift: int = 0) -> None:
          pass
  
  
  LoggerFnType = Union[
      BaseLoggerFnType,
      LevelessLoggerFnType,
      SemanticLoggerFnType]
  
  
  def indent(
          key: str,
          value: Optional[str] = None,
          color: Optional[str] = None,
          level: int = 0) -> str:
      """
      Indent a key/value message.
  
      If both ``key`` and ``value`` are specified, ``{key}: {value}``
      message is rendered. Otherwise, just ``key`` is used alone. If
      ``value`` contains multiple lines, each but the very first line is
      indented by one extra level.
  
      :param value: optional value to print at right side of ``key``.
      :param color: optional color to apply on ``key``.
      :param level: number of indentation levels. Each level is indented
                    by :py:data:`INDENT` spaces.
      """
  
      indent = ' ' * INDENT * level
      deeper = ' ' * INDENT * (level + 1)
  
      # Colorize
      if color is not None:
          key = style(key, fg=color)
  
      # Handle key only
      if value is None:
          message = key
  
      # Handle key + value
      else:
          # Multiline content indented deeper
          if isinstance(value, str):
              lines = value.splitlines()
              if len(lines) > 1:
                  value = ''.join([f"\n{deeper}{line}" for line in lines])
  
          message = f'{key}: {value}'
  
      return indent + message
  
  
  class Config(object):
      """ User configuration """
  
      def __init__(self) -> None:
          """ Initialize config directory path """
          self.path = os.path.expanduser(CONFIG_PATH)
          if not os.path.exists(self.path):
              try:
                  os.makedirs(self.path)
              except OSError as error:
                  raise GeneralError(
                      f"Failed to create config '{self.path}'.\n{error}")
  
      def last_run(self, run_id: Optional[str] = None) -> Optional[str]:
          """ Get and set last run id """
          symlink = os.path.join(self.path, 'last-run')
          if run_id:
              try:
                  os.remove(symlink)
              except OSError:
                  pass
              try:
                  os.symlink(run_id, symlink)
              except OSError as error:
                  raise GeneralError(
                      f"Unable to save last run '{self.path}'.\n{error}")
              return run_id
          if os.path.islink(symlink):
              return os.path.realpath(symlink)
          return None
  
  
  class StreamLogger(Thread):
      """
      Reading pipes of running process in threads.
  
      Code based on:
      https://github.com/packit/packit/blob/main/packit/utils/logging.py#L10
      """
  
      def __init__(self,
                   stream: Optional[IO[bytes]],
                   log_header: str,
                   logger: BaseLoggerFnType) -> None:
          super().__init__(daemon=True)
          self.stream = stream
          self.output: List[str] = []
          self.log_header = log_header
          self.logger = logger
  
      def run(self) -> None:
          if self.stream is None:
              return
  
          for _line in self.stream:
              line = _line.decode('utf-8', errors='replace')
              if line != '':
                  self.logger(
                      self.log_header,
                      line.rstrip('\n'),
                      'yellow',
                      level=3)
              self.output.append(line)
  
      def get_output(self) -> str:
          return "".join(self.output)
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Common
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  CommonDerivedType = TypeVar('CommonDerivedType', bound='Common')
  
  
  class CommandOutput(NamedTuple):
      stdout: Optional[str]
      stderr: Optional[str]
  
  
  class Common(object):
      """
      Common shared stuff
  
      Takes care of command line context, options and workdir handling.
      Provides logging functions info(), verbose() and debug().
      Implements read() and write() for comfortable file access.
      Provides the run() method for easy command execution.
      """
  
      # Command line context, options and workdir
      _context: Optional[click.Context] = None
      _options: Dict[str, Any] = dict()
      _workdir: WorkdirType = None
  
      def __init__(
              self,
              parent: Optional[CommonDerivedType] = None,
              name: Optional[str] = None,
              workdir: WorkdirArgumentType = None,
              context: Optional[click.Context] = None,
              relative_indent: int = 1):
          """
          Initialize name and relation with the parent object
  
          Prepare the workdir for provided id / directory path
          or generate a new workdir name if workdir=True given.
          Store command line context and options for future use
          if context is provided.
          """
          # Use lowercase class name as the default name
          self.name = name or self.__class__.__name__.lower()
          self.parent = parent
  
          # Relative log indent level shift against the parent
          self._relative_indent = relative_indent
  
          # Store command line context
          if context:
              self._save_context(context)
  
          # Initialize the workdir if requested
          self._workdir_load(workdir)
  
      def __str__(self) -> str:
          """ Name is the default string representation """
          return self.name
  
      @classmethod
      def _save_context(cls, context: click.Context) -> None:
          """ Save provided command line context and options for future use """
          cls._context = context
          cls._options = context.params
  
      @overload
      @classmethod
      def _opt(cls, option: str) -> Any:
          pass
  
      @overload
      @classmethod
      def _opt(cls, option: str, default: T) -> T:
          pass
  
      @classmethod
      def _opt(cls, option: str, default: Any = None) -> Any:
          """ Get an option from the command line context (class version) """
          return cls._options.get(option, default)
  
      def _fmf_context(self) -> FmfContextType:
          """ Return the current fmf context """
          if self._context is None:
              return dict()
  
          try:
              return cast(FmfContextType, self._context.obj.fmf_context)
          except AttributeError:
              return dict()
  
      def opt(self, option: str, default: Optional[Any] = None) -> Any:
          """
          Get an option from the command line options
  
          Checks also parent options. For flags (boolean values) parent's
          True wins over child's False (e.g. run --quiet enables quiet
          mode for all included plans and steps).
  
          For options that can be used multiple times, the child overrides
          the parent if it was defined (e.g. run -av provision -vvv runs
          all steps except for provision in mildly verbose mode, provision
          is run with the most verbosity).
  
          Environment variables override command line options.
          """
          # Translate dashes to underscores to match click's conversion
          option = option.replace('-', '_')
          # Check the environment first
          if option == 'debug':
              try:
                  debug = os.environ['TMT_DEBUG']
                  return int(debug)
              except ValueError:
                  raise GeneralError(
                      f"Invalid debug level '{debug}', use an integer.")
              except KeyError:
                  pass
  
          # Get local option
          local = self._options.get(option, default)
          # Check parent option
          parent = None
          if self.parent:
              parent = self.parent.opt(option)
          # Special handling for special flags (parent's yes always wins)
          if option in ['quiet', 'force', 'dry']:
              return parent if parent else local
          # Special handling for counting options (child overrides the
          # parent if it was defined)
          elif option in ['debug', 'verbose']:
              winner = local if local else parent
              if winner is None:
                  winner = 0
              return winner
          else:
              return parent if parent is not None else local
  
      def _level(self) -> int:
          """ Hierarchy level """
          if self.parent is None:
              return -1
          else:
              return self.parent._level() + self._relative_indent
  
      def _indent(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0) -> str:
          """ Indent message according to the object hierarchy """
  
          return indent(
              key,
              value=value,
              color=color,
              level=self._level() + shift)
  
      def _log(self, message: str) -> None:
          """ Append provided message to the current log """
          # Nothing to do if there is no workdir
          if self.workdir is None:
              return
  
          # Store log only in the top parent
          if self.parent:
              self.parent._log(message)
          else:
              with open(os.path.join(self.workdir, LOG_FILENAME), 'a') as log:
                  log.write(datetime.datetime.utcnow().strftime('%H:%M:%S') + ' '
                            + remove_color(message) + '\n')
  
      def print(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          """ Print a message regardless the quiet mode """
          self._log(self._indent(key, value, color=None, shift=shift))
          echo(self._indent(key, value, color, shift), err=err)
  
      def info(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          """ Show a message unless in quiet mode """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet'):
              echo(self._indent(key, value, color, shift), err=err)
  
      def verbose(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          """
          Show message if in requested verbose mode level
  
          In quiet mode verbose messages are not displayed.
          """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet') and self.opt('verbose') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def debug(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          """
          Show message if in requested debug mode level
  
          In quiet mode debug messages are not displayed.
          """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet') and self.opt('debug') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def warn(self, message: str, shift: int = 0) -> None:
          """ Show a yellow warning message on info level, send to stderr """
          self.info('warn', message, color='yellow', shift=shift, err=True)
  
      def fail(self, message: str, shift: int = 0) -> None:
          """ Show a red failure message on info level, send to stderr """
          self.info('fail', message, color='red', shift=shift, err=True)
  
      def _run(self,
               command: Union[str, List[str]],
               cwd: Optional[str],
               shell: bool,
               env: Optional[EnvironmentType],
               log: Optional[BaseLoggerFnType],
               join: bool = False,
               interactive: bool = False,
               timeout: Optional[int] = None) -> CommandOutput:
          """
          Run command, capture the output
  
          By default stdout and stderr are captured separately.
          Use join=True to merge stderr into stdout.
          Use timeout=<seconds> to finish process after given time
          """
          # By default command ouput is logged using debug
          if not log:
              log = self.debug
          # Prepare the environment
          if env:
              if not isinstance(env, dict):
                  raise GeneralError(f"Invalid environment '{env}'.")
              # Do not modify current process environment
              environment = os.environ.copy()
              environment.update(env)
          else:
              environment = None
          self.debug('environment', pprint.pformat(environment), level=4)
  
          # Run the command in interactive mode if requested
          if interactive:
              try:
                  subprocess.run(
                      command, cwd=cwd, shell=shell, env=environment, check=True)
              except subprocess.CalledProcessError:
                  # Interactive mode can return non-zero if the last command
                  # failed, ignore errors here
                  pass
              finally:
                  return CommandOutput(None, None)
  
          # Create the process
          try:
              process = subprocess.Popen(
                  command, cwd=cwd, shell=shell, env=environment,
                  stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
                  stderr=subprocess.STDOUT if join else subprocess.PIPE)
          except FileNotFoundError as error:
              raise RunError(
                  f"File '{error.filename}' not found.", command, 127)
  
          stdout_thread = StreamLogger(
              process.stdout, log_header='out', logger=log)
          stderr_thread = stdout_thread
          if not join:
              stderr_thread = StreamLogger(
                  process.stderr, log_header='err', logger=log)
          stdout_thread.start()
          if not join:
              stderr_thread.start()
          try:
              process.wait(timeout=timeout)
          except subprocess.TimeoutExpired:
              process.kill()
              process.returncode = PROCESS_TIMEOUT
          stdout_thread.join()
          if not join:
              stderr_thread.join()
  
  
  
  [... the rest is omitted ...]
files: tmt/utils.py
files: 
  """ Test Metadata Utilities """
  
  import contextlib
  import dataclasses
  import datetime
  import glob
  import io
  import os
  import pprint
  import re
  import shlex
  import shutil
  import subprocess
  import sys
  import unicodedata
  from collections import OrderedDict
  from functools import lru_cache
  from pathlib import Path
  from threading import Thread
  from typing import (IO, TYPE_CHECKING, Any, Dict, Generator, Iterable, List,
                      NamedTuple, Optional, Pattern, Tuple, Type, TypeVar, Union,
                      cast, overload)
  
  import click
  import fmf
  import requests
  from click import echo, style, wrap_text
  from requests.adapters import HTTPAdapter
  from requests.packages.urllib3.util.retry import Retry
  from ruamel.yaml import YAML, scalarstring
  from ruamel.yaml.comments import CommentedMap
  
  if sys.version_info >= (3, 8):
      from typing import Literal, Protocol
  else:
      from typing_extensions import Literal, Protocol
  
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  log = fmf.utils.Logging('tmt').logger
  
  # Default workdir root and max
  WORKDIR_ROOT = '/var/tmp/tmt'
  WORKDIR_MAX = 1000
  
  # Log in workdir
  LOG_FILENAME = 'log.txt'
  
  # Maximum number of lines of stdout/stderr to show upon errors
  OUTPUT_LINES = 100
  # Default output width
  OUTPUT_WIDTH = 79
  
  # Hierarchy indent
  INDENT = 4
  
  # Default name and order for step plugins
  DEFAULT_NAME = 'default'
  DEFAULT_PLUGIN_ORDER = 50
  DEFAULT_PLUGIN_ORDER_MULTIHOST = 10
  DEFAULT_PLUGIN_ORDER_REQUIRES = 70
  DEFAULT_PLUGIN_ORDER_RECOMMENDS = 75
  
  # Config directory
  CONFIG_PATH = '~/.config/tmt'
  
  # Special process return code
  PROCESS_TIMEOUT = 124
  
  # Default select.select(timeout) in seconds
  DEFAULT_SELECT_TIMEOUT = 5
  
  # Shell options to be set for all run shell scripts
  SHELL_OPTIONS = 'set -eo pipefail'
  
  # A stand-in variable for generic use.
  T = TypeVar('T')
  
  # A FMF context type, representing name/values context.
  FmfContextType = Dict[str, List[str]]
  
  # A "environment" type, representing name/value environment variables.
  EnvironmentType = Dict[str, str]
  
  # Workdir argument type, can be True, a string, a path or None
  WorkdirArgumentType = Union[Literal[True], str, None]
  
  # Workdir type, can be None or a string
  WorkdirType = Optional[str]
  
  # Option to skip to initialize work tree in plan
  PLAN_SKIP_WORKTREE_INIT = 'plan_skip_worktree_init'
  
  
  class BaseLoggerFnType(Protocol):
      def __call__(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          pass
  
  
  class LevelessLoggerFnType(Protocol):
      def __call__(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          pass
  
  
  class SemanticLoggerFnType(Protocol):
      def __call__(self, message: str, shift: int = 0) -> None:
          pass
  
  
  LoggerFnType = Union[
      BaseLoggerFnType,
      LevelessLoggerFnType,
      SemanticLoggerFnType]
  
  
  def indent(
          key: str,
          value: Optional[str] = None,
          color: Optional[str] = None,
          level: int = 0) -> str:
      """
      Indent a key/value message.
  
      If both ``key`` and ``value`` are specified, ``{key}: {value}``
      message is rendered. Otherwise, just ``key`` is used alone. If
      ``value`` contains multiple lines, each but the very first line is
      indented by one extra level.
  
      :param value: optional value to print at right side of ``key``.
      :param color: optional color to apply on ``key``.
      :param level: number of indentation levels. Each level is indented
                    by :py:data:`INDENT` spaces.
      """
  
      indent = ' ' * INDENT * level
      deeper = ' ' * INDENT * (level + 1)
  
      # Colorize
      if color is not None:
          key = style(key, fg=color)
  
      # Handle key only
      if value is None:
          message = key
  
      # Handle key + value
      else:
          # Multiline content indented deeper
          if isinstance(value, str):
              lines = value.splitlines()
              if len(lines) > 1:
                  value = ''.join([f"\n{deeper}{line}" for line in lines])
  
          message = f'{key}: {value}'
  
      return indent + message
  
  
  class Config(object):
      """ User configuration """
  
      def __init__(self) -> None:
          """ Initialize config directory path """
          self.path = os.path.expanduser(CONFIG_PATH)
          if not os.path.exists(self.path):
              try:
                  os.makedirs(self.path)
              except OSError as error:
                  raise GeneralError(
                      f"Failed to create config '{self.path}'.\n{error}")
  
      def last_run(self, run_id: Optional[str] = None) -> Optional[str]:
          """ Get and set last run id """
          symlink = os.path.join(self.path, 'last-run')
          if run_id:
              try:
                  os.remove(symlink)
              except OSError:
                  pass
              try:
                  os.symlink(run_id, symlink)
              except OSError as error:
                  raise GeneralError(
                      f"Unable to save last run '{self.path}'.\n{error}")
              return run_id
          if os.path.islink(symlink):
              return os.path.realpath(symlink)
          return None
  
  
  class StreamLogger(Thread):
      """
      Reading pipes of running process in threads.
  
      Code based on:
      https://github.com/packit/packit/blob/main/packit/utils/logging.py#L10
      """
  
      def __init__(self,
                   stream: Optional[IO[bytes]],
                   log_header: str,
                   logger: BaseLoggerFnType) -> None:
          super().__init__(daemon=True)
          self.stream = stream
          self.output: List[str] = []
          self.log_header = log_header
          self.logger = logger
  
      def run(self) -> None:
          if self.stream is None:
              return
  
          for _line in self.stream:
              line = _line.decode('utf-8', errors='replace')
              if line != '':
                  self.logger(
                      self.log_header,
                      line.rstrip('\n'),
                      'yellow',
                      level=3)
              self.output.append(line)
  
      def get_output(self) -> str:
          return "".join(self.output)
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Common
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  CommonDerivedType = TypeVar('CommonDerivedType', bound='Common')
  
  
  class CommandOutput(NamedTuple):
      stdout: Optional[str]
      stderr: Optional[str]
  
  
  class Common(object):
      """
      Common shared stuff
  
      Takes care of command line context, options and workdir handling.
      Provides logging functions info(), verbose() and debug().
      Implements read() and write() for comfortable file access.
      Provides the run() method for easy command execution.
      """
  
      # Command line context, options and workdir
      _context: Optional[click.Context] = None
      _options: Dict[str, Any] = dict()
      _workdir: WorkdirType = None
  
      def __init__(
              self,
              parent: Optional[CommonDerivedType] = None,
              name: Optional[str] = None,
              workdir: WorkdirArgumentType = None,
              context: Optional[click.Context] = None,
              relative_indent: int = 1):
          """
          Initialize name and relation with the parent object
  
          Prepare the workdir for provided id / directory path
          or generate a new workdir name if workdir=True given.
          Store command line context and options for future use
          if context is provided.
          """
          # Use lowercase class name as the default name
          self.name = name or self.__class__.__name__.lower()
          self.parent = parent
  
          # Relative log indent level shift against the parent
          self._relative_indent = relative_indent
  
          # Store command line context
          if context:
              self._save_context(context)
  
          # Initialize the workdir if requested
          self._workdir_load(workdir)
  
      def __str__(self) -> str:
          """ Name is the default string representation """
          return self.name
  
      @classmethod
      def _save_context(cls, context: click.Context) -> None:
          """ Save provided command line context and options for future use """
          cls._context = context
          cls._options = context.params
  
      @overload
      @classmethod
      def _opt(cls, option: str) -> Any:
          pass
  
      @overload
      @classmethod
      def _opt(cls, option: str, default: T) -> T:
          pass
  
      @classmethod
      def _opt(cls, option: str, default: Any = None) -> Any:
          """ Get an option from the command line context (class version) """
          return cls._options.get(option, default)
  
      def _fmf_context(self) -> FmfContextType:
          """ Return the current fmf context """
          if self._context is None:
              return dict()
  
          try:
              return cast(FmfContextType, self._context.obj.fmf_context)
          except AttributeError:
              return dict()
  
      def opt(self, option: str, default: Optional[Any] = None) -> Any:
          """
          Get an option from the command line options
  
          Checks also parent options. For flags (boolean values) parent's
          True wins over child's False (e.g. run --quiet enables quiet
          mode for all included plans and steps).
  
          For options that can be used multiple times, the child overrides
          the parent if it was defined (e.g. run -av provision -vvv runs
          all steps except for provision in mildly verbose mode, provision
          is run with the most verbosity).
  
          Environment variables override command line options.
          """
          # Translate dashes to underscores to match click's conversion
          option = option.replace('-', '_')
          # Check the environment first
          if option == 'debug':
              try:
                  debug = os.environ['TMT_DEBUG']
                  return int(debug)
              except ValueError:
                  raise GeneralError(
                      f"Invalid debug level '{debug}', use an integer.")
              except KeyError:
                  pass
  
          # Get local option
          local = self._options.get(option, default)
          # Check parent option
          parent = None
          if self.parent:
              parent = self.parent.opt(option)
          # Special handling for special flags (parent's yes always wins)
          if option in ['quiet', 'force', 'dry']:
              return parent if parent else local
          # Special handling for counting options (child overrides the
          # parent if it was defined)
          elif option in ['debug', 'verbose']:
              winner = local if local else parent
              if winner is None:
                  winner = 0
              return winner
          else:
              return parent if parent is not None else local
  
      def _level(self) -> int:
          """ Hierarchy level """
          if self.parent is None:
              return -1
          else:
              return self.parent._level() + self._relative_indent
  
      def _indent(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0) -> str:
          """ Indent message according to the object hierarchy """
  
          return indent(
              key,
              value=value,
              color=color,
              level=self._level() + shift)
  
      def _log(self, message: str) -> None:
          """ Append provided message to the current log """
          # Nothing to do if there is no workdir
          if self.workdir is None:
              return
  
          # Store log only in the top parent
          if self.parent:
              self.parent._log(message)
          else:
              with open(os.path.join(self.workdir, LOG_FILENAME), 'a') as log:
                  log.write(datetime.datetime.utcnow().strftime('%H:%M:%S') + ' '
                            + remove_color(message) + '\n')
  
      def print(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          """ Print a message regardless the quiet mode """
          self._log(self._indent(key, value, color=None, shift=shift))
          echo(self._indent(key, value, color, shift), err=err)
  
      def info(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          """ Show a message unless in quiet mode """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet'):
              echo(self._indent(key, value, color, shift), err=err)
  
      def verbose(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          """
          Show message if in requested verbose mode level
  
          In quiet mode verbose messages are not displayed.
          """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet') and self.opt('verbose') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def debug(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          """
          Show message if in requested debug mode level
  
          In quiet mode debug messages are not displayed.
          """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet') and self.opt('debug') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def warn(self, message: str, shift: int = 0) -> None:
          """ Show a yellow warning message on info level, send to stderr """
          self.info('warn', message, color='yellow', shift=shift, err=True)
  
      def fail(self, message: str, shift: int = 0) -> None:
          """ Show a red failure message on info level, send to stderr """
          self.info('fail', message, color='red', shift=shift, err=True)
  
      def _run(self,
               command: Union[str, List[str]],
               cwd: Optional[str],
               shell: bool,
               env: Optional[EnvironmentType],
               log: Optional[BaseLoggerFnType],
               join: bool = False,
               interactive: bool = False,
               timeout: Optional[int] = None) -> CommandOutput:
          """
          Run command, capture the output
  
          By default stdout and stderr are captured separately.
          Use join=True to merge stderr into stdout.
          Use timeout=<seconds> to finish process after given time
          """
          # By default command ouput is logged using debug
          if not log:
              log = self.debug
          # Prepare the environment
          if env:
              if not isinstance(env, dict):
                  raise GeneralError(f"Invalid environment '{env}'.")
              # Do not modify current process environment
              environment = os.environ.copy()
              environment.update(env)
          else:
              environment = None
          self.debug('environment', pprint.pformat(environment), level=4)
  
          # Run the command in interactive mode if requested
          if interactive:
              try:
                  subprocess.run(
                      command, cwd=cwd, shell=shell, env=environment, check=True)
              except subprocess.CalledProcessError:
                  # Interactive mode can return non-zero if the last command
                  # failed, ignore errors here
                  pass
              finally:
                  return CommandOutput(None, None)
  
          # Create the process
          try:
              process = subprocess.Popen(
                  command, cwd=cwd, shell=shell, env=environment,
                  stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
                  stderr=subprocess.STDOUT if join else subprocess.PIPE)
          except FileNotFoundError as error:
              raise RunError(
                  f"File '{error.filename}' not found.", command, 127)
  
          stdout_thread = StreamLogger(
              process.stdout, log_header='out', logger=log)
          stderr_thread = stdout_thread
          if not join:
              stderr_thread = StreamLogger(
                  process.stderr, log_header='err', logger=log)
          stdout_thread.start()
          if not join:
              stderr_thread.start()
          try:
              process.wait(timeout=timeout)
          except subprocess.TimeoutExpired:
              process.kill()
              process.returncode = PROCESS_TIMEOUT
          stdout_thread.join()
          if not join:
              stderr_thread.join()
  
          # Handle the exit code, return output
          if process.returncode != 0:
              raise RunError(
                  message=f"Command returned '{process.returncode}'.",
                  command=command,
                  returncode=process.returncode,
                  stdout=std
  
  [... the rest is omitted ...]
files: tmt/utils.py