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',
  
  [... 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',
  
  [... 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
                  
  
  [... 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,
  
  
  [... 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'>
        <
  
  [... 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'>
        <
  
  [... 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 ``
  
  [... 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
  
  [... the rest is omitted ...]
files: tmt/utils.py