files.new_contents: 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=
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/execute/internal.py
files.old_contents: 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=
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/execute/internal.py
files.new_contents: 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-back
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/artemis.py
files.old_contents: 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
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/artemis.py
files.new_contents: 
  """ 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):
  
  
  [... the rest is omitted ...]
files.new_path: tmt/utils.py
files.old_contents: 
  """ 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
  
  [... the rest is omitted ...]
files.old_path: tmt/utils.py