files: 
  """ Step Classes """
  
  import re
  import sys
  import textwrap
  import warnings
  from typing import (TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Type,
                      TypeVar, Union, cast, overload)
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  import click
  from click import echo
  
  import tmt.options
  import tmt.utils
  from tmt.options import show_step_method_hints
  
  if TYPE_CHECKING:
      import tmt.base
      from tmt.base import Plan
      from tmt.steps.provision import Guest
  
  
  DEFAULT_PLUGIN_METHOD_ORDER: int = 50
  
  
  # Supported steps and actions
  STEPS = ['discover', 'provision', 'prepare', 'execute', 'report', 'finish']
  ACTIONS = ['login', 'reboot']
  
  # Step phase order
  PHASE_START = 10
  PHASE_BASE = 50
  PHASE_END = 90
  
  
  class Phase(tmt.utils.Common):
      """ A phase of a step """
  
      def __init__(
              self,
              order: int = tmt.utils.DEFAULT_PLUGIN_ORDER,
              *args: Any,
              **kwargs: Any):
          super().__init__(*args, **kwargs)
          self.order: int = order
  
      def enabled_on_guest(self, guest: 'Guest') -> bool:
          """ Phases are enabled across all guests by default """
          return True
  
      @property
      def is_in_standalone_mode(self) -> bool:
          """
          True if the phase is in stand-alone mode.
  
          Stand-alone mode means that only this phase should be run as a part
          of the run (and not any other even if requested using --all).
          This is useful as some plugin options may completely change its
          behaviour from the regular behaviour based on options
          (e.g. listing images inside a provision plugin).
          """
          return False
  
  
  # A variable used to describe a generic type for all classes derived from Phase
  PhaseT = TypeVar('PhaseT', bound=Phase)
  
  
  class StepData(TypedDict, total=False):
      """
      Step data structure
      """
      name: str
      how: str
      order: Optional[int]
      tests: Optional[List['tmt.base.Test']]
  
  
  class Step(tmt.utils.Common):
      """ Common parent of all test steps """
  
      # Default implementation for all steps is "shell", but some
      # steps like provision may have better defaults for their
      # area of expertise.
      DEFAULT_HOW: str = 'shell'
  
      def __init__(
              self,
              plan: 'Plan',
              data: Optional[Union[StepData, List[StepData]]] = None,
              name: Optional[str] = None,
              workdir: tmt.utils.WorkdirArgumentType = None) -> None:
          """ Initialize and check the step data """
          super().__init__(name=name, parent=plan, workdir=workdir)
          # Initialize data
          self.plan: 'Plan' = plan
          self._status: Optional[str] = None
          self._phases: List[Phase] = []
  
          # Normalize `data` attribute to carry a list of step configuration data, one item per
          # distinct step configuration. Make sure all items have `name`` and `how` keys.
          self.data: List[StepData] = []
  
          # Create an empty step by default (can be updated from cli)
          if data is None:
              self.data = [{}]
  
          # Convert to list if only a single config provided
          elif isinstance(data, dict):
              self.data = [data]
  
          # List is as good as it gets
          elif isinstance(data, list):
              self.data = data
  
          # Shout about invalid configuration
          else:
              raise tmt.utils.GeneralError(
                  f"Invalid '{self}' config in '{self.plan}'.")
  
          # Final sanity checks
          for i, single_config_data in enumerate(self.data):
              # Add default unique names even to multiple configs so that the users
              # don't need to specify it if they don't care about the name
              if 'name' not in single_config_data:
                  single_config_data['name'] = f'{tmt.utils.DEFAULT_NAME}-{i}'
  
              # Set 'how' to the default if not specified
              if single_config_data.get('how') is None:
                  single_config_data['how'] = self.DEFAULT_HOW
  
      @property
      def enabled(self) -> Optional[bool]:
          """ True if the step is enabled """
          try:
              return self.name in self.plan.my_run._context.obj.steps
          except AttributeError:
              return None
  
      @property
      def plugins_in_standalone_mode(self) -> int:
          """
          The number of plugins in standalone mode.
  
          Stand-alone mode means that only this step should be run as a part
          of the run (and not any other even if requested using --all).
          This is useful as some step options may completely change its
          behaviour from the regular behaviour based on options
          (e.g. listing images inside provision).
          """
          return sum(phase.is_in_standalone_mode for phase in
  
  [... the rest is omitted ...]
files: tmt/steps/__init__.py
files: 
  """ Step Classes """
  
  import re
  import sys
  import textwrap
  import warnings
  from typing import (TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Type,
                      TypeVar, Union, cast, overload)
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  import click
  from click import echo
  
  import tmt.options
  import tmt.utils
  from tmt.options import show_step_method_hints
  
  if TYPE_CHECKING:
      import tmt.base
      from tmt.base import Plan
      from tmt.steps.provision import Guest
  
  
  DEFAULT_PLUGIN_METHOD_ORDER: int = 50
  
  
  # Supported steps and actions
  STEPS = ['discover', 'provision', 'prepare', 'execute', 'report', 'finish']
  ACTIONS = ['login', 'reboot']
  
  # Step phase order
  PHASE_START = 10
  PHASE_BASE = 50
  PHASE_END = 90
  
  
  class Phase(tmt.utils.Common):
      """ A phase of a step """
  
      def __init__(
              self,
              order: int = tmt.utils.DEFAULT_PLUGIN_ORDER,
              *args: Any,
              **kwargs: Any):
          super().__init__(*args, **kwargs)
          self.order: int = order
  
      def enabled_on_guest(self, guest: 'Guest') -> bool:
          """ Phases are enabled across all guests by default """
          return True
  
      @property
      def is_in_standalone_mode(self) -> bool:
          """
          True if the phase is in stand-alone mode.
  
          Stand-alone mode means that only this phase should be run as a part
          of the run (and not any other even if requested using --all).
          This is useful as some plugin options may completely change its
          behaviour from the regular behaviour based on options
          (e.g. listing images inside a provision plugin).
          """
          return False
  
  
  # A variable used to describe a generic type for all classes derived from Phase
  PhaseT = TypeVar('PhaseT', bound=Phase)
  
  
  class StepData(TypedDict, total=False):
      """
      Step data structure
      """
      name: str
      how: str
      order: Optional[int]
      tests: Optional[List['tmt.base.Test']]
  
  
  class Step(tmt.utils.Common):
      """ Common parent of all test steps """
  
      # Default implementation for all steps is "shell", but some
      # steps like provision may have better defaults for their
      # area of expertise.
      DEFAULT_HOW: str = 'shell'
  
      def __init__(
              self,
              plan: 'Plan',
              data: Optional[Union[StepData, List[StepData]]] = None,
              name: Optional[str] = None,
              workdir: tmt.utils.WorkdirArgumentType = None) -> None:
          """ Initialize and check the step data """
          super().__init__(name=name, parent=plan, workdir=workdir)
          # Initialize data
          self.plan: 'Plan' = plan
          self._status: Optional[str] = None
          self._phases: List[Phase] = []
  
          # Normalize `data` attribute to carry a list of step configuration data, one item per
          # distinct step configuration. Make sure all items have `name`` and `how` keys.
          self.data: List[StepData] = []
  
          # Create an empty step by default (can be updated from cli)
          if data is None:
              self.data = [{}]
  
          # Convert to list if only a single config provided
          elif isinstance(data, dict):
              self.data = [data]
  
          # List is as good as it gets
          elif isinstance(data, list):
              self.data = data
  
          # Shout about invalid configuration
          else:
              raise tmt.utils.GeneralError(
                  f"Invalid '{self}' config in '{self.plan}'.")
  
          # Final sanity checks
          for i, single_config_data in enumerate(self.data):
              # Add default unique names even to multiple configs so that the users
              # don't need to specify it if they don't care about the name
              if 'name' not in single_config_data:
                  single_config_data['name'] = f'{tmt.utils.DEFAULT_NAME}-{i}'
  
              # Set 'how' to the default if not specified
              if single_config_data.get('how') is None:
                  single_config_data['how'] = self.DEFAULT_HOW
  
      @property
      def enabled(self) -> Optional[bool]:
          """ True if the step is enabled """
          try:
              return self.name in self.plan.my_run._context.obj.steps
          except AttributeError:
              return None
  
      @property
      def plugins_in_standalone_mode(self) -> int:
          """
          The number of plugins in standalone mode.
  
          Stand-alone mode means that only this step should be run as a part
          of the run (and not any other even if requested using --all).
          This is useful as some step options may completely change its
          behaviour from the regular behaviour based on options
          (e.g. listing images inside provision).
          """
          return sum(phase.is_in_standalone_mode for phase in
  
  [... the rest is omitted ...]
files: tmt/steps/__init__.py
files: import glob
  import os
  import re
  import shutil
  import subprocess
  
  import click
  import fmf
  
  import tmt
  import tmt.beakerlib
  import tmt.steps
  import tmt.steps.discover
  from tmt.utils import git_clone
  
  
  @tmt.steps.provides_method('fmf')
  class DiscoverFmf(tmt.steps.discover.DiscoverPlugin):
      """
      Discover available tests from fmf metadata
  
      By default all available tests from the current repository are used
      so the minimal configuration looks like this:
  
          discover:
              how: fmf
  
      Full config example:
  
          discover:
              how: fmf
              url: https://github.com/teemtee/tmt
              ref: main
              path: /fmf/root
              test: /tests/basic
              filter: 'tier: 1'
  
      If no 'ref' is provided, the default branch from the origin is used.
  
      For DistGit repo one can extract source tarball first and discover
      tests from it by using 'distgit-source: true'. It can be used
      together with 'ref', 'path' and 'url', However 'ref' is not possible
      without using 'url'.
  
          discover:
              how: fmf
              dist-git-source: true
  
      Related config options (all optional):
      * dist-git-merge - set to True if you want to copy in extracted
        sources to the local repo
      * dist-git-init - set to True and 'fmf init' will be called inside
        extracted sources (at dist-git-extract or top directory)
      * dist-git-extract - directory (glob supported) to copy from
        extracted sources (defaults to inner fmf-root)
      * dist-git-remove-fmf-root - set to True to remove fmf root from
        extracted sources
  
      Selecting tests containing specified link is possible using 'link'
      option accepting RELATION:TARGET format of values. Regular
      expressions are supported for both relation and target part of the
      value. Relation can be omitted to target match any relation.
  
          discover:
              how: fmf
              link: verifies:.*issue/850$
  
      It is also possible to limit tests only to those that have changed
      in git since a given revision. This can be particularly useful when
      testing changes to tests themselves (e.g. in a pull request CI).
  
      Related config options (all optional):
      * modified-only - set to True if you want to filter modified tests
      * modified-url - fetched as "reference" remote in the test dir
      * modified-ref - the ref to compare against
  
      Example to compare local repo against upstream 'main' branch:
  
          discover:
              how: fmf
              modified-only: True
              modified-url: https://github.com/teemtee/tmt
              modified-ref: reference/main
  
      Note that internally the modified tests are appended to the list
      specified via 'test', so those tests will also be selected even if
      not modified.
      """
  
      # Supported keys
      _keys = [
          "url", "ref", "path", "test", "link", "filter",
          "modified-only", "modified-url", "modified-ref",
          "dist-git-init", "dist-git-remove-fmf-root", "dist-git-merge",
          "dist-git-extract",
          "fmf-id", "exclude"]
  
      REF_OPTION = click.option(
          '-r', '--ref', metavar='REVISION',
          help='Branch, tag or commit specifying the git revision.')
      TEST_OPTION = click.option(
          '-t', '--test', metavar='NAMES', multiple=True,
          help='Select tests by name.')
      FILTER_OPTION = click.option(
          '-F', '--filter', metavar='FILTERS', multiple=True,
          help='Include only tests matching the filter.')
      EXCLUDE_OPTION = click.option(
          '-x', '--exclude', metavar='[REGEXP]', multiple=True,
          help="Exclude a regular expression from search result.")
  
      # Options which require .git to be present for their functionality
      _REQUIRES_GIT = (
          "ref",
          "modified-url",
          "modified-only",
          "fmf-id",
          )
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          return [
              click.option(
                  '-u', '--url', metavar='REPOSITORY',
                  help='URL of the git repository with fmf metadata.'),
              cls.REF_OPTION,
              click.option(
                  '--modified-url', metavar='REPOSITORY',
                  help='URL of the reference git repository with fmf metadata.'),
              click.option(
                  '--modified-ref', metavar='REVISION',
                  help='Branch, tag or commit specifying the reference git '
                  'revision (if not provided, the default branch is used).'),
              click.option(
                  '-m', '--modified-only', is_flag=True,
                  help='If set, select only tests modified '
                  'since reference revision.'),
              click.option(
                  '-p', '--
  
  [... the rest is omitted ...]
files: tmt/steps/discover/fmf.py
files: import glob
  import os
  import re
  import shutil
  import subprocess
  
  import click
  import fmf
  
  import tmt
  import tmt.beakerlib
  import tmt.steps
  import tmt.steps.discover
  from tmt.utils import git_clone
  
  
  @tmt.steps.provides_method('fmf')
  class DiscoverFmf(tmt.steps.discover.DiscoverPlugin):
      """
      Discover available tests from fmf metadata
  
      By default all available tests from the current repository are used
      so the minimal configuration looks like this:
  
          discover:
              how: fmf
  
      Full config example:
  
          discover:
              how: fmf
              url: https://github.com/teemtee/tmt
              ref: main
              path: /fmf/root
              test: /tests/basic
              filter: 'tier: 1'
  
      If no 'ref' is provided, the default branch from the origin is used.
  
      For DistGit repo one can extract source tarball first and discover
      tests from it by using 'distgit-source: true'. It can be used
      together with 'ref', 'path' and 'url', However 'ref' is not possible
      without using 'url'.
  
          discover:
              how: fmf
              dist-git-source: true
  
      Related config options (all optional):
      * dist-git-merge - set to True if you want to copy in extracted
        sources to the local repo
      * dist-git-init - set to True and 'fmf init' will be called inside
        extracted sources (at dist-git-extract or top directory)
      * dist-git-extract - directory (glob supported) to copy from
        extracted sources (defaults to inner fmf-root)
      * dist-git-remove-fmf-root - set to True to remove fmf root from
        extracted sources
  
      Selecting tests containing specified link is possible using 'link'
      option accepting RELATION:TARGET format of values. Regular
      expressions are supported for both relation and target part of the
      value. Relation can be omitted to target match any relation.
  
          discover:
              how: fmf
              link: verifies:.*issue/850$
  
      It is also possible to limit tests only to those that have changed
      in git since a given revision. This can be particularly useful when
      testing changes to tests themselves (e.g. in a pull request CI).
  
      Related config options (all optional):
      * modified-only - set to True if you want to filter modified tests
      * modified-url - fetched as "reference" remote in the test dir
      * modified-ref - the ref to compare against
  
      Example to compare local repo against upstream 'main' branch:
  
          discover:
              how: fmf
              modified-only: True
              modified-url: https://github.com/teemtee/tmt
              modified-ref: reference/main
  
      Note that internally the modified tests are appended to the list
      specified via 'test', so those tests will also be selected even if
      not modified.
      """
  
      # Supported keys
      _keys = [
          "url", "ref", "path", "test", "link", "filter",
          "modified-only", "modified-url", "modified-ref",
          "dist-git-init", "dist-git-remove-fmf-root", "dist-git-merge",
          "dist-git-extract",
          "fmf-id", "exclude"]
  
      REF_OPTION = click.option(
          '-r', '--ref', metavar='REVISION',
          help='Branch, tag or commit specifying the git revision.')
      TEST_OPTION = click.option(
          '-t', '--test', metavar='NAMES', multiple=True,
          help='Select tests by name.')
      FILTER_OPTION = click.option(
          '-F', '--filter', metavar='FILTERS', multiple=True,
          help='Include only tests matching the filter.')
      EXCLUDE_OPTION = click.option(
          '-x', '--exclude', metavar='[REGEXP]', multiple=True,
          help="Exclude a regular expression from search result.")
  
      # Options which require .git to be present for their functionality
      _REQUIRES_GIT = (
          "ref",
          "modified-url",
          "modified-only",
          "fmf-id",
          )
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          return [
              click.option(
                  '-u', '--url', metavar='REPOSITORY',
                  help='URL of the git repository with fmf metadata.'),
              cls.REF_OPTION,
              click.option(
                  '--modified-url', metavar='REPOSITORY',
                  help='URL of the reference git repository with fmf metadata.'),
              click.option(
                  '--modified-ref', metavar='REVISION',
                  help='Branch, tag or commit specifying the reference git '
                  'revision (if not provided, the default branch is used).'),
              click.option(
                  '-m', '--modified-only', is_flag=True,
                  help='If set, select only tests modified '
                  'since reference revision.'),
              click.option(
                  '-p', '--
  
  [... the rest is omitted ...]
files: tmt/steps/discover/fmf.py
files: import copy
  import os
  from typing import TYPE_CHECKING, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.discover
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  @tmt.steps.provides_method('shell')
  class DiscoverShell(tmt.steps.discover.DiscoverPlugin):
      """
      Use provided list of shell script tests
  
      List of test cases to be executed can be defined manually directly
      in the plan as a list of dictionaries containing test name, actual
      test script and optionally a path to the test. Example config:
  
      discover:
          how: shell
          tests:
          - name: /help/main
            test: tmt --help
          - name: /help/test
            test: tmt test --help
          - name: /help/smoke
            test: ./smoke.sh
            path: /tests/shell
  
      For DistGit repo one can extract source tarball and use its code.
      It is extracted to TMT_SOURCE_DIR however no patches are applied
      (only source tarball is extracted).
  
      discover:
          how: shell
          dist-git-source: true
          tests:
          - name: /upstream
            test: cd $TMT_SOURCE_DIR/*/tests && make test
      """
  
      def show(self, keys: Optional[List[str]] = None) -> None:
          """ Show config details """
          super().show([])
          tests = self.get('tests')
          if tests:
              test_names = [test['name'] for test in tests]
              click.echo(tmt.utils.format('tests', test_names))
  
      def wake(self) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake()
          # Check provided tests, default to an empty list
          if 'tests' not in self.data:
              self.data['tests'] = []
          self._tests: List[tmt.base.Test] = []
  
      def go(self) -> None:
          """ Discover available tests """
          super(DiscoverShell, self).go()
          tests = fmf.Tree(dict(summary='tests'))
  
          # dist-git related
          assert self.workdir is not None
          sourcedir = os.path.join(self.workdir, 'source')
          dist_git_source = self.get('dist-git-source', False)
  
          # Check and process each defined shell test
          assert self.data['tests'] is not None
          for data in self.data['tests']:
              # Create data copy (we want to keep original data for save()
              data = copy.deepcopy(data)
              # Extract name, make sure it is present
              try:
                  name = data.pop('name')
              except KeyError:
                  raise tmt.utils.SpecificationError(
                      f"Missing test name in '{self.step.plan.name}'.")
              # Make sure that the test script is defined
              if 'test' not in data:
                  raise tmt.utils.SpecificationError(
                      f"Missing test script in '{self.step.plan.name}'.")
              # Prepare path to the test working directory (tree root by default)
              try:
                  data['path'] = f"/tests{data['path']}"
              except KeyError:
                  data['path'] = "/tests"
              # Apply default test duration unless provided
              if 'duration' not in data:
                  data['duration'] = tmt.base.DEFAULT_TEST_DURATION_L2
              # Add source dir path variable
              if dist_git_source:
                  data.setdefault('environment', {})[
                      'TMT_SOURCE_DIR'] = sourcedir
              # Create a simple fmf node, adjust its name
              tests.child(name, data)
  
          # Symlink tests directory to the plan work tree
          testdir = os.path.join(self.workdir, "tests")
          relative_path = os.path.relpath(self.step.plan.worktree, self.workdir)
          os.symlink(relative_path, testdir)
  
          if dist_git_source:
              try:
                  run_result = self.run(
                      ["git", "rev-parse", "--show-toplevel"],
                      cwd=self.step.plan.my_run.tree.root,
                      dry=True)[0]
                  assert run_result is not None
                  git_root = run_result.strip('\n')
              except tmt.utils.RunError:
                  raise tmt.utils.DiscoverError(
                      f"Directory '{self.step.plan.my_run.tree.root}' "
                      f"is not a git repository.")
              try:
                  self.extract_distgit_source(
                      git_root, sourcedir, self.get('dist-git-type'))
              except Exception as error:
                  raise tmt.utils.DiscoverError(
                      "Failed to process 'dist-git-source'.", original=error)
  
          # Use a tmt.Tree to apply possible command line filters
          tests = tmt.Tree(tree=tests).tests(conditions=["manual is False"])
          self._tests = tests
  
      def tests(self) -> List[tmt.base.Test]:
          return self._tests
  
  
  [... the rest is omitted ...]
files: tmt/steps/discover/shell.py
files: import copy
  import os
  from typing import TYPE_CHECKING, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.discover
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  @tmt.steps.provides_method('shell')
  class DiscoverShell(tmt.steps.discover.DiscoverPlugin):
      """
      Use provided list of shell script tests
  
      List of test cases to be executed can be defined manually directly
      in the plan as a list of dictionaries containing test name, actual
      test script and optionally a path to the test. Example config:
  
      discover:
          how: shell
          tests:
          - name: /help/main
            test: tmt --help
          - name: /help/test
            test: tmt test --help
          - name: /help/smoke
            test: ./smoke.sh
            path: /tests/shell
  
      For DistGit repo one can extract source tarball and use its code.
      It is extracted to TMT_SOURCE_DIR however no patches are applied
      (only source tarball is extracted).
  
      discover:
          how: shell
          dist-git-source: true
          tests:
          - name: /upstream
            test: cd $TMT_SOURCE_DIR/*/tests && make test
      """
  
      def show(self, keys: Optional[List[str]] = None) -> None:
          """ Show config details """
          super().show([])
          tests = self.get('tests')
          if tests:
              test_names = [test['name'] for test in tests]
              click.echo(tmt.utils.format('tests', test_names))
  
      def wake(self, keys: Optional[List[str]] = None) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys)
          # Check provided tests, default to an empty list
          if 'tests' not in self.data:
              self.data['tests'] = []
          self._tests: List[tmt.base.Test] = []
  
      def go(self) -> None:
          """ Discover available tests """
          super(DiscoverShell, self).go()
          tests = fmf.Tree(dict(summary='tests'))
  
          # dist-git related
          assert self.workdir is not None
          sourcedir = os.path.join(self.workdir, 'source')
          dist_git_source = self.get('dist-git-source', False)
  
          # Check and process each defined shell test
          assert self.data['tests'] is not None
          for data in self.data['tests']:
              # Create data copy (we want to keep original data for save()
              data = copy.deepcopy(data)
              # Extract name, make sure it is present
              try:
                  name = data.pop('name')
              except KeyError:
                  raise tmt.utils.SpecificationError(
                      f"Missing test name in '{self.step.plan.name}'.")
              # Make sure that the test script is defined
              if 'test' not in data:
                  raise tmt.utils.SpecificationError(
                      f"Missing test script in '{self.step.plan.name}'.")
              # Prepare path to the test working directory (tree root by default)
              try:
                  data['path'] = f"/tests{data['path']}"
              except KeyError:
                  data['path'] = "/tests"
              # Apply default test duration unless provided
              if 'duration' not in data:
                  data['duration'] = tmt.base.DEFAULT_TEST_DURATION_L2
              # Add source dir path variable
              if dist_git_source:
                  data.setdefault('environment', {})[
                      'TMT_SOURCE_DIR'] = sourcedir
              # Create a simple fmf node, adjust its name
              tests.child(name, data)
  
          # Symlink tests directory to the plan work tree
          testdir = os.path.join(self.workdir, "tests")
          relative_path = os.path.relpath(self.step.plan.worktree, self.workdir)
          os.symlink(relative_path, testdir)
  
          if dist_git_source:
              try:
                  run_result = self.run(
                      ["git", "rev-parse", "--show-toplevel"],
                      cwd=self.step.plan.my_run.tree.root,
                      dry=True)[0]
                  assert run_result is not None
                  git_root = run_result.strip('\n')
              except tmt.utils.RunError:
                  raise tmt.utils.DiscoverError(
                      f"Directory '{self.step.plan.my_run.tree.root}' "
                      f"is not a git repository.")
              try:
                  self.extract_distgit_source(
                      git_root, sourcedir, self.get('dist-git-type'))
              except Exception as error:
                  raise tmt.utils.DiscoverError(
                      "Failed to process 'dist-git-source'.", original=error)
  
          # Use a tmt.Tree to apply possible command line filters
          tests = tmt.Tree(tree=tests).tests(conditions=["manual is False"])
          self._tests = tests
  
      def tests(self) -> List[tmt.base.Test]:
          return self._tests
  
  
  [... the rest is omitted ...]
files: tmt/steps/discover/shell.py
files: import json
  import os
  import sys
  import time
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.execute
  import tmt.utils
  from tmt.steps.execute import TEST_OUTPUT_FILENAME, Script
  
  # Script handling reboots, in restraint compatible fashion
  TMT_REBOOT_SCRIPT = Script("/usr/local/bin/tmt-reboot",
                             aliases=[
                                 "/usr/local/bin/rstrnt-reboot",
                                 "/usr/local/bin/rhts-reboot"],
                             related_variables=[
                                 "TMT_REBOOT_COUNT",
                                 "REBOOTCOUNT",
                                 "RSTRNT_REBOOTCOUNT"]
                             )
  
  # Script handling result reporting, in restraint compatible fashion
  TMT_REPORT_RESULT_SCRIPT = Script("/usr/local/bin/tmt-report-result",
                                    aliases=[
                                        "/usr/local/bin/rstrnt-report-result",
                                        "/usr/local/bin/rhts-report-result"],
                                    related_variables=[]
                                    )
  
  # Script for archiving a file, usable for BEAKERLIB_COMMAND_SUBMIT_LOG
  TMT_FILE_SUBMIT_SCRIPT = Script("/usr/local/bin/tmt-file-submit",
                                  aliases=[
                                      "/usr/local/bin/rstrnt-report-log",
                                      "/usr/local/bin/rhts-submit-log",
                                      "/usr/local/bin/rhts_submit_log"],
                                  related_variables=[]
                                  )
  
  # Script handling text execution abortion, in restraint compatible fashion
  TMT_ABORT_SCRIPT = Script("/usr/local/bin/tmt-abort",
                            aliases=[
                                "/usr/local/bin/rstrnt-abort",
                                "/usr/local/bin/rhts-abort"],
                            related_variables=[]
                            )
  
  # File for requesting reboot
  REBOOT_REQUEST_FILENAME = "reboot_request"
  
  # List of all available scripts
  SCRIPTS = (TMT_FILE_SUBMIT_SCRIPT,
             TMT_REBOOT_SCRIPT,
             TMT_REPORT_RESULT_SCRIPT,
             TMT_ABORT_SCRIPT
             )
  
  
  @tmt.steps.provides_method('tmt')
  @tmt.steps.provides_method('shell.tmt', order=80)
  @tmt.steps.provides_method('beakerlib.tmt', order=80)
  class ExecuteInternal(tmt.steps.execute.ExecutePlugin):
      """
      Use the internal tmt executor to execute tests
  
      The internal tmt executor runs tests on the guest one by one, shows
      testing progress and supports interactive debugging as well. Test
      result is based on the script exit code (for shell tests) or the
      results file (for beakerlib tests).
      """
  
      # Supported keys
      _keys = ["script", "interactive"]
  
      def __init__(self, *args, **kwargs):
          super().__init__(*args, **kwargs)
          self._previous_progress_message = ""
          self.scripts = SCRIPTS
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          options = []
          # Shell script as a test
          options.append(click.option(
              '-s', '--script', metavar='SCRIPT', multiple=True,
              help='Shell script to be executed as a test.'))
          # Interactive mode
          options.append(click.option(
              '-i', '--interactive', is_flag=True,
              help='Run in interactive mode, do not capture output.'))
          # Disable interactive progress bar
          options.append(click.option(
              '--no-progress-bar', is_flag=True,
              help='Disable interactive progress bar showing the current test.'))
          return options + super().options(how)
  
      def wake(self):
          """ Wake up the plugin, process data, apply options """
          super().wake()
          # Make sure that script is a list
          tmt.utils.listify(self.data, keys=['script'])
  
      # TODO: consider switching to utils.updatable_message() - might need more
      # work, since use of _show_progress is split over several methods.
      def _show_progress(self, progress, test_name, finish=False):
          """
          Show an interactive progress bar in non-verbose mode.
  
          If the output is not an interactive terminal, or progress bar is
          disabled using an option, just output the message as info without
          utilising \r. If finish is True, overwrite the previous progress bar.
          """
          # Verbose mode outputs other information, using \r to
          # create a status bar wouldn't work.
          if self.opt('verbose'):
              return
  
          # No progress if terminal not attached or explicitly disabled
          if not sys.stdout.isatty() or self.opt('no-progress-bar'):
              return
  
          # For debug mode show just an info message (unless finishing)
          message = f"{test_name} [{progress}]" if not finish else ""
          if self.opt('debug'):
              if not finish:
                  self.info(message, shift=1)
              return
  
          # Show progress bar in an interactive shell.
          # We need to completely override the previous message, add
          # spaces if necessary.
          message = message.ljust(len(self._previous_progress_message))
          self._previous_progress_message = message
          message = self._indent('progress', message, color='cyan')
          sys.stdout
  
  [... the rest is omitted ...]
files: tmt/steps/execute/internal.py
files: import json
  import os
  import sys
  import time
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.execute
  import tmt.utils
  from tmt.steps.execute import TEST_OUTPUT_FILENAME, Script
  
  # Script handling reboots, in restraint compatible fashion
  TMT_REBOOT_SCRIPT = Script("/usr/local/bin/tmt-reboot",
                             aliases=[
                                 "/usr/local/bin/rstrnt-reboot",
                                 "/usr/local/bin/rhts-reboot"],
                             related_variables=[
                                 "TMT_REBOOT_COUNT",
                                 "REBOOTCOUNT",
                                 "RSTRNT_REBOOTCOUNT"]
                             )
  
  # Script handling result reporting, in restraint compatible fashion
  TMT_REPORT_RESULT_SCRIPT = Script("/usr/local/bin/tmt-report-result",
                                    aliases=[
                                        "/usr/local/bin/rstrnt-report-result",
                                        "/usr/local/bin/rhts-report-result"],
                                    related_variables=[]
                                    )
  
  # Script for archiving a file, usable for BEAKERLIB_COMMAND_SUBMIT_LOG
  TMT_FILE_SUBMIT_SCRIPT = Script("/usr/local/bin/tmt-file-submit",
                                  aliases=[
                                      "/usr/local/bin/rstrnt-report-log",
                                      "/usr/local/bin/rhts-submit-log",
                                      "/usr/local/bin/rhts_submit_log"],
                                  related_variables=[]
                                  )
  
  # Script handling text execution abortion, in restraint compatible fashion
  TMT_ABORT_SCRIPT = Script("/usr/local/bin/tmt-abort",
                            aliases=[
                                "/usr/local/bin/rstrnt-abort",
                                "/usr/local/bin/rhts-abort"],
                            related_variables=[]
                            )
  
  # File for requesting reboot
  REBOOT_REQUEST_FILENAME = "reboot_request"
  
  # List of all available scripts
  SCRIPTS = (TMT_FILE_SUBMIT_SCRIPT,
             TMT_REBOOT_SCRIPT,
             TMT_REPORT_RESULT_SCRIPT,
             TMT_ABORT_SCRIPT
             )
  
  
  @tmt.steps.provides_method('tmt')
  @tmt.steps.provides_method('shell.tmt', order=80)
  @tmt.steps.provides_method('beakerlib.tmt', order=80)
  class ExecuteInternal(tmt.steps.execute.ExecutePlugin):
      """
      Use the internal tmt executor to execute tests
  
      The internal tmt executor runs tests on the guest one by one, shows
      testing progress and supports interactive debugging as well. Test
      result is based on the script exit code (for shell tests) or the
      results file (for beakerlib tests).
      """
  
      # Supported keys
      _keys = ["script", "interactive"]
  
      def __init__(self, *args, **kwargs):
          super().__init__(*args, **kwargs)
          self._previous_progress_message = ""
          self.scripts = SCRIPTS
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for given method """
          options = []
          # Shell script as a test
          options.append(click.option(
              '-s', '--script', metavar='SCRIPT', multiple=True,
              help='Shell script to be executed as a test.'))
          # Interactive mode
          options.append(click.option(
              '-i', '--interactive', is_flag=True,
              help='Run in interactive mode, do not capture output.'))
          # Disable interactive progress bar
          options.append(click.option(
              '--no-progress-bar', is_flag=True,
              help='Disable interactive progress bar showing the current test.'))
          return options + super().options(how)
  
      def wake(self, keys=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys)
          # Make sure that script is a list
          tmt.utils.listify(self.data, keys=['script'])
  
      # TODO: consider switching to utils.updatable_message() - might need more
      # work, since use of _show_progress is split over several methods.
      def _show_progress(self, progress, test_name, finish=False):
          """
          Show an interactive progress bar in non-verbose mode.
  
          If the output is not an interactive terminal, or progress bar is
          disabled using an option, just output the message as info without
          utilising \r. If finish is True, overwrite the previous progress bar.
          """
          # Verbose mode outputs other information, using \r to
          # create a status bar wouldn't work.
          if self.opt('verbose'):
              return
  
          # No progress if terminal not attached or explicitly disabled
          if not sys.stdout.isatty() or self.opt('no-progress-bar'):
              return
  
          # For debug mode show just an info message (unless finishing)
          message = f"{test_name} [{progress}]" if not finish else ""
          if self.opt('debug'):
              if not finish:
                  self.info(message, shift=1)
              return
  
          # Show progress bar in an interactive shell.
          # We need to completely override the previous message, add
          # spaces if necessary.
          message = message.ljust(len(self._previous_progress_message))
          self._previous_progress_message = message
          message = self._indent('progress', message,
  
  [... the rest is omitted ...]
files: tmt/steps/execute/internal.py
files: from typing import Any, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.finish
  from tmt.steps.provision import Guest
  
  
  @tmt.steps.provides_method('shell')
  class FinishShell(tmt.steps.finish.FinishPlugin):
      """
      Perform finishing tasks using shell scripts
  
      Example config:
  
      finish:
          how: shell
          script:
              - upload-logs.sh || true
              - rm -rf /tmp/temporary-files
  
      Use the 'order' attribute to select in which order finishing tasks
      should happen if there are multiple configs. Default order is '50'.
      """
  
      # Supported keys
      _keys = ["script"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Finish command line options """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '-s', '--script', metavar='SCRIPT',
                  help='Shell script to be executed.')
              ]
          return options
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'script':
              return []
          return default
  
      # TODO: use better types once superclass gains its annotations
      def wake(self) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake()
  
          # Convert to list if single script provided
          tmt.utils.listify(self.data, keys=['script'])
  
      def go(self, guest: Guest) -> None:
          """ Perform finishing tasks on given guest """
          super().go(guest)
  
          # Give a short summary
          scripts = self.get('script')
          overview = fmf.utils.listed(scripts, 'script')
          self.info('overview', f'{overview} found', 'green')
  
          # Execute each script on the guest
          for script in scripts:
              self.verbose('script', script, 'green')
              guest.execute(script, cwd=self.step.plan.worktree)
  
  
  [... the rest is omitted ...]
files: tmt/steps/finish/shell.py
files: from typing import Any, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.finish
  from tmt.steps.provision import Guest
  
  
  @tmt.steps.provides_method('shell')
  class FinishShell(tmt.steps.finish.FinishPlugin):
      """
      Perform finishing tasks using shell scripts
  
      Example config:
  
      finish:
          how: shell
          script:
              - upload-logs.sh || true
              - rm -rf /tmp/temporary-files
  
      Use the 'order' attribute to select in which order finishing tasks
      should happen if there are multiple configs. Default order is '50'.
      """
  
      # Supported keys
      _keys = ["script"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Finish command line options """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '-s', '--script', metavar='SCRIPT',
                  help='Shell script to be executed.')
              ]
          return options
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'script':
              return []
          return default
  
      # TODO: use better types once superclass gains its annotations
      def wake(self, keys: Optional[List[str]] = None) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys)
  
          # Convert to list if single script provided
          tmt.utils.listify(self.data, keys=['script'])
  
      def go(self, guest: Guest) -> None:
          """ Perform finishing tasks on given guest """
          super().go(guest)
  
          # Give a short summary
          scripts = self.get('script')
          overview = fmf.utils.listed(scripts, 'script')
          self.info('overview', f'{overview} found', 'green')
  
          # Execute each script on the guest
          for script in scripts:
              self.verbose('script', script, 'green')
              guest.execute(script, cwd=self.step.plan.worktree)
  
  
  [... the rest is omitted ...]
files: tmt/steps/finish/shell.py
files: import sys
  from typing import Any, List, Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  import tmt.utils
  from tmt.steps import Step
  from tmt.steps.provision import Guest
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  StepDataType = TypedDict(
      'StepDataType',
      {
          'playbook': List[str],
          'playbooks': List[str]
          }
      )
  
  
  @tmt.steps.provides_method('ansible')
  class PrepareAnsible(tmt.steps.prepare.PreparePlugin):  # type: ignore[misc]
      """
      Prepare guest using ansible
  
      Single playbook config:
  
          prepare:
              how: ansible
              playbook: ansible/packages.yml
  
      Multiple playbooks config:
  
          prepare:
              how: ansible
              playbook:
                - playbook/one.yml
                - playbook/two.yml
                - playbook/three.yml
              extra-args: '-vvv'
  
      The playbook path should be relative to the metadata tree root.
      Use 'order' attribute to select in which order preparation should
      happen if there are multiple configs. Default order is '50'.
      Default order of required packages installation is '70'.
      """
  
      # Supported keys
      _keys = ["playbook", "extra-args"]
  
      def __init__(self, step: Step, data: StepDataType) -> None:
          """ Store plugin name, data and parent step """
          super().__init__(step, data)
          # Rename plural playbooks to singular
          if 'playbooks' in self.data:
              self.data['playbook'] = self.data.pop('playbooks')
  
      # TODO: fix types once superclass gains its annotations
      @classmethod
      def options(cls, how: Optional[str] = None) -> Any:
          """ Prepare command line options """
          return [
              click.option(
                  '-p', '--playbook', metavar='PLAYBOOK', multiple=True,
                  help='Path to an ansible playbook to run.'),
              click.option(
                  '--extra-args', metavar='EXTRA-ARGS',
                  help='Optional arguments for ansible-playbook.')
              ] + super().options(how)
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'playbook':
              return []
          return default
  
      # TODO: use better types once superclass gains its annotations
      def wake(self) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake()
  
          # Convert to list if necessary
          tmt.utils.listify(self.data, keys=['playbook'])
  
      def go(self, guest: Guest) -> None:
          """ Prepare the guests """
          super().go(guest)
  
          # Apply each playbook on the guest
          for playbook in self.get('playbook'):
              self.info('playbook', playbook, 'green')
              guest.ansible(playbook, self.get('extra-args'))
  
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/ansible.py
files: import sys
  from typing import Any, List, Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  import tmt.utils
  from tmt.steps import Step
  from tmt.steps.provision import Guest
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  StepDataType = TypedDict(
      'StepDataType',
      {
          'playbook': List[str],
          'playbooks': List[str]
          }
      )
  
  
  @tmt.steps.provides_method('ansible')
  class PrepareAnsible(tmt.steps.prepare.PreparePlugin):  # type: ignore[misc]
      """
      Prepare guest using ansible
  
      Single playbook config:
  
          prepare:
              how: ansible
              playbook: ansible/packages.yml
  
      Multiple playbooks config:
  
          prepare:
              how: ansible
              playbook:
                - playbook/one.yml
                - playbook/two.yml
                - playbook/three.yml
              extra-args: '-vvv'
  
      The playbook path should be relative to the metadata tree root.
      Use 'order' attribute to select in which order preparation should
      happen if there are multiple configs. Default order is '50'.
      Default order of required packages installation is '70'.
      """
  
      # Supported keys
      _keys = ["playbook", "extra-args"]
  
      def __init__(self, step: Step, data: StepDataType) -> None:
          """ Store plugin name, data and parent step """
          super().__init__(step, data)
          # Rename plural playbooks to singular
          if 'playbooks' in self.data:
              self.data['playbook'] = self.data.pop('playbooks')
  
      # TODO: fix types once superclass gains its annotations
      @classmethod
      def options(cls, how: Optional[str] = None) -> Any:
          """ Prepare command line options """
          return [
              click.option(
                  '-p', '--playbook', metavar='PLAYBOOK', multiple=True,
                  help='Path to an ansible playbook to run.'),
              click.option(
                  '--extra-args', metavar='EXTRA-ARGS',
                  help='Optional arguments for ansible-playbook.')
              ] + super().options(how)
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'playbook':
              return []
          return default
  
      # TODO: use better types once superclass gains its annotations
      def wake(self, keys: Optional[List[str]] = None) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys)
  
          # Convert to list if necessary
          tmt.utils.listify(self.data, keys=['playbook'])
  
      def go(self, guest: Guest) -> None:
          """ Prepare the guests """
          super().go(guest)
  
          # Apply each playbook on the guest
          for playbook in self.get('playbook'):
              self.info('playbook', playbook, 'green')
              guest.ansible(playbook, self.get('extra-args'))
  
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/ansible.py
files: import os
  import re
  import shutil
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  
  COPR_URL = 'https://copr.fedorainfracloud.org/coprs'
  
  
  class InstallBase(tmt.utils.Common):
      """ Base class for installation implementations """
  
      # Each installer knows its package manager and copr plugin
      package_manager = None
      copr_plugin = None
  
      def __init__(self, parent, guest):
          """ Initialize installation data """
          super().__init__(parent=parent, relative_indent=0)
          self.guest = guest
  
          # Get package related data from the plugin
          self.packages = self.parent.get("package", [])
          self.directories = self.parent.get("directory", [])
          self.exclude = self.parent.get("exclude", [])
          if not self.packages and not self.directories:
              self.debug("No packages for installation found.", level=3)
  
          # Prepare package lists and installation command
          self.prepare_packages()
          self.prepare_sudo()
          self.prepare_command()
  
      def prepare_packages(self):
          """ Process package names and directories """
          self.local_packages = []
          self.debuginfo_packages = []
          self.repository_packages = []
  
          # Detect local, debuginfo and repository packages
          for package in self.packages:
              if package.endswith('.rpm'):
                  self.local_packages.append(package)
              elif re.search(r"-debug(info|source)(\.|$)", package):
                  # Strip the '-debuginfo' string from package name
                  # (installing with it doesn't work on RHEL7)
                  package = re.sub(r"-debuginfo((?=\.)|$)", "", package)
                  self.debuginfo_packages.append(package)
              else:
                  self.repository_packages.append(package)
  
          # Check rpm packages in local directories
          for directory in self.directories:
              self.info('directory', directory, 'green')
              if not os.path.isdir(directory):
                  raise tmt.utils.PrepareError(f"Packages directory '{directory}' not found.")
              for filename in os.listdir(directory):
                  if filename.endswith('.rpm'):
                      self.debug(f"Found rpm '{filename}'.", level=3)
                      self.local_packages.append(os.path.join(directory, filename))
  
      def prepare_sudo(self):
          """ Check if sudo is needed for installation """
          self.debug('Check if sudo is necessary.', level=2)
          user = self.guest.execute('whoami')[0].strip()
          self.sudo = '' if user == 'root' else 'sudo '
  
      def prepare_command(self):
          """ Prepare installation command"""
          raise NotImplementedError
  
      def prepare_repository(self, **kwargs):
          """ Configure additional repository """
          raise NotImplementedError
  
      def list_packages(self, package_list, title):
          """ Show package info and return quoted package names """
          # Show a brief summary by default
          if not self.opt('verbose'):
              summary = fmf.utils.listed(package_list, max=3)
              self.info(title, summary, 'green')
          # Provide a full list of packages in verbose mode
          else:
              summary = fmf.utils.listed(package_list, 'package')
              self.info(title, summary + ' requested', 'green')
              for package in sorted(package_list):
                  self.verbose(package, shift=1)
          # Return quoted package names
          return " ".join([tmt.utils.quote(package) for package in package_list])
  
      def enable_copr_epel6(self, copr):
          """ Manually enable copr repositories for epel6 """
          # Parse the copr repo name
          matched = re.match("^(@)?([^/]+)/([^/]+)$", copr)
          if not matched:
              raise tmt.utils.PrepareError(f"Invalid copr repository '{copr}'.")
          group, name, project = matched.groups()
          group = 'group_' if group else ''
          # Prepare the repo file url
          parts = [COPR_URL] + (['g'] if group else [])
          parts += [name, project, 'repo', 'epel-6']
          parts += [f"{group}{name}-{project}-epel-6.repo"]
          url = '/'.join(parts)
          # Download the repo file on guest
          try:
              self.guest.execute(f'curl -LOf {url}', cwd='/etc/yum.repos.d')
          except tmt.utils.RunError as error:
              if 'not found' in error.stderr.lower():
                  raise tmt.utils.PrepareError(
                      f"Copr repository '{copr}' not found.")
              raise
  
      def enable_copr(self):
          """ Enable requested copr repositories """
          coprs = self.parent.get('copr')
          if not coprs:
              return
          # Try to install copr plugin
          self.debug('Make sure the copr plugin is available.')
          try:
              self.guest.execute(
                  f'rpm -q {self.copr_plugin} || {self.command} install -y {self.copr_plugin}')
          # Enable repositories manually for epel
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/install.py
files: import os
  import re
  import shutil
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  
  COPR_URL = 'https://copr.fedorainfracloud.org/coprs'
  
  
  class InstallBase(tmt.utils.Common):
      """ Base class for installation implementations """
  
      # Each installer knows its package manager and copr plugin
      package_manager = None
      copr_plugin = None
  
      def __init__(self, parent, guest):
          """ Initialize installation data """
          super().__init__(parent=parent, relative_indent=0)
          self.guest = guest
  
          # Get package related data from the plugin
          self.packages = self.parent.get("package", [])
          self.directories = self.parent.get("directory", [])
          self.exclude = self.parent.get("exclude", [])
          if not self.packages and not self.directories:
              self.debug("No packages for installation found.", level=3)
  
          # Prepare package lists and installation command
          self.prepare_packages()
          self.prepare_sudo()
          self.prepare_command()
  
      def prepare_packages(self):
          """ Process package names and directories """
          self.local_packages = []
          self.debuginfo_packages = []
          self.repository_packages = []
  
          # Detect local, debuginfo and repository packages
          for package in self.packages:
              if package.endswith('.rpm'):
                  self.local_packages.append(package)
              elif re.search(r"-debug(info|source)(\.|$)", package):
                  # Strip the '-debuginfo' string from package name
                  # (installing with it doesn't work on RHEL7)
                  package = re.sub(r"-debuginfo((?=\.)|$)", "", package)
                  self.debuginfo_packages.append(package)
              else:
                  self.repository_packages.append(package)
  
          # Check rpm packages in local directories
          for directory in self.directories:
              self.info('directory', directory, 'green')
              if not os.path.isdir(directory):
                  raise tmt.utils.PrepareError(f"Packages directory '{directory}' not found.")
              for filename in os.listdir(directory):
                  if filename.endswith('.rpm'):
                      self.debug(f"Found rpm '{filename}'.", level=3)
                      self.local_packages.append(os.path.join(directory, filename))
  
      def prepare_sudo(self):
          """ Check if sudo is needed for installation """
          self.debug('Check if sudo is necessary.', level=2)
          user = self.guest.execute('whoami')[0].strip()
          self.sudo = '' if user == 'root' else 'sudo '
  
      def prepare_command(self):
          """ Prepare installation command"""
          raise NotImplementedError
  
      def prepare_repository(self, **kwargs):
          """ Configure additional repository """
          raise NotImplementedError
  
      def list_packages(self, package_list, title):
          """ Show package info and return quoted package names """
          # Show a brief summary by default
          if not self.opt('verbose'):
              summary = fmf.utils.listed(package_list, max=3)
              self.info(title, summary, 'green')
          # Provide a full list of packages in verbose mode
          else:
              summary = fmf.utils.listed(package_list, 'package')
              self.info(title, summary + ' requested', 'green')
              for package in sorted(package_list):
                  self.verbose(package, shift=1)
          # Return quoted package names
          return " ".join([tmt.utils.quote(package) for package in package_list])
  
      def enable_copr_epel6(self, copr):
          """ Manually enable copr repositories for epel6 """
          # Parse the copr repo name
          matched = re.match("^(@)?([^/]+)/([^/]+)$", copr)
          if not matched:
              raise tmt.utils.PrepareError(f"Invalid copr repository '{copr}'.")
          group, name, project = matched.groups()
          group = 'group_' if group else ''
          # Prepare the repo file url
          parts = [COPR_URL] + (['g'] if group else [])
          parts += [name, project, 'repo', 'epel-6']
          parts += [f"{group}{name}-{project}-epel-6.repo"]
          url = '/'.join(parts)
          # Download the repo file on guest
          try:
              self.guest.execute(f'curl -LOf {url}', cwd='/etc/yum.repos.d')
          except tmt.utils.RunError as error:
              if 'not found' in error.stderr.lower():
                  raise tmt.utils.PrepareError(
                      f"Copr repository '{copr}' not found.")
              raise
  
      def enable_copr(self):
          """ Enable requested copr repositories """
          coprs = self.parent.get('copr')
          if not coprs:
              return
          # Try to install copr plugin
          self.debug('Make sure the copr plugin is available.')
          try:
              self.guest.execute(
                  f'rpm -q {self.copr_plugin} || {self.command} install -y {self.copr_plugin}')
          # Enable repositories manually for epel
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/install.py
files: from typing import Any, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  import tmt.utils
  from tmt.steps.provision import Guest
  
  
  # TODO: drop ignore once type annotations between modules enabled
  @tmt.steps.provides_method('shell')
  class PrepareShell(tmt.steps.prepare.PreparePlugin):  # type: ignore[misc]
      """
      Prepare guest using shell scripts
  
      Example config:
  
      prepare:
          how: shell
          script:
          - sudo dnf install -y 'dnf-command(copr)'
          - sudo dnf copr enable -y psss/tmt
          - sudo dnf install -y tmt
  
      Use 'order' attribute to select in which order preparation should
      happen if there are multiple configs. Default order is '50'.
      Default order of required packages installation is '70'.
      """
  
      # Supported keys
      _keys = ["script"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> Any:
          """ Prepare command line options """
          return [
              click.option(
                  '-s', '--script', metavar='SCRIPT',
                  help='Shell script to be executed.')
              ] + super().options(how)
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'script':
              return []
          return default
  
      def wake(self) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake()
  
          # Convert to list if single script provided
          tmt.utils.listify(self.data, keys=['script'])
  
      def go(self, guest: Guest) -> None:
          """ Prepare the guests """
          super().go(guest)
  
          # Give a short summary
          scripts = self.get('script')
          overview = fmf.utils.listed(scripts, 'script')
          self.info('overview', f'{overview} found', 'green')
  
          # Execute each script on the guest (with default shell options)
          for script in scripts:
              self.verbose('script', script, 'green')
              script_with_options = f'{tmt.utils.SHELL_OPTIONS}; {script}'
              guest.execute(script_with_options, cwd=self.step.plan.worktree)
  
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/shell.py
files: from typing import Any, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  import tmt.utils
  from tmt.steps.provision import Guest
  
  
  # TODO: drop ignore once type annotations between modules enabled
  @tmt.steps.provides_method('shell')
  class PrepareShell(tmt.steps.prepare.PreparePlugin):  # type: ignore[misc]
      """
      Prepare guest using shell scripts
  
      Example config:
  
      prepare:
          how: shell
          script:
          - sudo dnf install -y 'dnf-command(copr)'
          - sudo dnf copr enable -y psss/tmt
          - sudo dnf install -y tmt
  
      Use 'order' attribute to select in which order preparation should
      happen if there are multiple configs. Default order is '50'.
      Default order of required packages installation is '70'.
      """
  
      # Supported keys
      _keys = ["script"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> Any:
          """ Prepare command line options """
          return [
              click.option(
                  '-s', '--script', metavar='SCRIPT',
                  help='Shell script to be executed.')
              ] + super().options(how)
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'script':
              return []
          return default
  
      def wake(self, keys: Optional[List[str]] = None) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys)
  
          # Convert to list if single script provided
          tmt.utils.listify(self.data, keys=['script'])
  
      def go(self, guest: Guest) -> None:
          """ Prepare the guests """
          super().go(guest)
  
          # Give a short summary
          scripts = self.get('script')
          overview = fmf.utils.listed(scripts, 'script')
          self.info('overview', f'{overview} found', 'green')
  
          # Execute each script on the guest (with default shell options)
          for script in scripts:
              self.verbose('script', script, 'green')
              script_with_options = f'{tmt.utils.SHELL_OPTIONS}; {script}'
              guest.execute(script_with_options, cwd=self.step.plan.worktree)
  
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/shell.py
files: import collections
  import dataclasses
  import datetime
  import enum
  import os
  import random
  import re
  import shlex
  import string
  import subprocess
  import tempfile
  from shlex import quote
  from typing import Any, Dict, List, Optional, Type, Union, cast
  
  import click
  import fmf
  
  import tmt
  import tmt.plugins
  import tmt.steps
  import tmt.utils
  from tmt.steps import Action
  
  # Timeout in seconds of waiting for a connection after reboot
  CONNECTION_TIMEOUT = 5 * 60
  
  # When waiting for guest to recover from reboot, try re-connecting every
  # this many seconds.
  RECONNECT_WAIT_TICK = 5
  RECONNECT_WAIT_TICK_INCREASE = 1.0
  
  # Default rsync options
  DEFAULT_RSYNC_OPTIONS = [
      "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  DEFAULT_RSYNC_PUSH_OPTIONS = ["-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  DEFAULT_RSYNC_PULL_OPTIONS = ["-R", "-r", "-z", "--links", "--safe-links", "--protect-args"]
  
  
  class CheckRsyncOutcome(enum.Enum):
      ALREADY_INSTALLED = 'already-installed'
      INSTALLED = 'installed'
  
  
  class Provision(tmt.steps.Step):
      """ Provision an environment for testing or use localhost. """
  
      # Default implementation for provision is a virtual machine
      DEFAULT_HOW = 'virtual'
  
      def __init__(self, plan: 'tmt.Plan', data: tmt.steps.StepData) -> None:
          """ Initialize provision step data """
          super().__init__(plan=plan, data=data)
          # Check that the names are unique
          names = [data['name'] for data in self.data]
          names_count: Dict[str, int] = collections.defaultdict(int)
          for name in names:
              names_count[name] += 1
          if len(self.data) > 1 and len(self.data) != len(names_count):
              duplicate = [k for k, v in names_count.items() if v > 1]
              duplicate_string = ', '.join(duplicate)
              raise tmt.utils.GeneralError(
                  f"Provision step names must be unique for multihost testing. "
                  f"Duplicate names: {duplicate_string} in plan '{plan.name}'.")
          # List of provisioned guests and loaded guest data
          self._guests: List['Guest'] = []
          self._guest_data: Dict[str, 'GuestData'] = {}
          self.is_multihost = False
  
      def load(self) -> None:
          """ Load guest data from the workdir """
          super().load()
          try:
              raw_guest_data = tmt.utils.yaml_to_dict(self.read('guests.yaml'))
  
              self._guest_data = {
                  name: tmt.utils.SerializableContainer.unserialize(guest_data)
                  for name, guest_data in raw_guest_data.items()
                  }
  
          except tmt.utils.FileError:
              self.debug('Provisioned guests not found.', level=2)
  
      def save(self) -> None:
          """ Save guest data to the workdir """
          super().save()
          try:
              raw_guest_data = {guest.name: guest.save().to_serialized()
                                for guest in self.guests()}
  
              self.write('guests.yaml', tmt.utils.dict_to_yaml(raw_guest_data))
          except tmt.utils.FileError:
              self.debug('Failed to save provisioned guests.')
  
      def wake(self) -> None:
          """ Wake up the step (process workdir and command line) """
          super().wake()
  
          # Choose the right plugin and wake it up
          for data in self.data:
              # TODO: with generic BasePlugin, delegate() should return more fitting type,
              # not the base class.
              plugin = cast(ProvisionPlugin, ProvisionPlugin.delegate(self, data))
              self._phases.append(plugin)
              # If guest data loaded, perform a complete wake up
              plugin.wake(data=self._guest_data.get(plugin.name))
  
              guest = plugin.guest()
              if guest:
                  self._guests.append(guest)
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Provision wake up complete (already done before).', level=2)
          # Save status and step data (now we know what to do)
          else:
              self.status('todo')
              self.save()
  
      def show(self) -> None:
          """ Show discover details """
          for data in self.data:
              ProvisionPlugin.delegate(self, data).show()
  
      def summary(self) -> None:
          """ Give a concise summary of the provisioning """
          # Summary of provisioned guests
          guests = fmf.utils.listed(self.guests(), 'guest')
          self.info('summary', f'{guests} provisioned', 'green', shift=1)
          # Guest list in verbose mode
          for guest in self.guests():
              if not guest.name.startswith(tmt.utils.DEFAULT_NAME):
                  self.verbose
  
  [... the rest is omitted ...]
files: tmt/steps/provision/__init__.py
files: import collections
  import dataclasses
  import datetime
  import enum
  import os
  import random
  import re
  import shlex
  import string
  import subprocess
  import tempfile
  from shlex import quote
  from typing import Any, Dict, List, Optional, Type, Union, cast
  
  import click
  import fmf
  
  import tmt
  import tmt.plugins
  import tmt.steps
  import tmt.utils
  from tmt.steps import Action
  
  # Timeout in seconds of waiting for a connection after reboot
  CONNECTION_TIMEOUT = 5 * 60
  
  # When waiting for guest to recover from reboot, try re-connecting every
  # this many seconds.
  RECONNECT_WAIT_TICK = 5
  RECONNECT_WAIT_TICK_INCREASE = 1.0
  
  # Default rsync options
  DEFAULT_RSYNC_OPTIONS = [
      "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  DEFAULT_RSYNC_PUSH_OPTIONS = ["-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  DEFAULT_RSYNC_PULL_OPTIONS = ["-R", "-r", "-z", "--links", "--safe-links", "--protect-args"]
  
  
  class CheckRsyncOutcome(enum.Enum):
      ALREADY_INSTALLED = 'already-installed'
      INSTALLED = 'installed'
  
  
  class Provision(tmt.steps.Step):
      """ Provision an environment for testing or use localhost. """
  
      # Default implementation for provision is a virtual machine
      DEFAULT_HOW = 'virtual'
  
      def __init__(self, plan: 'tmt.Plan', data: tmt.steps.StepData) -> None:
          """ Initialize provision step data """
          super().__init__(plan=plan, data=data)
          # Check that the names are unique
          names = [data['name'] for data in self.data]
          names_count: Dict[str, int] = collections.defaultdict(int)
          for name in names:
              names_count[name] += 1
          if len(self.data) > 1 and len(self.data) != len(names_count):
              duplicate = [k for k, v in names_count.items() if v > 1]
              duplicate_string = ', '.join(duplicate)
              raise tmt.utils.GeneralError(
                  f"Provision step names must be unique for multihost testing. "
                  f"Duplicate names: {duplicate_string} in plan '{plan.name}'.")
          # List of provisioned guests and loaded guest data
          self._guests: List['Guest'] = []
          self._guest_data: Dict[str, 'GuestData'] = {}
          self.is_multihost = False
  
      def load(self) -> None:
          """ Load guest data from the workdir """
          super().load()
          try:
              raw_guest_data = tmt.utils.yaml_to_dict(self.read('guests.yaml'))
  
              self._guest_data = {
                  name: tmt.utils.SerializableContainer.unserialize(guest_data)
                  for name, guest_data in raw_guest_data.items()
                  }
  
          except tmt.utils.FileError:
              self.debug('Provisioned guests not found.', level=2)
  
      def save(self) -> None:
          """ Save guest data to the workdir """
          super().save()
          try:
              raw_guest_data = {guest.name: guest.save().to_serialized()
                                for guest in self.guests()}
  
              self.write('guests.yaml', tmt.utils.dict_to_yaml(raw_guest_data))
          except tmt.utils.FileError:
              self.debug('Failed to save provisioned guests.')
  
      def wake(self) -> None:
          """ Wake up the step (process workdir and command line) """
          super().wake()
  
          # Choose the right plugin and wake it up
          for data in self.data:
              # TODO: with generic BasePlugin, delegate() should return more fitting type,
              # not the base class.
              plugin = cast(ProvisionPlugin, ProvisionPlugin.delegate(self, data))
              self._phases.append(plugin)
              # If guest data loaded, perform a complete wake up
              plugin.wake(data=self._guest_data.get(plugin.name))
  
              guest = plugin.guest()
              if guest:
                  self._guests.append(guest)
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Provision wake up complete (already done before).', level=2)
          # Save status and step data (now we know what to do)
          else:
              self.status('todo')
              self.save()
  
      def show(self) -> None:
          """ Show discover details """
          for data in self.data:
              ProvisionPlugin.delegate(self, data).show()
  
      def summary(self) -> None:
          """ Give a concise summary of the provisioning """
          # Summary of provisioned guests
          guests = fmf.utils.listed(self.guests(), 'guest')
          self.info('summary', f'{guests} provisioned', 'green', shift=1)
          # Guest list in verbose mode
          for guest in self.guests():
              if not guest.name.startswith(tmt.utils.DEFAULT_NAME):
                  self.verbose
  
  [... the rest is omitted ...]
files: tmt/steps/provision/__init__.py
files: import dataclasses
  import datetime
  import sys
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import requests
  
  import tmt
  import tmt.options
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import ProvisionError, retry_session, updatable_message
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  
  # List of Artemis API versions supported and understood by this plugin.
  # Since API gains support for new features over time, it is important to
  # know when particular feature became available, and avoid using it with
  # older APIs.
  SUPPORTED_API_VERSIONS = (
      # NEW: added hostname HW constraint
      '0.0.38',
      # NEW: virtualization HW constraint
      '0.0.37',
      # NEW: boot.method HW constraint
      '0.0.32',
      # NEW: network HW constraint
      '0.0.28'
      )
  
  
  # The default Artemis API version - the most recent supported versions
  # should be perfectly fine.
  DEFAULT_API_VERSION = SUPPORTED_API_VERSIONS[0]
  
  DEFAULT_API_URL = 'http://127.0.0.1:8001'
  DEFAULT_USER = 'root'
  DEFAULT_ARCH = 'x86_64'
  DEFAULT_PRIORITY_GROUP = 'default-priority'
  DEFAULT_KEYNAME = 'default'
  DEFAULT_PROVISION_TIMEOUT = 600
  DEFAULT_PROVISION_TICK = 60
  DEFAULT_API_TIMEOUT = 10
  DEFAULT_API_RETRIES = 10
  # Should lead to delays of 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256 seconds
  DEFAULT_RETRY_BACKOFF_FACTOR = 1
  
  # Type annotation for "data" package describing a guest instance. Passed
  # between load() and save() calls
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class ArtemisGuestData(tmt.steps.provision.GuestSshData):
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      # API
      api_url: str = DEFAULT_API_URL
      api_version: str = DEFAULT_API_VERSION
  
      # Guest request properties
      arch: str = DEFAULT_ARCH
      image: Optional[str] = None
      hardware: Optional[Any] = None
      pool: Optional[str] = None
      priority_group: str = DEFAULT_PRIORITY_GROUP
      keyname: str = DEFAULT_KEYNAME
      user_data: Dict[str, str] = dataclasses.field(default_factory=dict)
  
      # Provided by Artemis response
      guestname: Optional[str] = None
  
      # Timeouts and deadlines
      provision_timeout: int = DEFAULT_PROVISION_TIMEOUT
      provision_tick: int = DEFAULT_PROVISION_TICK
      api_timeout: int = DEFAULT_API_TIMEOUT
      api_retries: int = DEFAULT_API_RETRIES
      api_retry_backoff_factor: int = DEFAULT_RETRY_BACKOFF_FACTOR
  
  
  GUEST_STATE_COLOR_DEFAULT = 'green'
  
  GUEST_STATE_COLORS = {
      'routing': 'yellow',
      'provisioning': 'magenta',
      'promised': 'blue',
      'preparing': 'cyan',
      'cancelled': 'red',
      'error': 'red'
  }
  
  
  # Type annotation for Artemis API `GET /guests/$guestname` response.
  # Partial, not all fields necessary since plugin ignores most of them.
  GuestInspectType = TypedDict(
      'GuestInspectType', {
          'state': str,
          'address': Optional[str]
          }
      )
  
  
  class ArtemisAPI:
      def __init__(self, guest: 'GuestArtemis') -> None:
          self._guest = guest
  
          self.http_session = retry_session.create(
              retries=guest.api_retries,
              backoff_factor=guest.api_retry_backoff_factor,
              allowed_methods=('HEAD', 'GET', 'POST', 'DELETE', 'PUT'),
              status_forcelist=(
                  429,  # Too Many Requests
                  500,  # Internal Server Error
                  502,  # Bad Gateway
                  503,  # Service Unavailable
                  504   # Gateway Timeout
                  ),
              timeout=guest.api_timeout
              )
  
      def query(
              self,
              path: str,
              method: str = 'get',
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Base helper for Artemis API queries.
  
          Trivial dispatcher per method, returning retrieved response.
  
          :param path: API path to contact.
          :param method: HTTP method to use.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          url = f'{self._guest.api_url}{path}
  
  [... the rest is omitted ...]
files: tmt/steps/provision/artemis.py
files: import dataclasses
  import datetime
  import sys
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import requests
  
  import tmt
  import tmt.options
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import ProvisionError, retry_session, updatable_message
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  
  # List of Artemis API versions supported and understood by this plugin.
  # Since API gains support for new features over time, it is important to
  # know when particular feature became available, and avoid using it with
  # older APIs.
  SUPPORTED_API_VERSIONS = (
      # NEW: added hostname HW constraint
      '0.0.38',
      # NEW: virtualization HW constraint
      '0.0.37',
      # NEW: boot.method HW constraint
      '0.0.32',
      # NEW: network HW constraint
      '0.0.28'
      )
  
  
  # The default Artemis API version - the most recent supported versions
  # should be perfectly fine.
  DEFAULT_API_VERSION = SUPPORTED_API_VERSIONS[0]
  
  DEFAULT_API_URL = 'http://127.0.0.1:8001'
  DEFAULT_USER = 'root'
  DEFAULT_ARCH = 'x86_64'
  DEFAULT_PRIORITY_GROUP = 'default-priority'
  DEFAULT_KEYNAME = 'default'
  DEFAULT_PROVISION_TIMEOUT = 600
  DEFAULT_PROVISION_TICK = 60
  DEFAULT_API_TIMEOUT = 10
  DEFAULT_API_RETRIES = 10
  # Should lead to delays of 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256 seconds
  DEFAULT_RETRY_BACKOFF_FACTOR = 1
  
  # Type annotation for "data" package describing a guest instance. Passed
  # between load() and save() calls
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class ArtemisGuestData(tmt.steps.provision.GuestSshData):
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      # API
      api_url: str = DEFAULT_API_URL
      api_version: str = DEFAULT_API_VERSION
  
      # Guest request properties
      arch: str = DEFAULT_ARCH
      image: Optional[str] = None
      hardware: Optional[Any] = None
      pool: Optional[str] = None
      priority_group: str = DEFAULT_PRIORITY_GROUP
      keyname: str = DEFAULT_KEYNAME
      user_data: Dict[str, str] = dataclasses.field(default_factory=dict)
  
      # Provided by Artemis response
      guestname: Optional[str] = None
  
      # Timeouts and deadlines
      provision_timeout: int = DEFAULT_PROVISION_TIMEOUT
      provision_tick: int = DEFAULT_PROVISION_TICK
      api_timeout: int = DEFAULT_API_TIMEOUT
      api_retries: int = DEFAULT_API_RETRIES
      api_retry_backoff_factor: int = DEFAULT_RETRY_BACKOFF_FACTOR
  
  
  GUEST_STATE_COLOR_DEFAULT = 'green'
  
  GUEST_STATE_COLORS = {
      'routing': 'yellow',
      'provisioning': 'magenta',
      'promised': 'blue',
      'preparing': 'cyan',
      'cancelled': 'red',
      'error': 'red'
  }
  
  
  # Type annotation for Artemis API `GET /guests/$guestname` response.
  # Partial, not all fields necessary since plugin ignores most of them.
  GuestInspectType = TypedDict(
      'GuestInspectType', {
          'state': str,
          'address': Optional[str]
          }
      )
  
  
  class ArtemisAPI:
      def __init__(self, guest: 'GuestArtemis') -> None:
          self._guest = guest
  
          self.http_session = retry_session.create(
              retries=guest.api_retries,
              backoff_factor=guest.api_retry_backoff_factor,
              allowed_methods=('HEAD', 'GET', 'POST', 'DELETE', 'PUT'),
              status_forcelist=(
                  429,  # Too Many Requests
                  500,  # Internal Server Error
                  502,  # Bad Gateway
                  503,  # Service Unavailable
                  504   # Gateway Timeout
                  ),
              timeout=guest.api_timeout
              )
  
      def query(
              self,
              path: str,
              method: str = 'get',
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Base helper for Artemis API queries.
  
          Trivial dispatcher per method, returning retrieved response.
  
          :param path: API path to contact.
          :param method: HTTP method to use.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          url = f'{self._guest.api_url}{path}
  
  [... the rest is omitted ...]
files: tmt/steps/provision/artemis.py
files: from typing import Any, List, Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.steps.provision import GuestSshData
  
  
  @tmt.steps.provides_method('connect')
  class ProvisionConnect(tmt.steps.provision.ProvisionPlugin):
      """
      Connect to a provisioned guest using ssh
  
      Private key authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              key: /home/psss/.ssh/example_rsa
  
      Password authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              password: secret
  
      User defaults to 'root', so if you have private key correctly set
      the minimal configuration can look like this:
  
          provision:
              how: connect
              guest: host.example.org
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["guest", "key", "user", "password", "port"]
  
      @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(
                  '-g', '--guest', metavar='GUEST',
                  help='Select remote host to connect to (hostname or ip).'),
              click.option(
                  '-P', '--port', metavar='PORT',
                  help='Use specific port to connect to.'),
              click.option(
                  '-k', '--key', metavar='PRIVATE_KEY',
                  help='Private key for login into the guest system.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-p', '--password', metavar='PASSWORD',
                  help='Password for login into the guest system.'),
              ]
          return options
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          # User root as the default user
          if option == 'user':
              return 'root'
          # No other defaults available
          return default
  
      def wake(self, data: Optional[GuestSshData] = None) -> None:  # type: ignore[override]
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
          if data:
              self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def go(self) -> None:
          """ Prepare the connection """
          super().go()
  
          # Check connection details
          guest = self.get('guest')
          user = self.get('user')
          key = self.get('key')
          password = self.get('password')
          port = self.get('port')
  
          # Check guest and auth info
          if not guest:
              raise tmt.utils.SpecificationError(
                  'Provide a host name or an ip address to connect.')
          data = GuestSshData(
              role=self.get('role'),
              guest=guest,
              user=user
              )
          self.info('guest', guest, 'green')
          self.info('user', user, 'green')
          if port:
              self.info('port', port, 'green')
              data.port = port
  
          # Use provided password for authentication
          if password:
              self.info('password', password, 'green')
              self.debug('Using password authentication.')
              data.password = password
          # Default to using a private key (user can have configured one)
          else:
              self.info('key', key or 'not provided', 'green')
              self.debug('Using private key authentication.')
              data.key = [key]
  
          # And finally create the guest
          self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def guest(self) -> Optional[tmt.GuestSsh]:
          """ Return the provisioned guest """
          return self._guest
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/connect.py
files: from typing import Any, List, Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.steps.provision import GuestSshData
  
  
  @tmt.steps.provides_method('connect')
  class ProvisionConnect(tmt.steps.provision.ProvisionPlugin):
      """
      Connect to a provisioned guest using ssh
  
      Private key authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              key: /home/psss/.ssh/example_rsa
  
      Password authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              password: secret
  
      User defaults to 'root', so if you have private key correctly set
      the minimal configuration can look like this:
  
          provision:
              how: connect
              guest: host.example.org
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["guest", "key", "user", "password", "port"]
  
      @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(
                  '-g', '--guest', metavar='GUEST',
                  help='Select remote host to connect to (hostname or ip).'),
              click.option(
                  '-P', '--port', metavar='PORT',
                  help='Use specific port to connect to.'),
              click.option(
                  '-k', '--key', metavar='PRIVATE_KEY',
                  help='Private key for login into the guest system.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-p', '--password', metavar='PASSWORD',
                  help='Password for login into the guest system.'),
              ]
          return options
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          # User root as the default user
          if option == 'user':
              return 'root'
          # No other defaults available
          return default
  
      # More specific type is a violation of Liskov substitution principle, and mypy
      # complains about it - rightfully so. Ignoring the issue which should be resolved
      # with https://github.com/teemtee/tmt/pull/1439.
      def wake(self, keys: Optional[List[str]] = None,  # type: ignore[override]
               data: Optional[GuestSshData] = None) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          if data:
              self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def go(self) -> None:
          """ Prepare the connection """
          super().go()
  
          # Check connection details
          guest = self.get('guest')
          user = self.get('user')
          key = self.get('key')
          password = self.get('password')
          port = self.get('port')
  
          # Check guest and auth info
          if not guest:
              raise tmt.utils.SpecificationError(
                  'Provide a host name or an ip address to connect.')
          data = GuestSshData(
              role=self.get('role'),
              guest=guest,
              user=user
              )
          self.info('guest', guest, 'green')
          self.info('user', user, 'green')
          if port:
              self.info('port', port, 'green')
              data.port = port
  
          # Use provided password for authentication
          if password:
              self.info('password', password, 'green')
              self.debug('Using password authentication.')
              data.password = password
          # Default to using a private key (user can have configured one)
          else:
              self.info('key', key or 'not provided', 'green')
              self.debug('Using private key authentication.')
              data.key = [key]
  
          # And finally create the guest
          self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def guest(self) -> Optional[tmt.GuestSsh]:
          """ Return the provisioned guest """
          return self._guest
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/connect.py
files: import tmt
  import tmt.steps
  import tmt.steps.provision
  
  
  @tmt.steps.provides_method('local')
  class ProvisionLocal(tmt.steps.provision.ProvisionPlugin):
      """
      Use local host for test execution
  
      In general it is not recommended to run tests on your local machine
      as there might be security risks. Run only those tests which you
      know are safe so that you don't destroy your laptop ;-)
  
      Example config:
  
          provision:
              how: local
  
      Note that 'tmt run' is expected to be executed under a regular user.
      If there are admin rights required (for example in the prepare step)
      you might be asked for a sudo password.
      """
  
      # Guest instance
      _guest = None
  
      def wake(self, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
          if data:
              self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Create a GuestLocal instance
          data = tmt.steps.provision.GuestSshData(
              guest='localhost',
              role=self.get('role')
              )
          self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestLocal.requires()
  
  
  class GuestLocal(tmt.Guest):
      """ Local Host """
  
      localhost = True
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, stderr = self.run(
              ['sudo', '-E', 'ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'local', '-i', 'localhost,', playbook],
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def execute(self, command, **kwargs):
          """ Execute command on localhost """
          # Prepare the environment (plan/cli variables override)
          environment = dict()
          environment.update(kwargs.pop('env', dict()))
          environment.update(self.parent.plan.environment)
          # Run the command under the prepared environment
          return self.run(command, env=environment, shell=True, **kwargs)
  
      def stop(self):
          """ Stop the guest """
  
          self.debug(f"Doing nothing to stop guest '{self.guest}'.")
  
      def reboot(self, hard=False, command=None, timeout=None):
          """ Reboot the guest, return True if successful """
  
          self.debug(f"Doing nothing to reboot guest '{self.guest}'.")
  
          return False
  
      def push(self, source=None, destination=None, options=None):
          """ Nothing to be done to push workdir """
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """ Nothing to be done to pull workdir """
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/local.py
files: import tmt
  import tmt.steps
  import tmt.steps.provision
  
  
  @tmt.steps.provides_method('local')
  class ProvisionLocal(tmt.steps.provision.ProvisionPlugin):
      """
      Use local host for test execution
  
      In general it is not recommended to run tests on your local machine
      as there might be security risks. Run only those tests which you
      know are safe so that you don't destroy your laptop ;-)
  
      Example config:
  
          provision:
              how: local
  
      Note that 'tmt run' is expected to be executed under a regular user.
      If there are admin rights required (for example in the prepare step)
      you might be asked for a sudo password.
      """
  
      # Guest instance
      _guest = None
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          if data:
              self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Create a GuestLocal instance
          data = tmt.steps.provision.GuestSshData(
              guest='localhost',
              role=self.get('role')
              )
          self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestLocal.requires()
  
  
  class GuestLocal(tmt.Guest):
      """ Local Host """
  
      localhost = True
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, stderr = self.run(
              ['sudo', '-E', 'ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'local', '-i', 'localhost,', playbook],
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def execute(self, command, **kwargs):
          """ Execute command on localhost """
          # Prepare the environment (plan/cli variables override)
          environment = dict()
          environment.update(kwargs.pop('env', dict()))
          environment.update(self.parent.plan.environment)
          # Run the command under the prepared environment
          return self.run(command, env=environment, shell=True, **kwargs)
  
      def stop(self):
          """ Stop the guest """
  
          self.debug(f"Doing nothing to stop guest '{self.guest}'.")
  
      def reboot(self, hard=False, command=None, timeout=None):
          """ Reboot the guest, return True if successful """
  
          self.debug(f"Doing nothing to reboot guest '{self.guest}'.")
  
          return False
  
      def push(self, source=None, destination=None, options=None):
          """ Nothing to be done to push workdir """
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """ Nothing to be done to pull workdir """
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/local.py
files: import dataclasses
  import os
  from shlex import quote
  from typing import Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  
  # Timeout in seconds of waiting for a connection
  CONNECTION_TIMEOUT = 60
  
  # Defaults
  DEFAULT_IMAGE = "fedora"
  DEFAULT_USER = "root"
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class PodmanGuestData(tmt.steps.provision.GuestData):  # type: ignore[misc]
      image: str = DEFAULT_IMAGE
      user: str = DEFAULT_USER
      force_pull: bool = False
  
      container: Optional[str] = None
  
  
  @tmt.steps.provides_method('container')
  class ProvisionPodman(tmt.steps.provision.ProvisionPlugin):
      """
      Create a new container using podman
  
      Example config:
  
          provision:
              how: container
              image: fedora:latest
  
      In order to always pull the fresh container image use 'pull: true'.
  
      In order to run the container with different user as the default 'root',
      use 'user: USER'.
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["image", "container", "pull", "user"]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for connect """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to use. Short name or complete url.'),
              click.option(
                  '-c', '--container', metavar='NAME',
                  help='Name or id of an existing container to be used.'),
              click.option(
                  '-p', '--pull', is_flag=True,
                  help='Force pulling a fresh container image.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='User to use for all container operations.')
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          if option == 'pull':
              return PodmanGuestData().force_pull
  
          return getattr(PodmanGuestData(), option.replace('-', '_'), default)
  
      def wake(self, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
          # Wake up podman instance
          if data:
              guest = GuestContainer(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Show which image we are using
          pull = ' (force pull)' if self.get('pull') else ''
          self.info('image', f"{self.get('image')}{pull}", 'green')
  
          # Prepare data for the guest instance
          data_from_options = {
              key: self.get(key)
              for key in PodmanGuestData.keys()
              if key != 'force_pull'
              }
  
          data_from_options['force_pull'] = self.get('pull')
  
          data = PodmanGuestData(**data_from_options)
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestContainer(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestContainer.requires()
  
  
  class GuestContainer(tmt.Guest):
      """ Container Instance """
  
      _data_class = PodmanGuestData
  
      image: Optional[str]
      container: Optional[str]
      user: str
      force_pull: bool
  
      def wake(self):
          """ Wake up the guest """
          self.debug(
              f"Waking up container '{self.container}'.", level=2, shift=0)
  
      def start(self):
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Check if the image is available
          image_id = self.podman(
              ['images', '-q', self.image],
              message=f"Check for container image '{self.image}'.")[0].strip()
  
          # Pull image if not available or pull forced
          if not image_id or self.force_pull:
              self.podman(
                  ['pull', '-q', self.image],
                  message=f"Pull image '{self.image}'.")
  
          # Mount the whole plan directory in the container
          workdir = self.parent.plan.workdir
  
          self.container = self.guest = self._tmt_name()
          self.verbose('name', self.container, 'green')
  
          # FIXME: Workaround for BZ#1900021 (f34 container on centos-8)
          workaround = ['--security-opt', 'seccomp=unconfined']
  
          # Run the container
          self.debug(f"Start container '{self.image}'.")
          self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f'{workdir}:{workdir}:z',
               '-
  
  [... the rest is omitted ...]
files: tmt/steps/provision/podman.py
files: import dataclasses
  import os
  from shlex import quote
  from typing import Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  
  # Timeout in seconds of waiting for a connection
  CONNECTION_TIMEOUT = 60
  
  # Defaults
  DEFAULT_IMAGE = "fedora"
  DEFAULT_USER = "root"
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class PodmanGuestData(tmt.steps.provision.GuestData):  # type: ignore[misc]
      image: str = DEFAULT_IMAGE
      user: str = DEFAULT_USER
      force_pull: bool = False
  
      container: Optional[str] = None
  
  
  @tmt.steps.provides_method('container')
  class ProvisionPodman(tmt.steps.provision.ProvisionPlugin):
      """
      Create a new container using podman
  
      Example config:
  
          provision:
              how: container
              image: fedora:latest
  
      In order to always pull the fresh container image use 'pull: true'.
  
      In order to run the container with different user as the default 'root',
      use 'user: USER'.
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["image", "container", "pull", "user"]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for connect """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to use. Short name or complete url.'),
              click.option(
                  '-c', '--container', metavar='NAME',
                  help='Name or id of an existing container to be used.'),
              click.option(
                  '-p', '--pull', is_flag=True,
                  help='Force pulling a fresh container image.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='User to use for all container operations.')
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          if option == 'pull':
              return PodmanGuestData().force_pull
  
          return getattr(PodmanGuestData(), option.replace('-', '_'), default)
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          # Wake up podman instance
          if data:
              guest = GuestContainer(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Show which image we are using
          pull = ' (force pull)' if self.get('pull') else ''
          self.info('image', f"{self.get('image')}{pull}", 'green')
  
          # Prepare data for the guest instance
          data_from_options = {
              key: self.get(key)
              for key in PodmanGuestData.keys()
              if key != 'force_pull'
              }
  
          data_from_options['force_pull'] = self.get('pull')
  
          data = PodmanGuestData(**data_from_options)
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestContainer(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestContainer.requires()
  
  
  class GuestContainer(tmt.Guest):
      """ Container Instance """
  
      _data_class = PodmanGuestData
  
      image: Optional[str]
      container: Optional[str]
      user: str
      force_pull: bool
  
      def wake(self):
          """ Wake up the guest """
          self.debug(
              f"Waking up container '{self.container}'.", level=2, shift=0)
  
      def start(self):
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Check if the image is available
          image_id = self.podman(
              ['images', '-q', self.image],
              message=f"Check for container image '{self.image}'.")[0].strip()
  
          # Pull image if not available or pull forced
          if not image_id or self.force_pull:
              self.podman(
                  ['pull', '-q', self.image],
                  message=f"Pull image '{self.image}'.")
  
          # Mount the whole plan directory in the container
          workdir = self.parent.plan.workdir
  
          self.container = self.guest = self._tmt_name()
          self.verbose('name', self.container, 'green')
  
          # FIXME: Workaround for BZ#1900021 (f34 container on centos-8)
          workaround = ['--security-opt', 'seccomp=unconfined']
  
          # Run the container
          self.debug(f"Start container '{self.image}'.")
          self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f'{workdir}:
  
  [... the rest is omitted ...]
files: tmt/steps/provision/podman.py
files: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  from typing import Optional
  
  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
  
  
  def import_testcloud():
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
    - {public_key}
  ssh_pwauth: true
  disable_root: false
  runcmd:
    - sed -i -e '/^.*PermitRootLogin/s/^.*$/PermitRootLogin yes/'
      -e '/^.*UseDNS/s/^.*$/UseDNS no/'
      -e '/^.*GSSAPIAuthentication/s/^.*$/GSSAPIAuthentication no/'
      /etc/ssh/sshd_config
    - systemctl reload sshd
    - [sh, -c, 'if [ ! -f /etc/systemd/network/20-tc-usernet.network ] &&
    systemctl status systemd-networkd | grep -q "enabled;\\svendor\\spreset:\\senabled";
    then mkdir -p /etc/systemd/network/ &&
    echo "[Match]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "Name=en*" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "[Network]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "DHCP=yes" >> /etc/systemd/network/20-tc-usernet.network; fi']
    - [sh, -c, 'if systemctl status systemd-networkd |
    grep -q "enabled;\\svendor\\spreset:\\senabled"; then
    systemctl restart systemd-networkd; fi']
    - [sh, -c, 'if cat /etc/os-release |
    grep -q platform:el8; then systemctl restart sshd; fi']
  """
  
  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
  
  [... the rest is omitted ...]
files: tmt/steps/provision/testcloud.py
files: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  from typing import Optional
  
  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
  
  
  def import_testcloud():
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
    - {public_key}
  ssh_pwauth: true
  disable_root: false
  runcmd:
    - sed -i -e '/^.*PermitRootLogin/s/^.*$/PermitRootLogin yes/'
      -e '/^.*UseDNS/s/^.*$/UseDNS no/'
      -e '/^.*GSSAPIAuthentication/s/^.*$/GSSAPIAuthentication no/'
      /etc/ssh/sshd_config
    - systemctl reload sshd
    - [sh, -c, 'if [ ! -f /etc/systemd/network/20-tc-usernet.network ] &&
    systemctl status systemd-networkd | grep -q "enabled;\\svendor\\spreset:\\senabled";
    then mkdir -p /etc/systemd/network/ &&
    echo "[Match]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "Name=en*" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "[Network]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "DHCP=yes" >> /etc/systemd/network/20-tc-usernet.network; fi']
    - [sh, -c, 'if systemctl status systemd-networkd |
    grep -q "enabled;\\svendor\\spreset:\\senabled"; then
    systemctl restart systemd-networkd; fi']
    - [sh, -c, 'if cat /etc/os-release |
    grep -q platform:el8; then systemctl restart sshd; fi']
  """
  
  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
  
  [... the rest is omitted ...]
files: tmt/steps/provision/testcloud.py