files: # See https://pre-commit.com for more information
  # See https://pre-commit.com/hooks.html for more hooks
  repos:
    - repo: https://github.com/pre-commit/mirrors-autopep8
      rev: 'v1.5.7'
      hooks:
        - id: autopep8
          args:
            - --recursive
            - --in-place
            - --aggressive
            - --aggressive
            - --hang-closing
            - --max-line-length=99
  
    - repo: https://github.com/PyCQA/isort
      rev: "5.8.0"
      hooks:
        - id: isort
  
    - repo: https://github.com/pycqa/flake8
      rev: "4.0.1"
      hooks:
        - id: flake8
          args:
            - --max-line-length=99
          files: >
            (?x)^(
              examples/.*|
              tests/.*|
              tmt/.*
            )$
  
    - repo: https://github.com/pre-commit/pre-commit-hooks
      rev: "v3.4.0"
      hooks:
        - id: end-of-file-fixer
        - id: mixed-line-ending
        - id: trailing-whitespace
  
    - repo: https://github.com/pre-commit/mirrors-mypy
      rev: "v0.961"
      hooks:
        - id: mypy
          additional_dependencies: [types-click]
          pass_filenames: false
          args:
            - --config-file
            - mypy.ini
  
    - repo: https://github.com/python-jsonschema/check-jsonschema
      rev: 0.15.0
      hooks:
        - id: check-metaschema
          name: "Check JSON schemas validity"
          files: ^tmt/schemas/.*\.yaml
  
    - repo: https://github.com/adrienverge/yamllint
      rev: v1.26.3
      hooks:
        - id: yamllint
          files: ^tmt/schemas/.*\.yaml
  
    # Yet another static analysis - these hooks use regular expressions to
    # process Python code, and offer interesting "metalinters", checks for
    # what we do to appease flake8 and mypy linters.
    - repo: https://github.com/pre-commit/pygrep-hooks
      rev: v1.9.0
      hooks:
        # Enforce `noqa` and `type: ignore` to always appear with specific
        # error code(s).
        - id: python-check-blanket-noqa
        - id: python-check-blanket-type-ignore
  
        # Other potentially useful hooks for future consideration:
        #
        # - id: python-check-mock-methods
        # - id: python-no-eval
        # - id: python-no-log-warn
        # - id: python-use-type-annotations
        # - id: text-unicode-replacement-char
  
  
  [... the rest is omitted ...]
files: .pre-commit-config.yaml
files: # See https://pre-commit.com for more information
  # See https://pre-commit.com/hooks.html for more hooks
  repos:
    - repo: https://github.com/pre-commit/mirrors-autopep8
      rev: 'v1.5.7'
      hooks:
        - id: autopep8
          args:
            - --recursive
            - --in-place
            - --aggressive
            - --aggressive
            - --hang-closing
            - --max-line-length=99
  
    - repo: https://github.com/PyCQA/isort
      rev: "5.8.0"
      hooks:
        - id: isort
  
    - repo: https://github.com/pycqa/flake8
      rev: "4.0.1"
      hooks:
        - id: flake8
          args:
            - --max-line-length=99
          files: >
            (?x)^(
              examples/.*|
              tests/.*|
              tmt/.*
            )$
  
    - repo: https://github.com/pre-commit/pre-commit-hooks
      rev: "v3.4.0"
      hooks:
        - id: end-of-file-fixer
        - id: mixed-line-ending
        - id: trailing-whitespace
  
    - repo: https://github.com/pre-commit/mirrors-mypy
      rev: "v0.961"
      hooks:
        - id: mypy
          additional_dependencies: [types-click]
          pass_filenames: false
          args:
            - --config-file
            - mypy.ini
  
    - repo: https://github.com/python-jsonschema/check-jsonschema
      rev: 0.15.0
      hooks:
        - id: check-metaschema
          name: "Check JSON schemas validity"
          files: ^tmt/schemas/.*\.yaml
  
    - repo: https://github.com/adrienverge/yamllint
      rev: v1.26.3
      hooks:
        - id: yamllint
          files: ^tmt/schemas/.*\.yaml
  
  
  [... the rest is omitted ...]
files: .pre-commit-config.yaml
files: # coding: utf-8
  
  """ Command line interface for the Test Management Tool """
  
  import collections
  import dataclasses
  import os
  import subprocess
  import sys
  from typing import TYPE_CHECKING, Any, Callable, List, Optional, Set
  
  import click
  import fmf
  from click import echo, style
  from fmf.utils import listed
  
  import tmt
  import tmt.convert
  import tmt.export
  import tmt.identifier
  import tmt.options
  import tmt.plugins
  import tmt.steps
  import tmt.templates
  import tmt.utils
  
  if TYPE_CHECKING:
      import tmt.steps.discover
      import tmt.steps.execute
  
  # Explore available plugins (need to detect all supported methods first)
  tmt.plugins.explore()
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Click Context Object Container
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  
  @dataclasses.dataclass
  class ContextObject:
      """ Click Context Object Container """
      common: tmt.utils.Common
      fmf_context: tmt.utils.FmfContextType
      steps: Set[tmt.steps.Step]
      tree: tmt.Tree
      clean: Optional[tmt.Clean] = None
      run: Optional[tmt.Run] = None
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Custom Group
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  class CustomGroup(click.Group):
      """ Custom Click Group """
  
      def list_commands(self, context: click.core.Context) -> List[str]:
          """ Prevent alphabetical sorting """
          return list(self.commands.keys())
  
      def get_command(self, context: click.core.Context, cmd_name: str) -> Optional[click.Command]:
          """ Allow command shortening """
          # Backward-compatible 'test convert' (just temporary for now FIXME)
          cmd_name = cmd_name.replace('convert', 'import')
          # Support both story & stories
          cmd_name = cmd_name.replace('story', 'stories')
          found = click.Group.get_command(self, context, cmd_name)
          if found is not None:
              return found
          matches = [command for command in self.list_commands(context)
                     if command.startswith(cmd_name)]
          if not matches:
              return None
          elif len(matches) == 1:
              return click.Group.get_command(self, context, matches[0])
          context.fail('Did you mean {}?'.format(
              listed(sorted(matches), join='or')))
          return None
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Common Options
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  def verbose_debug_quiet(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Verbose, debug and quiet output """
      for option in reversed(tmt.options.verbose_debug_quiet):
          function = option(function)
      return function
  
  
  def force_dry(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Force and dry actions """
      for option in reversed(tmt.options.force_dry):
          function = option(function)
      return function
  
  
  def fix(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Fix action """
      function = tmt.options.fix(function)
      return function
  
  
  def workdir_root(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Where tmt workdirs are to be found """
      return tmt.options.workdir_root(function)
  
  
  def name_filter_condition(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Common filter options (short & long) """
      options = [
          click.argument(
              'names', nargs=-1, metavar='[REGEXP|.]'),
          click.option(
              '-f', '--filter', 'filters', metavar='FILTER', multiple=True,
              help="Apply advanced filter (see 'pydoc fmf.filter')."),
          click.option(
              '-c', '--condition', 'conditions', metavar="EXPR", multiple=True,
              help="Use arbitrary Python expression for filtering."),
          click.option(
              '--enabled', is_flag=True,
              help="Show only enabled tests, plans or stories."),
          click.option(
              '--disabled', is_flag=True,
              help="Show only disabled tests, plans or stories."),
          click.option(
              '--link', 'links', metavar="RELATION:TARGET", multiple=True,
              help="Filter by linked objects (regular expressions are "
                   "supported for both relation and target)."),
          click.option(
              '-x', '--exclude', 'exclude', metavar='[REGEXP]', multiple=True,
              help="Exclude a regular expression from search result."),
          ]
  
      for option in reversed(options):
          function = option(function)
      return function
  
  
  def source(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Option to select fmf objects by their sources """
      return click.option(
          '--source', is_flag=True, help="Select by fmf source file names instead of object names."
          )(function)
  
  
  def name_filter_condition_long(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Common filter options (long only) """
      options = [
          click.
  
  [... the rest is omitted ...]
files: tmt/cli.py
files: # coding: utf-8
  
  """ Command line interface for the Test Management Tool """
  
  import collections
  import dataclasses
  import os
  import subprocess
  import sys
  from typing import TYPE_CHECKING, Any, Callable, List, Optional, Set
  
  import click
  import fmf
  from click import echo, style
  from fmf.utils import listed
  
  import tmt
  import tmt.convert
  import tmt.export
  import tmt.identifier
  import tmt.options
  import tmt.plugins
  import tmt.steps
  import tmt.templates
  import tmt.utils
  
  if TYPE_CHECKING:
      import tmt.steps.discover
      import tmt.steps.execute
  
  # Explore available plugins (need to detect all supported methods first)
  tmt.plugins.explore()
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Click Context Object Container
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  
  @dataclasses.dataclass
  class ContextObject:
      """ Click Context Object Container """
      common: tmt.utils.Common
      fmf_context: tmt.utils.FmfContextType
      steps: Set[tmt.steps.Step]
      tree: tmt.Tree
      clean: Optional[tmt.Clean] = None
      run: Optional[tmt.Run] = None
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Custom Group
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  class CustomGroup(click.Group):
      """ Custom Click Group """
  
      def list_commands(self, context: click.core.Context) -> List[str]:
          """ Prevent alphabetical sorting """
          return list(self.commands.keys())
  
      def get_command(self, context: click.core.Context, cmd_name: str) -> Optional[click.Command]:
          """ Allow command shortening """
          # Backward-compatible 'test convert' (just temporary for now FIXME)
          cmd_name = cmd_name.replace('convert', 'import')
          # Support both story & stories
          cmd_name = cmd_name.replace('story', 'stories')
          found = click.Group.get_command(self, context, cmd_name)
          if found is not None:
              return found
          matches = [command for command in self.list_commands(context)
                     if command.startswith(cmd_name)]
          if not matches:
              return None
          elif len(matches) == 1:
              return click.Group.get_command(self, context, matches[0])
          context.fail('Did you mean {}?'.format(
              listed(sorted(matches), join='or')))
          return None
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Common Options
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  def verbose_debug_quiet(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Verbose, debug and quiet output """
      for option in reversed(tmt.options.verbose_debug_quiet):
          function = option(function)
      return function
  
  
  def force_dry(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Force and dry actions """
      for option in reversed(tmt.options.force_dry):
          function = option(function)
      return function
  
  
  def fix(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Fix action """
      function = tmt.options.fix(function)
      return function
  
  
  def workdir_root(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Where tmt workdirs are to be found """
      return tmt.options.workdir_root(function)
  
  
  def name_filter_condition(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Common filter options (short & long) """
      options = [
          click.argument(
              'names', nargs=-1, metavar='[REGEXP|.]'),
          click.option(
              '-f', '--filter', 'filters', metavar='FILTER', multiple=True,
              help="Apply advanced filter (see 'pydoc fmf.filter')."),
          click.option(
              '-c', '--condition', 'conditions', metavar="EXPR", multiple=True,
              help="Use arbitrary Python expression for filtering."),
          click.option(
              '--enabled', is_flag=True,
              help="Show only enabled tests, plans or stories."),
          click.option(
              '--disabled', is_flag=True,
              help="Show only disabled tests, plans or stories."),
          click.option(
              '--link', 'links', metavar="RELATION:TARGET", multiple=True,
              help="Filter by linked objects (regular expressions are "
                   "supported for both relation and target)."),
          click.option(
              '-x', '--exclude', 'exclude', metavar='[REGEXP]', multiple=True,
              help="Exclude a regular expression from search result."),
          ]
  
      for option in reversed(options):
          function = option(function)
      return function
  
  
  def source(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Option to select fmf objects by their sources """
      return click.option(
          '--source', is_flag=True, help="Select by fmf source file names instead of object names."
          )(function)
  
  
  def name_filter_condition_long(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Common filter options (long only) """
      options = [
          click.
  
  [... the rest is omitted ...]
files: tmt/cli.py
files: # coding: utf-8
  
  """ Export metadata into nitrate """
  
  
  import email.utils
  import os
  import re
  import traceback
  import types
  import xmlrpc.client
  from functools import lru_cache
  from typing import Any, Dict, Generator, List, Optional, Tuple, Union, cast
  
  import fmf
  from click import echo, secho, style
  
  import tmt
  import tmt.utils
  from tmt.convert import SYSTEM_OTHER, add_link
  from tmt.identifier import ID_KEY, add_uuid_if_not_defined
  from tmt.utils import ConvertError, check_git_url, markdown_to_html
  
  bugzilla: Optional[types.ModuleType] = None
  gssapi: Optional[types.ModuleType] = None
  nitrate: Optional[types.ModuleType] = None
  
  PolarionException: Any = None
  PolarionTestCase: Any = None
  PolarionWorkItem: Any = None
  
  DEFAULT_PRODUCT: Any = None
  
  SectionsReturnType = Tuple[str, str, str, str]
  HeadingsType = List[List[Union[int, str]]]
  SectionsHeadingsType = Dict[str, HeadingsType]
  
  log = fmf.utils.Logging('tmt').logger
  
  WARNING = """
  Test case has been migrated to git. Any changes made here might be overwritten.
  See: https://tmt.readthedocs.io/en/latest/questions.html#nitrate-migration
  """.lstrip()
  
  # For linking bugs
  BUGZILLA_XMLRPC_URL = "https://bugzilla.redhat.com/xmlrpc.cgi"
  NITRATE_TRACKER_ID = 69  # ID of nitrate in RH's bugzilla
  POLARION_TRACKER_ID = 117  # ID of polarion in RH's bugzilla
  RE_BUGZILLA_URL = r'bugzilla.redhat.com/show_bug.cgi\?id=(\d+)'
  RE_POLARION_URL = r'.*/polarion/#/project/.*/workitem\?id=(.*)'
  LEGACY_POLARION_PROJECTS = set(['RedHatEnterpriseLinux7'])
  
  
  def import_nitrate() -> Any:
      """ Conditionally import the nitrate module """
      # Need to import nitrate only when really needed. Otherwise we get
      # traceback when nitrate not installed or config file not available.
      # And we want to keep the core tmt package with minimal dependencies.
      try:
          global nitrate, DEFAULT_PRODUCT, gssapi
          import gssapi
          import nitrate
          assert nitrate
          DEFAULT_PRODUCT = nitrate.Product(name='RHEL Tests')
          return nitrate
      except ImportError:
          raise ConvertError(
              "Install tmt-test-convert to export tests to nitrate.")
      except nitrate.NitrateError as error:  # type: ignore[union-attr]  # nitrate is no longer None
          raise ConvertError(error)
  
  
  def import_polarion() -> None:
      """ Import polarion python api - pylero """
      try:
          global PolarionException, PolarionTestCase, PolarionWorkItem
          from pylero.exceptions import PyleroLibException as PolarionException
          from pylero.work_item import TestCase as PolarionTestCase
          from pylero.work_item import _WorkItem as PolarionWorkItem
      except ImportError:
          raise ConvertError(
              "Run 'pip install tmt[export-polarion]' so pylero is installed.")
      except PolarionException as exc:
          log.debug(traceback.format_exc())
          raise ConvertError("Failed to login with pylero", original=exc)
  
  
  def get_bz_instance() -> Any:
      """ Import the bugzilla module and return BZ instance """
      try:
          import bugzilla
      except ImportError:
          raise ConvertError(
              "Install 'tmt-test-convert' to link test to the bugzilla.")
  
      try:
          bz_instance = bugzilla.Bugzilla(url=BUGZILLA_XMLRPC_URL)
      except Exception as exc:
          log.debug(traceback.format_exc())
          raise ConvertError(
              "Couldn't initialize the Bugzilla client.", original=exc)
  
      if not bz_instance.logged_in:
          raise ConvertError(
              "Not logged to Bugzilla, check 'man bugzilla' section "
              "'AUTHENTICATION CACHE AND API KEYS'.")
      return bz_instance
  
  
  def _nitrate_find_fmf_testcases(test: 'tmt.Test') -> Generator[Any, None, None]:
      """
      Find all Nitrate test cases with the same fmf identifier
  
      All component general plans are explored for possible duplicates.
      """
      assert nitrate
      for component in test.component:
          try:
              for testcase in find_general_plan(component).testcases:
                  struct_field = tmt.utils.StructuredField(testcase.notes)
                  try:
                      fmf_id = tmt.utils.yaml_to_dict(struct_field.get('fmf'))
                      if fmf_id == test.fmf_id:
                          echo(style(
                              f"Existing test case '{testcase.identifier}' "
                              f"found for given fmf id.", fg='magenta'))
                          yield testcase
                  except tmt.utils.StructuredFieldError:
                      pass
          except nitrate.Nit
  
  [... the rest is omitted ...]
files: tmt/export.py
files: # coding: utf-8
  
  """ Export metadata into nitrate """
  
  
  import email.utils
  import os
  import re
  import traceback
  import types
  import xmlrpc.client
  from functools import lru_cache
  from typing import Any, Dict, Generator, List, Optional, Tuple, Union, cast
  
  import fmf
  from click import echo, secho, style
  
  import tmt
  import tmt.utils
  from tmt.convert import SYSTEM_OTHER, add_link
  from tmt.identifier import ID_KEY, add_uuid_if_not_defined
  from tmt.utils import ConvertError, check_git_url, markdown_to_html
  
  bugzilla: Optional[types.ModuleType] = None
  gssapi: Optional[types.ModuleType] = None
  nitrate: Optional[types.ModuleType] = None
  
  PolarionException: Any = None
  PolarionTestCase: Any = None
  PolarionWorkItem: Any = None
  
  DEFAULT_PRODUCT: Any = None
  
  SectionsReturnType = Tuple[str, str, str, str]
  HeadingsType = List[List[Union[int, str]]]
  SectionsHeadingsType = Dict[str, HeadingsType]
  
  log = fmf.utils.Logging('tmt').logger
  
  WARNING = """
  Test case has been migrated to git. Any changes made here might be overwritten.
  See: https://tmt.readthedocs.io/en/latest/questions.html#nitrate-migration
  """.lstrip()
  
  # For linking bugs
  BUGZILLA_XMLRPC_URL = "https://bugzilla.redhat.com/xmlrpc.cgi"
  NITRATE_TRACKER_ID = 69  # ID of nitrate in RH's bugzilla
  POLARION_TRACKER_ID = 117  # ID of polarion in RH's bugzilla
  RE_BUGZILLA_URL = r'bugzilla.redhat.com/show_bug.cgi\?id=(\d+)'
  RE_POLARION_URL = r'.*/polarion/#/project/.*/workitem\?id=(.*)'
  LEGACY_POLARION_PROJECTS = set(['RedHatEnterpriseLinux7'])
  
  
  def import_nitrate() -> Any:
      """ Conditionally import the nitrate module """
      # Need to import nitrate only when really needed. Otherwise we get
      # traceback when nitrate not installed or config file not available.
      # And we want to keep the core tmt package with minimal dependencies.
      try:
          global nitrate, DEFAULT_PRODUCT, gssapi
          import gssapi
          import nitrate
          assert nitrate
          DEFAULT_PRODUCT = nitrate.Product(name='RHEL Tests')
          return nitrate
      except ImportError:
          raise ConvertError(
              "Install tmt-test-convert to export tests to nitrate.")
      except nitrate.NitrateError as error:  # type: ignore
          raise ConvertError(error)
  
  
  def import_polarion() -> None:
      """ Import polarion python api - pylero """
      try:
          global PolarionException, PolarionTestCase, PolarionWorkItem
          from pylero.exceptions import PyleroLibException as PolarionException
          from pylero.work_item import TestCase as PolarionTestCase
          from pylero.work_item import _WorkItem as PolarionWorkItem
      except ImportError:
          raise ConvertError(
              "Run 'pip install tmt[export-polarion]' so pylero is installed.")
      except PolarionException as exc:
          log.debug(traceback.format_exc())
          raise ConvertError("Failed to login with pylero", original=exc)
  
  
  def get_bz_instance() -> Any:
      """ Import the bugzilla module and return BZ instance """
      try:
          import bugzilla
      except ImportError:
          raise ConvertError(
              "Install 'tmt-test-convert' to link test to the bugzilla.")
  
      try:
          bz_instance = bugzilla.Bugzilla(url=BUGZILLA_XMLRPC_URL)
      except Exception as exc:
          log.debug(traceback.format_exc())
          raise ConvertError(
              "Couldn't initialize the Bugzilla client.", original=exc)
  
      if not bz_instance.logged_in:
          raise ConvertError(
              "Not logged to Bugzilla, check 'man bugzilla' section "
              "'AUTHENTICATION CACHE AND API KEYS'.")
      return bz_instance
  
  
  def _nitrate_find_fmf_testcases(test: 'tmt.Test') -> Generator[Any, None, None]:
      """
      Find all Nitrate test cases with the same fmf identifier
  
      All component general plans are explored for possible duplicates.
      """
      assert nitrate
      for component in test.component:
          try:
              for testcase in find_general_plan(component).testcases:
                  struct_field = tmt.utils.StructuredField(testcase.notes)
                  try:
                      fmf_id = tmt.utils.yaml_to_dict(struct_field.get('fmf'))
                      if fmf_id == test.fmf_id:
                          echo(style(
                              f"Existing test case '{testcase.identifier}' "
                              f"found for given fmf id.", fg='magenta'))
                          yield testcase
                  except tmt.utils.StructuredFieldError:
                      pass
          except nitrate.NitrateError:
              pass
  
  
  def convert_manual_to_n
  
  [... the rest is omitted ...]
files: tmt/export.py
files: import tmt
  import tmt.steps
  import tmt.steps.finish
  import tmt.steps.prepare.ansible
  from tmt.steps.prepare.ansible import PrepareAnsible
  
  
  @tmt.steps.provides_method('ansible')
  class FinishAnsible(tmt.steps.finish.FinishPlugin, PrepareAnsible):
      """
      Perform finishing tasks using ansible
  
      Single playbook config:
  
          finish:
              how: ansible
              playbook: ansible/packages.yml
  
      Multiple playbooks config:
  
          finish:
              how: ansible
              playbook:
                - playbook/one.yml
                - playbook/two.yml
                - playbook/three.yml
  
      The playbook path should be relative to the metadata tree root.
      Use 'order' attribute to select in which order finishing tasks
      should happen if there are multiple configs. Default order is '50'.
      """
  
      _data_class = tmt.steps.prepare.ansible.PrepareAnsibleData
  
      # Assigning class methods seems to cause trouble to mypy
      # See also: https://github.com/python/mypy/issues/6700
      base_command = tmt.steps.finish.FinishPlugin.base_command  # type: ignore[assignment]
  
  
  [... the rest is omitted ...]
files: tmt/steps/finish/ansible.py
files: import tmt
  import tmt.steps
  import tmt.steps.finish
  import tmt.steps.prepare.ansible
  from tmt.steps.prepare.ansible import PrepareAnsible
  
  
  @tmt.steps.provides_method('ansible')
  class FinishAnsible(tmt.steps.finish.FinishPlugin, PrepareAnsible):
      """
      Perform finishing tasks using ansible
  
      Single playbook config:
  
          finish:
              how: ansible
              playbook: ansible/packages.yml
  
      Multiple playbooks config:
  
          finish:
              how: ansible
              playbook:
                - playbook/one.yml
                - playbook/two.yml
                - playbook/three.yml
  
      The playbook path should be relative to the metadata tree root.
      Use 'order' attribute to select in which order finishing tasks
      should happen if there are multiple configs. Default order is '50'.
      """
  
      _data_class = tmt.steps.prepare.ansible.PrepareAnsibleData
  
      # Assigning class methods seems to cause trouble to mypy
      # See also: https://github.com/python/mypy/issues/6700
      base_command = tmt.steps.finish.FinishPlugin.base_command  # type: ignore
  
  
  [... the rest is omitted ...]
files: tmt/steps/finish/ansible.py
files: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  import types
  from typing import Any, List, Optional, cast
  
  import click
  import requests
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  libvirt: Optional[types.ModuleType] = None
  testcloud: Optional[types.ModuleType] = None
  
  
  def import_testcloud() -> None:
      """
      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']
  """
  
  COREOS_DATA = """variant: fcos
  version: 1.4.0
  passwd:
    users:
      - name: {user_name}
        ssh_authorized_keys:
          - {public_key}
  """
  
  # Libvirt domain XML template related variables
  DOMAIN_TEMPLATE_NAME = 'domain-template.jinja'
  DOMAIN_TEMPLATE_FILE = os.path.join(TESTCLOUD_DATA, DOMAIN_TEMPLATE_NAME)
  DOMAIN_TEMPLATE = """<domain type='{{ virt_type }}' xmlns:qemu='http://libvirt.org/schemas/domain/qemu/1.0'>
    <name>{{ domain_name }}</name>
    <uuid>{{ uuid }}</uuid>
    <memory unit='KiB'>{{ memory }}</memory>
    <currentMemory unit='KiB'>{{ memory }}</currentMemory>
    <vcpu placement='static'>2</vcpu>
    <os>
      <type arch='{{ arch }}' machine='{{ model }}'>hvm</type>
      {{ uefi_loader }}
      <boot dev='hd'/>
    </os>
    {{ cpu }}
    {{ extra_specs }}
    <clock offset='utc'>
      <timer name='rtc' tickpolicy='catchup'/>
      <timer name='pit' tickpolicy='delay'/>
      <timer name='hpet' present='no'/>
    </clock>
    <on_poweroff>destroy</on_poweroff>
    <on_reboot>restart</on_reboot>
    <on_crash>restart</on_crash>
    <devices>
      <emulator>{{ emulator_path }}</emulator>
      <disk type='file' device='disk'>
        <driver name='qemu' type='qcow2' cache='unsafe'/>
        <source file="{{ disk }}"/>
        <target dev='vda' bus='virtio'/>
        {{ boot_drive_address }}
      </disk>
      <disk type='file' device='disk'>
        <driver name='qemu' type='raw'/>
        <source file="{{ seed }}"/>
        <target dev='vdb' bus='virtio'/>
        <address type='pci'
  
  [... the rest is omitted ...]
files: tmt/steps/provision/testcloud.py
files: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  import types
  from typing import Any, List, Optional, cast
  
  import click
  import requests
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  libvirt: Optional[types.ModuleType] = None
  testcloud: Optional[types.ModuleType] = None
  
  
  def import_testcloud() -> None:
      """
      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']
  """
  
  COREOS_DATA = """variant: fcos
  version: 1.4.0
  passwd:
    users:
      - name: {user_name}
        ssh_authorized_keys:
          - {public_key}
  """
  
  # Libvirt domain XML template related variables
  DOMAIN_TEMPLATE_NAME = 'domain-template.jinja'
  DOMAIN_TEMPLATE_FILE = os.path.join(TESTCLOUD_DATA, DOMAIN_TEMPLATE_NAME)
  DOMAIN_TEMPLATE = """<domain type='{{ virt_type }}' xmlns:qemu='http://libvirt.org/schemas/domain/qemu/1.0'>
    <name>{{ domain_name }}</name>
    <uuid>{{ uuid }}</uuid>
    <memory unit='KiB'>{{ memory }}</memory>
    <currentMemory unit='KiB'>{{ memory }}</currentMemory>
    <vcpu placement='static'>2</vcpu>
    <os>
      <type arch='{{ arch }}' machine='{{ model }}'>hvm</type>
      {{ uefi_loader }}
      <boot dev='hd'/>
    </os>
    {{ cpu }}
    {{ extra_specs }}
    <clock offset='utc'>
      <timer name='rtc' tickpolicy='catchup'/>
      <timer name='pit' tickpolicy='delay'/>
      <timer name='hpet' present='no'/>
    </clock>
    <on_poweroff>destroy</on_poweroff>
    <on_reboot>restart</on_reboot>
    <on_crash>restart</on_crash>
    <devices>
      <emulator>{{ emulator_path }}</emulator>
      <disk type='file' device='disk'>
        <driver name='qemu' type='qcow2' cache='unsafe'/>
        <source file="{{ disk }}"/>
        <target dev='vda' bus='virtio'/>
        {{ boot_drive_address }}
      </disk>
      <disk type='file' device='disk'>
        <driver name='qemu' type='raw'/>
        <source file="{{ seed }}"/>
        <target dev='vdb' bus='virtio'/>
        <address type='pci'
  
  [... the rest is omitted ...]
files: tmt/steps/provision/testcloud.py
files: import dataclasses
  import os
  import types
  from typing import Any, List, Optional, cast, overload
  
  import click
  
  import tmt
  import tmt.options
  import tmt.steps
  import tmt.steps.report
  
  DEFAULT_NAME = "junit.xml"
  
  junit_xml: Optional[types.ModuleType] = None
  
  # Thanks to import burried in a function, we don't get the regular missing annotations
  # but "name ... is not defined". Define a special type to follow the test suite instance
  # around, and if junit_xml ever gets annotations, we can replace it with provided type.
  JunitTestSuite = Any
  
  
  def import_junit_xml() -> None:
      """
      Import junit_xml module only when needed
  
      Until we have a separate package for each plugin.
      """
      global junit_xml
      try:
          import junit_xml
      except ImportError:
          raise tmt.utils.ReportError(
              "Missing 'junit-xml', fixable by 'pip install tmt[report-junit]'.")
  
  
  @overload
  def duration_to_seconds(duration: str) -> int: pass
  
  
  @overload
  def duration_to_seconds(duration: None) -> None: pass
  
  
  def duration_to_seconds(duration: Optional[str]) -> Optional[int]:
      """ Convert valid duration string in to seconds """
      if duration is None:
          return None
      try:
          h, m, s = duration.split(':')
          return int(h) * 3600 + int(m) * 60 + int(s)
      except Exception as error:
          raise tmt.utils.ReportError(
              f"Malformed duration '{duration}' ({error}).")
  
  
  def make_junit_xml(report: "tmt.steps.report.ReportPlugin") -> JunitTestSuite:
      """ Create junit xml object """
      import_junit_xml()
      assert junit_xml
  
      suite = junit_xml.TestSuite(report.step.plan.name)
  
      for result in report.step.plan.execute.results():
          try:
              main_log = report.step.plan.execute.read(result.log[0])
          except (IndexError, AttributeError):
              main_log = None
          case = junit_xml.TestCase(
              result.name,
              classname=None,
              elapsed_sec=duration_to_seconds(result.duration),
              stdout=main_log)
          # Map tmt OUTCOME to JUnit states
          if result.result == "error":
              case.add_error_info(result.result)
          elif result.result == "fail":
              case.add_failure_info(result.result)
          elif result.result == "info":
              case.add_skipped_info(result.result)
          elif result.result == "warn":
              case.add_error_info(result.result)
          # Passed state is the default
          suite.test_cases.append(case)
  
      return cast(JunitTestSuite, suite)
  
  
  @dataclasses.dataclass
  class ReportJUnitData(tmt.steps.StepData):
      file: Optional[str] = None
  
  
  @tmt.steps.provides_method('junit')
  class ReportJUnit(tmt.steps.report.ReportPlugin):
      """
      Write test results in JUnit format
  
      When FILE is not specified output is written to the 'junit.xml'
      located in the current workdir.
      """
  
      _data_class = ReportJUnitData
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for connect """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '--file', metavar='FILE',
                  help='Path to the file to store junit to'),
              ]
          return options
  
      def go(self) -> None:
          """ Read executed tests and write junit """
          super().go()
  
          suite = make_junit_xml(self)
  
          assert self.workdir is not None
          f_path = self.get("file", os.path.join(self.workdir, DEFAULT_NAME))
          try:
              with open(f_path, 'w') as fw:
                  if hasattr(junit_xml, 'to_xml_report_file'):
                      junit_xml.to_xml_report_file(fw, [suite])  # type: ignore[union-attr]
                  else:
                      # For older junit-xml
                      junit_xml.TestSuite.to_file(fw, [suite])  # type: ignore[union-attr]
              self.info("output", f_path, 'yellow')
          except Exception as error:
              raise tmt.utils.ReportError(
                  f"Failed to write the output '{f_path}' ({error}).")
  
  
  [... the rest is omitted ...]
files: tmt/steps/report/junit.py
files: import dataclasses
  import os
  import types
  from typing import Any, List, Optional, cast, overload
  
  import click
  
  import tmt
  import tmt.options
  import tmt.steps
  import tmt.steps.report
  
  DEFAULT_NAME = "junit.xml"
  
  junit_xml: Optional[types.ModuleType] = None
  
  # Thanks to import burried in a function, we don't get the regular missing annotations
  # but "name ... is not defined". Define a special type to follow the test suite instance
  # around, and if junit_xml ever gets annotations, we can replace it with provided type.
  JunitTestSuite = Any
  
  
  def import_junit_xml() -> None:
      """
      Import junit_xml module only when needed
  
      Until we have a separate package for each plugin.
      """
      global junit_xml
      try:
          import junit_xml
      except ImportError:
          raise tmt.utils.ReportError(
              "Missing 'junit-xml', fixable by 'pip install tmt[report-junit]'.")
  
  
  @overload
  def duration_to_seconds(duration: str) -> int: pass
  
  
  @overload
  def duration_to_seconds(duration: None) -> None: pass
  
  
  def duration_to_seconds(duration: Optional[str]) -> Optional[int]:
      """ Convert valid duration string in to seconds """
      if duration is None:
          return None
      try:
          h, m, s = duration.split(':')
          return int(h) * 3600 + int(m) * 60 + int(s)
      except Exception as error:
          raise tmt.utils.ReportError(
              f"Malformed duration '{duration}' ({error}).")
  
  
  def make_junit_xml(report: "tmt.steps.report.ReportPlugin") -> JunitTestSuite:
      """ Create junit xml object """
      import_junit_xml()
      assert junit_xml
  
      suite = junit_xml.TestSuite(report.step.plan.name)
  
      for result in report.step.plan.execute.results():
          try:
              main_log = report.step.plan.execute.read(result.log[0])
          except (IndexError, AttributeError):
              main_log = None
          case = junit_xml.TestCase(
              result.name,
              classname=None,
              elapsed_sec=duration_to_seconds(result.duration),
              stdout=main_log)
          # Map tmt OUTCOME to JUnit states
          if result.result == "error":
              case.add_error_info(result.result)
          elif result.result == "fail":
              case.add_failure_info(result.result)
          elif result.result == "info":
              case.add_skipped_info(result.result)
          elif result.result == "warn":
              case.add_error_info(result.result)
          # Passed state is the default
          suite.test_cases.append(case)
  
      return cast(JunitTestSuite, suite)
  
  
  @dataclasses.dataclass
  class ReportJUnitData(tmt.steps.StepData):
      file: Optional[str] = None
  
  
  @tmt.steps.provides_method('junit')
  class ReportJUnit(tmt.steps.report.ReportPlugin):
      """
      Write test results in JUnit format
  
      When FILE is not specified output is written to the 'junit.xml'
      located in the current workdir.
      """
  
      _data_class = ReportJUnitData
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for connect """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '--file', metavar='FILE',
                  help='Path to the file to store junit to'),
              ]
          return options
  
      def go(self) -> None:
          """ Read executed tests and write junit """
          super().go()
  
          suite = make_junit_xml(self)
  
          assert self.workdir is not None
          f_path = self.get("file", os.path.join(self.workdir, DEFAULT_NAME))
          try:
              with open(f_path, 'w') as fw:
                  if hasattr(junit_xml, 'to_xml_report_file'):
                      junit_xml.to_xml_report_file(fw, [suite])  # type: ignore
                  else:
                      # For older junit-xml
                      junit_xml.TestSuite.to_file(fw, [suite])  # type: ignore
              self.info("output", f_path, 'yellow')
          except Exception as error:
              raise tmt.utils.ReportError(
                  f"Failed to write the output '{f_path}' ({error}).")
  
  
  [... the rest is omitted ...]
files: tmt/steps/report/junit.py
files: 
  """ Test Metadata Utilities """
  
  import contextlib
  import copy
  import dataclasses
  import datetime
  import functools
  import glob
  import io
  import os
  import pprint
  import re
  import shlex
  import shutil
  import subprocess
  import sys
  import time
  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, Callable, Dict, Generator,
                      Iterable, List, NamedTuple, Optional, Pattern, Tuple, Type,
                      TypeVar, Union, cast, overload)
  
  import click
  import fmf
  import jsonschema
  import pkg_resources
  import requests
  import requests.adapters
  import requests.packages.urllib3.util.retry
  import urllib3.exceptions
  from click import echo, style, wrap_text
  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
      import tmt.steps
  
  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
  
  # Default shell and options to be set for all shell scripts
  DEFAULT_SHELL = "/bin/bash"
  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
  
  # Defaults for HTTP/HTTPS retries for getting environment file
  # Retry with exponential backoff, maximum duration ~511 seconds
  ENVFILE_RETRY_SESSION_RETRIES: int = 10
  ENVFILE_RETRY_SESSION_BACKOFF_FACTOR: float = 1
  
  # Default for wait()-related options
  DEFAULT_WAIT_TICK: float = 30.0
  DEFAULT_WAIT_TICK_INCREASE: float = 1.0
  
  # 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.
  
      :
  
  [... the rest is omitted ...]
files: tmt/utils.py
files: 
  """ Test Metadata Utilities """
  
  import contextlib
  import copy
  import dataclasses
  import datetime
  import functools
  import glob
  import io
  import os
  import pprint
  import re
  import shlex
  import shutil
  import subprocess
  import sys
  import time
  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, Callable, Dict, Generator,
                      Iterable, List, NamedTuple, Optional, Pattern, Tuple, Type,
                      TypeVar, Union, cast, overload)
  
  import click
  import fmf
  import jsonschema
  import pkg_resources
  import requests
  import requests.adapters
  import requests.packages.urllib3.util.retry
  import urllib3.exceptions
  from click import echo, style, wrap_text
  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
      import tmt.steps
  
  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
  
  # Default shell and options to be set for all shell scripts
  DEFAULT_SHELL = "/bin/bash"
  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
  
  # Defaults for HTTP/HTTPS retries for getting environment file
  # Retry with exponential backoff, maximum duration ~511 seconds
  ENVFILE_RETRY_SESSION_RETRIES: int = 10
  ENVFILE_RETRY_SESSION_BACKOFF_FACTOR: float = 1
  
  # Default for wait()-related options
  DEFAULT_WAIT_TICK: float = 30.0
  DEFAULT_WAIT_TICK_INCREASE: float = 1.0
  
  # 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.
  
      :
  
  [... the rest is omitted ...]
files: tmt/utils.py