files: # coding: utf-8
  
  import datetime
  import os
  import queue
  import re
  import threading
  import time
  import unittest
  import unittest.mock
  
  import pytest
  
  import tmt
  import tmt.plugins
  import tmt.steps.discover
  from tmt.utils import (Common, GeneralError, StructuredField,
                         StructuredFieldError, WaitingIncomplete,
                         WaitingTimedOutError, duration_to_seconds, listify,
                         public_git_url, validate_git_status, wait)
  
  run = Common().run
  
  
  @pytest.fixture
  def local_git_repo(tmpdir):
      origin = tmpdir.join('origin')
      os.makedirs(origin)
  
      run('git init'.split(), cwd=origin)
      run(
          'git config --local user.email lzachar@redhat.com'.split(),
          cwd=origin)
      run(
          'git config --local user.name LZachar'.split(),
          cwd=origin)
      # We need to be able to push, --bare repo is another option here however
      # that would require to add separate fixture for bare repo (unusable for
      # local changes)
      run(
          'git config --local receive.denyCurrentBranch ignore'.split(),
          cwd=origin)
      origin.join('README').write('something to have in the repo')
      run('git add -A'.split(), cwd=origin)
      run(
          'git commit -m initial_commit'.split(),
          cwd=origin)
      return origin
  
  
  @pytest.fixture
  def origin_and_local_git_repo(local_git_repo):
      top_dir = local_git_repo.dirpath()
      fork_dir = top_dir.join('fork')
      run(
          f'git clone {local_git_repo} {fork_dir}'.split(),
          cwd=top_dir)
      run(
          'git config --local user.email lzachar@redhat.com'.split(),
          cwd=fork_dir)
      run(
          'git config --local user.name LZachar'.split(),
          cwd=fork_dir)
      return local_git_repo, fork_dir
  
  
  def test_public_git_url():
      """ Verify url conversion """
      examples = [
          {
              'original': 'git@github.com:teemtee/tmt.git',
              'expected': 'https://github.com/teemtee/tmt.git',
              }, {
              'original': 'ssh://psplicha@pkgs.devel.redhat.com/tests/bash',
              'expected': 'git://pkgs.devel.redhat.com/tests/bash',
              }, {
              'original': 'git+ssh://psplicha@pkgs.devel.redhat.com/tests/bash',
              'expected': 'git://pkgs.devel.redhat.com/tests/bash',
              }, {
              'original': 'ssh://pkgs.devel.redhat.com/tests/bash',
              'expected': 'git://pkgs.devel.redhat.com/tests/bash',
              }, {
              'original': 'git+ssh://psss@pkgs.fedoraproject.org/tests/shell',
              'expected': 'https://pkgs.fedoraproject.org/tests/shell',
              }, {
              'original': 'ssh://psss@pkgs.fedoraproject.org/tests/shell',
              'expected': 'https://pkgs.fedoraproject.org/tests/shell',
              }, {
              'original': 'ssh://git@pagure.io/fedora-ci/metadata.git',
              'expected': 'https://pagure.io/fedora-ci/metadata.git',
              },
          ]
      for example in examples:
          assert public_git_url(example['original']) == example['expected']
  
  
  def test_listify():
      """ Check listify functionality """
      assert listify(['abc']) == ['abc']
      assert listify('abc') == ['abc']
      assert listify('a b c') == ['a b c']
      assert listify('a b c', split=True) == ['a', 'b', 'c']
      assert listify(dict(a=1, b=2)) == dict(a=[1], b=[2])
      assert listify(dict(a=1, b=2), keys=['a']) == dict(a=[1], b=2)
  
  
  def test_config():
      """ Config smoke test """
      run = '/var/tmp/tmt/test'
      config1 = tmt.utils.Config()
      config1.last_run(run)
      config2 = tmt.utils.Config()
      assert config2.last_run() == run
  
  
  def test_last_run_race(tmpdir, monkeypatch):
      """ Race in last run symlink should't be fatal """
      monkeypatch.setattr(tmt.utils, 'CONFIG_PATH', tmpdir.mkdir('config'))
      mock_logger = unittest.mock.MagicMock()
      monkeypatch.setattr(tmt.utils.log, 'warning', mock_logger)
      config = tmt.utils.Config()
      results = queue.Queue()
      threads = []
  
      def create_last_run(config, counter):
          try:
              val = config.last_run(tmpdir.mkdir(f"run-{counter}"))
              results.put(val)
          except Exception as err:
              results.put(err)
  
      total = 20
      for i in range(total):
          threads.append(threading.Thread(target=create_last_run, args=(config, i)))
      for t in threads:
          t.start()
      for t in threads:
          t.join()
  
      all_good = True
      for t in threads:
          value = results.get()
          if isinstance(value, Exception):
              # Print exception for logging
              print(value)
              all_good = False
      assert all_good
      # Getting into race is not certain, do not assert
      # assert mock_logger.called
      assert config.last_run(), "Some run was stored as last run"
  
  
  def test_workdir_root_full(tmpdir, monkeypatch):
      """ Raise if all ids lower than WORKDIR_MAX are exceeded """
      monkeypatch.setattr(tmt.utils, 'WORKDIR_ROOT', tmpdir)
      monkeypatch.setattr(tmt.utils, 'WORKDIR_MAX', 1)
      possible_workdir = str(tmpdir.join
  
  [... the rest is omitted ...]
files: tests/unit/test_utils.py
files: # coding: utf-8
  
  import datetime
  import os
  import queue
  import re
  import threading
  import time
  import unittest
  import unittest.mock
  
  import pytest
  
  import tmt
  from tmt.utils import (Common, GeneralError, StructuredField,
                         StructuredFieldError, WaitingIncomplete,
                         WaitingTimedOutError, duration_to_seconds, listify,
                         public_git_url, validate_git_status, wait)
  
  run = Common().run
  
  
  @pytest.fixture
  def local_git_repo(tmpdir):
      origin = tmpdir.join('origin')
      os.makedirs(origin)
  
      run('git init'.split(), cwd=origin)
      run(
          'git config --local user.email lzachar@redhat.com'.split(),
          cwd=origin)
      run(
          'git config --local user.name LZachar'.split(),
          cwd=origin)
      # We need to be able to push, --bare repo is another option here however
      # that would require to add separate fixture for bare repo (unusable for
      # local changes)
      run(
          'git config --local receive.denyCurrentBranch ignore'.split(),
          cwd=origin)
      origin.join('README').write('something to have in the repo')
      run('git add -A'.split(), cwd=origin)
      run(
          'git commit -m initial_commit'.split(),
          cwd=origin)
      return origin
  
  
  @pytest.fixture
  def origin_and_local_git_repo(local_git_repo):
      top_dir = local_git_repo.dirpath()
      fork_dir = top_dir.join('fork')
      run(
          f'git clone {local_git_repo} {fork_dir}'.split(),
          cwd=top_dir)
      run(
          'git config --local user.email lzachar@redhat.com'.split(),
          cwd=fork_dir)
      run(
          'git config --local user.name LZachar'.split(),
          cwd=fork_dir)
      return local_git_repo, fork_dir
  
  
  def test_public_git_url():
      """ Verify url conversion """
      examples = [
          {
              'original': 'git@github.com:teemtee/tmt.git',
              'expected': 'https://github.com/teemtee/tmt.git',
              }, {
              'original': 'ssh://psplicha@pkgs.devel.redhat.com/tests/bash',
              'expected': 'git://pkgs.devel.redhat.com/tests/bash',
              }, {
              'original': 'git+ssh://psplicha@pkgs.devel.redhat.com/tests/bash',
              'expected': 'git://pkgs.devel.redhat.com/tests/bash',
              }, {
              'original': 'ssh://pkgs.devel.redhat.com/tests/bash',
              'expected': 'git://pkgs.devel.redhat.com/tests/bash',
              }, {
              'original': 'git+ssh://psss@pkgs.fedoraproject.org/tests/shell',
              'expected': 'https://pkgs.fedoraproject.org/tests/shell',
              }, {
              'original': 'ssh://psss@pkgs.fedoraproject.org/tests/shell',
              'expected': 'https://pkgs.fedoraproject.org/tests/shell',
              }, {
              'original': 'ssh://git@pagure.io/fedora-ci/metadata.git',
              'expected': 'https://pagure.io/fedora-ci/metadata.git',
              },
          ]
      for example in examples:
          assert public_git_url(example['original']) == example['expected']
  
  
  def test_listify():
      """ Check listify functionality """
      assert listify(['abc']) == ['abc']
      assert listify('abc') == ['abc']
      assert listify('a b c') == ['a b c']
      assert listify('a b c', split=True) == ['a', 'b', 'c']
      assert listify(dict(a=1, b=2)) == dict(a=[1], b=[2])
      assert listify(dict(a=1, b=2), keys=['a']) == dict(a=[1], b=2)
  
  
  def test_config():
      """ Config smoke test """
      run = '/var/tmp/tmt/test'
      config1 = tmt.utils.Config()
      config1.last_run(run)
      config2 = tmt.utils.Config()
      assert config2.last_run() == run
  
  
  def test_last_run_race(tmpdir, monkeypatch):
      """ Race in last run symlink should't be fatal """
      monkeypatch.setattr(tmt.utils, 'CONFIG_PATH', tmpdir.mkdir('config'))
      mock_logger = unittest.mock.MagicMock()
      monkeypatch.setattr(tmt.utils.log, 'warning', mock_logger)
      config = tmt.utils.Config()
      results = queue.Queue()
      threads = []
  
      def create_last_run(config, counter):
          try:
              val = config.last_run(tmpdir.mkdir(f"run-{counter}"))
              results.put(val)
          except Exception as err:
              results.put(err)
  
      total = 20
      for i in range(total):
          threads.append(threading.Thread(target=create_last_run, args=(config, i)))
      for t in threads:
          t.start()
      for t in threads:
          t.join()
  
      all_good = True
      for t in threads:
          value = results.get()
          if isinstance(value, Exception):
              # Print exception for logging
              print(value)
              all_good = False
      assert all_good
      # Getting into race is not certain, do not assert
      # assert mock_logger.called
      assert config.last_run(), "Some run was stored as last run"
  
  
  def test_workdir_root_full(tmpdir, monkeypatch):
      """ Raise if all ids lower than WORKDIR_MAX are exceeded """
      monkeypatch.setattr(tmt.utils, 'WORKDIR_ROOT', tmpdir)
      monkeypatch.setattr(tmt.utils, 'WORKDIR_MAX', 1)
      possible_workdir = str(tmpdir.join('run-001'))
      # First call success
      common1 =
  
  [... the rest is omitted ...]
files: tests/unit/test_utils.py
files: # coding: utf-8
  
  """ Handle Plugins """
  
  import importlib
  import os
  import pkgutil
  import sys
  from typing import Any, Generator, Optional
  
  if sys.version_info < (3, 9):
      from importlib_metadata import entry_points
  else:
      from importlib.metadata import entry_points
  
  import fmf
  
  import tmt
  from tmt.steps import STEPS
  
  log = fmf.utils.Logging('tmt').logger
  
  # Two possibilities to load additional plugins:
  # entry_points (setup_tools)
  ENTRY_POINT_NAME = 'tmt.plugin'
  # Directories with module in environment variable
  ENVIRONMENT_NAME = 'TMT_PLUGINS'
  
  
  def explore() -> None:
      """ Explore all available plugins """
  
      # Check all tmt steps for native plugins
      root = os.path.dirname(os.path.realpath(tmt.__file__))
      for step in STEPS:
          for module in discover(os.path.join(root, 'steps', step)):
              import_(f'tmt.steps.{step}.{module}')
      # Check for possible plugins in the 'plugins' directory
      for module in discover(os.path.join(root, 'plugins')):
          import_(f'tmt.plugins.{module}')
  
      # Check environment variable for user plugins
      try:
          paths = [
              os.path.realpath(os.path.expandvars(os.path.expanduser(path)))
              for path in os.environ[ENVIRONMENT_NAME].split(os.pathsep)]
      except KeyError:
          log.debug(f'No custom plugin locations detected in {ENVIRONMENT_NAME}.')
          paths = []
      for path in paths:
          for module in discover(path):
              if path not in sys.path:
                  sys.path.insert(0, path)
              import_(module, path)
  
      # Import by entry_points
      try:
          for found in entry_points()[ENTRY_POINT_NAME]:
              log.debug(f'Loading plugin "{found.name}" ({found.value}).')
              found.load()
      except KeyError:
          log.debug(f'No custom plugins detected for "{ENTRY_POINT_NAME}".')
  
  
  def import_(module: str, path: Optional[str] = None) -> None:
      """ Attempt to import requested module """
      try:
          importlib.import_module(module)
          log.debug(f"Successfully imported the '{module}' module.")
      except ImportError as error:
          raise SystemExit(
              f"Failed to import the '{module}' module" +
              (f" from '{path}'." if path else ".") + f"\n({error})")
  
  
  def import_member(module_name: str, member_name: str) -> Any:
      """ Import member from given module, handle errors nicely """
      # Make sure the module is imported. It probably is, but really,
      # make sure of it.
      try:
          import_(module_name)
      except SystemExit as exc:
          raise tmt.utils.GeneralError(f"Failed to import module '{module_name}'.", original=exc)
  
      # Now the module should be available in `sys.modules` like any
      # other, and we can go and grab the class we need from it.
      if module_name not in sys.modules:
          raise tmt.utils.GeneralError(f"Failed to import module '{module_name}'.")
      module = sys.modules[module_name]
  
      # Get the member and return it
      if not hasattr(module, member_name):
          raise tmt.utils.GeneralError(f"No such member '{member_name}' in module '{module_name}'.")
      return getattr(module, member_name)
  
  
  def discover(path: str) -> Generator[str, None, None]:
      """ Discover available plugins for given paths """
      for _, name, package in pkgutil.iter_modules([path]):
          if not package:
              yield name
  
  
  [... the rest is omitted ...]
files: tmt/plugins/__init__.py
files: # coding: utf-8
  
  """ Handle Plugins """
  
  import importlib
  import os
  import pkgutil
  import sys
  from typing import Generator, Optional
  
  if sys.version_info < (3, 9):
      from importlib_metadata import entry_points
  else:
      from importlib.metadata import entry_points
  
  import fmf
  
  import tmt
  from tmt.steps import STEPS
  
  log = fmf.utils.Logging('tmt').logger
  
  # Two possibilities to load additional plugins:
  # entry_points (setup_tools)
  ENTRY_POINT_NAME = 'tmt.plugin'
  # Directories with module in environment variable
  ENVIRONMENT_NAME = 'TMT_PLUGINS'
  
  
  def explore() -> None:
      """ Explore all available plugins """
  
      # Check all tmt steps for native plugins
      root = os.path.dirname(os.path.realpath(tmt.__file__))
      for step in STEPS:
          for module in discover(os.path.join(root, 'steps', step)):
              import_(f'tmt.steps.{step}.{module}')
      # Check for possible plugins in the 'plugins' directory
      for module in discover(os.path.join(root, 'plugins')):
          import_(f'tmt.plugins.{module}')
  
      # Check environment variable for user plugins
      try:
          paths = [
              os.path.realpath(os.path.expandvars(os.path.expanduser(path)))
              for path in os.environ[ENVIRONMENT_NAME].split(os.pathsep)]
      except KeyError:
          log.debug(f'No custom plugin locations detected in {ENVIRONMENT_NAME}.')
          paths = []
      for path in paths:
          for module in discover(path):
              if path not in sys.path:
                  sys.path.insert(0, path)
              import_(module, path)
  
      # Import by entry_points
      try:
          for found in entry_points()[ENTRY_POINT_NAME]:
              log.debug(f'Loading plugin "{found.name}" ({found.value}).')
              found.load()
      except KeyError:
          log.debug(f'No custom plugins detected for "{ENTRY_POINT_NAME}".')
  
  
  def import_(module: str, path: Optional[str] = None) -> None:
      """ Attempt to import requested module """
      try:
          importlib.import_module(module)
          log.debug(f"Successfully imported the '{module}' module.")
      except ImportError as error:
          raise SystemExit(
              f"Failed to import the '{module}' module" +
              (f" from '{path}'." if path else ".") + f"\n({error})")
  
  
  def discover(path: str) -> Generator[str, None, None]:
      """ Discover available plugins for given paths """
      for _, name, package in pkgutil.iter_modules([path]):
          if not package:
              yield name
  
  
  [... the rest is omitted ...]
files: tmt/plugins/__init__.py
files: 
  """ Test Metadata Utilities """
  
  import contextlib
  import copy
  import dataclasses
  import datetime
  import functools
  import glob
  import io
  import os
  import pprint
  import re
  import shlex
  import shutil
  import subprocess
  import sys
  import time
  import unicodedata
  from collections import OrderedDict
  from functools import lru_cache
  from pathlib import Path
  from threading import Thread
  from typing import (IO, TYPE_CHECKING, Any, Callable, Dict, Generator,
                      Iterable, List, NamedTuple, Optional, Pattern, Tuple, Type,
                      TypeVar, Union, cast, overload)
  
  import click
  import fmf
  import jsonschema
  import pkg_resources
  import requests
  import requests.adapters
  import requests.packages.urllib3.util.retry
  import urllib3.exceptions
  from click import echo, style, wrap_text
  from ruamel.yaml import YAML, scalarstring
  from ruamel.yaml.comments import CommentedMap
  
  if sys.version_info >= (3, 8):
      from typing import Literal, Protocol
  else:
      from typing_extensions import Literal, Protocol
  
  
  if TYPE_CHECKING:
      import tmt.base
      import tmt.steps
  
  log = fmf.utils.Logging('tmt').logger
  
  # Default workdir root and max
  WORKDIR_ROOT = '/var/tmp/tmt'
  WORKDIR_MAX = 1000
  
  # Log in workdir
  LOG_FILENAME = 'log.txt'
  
  # Maximum number of lines of stdout/stderr to show upon errors
  OUTPUT_LINES = 100
  # Default output width
  OUTPUT_WIDTH = 79
  
  # Hierarchy indent
  INDENT = 4
  
  # Default name and order for step plugins
  DEFAULT_NAME = 'default'
  DEFAULT_PLUGIN_ORDER = 50
  DEFAULT_PLUGIN_ORDER_MULTIHOST = 10
  DEFAULT_PLUGIN_ORDER_REQUIRES = 70
  DEFAULT_PLUGIN_ORDER_RECOMMENDS = 75
  
  # Config directory
  CONFIG_PATH = '~/.config/tmt'
  
  # Special process return code
  PROCESS_TIMEOUT = 124
  
  # Default select.select(timeout) in seconds
  DEFAULT_SELECT_TIMEOUT = 5
  
  # Default shell and options to be set for all shell scripts
  DEFAULT_SHELL = "/bin/bash"
  SHELL_OPTIONS = 'set -eo pipefail'
  
  # Defaults for HTTP/HTTPS retries and timeouts (see `retry_session()`).
  DEFAULT_RETRY_SESSION_RETRIES: int = 3
  DEFAULT_RETRY_SESSION_BACKOFF_FACTOR: float = 0.1
  
  # Defaults for HTTP/HTTPS retries for getting environment file
  # Retry with exponential backoff, maximum duration ~511 seconds
  ENVFILE_RETRY_SESSION_RETRIES: int = 10
  ENVFILE_RETRY_SESSION_BACKOFF_FACTOR: float = 1
  
  # Default for wait()-related options
  DEFAULT_WAIT_TICK: float = 30.0
  DEFAULT_WAIT_TICK_INCREASE: float = 1.0
  
  # A stand-in variable for generic use.
  T = TypeVar('T')
  
  # A FMF context type, representing name/values context.
  FmfContextType = Dict[str, List[str]]
  
  # A "environment" type, representing name/value environment variables.
  EnvironmentType = Dict[str, str]
  
  # Workdir argument type, can be True, a string, a path or None
  WorkdirArgumentType = Union[Literal[True], str, None]
  
  # Workdir type, can be None or a string
  WorkdirType = Optional[str]
  
  # Option to skip to initialize work tree in plan
  PLAN_SKIP_WORKTREE_INIT = 'plan_skip_worktree_init'
  
  
  class BaseLoggerFnType(Protocol):
      def __call__(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          pass
  
  
  class LevelessLoggerFnType(Protocol):
      def __call__(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          pass
  
  
  class SemanticLoggerFnType(Protocol):
      def __call__(self, message: str, shift: int = 0) -> None:
          pass
  
  
  LoggerFnType = Union[
      BaseLoggerFnType,
      LevelessLoggerFnType,
      SemanticLoggerFnType]
  
  
  def indent(
          key: str,
          value: Optional[str] = None,
          color: Optional[str] = None,
          level: int = 0) -> str:
      """
      Indent a key/value message.
  
      If both ``key`` and ``value`` are specified, ``{key}: {value}``
      message is rendered. Otherwise, just ``key`` is used alone. If
      ``value`` contains multiple lines, each but the very first line is
      indented by one extra level.
  
      :param value: optional value to print at right side of ``key``.
      :param color: optional color to apply on ``key``.
      :param level: number of indentation levels. Each level is indented
                    by :py:data:`INDENT` spaces.
      """
  
      indent = ' ' * INDENT * level
      deeper = ' ' * INDENT * (level + 1)
  
      # Colorize
      if color is not None:
          key = style(key, fg=color)
  
      # Handle key only
      if value is None:
          message = key
  
      # Handle key + value
      else:
          # Multiline content indented deeper
          if isinstance(value, str):
              lines = value.splitlines()
              if len(lines) > 1:
                  value = ''.join([f"\n{deeper}{line}" for line in lines])
  
          message =
  
  [... the rest is omitted ...]
files: tmt/utils.py
files: 
  """ Test Metadata Utilities """
  
  import contextlib
  import copy
  import dataclasses
  import datetime
  import functools
  import glob
  import io
  import os
  import pprint
  import re
  import shlex
  import shutil
  import subprocess
  import sys
  import time
  import unicodedata
  from collections import OrderedDict
  from functools import lru_cache
  from pathlib import Path
  from threading import Thread
  from typing import (IO, TYPE_CHECKING, Any, Callable, Dict, Generator,
                      Iterable, List, NamedTuple, Optional, Pattern, Tuple, Type,
                      TypeVar, Union, cast, overload)
  
  import click
  import fmf
  import jsonschema
  import pkg_resources
  import requests
  import requests.adapters
  import requests.packages.urllib3.util.retry
  import urllib3.exceptions
  from click import echo, style, wrap_text
  from ruamel.yaml import YAML, scalarstring
  from ruamel.yaml.comments import CommentedMap
  
  if sys.version_info >= (3, 8):
      from typing import Literal, Protocol
  else:
      from typing_extensions import Literal, Protocol
  
  
  if TYPE_CHECKING:
      import tmt.base
      import tmt.steps
  
  log = fmf.utils.Logging('tmt').logger
  
  # Default workdir root and max
  WORKDIR_ROOT = '/var/tmp/tmt'
  WORKDIR_MAX = 1000
  
  # Log in workdir
  LOG_FILENAME = 'log.txt'
  
  # Maximum number of lines of stdout/stderr to show upon errors
  OUTPUT_LINES = 100
  # Default output width
  OUTPUT_WIDTH = 79
  
  # Hierarchy indent
  INDENT = 4
  
  # Default name and order for step plugins
  DEFAULT_NAME = 'default'
  DEFAULT_PLUGIN_ORDER = 50
  DEFAULT_PLUGIN_ORDER_MULTIHOST = 10
  DEFAULT_PLUGIN_ORDER_REQUIRES = 70
  DEFAULT_PLUGIN_ORDER_RECOMMENDS = 75
  
  # Config directory
  CONFIG_PATH = '~/.config/tmt'
  
  # Special process return code
  PROCESS_TIMEOUT = 124
  
  # Default select.select(timeout) in seconds
  DEFAULT_SELECT_TIMEOUT = 5
  
  # Default shell and options to be set for all shell scripts
  DEFAULT_SHELL = "/bin/bash"
  SHELL_OPTIONS = 'set -eo pipefail'
  
  # Defaults for HTTP/HTTPS retries and timeouts (see `retry_session()`).
  DEFAULT_RETRY_SESSION_RETRIES: int = 3
  DEFAULT_RETRY_SESSION_BACKOFF_FACTOR: float = 0.1
  
  # Defaults for HTTP/HTTPS retries for getting environment file
  # Retry with exponential backoff, maximum duration ~511 seconds
  ENVFILE_RETRY_SESSION_RETRIES: int = 10
  ENVFILE_RETRY_SESSION_BACKOFF_FACTOR: float = 1
  
  # Default for wait()-related options
  DEFAULT_WAIT_TICK: float = 30.0
  DEFAULT_WAIT_TICK_INCREASE: float = 1.0
  
  # A stand-in variable for generic use.
  T = TypeVar('T')
  
  # A FMF context type, representing name/values context.
  FmfContextType = Dict[str, List[str]]
  
  # A "environment" type, representing name/value environment variables.
  EnvironmentType = Dict[str, str]
  
  # Workdir argument type, can be True, a string, a path or None
  WorkdirArgumentType = Union[Literal[True], str, None]
  
  # Workdir type, can be None or a string
  WorkdirType = Optional[str]
  
  # Option to skip to initialize work tree in plan
  PLAN_SKIP_WORKTREE_INIT = 'plan_skip_worktree_init'
  
  
  class BaseLoggerFnType(Protocol):
      def __call__(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          pass
  
  
  class LevelessLoggerFnType(Protocol):
      def __call__(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          pass
  
  
  class SemanticLoggerFnType(Protocol):
      def __call__(self, message: str, shift: int = 0) -> None:
          pass
  
  
  LoggerFnType = Union[
      BaseLoggerFnType,
      LevelessLoggerFnType,
      SemanticLoggerFnType]
  
  
  def indent(
          key: str,
          value: Optional[str] = None,
          color: Optional[str] = None,
          level: int = 0) -> str:
      """
      Indent a key/value message.
  
      If both ``key`` and ``value`` are specified, ``{key}: {value}``
      message is rendered. Otherwise, just ``key`` is used alone. If
      ``value`` contains multiple lines, each but the very first line is
      indented by one extra level.
  
      :param value: optional value to print at right side of ``key``.
      :param color: optional color to apply on ``key``.
      :param level: number of indentation levels. Each level is indented
                    by :py:data:`INDENT` spaces.
      """
  
      indent = ' ' * INDENT * level
      deeper = ' ' * INDENT * (level + 1)
  
      # Colorize
      if color is not None:
          key = style(key, fg=color)
  
      # Handle key only
      if value is None:
          message = key
  
      # Handle key + value
      else:
          # Multiline content indented deeper
          if isinstance(value, str):
              lines = value.splitlines()
              if len(lines) > 1:
                  value = ''.join([f"\n{deeper}{line}" for line in lines])
  
          message =
  
  [... the rest is omitted ...]
files: tmt/utils.py