files: .. _classes:
  
  ===============
      Classes
  ===============
  
  
  Class Structure
  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  Here's the overview of core classes::
  
      Common
      ├── Tree
      ├── Core
      │   ├── Plan
      │   ├── Story
      │   └── Test
      ├── Step
      │   ├── Discover
      │   ├── Execute
      │   ├── Finish
      │   ├── Prepare
      │   ├── Provision
      │   └── Report
      ├── Plugin
      │   ├── DiscoverPlugin
      │   │   ├── DiscoverFmf
      │   │   └── DiscoverShell
      │   ├── ExecutePlugin
      │   │   ├── ExecuteDetach
      │   │   └── ExecuteInternal
      │   ├── FinishPlugin
      │   │   └── FinishShell
      │   ├── PreparePlugin
      │   │   ├── PrepareAnsible
      │   │   ├── PrepareInstall
      │   │   └── PrepareShell
      │   ├── ProvisionPlugin
      │   │   ├── ProvisionConnect
      │   │   ├── ProvisionLocal
      │   │   ├── ProvisionMinute
      │   │   ├── ProvisionPodman
      │   │   └── ProvisionTestcloud
      │   └── ReportPlugin
      │       ├── ReportDisplay
      │       └── ReportHtml
      ├── Guest
      │   ├── GuestContainer
      │   ├── GuestLocal
      │   ├── GuestMinute
      │   └── GuestTestcloud
      ├── Run
      ├── Status
      └── Clean
  
  
  Attributes
  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  Object hierarchy is following: ``Run`` -> ``Plans`` -> ``Steps``
  -> ``Plugins``, where the ``Run`` is on the top of this hierarchy.
  The objects have the ``parent`` attribute, that is pointing to
  the parent in which the current instance is contained.
  
  The ``node`` attribute of ``Test``, ``Plan`` and ``Story``
  instances references the original leaf node of the fmf metadata
  tree from which the respective test, plan or story have been
  created.
  
  In a similar way, the ``tree`` property of the ``Tree`` instance
  points to the original ``fmf.Tree`` from which it was initialized.
  
  
  Class Conversions
  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  Various internal objects and classes often need to be converted
  from their Python nature to data that can be saved, loaded or
  exported in different form. To facilitate these conversions, three
  families of helper methods are provided, each with its own set of
  use cases.
  
  ``to_spec``/``from_spec``
  ------------------------------------------------------------------
  
  This family of methods works with tmt *specification*, i.e. raw
  user-provided data coming from fmf files describing plans, tests,
  stories, or from command-line options. ``from_spec()`` shall be
  called to spawn objects representing the user input, while
  ``to_spec()`` should produce output one could find in fmf files.
  
  The default implementation comes from ``tmt.utils.SpecBasedContainer``
  class, all classes based on user input data should include this
  class among their bases.
  
  .. code-block:: python
  
     # Create an fmf id object from raw data
     fmf_id = tmt.base.FmfId.from_spec({'url': ..., 'ref': ...})
  
  
  ``to_serialized``/``from_serialized``/``unserialize``
  ------------------------------------------------------------------
  
  This family of methods is aiming at runtime objects that may be
  saved into and loaded from tmt working files, i.e. files tmt uses
  to store a state in its workdir, like `step.yaml` or `guests.yaml`.
  
  Third member of this family, ``unserialize``, is similar to
  ``from_serialized`` - both create an object from its serialized form,
  only ``unserialize`` is capable of detecting the class to instantiate
  while for using ``from_serialized``, one must already know which
  class to work with. ``unserialize`` then uses ``from_serialized``
  under the hood to do the heavy lifting when correct class is
  identified.
  
  The default implementation comes from ``tmt.utils.SerializableContainer``
  class, all classes that are being saved and loaded during tmt run
  should include this class among their bases.
  
  See https://en.wikipedia.org/wiki/Serialization for more details
  on the concept of serialization.
  
  .. code-block:: python
  
      # tmt.steps.discover.shell.DiscoverShellData wishes to unserialize its
      # `tests` a list of `TestDescription` objects rather than a list of
      # dictionaries (the default implementation).
      @classmethod
      def from_serialized(cls, serialized: Dict[str, Any]) -> 'DiscoverShellData':
          obj = super().from_serialized(serialized)
  
          obj.tests = [TestDescription.from_serialized(
              serialized_test) for serialized_test in serialized['tests']]
  
          return obj
  
     # A step saving its state...
     content: Dict[str, Any] = {
         'status': self.status(),
         'data': [datum.to_serialized() for datum in self.data]
     }
     self.write('step.yaml', tmt.utils.dict_to_yaml(content))
  
     # ... and loading it back.
     # Note the use of unserialize(): step data may have been serialized from
     # various different classes (derived from tmt.steps.provision.Guest),
     # and unserialize() will detect the correct class.
     raw_step_data: Dict[Any, Any] = tmt.utils.yaml_to_dict(self.read('step.yaml'))
     self.data = [
         StepData.unserialize(raw_datum) for raw_datum in raw_step_data['data']
     ]
  
  
  ``to_dict``
  ------------------------------------------------------------------
  
  Very special helper method: its use cases are not related to any
  input or output data, and most of the time, when in need of
  iterating over object's keys and/or values, one can use ``keys()``,
  ``values()`` or ``items()`` methods. It is used as a source of data
  for serialization and validation, but it usually has no use outside
  of default implementations.
  
  .. warning::
  
     If you think of using ``to_dict()``, please, think again and be sure
     you know what are you doing. Despite its output being sometimes
     perfectly compatible with output of ``to_serialized()`` or ``to_spec()``,
     it is not generaly true, and using it instead of proper methods may lead
     to unexpected exceptions.
  
  .. code-block:: python
  
     # tmt.base.FmfId's specification is basically just a mapping,
     # therefore `to_dict()` is good enough to produce a specification.
     def to_spec(self) -> Dict[str, Any]:
          return self.to_dict()
  
  
  Essential Classes
  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  .. automodule:: tmt
      :members:
      :undoc-members:
  
  
  [... the rest is omitted ...]
files: docs/classes.rst
files: .. _classes:
  
  ===============
      Classes
  ===============
  
  
  Class Structure
  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  Here's the overview of core classes::
  
      Common
      ├── Tree
      ├── Core
      │   ├── Plan
      │   ├── Story
      │   └── Test
      ├── Step
      │   ├── Discover
      │   ├── Execute
      │   ├── Finish
      │   ├── Prepare
      │   ├── Provision
      │   └── Report
      ├── Plugin
      │   ├── DiscoverPlugin
      │   │   ├── DiscoverFmf
      │   │   └── DiscoverShell
      │   ├── ExecutePlugin
      │   │   ├── ExecuteDetach
      │   │   └── ExecuteInternal
      │   ├── FinishPlugin
      │   │   └── FinishShell
      │   ├── PreparePlugin
      │   │   ├── PrepareAnsible
      │   │   ├── PrepareInstall
      │   │   └── PrepareShell
      │   ├── ProvisionPlugin
      │   │   ├── ProvisionConnect
      │   │   ├── ProvisionLocal
      │   │   ├── ProvisionMinute
      │   │   ├── ProvisionPodman
      │   │   └── ProvisionTestcloud
      │   └── ReportPlugin
      │       ├── ReportDisplay
      │       └── ReportHtml
      ├── Guest
      │   ├── GuestContainer
      │   ├── GuestLocal
      │   ├── GuestMinute
      │   └── GuestTestcloud
      ├── Run
      ├── Status
      └── Clean
  
  
  Attributes
  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  Object hierarchy is following: ``Run`` -> ``Plans`` -> ``Steps``
  -> ``Plugins``, where the ``Run`` is on the top of this hierarchy.
  The objects have the ``parent`` attribute, that is pointing to
  the parent in which the current instance is contained.
  
  The ``node`` attribute of ``Test``, ``Plan`` and ``Story``
  instances references the original leaf node of the fmf metadata
  tree from which the respective test, plan or story have been
  created.
  
  In a similar way, the ``tree`` property of the ``Tree`` instance
  points to the original ``fmf.Tree`` from which it was initialized.
  
  
  Essential Classes
  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  .. automodule:: tmt
      :members:
      :undoc-members:
  
  
  [... the rest is omitted ...]
files: docs/classes.rst
files: 
  """ Base Metadata Classes """
  
  import copy
  import dataclasses
  import enum
  import functools
  import os
  import re
  import shutil
  import subprocess
  import sys
  import time
  from typing import Any, ClassVar, Dict, List, Optional, Tuple, Union, cast
  
  import click
  import fmf
  import fmf.base
  from click import confirm, echo, style
  from fmf.utils import listed
  from ruamel.yaml.error import MarkedYAMLError
  
  import tmt.export
  import tmt.identifier
  import tmt.steps
  import tmt.steps.discover
  import tmt.steps.execute
  import tmt.steps.finish
  import tmt.steps.prepare
  import tmt.steps.provision
  import tmt.steps.report
  import tmt.templates
  import tmt.utils
  from tmt.utils import FmfContextType, verdict
  
  if sys.version_info >= (3, 8):
      from typing import Literal, TypedDict
  else:
      from typing_extensions import Literal, TypedDict
  
  # Default test duration is 5m for individual tests discovered from L1
  # metadata and 1h for scripts defined directly in plans (L2 metadata).
  DEFAULT_TEST_DURATION_L1 = '5m'
  DEFAULT_TEST_DURATION_L2 = '1h'
  DEFAULT_ORDER = 50
  
  # How many already existing lines should tmt run --follow show
  FOLLOW_LINES = 10
  
  # Obsoleted test keys
  OBSOLETED_TEST_KEYS = "relevancy coverage".split()
  
  # Unofficial temporary test keys
  EXTRA_TEST_KEYS = (
      "extra-nitrate extra-hardware extra-pepa "
      "extra-summary extra-task id".split())
  
  # Unofficial temporary story keys
  EXTRA_STORY_KEYS = ("id".split())
  
  SECTIONS_HEADINGS = {
      'Setup': ['<h1>Setup</h1>'],
      'Test': ['<h1>Test</h1>',
               '<h1>Test .*</h1>'],
      'Step': ['<h2>Step</h2>',
               '<h2>Test Step</h2>'],
      'Expect': ['<h2>Expect</h2>',
                 '<h2>Result</h2>',
                 '<h2>Expected Result</h2>'],
      'Cleanup': ['<h1>Cleanup</h1>']
      }
  
  # Extra keys used for identification in Result class
  EXTRA_RESULT_IDENTIFICATION_KEYS = ['extra-nitrate', 'extra-task']
  
  
  #
  # fmf id types
  #
  # See https://fmf.readthedocs.io/en/latest/concept.html#identifiers for
  # formal specification.
  #
  
  # A "raw" fmf id as stored in fmf node data, i.e. as a mapping with various keys.
  # Used for a brief moment, to annotate raw fmf data before they are converted
  # into FmfId instances.
  _RawFmfId = TypedDict('_RawFmfId', {
      'url': Optional[str],
      'ref': Optional[str],
      'path': Optional[str],
      'name': Optional[str]
  })
  
  
  # An internal fmf id representation.
  @dataclasses.dataclass
  class FmfId(tmt.utils.SpecBasedContainer, tmt.utils.SerializableContainer):
      # The list of valid fmf id keys
      keys: ClassVar[List[str]] = ['url', 'ref', 'path', 'name']
  
      url: Optional[str] = None
      ref: Optional[str] = None
      path: Optional[str] = None
      name: Optional[str] = None
  
      def to_spec(self) -> Dict[str, Any]:
          """ Convert to a form suitable for saving in a specification file """
  
          return self.to_dict()
  
      @classmethod
      def from_spec(cls, raw: _RawFmfId) -> 'FmfId':
          """ Convert from a specification file or from a CLI option """
  
          return FmfId(**{key: raw.get(key, None) for key in cls.keys})
  
      def validate(self) -> Tuple[bool, str]:
          """
          Validate fmf id and return a human readable error
  
          Return a tuple (boolean, message) as the result of validation.
          The boolean specifies the validation result and the message
          the validation error. In case the FMF id is valid, return an empty
          string as the message.
          """
          # Validate remote id and translate to human readable errors
          try:
              # Simple asdict() is not good enough, fmf does not like keys that exist but are `None`.
              # Don't include those.
              fmf.base.Tree.node({
                  key: value for key, value in self.items()
                  if value is not None
                  })
          except fmf.utils.GeneralError as error:
              # Map fmf errors to more user friendly alternatives
              error_map: List[Tuple[str, str]] = [
                  ('git clone', f"repo '{self.url}' cannot be cloned"),
                  ('git checkout', f"git ref '{self.ref}' is invalid"),
                  ('directory path', f"path '{self.path}' is invalid"),
                  ('tree root', f"No tree found in repo '{self.url}', missing an '.fmf' directory?")
                  ]
  
              stringified_error = str(error)
  
              errors = [message for needle, message in error_map if needle in stringified_error]
              return (False, errors[0] if errors else stringified_error)
  
          return (True, '')
  
  #
  # Various types describing constructs as stored in "raw" fmf node data.
  # Used for a brief moment, to annotate raw fmf data before they are converted
  # into their internal representations.
  #
  
  #
  # A type describing the raw form of the core `link` attribute. See
  # https://tmt.readthedocs.io/en/stable/spec/core.html#link for its
  # formal specification. Internally, a link is represented by a `Link`
  # class instance, and types below describe the raw data coming from Fmf
  # nodes and CLI options.
  
  
  # Link relations.
  _RawLinkRelationName = Literal[
      'verifies', 'verified-by',
      'implements', 'implemented-by',
      'documents', 'documented-by',
      'blocks', 'blocked-by',
      'duplicates', 'duplicated-by',
      'parent', 'child',
      'relates',
      # Special case: not a relation, but it can appear where relations appear in
      # link data structures.
      'note'
  ]
  
  # Link target - can be either a string (like test case name or URL), or an fmf id.
  _RawLinkTarget = Union[str, _RawFmfId]
  
  # Basic "relation-aware" link - essentialy a mapping with one key/value pair.
  _RawLinkRelation = Dict[_RawLinkRelationName, _RawLinkTarget]
  
  # A single link can be represented as a string or FMF ID (meaning only target is specified),
  # or a "relation-aware" link aka mapping defined above.
  _RawLink = Union[
      str,
      _RawFmfId,
      _RawLinkRelation
  ]
  
  # Collection of links - can be either a single link, or a list of links, and all
  # link forms may be used together.
  _RawLinks = Union[
      _RawLink,
      List[_RawLink]
  ]
  
  
  # A type describing `adjust` content. See
  # https://tmt.readthedocs.io/en/stable/spec/core.html#adjust for its formal specification.
  #
  # The type is incomplete in the sense it does not describe all keys it may contain,
  # like `environment` or `provision`, and focuses only on the keys defined for `adjust`
  # itself.
  _RawAdjustRule = TypedDict(
      'AdjustRuleType',
      {
          'when': Optional[str],
          'continue': Optional[bool],
          'because': Optional[str]
          }
  )
  
  # A type describing content accepted by various require-like keys: - a string, fmf id,
  # or a list with a mixture of these two types.
  #
  # TODO: what other keys accept this kind of fmf data??
  _RawRequire = Union[
      str,
      _RawFmfId,
      List[Union[str, _RawFmfId]]
  ]
  
  
  class Core(tmt.utils.Common):
      """
      General node object
  
      Corresponds to given fmf.Tree node.
      Implements common Test, Plan and Story methods.
      Also defines L0 metadata and its manipulation.
      """
  
      # Core attributes (supported across all levels)
      summary: Optional[str] = None
      description: Optional[str] = None
      enabled: bool = True
      order: int = DEFAULT_ORDER
      link: Optional['Links'] = None
      id: Optional[str] = None
      tag: List[str] = []
      tier: Optional[str] = None
      adjust: Optional[List[_RawAdjustRule]] = None
  
      KEYS_SHOW_ORDER = [
          # Basic stuff
          'summary',
          'description',
          'enabled',
          'order',
          'id',
  
          # Filtering and more
          'tag',
          'tier',
          'link',
          'adjust',
          ]
  
      # Normalization methods
      _normalize_tag = tmt.utils.LoadFmfKeysMixin._normalize_string_list
  
      # TODO: remove when schema becomes mandatory, `order` shall never be `null`
      def _normalize_order(self, value: Optional[int]) -> int:
          if value is None:
              return DEFAULT_ORDER
          return int(value)
  
      def _normalize_link(self, value: _RawLinks) -> 'Links':
          return Links(data=value)
  
      def _normalize_adjust(
              self, value: Union[_RawAdjustRule, List[_RawAdjustRule]]) -> List[_RawAdjustRule]:
          if value is None:
              return []
          return [value] if not isinstance(value, list) else value
  
      def _normalize_tier(self, value: Optional[Union[int, str]]) -> Optional[str]:
          if value is None:
              return None
          return str(value)
  
      def __init__(self,
                   *,
                   node: fmf.Tree,
                   parent: Optional[tmt.utils.Common] = None,
                   **kwargs: Any) -> None:
          """ Initialize the node """
          kwargs.setdefault('logger', self)
          super().__init__(node=node, parent=parent, name=node.name, **kwargs)
  
          self.node = node
  
  
  
  [... the rest is omitted ...]
files: tmt/base.py
files: 
  """ Base Metadata Classes """
  
  import copy
  import dataclasses
  import enum
  import functools
  import os
  import re
  import shutil
  import subprocess
  import sys
  import time
  from typing import Any, ClassVar, Dict, List, Optional, Tuple, Union, cast
  
  import click
  import fmf
  import fmf.base
  from click import confirm, echo, style
  from fmf.utils import listed
  from ruamel.yaml.error import MarkedYAMLError
  
  import tmt.export
  import tmt.identifier
  import tmt.steps
  import tmt.steps.discover
  import tmt.steps.execute
  import tmt.steps.finish
  import tmt.steps.prepare
  import tmt.steps.provision
  import tmt.steps.report
  import tmt.templates
  import tmt.utils
  from tmt.utils import FmfContextType, verdict
  
  if sys.version_info >= (3, 8):
      from typing import Literal, TypedDict
  else:
      from typing_extensions import Literal, TypedDict
  
  # Default test duration is 5m for individual tests discovered from L1
  # metadata and 1h for scripts defined directly in plans (L2 metadata).
  DEFAULT_TEST_DURATION_L1 = '5m'
  DEFAULT_TEST_DURATION_L2 = '1h'
  DEFAULT_ORDER = 50
  
  # How many already existing lines should tmt run --follow show
  FOLLOW_LINES = 10
  
  # Obsoleted test keys
  OBSOLETED_TEST_KEYS = "relevancy coverage".split()
  
  # Unofficial temporary test keys
  EXTRA_TEST_KEYS = (
      "extra-nitrate extra-hardware extra-pepa "
      "extra-summary extra-task id".split())
  
  # Unofficial temporary story keys
  EXTRA_STORY_KEYS = ("id".split())
  
  SECTIONS_HEADINGS = {
      'Setup': ['<h1>Setup</h1>'],
      'Test': ['<h1>Test</h1>',
               '<h1>Test .*</h1>'],
      'Step': ['<h2>Step</h2>',
               '<h2>Test Step</h2>'],
      'Expect': ['<h2>Expect</h2>',
                 '<h2>Result</h2>',
                 '<h2>Expected Result</h2>'],
      'Cleanup': ['<h1>Cleanup</h1>']
      }
  
  # Extra keys used for identification in Result class
  EXTRA_RESULT_IDENTIFICATION_KEYS = ['extra-nitrate', 'extra-task']
  
  
  #
  # fmf id types
  #
  # See https://fmf.readthedocs.io/en/latest/concept.html#identifiers for
  # formal specification.
  #
  
  # A "raw" fmf id as stored in fmf node data, i.e. as a mapping with various keys.
  # Used for a brief moment, to annotate raw fmf data before they are converted
  # into FmfId instances.
  _RawFmfId = TypedDict('_RawFmfId', {
      'url': Optional[str],
      'ref': Optional[str],
      'path': Optional[str],
      'name': Optional[str]
  })
  
  
  # An internal fmf id representation.
  @dataclasses.dataclass
  class FmfId(tmt.utils.SerializableContainer):
      # The list of valid fmf id keys
      keys: ClassVar[List[str]] = ['url', 'ref', 'path', 'name']
  
      url: Optional[str] = None
      ref: Optional[str] = None
      path: Optional[str] = None
      name: Optional[str] = None
  
      def to_dict(self) -> Dict[str, Any]:
          """ Return keys and values in the form of a dictionary """
  
          return dataclasses.asdict(self)
  
      def to_raw(self) -> Dict[str, Any]:
          """ Return keys and values as if they originated from fmf node """
  
          return self.to_dict()
  
      @classmethod
      def from_dict(cls, raw: _RawFmfId) -> 'FmfId':
          """ Construct an :py:class:`FmfId` from given input container """
  
          return FmfId(**{key: raw.get(key, None) for key in cls.keys})
  
      def validate(self) -> Tuple[bool, str]:
          """
          Validate fmf id and return a human readable error
  
          Return a tuple (boolean, message) as the result of validation.
          The boolean specifies the validation result and the message
          the validation error. In case the FMF id is valid, return an empty
          string as the message.
          """
          # Validate remote id and translate to human readable errors
          try:
              # Simple asdict() is not good enough, fmf does not like keys that exist but are `None`.
              # Don't include those.
              fmf.base.Tree.node({
                  key: value for key, value in self.to_dict().items()
                  if value is not None
                  })
          except fmf.utils.GeneralError as error:
              # Map fmf errors to more user friendly alternatives
              error_map: List[Tuple[str, str]] = [
                  ('git clone', f"repo '{self.url}' cannot be cloned"),
                  ('git checkout', f"git ref '{self.ref}' is invalid"),
                  ('directory path', f"path '{self.path}' is invalid"),
                  ('tree root', f"No tree found in repo '{self.url}', missing an '.fmf' directory?")
                  ]
  
              stringified_error = str(error)
  
              errors = [message for needle, message in error_map if needle in stringified_error]
              return (False, errors[0] if errors else stringified_error)
  
          return (True, '')
  
  #
  # Various types describing constructs as stored in "raw" fmf node data.
  # Used for a brief moment, to annotate raw fmf data before they are converted
  # into their internal representations.
  #
  
  #
  # A type describing the raw form of the core `link` attribute. See
  # https://tmt.readthedocs.io/en/stable/spec/core.html#link for its
  # formal specification. Internally, a link is represented by a `Link`
  # class instance, and types below describe the raw data coming from Fmf
  # nodes and CLI options.
  
  
  # Link relations.
  _RawLinkRelationName = Literal[
      'verifies', 'verified-by',
      'implements', 'implemented-by',
      'documents', 'documented-by',
      'blocks', 'blocked-by',
      'duplicates', 'duplicated-by',
      'parent', 'child',
      'relates',
      # Special case: not a relation, but it can appear where relations appear in
      # link data structures.
      'note'
  ]
  
  # Link target - can be either a string (like test case name or URL), or an fmf id.
  _RawLinkTarget = Union[str, _RawFmfId]
  
  # Basic "relation-aware" link - essentialy a mapping with one key/value pair.
  _RawLinkRelation = Dict[_RawLinkRelationName, _RawLinkTarget]
  
  # A single link can be represented as a string or FMF ID (meaning only target is specified),
  # or a "relation-aware" link aka mapping defined above.
  _RawLink = Union[
      str,
      _RawFmfId,
      _RawLinkRelation
  ]
  
  # Collection of links - can be either a single link, or a list of links, and all
  # link forms may be used together.
  _RawLinks = Union[
      _RawLink,
      List[_RawLink]
  ]
  
  
  # A type describing `adjust` content. See
  # https://tmt.readthedocs.io/en/stable/spec/core.html#adjust for its formal specification.
  #
  # The type is incomplete in the sense it does not describe all keys it may contain,
  # like `environment` or `provision`, and focuses only on the keys defined for `adjust`
  # itself.
  _RawAdjustRule = TypedDict(
      'AdjustRuleType',
      {
          'when': Optional[str],
          'continue': Optional[bool],
          'because': Optional[str]
          }
  )
  
  # A type describing content accepted by various require-like keys: - a string, fmf id,
  # or a list with a mixture of these two types.
  #
  # TODO: what other keys accept this kind of fmf data??
  _RawRequire = Union[
      str,
      _RawFmfId,
      List[Union[str, _RawFmfId]]
  ]
  
  
  class Core(tmt.utils.Common):
      """
      General node object
  
      Corresponds to given fmf.Tree node.
      Implements common Test, Plan and Story methods.
      Also defines L0 metadata and its manipulation.
      """
  
      # Core attributes (supported across all levels)
      summary: Optional[str] = None
      description: Optional[str] = None
      enabled: bool = True
      order: int = DEFAULT_ORDER
      link: Optional['Links'] = None
      id: Optional[str] = None
      tag: List[str] = []
      tier: Optional[str] = None
      adjust: Optional[List[_RawAdjustRule]] = None
  
      KEYS_SHOW_ORDER = [
          # Basic stuff
          'summary',
          'description',
          'enabled',
          'order',
          'id',
  
          # Filtering and more
          'tag',
          'tier',
          'link',
          'adjust',
          ]
  
      # Normalization methods
      _normalize_tag = tmt.utils.LoadFmfKeysMixin._normalize_string_list
  
      # TODO: remove when schema becomes mandatory, `order` shall never be `null`
      def _normalize_order(self, value: Optional[int]) -> int:
          if value is None:
              return DEFAULT_ORDER
          return int(value)
  
      def _normalize_link(self, value: _RawLinks) -> 'Links':
          return Links(data=value)
  
      def _normalize_adjust(
              self, value: Union[_RawAdjustRule, List[_RawAdjustRule]]) -> List[_RawAdjustRule]:
          if value is None:
              return []
          return [value] if not isinstance(value, list) else value
  
      def _normalize_tier(self, value: Optional[Union[int, str]]) -> Optional[str]:
          if value is None:
              return None
          return str(value)
  
      def __init__(self,
                   *,
                   node: fmf.Tree,
                   parent: Optional[tmt.utils.Common] = None,
                   **kwargs: Any) -> None:
          """ Initialize the node """
          kwargs.set
  
  [... the rest is omitted ...]
files: tmt/base.py
files: 
  """ Step Classes """
  
  import dataclasses
  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)
  
  
  _RawStepData = TypedDict('_RawStepData', {
      'how': str,
      'name': str
  }, total=False)
  
  RawStepDataArgument = Union[_RawStepData, List[_RawStepData]]
  
  
  T = TypeVar('T', bound='StepData')
  
  
  @dataclasses.dataclass
  class StepData(
          tmt.utils.SpecBasedContainer,
          tmt.utils.NormalizeKeysMixin,
          tmt.utils.SerializableContainer):
      """
      Keys necessary to describe, create, save and restore a step.
  
      Very basic set of keys shared across all steps.
  
      Provides basic functionality for transition between "raw" step data, which
      consists of fmf nodes and CLI options, and this container representation with
      keys and types more suitable for internal use.
  
      Implementation expects simple 1:1 relation between ``StepData`` attributes - keys -
      and their fmf/CLI sources, where keys replace options' dashes (``-``) with
      underscores (``_``). For example, to hold value of an fmf key ``foo-bar`` - or
      value of a corresponding CLI option, ``--foo-bar``, a step data class should
      declare key named ``foo_data``. All ``StepData`` methods would honor this mapping.
      """
  
      # TODO: we can easily add lists of keys for various verbosity levels...
      KEYS_SHOW_ORDER = ['name', 'how']
  
      name: str
      how: str
      order: int = tmt.utils.DEFAULT_PLUGIN_ORDER
      summary: Optional[str] = None
  
      def to_spec(self) -> _RawStepData:  # type: ignore[override]
          """ Convert to a form suitable for saving in a specification file """
  
          return cast(_RawStepData, {
              tmt.utils.key_to_option(key): value
              for key, value in self.items()
              })
  
      @classmethod
      def pre_normalization(cls, raw_data: _RawStepData, logger: tmt.utils.Common) -> None:
          """ Called before normalization, useful for tweaking raw data """
  
          logger.debug(f'{cls.__name__}: original raw data', str(raw_data), level=4)
  
      def post_normalization(self, raw_data: _RawStepData, logger: tmt.utils.Common) -> None:
          """ Called after normalization, useful for tweaking normalized data """
  
          pass
  
      @classmethod
      def from_spec(  # type: ignore[override]
              cls: Type[T],
              raw_data: _RawStepData,
              logger: tmt.utils.Common
              ) -> T:
          """ Convert from a specification file or from a CLI option """
  
          cls.pre_normalization(raw_data, logger)
  
          data = cls(name=raw_data['name'], how=raw_data['how'])
          data._load_keys(cast(Dict[str, Any], raw_data), cls.__name__, logger)
  
          data.post_normalization(raw_data, logger)
  
          return data
  
  
  @dataclasses.dataclass
  class WhereableStepData:
      """
      Keys shared by step data that may be limited to a particular guest.
  
      To be used as a mixin class, adds necessary keys.
  
      See [1] and [2] for specification.
  
      1. https://tmt.readthedocs.io/en/stable/spec/plans.html#where
      2. https://tmt.readthedocs.io/en/stable/spec/plans.html#spec-plans-prepare-where
      """
  
      where: Optional[str] = None
  
  
  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'
  
      # Refers to a base class for all plugins registered with this step.
      _plugin_base_class: Type['BasePlugin']
  
      #: Stores the normalized step data. Initialized first time step's `data`
      #: is accessed.
      #
      # The delayed initialization is necessary to support `how` changes via
      # command-line - code instantiating steps must be able to invalidate
      # and replace raw step data entries before they get normalized and become
      # the single source of information for plugins involved.
      _data: List[StepData]
  
      #: Stores the original raw step data. Initialized by :py:meth:`__init__`
      #: or :py:meth:`wake`, and serves as a source for normalization performed
      #: by :py:meth:`_normalize_data`.
      _raw_data: List[_RawStepData]
  
      def __init__(
              self,
              plan: 'Plan',
              data: Optional[RawStepDataArgument] = 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 raw data to be a list of step configuration data, one item per
          # distinct step configuration. Make sure all items have `name`` and `how` keys.
          #
          # NOTE: this is not a normalization step as performed by NormalizeKeysMixin.
          # Here we make sure the raw data can be consumed by the delegation code, we
          # do not modify any existing content of raw data items.
  
          # Create an empty step by default (can be updated from cli)
          if data is None:
              self._raw_data = [{}]
  
          # Convert to list if only a single config provided
          elif isinstance(data, dict):
              self._raw_data = [data]
  
          # List is as good as it gets
          elif isinstance(data, list):
              self._raw_data = data
  
          # Shout about invalid configuration
          else:
              raise tmt.utils.GeneralError(
                  f"Invalid '{self}' config in '{self.plan}'.")
  
          for i, raw_datum in enumerate(self._raw_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 raw_datum.get('name', None) is None:
                  raw_datum['name'] = f'{tmt.utils.DEFAULT_NAME}-{i}'
  
              # Set 'how' to the default if not specified
              if raw_datum.get('how', None) is None:
                  raw_datum['how'] = self.DEFAULT_HOW
  
      def _normalize_data(self, raw_data: List[_RawStepData]) -> List[StepData]:
          """
          Normalize step data entries.
  
          Every entry of ``raw_data`` is converted into an instance of
          :py:class:`StepData` or one of its subclasses. Particular class
          is derived from a plugin identified by raw data's ``how`` field
          and step's plugin registry.
          """
  
          data: List[StepData] = []
  
          for raw_datum in raw_data:
              plugin = self._plugin_base_class.delegate(self, raw_data=raw_datum)
  
              data.append(plugin.data)
  
          return data
  
      @property
      def data(self) -> List[StepData]:
          if not hasattr(self, '_data'):
              self._data = self._normalize_data(self._raw_data)
  
          return self._data
  
      @data.setter
      def data(self, data: List[StepData]) -> None:
          self._data = data
  
      @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
  
  [... the rest is omitted ...]
files: tmt/steps/__init__.py
files: 
  """ Step Classes """
  
  import dataclasses
  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)
  
  
  _RawStepData = TypedDict('_RawStepData', {
      'how': str,
      'name': str
  }, total=False)
  
  RawStepDataArgument = Union[_RawStepData, List[_RawStepData]]
  
  
  T = TypeVar('T', bound='StepData')
  
  
  @dataclasses.dataclass
  class StepData(tmt.utils.NormalizeKeysMixin, tmt.utils.SerializableContainer):
      """
      Keys necessary to describe, create, save and restore a step.
  
      Very basic set of keys shared across all steps.
  
      Provides basic functionality for transition between "raw" step data, which
      consists of fmf nodes and CLI options, and this container representation with
      keys and types more suitable for internal use.
  
      Implementation expects simple 1:1 relation between ``StepData`` attributes - keys -
      and their fmf/CLI sources, where keys replace options' dashes (``-``) with
      underscores (``_``). For example, to hold value of an fmf key ``foo-bar`` - or
      value of a corresponding CLI option, ``--foo-bar``, a step data class should
      declare key named ``foo_data``. All ``StepData`` methods would honor this mapping.
      """
  
      # TODO: we can easily add lists of keys for various verbosity levels...
      KEYS_SHOW_ORDER = ['name', 'how']
  
      name: str
      how: str
      order: int = tmt.utils.DEFAULT_PLUGIN_ORDER
      summary: Optional[str] = None
  
      def to_raw(self) -> _RawStepData:
          """
          Serialize step data instance to a raw representation.
  
          The returned value can be used to recreate step data when given
          to :py:meth:`from_raw`.
          """
  
          return cast(_RawStepData, {
              tmt.utils.key_to_option(key): value
              for key, value in self.items()
              })
  
      @classmethod
      def pre_normalization(cls, raw_data: _RawStepData, logger: tmt.utils.Common) -> None:
          """ Called before normalization, useful for tweaking raw data """
  
          logger.debug(f'{cls.__name__}: original raw data', str(raw_data), level=4)
  
      def post_normalization(self, raw_data: _RawStepData, logger: tmt.utils.Common) -> None:
          """ Called after normalization, useful for tweaking normalized data """
  
          pass
  
      @classmethod
      def from_raw(cls: Type[T], raw_data: _RawStepData, logger: tmt.utils.Common) -> T:
          """
          Unserialize step data instance from its a raw representation.
          """
  
          cls.pre_normalization(raw_data, logger)
  
          data = cls(name=raw_data['name'], how=raw_data['how'])
          data._load_keys(cast(Dict[str, Any], raw_data), cls.__name__, logger)
  
          data.post_normalization(raw_data, logger)
  
          return data
  
  
  @dataclasses.dataclass
  class WhereableStepData:
      """
      Keys shared by step data that may be limited to a particular guest.
  
      To be used as a mixin class, adds necessary keys.
  
      See [1] and [2] for specification.
  
      1. https://tmt.readthedocs.io/en/stable/spec/plans.html#where
      2. https://tmt.readthedocs.io/en/stable/spec/plans.html#spec-plans-prepare-where
      """
  
      where: Optional[str] = None
  
  
  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'
  
      # Refers to a base class for all plugins registered with this step.
      _plugin_base_class: Type['BasePlugin']
  
      #: Stores the normalized step data. Initialized first time step's `data`
      #: is accessed.
      #
      # The delayed initialization is necessary to support `how` changes via
      # command-line - code instantiating steps must be able to invalidate
      # and replace raw step data entries before they get normalized and become
      # the single source of information for plugins involved.
      _data: List[StepData]
  
      #: Stores the original raw step data. Initialized by :py:meth:`__init__`
      #: or :py:meth:`wake`, and serves as a source for normalization performed
      #: by :py:meth:`_normalize_data`.
      _raw_data: List[_RawStepData]
  
      def __init__(
              self,
              plan: 'Plan',
              data: Optional[RawStepDataArgument] = 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 raw data to be a list of step configuration data, one item per
          # distinct step configuration. Make sure all items have `name`` and `how` keys.
          #
          # NOTE: this is not a normalization step as performed by NormalizeKeysMixin.
          # Here we make sure the raw data can be consumed by the delegation code, we
          # do not modify any existing content of raw data items.
  
          # Create an empty step by default (can be updated from cli)
          if data is None:
              self._raw_data = [{}]
  
          # Convert to list if only a single config provided
          elif isinstance(data, dict):
              self._raw_data = [data]
  
          # List is as good as it gets
          elif isinstance(data, list):
              self._raw_data = data
  
          # Shout about invalid configuration
          else:
              raise tmt.utils.GeneralError(
                  f"Invalid '{self}' config in '{self.plan}'.")
  
          for i, raw_datum in enumerate(self._raw_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 raw_datum.get('name', None) is None:
                  raw_datum['name'] = f'{tmt.utils.DEFAULT_NAME}-{i}'
  
              # Set 'how' to the default if not specified
              if raw_datum.get('how', None) is None:
                  raw_datum['how'] = self.DEFAULT_HOW
  
      def _normalize_data(self, raw_data: List[_RawStepData]) -> List[StepData]:
          """
          Normalize step data entries.
  
          Every entry of ``raw_data`` is converted into an instance of
          :py:class:`StepData` or one of its subclasses. Particular class
          is derived from a plugin identified by raw data's ``how`` field
          and step's plugin registry.
          """
  
          data: List[StepData] = []
  
          for raw_datum in raw_data:
              plugin = self._plugin_base_class.delegate(self, raw_data=raw_datum)
  
              data.append(plugin.data)
  
          return data
  
      @property
      def data(self) -> List[StepData]:
          if not hasattr(self, '_data'):
              self._data = self._normalize_data(self._raw_data)
  
          return self._data
  
      @data.setter
      def data(self, data: List[StepData]) -> None:
          self._data = data
  
      @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
  
  [... the rest is omitted ...]
files: tmt/steps/__init__.py
files: import dataclasses
  import glob
  import os
  import re
  import shutil
  import subprocess
  from typing import List, Optional, cast
  
  import click
  import fmf
  
  import tmt
  import tmt.base
  import tmt.beakerlib
  import tmt.options
  import tmt.steps
  import tmt.steps.discover
  import tmt.utils
  
  
  @dataclasses.dataclass
  class DiscoverFmfStepData(tmt.steps.discover.DiscoverStepData):
      url: Optional[str] = None
      ref: Optional[str] = None
      path: Optional[str] = None
      test: List[str] = dataclasses.field(default_factory=list)
      link: List[str] = dataclasses.field(default_factory=list)
      filter: List[str] = dataclasses.field(default_factory=list)
      modified_only: bool = False
      modified_url: Optional[str] = None
      modified_ref: Optional[str] = None
      dist_git_init: bool = False
      dist_git_remove_fmf_root: bool = False
      dist_git_merge: bool = False
      dist_git_extract: Optional[str] = None
      fmf_id: bool = False
      exclude: List[str] = dataclasses.field(default_factory=list)
  
      # Legacy fields
      repository: Optional[str] = None
      revision: Optional[str] = None
  
      _normalize_test = tmt.utils.NormalizeKeysMixin._normalize_string_list
      _normalize_link = tmt.utils.NormalizeKeysMixin._normalize_string_list
      _normalize_filter = tmt.utils.NormalizeKeysMixin._normalize_string_list
      _normalize_exclude = tmt.utils.NormalizeKeysMixin._normalize_string_list
  
      def post_normalization(self, raw_data: tmt.steps._RawStepData,
                             logger: tmt.utils.Common) -> None:
          super().post_normalization(raw_data, logger)
  
          if self.repository:
              self.url = self.repository
  
          if self.revision:
              self.ref = self.revision
  
  
  @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.
      """
  
      _data_class = DiscoverFmfStepData
  
      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: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for given method """
          options = super().options(how)
          options[:0] = [
              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', '--path', metavar='ROOT',
                  help='Path to the metadata tree root.'),
              cls.TEST_OPTION,
              click.option(
                  '--link', metavar="RELATION:TARGET", multiple=True,
                  help="Filter by linked objects (regular expressions are "
                       "supported for both relation and target)."),
              cls.FILTER_OPTION,
              cls.EXCLUDE_OPTION,
              click.option(
                  '--fmf-id', default=False, is_flag=True,
                  help='Show fmf identifiers for tests discovered in plan.'),
              click.option(
                  '--sync-repo', default=False, is_flag=True,
                  help='Force the sync of the whole git repo. By default, the '
                       'repo is copied only if the used options require it.'),
              click.option(
                  '--dist-git-init', is_flag=True,
                  help='Initialize fmf root inside extracted sources '
                       '(at dist-git-extract or top directory).'),
              click.option(
                  '--dist-git-remove-fmf-root', is_flag=True,
                  help='Remove fmf root from extracted source '
                       '(top one or selected by copy-path, '
                       'happens before dist-git-extract.'),
              click.option(
                  '--dist-git-merge', is_flag=True,
                  help='Merge copied sources and plan fmf root.'),
              click.option(
                  '--dist-git-extract',
                  help='What to copy from extracted sources, globbing is '
                       'supported. Defaults to the top fmf root if it is '
                       'present, otherwise top directory (shortcut "/").'
                  ),
              ]
          return options
  
      @property
      def is_in_standalone_mode(self) -> bool:
          """ Enable standalone mode when listing fmf ids """
          if self.opt('fmf_id'):
              return True
          return super().is_in_standalone_mode
  
      def go(self) -> None:
          """ Discover available tests """
          super(DiscoverFmf, self).go()
  
          # Check url and path, prepare test directory
          url = self.get('url')
          path = self.get('path')
          # Save the test directory so that others can reference it
          ref = self.get('ref')
          assert self.workdir is not None
          self.testdir = os.path.join(self.workdir, 'tests')
          sourcedir = os.path.join(self.workdir, 'source')
          dist_git_source = self.get('dist-git-source', False)
          dist_git_merge = self.get('dist-git-merge', False)
          dist_git_init = self.get('dist-git-init', False)
          dist_git_extract = self.get('dist-git-extract', None)
          dist_git_remove_fmf_root = self.get('dist-git-remove-fmf-root', False)
  
          # Self checks
          if dist_git_source and not dist_git_merge and (ref or url):
              raise tmt.utils.DiscoverError(
                  "Cannot manipulate with dist-git without "
                  "the `--dist-git-merge` option.")
  
          def get_git_root(dir: str) -> str:
              stdout, _ = self.run(["git", "rev-parse", "--show-toplevel"], cwd=dir, dry=True)
              assert stdout is not None
              return stdout.strip("\n")
  
          # Raise an exception if --fmf-id uses w/o url and git root
          # doesn't exist for discovered plan
          if self.opt('fmf_id'):
              def assert_git_url(plan_name: Optional[str] = None) -> None:
  
  
  [... the rest is omitted ...]
files: tmt/steps/discover/fmf.py
files: import dataclasses
  import glob
  import os
  import re
  import shutil
  import subprocess
  from typing import List, Optional, cast
  
  import click
  import fmf
  
  import tmt
  import tmt.base
  import tmt.beakerlib
  import tmt.options
  import tmt.steps
  import tmt.steps.discover
  import tmt.utils
  
  
  @dataclasses.dataclass
  class DiscoverFmfStepData(tmt.steps.discover.DiscoverStepData):
      url: Optional[str] = None
      ref: Optional[str] = None
      path: Optional[str] = None
      test: List[str] = dataclasses.field(default_factory=list)
      link: List[str] = dataclasses.field(default_factory=list)
      filter: List[str] = dataclasses.field(default_factory=list)
      modified_only: bool = False
      modified_url: Optional[str] = None
      modified_ref: Optional[str] = None
      dist_git_init: bool = False
      dist_git_remove_fmf_root: bool = False
      dist_git_merge: bool = False
      dist_git_extract: Optional[str] = None
      fmf_id: bool = False
      exclude: List[str] = dataclasses.field(default_factory=list)
  
      # Legacy fields
      repository: Optional[str] = None
      revision: Optional[str] = None
  
      _normalize_test = tmt.utils.NormalizeKeysMixin._normalize_string_list
      _normalize_link = tmt.utils.NormalizeKeysMixin._normalize_string_list
      _normalize_filter = tmt.utils.NormalizeKeysMixin._normalize_string_list
      _normalize_exclude = tmt.utils.NormalizeKeysMixin._normalize_string_list
  
      def post_normalization(self, raw_data: tmt.steps._RawStepData,
                             logger: tmt.utils.Common) -> None:
          super().post_normalization(raw_data, logger)
  
          if self.repository:
              self.url = self.repository
  
          if self.revision:
              self.ref = self.revision
  
  
  @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.
      """
  
      _data_class = DiscoverFmfStepData
  
      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: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for given method """
          options = super().options(how)
          options[:0] = [
              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', '--path', metavar='ROOT',
                  help='Path to the metadata tree root.'),
              cls.TEST_OPTION,
              click.option(
                  '--link', metavar="RELATION:TARGET", multiple=True,
                  help="Filter by linked objects (regular expressions are "
                       "supported for both relation and target)."),
              cls.FILTER_OPTION,
              cls.EXCLUDE_OPTION,
              click.option(
                  '--fmf-id', default=False, is_flag=True,
                  help='Show fmf identifiers for tests discovered in plan.'),
              click.option(
                  '--sync-repo', default=False, is_flag=True,
                  help='Force the sync of the whole git repo. By default, the '
                       'repo is copied only if the used options require it.'),
              click.option(
                  '--dist-git-init', is_flag=True,
                  help='Initialize fmf root inside extracted sources '
                       '(at dist-git-extract or top directory).'),
              click.option(
                  '--dist-git-remove-fmf-root', is_flag=True,
                  help='Remove fmf root from extracted source '
                       '(top one or selected by copy-path, '
                       'happens before dist-git-extract.'),
              click.option(
                  '--dist-git-merge', is_flag=True,
                  help='Merge copied sources and plan fmf root.'),
              click.option(
                  '--dist-git-extract',
                  help='What to copy from extracted sources, globbing is '
                       'supported. Defaults to the top fmf root if it is '
                       'present, otherwise top directory (shortcut "/").'
                  ),
              ]
          return options
  
      @property
      def is_in_standalone_mode(self) -> bool:
          """ Enable standalone mode when listing fmf ids """
          if self.opt('fmf_id'):
              return True
          return super().is_in_standalone_mode
  
      def go(self) -> None:
          """ Discover available tests """
          super(DiscoverFmf, self).go()
  
          # Check url and path, prepare test directory
          url = self.get('url')
          path = self.get('path')
          # Save the test directory so that others can reference it
          ref = self.get('ref')
          assert self.workdir is not None
          self.testdir = os.path.join(self.workdir, 'tests')
          sourcedir = os.path.join(self.workdir, 'source')
          dist_git_source = self.get('dist-git-source', False)
          dist_git_merge = self.get('dist-git-merge', False)
          dist_git_init = self.get('dist-git-init', False)
          dist_git_extract = self.get('dist-git-extract', None)
          dist_git_remove_fmf_root = self.get('dist-git-remove-fmf-root', False)
  
          # Self checks
          if dist_git_source and not dist_git_merge and (ref or url):
              raise tmt.utils.DiscoverError(
                  "Cannot manipulate with dist-git without "
                  "the `--dist-git-merge` option.")
  
          def get_git_root(dir: str) -> str:
              stdout, _ = self.run(["git", "rev-parse", "--show-toplevel"], cwd=dir, dry=True)
              assert stdout is not None
              return stdout.strip("\n")
  
          # Raise an exception if --fmf-id uses w/o url and git root
          # doesn't exist for discovered plan
          if self.opt('fmf_id'):
              def assert_git_url(plan_name: Optional[str] = None) -> None:
  
  
  [... the rest is omitted ...]
files: tmt/steps/discover/fmf.py
files: import copy
  import dataclasses
  import os
  from typing import (TYPE_CHECKING, Any, Dict, List, Optional, Type, TypeVar,
                      Union, cast)
  
  import click
  import fmf
  
  import tmt
  import tmt.base
  import tmt.steps
  import tmt.steps.discover
  import tmt.utils
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  T = TypeVar('T', bound='TestDescription')
  
  
  @dataclasses.dataclass
  class TestDescription(
          tmt.utils.SpecBasedContainer,
          tmt.utils.NormalizeKeysMixin,
          tmt.utils.SerializableContainer):
      """
      Keys necessary to describe a shell-based test.
  
      Provides basic functionality for tansition between "raw" step data representation,
      which consists of keys and values given by fmf tree and CLI options, and this
      container representation for internal use.
      """
  
      name: str
  
      # TODO: following keys are copy & pasted from base.Test. It would be much, much better
      # to re-use the definitions from base.Test instead copying them here, but base.Test
      # does not support save/load operations. This is a known issue, introduced by a patch
      # transitioning step data to data classes, it is temporary, and it will be fixed as
      # soon as possible - nobody want's to keep two very same lists of attributes.
      test: str
  
      # Core attributes (supported across all levels)
      summary: Optional[str] = None
      description: Optional[str] = None
      enabled: bool = True
      # TODO: ugly circular dependency (see tmt.base.DEFAULT_ORDER)
      order: int = 50
      link: Optional['tmt.base.Links'] = None
      id: Optional[str] = None
      tag: List[str] = dataclasses.field(default_factory=list)
      tier: Optional[str] = None
      adjust: Optional[List['tmt.base._RawAdjustRule']] = None
  
      # Basic test information
      contact: List[str] = dataclasses.field(default_factory=list)
      component: List[str] = dataclasses.field(default_factory=list)
  
      # Test execution data
      path: Optional[str] = None
      framework: Optional[str] = None
      manual: bool = False
      require: List[Union[str, 'tmt.base.FmfId']] = dataclasses.field(default_factory=list)
      recommend: List[str] = dataclasses.field(default_factory=list)
      environment: tmt.utils.EnvironmentType = dataclasses.field(default_factory=dict)
      duration: str = '1h'
      result: str = 'respect'
  
      # We can't even re-use normalization callbacks from tmt.base because of the cyclic import :/
      _normalize_tag = tmt.utils.LoadFmfKeysMixin._normalize_string_list
      _normalize_contact = tmt.utils.LoadFmfKeysMixin._normalize_string_list
      _normalize_component = tmt.utils.LoadFmfKeysMixin._normalize_string_list
      _normalize_recommend = tmt.utils.LoadFmfKeysMixin._normalize_string_list
  
      def _normalize_order(self, value: Optional[int]) -> int:
          if value is None:
              # TODO: ugly circular dependency (see tmt.base.DEFAULT_ORDER)
              return 50
          return int(value)
  
      def _normalize_link(self, value: 'tmt.base._RawLinks') -> 'tmt.base.Links':
          import tmt.base
  
          return tmt.base.Links(data=value)
  
      def _normalize_adjust(self,
                            value: Optional[Union[
                                'tmt.base._RawAdjustRule',
                                List['tmt.base._RawAdjustRule']]]
                            ) -> List['tmt.base._RawAdjustRule']:
          if value is None:
              return []
          return [value] if not isinstance(value, list) else value
  
      def _normalize_tier(self, value: Optional[Union[int, str]]) -> Optional[str]:
          if value is None:
              return None
          return str(value)
  
      def _normalize_require(
              self, value: Optional['tmt.base._RawRequire']) -> List[Union[str, 'tmt.base.FmfId']]:
          if value is None:
              return []
          # TODO: remove ignore when base.py becomes annotated
          return [value] if isinstance(value, str) else value  # type: ignore[no-any-return]
  
      # Our own implementation, parent uses `name` and `how`, and tests don't have any `how`.
      @classmethod
      def from_spec(  # type: ignore[override]
              cls: Type[T],
              raw_data: Dict[str, Any],
              logger: tmt.utils.Common
              ) -> T:
          """ Convert from a specification file or from a CLI option """
  
          data = cls(name=raw_data['name'], test=raw_data['test'])
          data._load_keys(raw_data, cls.__name__, logger)
  
          return data
  
      def to_spec(self) -> Dict[str, Any]:
          """ Convert to a form suitable for saving in a specification file """
  
          data = super().to_spec()
          data['link'] = self.link.to_spec() if self.link else None
          data['require'] = [
              require if isinstance(require, str) else require.to_spec()
              for require in self.require
              ]
  
          return data
  
      def to_serialized(self) -> Dict[str, Any]:
          """ Convert to a form suitable for saving in a file """
  
          data = super().to_serialized()
  
          # Using `to_spec()` on purpose: `Links` does not provide serialization
          # methods, because specification of links is already good enough. We
          # can use existing `to_spec()` method, and undo it with a simple
          # `Links(...)` call.
          data['link'] = self.link.to_spec() if self.link else None
          data['require'] = [
              require if isinstance(require, str) else require.to_serialized()
              for require in self.require
              ]
  
          return data
  
      @classmethod
      def from_serialized(cls, serialized: Dict[str, Any]) -> 'TestDescription':
          """ Convert from a serialized form loaded from a file """
  
          obj = super().from_serialized(serialized)
          obj.link = tmt.base.Links(serialized['link'])
          obj.require = [
              require if isinstance(require, str) else tmt.base.FmfId.from_serialized(require)
              for require in serialized['require']
              ]
  
          return obj
  
  
  @dataclasses.dataclass
  class DiscoverShellData(tmt.steps.discover.DiscoverStepData):
      tests: List[TestDescription] = dataclasses.field(default_factory=list)
  
      def _normalize_tests(self, value: List[Dict[str, Any]]
                           ) -> List[TestDescription]:
          return [TestDescription.from_spec(raw_datum, tmt.utils.Common()) for raw_datum in value]
  
      def to_serialized(self) -> Dict[str, Any]:
          """ Convert to a form suitable for saving in a file """
  
          serialized = super().to_serialized()
  
          serialized['tests'] = [test.to_serialized() for test in self.tests]
  
          return serialized
  
      @classmethod
      def from_serialized(cls, serialized: Dict[str, Any]) -> 'DiscoverShellData':
          """ Convert from a serialized form loaded from a file """
  
          obj = super().from_serialized(serialized)
  
          obj.tests = [TestDescription.from_serialized(
              serialized_test) for serialized_test in serialized['tests']]
  
          return obj
  
  
  @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
      """
  
      _data_class = DiscoverShellData
  
      _tests: List[tmt.base.Test] = []
  
      def show(self, keys: Optional[List[str]] = None) -> None:
          """ Show config details """
          super().show([])
          tests = cast(List[TestDescription], self.get('tests'))
          if tests:
              test_names = [test.name for test in tests]
              click.echo(tmt.utils.format('tests', test_names))
  
      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
          for data in cast(DiscoverShellData, 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
              # TODO: can this ever happen? With annotations, `name: str` and `test: str`, nothing
              # should ever assign `None` there and pass the test.
              if not data.name:
                  raise tmt.utils.SpecificationError(
                      f"Missing test name in '{self.step.plan.name}'.")
              # Make sure that the test script is defined
              if not data.test:
                  raise tmt.utils.SpecificationError(
                      f"Missing test script in '{self.step.plan.name}'.")
              #
  
  [... the rest is omitted ...]
files: tmt/steps/discover/shell.py
files: import copy
  import dataclasses
  import os
  from typing import (TYPE_CHECKING, Any, Dict, List, Optional, Type, TypeVar,
                      Union, cast)
  
  import click
  import fmf
  
  import tmt
  import tmt.base
  import tmt.steps
  import tmt.steps.discover
  import tmt.utils
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  T = TypeVar('T', bound='TestDescription')
  
  
  @dataclasses.dataclass
  class TestDescription(tmt.utils.NormalizeKeysMixin, tmt.utils.SerializableContainer):
      """
      Keys necessary to describe a shell-based test.
  
      Provides basic functionality for tansition between "raw" step data representation,
      which consists of keys and values given by fmf tree and CLI options, and this
      container representation for internal use.
      """
  
      name: str
  
      # TODO: following keys are copy & pasted from base.Test. It would be much, much better
      # to re-use the definitions from base.Test instead copying them here, but base.Test
      # does not support save/load operations. This is a known issue, introduced by a patch
      # transitioning step data to data classes, it is temporary, and it will be fixed as
      # soon as possible - nobody want's to keep two very same lists of attributes.
      test: str
  
      # Core attributes (supported across all levels)
      summary: Optional[str] = None
      description: Optional[str] = None
      enabled: bool = True
      # TODO: ugly circular dependency (see tmt.base.DEFAULT_ORDER)
      order: int = 50
      link: Optional['tmt.base.Links'] = None
      id: Optional[str] = None
      tag: List[str] = dataclasses.field(default_factory=list)
      tier: Optional[str] = None
      adjust: Optional[List['tmt.base._RawAdjustRule']] = None
  
      # Basic test information
      contact: List[str] = dataclasses.field(default_factory=list)
      component: List[str] = dataclasses.field(default_factory=list)
  
      # Test execution data
      path: Optional[str] = None
      framework: Optional[str] = None
      manual: bool = False
      require: List[Union[str, 'tmt.base.FmfId']] = dataclasses.field(default_factory=list)
      recommend: List[str] = dataclasses.field(default_factory=list)
      environment: tmt.utils.EnvironmentType = dataclasses.field(default_factory=dict)
      duration: str = '1h'
      result: str = 'respect'
  
      # We can't even re-use normalization callbacks from tmt.base because of the cyclic import :/
      _normalize_tag = tmt.utils.LoadFmfKeysMixin._normalize_string_list
      _normalize_contact = tmt.utils.LoadFmfKeysMixin._normalize_string_list
      _normalize_component = tmt.utils.LoadFmfKeysMixin._normalize_string_list
      _normalize_recommend = tmt.utils.LoadFmfKeysMixin._normalize_string_list
  
      def _normalize_order(self, value: Optional[int]) -> int:
          if value is None:
              # TODO: ugly circular dependency (see tmt.base.DEFAULT_ORDER)
              return 50
          return int(value)
  
      def _normalize_link(self, value: 'tmt.base._RawLinks') -> 'tmt.base.Links':
          import tmt.base
  
          return tmt.base.Links(data=value)
  
      def _normalize_adjust(self,
                            value: Optional[Union[
                                'tmt.base._RawAdjustRule',
                                List['tmt.base._RawAdjustRule']]]
                            ) -> List['tmt.base._RawAdjustRule']:
          if value is None:
              return []
          return [value] if not isinstance(value, list) else value
  
      def _normalize_tier(self, value: Optional[Union[int, str]]) -> Optional[str]:
          if value is None:
              return None
          return str(value)
  
      def _normalize_require(
              self, value: Optional['tmt.base._RawRequire']) -> List[Union[str, 'tmt.base.FmfId']]:
          if value is None:
              return []
          # TODO: remove ignore when base.py becomes annotated
          return [value] if isinstance(value, str) else value  # type: ignore[no-any-return]
  
      # Our own implementation, parent uses `name` and `how`, and tests don't have any `how`.
      @classmethod
      def from_raw(cls: Type[T], raw_data: Dict[str, Any], logger: tmt.utils.Common) -> T:
          """
          Unserialize step data instance from its a raw representation.
          """
  
          data = cls(name=raw_data['name'], test=raw_data['test'])
          data._load_keys(raw_data, cls.__name__, logger)
  
          return data
  
      def to_raw(self) -> Dict[str, Any]:
          data = super().to_dict()
          data['link'] = self.link.to_raw() if self.link else None
          data['require'] = [
              require if isinstance(require, str) else require.to_raw()
              for require in self.require
              ]
  
          return data
  
      def to_serialized(self) -> Dict[str, Any]:
          data = super().to_serialized()
  
          data['link'] = self.link.to_raw() if self.link else None
          data['require'] = [
              require if isinstance(require, str) else require.to_serialized()
              for require in self.require
              ]
  
          return data
  
      @classmethod
      def from_serialized(cls, serialized: Dict[str, Any]) -> 'TestDescription':
          obj = super().from_serialized(serialized)
          obj.link = tmt.base.Links(serialized['link'])
          obj.require = [
              require if isinstance(require, str) else tmt.base.FmfId.from_serialized(require)
              for require in serialized['require']
              ]
  
          return obj
  
  
  @dataclasses.dataclass
  class DiscoverShellData(tmt.steps.discover.DiscoverStepData):
      tests: List[TestDescription] = dataclasses.field(default_factory=list)
  
      def _normalize_tests(self, value: List[Dict[str, Any]]
                           ) -> List[TestDescription]:
          return [TestDescription.from_raw(raw_datum, tmt.utils.Common()) for raw_datum in value]
  
      def to_serialized(self) -> Dict[str, Any]:
          serialized = super().to_serialized()
  
          serialized['tests'] = [test.to_serialized() for test in self.tests]
  
          return serialized
  
      @classmethod
      def from_serialized(cls, serialized: Dict[str, Any]) -> 'DiscoverShellData':
          obj = super().from_serialized(serialized)
  
          obj.tests = [TestDescription.from_serialized(
              serialized_test) for serialized_test in serialized['tests']]
  
          return obj
  
  
  @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
      """
  
      _data_class = DiscoverShellData
  
      _tests: List[tmt.base.Test] = []
  
      def show(self, keys: Optional[List[str]] = None) -> None:
          """ Show config details """
          super().show([])
          tests = cast(List[TestDescription], self.get('tests'))
          if tests:
              test_names = [test.name for test in tests]
              click.echo(tmt.utils.format('tests', test_names))
  
      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
          for data in cast(DiscoverShellData, 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
              # TODO: can this ever happen? With annotations, `name: str` and `test: str`, nothing
              # should ever assign `None` there and pass the test.
              if not data.name:
                  raise tmt.utils.SpecificationError(
                      f"Missing test name in '{self.step.plan.name}'.")
              # Make sure that the test script is defined
              if not data.test:
                  raise tmt.utils.SpecificationError(
                      f"Missing test script in '{self.step.plan.name}'.")
              # Prepare path to the test working directory (tree root by default)
              data.path = f"/tests{data.path}" if data.path else '/tests'
              # Apply default test duration unless provided
              if not data.duration:
                  data.duration = tmt.base.DEFAULT_TEST_DURATION_L2
              # Add source dir path variable
              if dist_git_source:
                  data.environment['TMT_SOURCE_DIR'] = sourcedir
              # Create a simple fmf node, adjust its name
              tests.child(data.name, data.to_raw())
  
          # Symlink tests directory to the plan work tree
          testdir = os.path.join(self.workdir, "tests")
          relative_path = os.path.relpath(self.step.plan
  
  [... the rest is omitted ...]
files: tmt/steps/discover/shell.py
files: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  import types
  from typing import Any, List, Optional, cast
  
  import click
  import fmf
  import requests
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  libvirt: Optional[types.ModuleType] = None
  testcloud: Optional[types.ModuleType] = None
  
  
  def import_testcloud() -> None:
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
    - {public_key}
  ssh_pwauth: true
  disable_root: false
  runcmd:
    - sed -i -e '/^.*PermitRootLogin/s/^.*$/PermitRootLogin yes/'
      -e '/^.*UseDNS/s/^.*$/UseDNS no/'
      -e '/^.*GSSAPIAuthentication/s/^.*$/GSSAPIAuthentication no/'
      /etc/ssh/sshd_config
    - systemctl reload sshd
    - [sh, -c, 'if [ ! -f /etc/systemd/network/20-tc-usernet.network ] &&
    systemctl status systemd-networkd | grep -q "enabled;\\svendor\\spreset:\\senabled";
    then mkdir -p /etc/systemd/network/ &&
    echo "[Match]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "Name=en*" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "[Network]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "DHCP=yes" >> /etc/systemd/network/20-tc-usernet.network; fi']
    - [sh, -c, 'if systemctl status systemd-networkd |
    grep -q "enabled;\\svendor\\spreset:\\senabled"; then
    systemctl restart systemd-networkd; fi']
    - [sh, -c, 'if cat /etc/os-release |
    grep -q platform:el8; then systemctl restart sshd; fi']
  """
  
  COREOS_DATA = """variant: fcos
  version: 1.4.0
  passwd:
    users:
      - name: {user_name}
        ssh_authorized_keys:
          - {public_key}
  """
  
  # Libvirt domain XML template related variables
  DOMAIN_TEMPLATE_NAME = 'domain-template.jinja'
  DOMAIN_TEMPLATE_FILE = os.path.join(TESTCLOUD_DATA, DOMAIN_TEMPLATE_NAME)
  DOMAIN_TEMPLATE = """<domain type='{{ virt_type }}' xmlns:qemu='http://libvirt.org/schemas/domain/qemu/1.0'>
    <name>{{ domain_name }}</name>
    <uuid>{{ uuid }}</uuid>
    <memory unit='KiB'>{{ memory }}</memory>
    <currentMemory unit='KiB'>{{ memory }}</currentMemory>
    <vcpu placement='static'>2</vcpu>
    <os>
      <type arch='{{ arch }}' machine='{{ model }}'>hvm</type>
      {{ uefi_loader }}
      <boot dev='hd'/>
    </os>
    {{ cpu }}
    {{ extra_specs }}
    <clock offset='utc'>
      <timer name='rtc' tickpolicy='catchup'/>
      <timer name='pit' tickpolicy='delay'/>
      <timer name='hpet' present='no'/>
    </clock>
    <on_poweroff>destroy</on_poweroff>
    <on_reboot>restart</on_reboot>
    <on_crash>restart</on_crash>
    <devices>
      <emulator>{{ emulator_path }}</emulator>
      <disk type='file' device='disk'>
        <driver name='qemu' type='qcow2' cache='unsafe'/>
        <source file="{{ disk }}"/>
        <target dev='vda' bus='virtio'/>
        {{ boot_drive_address }}
      </disk>
      <disk type='file' device='disk'>
        <driver name='qemu' type='raw'/>
        <source file="{{ seed }}"/>
        <target dev='vdb' bus='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x08' function='0x0'/>
      </disk>
      <interface type='{{ network_type }}'>
        <mac address="{{ mac_address }}"/>
        {{ network_source }}
        {{ ip_setup }}
        <model type='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>
      </interface>
      <serial type='pty'>
        <target port='0'/>
      </serial>
      <console type='pty'>
        <target type='serial' port='0'/>
      </console>
      <input type="keyboard" bus="virtio"/>
      <rng model='virtio'>
        <backend model='random'>/dev/urandom</backend>
      </rng>
    </devices>
    {{ qemu_args }}
  </domain>
  """  # noqa: E501
  
  # VM defaults
  DEFAULT_BOOT_TIMEOUT = 60      # seconds
  DEFAULT_CONNECT_TIMEOUT = 60   # seconds
  NON_KVM_ADDITIONAL_WAIT = 10       # seconds
  NON_KVM_TIMEOUT_COEF = 10          # times
  
  # SSH key type, set None for ssh-keygen default one
  SSH_KEYGEN_TYPE = "ecdsa"
  
  DEFAULT_USER = 'root'
  DEFAULT_MEMORY = 2048
  DEFAULT_DISK = 10
  DEFAULT_IMAGE = 'fedora'
  DEFAULT_CONNECTION = 'session'
  DEFAULT_ARCH = platform.machine()
  
  
  @dataclasses.dataclass
  class TestcloudGuestData(tmt.steps.provision.GuestSshData):
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      image: str = DEFAULT_IMAGE
      memory: int = DEFAULT_MEMORY
      disk: int = DEFAULT_DISK
      connection: str = DEFAULT_CONNECTION
      arch: str = DEFAULT_ARCH
  
      image_url: Optional[str] = None
      instance_name: Optional[str] = None
  
  
  @dataclasses.dataclass
  class ProvisionTestcloudData(TestcloudGuestData, tmt.steps.StepData):
      pass
  
  
  @tmt.steps.provides_method('virtual.testcloud')
  class ProvisionTestcloud(tmt.steps.provision.ProvisionPlugin):
      """
      Local virtual machine using testcloud
  
      Minimal config which uses the latest fedora image:
  
          provision:
              how: virtual
  
      Here's a full config example:
  
          provision:
              how: virtual
              image: fedora
              user: root
              memory: 2048
  
      As the image use 'fedora' for the latest released Fedora compose,
      'rawhide' for the latest Rawhide compose, short aliases such as
      'fedora-32', 'f-32' or 'f32' for specific release or a full url to
      the qcow2 image for example from:
  
          https://kojipkgs.fedoraproject.org/compose/
  
      Short names are also provided for 'centos', 'centos-stream',
      'debian' and 'ubuntu' (e.g. 'centos-8' or 'c8').
  
      Supported Fedora CoreOS images are:
  
          fedora-coreos
          fedora-coreos-stable
          fedora-coreos-testing
          fedora-coreos-next
  
      Use the full path for images stored on local disk, for example:
  
          /var/tmp/images/Fedora-Cloud-Base-31-1.9.x86_64.qcow2
  
      In addition to the qcow2 format, vagrant boxes can be used as well,
      testcloud will take care of unpacking the image for you.
      """
  
      _data_class = ProvisionTestcloudData
  
      # Guest instance
      _guest = None
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for testcloud """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to be used. Provide a short name, '
                       'full path to a local file or a complete url.'),
              click.option(
                  '-m', '--memory', metavar='MEMORY', type=int,
                  help='Set available memory in MB, 2048 MB by default.'),
              click.option(
                  '-D', '--disk', metavar='MEMORY', type=int,
                  help='Specify disk size in GB, 10 GB by default.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-c', '--connection',
                  type=click.Choice(['session', 'system']),
                  help="What session type to use, 'session' by default."),
              click.option(
                  '-a', '--arch',
                  type=click.Choice(['x86_64', 'aarch64', 's
  
  [... the rest is omitted ...]
files: tmt/steps/provision/testcloud.py
files: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  import types
  from typing import Any, List, Optional, cast
  
  import click
  import fmf
  import requests
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  libvirt: Optional[types.ModuleType] = None
  testcloud: Optional[types.ModuleType] = None
  
  
  def import_testcloud() -> None:
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
    - {public_key}
  ssh_pwauth: true
  disable_root: false
  runcmd:
    - sed -i -e '/^.*PermitRootLogin/s/^.*$/PermitRootLogin yes/'
      -e '/^.*UseDNS/s/^.*$/UseDNS no/'
      -e '/^.*GSSAPIAuthentication/s/^.*$/GSSAPIAuthentication no/'
      /etc/ssh/sshd_config
    - systemctl reload sshd
    - [sh, -c, 'if [ ! -f /etc/systemd/network/20-tc-usernet.network ] &&
    systemctl status systemd-networkd | grep -q "enabled;\\svendor\\spreset:\\senabled";
    then mkdir -p /etc/systemd/network/ &&
    echo "[Match]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "Name=en*" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "[Network]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "DHCP=yes" >> /etc/systemd/network/20-tc-usernet.network; fi']
    - [sh, -c, 'if systemctl status systemd-networkd |
    grep -q "enabled;\\svendor\\spreset:\\senabled"; then
    systemctl restart systemd-networkd; fi']
    - [sh, -c, 'if cat /etc/os-release |
    grep -q platform:el8; then systemctl restart sshd; fi']
  """
  
  COREOS_DATA = """variant: fcos
  version: 1.4.0
  passwd:
    users:
      - name: {user_name}
        ssh_authorized_keys:
          - {public_key}
  """
  
  # Libvirt domain XML template related variables
  DOMAIN_TEMPLATE_NAME = 'domain-template.jinja'
  DOMAIN_TEMPLATE_FILE = os.path.join(TESTCLOUD_DATA, DOMAIN_TEMPLATE_NAME)
  DOMAIN_TEMPLATE = """<domain type='{{ virt_type }}' xmlns:qemu='http://libvirt.org/schemas/domain/qemu/1.0'>
    <name>{{ domain_name }}</name>
    <uuid>{{ uuid }}</uuid>
    <memory unit='KiB'>{{ memory }}</memory>
    <currentMemory unit='KiB'>{{ memory }}</currentMemory>
    <vcpu placement='static'>2</vcpu>
    <os>
      <type arch='{{ arch }}' machine='{{ model }}'>hvm</type>
      {{ uefi_loader }}
      <boot dev='hd'/>
    </os>
    {{ cpu }}
    {{ extra_specs }}
    <clock offset='utc'>
      <timer name='rtc' tickpolicy='catchup'/>
      <timer name='pit' tickpolicy='delay'/>
      <timer name='hpet' present='no'/>
    </clock>
    <on_poweroff>destroy</on_poweroff>
    <on_reboot>restart</on_reboot>
    <on_crash>restart</on_crash>
    <devices>
      <emulator>{{ emulator_path }}</emulator>
      <disk type='file' device='disk'>
        <driver name='qemu' type='qcow2' cache='unsafe'/>
        <source file="{{ disk }}"/>
        <target dev='vda' bus='virtio'/>
        {{ boot_drive_address }}
      </disk>
      <disk type='file' device='disk'>
        <driver name='qemu' type='raw'/>
        <source file="{{ seed }}"/>
        <target dev='vdb' bus='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x08' function='0x0'/>
      </disk>
      <interface type='{{ network_type }}'>
        <mac address="{{ mac_address }}"/>
        {{ network_source }}
        {{ ip_setup }}
        <model type='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>
      </interface>
      <serial type='pty'>
        <target port='0'/>
      </serial>
      <console type='pty'>
        <target type='serial' port='0'/>
      </console>
      <input type="keyboard" bus="virtio"/>
      <rng model='virtio'>
        <backend model='random'>/dev/urandom</backend>
      </rng>
    </devices>
    {{ qemu_args }}
  </domain>
  """  # noqa: E501
  
  # VM defaults
  DEFAULT_BOOT_TIMEOUT = 60      # seconds
  DEFAULT_CONNECT_TIMEOUT = 60   # seconds
  NON_KVM_ADDITIONAL_WAIT = 10       # seconds
  NON_KVM_TIMEOUT_COEF = 10          # times
  
  # SSH key type, set None for ssh-keygen default one
  SSH_KEYGEN_TYPE = "ecdsa"
  
  DEFAULT_USER = 'root'
  DEFAULT_MEMORY = 2048
  DEFAULT_DISK = 10
  DEFAULT_IMAGE = 'fedora'
  DEFAULT_CONNECTION = 'session'
  DEFAULT_ARCH = platform.machine()
  
  
  @dataclasses.dataclass
  class TestcloudGuestData(tmt.steps.provision.GuestSshData):
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      image: str = DEFAULT_IMAGE
      memory: int = DEFAULT_MEMORY
      disk: int = DEFAULT_DISK
      connection: str = DEFAULT_CONNECTION
      arch: str = DEFAULT_ARCH
  
      image_url: Optional[str] = None
      instance_name: Optional[str] = None
  
  
  @dataclasses.dataclass
  class ProvisionTestcloudData(TestcloudGuestData, tmt.steps.StepData):
      pass
  
  
  @tmt.steps.provides_method('virtual.testcloud')
  class ProvisionTestcloud(tmt.steps.provision.ProvisionPlugin):
      """
      Local virtual machine using testcloud
  
      Minimal config which uses the latest fedora image:
  
          provision:
              how: virtual
  
      Here's a full config example:
  
          provision:
              how: virtual
              image: fedora
              user: root
              memory: 2048
  
      As the image use 'fedora' for the latest released Fedora compose,
      'rawhide' for the latest Rawhide compose, short aliases such as
      'fedora-32', 'f-32' or 'f32' for specific release or a full url to
      the qcow2 image for example from:
  
          https://kojipkgs.fedoraproject.org/compose/
  
      Short names are also provided for 'centos', 'centos-stream',
      'debian' and 'ubuntu' (e.g. 'centos-8' or 'c8').
  
      Supported Fedora CoreOS images are:
  
          fedora-coreos
          fedora-coreos-stable
          fedora-coreos-testing
          fedora-coreos-next
  
      Use the full path for images stored on local disk, for example:
  
          /var/tmp/images/Fedora-Cloud-Base-31-1.9.x86_64.qcow2
  
      In addition to the qcow2 format, vagrant boxes can be used as well,
      testcloud will take care of unpacking the image for you.
      """
  
      _data_class = ProvisionTestcloudData
  
      # Guest instance
      _guest = None
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for testcloud """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to be used. Provide a short name, '
                       'full path to a local file or a complete url.'),
              click.option(
                  '-m', '--memory', metavar='MEMORY', type=int,
                  help='Set available memory in MB, 2048 MB by default.'),
              click.option(
                  '-D', '--disk', metavar='MEMORY', type=int,
                  help='Specify disk size in GB, 10 GB by default.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-c', '--connection',
                  type=click.Choice(['session', 'system']),
                  help="What session type to use, 'session' by default."),
              click.option(
                  '-a', '--arch',
                  type=click.Choice(['x86_64', 'aarch64', 's
  
  [... the rest is omitted ...]
files: tmt/steps/provision/testcloud.py
files: 
  """ Test Metadata Utilities """
  
  import contextlib
  import copy
  import dataclasses
  import datetime
  import functools
  import glob
  import io
  import os
  import pprint
  import re
  import shlex
  import shutil
  import subprocess
  import sys
  import time
  import unicodedata
  from collections import OrderedDict
  from functools import lru_cache
  from pathlib import Path
  from threading import Thread
  from typing import (IO, TYPE_CHECKING, Any, Callable, Dict, Generator,
                      Iterable, List, NamedTuple, Optional, Pattern, Tuple, Type,
                      TypeVar, Union, cast, overload)
  
  import click
  import fmf
  import jsonschema
  import pkg_resources
  import requests
  import requests.adapters
  import requests.packages.urllib3.util.retry
  import urllib3.exceptions
  from click import echo, style, wrap_text
  from ruamel.yaml import YAML, scalarstring
  from ruamel.yaml.comments import CommentedMap
  from ruamel.yaml.parser import ParserError
  
  if sys.version_info >= (3, 8):
      from typing import Literal, Protocol
  else:
      from typing_extensions import Literal, Protocol
  
  
  if TYPE_CHECKING:
      import tmt.base
      import tmt.steps
  
  log = fmf.utils.Logging('tmt').logger
  
  # Default workdir root and max
  WORKDIR_ROOT = '/var/tmp/tmt'
  WORKDIR_MAX = 1000
  
  # Log in workdir
  LOG_FILENAME = 'log.txt'
  
  # Maximum number of lines of stdout/stderr to show upon errors
  OUTPUT_LINES = 100
  # Default output width
  OUTPUT_WIDTH = 79
  
  # Hierarchy indent
  INDENT = 4
  
  # Default name and order for step plugins
  DEFAULT_NAME = 'default'
  DEFAULT_PLUGIN_ORDER = 50
  DEFAULT_PLUGIN_ORDER_MULTIHOST = 10
  DEFAULT_PLUGIN_ORDER_REQUIRES = 70
  DEFAULT_PLUGIN_ORDER_RECOMMENDS = 75
  
  # Config directory
  CONFIG_PATH = '~/.config/tmt'
  
  # Special process return code
  PROCESS_TIMEOUT = 124
  
  # Default select.select(timeout) in seconds
  DEFAULT_SELECT_TIMEOUT = 5
  
  # Default shell and options to be set for all shell scripts
  DEFAULT_SHELL = "/bin/bash"
  SHELL_OPTIONS = 'set -eo pipefail'
  
  # Defaults for HTTP/HTTPS retries and timeouts (see `retry_session()`).
  DEFAULT_RETRY_SESSION_RETRIES: int = 3
  DEFAULT_RETRY_SESSION_BACKOFF_FACTOR: float = 0.1
  
  # Defaults for HTTP/HTTPS retries for getting environment file
  # Retry with exponential backoff, maximum duration ~511 seconds
  ENVFILE_RETRY_SESSION_RETRIES: int = 10
  ENVFILE_RETRY_SESSION_BACKOFF_FACTOR: float = 1
  
  # Default for wait()-related options
  DEFAULT_WAIT_TICK: float = 30.0
  DEFAULT_WAIT_TICK_INCREASE: float = 1.0
  
  # A stand-in variable for generic use.
  T = TypeVar('T')
  
  # A FMF context type, representing name/values context.
  FmfContextType = Dict[str, List[str]]
  
  # A "environment" type, representing name/value environment variables.
  EnvironmentType = Dict[str, str]
  
  # Workdir argument type, can be True, a string, a path or None
  WorkdirArgumentType = Union[Literal[True], str, None]
  
  # Workdir type, can be None or a string
  WorkdirType = Optional[str]
  
  # Option to skip to initialize work tree in plan
  PLAN_SKIP_WORKTREE_INIT = 'plan_skip_worktree_init'
  
  
  class BaseLoggerFnType(Protocol):
      def __call__(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          pass
  
  
  class LevelessLoggerFnType(Protocol):
      def __call__(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          pass
  
  
  class SemanticLoggerFnType(Protocol):
      def __call__(self, message: str, shift: int = 0) -> None:
          pass
  
  
  LoggerFnType = Union[
      BaseLoggerFnType,
      LevelessLoggerFnType,
      SemanticLoggerFnType]
  
  
  def indent(
          key: str,
          value: Optional[str] = None,
          color: Optional[str] = None,
          level: int = 0) -> str:
      """
      Indent a key/value message.
  
      If both ``key`` and ``value`` are specified, ``{key}: {value}``
      message is rendered. Otherwise, just ``key`` is used alone. If
      ``value`` contains multiple lines, each but the very first line is
      indented by one extra level.
  
      :param value: optional value to print at right side of ``key``.
      :param color: optional color to apply on ``key``.
      :param level: number of indentation levels. Each level is indented
                    by :py:data:`INDENT` spaces.
      """
  
      indent = ' ' * INDENT * level
      deeper = ' ' * INDENT * (level + 1)
  
      # Colorize
      if color is not None:
          key = style(key, fg=color)
  
      # Handle key only
      if value is None:
          message = key
  
      # Handle key + value
      else:
          # Multiline content indented deeper
          if isinstance(value, str):
              lines = value.splitlines()
              if len(lines) > 1:
                  value = ''.join([f"\n{deeper}{line}" for line in lines])
  
          message = f'{key}: {value}'
  
      return indent + message
  
  
  class Config:
      """ User configuration """
  
      def __init__(self) -> None:
          """ Initialize config directory path """
          self.path = os.path.expanduser(CONFIG_PATH)
          if not os.path.exists(self.path):
              try:
                  os.makedirs(self.path)
              except OSError as error:
                  raise GeneralError(
                      f"Failed to create config '{self.path}'.\n{error}")
  
      def last_run(self, run_id: Optional[str] = None) -> Optional[str]:
          """ Get and set last run id """
          symlink = os.path.join(self.path, 'last-run')
          if run_id:
              try:
                  os.remove(symlink)
              except OSError:
                  pass
              try:
                  os.symlink(run_id, symlink)
              except FileExistsError:
                  # Race when tmt runs in parallel
                  log.warning(f"Race condition, unable to save last run '{run_id}'.")
              except OSError as error:
                  raise GeneralError(
                      f"Unable to save last run '{self.path}'.\n{error}")
              return run_id
          if os.path.islink(symlink):
              return os.path.realpath(symlink)
          return None
  
  
  class StreamLogger(Thread):
      """
      Reading pipes of running process in threads.
  
      Code based on:
      https://github.com/packit/packit/blob/main/packit/utils/logging.py#L10
      """
  
      def __init__(self,
                   stream: Optional[IO[bytes]],
                   log_header: str,
                   logger: BaseLoggerFnType) -> None:
          super().__init__(daemon=True)
          self.stream = stream
          self.output: List[str] = []
          self.log_header = log_header
          self.logger = logger
  
      def run(self) -> None:
          if self.stream is None:
              return
  
          for _line in self.stream:
              line = _line.decode('utf-8', errors='replace')
              if line != '':
                  self.logger(
                      self.log_header,
                      line.rstrip('\n'),
                      'yellow',
                      level=3)
              self.output.append(line)
  
      def get_output(self) -> str:
          return "".join(self.output)
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Common
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  CommonDerivedType = TypeVar('CommonDerivedType', bound='Common')
  
  
  class CommandOutput(NamedTuple):
      stdout: Optional[str]
      stderr: Optional[str]
  
  
  class Common:
      """
      Common shared stuff
  
      Takes care of command line context, options and workdir handling.
      Provides logging functions info(), verbose() and debug().
      Implements read() and write() for comfortable file access.
      Provides the run() method for easy command execution.
      """
  
      # Command line context, options and workdir
      _context: Optional[click.Context] = None
      # When set to true, _opt will be ignored (default will be returned)
      ignore_class_options: bool = False
      _options: Dict[str, Any] = dict()
      _workdir: WorkdirType = None
  
      # TODO: must be declared outside of __init__(), because it must exist before
      # __init__() gets called to allow logging helpers work correctly when used
      # from mixins. But that's not very clean, is it? :( Maybe decoupling logging
      # from Common class would help, such a class would be able to initialize
      # itself without involving the rest of Common code. On the other hand,
      # Common owns workdir, for example, whose value affects logging too, so no
      # clear solution so far.
      #
      # Note: cannot use CommonDerivedType - it's a TypeVar filled in by the type
      # given to __init__() and therefore the type it's representing *now* is
      # unknown. but we know `parent` will be derived from `Common` class, so it's
      # mostly fine.
      parent: Optional['Common'] = None
  
      def __init__(
              self,
              parent: Optional[CommonDerivedType] = None,
              name: Optional[str] = None,
              workdir: WorkdirArgumentType = None,
              context: Optional[click.Context] = None
  
  [... the rest is omitted ...]
files: tmt/utils.py
files: 
  """ Test Metadata Utilities """
  
  import contextlib
  import copy
  import dataclasses
  import datetime
  import functools
  import glob
  import io
  import os
  import pprint
  import re
  import shlex
  import shutil
  import subprocess
  import sys
  import time
  import unicodedata
  from collections import OrderedDict
  from functools import lru_cache
  from pathlib import Path
  from threading import Thread
  from typing import (IO, TYPE_CHECKING, Any, Callable, Dict, Generator,
                      Iterable, List, NamedTuple, Optional, Pattern, Tuple, Type,
                      TypeVar, Union, cast, overload)
  
  import click
  import fmf
  import jsonschema
  import pkg_resources
  import requests
  import requests.adapters
  import requests.packages.urllib3.util.retry
  import urllib3.exceptions
  from click import echo, style, wrap_text
  from ruamel.yaml import YAML, scalarstring
  from ruamel.yaml.comments import CommentedMap
  from ruamel.yaml.parser import ParserError
  
  if sys.version_info >= (3, 8):
      from typing import Literal, Protocol
  else:
      from typing_extensions import Literal, Protocol
  
  
  if TYPE_CHECKING:
      import tmt.base
      import tmt.steps
  
  log = fmf.utils.Logging('tmt').logger
  
  # Default workdir root and max
  WORKDIR_ROOT = '/var/tmp/tmt'
  WORKDIR_MAX = 1000
  
  # Log in workdir
  LOG_FILENAME = 'log.txt'
  
  # Maximum number of lines of stdout/stderr to show upon errors
  OUTPUT_LINES = 100
  # Default output width
  OUTPUT_WIDTH = 79
  
  # Hierarchy indent
  INDENT = 4
  
  # Default name and order for step plugins
  DEFAULT_NAME = 'default'
  DEFAULT_PLUGIN_ORDER = 50
  DEFAULT_PLUGIN_ORDER_MULTIHOST = 10
  DEFAULT_PLUGIN_ORDER_REQUIRES = 70
  DEFAULT_PLUGIN_ORDER_RECOMMENDS = 75
  
  # Config directory
  CONFIG_PATH = '~/.config/tmt'
  
  # Special process return code
  PROCESS_TIMEOUT = 124
  
  # Default select.select(timeout) in seconds
  DEFAULT_SELECT_TIMEOUT = 5
  
  # Default shell and options to be set for all shell scripts
  DEFAULT_SHELL = "/bin/bash"
  SHELL_OPTIONS = 'set -eo pipefail'
  
  # Defaults for HTTP/HTTPS retries and timeouts (see `retry_session()`).
  DEFAULT_RETRY_SESSION_RETRIES: int = 3
  DEFAULT_RETRY_SESSION_BACKOFF_FACTOR: float = 0.1
  
  # Defaults for HTTP/HTTPS retries for getting environment file
  # Retry with exponential backoff, maximum duration ~511 seconds
  ENVFILE_RETRY_SESSION_RETRIES: int = 10
  ENVFILE_RETRY_SESSION_BACKOFF_FACTOR: float = 1
  
  # Default for wait()-related options
  DEFAULT_WAIT_TICK: float = 30.0
  DEFAULT_WAIT_TICK_INCREASE: float = 1.0
  
  # A stand-in variable for generic use.
  T = TypeVar('T')
  
  # A FMF context type, representing name/values context.
  FmfContextType = Dict[str, List[str]]
  
  # A "environment" type, representing name/value environment variables.
  EnvironmentType = Dict[str, str]
  
  # Workdir argument type, can be True, a string, a path or None
  WorkdirArgumentType = Union[Literal[True], str, None]
  
  # Workdir type, can be None or a string
  WorkdirType = Optional[str]
  
  # Option to skip to initialize work tree in plan
  PLAN_SKIP_WORKTREE_INIT = 'plan_skip_worktree_init'
  
  
  class BaseLoggerFnType(Protocol):
      def __call__(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          pass
  
  
  class LevelessLoggerFnType(Protocol):
      def __call__(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          pass
  
  
  class SemanticLoggerFnType(Protocol):
      def __call__(self, message: str, shift: int = 0) -> None:
          pass
  
  
  LoggerFnType = Union[
      BaseLoggerFnType,
      LevelessLoggerFnType,
      SemanticLoggerFnType]
  
  
  def indent(
          key: str,
          value: Optional[str] = None,
          color: Optional[str] = None,
          level: int = 0) -> str:
      """
      Indent a key/value message.
  
      If both ``key`` and ``value`` are specified, ``{key}: {value}``
      message is rendered. Otherwise, just ``key`` is used alone. If
      ``value`` contains multiple lines, each but the very first line is
      indented by one extra level.
  
      :param value: optional value to print at right side of ``key``.
      :param color: optional color to apply on ``key``.
      :param level: number of indentation levels. Each level is indented
                    by :py:data:`INDENT` spaces.
      """
  
      indent = ' ' * INDENT * level
      deeper = ' ' * INDENT * (level + 1)
  
      # Colorize
      if color is not None:
          key = style(key, fg=color)
  
      # Handle key only
      if value is None:
          message = key
  
      # Handle key + value
      else:
          # Multiline content indented deeper
          if isinstance(value, str):
              lines = value.splitlines()
              if len(lines) > 1:
                  value = ''.join([f"\n{deeper}{line}" for line in lines])
  
          message = f'{key}: {value}'
  
      return indent + message
  
  
  class Config:
      """ User configuration """
  
      def __init__(self) -> None:
          """ Initialize config directory path """
          self.path = os.path.expanduser(CONFIG_PATH)
          if not os.path.exists(self.path):
              try:
                  os.makedirs(self.path)
              except OSError as error:
                  raise GeneralError(
                      f"Failed to create config '{self.path}'.\n{error}")
  
      def last_run(self, run_id: Optional[str] = None) -> Optional[str]:
          """ Get and set last run id """
          symlink = os.path.join(self.path, 'last-run')
          if run_id:
              try:
                  os.remove(symlink)
              except OSError:
                  pass
              try:
                  os.symlink(run_id, symlink)
              except FileExistsError:
                  # Race when tmt runs in parallel
                  log.warning(f"Race condition, unable to save last run '{run_id}'.")
              except OSError as error:
                  raise GeneralError(
                      f"Unable to save last run '{self.path}'.\n{error}")
              return run_id
          if os.path.islink(symlink):
              return os.path.realpath(symlink)
          return None
  
  
  class StreamLogger(Thread):
      """
      Reading pipes of running process in threads.
  
      Code based on:
      https://github.com/packit/packit/blob/main/packit/utils/logging.py#L10
      """
  
      def __init__(self,
                   stream: Optional[IO[bytes]],
                   log_header: str,
                   logger: BaseLoggerFnType) -> None:
          super().__init__(daemon=True)
          self.stream = stream
          self.output: List[str] = []
          self.log_header = log_header
          self.logger = logger
  
      def run(self) -> None:
          if self.stream is None:
              return
  
          for _line in self.stream:
              line = _line.decode('utf-8', errors='replace')
              if line != '':
                  self.logger(
                      self.log_header,
                      line.rstrip('\n'),
                      'yellow',
                      level=3)
              self.output.append(line)
  
      def get_output(self) -> str:
          return "".join(self.output)
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Common
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  CommonDerivedType = TypeVar('CommonDerivedType', bound='Common')
  
  
  class CommandOutput(NamedTuple):
      stdout: Optional[str]
      stderr: Optional[str]
  
  
  class Common:
      """
      Common shared stuff
  
      Takes care of command line context, options and workdir handling.
      Provides logging functions info(), verbose() and debug().
      Implements read() and write() for comfortable file access.
      Provides the run() method for easy command execution.
      """
  
      # Command line context, options and workdir
      _context: Optional[click.Context] = None
      # When set to true, _opt will be ignored (default will be returned)
      ignore_class_options: bool = False
      _options: Dict[str, Any] = dict()
      _workdir: WorkdirType = None
  
      # TODO: must be declared outside of __init__(), because it must exist before
      # __init__() gets called to allow logging helpers work correctly when used
      # from mixins. But that's not very clean, is it? :( Maybe decoupling logging
      # from Common class would help, such a class would be able to initialize
      # itself without involving the rest of Common code. On the other hand,
      # Common owns workdir, for example, whose value affects logging too, so no
      # clear solution so far.
      #
      # Note: cannot use CommonDerivedType - it's a TypeVar filled in by the type
      # given to __init__() and therefore the type it's representing *now* is
      # unknown. but we know `parent` will be derived from `Common` class, so it's
      # mostly fine.
      parent: Optional['Common'] = None
  
      def __init__(
              self,
              parent: Optional[CommonDerivedType] = None,
              name: Optional[str] = None,
              workdir: WorkdirArgumentType = None,
              context: Optional[click.Context] = None
  
  [... the rest is omitted ...]
files: tmt/utils.py