files.new_contents: # 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.new_path: .pre-commit-config.yaml
files.old_contents: # 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.old_path: .pre-commit-config.yaml
files.new_contents: # 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.argument(
              'names', nargs=-1, metavar='[REGEXP|.]'),
          click.option(
              '--filter', 'filters', metavar='FILTER', multiple=True,
              help="Apply advanced filter (see 'pydoc fmf.filter')."),
          click.option(
              '--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(
              '--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 implemented_verified_documented(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Common story options """
  
      options = [
          click.option(
              '--implemented', is_flag=True,
              help='Implemented stories only.'),
          click.option(
              '--unimplemented', is_flag=True,
              help='Unimplemented stories only.'),
          click.option(
              '--verified', is_flag=True,
              help='Stories verified by tests.'),
          click.option(
              '--unverified', is_flag=True,
              help='Stories not verified by tests.'),
          click.option(
              '--documented', is_flag=True,
              help='Documented stories only.'),
          click.option(
              '--undocumented', is_flag=True,
              help='Undocumented stories only.'),
          click.option(
              '--covered', is_flag=True,
              help='Covered stories only.'),
          click.option(
              '--uncovered', is_flag=True,
              help='Uncovered stories only.'),
          ]
  
      for option in reversed(options):
          function = option(function)
      return function
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Main
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  @click.group(invoke_without_command=True, cls=CustomGroup)
  @click.pass_context
  @click.option(
      '-r', '--root', metavar='PATH', show_default=True,
      help="Path to the tree root, '.' by default.")
  @click.option(
      '-c', '--context', metavar='DATA', multiple=True,
      help='Set the fmf context. Use KEY=VAL or KEY=VAL1,VAL2... format '
           'to define individual dimensions or the @FILE notation to load data '
           'from provided yaml file. Can be specified multiple times. ')
  @verbose_debug_quiet
  @click.option(
      '--version', is_flag=True,
      help='Show tmt version and commit hash.')
  def main(
          click_contex: click.core.Context,
          root: str,
          context: List[str],
          **kwargs: Any) -> None:
      """ Test Management Tool """
      # Show current tmt version and exit
      if kwargs.get('version'):
          print(f"tmt version: {tmt.__version__}")
          raise SystemExit(0)
  
      # Disable coloring if NO_COLOR is set
      if 'NO_COLOR' in os.environ:
          click_contex.color = False
  
      # Save click context and fmf context for future use
      tmt.utils.Common._save_context(click_contex)
  
      # Initialize metadata tree (from given path or current directory)
      tree = tmt.Tree(root or os.curdir)
  
      # TODO: context object details need checks
      click_contex.obj = ContextObject(
          common=tmt.utils.Common(),
          fmf_context=tmt.utils.context_to_dict(context),
          steps=set(),
          tree=tree
          )
  
      # Show overview of available tests, plans and stories
      if click_contex.invoked_subcommand is None:
          tmt.Test.overview(tree)
          tmt.Plan.overview(tree)
          tmt.Story.overview(tree)
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Run
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  @main.group(chain=True, invoke_without_command=True, cls=CustomGroup)
  @click.pass_context
  @click.option(
      '-i', '--id', 'id_', help='Run id (name or directory path).', metavar="ID")
  @click.option(
      '-l', '--last', help='Execute the last run once again.', is_flag=True)
  @click.option(
      '-r', '--rm', '--remove', 'remove', is_flag=True,
      help='Remove the workdir when test run is finished.')
  @click.option(
      '--scratch', is_flag=True,
      help='Remove the run workdir before executing to start from scratch.')
  @click.option(
      '--follow', is_flag=True,
      help='Output the logfile as it grows.')
  @click.option(
      '-a', '--all', help='Run all steps, customize some.', is_flag=True)
  @click.option(
      '-u', '--until', type=click.Choice(tmt.steps.STEPS), metavar='STEP',
      help='Enable given step and all preceding steps.')
  @click.option(
      '-s', '--since', type=click.Choice(tmt.steps.STEPS), metavar='STEP',
      help='Enable given step and all following steps.')
  @click.option(
      '-A', '--after', type=click.Choice(tmt.steps.STEPS), metavar='STEP',
      help='Enable all steps after the given one.')
  @click.option(
      '-B', '--before', type=click.Choice(tmt.steps.STEPS), metavar='STEP',
      help='Enable all steps before the given one.')
  @click.option(
      '-S', '--skip', type=click.Choice(tmt.steps.STEPS), metavar='STEP',
      help='Skip given step(s) during test run execution.', multiple=True)
  @click.option(
      '-e', '--environment', metavar='KEY=VALUE|@FILE', multiple=True,
      help='Set environment variable. Can be specified multiple times. The '
           '"@" prefix marks a file to load (yaml or dotenv formats supported).')
  @click.option(
      '--environment-file', metavar='FILE|URL', multiple=True,
      help='Set environment variables from file or url (yaml or dotenv formats '
           'are supported). Can be specified multiple times.')
  @verbose_debug_quiet
  @force_dry
  def run(context: click.core.Context, id_: str, **kwargs: Any) -> None:
      """ Run test steps. """
      # Initialize
      run = tmt.Run(id_, context.obj.tree, context=context)
      context.obj.run = run
  
  
  # Steps options
  run.add_command(tmt.steps.discover.DiscoverPlugin.command())
  run.add_command(tmt.steps.provision.ProvisionPlugin.command())
  run.add_command(tmt.steps.prepare.PreparePlugin.command())
  run.add_command(tmt.steps.execute.ExecutePlugin.command())
  run.add_command(tmt.steps.report.ReportPlugin.command())
  run.add_command(tmt.steps.finish.FinishPlugin.command())
  run.add_command(tmt.steps.Login.command())
  run.add_command(tmt.steps.Reboot.command())
  
  
  @run.command(name='plans')
  @click.pass_context
  @click.option(
      '-n', '--name', 'names', metavar='[REGEXP|.]', multiple=True,
      help="Regular expression to match plan name or '.' for current directory.")
  @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(
      '--link', 'links', metavar="RELATION:TARGET", multiple=True,
      help="Filter by linked objects (regular expressions are "
           "supported for both relation and target).")
  @click.option(
      '--default', is_flag=True,
      help="Use default plans even if others are available.")
  @verbose_debug_quiet
  def run_plans(context: click.core.Context, **kwargs: Any) -> None:
      """
      Select plans which should be executed.
  
      Regular expression can be used to filter plans by name.
      Use '.' to select plans under the current working directory.
      """
      tmt.base.Plan._save_context(context)
  
  
  @run.command(name='tests')
  @click.pass_context
  @click.option(
      '-n', '--name', 'names', metavar='[REGEXP|.]', multiple=True,
      help="Regular expression to match test name or '.' for current directory.")
  @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(
      '--link', 'links', metavar="RELATION:TARGET", multiple=True,
      help="Filter by linked objects (regular expressions are "
           "supported for both relation and target).")
  @verbose_debug_quiet
  def run_tests(context: click.core.Context, **kwargs: Any) -> None:
      """
      Select tests which should be executed.
  
      Regular expression can be used to filter tests by name.
      Use '.' to select tests under the current working directory.
      """
      tmt.base.Test._save_context(context)
  
  
  # FIXME: click 8.0 renamed resultcallback to result_callback. The former
  #        name will be removed in click 8.1. However, click 8.0 will not
  #        be added to F33 and F34. Get rid of this workaround once
  #        all Fedora + EPEL releases have click 8.0 or newer available.
  run_callback = run.result_callback
  if run_callback is None:
      run_callback = run.resultcallback
  
  
  # TODO: commands is unknown, needs revisit
  @run_callback()  # type: ignore[misc]
  @click.pass_context
  def finito(
          click_context: click.core.Context,
          commands: Any,
          *args: Any,
          **kwargs: Any) -> None:
      """ Run tests if run defined """
      if hasattr(click_context.obj, 'run'):
          click_context.obj.run.go()
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Test
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  @main.group(invoke_without_command=True, cls=CustomGroup)
  @click.pass_context
  @verbose_debug_quiet
  def tests(context: click.core.Context, **kwargs: Any) -> None:
      """
      Manage tests (L1 metadata).
  
      Check available tests, inspect their metadata.
      Convert old metadata into the new fmf format.
      """
  
      # Show overview of available tests
      if context.invoked_subcommand is None:
          tmt.Test.overview(context.obj.tree)
  
  
  @tests.command(name='ls')
  @click.pass_context
  @name_filter_condition
  @verbose_debug_quiet
  def tests_ls(context: click.core.Context, **kwargs: Any) -> None:
      """
      List available tests.
  
      Regular expression can be used to filter tests by name.
      Use '.' to select tests under the current working directory.
      """
      tmt.Test._save_context(context)
      for test in context.obj.tree.tests():
          test.ls()
  
  
  @tests.command(name='show')
  @click.pass_context
  @name_filter_condition
  @verbose_debug_quiet
  def tests_show(context: click.core.Context, **kwargs: Any) -> None:
      """
      Show test details.
  
      Regular expression can be used to filter tests by name.
      Use '.' to select tests under the current working directory.
      """
      tmt.Test._save_context(context)
      for test in context.obj.tree.tests():
          test.show()
          echo()
  
  
  @tests.command(name='lint')
  @click.pass_context
  @name_filter_condition
  @source
  @fix
  @verbose_debug_quiet
  def tests_lint(context: click.core.Context, **kwargs: Any) -> None:
      """
      Check tests against the L1 metadata specification.
  
      Regular expression can be used to filter tests for linting.
      Use '.' to select tests under the current working directory.
      """
      # FIXME: Workaround https://github.com/pallets/click/pull/1840 for click 7
      context.params.update(**kwargs)
      tmt.Test._save_context(context)
      exit_code = 0
      for test in context.obj.tree.tests():
          if not test.lint():
              exit_code = 1
          echo()
      raise SystemExit(exit_code)
  
  
  _test_templates = listed(tmt.templates.TEST, join='or')
  
  
  @tests.command(name='create')
  @click.pass_context
  @click.argument('name')
  @click.option(
      '-t', '--template', metavar='TEMPLATE',
      help='Test template ({}).'.format(_test_templates),
      prompt='Template ({})'.format(_test_templates))
  @verbose_debug_quiet
  @force_dry
  def tests_create(
          context: click.core.Context,
          name: str,
          template: str,
          force: bool,
          **kwargs: Any) -> None:
      """
      Create a new test based on given template.
  
      Specify directory name or use '.' to create tests under the
      current working directory.
      """
      tmt.Test._save_context(context)
      tmt.Test.create(name, template, context.obj.tree.root, force)
  
  
  @tests.command(name='import')
  @click.pass_context
  @click.argument('paths', nargs=-1, metavar='[PATH]...')
  @click.option(
      '--nitrate / --no-nitrate', default=True,
      help='Import test metadata from Nitrate')
  @click.option(
      '--purpose / --no-purpose', default=True,
      help='Migrate description from PURPOSE file')
  @click.option(
      '--makefile / --no-makefile', default=True,
      help='Convert Beaker Makefile metadata')
  @click.option(
      '--restraint / --no-restraint', default=False,
      help='Convert restraint metadata file')
  @click.option(
      '--general / --no-general', default=True,
      help='Detect components from linked nitrate general plans '
           '(overrides Makefile/restraint component).')
  @click.option(
      '--type', 'types', metavar='TYPE', default=['multihost'], multiple=True,
      show_default=True,
      help="Convert selected types from Makefile into tags. "
           "Use 'all' to convert all detected types.")
  @click.option(
      '--disabled', default=False, is_flag=True,
      help='Import disabled test cases from Nitrate as well.')
  @click.option(
      '--manual', default=False, is_flag=True,
      help='Import manual test cases from Nitrate.')
  @click.option(
      '--plan', metavar='PLAN
  
  [... the rest is omitted ...]
files.new_path: tmt/cli.py
files.old_contents: # 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.argument(
              'names', nargs=-1, metavar='[REGEXP|.]'),
          click.option(
              '--filter', 'filters', metavar='FILTER', multiple=True,
              help="Apply advanced filter (see 'pydoc fmf.filter')."),
          click.option(
              '--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(
              '--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 implemented_verified_documented(function: Callable[..., Any]) -> Callable[..., Any]:
      """ Common story options """
  
      options = [
          click.option(
              '--implemented', is_flag=True,
              help='Implemented stories only.'),
          click.option(
              '--unimplemented', is_flag=True,
              help='Unimplemented stories only.'),
          click.option(
              '--verified', is_flag=True,
              help='Stories verified by tests.'),
          click.option(
              '--unverified', is_flag=True,
              help='Stories not verified by tests.'),
          click.option(
              '--documented', is_flag=True,
              help='Documented stories only.'),
          click.option(
              '--undocumented', is_flag=True,
              help='Undocumented stories only.'),
          click.option(
              '--covered', is_flag=True,
              help='Covered stories only.'),
          click.option(
              '--uncovered', is_flag=True,
              help='Uncovered stories only.'),
          ]
  
      for option in reversed(options):
          function = option(function)
      return function
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Main
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  @click.group(invoke_without_command=True, cls=CustomGroup)
  @click.pass_context
  @click.option(
      '-r', '--root', metavar='PATH', show_default=True,
      help="Path to the tree root, '.' by default.")
  @click.option(
      '-c', '--context', metavar='DATA', multiple=True,
      help='Set the fmf context. Use KEY=VAL or KEY=VAL1,VAL2... format '
           'to define individual dimensions or the @FILE notation to load data '
           'from provided yaml file. Can be specified multiple times. ')
  @verbose_debug_quiet
  @click.option(
      '--version', is_flag=True,
      help='Show tmt version and commit hash.')
  def main(
          click_contex: click.core.Context,
          root: str,
          context: List[str],
          **kwargs: Any) -> None:
      """ Test Management Tool """
      # Show current tmt version and exit
      if kwargs.get('version'):
          print(f"tmt version: {tmt.__version__}")
          raise SystemExit(0)
  
      # Disable coloring if NO_COLOR is set
      if 'NO_COLOR' in os.environ:
          click_contex.color = False
  
      # Save click context and fmf context for future use
      tmt.utils.Common._save_context(click_contex)
  
      # Initialize metadata tree (from given path or current directory)
      tree = tmt.Tree(root or os.curdir)
  
      # TODO: context object details need checks
      click_contex.obj = ContextObject(
          common=tmt.utils.Common(),
          fmf_context=tmt.utils.context_to_dict(context),
          steps=set(),
          tree=tree
          )
  
      # Show overview of available tests, plans and stories
      if click_contex.invoked_subcommand is None:
          tmt.Test.overview(tree)
          tmt.Plan.overview(tree)
          tmt.Story.overview(tree)
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Run
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  @main.group(chain=True, invoke_without_command=True, cls=CustomGroup)
  @click.pass_context
  @click.option(
      '-i', '--id', 'id_', help='Run id (name or directory path).', metavar="ID")
  @click.option(
      '-l', '--last', help='Execute the last run once again.', is_flag=True)
  @click.option(
      '-r', '--rm', '--remove', 'remove', is_flag=True,
      help='Remove the workdir when test run is finished.')
  @click.option(
      '--scratch', is_flag=True,
      help='Remove the run workdir before executing to start from scratch.')
  @click.option(
      '--follow', is_flag=True,
      help='Output the logfile as it grows.')
  @click.option(
      '-a', '--all', help='Run all steps, customize some.', is_flag=True)
  @click.option(
      '-u', '--until', type=click.Choice(tmt.steps.STEPS), metavar='STEP',
      help='Enable given step and all preceding steps.')
  @click.option(
      '-s', '--since', type=click.Choice(tmt.steps.STEPS), metavar='STEP',
      help='Enable given step and all following steps.')
  @click.option(
      '-A', '--after', type=click.Choice(tmt.steps.STEPS), metavar='STEP',
      help='Enable all steps after the given one.')
  @click.option(
      '-B', '--before', type=click.Choice(tmt.steps.STEPS), metavar='STEP',
      help='Enable all steps before the given one.')
  @click.option(
      '-S', '--skip', type=click.Choice(tmt.steps.STEPS), metavar='STEP',
      help='Skip given step(s) during test run execution.', multiple=True)
  @click.option(
      '-e', '--environment', metavar='KEY=VALUE|@FILE', multiple=True,
      help='Set environment variable. Can be specified multiple times. The '
           '"@" prefix marks a file to load (yaml or dotenv formats supported).')
  @click.option(
      '--environment-file', metavar='FILE|URL', multiple=True,
      help='Set environment variables from file or url (yaml or dotenv formats '
           'are supported). Can be specified multiple times.')
  @verbose_debug_quiet
  @force_dry
  def run(context: click.core.Context, id_: str, **kwargs: Any) -> None:
      """ Run test steps. """
      # Initialize
      run = tmt.Run(id_, context.obj.tree, context=context)
      context.obj.run = run
  
  
  # Steps options
  run.add_command(tmt.steps.discover.DiscoverPlugin.command())
  run.add_command(tmt.steps.provision.ProvisionPlugin.command())
  run.add_command(tmt.steps.prepare.PreparePlugin.command())
  run.add_command(tmt.steps.execute.ExecutePlugin.command())
  run.add_command(tmt.steps.report.ReportPlugin.command())
  run.add_command(tmt.steps.finish.FinishPlugin.command())
  run.add_command(tmt.steps.Login.command())
  run.add_command(tmt.steps.Reboot.command())
  
  
  @run.command(name='plans')
  @click.pass_context
  @click.option(
      '-n', '--name', 'names', metavar='[REGEXP|.]', multiple=True,
      help="Regular expression to match plan name or '.' for current directory.")
  @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(
      '--link', 'links', metavar="RELATION:TARGET", multiple=True,
      help="Filter by linked objects (regular expressions are "
           "supported for both relation and target).")
  @click.option(
      '--default', is_flag=True,
      help="Use default plans even if others are available.")
  @verbose_debug_quiet
  def run_plans(context: click.core.Context, **kwargs: Any) -> None:
      """
      Select plans which should be executed.
  
      Regular expression can be used to filter plans by name.
      Use '.' to select plans under the current working directory.
      """
      tmt.base.Plan._save_context(context)
  
  
  @run.command(name='tests')
  @click.pass_context
  @click.option(
      '-n', '--name', 'names', metavar='[REGEXP|.]', multiple=True,
      help="Regular expression to match test name or '.' for current directory.")
  @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(
      '--link', 'links', metavar="RELATION:TARGET", multiple=True,
      help="Filter by linked objects (regular expressions are "
           "supported for both relation and target).")
  @verbose_debug_quiet
  def run_tests(context: click.core.Context, **kwargs: Any) -> None:
      """
      Select tests which should be executed.
  
      Regular expression can be used to filter tests by name.
      Use '.' to select tests under the current working directory.
      """
      tmt.base.Test._save_context(context)
  
  
  # FIXME: click 8.0 renamed resultcallback to result_callback. The former
  #        name will be removed in click 8.1. However, click 8.0 will not
  #        be added to F33 and F34. Get rid of this workaround once
  #        all Fedora + EPEL releases have click 8.0 or newer available.
  run_callback = run.result_callback
  if run_callback is None:
      run_callback = run.resultcallback
  
  
  # TODO: commands is unknown, needs revisit
  @run_callback()  # type: ignore
  @click.pass_context
  def finito(
          click_context: click.core.Context,
          commands: Any,
          *args: Any,
          **kwargs: Any) -> None:
      """ Run tests if run defined """
      if hasattr(click_context.obj, 'run'):
          click_context.obj.run.go()
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Test
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  @main.group(invoke_without_command=True, cls=CustomGroup)
  @click.pass_context
  @verbose_debug_quiet
  def tests(context: click.core.Context, **kwargs: Any) -> None:
      """
      Manage tests (L1 metadata).
  
      Check available tests, inspect their metadata.
      Convert old metadata into the new fmf format.
      """
  
      # Show overview of available tests
      if context.invoked_subcommand is None:
          tmt.Test.overview(context.obj.tree)
  
  
  @tests.command(name='ls')
  @click.pass_context
  @name_filter_condition
  @verbose_debug_quiet
  def tests_ls(context: click.core.Context, **kwargs: Any) -> None:
      """
      List available tests.
  
      Regular expression can be used to filter tests by name.
      Use '.' to select tests under the current working directory.
      """
      tmt.Test._save_context(context)
      for test in context.obj.tree.tests():
          test.ls()
  
  
  @tests.command(name='show')
  @click.pass_context
  @name_filter_condition
  @verbose_debug_quiet
  def tests_show(context: click.core.Context, **kwargs: Any) -> None:
      """
      Show test details.
  
      Regular expression can be used to filter tests by name.
      Use '.' to select tests under the current working directory.
      """
      tmt.Test._save_context(context)
      for test in context.obj.tree.tests():
          test.show()
          echo()
  
  
  @tests.command(name='lint')
  @click.pass_context
  @name_filter_condition
  @source
  @fix
  @verbose_debug_quiet
  def tests_lint(context: click.core.Context, **kwargs: Any) -> None:
      """
      Check tests against the L1 metadata specification.
  
      Regular expression can be used to filter tests for linting.
      Use '.' to select tests under the current working directory.
      """
      # FIXME: Workaround https://github.com/pallets/click/pull/1840 for click 7
      context.params.update(**kwargs)
      tmt.Test._save_context(context)
      exit_code = 0
      for test in context.obj.tree.tests():
          if not test.lint():
              exit_code = 1
          echo()
      raise SystemExit(exit_code)
  
  
  _test_templates = listed(tmt.templates.TEST, join='or')
  
  
  @tests.command(name='create')
  @click.pass_context
  @click.argument('name')
  @click.option(
      '-t', '--template', metavar='TEMPLATE',
      help='Test template ({}).'.format(_test_templates),
      prompt='Template ({})'.format(_test_templates))
  @verbose_debug_quiet
  @force_dry
  def tests_create(
          context: click.core.Context,
          name: str,
          template: str,
          force: bool,
          **kwargs: Any) -> None:
      """
      Create a new test based on given template.
  
      Specify directory name or use '.' to create tests under the
      current working directory.
      """
      tmt.Test._save_context(context)
      tmt.Test.create(name, template, context.obj.tree.root, force)
  
  
  @tests.command(name='import')
  @click.pass_context
  @click.argument('paths', nargs=-1, metavar='[PATH]...')
  @click.option(
      '--nitrate / --no-nitrate', default=True,
      help='Import test metadata from Nitrate')
  @click.option(
      '--purpose / --no-purpose', default=True,
      help='Migrate description from PURPOSE file')
  @click.option(
      '--makefile / --no-makefile', default=True,
      help='Convert Beaker Makefile metadata')
  @click.option(
      '--restraint / --no-restraint', default=False,
      help='Convert restraint metadata file')
  @click.option(
      '--general / --no-general', default=True,
      help='Detect components from linked nitrate general plans '
           '(overrides Makefile/restraint component).')
  @click.option(
      '--type', 'types', metavar='TYPE', default=['multihost'], multiple=True,
      show_default=True,
      help="Convert selected types from Makefile into tags. "
           "Use 'all' to convert all detected types.")
  @click.option(
      '--disabled', default=False, is_flag=True,
      help='Import disabled test cases from Nitrate as well.')
  @click.option(
      '--manual', default=False, is_flag=True,
      help='Import manual test cases from Nitrate.')
  @click.option(
      '--plan', metavar='PLAN', type
  
  [... the rest is omitted ...]
files.old_path: tmt/cli.py
files.new_contents: # 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.NitrateError:
              pass
  
  
  def convert_manual_to_nitrate(test_md: str) -> SectionsReturnType:
      """
      Convert Markdown document to html sections.
  
      These sections can be exported to nitrate.
      Expects: Markdown document as a file.
      Returns: tuple of (step, expect, setup, cleanup) sections
      as html strings.
      """
  
      sections_headings: SectionsHeadingsType = {
          heading: []
          for heading_list in tmt.base.SECTIONS_HEADINGS.values()
          for heading in heading_list
          }
  
      html = markdown_to_html(test_md)
      html_splitlines = html.splitlines()
  
      for key in sections_headings.keys():
          result: HeadingsType = []
          i = 0
          while html_splitlines:
              try:
                  if re.search("^" + key + "$", html_splitlines[i]):
                      html_content = str()
                      if key.startswith('<h1>Test'):
                          html_content = html_splitlines[i].\
                              replace('<h1>', '<b>').\
                              replace('</h1>', '</b>')
                      for j in range(i + 1, len(html_splitlines)):
                          if re.search("^<h[1-4]>(.+?)</h[1-4]>$",
                                       html_splitlines[j]):
                              result.append([i, html_content])
                              i = j - 1
                              break
                          html_content += html_splitlines[j] + "\n"
                          # Check end of the file
                          if j + 1 == len(html_splitlines):
                              result.append([i, html_content])
              except IndexError:
                  sections_headings[key] = result
                  break
              i += 1
              if i >= len(html_splitlines):
                  sections_headings[key] = result
                  break
  
      def concatenate_headings_content(headings: Tuple[str, ...]) -> HeadingsType:
          content = list()
          for v in headings:
              content += sections_headings[v]
          return content
  
      def enumerate_content(content: HeadingsType) -> HeadingsType:
          # for sorting convert the index to integer, but keep whole list as list of strings
          content.sort(key=lambda a: int(a[0]))
          for c in range(len(content)):
              content[c][1] = f"<p>Step {c + 1}.</p>" + str(content[c][1])
          return content
  
      sorted_test = sorted(concatenate_headings_content((
          '<h1>Test</h1>',
          '<h1>Test .*</h1>')))
  
      sorted_step = sorted(enumerate_content(concatenate_headings_content((
          '<h2>Step</h2>',
          '<h2>Test Step</h2>'))) + sorted_test)
      step = ''.join([f"{v[1]}" for v in sorted_step])
  
      sorted_expect = sorted(enumerate_content(concatenate_headings_content((
          '<h2>Expect</h2>',
          '<h2>Result</h2>',
          '<h2>Expected Result</h2>'))) + sorted_test)
      expect = ''.join([f"{v[1]}" for v in sorted_expect])
  
      def check_section_exists(text: str) -> str:
          try:
              return str(sections_headings[text][0][1])
          except (IndexError, KeyError):
              return ''
  
      setup = check_section_exists('<h1>Setup</h1>')
      cleanup = check_section_exists('<h1>Cleanup</h1>')
  
      return step, expect, setup, cleanup
  
  
  def bz_set_coverage(bug_ids: List[int], case_id: str, tracker_id: int) -> None:
      """ Set coverage in Bugzilla """
      bz_instance = get_bz_instance()
  
      overall_pass = True
      no_email = 1  # Do not send emails about the change
      get_bz_dict = {
          'ids': bug_ids,
          'include_fields': ['id', 'external_bugs', 'flags']}
      bugs_data = bz_instance._proxy.Bug.get(get_bz_dict)
      for bug in bugs_data['bugs']:
          # Process flag (might fail for some types)
          bug_id = bug['id']
          if 'qe_test_coverage+' not in set(
                  [x['name'] + x['status'] for x in bug['flags']]):
              try:
                  bz_instance._proxy.Flag.update({
                      'ids': [bug_id],
                      'nomail': no_email,
                      'updates': [{
                          'name': 'qe_test_coverage',
                          'status': '+'
                          }]
                      })
              except xmlrpc.client.Fault as err:
                  # TODO: Fix missing overall_result = False, breaks tests
                  # if bug used for testing is not changed
                  # BZ#1925518 can't have qe_test_coverage flag
                  log.debug(f"Update flag failed: {err}")
                  echo(style(
                      f"Failed to set qe_test_coverage+ flag for BZ#{bug_id}",
                      fg='red'))
          # Process external tracker - should succeed
          current = set(
              [b['ext_bz_bug_id'] for b in bug['external_bugs']
               if b['ext_bz_id'] == tracker_id])
          if case_id not in current:
              query = {
                  'bug_ids': [bug_id],
                  'nomail': no_email,
                  'external_bugs': [{
                      'ext_type_id': tracker_id,
                      'ext_bz_bug_id': case_id,
                      'ext_description': '',
                      }]
                  }
              try:
                  bz_instance._proxy.ExternalBugs.add_external_bug(query)
              except Exception as err:
                  log.debug(f"Link case failed: {err}")
                  echo(style(f"Failed to link to BZ#{bug_id}", fg='red'))
                  overall_pass = False
      if not overall_pass:
          raise ConvertError("Failed to link the case to bugs.")
  
      echo(style("Linked to Bugzilla: {}.".format(
          " ".join([f"BZ#{bz_id}" for bz_id in bug_ids])), fg='magenta'))
  
  
  def get_polarion_ids(
          query_result: List[Any],
          preferred_project: Optional[str] = None) -> Tuple[str, Optional[str]]:
      """ Return case and project ids from query results """
      if not query_result:
          return 'None', None
      if len(query_result) == 1:
          return query_result[0].work_item_id, query_result[0].project_id
  
      if preferred_project:
          try:
              return next(
                  item.work_item_id for item in query_result
                  if item.project_id == preferred_project), preferred_project
          except StopIteration:
              pass
  
      for result in query_result:
          # If multiple cases are found prefer cases from other projects
          # than these legacy ones
          if str(result.project_id) not in LEGACY_POLARION_PROJECTS:
              return result.work_item_id, result.project_id
  
      return query_result[0].work_item_id, query_result[0].project_id
  
  
  def find_polarion_case_ids(
          data: Dict[str, str],
          preferred_project: Optional[str] = None) -> Tuple[str, Optional[str]]:
      """ Find IDs for Polarion case from data dictionary """
      assert PolarionWorkItem
  
      # Search by UUID
      if data.get(ID_KEY):
          query_result = PolarionWorkItem.query(
              data.get(ID_KEY), fields=['work_item_id', 'project_id'])
          return get_polarion_ids(query_result, preferred_project)
      # Search by TCMS Case ID
      extra_nitrate = data.get('extra-nitrate')
      if extra_nitrate:
          nitrate_case_id_search = re.search(r'\d+', extra_nitrate)
          if not nitrate_case_id_search:
              raise ConvertError(
                  "Could not find a valid nitrate testcase ID in 'extra-nitrate' attribute")
          nitrate_case_id = str(int(nitrate_case_id_search.group()))
          query_result = PolarionWorkItem.query(
              f"tcmscaseid:{nitrate_case_id}", fields=['work_item_id', 'project_id'])
          return get_polarion_ids(query_result, preferred_project)
      # Search by extra task
      if data.get('extra-task'):
          query_result = PolarionWorkItem.query(
              data.get('extra-task'), fields=['work_item_id', 'project_id'])
          return get_polarion_ids(query_result, preferred_project)
      return 'None', None
  
  
  def get_polarion_case(data: Dict[str, str], preferred_project: Optional[str] = None) -> Any:
      """ Get Polarion case through couple different methods """
      import_polarion()
  
      assert PolarionTestCase
      assert PolarionException
  
      polarion_id, project_id = find_polarion_case_ids(data, preferred_project)
  
      try:
          polarion_case = PolarionTestCase(
              project_id=project_id, work_item_id=polarion_id)
          echo(style(
              f"Test case '{str(polarion_case.work_item_id)}' found.",
              fg='blue'))
          return polarion_case
      except PolarionException:
          return None
  
  
  def enabled_somewhere(test: 'tmt.Test') -> bool:
      """ True if the test is enabled for some context (adjust rules) """
      # node.original_data are fmf data _before_ adjust was processed
      try:
          if test.node.original_data['enabled']:
              return True
      except KeyError:
          # Key defaults to True so we already know the outcome
          return True
  
      # Some rule in adjust enables the test
      try:
          adjust_rules = test.node.original_data['adjust']
          # TODO: Should not be necessary once we normalize data
          if isinstance(adjust_rules, dict):
              adjust_rules = [adjust_rules]
          for rule in adjust_rules:
              try:
                  if rule['enabled']:
                      return True
              except KeyError:
                  pass
      except KeyError:
          pass
      # At this point nothing enables the test
      return False
  
  
  def export_to_nitrate(test: 'tmt.Test') -> None:
      """ Export fmf metadata to nitrate test cases """
      import_nitrate()
      assert nitrate
      assert gssapi
  
      # Check command line options
      create = test.opt('create')
      general = test.opt('general')
      link_runs = test.opt('link_runs')
      duplicate = test.opt('duplicate')
      link_bugzilla = test.opt('bugzilla')
      ignore_git_validation = test.opt('ignore_git_validation')
      dry_mode = test.opt('dry')
  
      if link_runs:
          general = True
  
      # Check git is already correct
      valid, error_msg = tmt.utils.validate_git_status(test)
      if not valid:
          if ignore_git_validation:
              secho(f"Exporting regardless '{error_msg}'.", fg='red')
          else:
              raise ConvertError(
                  f"Can't export due '{error_msg}'.\n"
                  "Use --ignore-git-validation on your own risk to export regardless.")
  
      # Check nitrate test case
      try:
          nitrate_id = test.node.get('extra-nitrate')[3:]
          nitrate_case = nitrate.TestCase(int(nitrate_id))
          nitrate_case.summary  # Make sure we connect to the server now
          echo(style(f"Test case '{nitrate_case.identifier}' found.", fg='blue'))
      except TypeError:
          # Create a new nitrate test case
          if create:
              nitrate_case = None
              # Check for existing Nitrate tests with the same fmf id
              if not duplicate:
                  testcases = _nitrate_find_fmf_testcases(test)
                  try:
                      # Select the first found testcase if any
                      nitrate_case = next(testcases)
                  except StopIteration:
                      pass
              if not nitrate_case:
                  # Summary for TCMS case
                  extra_summary = prepare_extra_summary(test)
                  if not dry_mode:
                      nitrate_case = create_nitrate_case(extra_summary)
                  else:
                      echo(style(
                          f"Test case '{extra_summary}' created.", fg='blue'))
                  test._metadata['extra-summary'] = extra_summary
              # Either newly created or duplicate with missing extra-nitrate
              if nitrate_case:
                  echo(style("Append the nitrate test case id.", fg='green'))
                  if not dry_mode:
                      with test.node as data:
                          data["extra-nitrate"] = nitrate_case.identifier
          else:
              raise ConvertError(f"Nitrate test case id not found for {test}"
                                 " (You can use --create option to enforce"
                                 " creating testcases)")
      except (nitrate.NitrateError, gssapi.raw.misc.GSSError) as error:
          raise ConvertError(error)
  
      # Check if URL is accessible, to be able to reach from nitrate
      check_git_url(test.fmf_id['url'])
  
      # Summary
      try:
          summary = (test._metadata.get('extra-summary')
                     or test._metadata.get('extra-task')
                     or prepare_extra_summary(test))
      except ConvertError:
          summary = test.name
      if not dry_mode:
          nitrate_case.summary = summary
      echo(style('summary: ', fg='green') + summary)
  
      # Script
      if test.node.get('extra-task'):
          if not dry_mode:
              nitrate_case.script = test.node.get('extra-task')
          echo(style('script: ', fg='green') + test.node.get('extra-task'))
  
      # Components and General plan
      # First remove any components that are already there
      if not dry_mode:
          nitrate_case.components.clear()
      # Only these general plans should stay
      expected_general_plans = set()
      # Then add fmf ones
      if test.component:
          echo(style('components: ', fg='green') + ' '.join(test.component))
          for component in test.component:
              try:
                  nitrate_component = nitrate.Component(
                      name=component, product=DEFAULT_PRODUCT.id)
                  if not dry_mode:
                      nitrate_case.components.add(nitrate_component)
              except nitrate.xmlrpc_driver.NitrateError as error:
                  log.debug(error)
                  echo(style(
                      f"Failed to add component '{component}'.", fg='red'))
              if general:
                  try:
                      general_plan = find_general_plan(component)
                      expected_general_plans.add(general_plan)
                      echo(style(
                          f"Linked to general plan '{general_plan}'.",
                          fg='magenta'))
                      if not dry_mode:
                          nitrate_case.testplans.add(general_plan)
                      if link_runs:
                          add_to_nitrate_runs(
                              nitrate_case, general_plan, test, dry_mode)
                  except nitrate.NitrateError as error:
                      log.debug(error)
                      echo(style(
                          f"Failed to find general test plan for '{component}'.",
                          fg='red'))
      # Remove unexpected general plans
      if general and nitrate_case:
          # Remove also all general plans linked to testcase
          for nitrate_plan in [plan for plan in nitrate_case.testplans]:
              if (nitrate_plan.type.name == "General"
                      and nitrate_plan not in expected_general_plans):
                  echo(style(
                      f"Removed general plan '{nitrate_plan}'.", fg='red'))
                  if not dry_mode:
                      nitrate_case.testplans.remove(nitrate_plan)
  
  
  
  [... the rest is omitted ...]
files.new_path: tmt/export.py
files.old_contents: # 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_nitrate(test_md: str) -> SectionsReturnType:
      """
      Convert Markdown document to html sections.
  
      These sections can be exported to nitrate.
      Expects: Markdown document as a file.
      Returns: tuple of (step, expect, setup, cleanup) sections
      as html strings.
      """
  
      sections_headings: SectionsHeadingsType = {
          heading: []
          for heading_list in tmt.base.SECTIONS_HEADINGS.values()
          for heading in heading_list
          }
  
      html = markdown_to_html(test_md)
      html_splitlines = html.splitlines()
  
      for key in sections_headings.keys():
          result: HeadingsType = []
          i = 0
          while html_splitlines:
              try:
                  if re.search("^" + key + "$", html_splitlines[i]):
                      html_content = str()
                      if key.startswith('<h1>Test'):
                          html_content = html_splitlines[i].\
                              replace('<h1>', '<b>').\
                              replace('</h1>', '</b>')
                      for j in range(i + 1, len(html_splitlines)):
                          if re.search("^<h[1-4]>(.+?)</h[1-4]>$",
                                       html_splitlines[j]):
                              result.append([i, html_content])
                              i = j - 1
                              break
                          html_content += html_splitlines[j] + "\n"
                          # Check end of the file
                          if j + 1 == len(html_splitlines):
                              result.append([i, html_content])
              except IndexError:
                  sections_headings[key] = result
                  break
              i += 1
              if i >= len(html_splitlines):
                  sections_headings[key] = result
                  break
  
      def concatenate_headings_content(headings: Tuple[str, ...]) -> HeadingsType:
          content = list()
          for v in headings:
              content += sections_headings[v]
          return content
  
      def enumerate_content(content: HeadingsType) -> HeadingsType:
          # for sorting convert the index to integer, but keep whole list as list of strings
          content.sort(key=lambda a: int(a[0]))
          for c in range(len(content)):
              content[c][1] = f"<p>Step {c + 1}.</p>" + str(content[c][1])
          return content
  
      sorted_test = sorted(concatenate_headings_content((
          '<h1>Test</h1>',
          '<h1>Test .*</h1>')))
  
      sorted_step = sorted(enumerate_content(concatenate_headings_content((
          '<h2>Step</h2>',
          '<h2>Test Step</h2>'))) + sorted_test)
      step = ''.join([f"{v[1]}" for v in sorted_step])
  
      sorted_expect = sorted(enumerate_content(concatenate_headings_content((
          '<h2>Expect</h2>',
          '<h2>Result</h2>',
          '<h2>Expected Result</h2>'))) + sorted_test)
      expect = ''.join([f"{v[1]}" for v in sorted_expect])
  
      def check_section_exists(text: str) -> str:
          try:
              return str(sections_headings[text][0][1])
          except (IndexError, KeyError):
              return ''
  
      setup = check_section_exists('<h1>Setup</h1>')
      cleanup = check_section_exists('<h1>Cleanup</h1>')
  
      return step, expect, setup, cleanup
  
  
  def bz_set_coverage(bug_ids: List[int], case_id: str, tracker_id: int) -> None:
      """ Set coverage in Bugzilla """
      bz_instance = get_bz_instance()
  
      overall_pass = True
      no_email = 1  # Do not send emails about the change
      get_bz_dict = {
          'ids': bug_ids,
          'include_fields': ['id', 'external_bugs', 'flags']}
      bugs_data = bz_instance._proxy.Bug.get(get_bz_dict)
      for bug in bugs_data['bugs']:
          # Process flag (might fail for some types)
          bug_id = bug['id']
          if 'qe_test_coverage+' not in set(
                  [x['name'] + x['status'] for x in bug['flags']]):
              try:
                  bz_instance._proxy.Flag.update({
                      'ids': [bug_id],
                      'nomail': no_email,
                      'updates': [{
                          'name': 'qe_test_coverage',
                          'status': '+'
                          }]
                      })
              except xmlrpc.client.Fault as err:
                  # TODO: Fix missing overall_result = False, breaks tests
                  # if bug used for testing is not changed
                  # BZ#1925518 can't have qe_test_coverage flag
                  log.debug(f"Update flag failed: {err}")
                  echo(style(
                      f"Failed to set qe_test_coverage+ flag for BZ#{bug_id}",
                      fg='red'))
          # Process external tracker - should succeed
          current = set(
              [b['ext_bz_bug_id'] for b in bug['external_bugs']
               if b['ext_bz_id'] == tracker_id])
          if case_id not in current:
              query = {
                  'bug_ids': [bug_id],
                  'nomail': no_email,
                  'external_bugs': [{
                      'ext_type_id': tracker_id,
                      'ext_bz_bug_id': case_id,
                      'ext_description': '',
                      }]
                  }
              try:
                  bz_instance._proxy.ExternalBugs.add_external_bug(query)
              except Exception as err:
                  log.debug(f"Link case failed: {err}")
                  echo(style(f"Failed to link to BZ#{bug_id}", fg='red'))
                  overall_pass = False
      if not overall_pass:
          raise ConvertError("Failed to link the case to bugs.")
  
      echo(style("Linked to Bugzilla: {}.".format(
          " ".join([f"BZ#{bz_id}" for bz_id in bug_ids])), fg='magenta'))
  
  
  def get_polarion_ids(
          query_result: List[Any],
          preferred_project: Optional[str] = None) -> Tuple[str, Optional[str]]:
      """ Return case and project ids from query results """
      if not query_result:
          return 'None', None
      if len(query_result) == 1:
          return query_result[0].work_item_id, query_result[0].project_id
  
      if preferred_project:
          try:
              return next(
                  item.work_item_id for item in query_result
                  if item.project_id == preferred_project), preferred_project
          except StopIteration:
              pass
  
      for result in query_result:
          # If multiple cases are found prefer cases from other projects
          # than these legacy ones
          if str(result.project_id) not in LEGACY_POLARION_PROJECTS:
              return result.work_item_id, result.project_id
  
      return query_result[0].work_item_id, query_result[0].project_id
  
  
  def find_polarion_case_ids(
          data: Dict[str, str],
          preferred_project: Optional[str] = None) -> Tuple[str, Optional[str]]:
      """ Find IDs for Polarion case from data dictionary """
      assert PolarionWorkItem
  
      # Search by UUID
      if data.get(ID_KEY):
          query_result = PolarionWorkItem.query(
              data.get(ID_KEY), fields=['work_item_id', 'project_id'])
          return get_polarion_ids(query_result, preferred_project)
      # Search by TCMS Case ID
      extra_nitrate = data.get('extra-nitrate')
      if extra_nitrate:
          nitrate_case_id_search = re.search(r'\d+', extra_nitrate)
          if not nitrate_case_id_search:
              raise ConvertError(
                  "Could not find a valid nitrate testcase ID in 'extra-nitrate' attribute")
          nitrate_case_id = str(int(nitrate_case_id_search.group()))
          query_result = PolarionWorkItem.query(
              f"tcmscaseid:{nitrate_case_id}", fields=['work_item_id', 'project_id'])
          return get_polarion_ids(query_result, preferred_project)
      # Search by extra task
      if data.get('extra-task'):
          query_result = PolarionWorkItem.query(
              data.get('extra-task'), fields=['work_item_id', 'project_id'])
          return get_polarion_ids(query_result, preferred_project)
      return 'None', None
  
  
  def get_polarion_case(data: Dict[str, str], preferred_project: Optional[str] = None) -> Any:
      """ Get Polarion case through couple different methods """
      import_polarion()
  
      assert PolarionTestCase
      assert PolarionException
  
      polarion_id, project_id = find_polarion_case_ids(data, preferred_project)
  
      try:
          polarion_case = PolarionTestCase(
              project_id=project_id, work_item_id=polarion_id)
          echo(style(
              f"Test case '{str(polarion_case.work_item_id)}' found.",
              fg='blue'))
          return polarion_case
      except PolarionException:
          return None
  
  
  def enabled_somewhere(test: 'tmt.Test') -> bool:
      """ True if the test is enabled for some context (adjust rules) """
      # node.original_data are fmf data _before_ adjust was processed
      try:
          if test.node.original_data['enabled']:
              return True
      except KeyError:
          # Key defaults to True so we already know the outcome
          return True
  
      # Some rule in adjust enables the test
      try:
          adjust_rules = test.node.original_data['adjust']
          # TODO: Should not be necessary once we normalize data
          if isinstance(adjust_rules, dict):
              adjust_rules = [adjust_rules]
          for rule in adjust_rules:
              try:
                  if rule['enabled']:
                      return True
              except KeyError:
                  pass
      except KeyError:
          pass
      # At this point nothing enables the test
      return False
  
  
  def export_to_nitrate(test: 'tmt.Test') -> None:
      """ Export fmf metadata to nitrate test cases """
      import_nitrate()
      assert nitrate
      assert gssapi
  
      # Check command line options
      create = test.opt('create')
      general = test.opt('general')
      link_runs = test.opt('link_runs')
      duplicate = test.opt('duplicate')
      link_bugzilla = test.opt('bugzilla')
      ignore_git_validation = test.opt('ignore_git_validation')
      dry_mode = test.opt('dry')
  
      if link_runs:
          general = True
  
      # Check git is already correct
      valid, error_msg = tmt.utils.validate_git_status(test)
      if not valid:
          if ignore_git_validation:
              secho(f"Exporting regardless '{error_msg}'.", fg='red')
          else:
              raise ConvertError(
                  f"Can't export due '{error_msg}'.\n"
                  "Use --ignore-git-validation on your own risk to export regardless.")
  
      # Check nitrate test case
      try:
          nitrate_id = test.node.get('extra-nitrate')[3:]
          nitrate_case = nitrate.TestCase(int(nitrate_id))
          nitrate_case.summary  # Make sure we connect to the server now
          echo(style(f"Test case '{nitrate_case.identifier}' found.", fg='blue'))
      except TypeError:
          # Create a new nitrate test case
          if create:
              nitrate_case = None
              # Check for existing Nitrate tests with the same fmf id
              if not duplicate:
                  testcases = _nitrate_find_fmf_testcases(test)
                  try:
                      # Select the first found testcase if any
                      nitrate_case = next(testcases)
                  except StopIteration:
                      pass
              if not nitrate_case:
                  # Summary for TCMS case
                  extra_summary = prepare_extra_summary(test)
                  if not dry_mode:
                      nitrate_case = create_nitrate_case(extra_summary)
                  else:
                      echo(style(
                          f"Test case '{extra_summary}' created.", fg='blue'))
                  test._metadata['extra-summary'] = extra_summary
              # Either newly created or duplicate with missing extra-nitrate
              if nitrate_case:
                  echo(style("Append the nitrate test case id.", fg='green'))
                  if not dry_mode:
                      with test.node as data:
                          data["extra-nitrate"] = nitrate_case.identifier
          else:
              raise ConvertError(f"Nitrate test case id not found for {test}"
                                 " (You can use --create option to enforce"
                                 " creating testcases)")
      except (nitrate.NitrateError, gssapi.raw.misc.GSSError) as error:
          raise ConvertError(error)
  
      # Check if URL is accessible, to be able to reach from nitrate
      check_git_url(test.fmf_id['url'])
  
      # Summary
      try:
          summary = (test._metadata.get('extra-summary')
                     or test._metadata.get('extra-task')
                     or prepare_extra_summary(test))
      except ConvertError:
          summary = test.name
      if not dry_mode:
          nitrate_case.summary = summary
      echo(style('summary: ', fg='green') + summary)
  
      # Script
      if test.node.get('extra-task'):
          if not dry_mode:
              nitrate_case.script = test.node.get('extra-task')
          echo(style('script: ', fg='green') + test.node.get('extra-task'))
  
      # Components and General plan
      # First remove any components that are already there
      if not dry_mode:
          nitrate_case.components.clear()
      # Only these general plans should stay
      expected_general_plans = set()
      # Then add fmf ones
      if test.component:
          echo(style('components: ', fg='green') + ' '.join(test.component))
          for component in test.component:
              try:
                  nitrate_component = nitrate.Component(
                      name=component, product=DEFAULT_PRODUCT.id)
                  if not dry_mode:
                      nitrate_case.components.add(nitrate_component)
              except nitrate.xmlrpc_driver.NitrateError as error:
                  log.debug(error)
                  echo(style(
                      f"Failed to add component '{component}'.", fg='red'))
              if general:
                  try:
                      general_plan = find_general_plan(component)
                      expected_general_plans.add(general_plan)
                      echo(style(
                          f"Linked to general plan '{general_plan}'.",
                          fg='magenta'))
                      if not dry_mode:
                          nitrate_case.testplans.add(general_plan)
                      if link_runs:
                          add_to_nitrate_runs(
                              nitrate_case, general_plan, test, dry_mode)
                  except nitrate.NitrateError as error:
                      log.debug(error)
                      echo(style(
                          f"Failed to find general test plan for '{component}'.",
                          fg='red'))
      # Remove unexpected general plans
      if general and nitrate_case:
          # Remove also all general plans linked to testcase
          for nitrate_plan in [plan for plan in nitrate_case.testplans]:
              if (nitrate_plan.type.name == "General"
                      and nitrate_plan not in expected_general_plans):
                  echo(style(
                      f"Removed general plan '{nitrate_plan}'.", fg='red'))
                  if not dry_mode:
                      nitrate_case.testplans.remove(nitrate_plan)
  
      # Tags
      # Convert 'tier' attribute
  
  [... the rest is omitted ...]
files.old_path: tmt/export.py
files.new_contents: 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.new_path: tmt/steps/finish/ansible.py
files.old_contents: 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.old_path: tmt/steps/finish/ansible.py
files.new_contents: # 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' domain='0x0000' bus='0x00' slot='0x08' function='0x0'/>
      </disk>
      <interface type='{{ network_type }}'>
        <mac address="{{ mac_address }}"/>
        {{ network_source }}
        {{ ip_setup }}
        <model type='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>
      </interface>
      <serial type='pty'>
        <target port='0'/>
      </serial>
      <console type='pty'>
        <target type='serial' port='0'/>
      </console>
      <input type="keyboard" bus="virtio"/>
      <rng model='virtio'>
        <backend model='random'>/dev/urandom</backend>
      </rng>
    </devices>
    {{ qemu_args }}
  </domain>
  """  # noqa: E501
  
  # VM defaults
  DEFAULT_BOOT_TIMEOUT = 60      # seconds
  DEFAULT_CONNECT_TIMEOUT = 60   # seconds
  NON_KVM_ADDITIONAL_WAIT = 10       # seconds
  NON_KVM_TIMEOUT_COEF = 10          # times
  
  # SSH key type, set None for ssh-keygen default one
  SSH_KEYGEN_TYPE = "ecdsa"
  
  DEFAULT_USER = 'root'
  DEFAULT_MEMORY = 2048
  DEFAULT_DISK = 10
  DEFAULT_IMAGE = 'fedora'
  DEFAULT_CONNECTION = 'session'
  DEFAULT_ARCH = platform.machine()
  
  
  @dataclasses.dataclass
  class TestcloudGuestData(tmt.steps.provision.GuestSshData):
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      image: str = DEFAULT_IMAGE
      memory: int = DEFAULT_MEMORY
      disk: int = DEFAULT_DISK
      connection: str = DEFAULT_CONNECTION
      arch: str = DEFAULT_ARCH
  
      image_url: Optional[str] = None
      instance_name: Optional[str] = None
  
  
  @dataclasses.dataclass
  class ProvisionTestcloudData(TestcloudGuestData, tmt.steps.StepData):
      pass
  
  
  @tmt.steps.provides_method('virtual.testcloud')
  class ProvisionTestcloud(tmt.steps.provision.ProvisionPlugin):
      """
      Local virtual machine using testcloud
  
      Minimal config which uses the latest fedora image:
  
          provision:
              how: virtual
  
      Here's a full config example:
  
          provision:
              how: virtual
              image: fedora
              user: root
              memory: 2048
  
      As the image use 'fedora' for the latest released Fedora compose,
      'rawhide' for the latest Rawhide compose, short aliases such as
      'fedora-32', 'f-32' or 'f32' for specific release or a full url to
      the qcow2 image for example from:
  
          https://kojipkgs.fedoraproject.org/compose/
  
      Short names are also provided for 'centos', 'centos-stream',
      'debian' and 'ubuntu' (e.g. 'centos-8' or 'c8').
  
      Supported Fedora CoreOS images are:
  
          fedora-coreos
          fedora-coreos-stable
          fedora-coreos-testing
          fedora-coreos-next
  
      Use the full path for images stored on local disk, for example:
  
          /var/tmp/images/Fedora-Cloud-Base-31-1.9.x86_64.qcow2
  
      In addition to the qcow2 format, vagrant boxes can be used as well,
      testcloud will take care of unpacking the image for you.
      """
  
      _data_class = ProvisionTestcloudData
  
      # Guest instance
      _guest = None
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for testcloud """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to be used. Provide a short name, '
                       'full path to a local file or a complete url.'),
              click.option(
                  '-m', '--memory', metavar='MEMORY', type=int,
                  help='Set available memory in MB, 2048 MB by default.'),
              click.option(
                  '-D', '--disk', metavar='MEMORY', type=int,
                  help='Specify disk size in GB, 10 GB by default.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-c', '--connection',
                  type=click.Choice(['session', 'system']),
                  help="What session type to use, 'session' by default."),
              click.option(
                  '-a', '--arch',
                  type=click.Choice(['x86_64', 'aarch64', 's390x', 'ppc64le']),
                  help="What architecture to virtualize, host arch by default."),
              ]
          return options
  
      # TODO: Revisit this `type: ignore` once `Guest` becomes a generic type
      def wake(self, data: Optional[TestcloudGuestData] = None) -> None:  # type: ignore[override]
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
  
          # Wake up testcloud instance
          if data:
              guest = GuestTestcloud(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self) -> None:
          """ Provision the testcloud instance """
          super().go()
  
          # Give info about provided data
          data = TestcloudGuestData(**{
              key: self.get(key)
              for key in TestcloudGuestData.keys()
              })
  
          # Once plan schema is enforced this won't be necessary
          # click enforces int for cmdline and schema validation
          # will make sure 'int' gets from plan data.
          # Another key is 'port' however that is not exposed to the cli
          for int_key in ["memory", "disk", "port"]:
              value = getattr(data, int_key)
              if value is not None:
                  try:
                      setattr(data, int_key, int(value))
                  except ValueError:
                      raise tmt.utils.SpecificationError(
                          f"Value '{value}' cannot be converted to int for '{int_key}' attribute.")
  
          for key, value in data.to_dict().items():
              if key == 'memory':
                  self.info('memory', f"{value} MB", 'green')
              elif key == 'disk':
                  self.info('disk', f"{value} GB", 'green')
              elif key == 'connection':
                  self.verbose('connection', value, 'green')
              elif value is not None:
                  self.info(key, value, 'green')
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestTestcloud(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self) -> Optional[tmt.Guest]:
          """ Return the provisioned guest """
          return self._guest
  
      @classmethod
      def clean_images(cls, clean: Any, dry: Any) -> bool:
          """ Remove the testcloud images """
          clean.info('testcloud', shift=1, color='green')
          if not os.path.exists(TESTCLOUD_IMAGES):
              clean.warn(
                  f"Directory '{TESTCLOUD_IMAGES}' does not exist.", shift=2)
              return True
          successful = True
          for image in os.listdir(TESTCLOUD_IMAGES):
              image = os.path.join(TESTCLOUD_IMAGES, image)
              if dry:
                  clean.verbose(f"Would remove '{image}'.", shift=2)
              else:
                  clean.verbose(f"Removing '{image}'.", shift=2)
                  try:
                      os.remove(image)
                  except OSError:
                      clean.error(f"Failed to remove '{image}'.", shift=2)
                      successful = False
          return successful
  
  
  class GuestTestcloud(tmt.GuestSsh):
      """
      Testcloud Instance
  
      The following keys are expected in the 'data' dictionary::
  
          image ...... qcov image name or url
          user ....... user name to log in
          memory ..... memory size for vm
          disk ....... disk size for vm
          connection . either session (default) or system, to be passed to qemu
          arch ....... architecture for the VM, host arch is the default
      """
  
      # TODO: Revisit this `type: ignore` once `Guest` becomes a generic type
      _data_class = TestcloudGuestData  # type: ignore[assignment]
  
      image: str
      image_url: Optional[str]
      instance_name: Optional[str]
      memory: int
      disk: str
      connection: str
      arch: str
  
      # Not to be saved, recreated from image_url/instance_name/... every
      # time guest is instantiated.
      _image: Optional['testcloud.image.Image'] = None  # type: ignore[name-defined]
      _instance: Optional['testcloud.instance.Instance'] = None  # type: ignore[name-defined]
  
      def _get_url(self, url: str, message: str) -> requests.Response:
          """ Get url, retry when fails, return response """
  
          def try_get_url() -> requests.Response:
              try:
                  with retry_session() as session:
                      response = session.get(url)
  
                  if response.ok:
                      return response
  
              except requests.RequestException:
                  pass
              finally:
                  raise tmt.utils.WaitingIncomplete()
  
          try:
              return tmt.utils.wait(
                  self, try_get_url, datetime.timedelta(
                      seconds=DEFAULT_CONNECT_TIMEOUT), tick=1)
  
          except tmt.utils.WaitingTimedOutError:
              raise ProvisionError(
                  f'Failed to {message} in {DEFAULT_CONNECT_TIMEOUT}s.')
  
      def _guess_image_url(self, name: str) -> str:
          """ Guess image url for given name """
  
          # Try to check if given url is a local file
          if os.path.isabs(name) and os.path.isfile(name):
              return f'file://{name}'
  
          name = name.lower().strip()
          url = None
  
          # Map fedora aliases (e.g. rawhide, fedora, fedora-32, f-32, f32)
          matched_fedora = re.match(r'^f(edora)?-?(\d+)$', name)
          # Map fedora coreos aliases (e.g. stable, next or testing)
          matched_fedora_coreos = re.match(r'^f(edora-coreos)?-?(stable|testing|next)$', name)
          # Map centos aliases (e.g. centos:X, centos, centos-stream:X)
          matched_centos = [re.match(r'^c(entos)?-?(\d+)$', name),
                            re.match(r'^c(entos-stream)?-?(\d+)$', name)]
          matched_ubuntu = re.match(r'^u(buntu)?-?(\w+)$', name)
          matched_debian = re.match(r'^d(ebian)?-?(\w+)$', name)
  
          assert testcloud is not None
          # Plain name match means we want the latest release
          if name == 'fedora':
              url = testcloud.util.get_fedora_image_url("latest", self.arch)
          elif name == 'fedora-coreos':
              url = testcloud.util.get_fedora_image_url("stable", self.arch)
          elif name == 'centos':
              url = testcloud.util.get_centos_image_url("latest", self.arch)
          elif name == 'centos-stream':
              url = testcloud.util.get_centos_image_url(
                  "latest", stream=True, arch=self.arch)
          elif name == 'ubuntu':
              url = testcloud.util.get_ubuntu_image_url("latest", self.arch)
          elif name == 'debian':
              url = testcloud.util.get_debian_image_url("latest", self.arch)
  
          elif matched_fedora:
              url = testcloud.util.get_fedora_image_url(
                  matched_fedora.group(2), self.arch)
          elif matched_fedora_coreos:
              url = testcloud.util.get_fedora_image_url(
                  matched_fedora_coreos.group(2), self.arch)
          elif matched_centos[0]:
              url = testcloud.util.get_centos_image_url(
                  matched_centos[0].group(2), self.arch)
          elif matched_centos[1]:
              url = testcloud.util.get_centos_image_url(
                  matched_centos[1].group(2), stream=True, arch=self.arch)
          elif matched_ubuntu:
              url = testcloud.util.get_ubuntu_image_url(
                  matched_ubuntu.group(2), self.arch)
          elif matched_debian:
              url = testcloud.util.get_debian_image_url(
                  matched_debian.group(2), self.arch)
          elif 'rawhide' in name:
              url = testcloud.util.get_fedora_image_url("rawhide", self.arch)
  
          if not url:
              raise ProvisionError(f"Could not map '{name}' to compose.")
          return cast(str, url)
  
      @staticmethod
      def _create_template() -> None:
          """ Create libvirt domain template """
          # Write always to ovewrite possible outdated version
          with open(DOMAIN_TEMPLATE_FILE, 'w') as template:
              template.write(DOMAIN_TEMPLATE)
  
      def wake(self) -> None:
          """ Wake up the guest """
          self.debug(
              f"Waking up testcloud instance '{self.instance_name}'.",
              level=2, shift=0)
          self.prepare_config()
          assert testcloud is not None
          self._image = testcloud.image.Image(self.image_url)
          self._instance = testcloud.instance.Instance(
              self.instance_name, image=self._image,
              connection=f"qemu:///{self.connection}", desired_arch=self.arch)
  
      def prepare_ssh_key(self, key_type: Optional[str] = None) -> None:
          """ Prepare ssh key for authentication """
          # Create ssh key paths
          key_name = "id_{}".format(key_type if key_type is not None else 'rsa')
          assert self.workdir is not None
          self.key = [os.path.join(self.workdir, key_name)]
          self.pubkey = os.path.join(self.workdir, f'{key_name}.pub')
  
          # Generate ssh key
          self.debug('Generating an ssh key.')
          command = ["ssh-keygen", "-f", self.key[0], "-N", ""]
          if key_type is not None:
              command.extend(["-t", key_type])
          self.run(command)
          with open(self.pubkey, 'r') as pubkey:
              self.config.USER_DATA = USER_DATA.format(
                  user_name=self.user, public_key=pubkey.read())
          with open(self.pubkey, 'r') as pubkey:
              self.config.COREOS_DATA = COREOS_DATA.format(
                  user_name=self.user, public_key=pubkey.read())
  
      def prepare_config(self) -> None:
          """ Prepare common configuration """
         
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/testcloud.py
files.old_contents: # 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' domain='0x0000' bus='0x00' slot='0x08' function='0x0'/>
      </disk>
      <interface type='{{ network_type }}'>
        <mac address="{{ mac_address }}"/>
        {{ network_source }}
        {{ ip_setup }}
        <model type='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>
      </interface>
      <serial type='pty'>
        <target port='0'/>
      </serial>
      <console type='pty'>
        <target type='serial' port='0'/>
      </console>
      <input type="keyboard" bus="virtio"/>
      <rng model='virtio'>
        <backend model='random'>/dev/urandom</backend>
      </rng>
    </devices>
    {{ qemu_args }}
  </domain>
  """  # noqa: E501
  
  # VM defaults
  DEFAULT_BOOT_TIMEOUT = 60      # seconds
  DEFAULT_CONNECT_TIMEOUT = 60   # seconds
  NON_KVM_ADDITIONAL_WAIT = 10       # seconds
  NON_KVM_TIMEOUT_COEF = 10          # times
  
  # SSH key type, set None for ssh-keygen default one
  SSH_KEYGEN_TYPE = "ecdsa"
  
  DEFAULT_USER = 'root'
  DEFAULT_MEMORY = 2048
  DEFAULT_DISK = 10
  DEFAULT_IMAGE = 'fedora'
  DEFAULT_CONNECTION = 'session'
  DEFAULT_ARCH = platform.machine()
  
  
  @dataclasses.dataclass
  class TestcloudGuestData(tmt.steps.provision.GuestSshData):
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      image: str = DEFAULT_IMAGE
      memory: int = DEFAULT_MEMORY
      disk: int = DEFAULT_DISK
      connection: str = DEFAULT_CONNECTION
      arch: str = DEFAULT_ARCH
  
      image_url: Optional[str] = None
      instance_name: Optional[str] = None
  
  
  @dataclasses.dataclass
  class ProvisionTestcloudData(TestcloudGuestData, tmt.steps.StepData):
      pass
  
  
  @tmt.steps.provides_method('virtual.testcloud')
  class ProvisionTestcloud(tmt.steps.provision.ProvisionPlugin):
      """
      Local virtual machine using testcloud
  
      Minimal config which uses the latest fedora image:
  
          provision:
              how: virtual
  
      Here's a full config example:
  
          provision:
              how: virtual
              image: fedora
              user: root
              memory: 2048
  
      As the image use 'fedora' for the latest released Fedora compose,
      'rawhide' for the latest Rawhide compose, short aliases such as
      'fedora-32', 'f-32' or 'f32' for specific release or a full url to
      the qcow2 image for example from:
  
          https://kojipkgs.fedoraproject.org/compose/
  
      Short names are also provided for 'centos', 'centos-stream',
      'debian' and 'ubuntu' (e.g. 'centos-8' or 'c8').
  
      Supported Fedora CoreOS images are:
  
          fedora-coreos
          fedora-coreos-stable
          fedora-coreos-testing
          fedora-coreos-next
  
      Use the full path for images stored on local disk, for example:
  
          /var/tmp/images/Fedora-Cloud-Base-31-1.9.x86_64.qcow2
  
      In addition to the qcow2 format, vagrant boxes can be used as well,
      testcloud will take care of unpacking the image for you.
      """
  
      _data_class = ProvisionTestcloudData
  
      # Guest instance
      _guest = None
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for testcloud """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to be used. Provide a short name, '
                       'full path to a local file or a complete url.'),
              click.option(
                  '-m', '--memory', metavar='MEMORY', type=int,
                  help='Set available memory in MB, 2048 MB by default.'),
              click.option(
                  '-D', '--disk', metavar='MEMORY', type=int,
                  help='Specify disk size in GB, 10 GB by default.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-c', '--connection',
                  type=click.Choice(['session', 'system']),
                  help="What session type to use, 'session' by default."),
              click.option(
                  '-a', '--arch',
                  type=click.Choice(['x86_64', 'aarch64', 's390x', 'ppc64le']),
                  help="What architecture to virtualize, host arch by default."),
              ]
          return options
  
      # TODO: Revisit this `type: ignore` once `Guest` becomes a generic type
      def wake(self, data: Optional[TestcloudGuestData] = None) -> None:  # type: ignore[override]
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
  
          # Wake up testcloud instance
          if data:
              guest = GuestTestcloud(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self) -> None:
          """ Provision the testcloud instance """
          super().go()
  
          # Give info about provided data
          data = TestcloudGuestData(**{
              key: self.get(key)
              for key in TestcloudGuestData.keys()
              })
  
          # Once plan schema is enforced this won't be necessary
          # click enforces int for cmdline and schema validation
          # will make sure 'int' gets from plan data.
          # Another key is 'port' however that is not exposed to the cli
          for int_key in ["memory", "disk", "port"]:
              value = getattr(data, int_key)
              if value is not None:
                  try:
                      setattr(data, int_key, int(value))
                  except ValueError:
                      raise tmt.utils.SpecificationError(
                          f"Value '{value}' cannot be converted to int for '{int_key}' attribute.")
  
          for key, value in data.to_dict().items():
              if key == 'memory':
                  self.info('memory', f"{value} MB", 'green')
              elif key == 'disk':
                  self.info('disk', f"{value} GB", 'green')
              elif key == 'connection':
                  self.verbose('connection', value, 'green')
              elif value is not None:
                  self.info(key, value, 'green')
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestTestcloud(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self) -> Optional[tmt.Guest]:
          """ Return the provisioned guest """
          return self._guest
  
      @classmethod
      def clean_images(cls, clean: Any, dry: Any) -> bool:
          """ Remove the testcloud images """
          clean.info('testcloud', shift=1, color='green')
          if not os.path.exists(TESTCLOUD_IMAGES):
              clean.warn(
                  f"Directory '{TESTCLOUD_IMAGES}' does not exist.", shift=2)
              return True
          successful = True
          for image in os.listdir(TESTCLOUD_IMAGES):
              image = os.path.join(TESTCLOUD_IMAGES, image)
              if dry:
                  clean.verbose(f"Would remove '{image}'.", shift=2)
              else:
                  clean.verbose(f"Removing '{image}'.", shift=2)
                  try:
                      os.remove(image)
                  except OSError:
                      clean.error(f"Failed to remove '{image}'.", shift=2)
                      successful = False
          return successful
  
  
  class GuestTestcloud(tmt.GuestSsh):
      """
      Testcloud Instance
  
      The following keys are expected in the 'data' dictionary::
  
          image ...... qcov image name or url
          user ....... user name to log in
          memory ..... memory size for vm
          disk ....... disk size for vm
          connection . either session (default) or system, to be passed to qemu
          arch ....... architecture for the VM, host arch is the default
      """
  
      # TODO: Revisit this `type: ignore` once `Guest` becomes a generic type
      _data_class = TestcloudGuestData  # type: ignore
  
      image: str
      image_url: Optional[str]
      instance_name: Optional[str]
      memory: int
      disk: str
      connection: str
      arch: str
  
      # Not to be saved, recreated from image_url/instance_name/... every
      # time guest is instantiated.
      _image: Optional['testcloud.image.Image'] = None  # type: ignore
      _instance: Optional['testcloud.instance.Instance'] = None  # type: ignore
  
      def _get_url(self, url: str, message: str) -> requests.Response:
          """ Get url, retry when fails, return response """
  
          def try_get_url() -> requests.Response:
              try:
                  with retry_session() as session:
                      response = session.get(url)
  
                  if response.ok:
                      return response
  
              except requests.RequestException:
                  pass
              finally:
                  raise tmt.utils.WaitingIncomplete()
  
          try:
              return tmt.utils.wait(
                  self, try_get_url, datetime.timedelta(
                      seconds=DEFAULT_CONNECT_TIMEOUT), tick=1)
  
          except tmt.utils.WaitingTimedOutError:
              raise ProvisionError(
                  f'Failed to {message} in {DEFAULT_CONNECT_TIMEOUT}s.')
  
      def _guess_image_url(self, name: str) -> str:
          """ Guess image url for given name """
  
          # Try to check if given url is a local file
          if os.path.isabs(name) and os.path.isfile(name):
              return f'file://{name}'
  
          name = name.lower().strip()
          url = None
  
          # Map fedora aliases (e.g. rawhide, fedora, fedora-32, f-32, f32)
          matched_fedora = re.match(r'^f(edora)?-?(\d+)$', name)
          # Map fedora coreos aliases (e.g. stable, next or testing)
          matched_fedora_coreos = re.match(r'^f(edora-coreos)?-?(stable|testing|next)$', name)
          # Map centos aliases (e.g. centos:X, centos, centos-stream:X)
          matched_centos = [re.match(r'^c(entos)?-?(\d+)$', name),
                            re.match(r'^c(entos-stream)?-?(\d+)$', name)]
          matched_ubuntu = re.match(r'^u(buntu)?-?(\w+)$', name)
          matched_debian = re.match(r'^d(ebian)?-?(\w+)$', name)
  
          assert testcloud is not None
          # Plain name match means we want the latest release
          if name == 'fedora':
              url = testcloud.util.get_fedora_image_url("latest", self.arch)
          elif name == 'fedora-coreos':
              url = testcloud.util.get_fedora_image_url("stable", self.arch)
          elif name == 'centos':
              url = testcloud.util.get_centos_image_url("latest", self.arch)
          elif name == 'centos-stream':
              url = testcloud.util.get_centos_image_url(
                  "latest", stream=True, arch=self.arch)
          elif name == 'ubuntu':
              url = testcloud.util.get_ubuntu_image_url("latest", self.arch)
          elif name == 'debian':
              url = testcloud.util.get_debian_image_url("latest", self.arch)
  
          elif matched_fedora:
              url = testcloud.util.get_fedora_image_url(
                  matched_fedora.group(2), self.arch)
          elif matched_fedora_coreos:
              url = testcloud.util.get_fedora_image_url(
                  matched_fedora_coreos.group(2), self.arch)
          elif matched_centos[0]:
              url = testcloud.util.get_centos_image_url(
                  matched_centos[0].group(2), self.arch)
          elif matched_centos[1]:
              url = testcloud.util.get_centos_image_url(
                  matched_centos[1].group(2), stream=True, arch=self.arch)
          elif matched_ubuntu:
              url = testcloud.util.get_ubuntu_image_url(
                  matched_ubuntu.group(2), self.arch)
          elif matched_debian:
              url = testcloud.util.get_debian_image_url(
                  matched_debian.group(2), self.arch)
          elif 'rawhide' in name:
              url = testcloud.util.get_fedora_image_url("rawhide", self.arch)
  
          if not url:
              raise ProvisionError(f"Could not map '{name}' to compose.")
          return cast(str, url)
  
      @staticmethod
      def _create_template() -> None:
          """ Create libvirt domain template """
          # Write always to ovewrite possible outdated version
          with open(DOMAIN_TEMPLATE_FILE, 'w') as template:
              template.write(DOMAIN_TEMPLATE)
  
      def wake(self) -> None:
          """ Wake up the guest """
          self.debug(
              f"Waking up testcloud instance '{self.instance_name}'.",
              level=2, shift=0)
          self.prepare_config()
          assert testcloud is not None
          self._image = testcloud.image.Image(self.image_url)
          self._instance = testcloud.instance.Instance(
              self.instance_name, image=self._image,
              connection=f"qemu:///{self.connection}", desired_arch=self.arch)
  
      def prepare_ssh_key(self, key_type: Optional[str] = None) -> None:
          """ Prepare ssh key for authentication """
          # Create ssh key paths
          key_name = "id_{}".format(key_type if key_type is not None else 'rsa')
          assert self.workdir is not None
          self.key = [os.path.join(self.workdir, key_name)]
          self.pubkey = os.path.join(self.workdir, f'{key_name}.pub')
  
          # Generate ssh key
          self.debug('Generating an ssh key.')
          command = ["ssh-keygen", "-f", self.key[0], "-N", ""]
          if key_type is not None:
              command.extend(["-t", key_type])
          self.run(command)
          with open(self.pubkey, 'r') as pubkey:
              self.config.USER_DATA = USER_DATA.format(
                  user_name=self.user, public_key=pubkey.read())
          with open(self.pubkey, 'r') as pubkey:
              self.config.COREOS_DATA = COREOS_DATA.format(
                  user_name=self.user, public_key=pubkey.read())
  
      def prepare_config(self) -> None:
          """ Prepare common configuration """
          import_testcloud()
  
          #
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/testcloud.py
files.new_contents: 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.new_path: tmt/steps/report/junit.py
files.old_contents: 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.old_path: tmt/steps/report/junit.py
files.new_contents: 
  """ 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.
  
      :param value: optional value to print at right side of ``key``.
      :param color: optional color to apply on ``key``.
      :param level: number of indentation levels. Each level is indented
                    by :py:data:`INDENT` spaces.
      """
  
      indent = ' ' * INDENT * level
      deeper = ' ' * INDENT * (level + 1)
  
      # Colorize
      if color is not None:
          key = style(key, fg=color)
  
      # Handle key only
      if value is None:
          message = key
  
      # Handle key + value
      else:
          # Multiline content indented deeper
          if isinstance(value, str):
              lines = value.splitlines()
              if len(lines) > 1:
                  value = ''.join([f"\n{deeper}{line}" for line in lines])
  
          message = f'{key}: {value}'
  
      return indent + message
  
  
  class Config:
      """ User configuration """
  
      def __init__(self) -> None:
          """ Initialize config directory path """
          self.path = os.path.expanduser(CONFIG_PATH)
          if not os.path.exists(self.path):
              try:
                  os.makedirs(self.path)
              except OSError as error:
                  raise GeneralError(
                      f"Failed to create config '{self.path}'.\n{error}")
  
      def last_run(self, run_id: Optional[str] = None) -> Optional[str]:
          """ Get and set last run id """
          symlink = os.path.join(self.path, 'last-run')
          if run_id:
              try:
                  os.remove(symlink)
              except OSError:
                  pass
              try:
                  os.symlink(run_id, symlink)
              except FileExistsError:
                  # Race when tmt runs in parallel
                  log.warning(f"Race condition, unable to save last run '{run_id}'.")
              except OSError as error:
                  raise GeneralError(
                      f"Unable to save last run '{self.path}'.\n{error}")
              return run_id
          if os.path.islink(symlink):
              return os.path.realpath(symlink)
          return None
  
  
  class StreamLogger(Thread):
      """
      Reading pipes of running process in threads.
  
      Code based on:
      https://github.com/packit/packit/blob/main/packit/utils/logging.py#L10
      """
  
      def __init__(self,
                   stream: Optional[IO[bytes]],
                   log_header: str,
                   logger: BaseLoggerFnType) -> None:
          super().__init__(daemon=True)
          self.stream = stream
          self.output: List[str] = []
          self.log_header = log_header
          self.logger = logger
  
      def run(self) -> None:
          if self.stream is None:
              return
  
          for _line in self.stream:
              line = _line.decode('utf-8', errors='replace')
              if line != '':
                  self.logger(
                      self.log_header,
                      line.rstrip('\n'),
                      'yellow',
                      level=3)
              self.output.append(line)
  
      def get_output(self) -> str:
          return "".join(self.output)
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Common
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  CommonDerivedType = TypeVar('CommonDerivedType', bound='Common')
  
  
  class CommandOutput(NamedTuple):
      stdout: Optional[str]
      stderr: Optional[str]
  
  
  class Common:
      """
      Common shared stuff
  
      Takes care of command line context, options and workdir handling.
      Provides logging functions info(), verbose() and debug().
      Implements read() and write() for comfortable file access.
      Provides the run() method for easy command execution.
      """
  
      # Command line context, options and workdir
      _context: Optional[click.Context] = None
      # When set to true, _opt will be ignored (default will be returned)
      ignore_class_options: bool = False
      _options: Dict[str, Any] = dict()
      _workdir: WorkdirType = None
  
      # TODO: must be declared outside of __init__(), because it must exist before
      # __init__() gets called to allow logging helpers work correctly when used
      # from mixins. But that's not very clean, is it? :( Maybe decoupling logging
      # from Common class would help, such a class would be able to initialize
      # itself without involving the rest of Common code. On the other hand,
      # Common owns workdir, for example, whose value affects logging too, so no
      # clear solution so far.
      #
      # Note: cannot use CommonDerivedType - it's a TypeVar filled in by the type
      # given to __init__() and therefore the type it's representing *now* is
      # unknown. but we know `parent` will be derived from `Common` class, so it's
      # mostly fine.
      parent: Optional['Common'] = None
  
      def __init__(
              self,
              parent: Optional[CommonDerivedType] = None,
              name: Optional[str] = None,
              workdir: WorkdirArgumentType = None,
              context: Optional[click.Context] = None,
              relative_indent: int = 1,
              **kwargs: Any) -> None:
          """
          Initialize name and relation with the parent object
  
          Prepare the workdir for provided id / directory path
          or generate a new workdir name if workdir=True given.
          Store command line context and options for future use
          if context is provided.
          """
          # Use lowercase class name as the default name
          self.name = name or self.__class__.__name__.lower()
          self.parent = parent
  
          # Relative log indent level shift against the parent
          self._relative_indent = relative_indent
  
          # Store command line context
          if context:
              self._save_context_to_instance(context)
  
          # Initialize the workdir if requested
          self._workdir_load(workdir)
  
      def __str__(self) -> str:
          """ Name is the default string representation """
          return self.name
  
      @classmethod
      def _save_context(cls, context: click.Context) -> None:
          """ Save provided command line context and options for future use """
          cls._context = context
          cls._options = context.params
  
      def _save_context_to_instance(self, context: click.Context) -> None:
          """ Save provided command line context and options to the instance """
          self._context = context
          self._options = context.params
  
      @overload
      @classmethod
      def _opt(cls, option: str) -> Any:
          pass
  
      @overload
      @classmethod
      def _opt(cls, option: str, default: T) -> T:
          pass
  
      @classmethod
      def _opt(cls, option: str, default: Any = None) -> Any:
          """ Get an option from the command line context (class version) """
          if cls.ignore_class_options:
              return default
          return cls._options.get(option, default)
  
      def _fmf_context(self) -> FmfContextType:
          """ Return the current fmf context """
          if self._context is None:
              return dict()
  
          try:
              return cast(FmfContextType, self._context.obj.fmf_context)
          except AttributeError:
              return dict()
  
      def opt(self, option: str, default: Optional[Any] = None) -> Any:
          """
          Get an option from the command line options
  
          Checks also parent options. For flags (boolean values) parent's
          True wins over child's False (e.g. run --quiet enables quiet
          mode for all included plans and steps).
  
          For options that can be used multiple times, the child overrides
          the parent if it was defined (e.g. run -av provision -vvv runs
          all steps except for provision in mildly verbose mode, provision
          is run with the most verbosity).
  
          Environment variables override command line options.
          """
          # Translate dashes to underscores to match click's conversion
          option = option.replace('-', '_')
          # Check the environment first
          if option == 'debug':
              try:
                  debug = os.environ['TMT_DEBUG']
                  return int(debug)
              except ValueError:
                  raise GeneralError(
                      f"Invalid debug level '{debug}', use an integer.")
              except KeyError:
                  pass
  
          # Get local option
          local = self._options.get(option, default)
          # Check parent option
          parent = None
          if self.parent:
              parent = self.parent.opt(option)
          # Special handling for special flags (parent's yes always wins)
          if option in ['quiet', 'force', 'dry']:
              return parent if parent else local
          # Special handling for counting options (child overrides the
          # parent if it was defined)
          elif option in ['debug', 'verbose']:
              winner = local if local else parent
              if winner is None:
                  winner = 0
              return winner
          else:
              return parent if parent is not None else local
  
      def _level(self) -> int:
          """ Hierarchy level """
          if self.parent is None:
              return -1
          else:
              return self.parent._level() + self._relative_indent
  
      def _indent(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0) -> str:
          """ Indent message according to the object hierarchy """
  
          return indent(
              key,
              value=value,
              color=color,
              level=self._level() + shift)
  
      def _log(self, message: str) -> None:
          """ Append provided message to the current log """
          # Nothing to do if there is no workdir
          if self.workdir is None:
              return
  
          # Store log only in the top parent
          if self.parent:
              self.parent._log(message)
          else:
              with open(os.path.join(self.workdir, LOG_FILENAME), 'a') as log:
                  log.write(datetime.datetime.utcnow().strftime('%H:%M:%S') + ' '
                            + remove_color(message) + '\n')
  
      def print(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          """ Print a message regardless the quiet mode """
          self._log(self._indent(key, value, color=None, shift=shift))
          echo(self._indent(key, value, color, shift), err=err)
  
      def info(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          """ Show a message unless in quiet mode """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet'):
              echo(self._indent(key, value, color, shift), err=err)
  
      def verbose(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          """
          Show message if in requested verbose mode level
  
          In quiet mode verbose messages are not displayed.
          """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet') and self.opt('verbose') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def debug(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          """
          Show message if in requested debug mode level
  
          In quiet mode debug messages are not displayed.
          """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet') and self.opt('debug') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def warn(self, message: str, shift: int = 0) -> None:
          """ Show a yellow warning message on info level, send to stderr """
          self.info('warn', message, color='yellow', shift=shift, err=True)
  
      def fail(self, message: str, shift: int = 0) -> None:
          """ Show a red failure message on info level, send to stderr """
          self.info('fail', message, color='red', shift=shift, err=True)
  
      def _run(self,
               command: Union[str, List[str]],
               cwd: Optional[str],
               shell: bool,
               env: Optional[EnvironmentType],
               log: Optional[BaseLoggerFnType],
               join: bool = False,
               interactive: bool = False,
               timeout: Optional[int] = None) -> CommandOutput:
          """
          Run command, capture the output
  
          By default stdout and stderr are captured separately.
          Use join=True to merge stderr into stdout.
          Use timeout=<seconds> to finish process after given time
          """
          # By default command ouput is logged using debug
          if not log:
              log = self.debug
          # Prepare the environment
          if env:
              if not isinstance(env, dict):
                  raise GeneralError(f"Invalid environment '{env}'.")
              # Do not modify current process environment
              environment = os.environ.copy()
              environment.update(env)
          else:
              environment = None
          self.debug('environment', pprint.pformat(environment), level=4)
  
          # Set only for shell=True as it would affect command
          executable = DEFAULT_SHELL if shell else None
  
          # Run the command in interactive mode if requested
          if interactive:
              try:
                  subprocess.run(
                      command, cwd=cwd, shell=shell, env=environment, check=True,
                      executable=executable)
              except subprocess.CalledProcessError:
                  # Interactive mode can return non-zero if the last command
                  # failed, ignore errors here
                  pass
              finally:
                  return CommandOutput(None, None)
  
          # Create the process
          try:
              process = subprocess.Popen(
                  command, cwd=cwd, shell=shell, env=environment,
                  stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
                  stderr=subprocess.STDOUT if join else subprocess.PIPE,
                  executable=executable)
          except FileNotFoundError as error:
              raise RunError(
                  f"File '{error.filename}' not found.", command, 127)
  
          stdout_thread = StreamLogger(
              process.stdout, log_header='out', logger=log)
          stderr_thread = stdout_thread
          if not join:
              stderr_thread = StreamLogger(
                  process.stderr, log_header='err', logger=log)
          stdout_thread.start()
          if not join:
              stderr_thread.start()
          try:
              process.wait(timeout=timeout)
          except subprocess.TimeoutExpired:
              process.kill()
              process.returncode = PROCESS_TIMEOUT
          stdout_thread.join()
          if not join:
              stderr_thread.join()
  
          # Handle the exit code, return output
          if process.returncode != 0:
              raise RunError(
                  message=f"Command returned '{process.returncode}'.",
                  command=command,
                  returncode=process.returncode,
  
  
  [... the rest is omitted ...]
files.new_path: tmt/utils.py
files.old_contents: 
  """ 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.
  
      :param value: optional value to print at right side of ``key``.
      :param color: optional color to apply on ``key``.
      :param level: number of indentation levels. Each level is indented
                    by :py:data:`INDENT` spaces.
      """
  
      indent = ' ' * INDENT * level
      deeper = ' ' * INDENT * (level + 1)
  
      # Colorize
      if color is not None:
          key = style(key, fg=color)
  
      # Handle key only
      if value is None:
          message = key
  
      # Handle key + value
      else:
          # Multiline content indented deeper
          if isinstance(value, str):
              lines = value.splitlines()
              if len(lines) > 1:
                  value = ''.join([f"\n{deeper}{line}" for line in lines])
  
          message = f'{key}: {value}'
  
      return indent + message
  
  
  class Config:
      """ User configuration """
  
      def __init__(self) -> None:
          """ Initialize config directory path """
          self.path = os.path.expanduser(CONFIG_PATH)
          if not os.path.exists(self.path):
              try:
                  os.makedirs(self.path)
              except OSError as error:
                  raise GeneralError(
                      f"Failed to create config '{self.path}'.\n{error}")
  
      def last_run(self, run_id: Optional[str] = None) -> Optional[str]:
          """ Get and set last run id """
          symlink = os.path.join(self.path, 'last-run')
          if run_id:
              try:
                  os.remove(symlink)
              except OSError:
                  pass
              try:
                  os.symlink(run_id, symlink)
              except FileExistsError:
                  # Race when tmt runs in parallel
                  log.warning(f"Race condition, unable to save last run '{run_id}'.")
              except OSError as error:
                  raise GeneralError(
                      f"Unable to save last run '{self.path}'.\n{error}")
              return run_id
          if os.path.islink(symlink):
              return os.path.realpath(symlink)
          return None
  
  
  class StreamLogger(Thread):
      """
      Reading pipes of running process in threads.
  
      Code based on:
      https://github.com/packit/packit/blob/main/packit/utils/logging.py#L10
      """
  
      def __init__(self,
                   stream: Optional[IO[bytes]],
                   log_header: str,
                   logger: BaseLoggerFnType) -> None:
          super().__init__(daemon=True)
          self.stream = stream
          self.output: List[str] = []
          self.log_header = log_header
          self.logger = logger
  
      def run(self) -> None:
          if self.stream is None:
              return
  
          for _line in self.stream:
              line = _line.decode('utf-8', errors='replace')
              if line != '':
                  self.logger(
                      self.log_header,
                      line.rstrip('\n'),
                      'yellow',
                      level=3)
              self.output.append(line)
  
      def get_output(self) -> str:
          return "".join(self.output)
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Common
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  CommonDerivedType = TypeVar('CommonDerivedType', bound='Common')
  
  
  class CommandOutput(NamedTuple):
      stdout: Optional[str]
      stderr: Optional[str]
  
  
  class Common:
      """
      Common shared stuff
  
      Takes care of command line context, options and workdir handling.
      Provides logging functions info(), verbose() and debug().
      Implements read() and write() for comfortable file access.
      Provides the run() method for easy command execution.
      """
  
      # Command line context, options and workdir
      _context: Optional[click.Context] = None
      # When set to true, _opt will be ignored (default will be returned)
      ignore_class_options: bool = False
      _options: Dict[str, Any] = dict()
      _workdir: WorkdirType = None
  
      # TODO: must be declared outside of __init__(), because it must exist before
      # __init__() gets called to allow logging helpers work correctly when used
      # from mixins. But that's not very clean, is it? :( Maybe decoupling logging
      # from Common class would help, such a class would be able to initialize
      # itself without involving the rest of Common code. On the other hand,
      # Common owns workdir, for example, whose value affects logging too, so no
      # clear solution so far.
      #
      # Note: cannot use CommonDerivedType - it's a TypeVar filled in by the type
      # given to __init__() and therefore the type it's representing *now* is
      # unknown. but we know `parent` will be derived from `Common` class, so it's
      # mostly fine.
      parent: Optional['Common'] = None
  
      def __init__(
              self,
              parent: Optional[CommonDerivedType] = None,
              name: Optional[str] = None,
              workdir: WorkdirArgumentType = None,
              context: Optional[click.Context] = None,
              relative_indent: int = 1,
              **kwargs: Any) -> None:
          """
          Initialize name and relation with the parent object
  
          Prepare the workdir for provided id / directory path
          or generate a new workdir name if workdir=True given.
          Store command line context and options for future use
          if context is provided.
          """
          # Use lowercase class name as the default name
          self.name = name or self.__class__.__name__.lower()
          self.parent = parent
  
          # Relative log indent level shift against the parent
          self._relative_indent = relative_indent
  
          # Store command line context
          if context:
              self._save_context_to_instance(context)
  
          # Initialize the workdir if requested
          self._workdir_load(workdir)
  
      def __str__(self) -> str:
          """ Name is the default string representation """
          return self.name
  
      @classmethod
      def _save_context(cls, context: click.Context) -> None:
          """ Save provided command line context and options for future use """
          cls._context = context
          cls._options = context.params
  
      def _save_context_to_instance(self, context: click.Context) -> None:
          """ Save provided command line context and options to the instance """
          self._context = context
          self._options = context.params
  
      @overload
      @classmethod
      def _opt(cls, option: str) -> Any:
          pass
  
      @overload
      @classmethod
      def _opt(cls, option: str, default: T) -> T:
          pass
  
      @classmethod
      def _opt(cls, option: str, default: Any = None) -> Any:
          """ Get an option from the command line context (class version) """
          if cls.ignore_class_options:
              return default
          return cls._options.get(option, default)
  
      def _fmf_context(self) -> FmfContextType:
          """ Return the current fmf context """
          if self._context is None:
              return dict()
  
          try:
              return cast(FmfContextType, self._context.obj.fmf_context)
          except AttributeError:
              return dict()
  
      def opt(self, option: str, default: Optional[Any] = None) -> Any:
          """
          Get an option from the command line options
  
          Checks also parent options. For flags (boolean values) parent's
          True wins over child's False (e.g. run --quiet enables quiet
          mode for all included plans and steps).
  
          For options that can be used multiple times, the child overrides
          the parent if it was defined (e.g. run -av provision -vvv runs
          all steps except for provision in mildly verbose mode, provision
          is run with the most verbosity).
  
          Environment variables override command line options.
          """
          # Translate dashes to underscores to match click's conversion
          option = option.replace('-', '_')
          # Check the environment first
          if option == 'debug':
              try:
                  debug = os.environ['TMT_DEBUG']
                  return int(debug)
              except ValueError:
                  raise GeneralError(
                      f"Invalid debug level '{debug}', use an integer.")
              except KeyError:
                  pass
  
          # Get local option
          local = self._options.get(option, default)
          # Check parent option
          parent = None
          if self.parent:
              parent = self.parent.opt(option)
          # Special handling for special flags (parent's yes always wins)
          if option in ['quiet', 'force', 'dry']:
              return parent if parent else local
          # Special handling for counting options (child overrides the
          # parent if it was defined)
          elif option in ['debug', 'verbose']:
              winner = local if local else parent
              if winner is None:
                  winner = 0
              return winner
          else:
              return parent if parent is not None else local
  
      def _level(self) -> int:
          """ Hierarchy level """
          if self.parent is None:
              return -1
          else:
              return self.parent._level() + self._relative_indent
  
      def _indent(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0) -> str:
          """ Indent message according to the object hierarchy """
  
          return indent(
              key,
              value=value,
              color=color,
              level=self._level() + shift)
  
      def _log(self, message: str) -> None:
          """ Append provided message to the current log """
          # Nothing to do if there is no workdir
          if self.workdir is None:
              return
  
          # Store log only in the top parent
          if self.parent:
              self.parent._log(message)
          else:
              with open(os.path.join(self.workdir, LOG_FILENAME), 'a') as log:
                  log.write(datetime.datetime.utcnow().strftime('%H:%M:%S') + ' '
                            + remove_color(message) + '\n')
  
      def print(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          """ Print a message regardless the quiet mode """
          self._log(self._indent(key, value, color=None, shift=shift))
          echo(self._indent(key, value, color, shift), err=err)
  
      def info(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          """ Show a message unless in quiet mode """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet'):
              echo(self._indent(key, value, color, shift), err=err)
  
      def verbose(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          """
          Show message if in requested verbose mode level
  
          In quiet mode verbose messages are not displayed.
          """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet') and self.opt('verbose') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def debug(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          """
          Show message if in requested debug mode level
  
          In quiet mode debug messages are not displayed.
          """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet') and self.opt('debug') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def warn(self, message: str, shift: int = 0) -> None:
          """ Show a yellow warning message on info level, send to stderr """
          self.info('warn', message, color='yellow', shift=shift, err=True)
  
      def fail(self, message: str, shift: int = 0) -> None:
          """ Show a red failure message on info level, send to stderr """
          self.info('fail', message, color='red', shift=shift, err=True)
  
      def _run(self,
               command: Union[str, List[str]],
               cwd: Optional[str],
               shell: bool,
               env: Optional[EnvironmentType],
               log: Optional[BaseLoggerFnType],
               join: bool = False,
               interactive: bool = False,
               timeout: Optional[int] = None) -> CommandOutput:
          """
          Run command, capture the output
  
          By default stdout and stderr are captured separately.
          Use join=True to merge stderr into stdout.
          Use timeout=<seconds> to finish process after given time
          """
          # By default command ouput is logged using debug
          if not log:
              log = self.debug
          # Prepare the environment
          if env:
              if not isinstance(env, dict):
                  raise GeneralError(f"Invalid environment '{env}'.")
              # Do not modify current process environment
              environment = os.environ.copy()
              environment.update(env)
          else:
              environment = None
          self.debug('environment', pprint.pformat(environment), level=4)
  
          # Set only for shell=True as it would affect command
          executable = DEFAULT_SHELL if shell else None
  
          # Run the command in interactive mode if requested
          if interactive:
              try:
                  subprocess.run(
                      command, cwd=cwd, shell=shell, env=environment, check=True,
                      executable=executable)
              except subprocess.CalledProcessError:
                  # Interactive mode can return non-zero if the last command
                  # failed, ignore errors here
                  pass
              finally:
                  return CommandOutput(None, None)
  
          # Create the process
          try:
              process = subprocess.Popen(
                  command, cwd=cwd, shell=shell, env=environment,
                  stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
                  stderr=subprocess.STDOUT if join else subprocess.PIPE,
                  executable=executable)
          except FileNotFoundError as error:
              raise RunError(
                  f"File '{error.filename}' not found.", command, 127)
  
          stdout_thread = StreamLogger(
              process.stdout, log_header='out', logger=log)
          stderr_thread = stdout_thread
          if not join:
              stderr_thread = StreamLogger(
                  process.stderr, log_header='err', logger=log)
          stdout_thread.start()
          if not join:
              stderr_thread.start()
          try:
              process.wait(timeout=timeout)
          except subprocess.TimeoutExpired:
              process.kill()
              process.returncode = PROCESS_TIMEOUT
          stdout_thread.join()
          if not join:
              stderr_thread.join()
  
          # Handle the exit code, return output
          if process.returncode != 0:
              raise RunError(
                  message=f"Command returned '{process.returncode}'.",
                  command=command,
                  returncode=process.returncode,
  
  
  [... the rest is omitted ...]
files.old_path: tmt/utils.py