files: import contextlib
  import os
  import stat
  import sys
  import time
  
  import click
  
  import tmt
  import tmt.utils
  from tmt.steps.execute import TEST_OUTPUT_FILENAME
  from tmt.steps.provision import DEFAULT_RSYNC_OPTIONS
  from tmt.steps.provision.local import GuestLocal
  
  REBOOT_VARIABLES = (
      "TMT_REBOOT_COUNT",
      "REBOOTCOUNT",
      "RSTRNT_REBOOTCOUNT")
  REBOOT_SCRIPT_PATHS = (
      "/usr/local/bin/rstrnt-reboot",
      "/usr/local/bin/rhts-reboot",
      "/usr/local/bin/tmt-reboot")
  REBOOT_REQUEST_FILENAME = "reboot_request"
  REBOOT_SCRIPT = f"""\
  #!/bin/sh
  touch "$TMT_TEST_DATA/{REBOOT_REQUEST_FILENAME}"
  """
  REBOOT_BACKUP_EXT = ".tmt.backup"
  REBOOT_SETUP_NAME = "reboot_setup"
  REBOOT_SETUP_SCRIPT = f"""\
  #!/bin/sh
  for file in {" ".join(REBOOT_SCRIPT_PATHS)}; do
      # Backup existing file, keep track of what has been backed-up
      if [ -e "$file" ]; then
          mv "$file" "$file{REBOOT_BACKUP_EXT}"
          echo "$file"
      fi
      cp "$1" "$file" && chmod +x "$file"
  done
  """
  REBOOT_TEARDOWN_NAME = "reboot_teardown"
  REBOOT_TEARDOWN_SCRIPT = f"""\
  #!/bin/sh
  for file in {" ".join(REBOOT_SCRIPT_PATHS)}; do
      rm "$file"
  done
  
  # We pass back the backed up files
  for file in "$@"; do
      mv "$file{REBOOT_BACKUP_EXT}" "$file"
  done
  """
  REBOOT_TEMPLATE_NAME = "reboot_template"
  
  FILE_SUBMIT_SCRIPT = """\
  #!/bin/sh
  FILENAME="$2"
  [ -d "$TMT_TEST_DATA" ] || mkdir -p "$TMT_TEST_DATA"
  cp "$FILENAME" "$TMT_TEST_DATA"
  echo "File $FILENAME stored to $TMT_TEST_DATA"
  """
  FILE_SUBMIT_NAME = "tmt-file-submit"
  
  
  class ExecuteInternal(tmt.steps.execute.ExecutePlugin):
      """
      Use the internal tmt executor to execute tests
  
      The internal tmt executor runs tests on the guest one by one, shows
      testing progress and supports interactive debugging as well. Test
      result is based on the script exit code (for shell tests) or the
      results file (for beakerlib tests).
      """
  
      # Supported methods
      _methods = [
          tmt.steps.Method(name='tmt', doc=__doc__, order=50),
          tmt.steps.Method(name='shell.tmt', doc=__doc__, order=80),
          tmt.steps.Method(name='beakerlib.tmt', doc=__doc__, order=80),
          ]
  
      # Supported keys
      _keys = ["script", "interactive"]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          options = []
          # Shell script as a test
          options.append(click.option(
              '-s', '--script', metavar='SCRIPT', multiple=True,
              help='Shell script to be executed as a test.'))
          # Interactive mode
          options.append(click.option(
              '-i', '--interactive', is_flag=True,
              help='Run in interactive mode, do not capture output.'))
          # Disable interactive progress bar
          options.append(click.option(
              '--no-progress-bar', is_flag=True,
              help='Disable interactive progress bar showing the current test.'))
          return options + super().options(how)
  
      def wake(self, keys=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys)
          # Make sure that script is a list
          tmt.utils.listify(self.data, keys=['script'])
  
      # TODO: consider switching to utils.updatable_message() - might need more
      # work, since use of _show_progress is split over several methods.
      def _show_progress(self, progress, test_name, finish=False):
          """
          Show an interactive progress bar in non-verbose mode.
  
          If the output is not an interactive terminal, or progress bar is
          disabled using an option, just output the message as info without
          utilising \r. If finish is True, overwrite the previous progress bar.
          """
          # Verbose mode outputs other information, using \r to
          # create a status bar wouldn't work.
          if self.opt('verbose'):
              return
  
          # No progress if terminal not attached or explicitly disabled
          if not sys.stdout.isatty() or self.opt('no-progress-bar'):
              return
  
          # For debug mode show just an info message (unless finishing)
          message = f"{test_name} [{progress}]" if not finish else ""
          if self.opt('debug'):
              if not finish:
                  self.info(message, shift=1)
              return
  
          # Show progress bar in an interactive shell.
          try:
              # We need to completely override the previous message, add
              # spaces if necessary.
              message = message.ljust(len(self._previous_progress_message))
          except AttributeError:
              # First iteration, previous message not set
              pass
          self._previous_progress_message = message
          message = self._indent('progress', message, color='cyan')
          sys.stdout.write(f"\r{message}")
          if finish:
              # The progress has been overwritten, return back to the start
              sys.stdout.write(f"\r")
          sys.stdout.flush()
  
      def execute(self, test, guest, progress):
          """ Run test on the guest """
          # Provide info/debug message
          self._show_progress(progress, test.name)
          self.verbose(
              'test', test.summary or test.name, color='cyan', shift=1, level=2)
          self.debug(f"Execute '{test.name}' as a '{test.framework}' test.")
  
          # Test will be executed in the workdir
          workdir = os.path.join(
              self.step.plan.discover.workdir, test.path.lstrip('/'))
          self.debug(f"Use workdir '{workdir}'.", level=3)
  
          # Create data directory, prepare environment
          data_directory = self.data_path(test, full=True, create=True)
          environment = test.environment.copy()
          environment["TMT_TREE"] = self.parent.plan.worktree
          environment["TMT_TEST_DATA"] = os.path.join(
              data_directory, tmt.steps.execute.TEST_DATA)
          # Set all supported reboot variables
          for reboot_variable in REBOOT_VARIABLES:
              environment[reboot_variable] = str(test._reboot_count)
          # Variables related to beakerlib tests
          if test.framework == 'beakerlib':
              environment['BEAKERLIB_DIR'] = data_directory
              environment['BEAKERLIB_COMMAND_SUBMIT_LOG'] = (
                  f"bash {self.step.workdir}/{FILE_SUBMIT_NAME}")
  
          # Prepare the test command (use default options for shell tests)
          if test.framework == "shell":
              command = f"{tmt.utils.SHELL_OPTIONS}; {test.test}"
          else:
              command = test.test
  
          # Prepare custom function to log output in verbose mode
          def log(key, value=None, color=None, shift=1, level=1):
              self.verbose(key, value, color, shift=2, level=3)
  
          # Execute the test, save the output and return code
          timeout = ''
          start = time.time()
          try:
              stdout = guest.execute(
                  command, cwd=workdir, env=environment,
                  join=True, interactive=self.get('interactive'), log=log,
                  timeout=tmt.utils.duration_to_seconds(test.duration))
              test.returncode = 0
          except tmt.utils.RunError as error:
              stdout = error.stdout
              test.returncode = error.returncode
              if test.returncode == tmt.utils.PROCESS_TIMEOUT:
                  timeout = ' (timeout)'
                  self.debug(f"Test duration '{test.duration}' exceeded.")
          end = time.time()
          self.write(
              self.data_path(test, TEST_OUTPUT_FILENAME, full=True),
              stdout or '', mode='a', level=3)
          test.real_duration = self.test_duration(start, end)
          duration = click.style(test.real_duration, fg='cyan')
          shift = 1 if self.opt('verbose') < 2 else 2
          self.verbose(
              f"{duration} {test.name} [{progress}]{timeout}", shift=shift)
  
      def check(self, test):
          """ Check the test result """
          self.debug(f"Check result of '{test.name}'.")
          if test.framework == 'beakerlib':
              return self.check_beakerlib(test)
          else:
              return self.check_shell(test)
  
      def _setup_reboot_script(self, guest):
          """ Set up reboot script on the guest. """
          def setup_file(filename, content):
              file_path = os.path.join(self.workdir, filename)
              with open(file_path, 'w') as template:
                  template.write(content)
              # Make it executable, keep perms when pushing
              perms = os.stat(file_path)
              os.chmod(file_path, perms.st_mode | stat.S_IEXEC)
              return file_path
  
          template = setup_file(REBOOT_TEMPLATE_NAME, REBOOT_SCRIPT)
          setup = setup_file(REBOOT_SETUP_NAME, REBOOT_SETUP_SCRIPT)
          teardown = setup_file(REBOOT_TEARDOWN_NAME, REBOOT_TEARDOWN_SCRIPT)
          guest.push(self.workdir, options=DEFAULT_RSYNC_OPTIONS + ["-p"])
          return template, setup, teardown
  
      @contextlib.contextmanager
      def _setup_reboot(self, guest):
          """ Prepare the guest environment for potential reboot """
          # Ignore local provision
          if isinstance(guest, GuestLocal):
              yield
              return
          backed_up = []
          template, setup, teardown = None, None, None
          try:
              self.debug("Setup our reboot script implementations.", level=2)
              template, setup, teardown = self._setup_reboot_script(guest)
              try:
                  stdout = guest.execute(f"\"{setup}\" \"{template}\"")[0]
                  backed_up.extend(stdout.splitlines())
              except tmt.utils.RunError as error:
                  if "Read-only file system" not in error.stderr:
                      raise error
              yield
          finally:
              self.debug("Remove our reboot script implementations.", level=2)
              # FIXME: This part may not be executed if connection to the guest
              #        drops in the middle and the guest may be left in an
              #        inconsistent state.
              if teardown:
                  backed_up_files = " ".join(backed_up)
                  guest.execute(f"\"{teardown}\" {backed_up_files}")
  
      def _handle_reboot(self, test, guest):
          """
          Reboot the guest if the test requested it.
  
          Check for presence of a file signalling reboot request
          and orchestrate the reboot if it was requested. Also increment
          REBOOTCOUNT variable, reset it to 0 if no reboot was requested
          (going forward to the next test). Return whether reboot was done.
          """
          test_data = os.path.join(
              self.data_path(test, full=True), tmt.steps.execute.TEST_DATA)
          reboot_request_path = os.path.join(test_data, REBOOT_REQUEST_FILENAME)
          if os.path.exists(reboot_request_path):
              test._reboot_count += 1
              self.debug(f"Reboot during test '{test}' "
                         f"with reboot count {test
  
  [... the rest is omitted ...]
files: tmt/steps/execute/internal.py
files: import contextlib
  import os
  import stat
  import sys
  import time
  
  import click
  
  import tmt
  import tmt.utils
  from tmt.steps.execute import TEST_OUTPUT_FILENAME
  from tmt.steps.provision import DEFAULT_RSYNC_OPTIONS
  from tmt.steps.provision.local import GuestLocal
  
  REBOOT_VARIABLES = (
      "TMT_REBOOT_COUNT",
      "REBOOTCOUNT",
      "RSTRNT_REBOOTCOUNT")
  REBOOT_SCRIPT_PATHS = (
      "/usr/local/bin/rstrnt-reboot",
      "/usr/local/bin/rhts-reboot",
      "/usr/local/bin/tmt-reboot")
  REBOOT_REQUEST_FILENAME = "reboot_request"
  REBOOT_SCRIPT = f"""\
  #!/bin/sh
  touch "$TMT_TEST_DATA/{REBOOT_REQUEST_FILENAME}"
  """
  REBOOT_BACKUP_EXT = ".tmt.backup"
  REBOOT_SETUP_NAME = "reboot_setup"
  REBOOT_SETUP_SCRIPT = f"""\
  #!/bin/sh
  for file in {" ".join(REBOOT_SCRIPT_PATHS)}; do
      # Backup existing file, keep track of what has been backed-up
      if [ -e "$file" ]; then
          mv "$file" "$file{REBOOT_BACKUP_EXT}"
          echo "$file"
      fi
      cp "$1" "$file" && chmod +x "$file"
  done
  """
  REBOOT_TEARDOWN_NAME = "reboot_teardown"
  REBOOT_TEARDOWN_SCRIPT = f"""\
  #!/bin/sh
  for file in {" ".join(REBOOT_SCRIPT_PATHS)}; do
      rm "$file"
  done
  
  # We pass back the backed up files
  for file in "$@"; do
      mv "$file{REBOOT_BACKUP_EXT}" "$file"
  done
  """
  REBOOT_TEMPLATE_NAME = "reboot_template"
  
  FILE_SUBMIT_SCRIPT = """\
  #!/bin/sh
  FILENAME="$2"
  [ -d "$TMT_TEST_DATA" ] || mkdir -p "$TMT_TEST_DATA"
  cp "$FILENAME" "$TMT_TEST_DATA"
  echo "File $FILENAME stored to $TMT_TEST_DATA"
  """
  FILE_SUBMIT_NAME = "tmt-file-submit"
  
  
  class ExecuteInternal(tmt.steps.execute.ExecutePlugin):
      """
      Use the internal tmt executor to execute tests
  
      The internal tmt executor runs tests on the guest one by one, shows
      testing progress and supports interactive debugging as well. Test
      result is based on the script exit code (for shell tests) or the
      results file (for beakerlib tests).
      """
  
      # Supported methods
      _methods = [
          tmt.steps.Method(name='tmt', doc=__doc__, order=50),
          tmt.steps.Method(name='shell.tmt', doc=__doc__, order=80),
          tmt.steps.Method(name='beakerlib.tmt', doc=__doc__, order=80),
          ]
  
      # Supported keys
      _keys = ["script", "interactive"]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          options = []
          # Shell script as a test
          options.append(click.option(
              '-s', '--script', metavar='SCRIPT', multiple=True,
              help='Shell script to be executed as a test.'))
          # Interactive mode
          options.append(click.option(
              '-i', '--interactive', is_flag=True,
              help='Run in interactive mode, do not capture output.'))
          # Disable interactive progress bar
          options.append(click.option(
              '--no-progress-bar', is_flag=True,
              help='Disable interactive progress bar showing the current test.'))
          return options + super().options(how)
  
      def wake(self, keys=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys)
          # Make sure that script is a list
          tmt.utils.listify(self.data, keys=['script'])
  
      def _show_progress(self, progress, test_name, finish=False):
          """
          Show an interactive progress bar in non-verbose mode.
  
          If the output is not an interactive terminal, or progress bar is
          disabled using an option, just output the message as info without
          utilising \r. If finish is True, overwrite the previous progress bar.
          """
          # Verbose mode outputs other information, using \r to
          # create a status bar wouldn't work.
          if self.opt('verbose'):
              return
  
          # No progress if terminal not attached or explicitly disabled
          if not sys.stdout.isatty() or self.opt('no-progress-bar'):
              return
  
          # For debug mode show just an info message (unless finishing)
          message = f"{test_name} [{progress}]" if not finish else ""
          if self.opt('debug'):
              if not finish:
                  self.info(message, shift=1)
              return
  
          # Show progress bar in an interactive shell.
          try:
              # We need to completely override the previous message, add
              # spaces if necessary.
              message = message.ljust(len(self._previous_progress_message))
          except AttributeError:
              # First iteration, previous message not set
              pass
          self._previous_progress_message = message
          message = self._indent('progress', message, color='cyan')
          sys.stdout.write(f"\r{message}")
          if finish:
              # The progress has been overwritten, return back to the start
              sys.stdout.write(f"\r")
          sys.stdout.flush()
  
      def execute(self, test, guest, progress):
          """ Run test on the guest """
          # Provide info/debug message
          self._show_progress(progress, test.name)
          self.verbose(
              'test', test.summary or test.name, color='cyan', shift=1, level=2)
          self.debug(f"Execute '{test.name}' as a '{test.framework}' test.")
  
          # Test will be executed in the workdir
          workdir = os.path.join(
              self.step.plan.discover.workdir, test.path.lstrip('/'))
          self.debug(f"Use workdir '{workdir}'.", level=3)
  
          # Create data directory, prepare environment
          data_directory = self.data_path(test, full=True, create=True)
          environment = test.environment.copy()
          environment["TMT_TREE"] = self.parent.plan.worktree
          environment["TMT_TEST_DATA"] = os.path.join(
              data_directory, tmt.steps.execute.TEST_DATA)
          # Set all supported reboot variables
          for reboot_variable in REBOOT_VARIABLES:
              environment[reboot_variable] = str(test._reboot_count)
          # Variables related to beakerlib tests
          if test.framework == 'beakerlib':
              environment['BEAKERLIB_DIR'] = data_directory
              environment['BEAKERLIB_COMMAND_SUBMIT_LOG'] = (
                  f"bash {self.step.workdir}/{FILE_SUBMIT_NAME}")
  
          # Prepare the test command (use default options for shell tests)
          if test.framework == "shell":
              command = f"{tmt.utils.SHELL_OPTIONS}; {test.test}"
          else:
              command = test.test
  
          # Prepare custom function to log output in verbose mode
          def log(key, value=None, color=None, shift=1, level=1):
              self.verbose(key, value, color, shift=2, level=3)
  
          # Execute the test, save the output and return code
          timeout = ''
          start = time.time()
          try:
              stdout = guest.execute(
                  command, cwd=workdir, env=environment,
                  join=True, interactive=self.get('interactive'), log=log,
                  timeout=tmt.utils.duration_to_seconds(test.duration))
              test.returncode = 0
          except tmt.utils.RunError as error:
              stdout = error.stdout
              test.returncode = error.returncode
              if test.returncode == tmt.utils.PROCESS_TIMEOUT:
                  timeout = ' (timeout)'
                  self.debug(f"Test duration '{test.duration}' exceeded.")
          end = time.time()
          self.write(
              self.data_path(test, TEST_OUTPUT_FILENAME, full=True),
              stdout or '', mode='a', level=3)
          test.real_duration = self.test_duration(start, end)
          duration = click.style(test.real_duration, fg='cyan')
          shift = 1 if self.opt('verbose') < 2 else 2
          self.verbose(
              f"{duration} {test.name} [{progress}]{timeout}", shift=shift)
  
      def check(self, test):
          """ Check the test result """
          self.debug(f"Check result of '{test.name}'.")
          if test.framework == 'beakerlib':
              return self.check_beakerlib(test)
          else:
              return self.check_shell(test)
  
      def _setup_reboot_script(self, guest):
          """ Set up reboot script on the guest. """
          def setup_file(filename, content):
              file_path = os.path.join(self.workdir, filename)
              with open(file_path, 'w') as template:
                  template.write(content)
              # Make it executable, keep perms when pushing
              perms = os.stat(file_path)
              os.chmod(file_path, perms.st_mode | stat.S_IEXEC)
              return file_path
  
          template = setup_file(REBOOT_TEMPLATE_NAME, REBOOT_SCRIPT)
          setup = setup_file(REBOOT_SETUP_NAME, REBOOT_SETUP_SCRIPT)
          teardown = setup_file(REBOOT_TEARDOWN_NAME, REBOOT_TEARDOWN_SCRIPT)
          guest.push(self.workdir, options=DEFAULT_RSYNC_OPTIONS + ["-p"])
          return template, setup, teardown
  
      @contextlib.contextmanager
      def _setup_reboot(self, guest):
          """ Prepare the guest environment for potential reboot """
          # Ignore local provision
          if isinstance(guest, GuestLocal):
              yield
              return
          backed_up = []
          template, setup, teardown = None, None, None
          try:
              self.debug("Setup our reboot script implementations.", level=2)
              template, setup, teardown = self._setup_reboot_script(guest)
              try:
                  stdout = guest.execute(f"\"{setup}\" \"{template}\"")[0]
                  backed_up.extend(stdout.splitlines())
              except tmt.utils.RunError as error:
                  if "Read-only file system" not in error.stderr:
                      raise error
              yield
          finally:
              self.debug("Remove our reboot script implementations.", level=2)
              # FIXME: This part may not be executed if connection to the guest
              #        drops in the middle and the guest may be left in an
              #        inconsistent state.
              if teardown:
                  backed_up_files = " ".join(backed_up)
                  guest.execute(f"\"{teardown}\" {backed_up_files}")
  
      def _handle_reboot(self, test, guest):
          """
          Reboot the guest if the test requested it.
  
          Check for presence of a file signalling reboot request
          and orchestrate the reboot if it was requested. Also increment
          REBOOTCOUNT variable, reset it to 0 if no reboot was requested
          (going forward to the next test). Return whether reboot was done.
          """
          test_data = os.path.join(
              self.data_path(test, full=True), tmt.steps.execute.TEST_DATA)
          reboot_request_path = os.path.join(test_data, REBOOT_REQUEST_FILENAME)
          if os.path.exists(reboot_request_path):
              test._reboot_count += 1
              self.debug(f"Reboot during test '{test}' "
                         f"with reboot count {test._reboot_count}.")
              # Reset the file
              os.remove(reboot_request_path)
              guest.push(test_data)
              try:
                  guest.reboot()
  
  
  [... the rest is omitted ...]
files: tmt/steps/execute/internal.py
files: import datetime
  import time
  from typing import Any, Dict, Optional, cast
  
  import click
  import requests
  
  import tmt
  from tmt.utils import ProvisionError, updatable_message
  
  # TODO: find out how to get this one into RPM builds.
  try:
      from typing_extensions import TypedDict
  
  except ImportError:
      TypedDict = None
  
  
  # 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 = (
      '0.0.28', '0.0.32'
      )
  
  # The default Artemis API version - the most recent supported versions
  # should be perfectly fine.
  DEFAULT_API_VERSION = SUPPORTED_API_VERSIONS[-1]
  
  # Type annotation for "data" package describing a guest instance. Passed
  # between load() and save() calls.
  if TypedDict is None:
      GuestDataType = Dict[str, Any]
  
  else:
      GuestDataType = TypedDict(
          'DataType',
          {
              # 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(
      GuestDataType, {
          '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.
  if TypedDict is None:
      GuestInspectType = Dict[str, Any]
  
  else:
      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(
              self,
              request: requests.PreparedRequest,
              **kwargs: Any) -> requests.Response:    # type: ignore
          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 = requests.packages.urllib3.util.retry.Retry(
              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)
  
  
  class ProvisionArtemis(tmt.steps.provision.ProvisionPlugin):
      """
      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']
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for Artemis """
          return [
              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=[]
                 
  
  [... the rest is omitted ...]
files: tmt/steps/provision/artemis.py
files: import datetime
  import time
  from typing import Any, Dict, Optional, cast
  
  import click
  import requests
  
  import tmt
  from tmt.utils import ProvisionError
  
  # TODO: find out how to get this one into RPM builds.
  try:
      from typing_extensions import TypedDict
  
  except ImportError:
      TypedDict = None
  
  
  # 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 = (
      '0.0.28', '0.0.32'
      )
  
  # The default Artemis API version - the most recent supported versions
  # should be perfectly fine.
  DEFAULT_API_VERSION = SUPPORTED_API_VERSIONS[-1]
  
  # Type annotation for "data" package describing a guest instance. Passed
  # between load() and save() calls.
  if TypedDict is None:
      GuestDataType = Dict[str, Any]
  
  else:
      GuestDataType = TypedDict(
          'DataType',
          {
              # 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(
      GuestDataType, {
          '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
          }
      )
  
  
  # Type annotation for Artemis API `GET /guests/$guestname` response.
  # Partial, not all fields necessary since plugin ignores most of them.
  if TypedDict is None:
      GuestInspectType = Dict[str, Any]
  
  else:
      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(
              self,
              request: requests.PreparedRequest,
              **kwargs: Any) -> requests.Response:    # type: ignore
          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 = requests.packages.urllib3.util.retry.Retry(
              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)
  
  
  class ProvisionArtemis(tmt.steps.provision.ProvisionPlugin):
      """
      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']
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for Artemis """
          return [
              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'
  
  [... the rest is omitted ...]
files: tmt/steps/provision/artemis.py
files: 
  """ Test Metadata Utilities """
  
  import contextlib
  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 Dict, Iterable, Optional
  
  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
  
  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'
  
  
  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):
          """ 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=None):
          """ 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, log_header, logger):
          super().__init__(daemon=True)
          self.stream = stream
          self.output = []
          self.log_header = log_header
          self.logger = logger
  
      def run(self):
          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):
          return "".join(self.output)
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Common
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  class Common(object):
      """
      Common shared stuff
  
      Takes care of command line context 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 and workdir
      _context = None
      _workdir = None
  
      def __init__(self, parent=None, name=None, workdir=None, context=None):
          """
          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 for future use if provided.
          """
          # Use lowercase class name as the default name
          self.name = name or self.__class__.__name__.lower()
          self.parent = parent
  
          # Store command line context
          if context:
              self._context = context
  
          # Initialize the workdir if requested
          self._workdir_load(workdir)
  
      def __str__(self):
          """ Name is the default string representation """
          return self.name
  
      @classmethod
      def _save_context(cls, context):
          """ Save provided command line context for future use """
          cls._context = context
  
      @classmethod
      def _opt(cls, option, default=None):
          """ Get an option from the command line context (class version) """
          if cls._context is None:
              return default
          return cls._context.params.get(option, default)
  
      def _fmf_context(self):
          """ Return the current fmf contex """
          try:
              return self._context.obj.fmf_context
          except AttributeError:
              return dict()
  
      def opt(self, option, default=None):
          """
          Get an option from the command line context
  
          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
  
          # Check local option
          local = default
          if self._context is not None:
              local = self._context.params.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):
          """ Hierarchy level """
          if self.parent is None:
              return -1
          else:
              return self.parent._level() + 1
  
      def _indent(self, key, value=None, color=None, shift=0):
          """ Indent message according to the object hierarchy """
  
          return indent(
              key,
              value=value,
              color=color,
              level=self._level() + shift)
  
      def _log(self, message):
          """ 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, value=None, color=None, shift=0, err=False):
          """ 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, value=None, color=None, shift=0, err=False):
          """ 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 warn(self, message, shift=0):
          """ 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, shift=0):
          """ Show a red failure message on info level, send to stderr """
          self.info('fail', message, color='red', shift=shift, err=True)
  
      def verbose(
              self, key, value=None, color=None, shift=0, level=1, err=False):
          """ Show message if in requested verbose mode level """
          self._log(self._indent(key, value, color=None, shift=shift))
          if self.opt('verbose') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def debug(self, key, value=None, color=None, shift=1, level=1, err=False):
          """ Show message if in requested debug mode level """
          self._log(self._indent(key, value, color=None, shift=shift))
          if self.opt('debug') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def _run(
              self, command, cwd, shell, env, log, join=False, interactive=False,
              timeout=None):
          """
          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:
  
  
  [... the rest is omitted ...]
files: tmt/utils.py
files: 
  """ Test Metadata Utilities """
  
  import contextlib
  import datetime
  import glob
  import io
  import os
  import pprint
  import re
  import shlex
  import shutil
  import subprocess
  import unicodedata
  from collections import OrderedDict
  from functools import lru_cache
  from pathlib import Path
  from threading import Thread
  from typing import Dict, Iterable
  
  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
  
  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'
  
  
  class Config(object):
      """ User configuration """
  
      def __init__(self):
          """ 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=None):
          """ 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, log_header, logger):
          super().__init__(daemon=True)
          self.stream = stream
          self.output = []
          self.log_header = log_header
          self.logger = logger
  
      def run(self):
          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):
          return "".join(self.output)
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Common
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  class Common(object):
      """
      Common shared stuff
  
      Takes care of command line context 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 and workdir
      _context = None
      _workdir = None
  
      def __init__(self, parent=None, name=None, workdir=None, context=None):
          """
          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 for future use if provided.
          """
          # Use lowercase class name as the default name
          self.name = name or self.__class__.__name__.lower()
          self.parent = parent
  
          # Store command line context
          if context:
              self._context = context
  
          # Initialize the workdir if requested
          self._workdir_load(workdir)
  
      def __str__(self):
          """ Name is the default string representation """
          return self.name
  
      @classmethod
      def _save_context(cls, context):
          """ Save provided command line context for future use """
          cls._context = context
  
      @classmethod
      def _opt(cls, option, default=None):
          """ Get an option from the command line context (class version) """
          if cls._context is None:
              return default
          return cls._context.params.get(option, default)
  
      def _fmf_context(self):
          """ Return the current fmf contex """
          try:
              return self._context.obj.fmf_context
          except AttributeError:
              return dict()
  
      def opt(self, option, default=None):
          """
          Get an option from the command line context
  
          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
  
          # Check local option
          local = default
          if self._context is not None:
              local = self._context.params.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):
          """ Hierarchy level """
          if self.parent is None:
              return -1
          else:
              return self.parent._level() + 1
  
      def _indent(self, key, value=None, color=None, shift=0):
          """ Indent message according to the object hierarchy """
          level = self._level() + shift
          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
  
      def _log(self, message):
          """ 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, value=None, color=None, shift=0, err=False):
          """ 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, value=None, color=None, shift=0, err=False):
          """ 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 warn(self, message, shift=0):
          """ 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, shift=0):
          """ Show a red failure message on info level, send to stderr """
          self.info('fail', message, color='red', shift=shift, err=True)
  
      def verbose(
              self, key, value=None, color=None, shift=0, level=1, err=False):
          """ Show message if in requested verbose mode level """
          self._log(self._indent(key, value, color=None, shift=shift))
          if self.opt('verbose') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def debug(self, key, value=None, color=None, shift=1, level=1, err=False):
          """ Show message if in requested debug mode level """
          self._log(self._indent(key, value, color=None, shift=shift))
          if self.opt('debug') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def _run(
              self, command, cwd, shell, env, log, join=False, interactive=False,
              timeout=None):
          """
          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 as error:
                  # Interactive mode can return non-zero if the last command
                  # failed, ignore errors here
                  pass
              finally:
                  return None if join else (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(
  
  
  [... the rest is omitted ...]
files: tmt/utils.py