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, Dict, Generator, 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:
      url: Optional[str] = None
      ref: Optional[str] = None
      path: Optional[str] = None
      name: Optional[str] = None
  
      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 dataclasses.asdict(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, the data is wrapped with `Link`
  # instances, the type below describes the raw data coming from Fmf
  # nodes.
  # TODO: `*Link._relations` would be much better, DRY, but that's allowed
  # since Python 3.11.
  _RawLinkRelation = 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'
  ]
  
  # A "relation": "link" subtype.
  #
  # An example from TMT docs says:
  #
  # link:
  #   verifies: /stories/cli/init/base
  #
  # link:
  #     blocked-by:
  #         url: https://github.com/teemtee/fmf
  #         name: /stories/select/filter/regexp
  #     note: Need to get the regexp filter working first.
  _RawLinkRelationAware = Dict[_RawLinkRelation, Union[str, _RawFmfId]]
  
  _RawLink = Union[
      # link: https://github.com/teemtee/tmt/issues/461
      str,
      _RawFmfId,
      _RawLinkRelationAware,
  
      # link:
      # - verifies: /stories/cli/init/base
      # - verifies: https://bugzilla.redhat.com/show_bug.cgi?id=1234
      List[Union[str, _RawFmfId, _RawLinkRelationAware]],
  ]
  
  
  # 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 ValidateFmfMixin:
      """
      Mixin adding validation of an fmf node.
  
      Loads a schema whose name is derived from class name, and uses fmf's validate()
      method to perform the validation.
      """
  
      def _validate_fmf_node(self, node: fmf.Tree, logger: tmt.utils.Common,
                             raise_on_validation_error: bool) -> None:
          """ Validate a given fmf node """
  
          errors = tmt.utils.validate_fmf_node(
              node, f'{self.__class__.__name__.lower()}.yaml')
  
          if errors:
              if raise_on_validation_error:
                  raise tmt.utils.SpecificationError(
                      f'fmf node {node.name} failed validation',
                      validation_errors=errors)
  
              for _, error_message in errors:
                  logger.warn(error_message, shift=1)
  
      def __init__(
              self,
              *,
              node: fmf.Tree,
              logger: tmt.utils.Common,
              skip_validation: bool = False,
              raise_on_validation_error: bool = False,
              **kwargs: Any) -> None:
          # Validate *before* letting next class in line touch the data.
          if not skip_validation:
              self._validate_fmf_node(node, logger, raise_on_validation_error)
  
          kwargs.setdefault('logger', self)
          super().__init__(node=node, **kwargs)
  
  
  # A type representing compatible sources of keys and values.
  KeySource = Union[Dict[str, Any], fmf.Tree]
  
  
  class LoadFmfKeysMixin:
      """
      Mixin adding support for loading fmf keys into object attributes.
  
      When invoked, annotated class-level variables are searched for in a given fmf node,
      and if the key of the same name as the variable exists, its value is "promoted" to
      instance variable.
  
      If a method named ``_normalize_<variable name>`` exists, it is called with the fmf
      key value as its single argument, and its return value is assigned to instance
      variable. This gives class chance to modify or transform the original value when
      needed, e.g. to convert the original value to a type more suitable for internal
      processing.
      """
  
      # If specified, keys would be iterated over in the order as listed here.
      KEYS_SHOW_ORDER: List[str] = []
  
      # NOTE: these could be static methods, self is probably useless, but that would
      # cause complications when classes assign these to their members. That makes them
      # no longer static as far as class is concerned, which means they get called with
      # `self` as the first argument. A workaround would be to assign staticmethod()-ized
      # version of them, but that's too much repetition.
      def _normalize_string_list(self, value: Union[None, str, List[str]]) -> List[str]:
          if value is None:
              return []
  
          return [value] if isinstance(value, str) else value
  
      def _normalize_environment(self, value: Optional[Dict[str, Any]]) -> tmt.utils.EnvironmentType:
          if value is None:
              return {}
  
          return {
              name: str(value) for name, value in value.items()
              }
  
      @classmethod
      def _iter_key_annotations(cls) -> Generator[Tuple[str, Any], None, None]:
          """
          Iterate over keys' type annotations.
  
          Keys are yielded in the order: keys declared by parent classes first, then
          keys declared by the class itself, all following the order in which keys
          were defined in their respective classes.
  
          Yields:
              pairs of key name and its annotations.
          """
  
          for base in cls.__bases__:
              for name, value in base.__dict__.get('__annotations__', {}).items():
                  # Skip special fields that are not keys.
                  if name == 'KEYS_SHOW_ORDER':
                      continue
  
                  yield (name, value)
  
          yield from cls.__dict__.get('__annotations__', {}).items()
  
      @classmethod
      def keys(cls) -> Generator[str, None, None]:
          """
          Iterate over key names.
  
          Keys are yielded in the order: keys declared by parent classes first, then
          keys declared by the class itself, all following the order in which keys
          were defined in their respective classes.
  
          Yields:
              key names.
          """
  
          for keyname, _ in cls._iter_key_annotations():
              yield keyname
  
      def items(self) -> Generator[Tuple[str, Any], None, None]:
          """
          Iterate over keys and their values.
  
          Keys are yielded in the order: keys declared by parent classes first, then
          keys declared by the class itself, all following the order in which keys
          were defined in their respective classes.
  
          Yields:
              pairs of key name and its value.
          """
  
          for keyname in self.keys():
              yield (keyname, getattr(self, keyname))
  
      # TODO: exists for backward compatibility for the transition period. Once full
      # type annotations land, there should be no need for extra _keys attribute.
      @classmethod
      def _keys(cls) -> List[str]:
          """ Return a list of names of object's keys. """
  
          return list(cls.keys())
  
      def _load_keys(
              self,
              node: fmf.Tree,
              logger: tmt.utils.Common) -> None:
          """ Extract values for class-level attributes, and verify they match declared types. """
  
          LOG_SHIFT, LOG_LEVEL = 2, 4
  
          debug = functools.partial(self.debug, shift=LOG_SHIFT, level=LOG_LEVEL)
  
          for keyname, keytype in self._iter_key_annotations():
              key_address = f'{node.name}{keyname}'
  
              # Do not indent this particular entry like the rest, so it could serve
              # as a "header" for a single key processing.
              logger.debug('key', key_address, shift=LOG_SHIFT - 1, level=LOG_LEVEL)
  
              debug('desired type', str(keytype))
  
              value: Any = None
  
              debug('dict', self.__dict__)
  
              if hasattr(self, keyname):
                  # If the key exists as instance's attribute already, it is because it's been
                  # declared with a default value, and the attribute now holds said default value.
                  default_value = getattr(self, keyname)
  
                  # If the default value is a mutable container, we cannot use it directly.
                  # Should we do so, the very same default value would be assigned to multiple
                  # instances/attributes instead of each instance having its own distinct container.
                  if isinstance(default_value, (list, dict)):
                      debug('detected mutable default')
                      default_value = copy.copy(default_value)
  
                  debug('default value', str(default_value))
                  debug('default value type', str(type(default_value)))
  
                  # try+except seems to work better than get(), especially when
                  # semantic of fmf.Tree.get() is slightly different than that
                  # of dict().get().
                  try:
                      value = node[keyname]
  
                  except KeyError:
                      value = default_value
  
                  debug('raw value', str(value))
                  debug('raw value type', str(type(value)))
  
              else:
                  value = node.get(keyname)
  
                  debug('raw value', str(value))
                  debug('raw value type', str(type(value)))
  
              debug('value', str(value))
              debug('value type', str(type(value)))
  
              normalize_callback = getattr(self, f'_normalize_{keyname}', None)
  
              if normalize_callback:
                  value = normalize_callback(value)
  
                  debug('normalized value', str(value))
                  debug('normalized value type', str(type(value)))
  
              debug('final value', str(value))
              debug('final value type', str(type(value)))
  
              setattr(self, keyname, value)
  
              # Apparently pointless, but makes the debugging output more readable.
              # There may be plenty of tests and plans and keys, a bit of spacing
              # can't hurt.
              debug('')
  
      def __init__(
              self,
              *,
              node: fmf.Tree,
              logger: tmt.utils.Common,
              **kwargs: Any) -> None:
          self._load_keys(node, logger)
  
          kwargs.setdefault('logger', self)
          super().__init__(node=node, **kwargs)
  
  
  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['Link'] = 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 = 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: _RawLink) -> 'Link':
          return Link(data=value)
  
      def _normalize_adjust(
              self, value: Union[_RawAdjustRule, List[_RawAdjustRule]]) -> List[_RawAdjustRule]:
          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
  
          # Verify id is not inherited from parent.
          if self.id and not tmt.identifier.key_defined_in_leaf(node, 'id'):
              raise tmt.utils.SpecificationError(
                  f"The 'id' key '{node.get('id')}' in '{self.name}' "
                  f"is inherited from parent, should be defined in a leaf.")
  
          # Store original metadata with applied defaults and including
          # keys which are not defined in the L1 metadata specification
          # Once the whole node has been initialized,
          # self._update_metadata() must be called to work correctly.
          self._metadata = self.node.data.copy()
  
      def __str__(self):
          """ Node name """
          return self.name
  
      def _update_metadata(self):
          """ Update the _metadata attribute """
          self._metadata.update(self.export(format_='dict'))
          self._metadata['name'] = self.name
  
      def _show_additional_keys(self):
          """ Show source files """
          if self.id is not None:
              echo(tmt.utils.format('id', self.id, key_color='magenta'))
          echo(tmt.utils.format(
              'sources', self.node.sources, key_color='magenta'))
  
      def _fmf_id(self):
          """ Show fmf identifier """
          echo(tmt.utils.format('fmf-id', self.fmf_id, key_color='magenta'))
  
      @property
      @functools.lru_cache(maxsize=None)
      def fmf_id(self):
          """ Return full fmf identifier of the node """
  
          def run(command):
              """ Run command, return output """
              cwd = fmf_root
              result = subprocess.run(
                  command.split(),
                  stdout=subprocess.PIPE,
                  stderr=subprocess.DEVNULL,
                  cwd=cwd)
              return result.stdout.strip().decode("utf-8")
  
          fmf_id = {'name': self.name}
  
          fmf_root = self.node.root
          # Prepare url (for now handle just the most common schemas)
          branch = run("git rev-parse --abbrev-ref --symbolic-full-name @{u}")
          try:
              remote_name = branch[:branch.index('/')]
          except ValueError:
              remote_name = 'origin'
          remote = run(f"git config --get remote.{remote_name}.url")
          fmf_id['url'] = tmt.utils.public_git_url(remote)
  
          # Get the ref (skip for master as it is the default)
          ref = run('git rev-parse --abbrev-ref HEAD')
          if ref != 'master':
              fmf_id['ref'] = ref
  
          # Construct path (if different from git root)
          git_root = run('git rev-parse --show-toplevel')
          if git_root != fmf_root:
              fmf_id['path'] = os.path.join(
                  '/', os.path.relpath(fmf_root, git_root))
  
          return fmf_id
  
      @classmethod
      def _save_context(cls, context):
          """ Save provided command line context for future use """
          super(Core, cls)._save_context(context)
  
          # Handle '.' as an alias for the current working directory
          names = cls._opt('names')
          if names is not None and '.' in names:
              root = context.obj.tree.root
              current = os.getcwd()
              # Handle special case when directly in the metadata root
              if current == root:
                  pattern = '/'
              # Prepare path from the tree root to the current directory
              else:
                  pattern = os.path.join('/', os.path.relpath(current, root))
                  # Prevent matching common prefix from other directories
                  pattern = f"{pattern}(/|$)"
              cls._context.params['names'] = tuple([
                  pattern if name == '.' else name for name in names])
  
      def name_and_summary(self):
          """ Node name and optional summary """
          if self.summary:
              return '{0} ({1})'.format(self.name, self.summary)
          else:
              return self.name
  
      def ls(self, summary=False):
          """ List node """
          echo(style(self.name, fg='red'))
          if summary and self.summary:
              echo(tmt.utils.format('summary', self.summary))
  
      def export(self, format_='yaml', keys=None):
          """ Export data into requested format (yaml or dict) """
          if keys is None:
              keys = self._keys()
  
          # Always include node name, add requested keys, ignore adjust
          data = dict(name=self.name)
          for key in keys:
              if key == 'adjust':
                  continue
  
              value = getattr(self, key)
  
              if key == 'link':
                  data[key] = value.links
  
              else:
                  data[key] = value
  
          # Choose proper format
          if format_ == 'dict':
              return data
          elif format_ == 'yaml':
              return tmt.utils.dict_to_yaml(data)
          else:
              raise tmt.utils.GeneralError(
                  f"Invalid export format '{format_}'.")
  
      def lint_keys(self, additional_keys):
          """ Return list of invalid keys used, empty when all good """
          known_keys = additional_keys + self._keys()
          return [key for key in self.node.get().keys() if key not in known_keys]
  
      def _lint_summary(self):
          """ Lint summary attribute """
          # Summary is advised with a resonable length
          if self.summary is None:
              verdict(None, "summary is very useful for quick inspection")
          elif len(self.summary) > 50:
              verdict(None, "summary should not exceed 50 characters")
          return True
  
      def has_link(self, link_object):
          """ Whether object contains specified link """
          def get_relation(from_link):
              # keys() = relation and optional note
              return list(set(from_link.keys()) - set(['note']))[0]
          if isinstance(link_object, Link):
              relation = get_relation(link_object)
              target = link_object[relation]
          else:
              # User text input
              parts = link_object.split(':', maxsplit=1)
             
  
  [... 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, Dict, Generator, 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>']
      }
  
  
  #
  # 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:
      url: Optional[str] = None
      ref: Optional[str] = None
      path: Optional[str] = None
      name: Optional[str] = None
  
      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 dataclasses.asdict(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, the data is wrapped with `Link`
  # instances, the type below describes the raw data coming from Fmf
  # nodes.
  # TODO: `*Link._relations` would be much better, DRY, but that's allowed
  # since Python 3.11.
  _RawLinkRelation = 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'
  ]
  
  # A "relation": "link" subtype.
  #
  # An example from TMT docs says:
  #
  # link:
  #   verifies: /stories/cli/init/base
  #
  # link:
  #     blocked-by:
  #         url: https://github.com/teemtee/fmf
  #         name: /stories/select/filter/regexp
  #     note: Need to get the regexp filter working first.
  _RawLinkRelationAware = Dict[_RawLinkRelation, Union[str, _RawFmfId]]
  
  _RawLink = Union[
      # link: https://github.com/teemtee/tmt/issues/461
      str,
      _RawFmfId,
      _RawLinkRelationAware,
  
      # link:
      # - verifies: /stories/cli/init/base
      # - verifies: https://bugzilla.redhat.com/show_bug.cgi?id=1234
      List[Union[str, _RawFmfId, _RawLinkRelationAware]],
  ]
  
  
  # 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 ValidateFmfMixin:
      """
      Mixin adding validation of an fmf node.
  
      Loads a schema whose name is derived from class name, and uses fmf's validate()
      method to perform the validation.
      """
  
      def _validate_fmf_node(self, node: fmf.Tree, logger: tmt.utils.Common,
                             raise_on_validation_error: bool) -> None:
          """ Validate a given fmf node """
  
          errors = tmt.utils.validate_fmf_node(
              node, f'{self.__class__.__name__.lower()}.yaml')
  
          if errors:
              if raise_on_validation_error:
                  raise tmt.utils.SpecificationError(
                      f'fmf node {node.name} failed validation',
                      validation_errors=errors)
  
              for _, error_message in errors:
                  logger.warn(error_message, shift=1)
  
      def __init__(
              self,
              *,
              node: fmf.Tree,
              logger: tmt.utils.Common,
              skip_validation: bool = False,
              raise_on_validation_error: bool = False,
              **kwargs: Any) -> None:
          # Validate *before* letting next class in line touch the data.
          if not skip_validation:
              self._validate_fmf_node(node, logger, raise_on_validation_error)
  
          kwargs.setdefault('logger', self)
          super().__init__(node=node, **kwargs)
  
  
  # A type representing compatible sources of keys and values.
  KeySource = Union[Dict[str, Any], fmf.Tree]
  
  
  class LoadFmfKeysMixin:
      """
      Mixin adding support for loading fmf keys into object attributes.
  
      When invoked, annotated class-level variables are searched for in a given fmf node,
      and if the key of the same name as the variable exists, its value is "promoted" to
      instance variable.
  
      If a method named ``_normalize_<variable name>`` exists, it is called with the fmf
      key value as its single argument, and its return value is assigned to instance
      variable. This gives class chance to modify or transform the original value when
      needed, e.g. to convert the original value to a type more suitable for internal
      processing.
      """
  
      # If specified, keys would be iterated over in the order as listed here.
      KEYS_SHOW_ORDER: List[str] = []
  
      # NOTE: these could be static methods, self is probably useless, but that would
      # cause complications when classes assign these to their members. That makes them
      # no longer static as far as class is concerned, which means they get called with
      # `self` as the first argument. A workaround would be to assign staticmethod()-ized
      # version of them, but that's too much repetition.
      def _normalize_string_list(self, value: Union[None, str, List[str]]) -> List[str]:
          if value is None:
              return []
  
          return [value] if isinstance(value, str) else value
  
      def _normalize_environment(self, value: Optional[Dict[str, Any]]) -> tmt.utils.EnvironmentType:
          if value is None:
              return {}
  
          return {
              name: str(value) for name, value in value.items()
              }
  
      @classmethod
      def _iter_key_annotations(cls) -> Generator[Tuple[str, Any], None, None]:
          """
          Iterate over keys' type annotations.
  
          Keys are yielded in the order: keys declared by parent classes first, then
          keys declared by the class itself, all following the order in which keys
          were defined in their respective classes.
  
          Yields:
              pairs of key name and its annotations.
          """
  
          for base in cls.__bases__:
              for name, value in base.__dict__.get('__annotations__', {}).items():
                  # Skip special fields that are not keys.
                  if name == 'KEYS_SHOW_ORDER':
                      continue
  
                  yield (name, value)
  
          yield from cls.__dict__.get('__annotations__', {}).items()
  
      @classmethod
      def keys(cls) -> Generator[str, None, None]:
          """
          Iterate over key names.
  
          Keys are yielded in the order: keys declared by parent classes first, then
          keys declared by the class itself, all following the order in which keys
          were defined in their respective classes.
  
          Yields:
              key names.
          """
  
          for keyname, _ in cls._iter_key_annotations():
              yield keyname
  
      def items(self) -> Generator[Tuple[str, Any], None, None]:
          """
          Iterate over keys and their values.
  
          Keys are yielded in the order: keys declared by parent classes first, then
          keys declared by the class itself, all following the order in which keys
          were defined in their respective classes.
  
          Yields:
              pairs of key name and its value.
          """
  
          for keyname in self.keys():
              yield (keyname, getattr(self, keyname))
  
      # TODO: exists for backward compatibility for the transition period. Once full
      # type annotations land, there should be no need for extra _keys attribute.
      @classmethod
      def _keys(cls) -> List[str]:
          """ Return a list of names of object's keys. """
  
          return list(cls.keys())
  
      def _load_keys(
              self,
              node: fmf.Tree,
              logger: tmt.utils.Common) -> None:
          """ Extract values for class-level attributes, and verify they match declared types. """
  
          LOG_SHIFT, LOG_LEVEL = 2, 4
  
          debug = functools.partial(self.debug, shift=LOG_SHIFT, level=LOG_LEVEL)
  
          for keyname, keytype in self._iter_key_annotations():
              key_address = f'{node.name}{keyname}'
  
              # Do not indent this particular entry like the rest, so it could serve
              # as a "header" for a single key processing.
              logger.debug('key', key_address, shift=LOG_SHIFT - 1, level=LOG_LEVEL)
  
              debug('desired type', str(keytype))
  
              value: Any = None
  
              debug('dict', self.__dict__)
  
              if hasattr(self, keyname):
                  # If the key exists as instance's attribute already, it is because it's been
                  # declared with a default value, and the attribute now holds said default value.
                  default_value = getattr(self, keyname)
  
                  # If the default value is a mutable container, we cannot use it directly.
                  # Should we do so, the very same default value would be assigned to multiple
                  # instances/attributes instead of each instance having its own distinct container.
                  if isinstance(default_value, (list, dict)):
                      debug('detected mutable default')
                      default_value = copy.copy(default_value)
  
                  debug('default value', str(default_value))
                  debug('default value type', str(type(default_value)))
  
                  # try+except seems to work better than get(), especially when
                  # semantic of fmf.Tree.get() is slightly different than that
                  # of dict().get().
                  try:
                      value = node[keyname]
  
                  except KeyError:
                      value = default_value
  
                  debug('raw value', str(value))
                  debug('raw value type', str(type(value)))
  
              else:
                  value = node.get(keyname)
  
                  debug('raw value', str(value))
                  debug('raw value type', str(type(value)))
  
              debug('value', str(value))
              debug('value type', str(type(value)))
  
              normalize_callback = getattr(self, f'_normalize_{keyname}', None)
  
              if normalize_callback:
                  value = normalize_callback(value)
  
                  debug('normalized value', str(value))
                  debug('normalized value type', str(type(value)))
  
              debug('final value', str(value))
              debug('final value type', str(type(value)))
  
              setattr(self, keyname, value)
  
              # Apparently pointless, but makes the debugging output more readable.
              # There may be plenty of tests and plans and keys, a bit of spacing
              # can't hurt.
              debug('')
  
      def __init__(
              self,
              *,
              node: fmf.Tree,
              logger: tmt.utils.Common,
              **kwargs: Any) -> None:
          self._load_keys(node, logger)
  
          kwargs.setdefault('logger', self)
          super().__init__(node=node, **kwargs)
  
  
  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['Link'] = 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 = 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: _RawLink) -> 'Link':
          return Link(data=value)
  
      def _normalize_adjust(
              self, value: Union[_RawAdjustRule, List[_RawAdjustRule]]) -> List[_RawAdjustRule]:
          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
  
          # Verify id is not inherited from parent.
          if self.id and not tmt.identifier.key_defined_in_leaf(node, 'id'):
              raise tmt.utils.SpecificationError(
                  f"The 'id' key '{node.get('id')}' in '{self.name}' "
                  f"is inherited from parent, should be defined in a leaf.")
  
          # Store original metadata with applied defaults and including
          # keys which are not defined in the L1 metadata specification
          # Once the whole node has been initialized,
          # self._update_metadata() must be called to work correctly.
          self._metadata = self.node.data.copy()
  
      def __str__(self):
          """ Node name """
          return self.name
  
      def _update_metadata(self):
          """ Update the _metadata attribute """
          self._metadata.update(self.export(format_='dict'))
          self._metadata['name'] = self.name
  
      def _show_additional_keys(self):
          """ Show source files """
          if self.id is not None:
              echo(tmt.utils.format('id', self.id, key_color='magenta'))
          echo(tmt.utils.format(
              'sources', self.node.sources, key_color='magenta'))
  
      def _fmf_id(self):
          """ Show fmf identifier """
          echo(tmt.utils.format('fmf-id', self.fmf_id, key_color='magenta'))
  
      @property
      @functools.lru_cache(maxsize=None)
      def fmf_id(self):
          """ Return full fmf identifier of the node """
  
          def run(command):
              """ Run command, return output """
              cwd = fmf_root
              result = subprocess.run(
                  command.split(),
                  stdout=subprocess.PIPE,
                  stderr=subprocess.DEVNULL,
                  cwd=cwd)
              return result.stdout.strip().decode("utf-8")
  
          fmf_id = {'name': self.name}
  
          fmf_root = self.node.root
          # Prepare url (for now handle just the most common schemas)
          branch = run("git rev-parse --abbrev-ref --symbolic-full-name @{u}")
          try:
              remote_name = branch[:branch.index('/')]
          except ValueError:
              remote_name = 'origin'
          remote = run(f"git config --get remote.{remote_name}.url")
          fmf_id['url'] = tmt.utils.public_git_url(remote)
  
          # Get the ref (skip for master as it is the default)
          ref = run('git rev-parse --abbrev-ref HEAD')
          if ref != 'master':
              fmf_id['ref'] = ref
  
          # Construct path (if different from git root)
          git_root = run('git rev-parse --show-toplevel')
          if git_root != fmf_root:
              fmf_id['path'] = os.path.join(
                  '/', os.path.relpath(fmf_root, git_root))
  
          return fmf_id
  
      @classmethod
      def _save_context(cls, context):
          """ Save provided command line context for future use """
          super(Core, cls)._save_context(context)
  
          # Handle '.' as an alias for the current working directory
          names = cls._opt('names')
          if names is not None and '.' in names:
              root = context.obj.tree.root
              current = os.getcwd()
              # Handle special case when directly in the metadata root
              if current == root:
                  pattern = '/'
              # Prepare path from the tree root to the current directory
              else:
                  pattern = os.path.join('/', os.path.relpath(current, root))
                  # Prevent matching common prefix from other directories
                  pattern = f"{pattern}(/|$)"
              cls._context.params['names'] = tuple([
                  pattern if name == '.' else name for name in names])
  
      def name_and_summary(self):
          """ Node name and optional summary """
          if self.summary:
              return '{0} ({1})'.format(self.name, self.summary)
          else:
              return self.name
  
      def ls(self, summary=False):
          """ List node """
          echo(style(self.name, fg='red'))
          if summary and self.summary:
              echo(tmt.utils.format('summary', self.summary))
  
      def export(self, format_='yaml', keys=None):
          """ Export data into requested format (yaml or dict) """
          if keys is None:
              keys = self._keys()
  
          # Always include node name, add requested keys, ignore adjust
          data = dict(name=self.name)
          for key in keys:
              if key == 'adjust':
                  continue
  
              value = getattr(self, key)
  
              if key == 'link':
                  data[key] = value.links
  
              else:
                  data[key] = value
  
          # Choose proper format
          if format_ == 'dict':
              return data
          elif format_ == 'yaml':
              return tmt.utils.dict_to_yaml(data)
          else:
              raise tmt.utils.GeneralError(
                  f"Invalid export format '{format_}'.")
  
      def lint_keys(self, additional_keys):
          """ Return list of invalid keys used, empty when all good """
          known_keys = additional_keys + self._keys()
          return [key for key in self.node.get().keys() if key not in known_keys]
  
      def _lint_summary(self):
          """ Lint summary attribute """
          # Summary is advised with a resonable length
          if self.summary is None:
              verdict(None, "summary is very useful for quick inspection")
          elif len(self.summary) > 50:
              verdict(None, "summary should not exceed 50 characters")
          return True
  
      def has_link(self, link_object):
          """ Whether object contains specified link """
          def get_relation(from_link):
              # keys() = relation and optional note
              return list(set(from_link.keys()) - set(['note']))[0]
          if isinstance(link_object, Link):
              relation = get_relation(link_object)
              target = link_object[relation]
          else:
              # User text input
              parts = link_object.split(':', maxsplit=1)
              if len(parts) == 1:
                  relation, target = ".*", parts[0]
              else:
                  relation, target = parts
  
  
  [... the rest is omitted ...]
files: tmt/base.py
files: # coding: utf-8
  
  """ Export metadata into nitrate """
  
  
  import email.utils
  import os
  import re
  import traceback
  import types
  import xmlrpc.client
  from functools import lru_cache
  from typing import Any, Dict, Generator, List, Optional, Tuple, Union, cast
  
  import fmf
  from click import echo, secho, style
  
  import tmt
  import tmt.utils
  from tmt.convert import SYSTEM_OTHER, add_link
  from tmt.identifier import ID_KEY, add_uuid_if_not_defined
  from tmt.utils import ConvertError, check_git_url, markdown_to_html
  
  bugzilla: Optional[types.ModuleType] = None
  gssapi: Optional[types.ModuleType] = None
  nitrate: Optional[types.ModuleType] = None
  
  PolarionException: Any = None
  PolarionTestCase: Any = None
  PolarionWorkItem: Any = None
  
  DEFAULT_PRODUCT: Any = None
  
  SectionsReturnType = Tuple[str, str, str, str]
  HeadingsType = List[List[Union[int, str]]]
  SectionsHeadingsType = Dict[str, HeadingsType]
  
  log = fmf.utils.Logging('tmt').logger
  
  WARNING = """
  Test case has been migrated to git. Any changes made here might be overwritten.
  See: https://tmt.readthedocs.io/en/latest/questions.html#nitrate-migration
  """.lstrip()
  
  # For linking bugs
  BUGZILLA_XMLRPC_URL = "https://bugzilla.redhat.com/xmlrpc.cgi"
  NITRATE_TRACKER_ID = 69  # ID of nitrate in RH's bugzilla
  POLARION_TRACKER_ID = 117  # ID of polarion in RH's bugzilla
  RE_BUGZILLA_URL = r'bugzilla.redhat.com/show_bug.cgi\?id=(\d+)'
  RE_POLARION_URL = r'.*/polarion/#/project/.*/workitem\?id=(.*)'
  LEGACY_POLARION_PROJECTS = set(['RedHatEnterpriseLinux7'])
  
  
  def import_nitrate() -> Any:
      """ Conditionally import the nitrate module """
      # Need to import nitrate only when really needed. Otherwise we get
      # traceback when nitrate not installed or config file not available.
      # And we want to keep the core tmt package with minimal dependencies.
      try:
          global nitrate, DEFAULT_PRODUCT, gssapi
          import gssapi
          import nitrate
          assert nitrate
          DEFAULT_PRODUCT = nitrate.Product(name='RHEL Tests')
          return nitrate
      except ImportError:
          raise ConvertError(
              "Install tmt-test-convert to export tests to nitrate.")
      except nitrate.NitrateError as error:  # type: ignore
          raise ConvertError(error)
  
  
  def import_polarion() -> None:
      """ Import polarion python api - pylero """
      try:
          global PolarionException, PolarionTestCase, PolarionWorkItem
          from pylero.exceptions import PyleroLibException as PolarionException
          from pylero.work_item import TestCase as PolarionTestCase
          from pylero.work_item import _WorkItem as PolarionWorkItem
      except ImportError:
          raise ConvertError(
              "Run 'pip install tmt[export-polarion]' so pylero is installed.")
      except PolarionException as exc:
          log.debug(traceback.format_exc())
          raise ConvertError("Failed to login with pylero", original=exc)
  
  
  def get_bz_instance() -> Any:
      """ Import the bugzilla module and return BZ instance """
      try:
          import bugzilla
      except ImportError:
          raise ConvertError(
              "Install 'tmt-test-convert' to link test to the bugzilla.")
  
      try:
          bz_instance = bugzilla.Bugzilla(url=BUGZILLA_XMLRPC_URL)
      except Exception as exc:
          log.debug(traceback.format_exc())
          raise ConvertError(
              "Couldn't initialize the Bugzilla client.", original=exc)
  
      if not bz_instance.logged_in:
          raise ConvertError(
              "Not logged to Bugzilla, check 'man bugzilla' section "
              "'AUTHENTICATION CACHE AND API KEYS'.")
      return bz_instance
  
  
  def _nitrate_find_fmf_testcases(test: 'tmt.Test') -> Generator[Any, None, None]:
      """
      Find all Nitrate test cases with the same fmf identifier
  
      All component general plans are explored for possible duplicates.
      """
      assert nitrate
      for component in test.component:
          try:
              for testcase in find_general_plan(component).testcases:
                  struct_field = tmt.utils.StructuredField(testcase.notes)
                  try:
                      fmf_id = tmt.utils.yaml_to_dict(struct_field.get('fmf'))
                      if fmf_id == test.fmf_id:
                          echo(style(
                              f"Existing test case '{testcase.identifier}' "
                              f"found for given fmf id.", fg='magenta'))
                          yield testcase
                  except tmt.utils.StructuredFieldError:
                      pass
          except nitrate.NitrateError:
              pass
  
  
  def convert_manual_to_nitrate(test_md: str) -> SectionsReturnType:
      """
      Convert Markdown document to html sections.
  
      These sections can be exported to nitrate.
      Expects: Markdown document as a file.
      Returns: tuple of (step, expect, setup, cleanup) sections
      as html strings.
      """
  
      sections_headings: SectionsHeadingsType = {
          heading: []
          for heading_list in tmt.base.SECTIONS_HEADINGS.values()
          for heading in heading_list
          }
  
      html = markdown_to_html(test_md)
      html_splitlines = html.splitlines()
  
      for key in sections_headings.keys():
          result: HeadingsType = []
          i = 0
          while html_splitlines:
              try:
                  if re.search("^" + key + "$", html_splitlines[i]):
                      html_content = str()
                      if key.startswith('<h1>Test'):
                          html_content = html_splitlines[i].\
                              replace('<h1>', '<b>').\
                              replace('</h1>', '</b>')
                      for j in range(i + 1, len(html_splitlines)):
                          if re.search("^<h[1-4]>(.+?)</h[1-4]>$",
                                       html_splitlines[j]):
                              result.append([i, html_content])
                              i = j - 1
                              break
                          html_content += html_splitlines[j] + "\n"
                          # Check end of the file
                          if j + 1 == len(html_splitlines):
                              result.append([i, html_content])
              except IndexError:
                  sections_headings[key] = result
                  break
              i += 1
              if i >= len(html_splitlines):
                  sections_headings[key] = result
                  break
  
      def concatenate_headings_content(headings: Tuple[str, ...]) -> HeadingsType:
          content = list()
          for v in headings:
              content += sections_headings[v]
          return content
  
      def enumerate_content(content: HeadingsType) -> HeadingsType:
          # for sorting convert the index to integer, but keep whole list as list of strings
          content.sort(key=lambda a: int(a[0]))
          for c in range(len(content)):
              content[c][1] = f"<p>Step {c + 1}.</p>" + str(content[c][1])
          return content
  
      sorted_test = sorted(concatenate_headings_content((
          '<h1>Test</h1>',
          '<h1>Test .*</h1>')))
  
      sorted_step = sorted(enumerate_content(concatenate_headings_content((
          '<h2>Step</h2>',
          '<h2>Test Step</h2>'))) + sorted_test)
      step = ''.join([f"{v[1]}" for v in sorted_step])
  
      sorted_expect = sorted(enumerate_content(concatenate_headings_content((
          '<h2>Expect</h2>',
          '<h2>Result</h2>',
          '<h2>Expected Result</h2>'))) + sorted_test)
      expect = ''.join([f"{v[1]}" for v in sorted_expect])
  
      def check_section_exists(text: str) -> str:
          try:
              return str(sections_headings[text][0][1])
          except (IndexError, KeyError):
              return ''
  
      setup = check_section_exists('<h1>Setup</h1>')
      cleanup = check_section_exists('<h1>Cleanup</h1>')
  
      return step, expect, setup, cleanup
  
  
  def bz_set_coverage(bug_ids: List[int], case_id: str, tracker_id: int) -> None:
      """ Set coverage in Bugzilla """
      bz_instance = get_bz_instance()
  
      overall_pass = True
      no_email = 1  # Do not send emails about the change
      get_bz_dict = {
          'ids': bug_ids,
          'include_fields': ['id', 'external_bugs', 'flags']}
      bugs_data = bz_instance._proxy.Bug.get(get_bz_dict)
      for bug in bugs_data['bugs']:
          # Process flag (might fail for some types)
          bug_id = bug['id']
          if 'qe_test_coverage+' not in set(
                  [x['name'] + x['status'] for x in bug['flags']]):
              try:
                  bz_instance._proxy.Flag.update({
                      'ids': [bug_id],
                      'nomail': no_email,
                      'updates': [{
                          'name': 'qe_test_coverage',
                          'status': '+'
                          }]
                      })
              except xmlrpc.client.Fault as err:
                  # TODO: Fix missing overall_result = False, breaks tests
                  # if bug used for testing is not changed
                  # BZ#1925518 can't have qe_test_coverage flag
                  log.debug(f"Update flag failed: {err}")
                  echo(style(
                      f"Failed to set qe_test_coverage+ flag for BZ#{bug_id}",
                      fg='red'))
          # Process external tracker - should succeed
          current = set(
              [b['ext_bz_bug_id'] for b in bug['external_bugs']
               if b['ext_bz_id'] == tracker_id])
          if case_id not in current:
              query = {
                  'bug_ids': [bug_id],
                  'nomail': no_email,
                  'external_bugs': [{
                      'ext_type_id': tracker_id,
                      'ext_bz_bug_id': case_id,
                      'ext_description': '',
                      }]
                  }
              try:
                  bz_instance._proxy.ExternalBugs.add_external_bug(query)
              except Exception as err:
                  log.debug(f"Link case failed: {err}")
                  echo(style(f"Failed to link to BZ#{bug_id}", fg='red'))
                  overall_pass = False
      if not overall_pass:
          raise ConvertError("Failed to link the case to bugs.")
  
      echo(style("Linked to Bugzilla: {}.".format(
          " ".join([f"BZ#{bz_id}" for bz_id in bug_ids])), fg='magenta'))
  
  
  def get_polarion_ids(
          query_result: List[Any],
          preferred_project: Optional[str] = None) -> Tuple[str, Optional[str]]:
      """ Return case and project ids from query results """
      if not query_result:
          return 'None', None
      if len(query_result) == 1:
          return query_result[0].work_item_id, query_result[0].project_id
  
      if preferred_project:
          try:
              return next(
                  item.work_item_id for item in query_result
                  if item.project_id == preferred_project), preferred_project
          except StopIteration:
              pass
  
      for result in query_result:
          # If multiple cases are found prefer cases from other projects
          # than these legacy ones
          if str(result.project_id) not in LEGACY_POLARION_PROJECTS:
              return result.work_item_id, result.project_id
  
      return query_result[0].work_item_id, query_result[0].project_id
  
  
  def find_polarion_case_ids(
          data: Dict[str, str],
          preferred_project: Optional[str] = None) -> Tuple[str, Optional[str]]:
      """ Find IDs for Polarion case from data dictionary """
      assert PolarionWorkItem
  
      # Search by UUID
      if data.get(ID_KEY):
          query_result = PolarionWorkItem.query(
              data.get(ID_KEY), fields=['work_item_id', 'project_id'])
          return get_polarion_ids(query_result, preferred_project)
      # Search by TCMS Case ID
      extra_nitrate = data.get('extra-nitrate')
      if extra_nitrate:
          nitrate_case_id_search = re.search(r'\d+', extra_nitrate)
          if not nitrate_case_id_search:
              raise ConvertError(
                  "Could not find a valid nitrate testcase ID in 'extra-nitrate' attribute")
          nitrate_case_id = str(int(nitrate_case_id_search.group()))
          query_result = PolarionWorkItem.query(
              f"tcmscaseid:{nitrate_case_id}", fields=['work_item_id', 'project_id'])
          return get_polarion_ids(query_result, preferred_project)
      # Search by extra task
      if data.get('extra-task'):
          query_result = PolarionWorkItem.query(
              data.get('extra-task'), fields=['work_item_id', 'project_id'])
          return get_polarion_ids(query_result, preferred_project)
      return 'None', None
  
  
  def get_polarion_case(data: Dict[str, str], preferred_project: Optional[str] = None) -> Any:
      """ Get Polarion case through couple different methods """
      import_polarion()
  
      assert PolarionTestCase
      assert PolarionException
  
      polarion_id, project_id = find_polarion_case_ids(data, preferred_project)
  
      try:
          polarion_case = PolarionTestCase(
              project_id=project_id, work_item_id=polarion_id)
          echo(style(
              f"Test case '{str(polarion_case.work_item_id)}' found.",
              fg='blue'))
          return polarion_case
      except PolarionException:
          return None
  
  
  def enabled_somewhere(test: 'tmt.Test') -> bool:
      """ True if the test is enabled for some context (adjust rules) """
      # node.original_data are fmf data _before_ adjust was processed
      try:
          if test.node.original_data['enabled']:
              return True
      except KeyError:
          # Key defaults to True so we already know the outcome
          return True
  
      # Some rule in adjust enables the test
      try:
          adjust_rules = test.node.original_data['adjust']
          # TODO: Should not be necessary once we normalize data
          if isinstance(adjust_rules, dict):
              adjust_rules = [adjust_rules]
          for rule in adjust_rules:
              try:
                  if rule['enabled']:
                      return True
              except KeyError:
                  pass
      except KeyError:
          pass
      # At this point nothing enables the test
      return False
  
  
  def export_to_nitrate(test: 'tmt.Test') -> None:
      """ Export fmf metadata to nitrate test cases """
      import_nitrate()
      assert nitrate
      assert gssapi
  
      # Check command line options
      create = test.opt('create')
      general = test.opt('general')
      link_runs = test.opt('link_runs')
      duplicate = test.opt('duplicate')
      link_bugzilla = test.opt('bugzilla')
      ignore_git_validation = test.opt('ignore_git_validation')
      dry_mode = test.opt('dry')
  
      if link_runs:
          general = True
  
      # Check git is already correct
      valid, error_msg = tmt.utils.validate_git_status(test)
      if not valid:
          if ignore_git_validation:
              secho(f"Exporting regardless '{error_msg}'.", fg='red')
          else:
              raise ConvertError(
                  f"Can't export due '{error_msg}'.\n"
                  "Use --ignore-git-validation on your own risk to export regardless.")
  
      # Check nitrate test case
      try:
          nitrate_id = test.node.get('extra-nitrate')[3:]
          nitrate_case = nitrate.TestCase(int(nitrate_id))
          nitrate_case.summary  # Make sure we connect to the server now
          echo(style(f"Test case '{nitrate_case.identifier}' found.", fg='blue'))
      except TypeError:
          # Create a new nitrate test case
          if create:
              nitrate_case = None
              # Check for existing Nitrate tests with the same fmf id
              if not duplicate:
                  testcases = _nitrate_find_fmf_testcases(test)
                  try:
                      # Select the first found testcase if any
                      nitrate_case = next(testcases)
                  except StopIteration:
                      pass
              if not nitrate_case:
                  # Summary for TCMS case
                  extra_summary = prepare_extra_summary(test)
                  if not dry_mode:
                      nitrate_case = create_nitrate_case(extra_summary)
                  else:
                      echo(style(
                          f"Test case '{extra_summary}' created.", fg='blue'))
                  test._metadata['extra-summary'] = extra_summary
              # Either newly created or duplicate with missing extra-nitrate
              if nitrate_case:
                  echo(style("Append the nitrate test case id.", fg='green'))
                  if not dry_mode:
                      with test.node as data:
                          data["extra-nitrate"] = nitrate_case.identifier
          else:
              raise ConvertError(f"Nitrate test case id not found for {test}"
                                 " (You can use --create option to enforce"
                                 " creating testcases)")
      except (nitrate.NitrateError, gssapi.raw.misc.GSSError) as error:
          raise ConvertError(error)
  
      # Check if URL is accessible, to be able to reach from nitrate
      check_git_url(test.fmf_id['url'])
  
      # Summary
      try:
          summary = (test._metadata.get('extra-summary')
                     or test._metadata.get('extra-task')
                     or prepare_extra_summary(test))
      except ConvertError:
          summary = test.name
      if not dry_mode:
          nitrate_case.summary = summary
      echo(style('summary: ', fg='green') + summary)
  
      # Script
      if test.node.get('extra-task'):
          if not dry_mode:
              nitrate_case.script = test.node.get('extra-task')
          echo(style('script: ', fg='green') + test.node.get('extra-task'))
  
      # Components and General plan
      # First remove any components that are already there
      if not dry_mode:
          nitrate_case.components.clear()
      # Only these general plans should stay
      expected_general_plans = set()
      # Then add fmf ones
      if test.component:
          echo(style('components: ', fg='green') + ' '.join(test.component))
          for component in test.component:
              try:
                  nitrate_component = nitrate.Component(
                      name=component, product=DEFAULT_PRODUCT.id)
                  if not dry_mode:
                      nitrate_case.components.add(nitrate_component)
              except nitrate.xmlrpc_driver.NitrateError as error:
                  log.debug(error)
                  echo(style(
                      f"Failed to add component '{component}'.", fg='red'))
              if general:
                  try:
                      general_plan = find_general_plan(component)
                      expected_general_plans.add(general_plan)
                      echo(style(
                          f"Linked to general plan '{general_plan}'.",
                          fg='magenta'))
                      if not dry_mode:
                          nitrate_case.testplans.add(general_plan)
                      if link_runs:
                          add_to_nitrate_runs(
                              nitrate_case, general_plan, test, dry_mode)
                  except nitrate.NitrateError as error:
                      log.debug(error)
                      echo(style(
                          f"Failed to find general test plan for '{component}'.",
                          fg='red'))
      # Remove unexpected general plans
      if general and nitrate_case:
          # Remove also all general plans linked to testcase
          for nitrate_plan in [plan for plan in nitrate_case.testplans]:
              if (nitrate_plan.type.name == "General"
                      and nitrate_plan not in expected_general_plans):
                  echo(style(
                      f"Removed general plan '{nitrate_plan}'.", fg='red'))
                  if not dry_mode:
                      nitrate_case.testplans.remove(nitrate_plan)
  
      # Tags
      # Convert 'tier' attribute into a Tier tag
      if test.tier is not None:
          test.tag.append(f"Tier{test.tier}")
      # Add special fmf-export tag
      test.tag.append('fmf-export')
      if not dry_mode:
          nitrate_case.tags.clear()
          nitrate_case.tags.add([nitrate.Tag(tag) for tag in test.tag])
      echo(style('tags: ', fg='green') + ' '.join(set(test.tag)))
  
      # Default tester
      if test.contact:
          try:
              # Need to pick one value, so picking the first contact
              email_address = email.utils.parseaddr(test.contact[0])[1]
              nitrate_user = nitrate.User(email_address)
              nitrate_user._fetch()  # To check that user exists
              if not dry_mode:
                  nitrate_case.tester = nitrate_user
              echo(style('default tester: ', fg='green') + email_address)
          except nitrate.NitrateError as error:
              log.debug(error)
              raise ConvertError(f"Nitrate issue: {error}")
  
      # Duration
      if not dry_mode:
          nitrate_case.time = test.duration
      echo(style('estimated time: ', fg='green') + test.duration)
  
      # Manual
      if not dry_mode:
          nitrate_case.automated = not test.manual
      echo(style('automated: ', fg='green') + ['auto', 'manual'][test.manual])
  
      # Status
      current_status = nitrate_case.status if nitrate_case else nitrate.CaseStatus('CONFIRMED')
      # Enable enabled tests
      if enabled_somewhere(test):
          if not dry_mode:
              nitrate_case.status = nitrate.CaseStatus('CONFIRMED')
          echo(style('status: ', fg='green') + 'CONFIRMED')
      # Disable disabled tests which are CONFIRMED
      elif current_status == nitrate.CaseStatus('CONFIRMED'):
          if not dry_mode:
              nitrate_case.status = nitrate.CaseStatus('DISABLED')
          echo(style('status: ', fg='green') + 'DISABLED')
      # Keep disabled tests in their states
      else:
          echo(style('status: ', fg='green') + str(current_status))
  
      # Environment
      if test.environment:
          environment = ' '.join(tmt.utils.shell_variables(test.environment))
          if not dry_mode:
              nitrate_case.arguments = environment
          echo(style('arguments: ', fg='green') + environment)
      else:
          # FIXME unable clear to set empty arguments
          # (possibly error in xmlrpc, BZ#1805687)
          if not dry_mode:
              nitrate_case.arguments = ' '
          echo(style('arguments: ', fg='green') + "' '")
  
      # Structured Field
      struct_field = tmt.utils.StructuredField(
          nitrate_case.notes if nitrate_case else '')
      echo(style('Structured Field: ', fg='green'))
  
      # Mapping of structured field sections to fmf case attributes
      section_to_attr = {
          'description': test.summary,
          'purpose-file': test.description,
          'hardware': test.node.get('extra-hardware'),
          'pepa': test.node.get('extra-pepa'),
          }
      for section, attribute in section_to_attr.items():
          if attribute is None:
              try:
                  struct_field.remove(section)
              except tmt.utils.StructuredFieldError:
                  pass
          else:
              struct_field.set
  
  [... the rest is omitted ...]
files: tmt/export.py
files: # coding: utf-8
  
  """ Export metadata into nitrate """
  
  
  import email.utils
  import os
  import re
  import traceback
  import types
  import xmlrpc.client
  from functools import lru_cache
  from typing import Any, Dict, Generator, List, Optional, Tuple, Union, cast
  
  import fmf
  from click import echo, secho, style
  
  import tmt
  import tmt.utils
  from tmt.convert import SYSTEM_OTHER, add_link
  from tmt.identifier import ID_KEY, add_uuid_if_not_defined
  from tmt.utils import ConvertError, check_git_url, markdown_to_html
  
  bugzilla: Optional[types.ModuleType] = None
  gssapi: Optional[types.ModuleType] = None
  nitrate: Optional[types.ModuleType] = None
  
  PolarionException: Any = None
  PolarionTestCase: Any = None
  PolarionWorkItem: Any = None
  
  DEFAULT_PRODUCT: Any = None
  
  SectionsReturnType = Tuple[str, str, str, str]
  HeadingsType = List[List[Union[int, str]]]
  SectionsHeadingsType = Dict[str, HeadingsType]
  
  log = fmf.utils.Logging('tmt').logger
  
  WARNING = """
  Test case has been migrated to git. Any changes made here might be overwritten.
  See: https://tmt.readthedocs.io/en/latest/questions.html#nitrate-migration
  """.lstrip()
  
  # For linking bugs
  BUGZILLA_XMLRPC_URL = "https://bugzilla.redhat.com/xmlrpc.cgi"
  NITRATE_TRACKER_ID = 69  # ID of nitrate in RH's bugzilla
  POLARION_TRACKER_ID = 117  # ID of polarion in RH's bugzilla
  RE_BUGZILLA_URL = r'bugzilla.redhat.com/show_bug.cgi\?id=(\d+)'
  RE_POLARION_URL = r'.*/polarion/#/project/.*/workitem\?id=(.*)'
  LEGACY_POLARION_PROJECTS = set(['RedHatEnterpriseLinux7'])
  
  
  def import_nitrate() -> Any:
      """ Conditionally import the nitrate module """
      # Need to import nitrate only when really needed. Otherwise we get
      # traceback when nitrate not installed or config file not available.
      # And we want to keep the core tmt package with minimal dependencies.
      try:
          global nitrate, DEFAULT_PRODUCT, gssapi
          import gssapi
          import nitrate
          assert nitrate
          DEFAULT_PRODUCT = nitrate.Product(name='RHEL Tests')
          return nitrate
      except ImportError:
          raise ConvertError(
              "Install tmt-test-convert to export tests to nitrate.")
      except nitrate.NitrateError as error:  # type: ignore
          raise ConvertError(error)
  
  
  def import_polarion() -> None:
      """ Import polarion python api - pylero """
      try:
          global PolarionException, PolarionTestCase, PolarionWorkItem
          from pylero.exceptions import PyleroLibException as PolarionException
          from pylero.work_item import TestCase as PolarionTestCase
          from pylero.work_item import _WorkItem as PolarionWorkItem
      except ImportError:
          raise ConvertError(
              "Run 'pip install tmt[export-polarion]' so pylero is installed.")
      except PolarionException as exc:
          log.debug(traceback.format_exc())
          raise ConvertError("Failed to login with pylero", original=exc)
  
  
  def get_bz_instance() -> Any:
      """ Import the bugzilla module and return BZ instance """
      try:
          import bugzilla
      except ImportError:
          raise ConvertError(
              "Install 'tmt-test-convert' to link test to the bugzilla.")
  
      try:
          bz_instance = bugzilla.Bugzilla(url=BUGZILLA_XMLRPC_URL)
      except Exception as exc:
          log.debug(traceback.format_exc())
          raise ConvertError(
              "Couldn't initialize the Bugzilla client.", original=exc)
  
      if not bz_instance.logged_in:
          raise ConvertError(
              "Not logged to Bugzilla, check 'man bugzilla' section "
              "'AUTHENTICATION CACHE AND API KEYS'.")
      return bz_instance
  
  
  def _nitrate_find_fmf_testcases(test: 'tmt.Test') -> Generator[Any, None, None]:
      """
      Find all Nitrate test cases with the same fmf identifier
  
      All component general plans are explored for possible duplicates.
      """
      assert nitrate
      for component in test.component:
          try:
              for testcase in find_general_plan(component).testcases:
                  struct_field = tmt.utils.StructuredField(testcase.notes)
                  try:
                      fmf_id = tmt.utils.yaml_to_dict(struct_field.get('fmf'))
                      if fmf_id == test.fmf_id:
                          echo(style(
                              f"Existing test case '{testcase.identifier}' "
                              f"found for given fmf id.", fg='magenta'))
                          yield testcase
                  except tmt.utils.StructuredFieldError:
                      pass
          except nitrate.NitrateError:
              pass
  
  
  def convert_manual_to_nitrate(test_md: str) -> SectionsReturnType:
      """
      Convert Markdown document to html sections.
  
      These sections can be exported to nitrate.
      Expects: Markdown document as a file.
      Returns: tuple of (step, expect, setup, cleanup) sections
      as html strings.
      """
  
      sections_headings: SectionsHeadingsType = {
          heading: []
          for heading_list in tmt.base.SECTIONS_HEADINGS.values()
          for heading in heading_list
          }
  
      html = markdown_to_html(test_md)
      html_splitlines = html.splitlines()
  
      for key in sections_headings.keys():
          result: HeadingsType = []
          i = 0
          while html_splitlines:
              try:
                  if re.search("^" + key + "$", html_splitlines[i]):
                      html_content = str()
                      if key.startswith('<h1>Test'):
                          html_content = html_splitlines[i].\
                              replace('<h1>', '<b>').\
                              replace('</h1>', '</b>')
                      for j in range(i + 1, len(html_splitlines)):
                          if re.search("^<h[1-4]>(.+?)</h[1-4]>$",
                                       html_splitlines[j]):
                              result.append([i, html_content])
                              i = j - 1
                              break
                          html_content += html_splitlines[j] + "\n"
                          # Check end of the file
                          if j + 1 == len(html_splitlines):
                              result.append([i, html_content])
              except IndexError:
                  sections_headings[key] = result
                  break
              i += 1
              if i >= len(html_splitlines):
                  sections_headings[key] = result
                  break
  
      def concatenate_headings_content(headings: Tuple[str, ...]) -> HeadingsType:
          content = list()
          for v in headings:
              content += sections_headings[v]
          return content
  
      def enumerate_content(content: HeadingsType) -> HeadingsType:
          # for sorting convert the index to integer, but keep whole list as list of strings
          content.sort(key=lambda a: int(a[0]))
          for c in range(len(content)):
              content[c][1] = f"<p>Step {c + 1}.</p>" + str(content[c][1])
          return content
  
      sorted_test = sorted(concatenate_headings_content((
          '<h1>Test</h1>',
          '<h1>Test .*</h1>')))
  
      sorted_step = sorted(enumerate_content(concatenate_headings_content((
          '<h2>Step</h2>',
          '<h2>Test Step</h2>'))) + sorted_test)
      step = ''.join([f"{v[1]}" for v in sorted_step])
  
      sorted_expect = sorted(enumerate_content(concatenate_headings_content((
          '<h2>Expect</h2>',
          '<h2>Result</h2>',
          '<h2>Expected Result</h2>'))) + sorted_test)
      expect = ''.join([f"{v[1]}" for v in sorted_expect])
  
      def check_section_exists(text: str) -> str:
          try:
              return str(sections_headings[text][0][1])
          except (IndexError, KeyError):
              return ''
  
      setup = check_section_exists('<h1>Setup</h1>')
      cleanup = check_section_exists('<h1>Cleanup</h1>')
  
      return step, expect, setup, cleanup
  
  
  def bz_set_coverage(bug_ids: List[int], case_id: str, tracker_id: int) -> None:
      """ Set coverage in Bugzilla """
      bz_instance = get_bz_instance()
  
      overall_pass = True
      no_email = 1  # Do not send emails about the change
      get_bz_dict = {
          'ids': bug_ids,
          'include_fields': ['id', 'external_bugs', 'flags']}
      bugs_data = bz_instance._proxy.Bug.get(get_bz_dict)
      for bug in bugs_data['bugs']:
          # Process flag (might fail for some types)
          bug_id = bug['id']
          if 'qe_test_coverage+' not in set(
                  [x['name'] + x['status'] for x in bug['flags']]):
              try:
                  bz_instance._proxy.Flag.update({
                      'ids': [bug_id],
                      'nomail': no_email,
                      'updates': [{
                          'name': 'qe_test_coverage',
                          'status': '+'
                          }]
                      })
              except xmlrpc.client.Fault as err:
                  # TODO: Fix missing overall_result = False, breaks tests
                  # if bug used for testing is not changed
                  # BZ#1925518 can't have qe_test_coverage flag
                  log.debug(f"Update flag failed: {err}")
                  echo(style(
                      f"Failed to set qe_test_coverage+ flag for BZ#{bug_id}",
                      fg='red'))
          # Process external tracker - should succeed
          current = set(
              [b['ext_bz_bug_id'] for b in bug['external_bugs']
               if b['ext_bz_id'] == tracker_id])
          if case_id not in current:
              query = {
                  'bug_ids': [bug_id],
                  'nomail': no_email,
                  'external_bugs': [{
                      'ext_type_id': tracker_id,
                      'ext_bz_bug_id': case_id,
                      'ext_description': '',
                      }]
                  }
              try:
                  bz_instance._proxy.ExternalBugs.add_external_bug(query)
              except Exception as err:
                  log.debug(f"Link case failed: {err}")
                  echo(style(f"Failed to link to BZ#{bug_id}", fg='red'))
                  overall_pass = False
      if not overall_pass:
          raise ConvertError("Failed to link the case to bugs.")
  
      echo(style("Linked to Bugzilla: {}.".format(
          " ".join([f"BZ#{bz_id}" for bz_id in bug_ids])), fg='magenta'))
  
  
  def get_polarion_ids(
          query_result: List[Any],
          preferred_project: Optional[str] = None) -> Tuple[str, Optional[str]]:
      """ Return case and project ids from query results """
      if not query_result:
          return 'None', None
      if len(query_result) == 1:
          return query_result[0].work_item_id, query_result[0].project_id
  
      if preferred_project:
          try:
              return next(
                  item.work_item_id for item in query_result
                  if item.project_id == preferred_project), preferred_project
          except StopIteration:
              pass
  
      for result in query_result:
          # If multiple cases are found prefer cases from other projects
          # than these legacy ones
          if str(result.project_id) not in LEGACY_POLARION_PROJECTS:
              return result.work_item_id, result.project_id
  
      return query_result[0].work_item_id, query_result[0].project_id
  
  
  def get_polarion_case(data: Dict[str, str], preferred_project: Optional[str] = None) -> Any:
      """ Get Polarion case through couple different methods """
      import_polarion()
      polarion_id = 'None'
      project_id = None
  
      assert PolarionWorkItem
      assert PolarionTestCase
      assert PolarionException
  
      # Search by UUID
      if data.get(ID_KEY):
          query_result = PolarionWorkItem.query(
              data.get(ID_KEY), fields=['work_item_id', 'project_id'])
          polarion_id, project_id = get_polarion_ids(query_result, preferred_project)
      # Search by TCMS Case ID
      extra_nitrate = data.get('extra-nitrate')
      if not project_id and extra_nitrate:
          nitrate_case_id_search = re.search(r'\d+', extra_nitrate)
          if not nitrate_case_id_search:
              raise ConvertError(
                  "Could not find a valid nitrate testcase ID in 'extra-nitrate' attribute")
          nitrate_case_id = str(int(nitrate_case_id_search.group()))
          query_result = PolarionWorkItem.query(
              f"tcmscaseid:{nitrate_case_id}", fields=['work_item_id', 'project_id'])
          polarion_id, project_id = get_polarion_ids(query_result, preferred_project)
      # Search by extra task
      if not project_id and data.get('extra-task'):
          query_result = PolarionWorkItem.query(
              data.get('extra-task'), fields=['work_item_id', 'project_id'])
          polarion_id, project_id = get_polarion_ids(query_result, preferred_project)
  
      try:
          polarion_case = PolarionTestCase(
              project_id=project_id, work_item_id=polarion_id)
          echo(style(
              f"Test case '{str(polarion_case.work_item_id)}' found.",
              fg='blue'))
          return polarion_case
      except PolarionException:
          return None
  
  
  def enabled_somewhere(test: 'tmt.Test') -> bool:
      """ True if the test is enabled for some context (adjust rules) """
      # node.original_data are fmf data _before_ adjust was processed
      try:
          if test.node.original_data['enabled']:
              return True
      except KeyError:
          # Key defaults to True so we already know the outcome
          return True
  
      # Some rule in adjust enables the test
      try:
          adjust_rules = test.node.original_data['adjust']
          # TODO: Should not be necessary once we normalize data
          if isinstance(adjust_rules, dict):
              adjust_rules = [adjust_rules]
          for rule in adjust_rules:
              try:
                  if rule['enabled']:
                      return True
              except KeyError:
                  pass
      except KeyError:
          pass
      # At this point nothing enables the test
      return False
  
  
  def export_to_nitrate(test: 'tmt.Test') -> None:
      """ Export fmf metadata to nitrate test cases """
      import_nitrate()
      assert nitrate
      assert gssapi
  
      # Check command line options
      create = test.opt('create')
      general = test.opt('general')
      link_runs = test.opt('link_runs')
      duplicate = test.opt('duplicate')
      link_bugzilla = test.opt('bugzilla')
      ignore_git_validation = test.opt('ignore_git_validation')
      dry_mode = test.opt('dry')
  
      if link_runs:
          general = True
  
      # Check git is already correct
      valid, error_msg = tmt.utils.validate_git_status(test)
      if not valid:
          if ignore_git_validation:
              secho(f"Exporting regardless '{error_msg}'.", fg='red')
          else:
              raise ConvertError(
                  f"Can't export due '{error_msg}'.\n"
                  "Use --ignore-git-validation on your own risk to export regardless.")
  
      # Check nitrate test case
      try:
          nitrate_id = test.node.get('extra-nitrate')[3:]
          nitrate_case = nitrate.TestCase(int(nitrate_id))
          nitrate_case.summary  # Make sure we connect to the server now
          echo(style(f"Test case '{nitrate_case.identifier}' found.", fg='blue'))
      except TypeError:
          # Create a new nitrate test case
          if create:
              nitrate_case = None
              # Check for existing Nitrate tests with the same fmf id
              if not duplicate:
                  testcases = _nitrate_find_fmf_testcases(test)
                  try:
                      # Select the first found testcase if any
                      nitrate_case = next(testcases)
                  except StopIteration:
                      pass
              if not nitrate_case:
                  # Summary for TCMS case
                  extra_summary = prepare_extra_summary(test)
                  if not dry_mode:
                      nitrate_case = create_nitrate_case(extra_summary)
                  else:
                      echo(style(
                          f"Test case '{extra_summary}' created.", fg='blue'))
                  test._metadata['extra-summary'] = extra_summary
              # Either newly created or duplicate with missing extra-nitrate
              if nitrate_case:
                  echo(style("Append the nitrate test case id.", fg='green'))
                  if not dry_mode:
                      with test.node as data:
                          data["extra-nitrate"] = nitrate_case.identifier
          else:
              raise ConvertError(f"Nitrate test case id not found for {test}"
                                 " (You can use --create option to enforce"
                                 " creating testcases)")
      except (nitrate.NitrateError, gssapi.raw.misc.GSSError) as error:
          raise ConvertError(error)
  
      # Check if URL is accessible, to be able to reach from nitrate
      check_git_url(test.fmf_id['url'])
  
      # Summary
      try:
          summary = (test._metadata.get('extra-summary')
                     or test._metadata.get('extra-task')
                     or prepare_extra_summary(test))
      except ConvertError:
          summary = test.name
      if not dry_mode:
          nitrate_case.summary = summary
      echo(style('summary: ', fg='green') + summary)
  
      # Script
      if test.node.get('extra-task'):
          if not dry_mode:
              nitrate_case.script = test.node.get('extra-task')
          echo(style('script: ', fg='green') + test.node.get('extra-task'))
  
      # Components and General plan
      # First remove any components that are already there
      if not dry_mode:
          nitrate_case.components.clear()
      # Only these general plans should stay
      expected_general_plans = set()
      # Then add fmf ones
      if test.component:
          echo(style('components: ', fg='green') + ' '.join(test.component))
          for component in test.component:
              try:
                  nitrate_component = nitrate.Component(
                      name=component, product=DEFAULT_PRODUCT.id)
                  if not dry_mode:
                      nitrate_case.components.add(nitrate_component)
              except nitrate.xmlrpc_driver.NitrateError as error:
                  log.debug(error)
                  echo(style(
                      f"Failed to add component '{component}'.", fg='red'))
              if general:
                  try:
                      general_plan = find_general_plan(component)
                      expected_general_plans.add(general_plan)
                      echo(style(
                          f"Linked to general plan '{general_plan}'.",
                          fg='magenta'))
                      if not dry_mode:
                          nitrate_case.testplans.add(general_plan)
                      if link_runs:
                          add_to_nitrate_runs(
                              nitrate_case, general_plan, test, dry_mode)
                  except nitrate.NitrateError as error:
                      log.debug(error)
                      echo(style(
                          f"Failed to find general test plan for '{component}'.",
                          fg='red'))
      # Remove unexpected general plans
      if general and nitrate_case:
          # Remove also all general plans linked to testcase
          for nitrate_plan in [plan for plan in nitrate_case.testplans]:
              if (nitrate_plan.type.name == "General"
                      and nitrate_plan not in expected_general_plans):
                  echo(style(
                      f"Removed general plan '{nitrate_plan}'.", fg='red'))
                  if not dry_mode:
                      nitrate_case.testplans.remove(nitrate_plan)
  
      # Tags
      # Convert 'tier' attribute into a Tier tag
      if test.tier is not None:
          test.tag.append(f"Tier{test.tier}")
      # Add special fmf-export tag
      test.tag.append('fmf-export')
      if not dry_mode:
          nitrate_case.tags.clear()
          nitrate_case.tags.add([nitrate.Tag(tag) for tag in test.tag])
      echo(style('tags: ', fg='green') + ' '.join(set(test.tag)))
  
      # Default tester
      if test.contact:
          try:
              # Need to pick one value, so picking the first contact
              email_address = email.utils.parseaddr(test.contact[0])[1]
              nitrate_user = nitrate.User(email_address)
              nitrate_user._fetch()  # To check that user exists
              if not dry_mode:
                  nitrate_case.tester = nitrate_user
              echo(style('default tester: ', fg='green') + email_address)
          except nitrate.NitrateError as error:
              log.debug(error)
              raise ConvertError(f"Nitrate issue: {error}")
  
      # Duration
      if not dry_mode:
          nitrate_case.time = test.duration
      echo(style('estimated time: ', fg='green') + test.duration)
  
      # Manual
      if not dry_mode:
          nitrate_case.automated = not test.manual
      echo(style('automated: ', fg='green') + ['auto', 'manual'][test.manual])
  
      # Status
      current_status = nitrate_case.status if nitrate_case else nitrate.CaseStatus('CONFIRMED')
      # Enable enabled tests
      if enabled_somewhere(test):
          if not dry_mode:
              nitrate_case.status = nitrate.CaseStatus('CONFIRMED')
          echo(style('status: ', fg='green') + 'CONFIRMED')
      # Disable disabled tests which are CONFIRMED
      elif current_status == nitrate.CaseStatus('CONFIRMED'):
          if not dry_mode:
              nitrate_case.status = nitrate.CaseStatus('DISABLED')
          echo(style('status: ', fg='green') + 'DISABLED')
      # Keep disabled tests in their states
      else:
          echo(style('status: ', fg='green') + str(current_status))
  
      # Environment
      if test.environment:
          environment = ' '.join(tmt.utils.shell_variables(test.environment))
          if not dry_mode:
              nitrate_case.arguments = environment
          echo(style('arguments: ', fg='green') + environment)
      else:
          # FIXME unable clear to set empty arguments
          # (possibly error in xmlrpc, BZ#1805687)
          if not dry_mode:
              nitrate_case.arguments = ' '
          echo(style('arguments: ', fg='green') + "' '")
  
      # Structured Field
      struct_field = tmt.utils.StructuredField(
          nitrate_case.notes if nitrate_case else '')
      echo(style('Structured Field: ', fg='green'))
  
      # Mapping of structured field sections to fmf case attributes
      section_to_attr = {
          'description': test.summary,
          'purpose-file': test.description,
          'hardware': test.node.get('extra-hardware'),
          'pepa': test.node.get('extra-pepa'),
          }
      for section, attribute in section_to_attr.items():
          if attribute is None:
              try:
                  struct_field.remove(section)
              except tmt.utils.StructuredFieldError:
                  pass
          else:
              struct_field.set(section, attribute)
              echo(style(section + ': ', fg='green') + attribute.strip())
  
      # fmf identifer
      fm
  
  [... the rest is omitted ...]
files: tmt/export.py
files: import datetime
  import os
  import xml.etree.ElementTree as ET
  from typing import Any, List, Optional
  
  import click
  from requests import post
  
  import tmt
  import tmt.steps
  
  from .junit import make_junit_xml
  
  DEFAULT_NAME = 'xunit.xml'
  
  
  @tmt.steps.provides_method('polarion')
  class ReportPolarion(tmt.steps.report.ReportPlugin):
      """
      Write test results into a xUnit file and upload to Polarion
      """
      _keys = ['file', 'no-upload', 'project-id', 'testrun-title']
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '--file', metavar='FILE', help='Path to the file to store xUnit in'),
              click.option(
                  '--upload / --no-upload', default=True, show_default=True,
                  help="Whether to upload results to Polarion"),
              click.option(
                  '--project-id', required=True, help='Use specific Polarion project ID'),
              click.option(
                  '--testrun-title', help='Use specific TestRun title')
              ]
          return options
  
      def go(self, *args: Any, **kwargs: Any) -> None:
          """ Go through executed tests and report into Polarion """
          super().go()
  
          from tmt.export import find_polarion_case_ids, import_polarion
          import_polarion()
          from tmt.export import PolarionWorkItem
          assert PolarionWorkItem
  
          title = self.get(
              'testrun_title',
              self.step.plan.name.rsplit('/', 1)[1] +
              datetime.datetime.now().strftime("%Y%m%d%H%M%S"))
          title = title.replace('-', '_')
          project_id = self.get('project-id')
          upload = self.get('upload')
  
          junit_suite = make_junit_xml(self)
          xml_tree = ET.fromstring(junit_suite.to_xml_string([junit_suite]))
  
          properties = {
              'polarion-project-id': project_id,
              'polarion-user-id': PolarionWorkItem._session.user_id,
              'polarion-testrun-id': title,
              'polarion-project-span-ids': project_id}
          testsuites_properties = ET.SubElement(xml_tree, 'properties')
          for name, value in properties.items():
              ET.SubElement(testsuites_properties, 'property', attrib={
                  'name': name, 'value': value})
  
          testsuite = xml_tree.find('testsuite')
          project_span_ids = xml_tree.find(
              '*property[@name="polarion-project-span-ids"]')
  
          for result in self.step.plan.execute.results():
              if not result.ids or not any(result.ids.values()):
                  raise tmt.utils.ReportError(
                      f"Test Case {result.name} is not exported to Polarion, "
                      "please run 'tmt tests export --how polarion' on it")
              work_item_id, test_project_id = find_polarion_case_ids(result.ids)
  
              if test_project_id is None:
                  raise tmt.utils.ReportError("Test case missing or not found in Polarion")
  
              assert work_item_id is not None
              assert project_span_ids is not None
  
              if test_project_id not in project_span_ids.attrib['value']:
                  project_span_ids.attrib['value'] += f',{test_project_id}'
  
              test_properties = {
                  'polarion-testcase-id': work_item_id,
                  'polarion-testcase-project-id': test_project_id}
  
              assert testsuite is not None
              test_case = testsuite.find(f"*[@name='{result.name}']")
              assert test_case is not None
              properties_elem = ET.SubElement(test_case, 'properties')
              for name, value in test_properties.items():
                  ET.SubElement(properties_elem, 'property', attrib={
                      'name': name, 'value': value})
  
          assert self.workdir is not None
  
          f_path = self.get("file", os.path.join(self.workdir, DEFAULT_NAME))
          with open(f_path, 'wb') as fw:
              ET.ElementTree(xml_tree).write(fw)
  
          if upload:
              server_url = str(PolarionWorkItem._session._server.url)
              polarion_import_url = (
                  f'{server_url}{"" if server_url.endswith("/") else "/"}'
                  'import/xunit')
              auth = (
                  PolarionWorkItem._session.user_id,
                  PolarionWorkItem._session.password)
  
              response = post(
                  polarion_import_url, auth=auth,
                  files={'file': ('xunit.xml', ET.tostring(xml_tree))})
              self.info(
                  f'Response code is {response.status_code} with text: {response.text}')
          else:
              self.info(f"xUnit file saved at: {f_path}")
              self.info("Polarion upload can be done manually using command:")
              self.info(
                  "curl -k -u <USER>:<PASSWORD> -X POST -F file=@<XUNIT_XML_FILE_PATH> "
                  "<POLARION_URL>/polarion/import/xunit")
  
  
  [... the rest is omitted ...]
files: tmt/steps/report/polarion.py
files: import datetime
  import os
  import xml.etree.ElementTree as ET
  from typing import Any, List, Optional
  
  import click
  from requests import post
  
  import tmt
  import tmt.steps
  
  from .junit import make_junit_xml
  
  DEFAULT_NAME = 'xunit.xml'
  
  
  @tmt.steps.provides_method('polarion')
  class ReportPolarion(tmt.steps.report.ReportPlugin):
      """
      Write test results into a xUnit file and upload to Polarion
      """
      _keys = ['file', 'no-upload', 'project-id', 'testrun-title']
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '--file', metavar='FILE', help='Path to the file to store xUnit in'),
              click.option(
                  '--upload / --no-upload', default=True, show_default=True,
                  help="Whether to upload results to Polarion"),
              click.option(
                  '--project-id', required=True, help='Use specific Polarion project ID'),
              click.option(
                  '--testrun-title', help='Use specific TestRun title')
              ]
          return options
  
      def go(self, *args: Any, **kwargs: Any) -> None:
          """ Go through executed tests and report into Polarion """
          super().go()
  
          from tmt.export import get_polarion_ids, import_polarion
          import_polarion()
          from tmt.export import PolarionWorkItem
          assert PolarionWorkItem
  
          title = self.get(
              'testrun_title',
              self.step.plan.name.rsplit('/', 1)[1] +
              datetime.datetime.now().strftime("%Y%m%d%H%M%S"))
          title = title.replace('-', '_')
          project_id = self.get('project-id')
          upload = self.get('upload')
  
          junit_suite = make_junit_xml(self)
          xml_tree = ET.fromstring(junit_suite.to_xml_string([junit_suite]))
  
          properties = {
              'polarion-project-id': project_id,
              'polarion-user-id': PolarionWorkItem._session.user_id,
              'polarion-testrun-id': title,
              'polarion-project-span-ids': project_id}
          testsuites_properties = ET.SubElement(xml_tree, 'properties')
          for name, value in properties.items():
              ET.SubElement(testsuites_properties, 'property', attrib={
                  'name': name, 'value': value})
  
          testsuite = xml_tree.find('testsuite')
          project_span_ids = xml_tree.find(
              '*property[@name="polarion-project-span-ids"]')
  
          for result in self.step.plan.execute.results():
              if not result.id:
                  raise tmt.utils.ReportError(
                      f"Test Case {result.name} is not exported to Polarion, "
                      "please run 'tmt tests export --how polarion' on it")
              work_item_id, test_project_id = get_polarion_ids(
                  PolarionWorkItem.query(
                      result.id, fields=['work_item_id', 'project_id']))
  
              assert work_item_id is not None
              assert test_project_id is not None
              assert project_span_ids is not None
  
              if test_project_id not in project_span_ids.attrib['value']:
                  project_span_ids.attrib['value'] += f',{test_project_id}'
  
              test_properties = {
                  'polarion-testcase-id': work_item_id,
                  'polarion-testcase-project-id': test_project_id}
  
              assert testsuite is not None
              test_case = testsuite.find(f"*[@name='{result.name}']")
              assert test_case is not None
              properties_elem = ET.SubElement(test_case, 'properties')
              for name, value in test_properties.items():
                  ET.SubElement(properties_elem, 'property', attrib={
                      'name': name, 'value': value})
  
          assert self.workdir is not None
  
          f_path = self.get("file", os.path.join(self.workdir, DEFAULT_NAME))
          with open(f_path, 'wb') as fw:
              ET.ElementTree(xml_tree).write(fw)
  
          if upload:
              server_url = str(PolarionWorkItem._session._server.url)
              polarion_import_url = (
                  f'{server_url}{"" if server_url.endswith("/") else "/"}'
                  'import/xunit')
              auth = (
                  PolarionWorkItem._session.user_id,
                  PolarionWorkItem._session.password)
  
              response = post(
                  polarion_import_url, auth=auth,
                  files={'file': ('xunit.xml', ET.tostring(xml_tree))})
              self.info(
                  f'Response code is {response.status_code} with text: {response.text}')
          else:
              self.info(f"xUnit file saved at: {f_path}")
              self.info("Polarion upload can be done manually using command:")
              self.info(
                  "curl -k -u <USER>:<PASSWORD> -X POST -F file=@<XUNIT_XML_FILE_PATH> "
                  "<POLARION_URL>/polarion/import/xunit")
  
  
  [... the rest is omitted ...]
files: tmt/steps/report/polarion.py