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('run-001'))
      # First call success
      common1 = Common()
      common1._workdir_init()
      assert common1.workdir == possible_workdir
      # Second call has no id to try
      with pytest.raises(GeneralError):
          Common()._workdir_init()
      # Removed run-001 should be used again
      common1._workdir_cleanup(common1.workdir)
      assert not os.path.exists(possible_workdir)
      common2 = Common()
      common2._workdir_init()
      assert common2.workdir == possible_workdir
  
  
  def test_workdir_root_race(tmpdir, monkeypatch):
      """ Avoid race in workdir creation """
      monkeypatch.setattr(tmt.utils, 'WORKDIR_ROOT', str(tmpdir))
      results = queue.Queue()
      threads = []
  
      def create_workdir():
          try:
              common = Common()
              common._workdir_init()
              results.put(common.workdir)
          except Exception as err:
              results.put(err)
  
      total = 30
      for _ in range(total):
          threads.append(threading.Thread(target=create_workdir))
      for t in threads:
          t.start()
      for t in threads:
          t.join()
  
      all_good = True
      unique_workdirs = set()
      for t in threads:
          value = results.get()
          if isinstance(value, str):
              unique_workdirs.add(value)
          else:
              # None or Exception: Record to the log and fail test
              print(value)
              all_good = False
      assert all_good, "No exception raised"
      assert len(unique_workdirs) == total, "Each workdir is unique"
  
  
  def test_duration_to_seconds():
      """ Check conversion from sleep time format to seconds """
      assert duration_to_seconds(5) == 5
      assert duration_to_seconds('5') == 5
      assert duration_to_seconds('5s') == 5
      assert duration_to_seconds('5m') == 300
      assert duration_to_seconds('5h') == 18000
      assert duration_to_seconds('5d') == 432000
      with pytest.raises(tmt.utils.SpecificationError):
          duration_to_seconds('bad')
  
  
  class test_structured_field(unittest.TestCase):
      """ Self Test """
  
      def setUp(self):
          self.header = "This is a header.\n"
          self.footer = "This is a footer.\n"
          self.start = (
              "[structured-field-start]\n"
              "This is StructuredField version 1. "
              "Please, edit with care.\n")
          self.end = "[structured-field-end]\n"
          self.zeroend = "[end]\n"
          self.one = "[one]\n1\n"
          self.two = "[two]\n2\n"
          self.three = "[three]\n3\n"
          self.sections = "\n".join([self.one, self.two, self.three])
  
      def test_everything(self):
          """ Everything """
          # Version 0
          text0 = "\n".join([
                  self.header,
                  self.sections, self.zeroend,
                  self.footer])
          inited0 = StructuredField(text0, version=0)
          loaded0 = StructuredField()
          loaded0.load(text0, version=0)
          self.assertEqual(inited0.save(), text0)
          self.assertEqual(loaded0.save(), text0)
          # Version 1
          text1 = "\n".join([
                  self.header,
                  self.start, self.sections, self.end,
                  self.footer])
          inited1 = StructuredField(text1)
          loaded1 = StructuredField()
          loaded1.load(text1)
          self.assertEqual(inited1.save(), text1)
          self.assertEqual(loaded1.save(), text1)
          # Common checks
          for field in [inited0, loaded0, inited1, loaded1]:
              self.assertEqual(field.header(), self.header)
              self.assertEqual(field.footer(), self.footer)
              self.assertEqual(field.sections(), ["one", "two", "three"])
              self.assertEqual(field.get("one"), "1\n")
              self.assertEqual(field.get("two"), "2\n")
              self.assertEqual(field.get("three"), "3\n")
  
      def test_no_header(self):
          """ No header """
          # Version 0
          text0 = "\n".join([self.sections, self.zeroend, self.footer])
          field0 = StructuredField(text0, version=0)
          self.assertEqual(field0.save(), text0)
          # Version 1
          text1 = "\n".join(
                  [self.start, self.sections, self.end, self.footer])
          field1 = StructuredField(text1)
          self.assertEqual(field1.save(), text1)
          # Common checks
          for field in [field0, field1]:
              self.assertEqual(field.header(), "")
              self.assertEqual(field.footer(), self.footer)
              self.assertEqual(field.get("one"), "1\n")
              self.assertEqual(field.get("two"), "2\n")
              self.assertEqual(field.get("three"), "3\n")
  
      def test_no_footer(self):
          """ No footer """
          # Version 0
          text0 = "\n".join([self.header, self.sections, self.zeroend])
          field0 = StructuredField(text0, version=0)
          self.assertEqual(field0.save(), text0)
          # Version 1
          text1 = "\n".join(
                  [self.header, self.start, self.sections, self.end])
          field1 = StructuredField(text1)
          self.assertEqual(field1.save(), text1)
          # Common checks
          for field in [field0, field1]:
              self.assertEqual(field.header(), self.header)
              self.assertEqual(field.footer(), "")
              self.assertEqual(field.get("one"), "1\n")
              self.assertEqual(field.get("two"), "2\n")
              self.assertEqual(field.get("three"), "3\n")
  
      def test_just_sections(self):
          """ Just sections """
          # Version 0
          text0 = "\n".join([self.sections, self.zeroend])
          field0 = StructuredField(text0, version=0)
          self.assertEqual(field0.save(), text0)
          # Version 1
          text1 = "\n".join([self.start, self.sections, self.end])
          field1 = StructuredField(text1)
          self.assertEqual(field1.save(), text1)
          # Common checks
          for field in [field0, field1]:
              self.assertEqual(field.header(), "")
              self.assertEqual(field.footer(), "")
              self.assertEqual(field.get("one"), "1\n")
              self.assertEqual(field.get("two"), "2\n")
              self.assertEqual(field.get("three"), "3\n")
  
      def test_plain_text(self):
          """ Plain text """
          text = "Some plain text.\n"
          field0 = StructuredField(text, version=0)
          field1 = StructuredField(text)
          for field in [field0, field1]:
              self.assertEqual(field.header(), text)
              self.assertEqual(field.footer(), "")
              self.assertEqual(field.save(), text)
              self.assertEqual(list(field), [])
              self.assertEqual(bool(field), False)
  
      def test_missing_end_tag(self):
          """ Missing end tag """
          text = "\n".join([self.header, self.sections, self.footer])
          self.assertRaises(StructuredFieldError, StructuredField, text, 0)
  
      def test_broken_field(self):
          """ Broken field"""
          text = "[structured-field-start]"
          self.assertRaises(StructuredFieldError, StructuredField, text)
  
      def test_set_content(self):
          """ Set section content """
          field0 = StructuredField(version=0)
          field1 = StructuredField()
          for field in [field0, field1]:
              field.set("one", "1")
              self.assertEqual(field.get("one"), "1\n")
              field.set("two", "2")
              self.assertEqual(field.get("two"), "2\n")
              field.set("three", "3")
              self.assertEqual(field.get("three"), "3\n")
          self.assertEqual(field0.save(), "\n".join(
              [self.sections, self.zeroend]))
          self.assertEqual(field1.save(), "\n".join(
              [self.start, self.sections, self.end]))
  
      def test_remove_section(self):
          """ Remove section """
          field0 = StructuredField(
              "\n".join([self.sections, self.zeroend]), version=0)
          field1 = StructuredField(
              "\n".join([self.start, self.sections, self.end]))
          for field in [field0, field1]:
              field.remove("one")
              field.remove("two")
          self.assertEqual(
              field0.save(), "\n".join([self.three, self.zeroend]))
          self.assertEqual(
              field1.save(), "\n".join([self.start, self.three, self.end]))
  
      def test_section_tag_escaping(self):
          """ Section tag escaping """
          field = StructuredField()
          field.set("section", "\n[content]\n")
          reloaded = StructuredField(field.save())
          self.assertTrue("section" in reloaded)
          self.assertTrue("content" not in reloaded)
          self.assertEqual(reloaded.get("section"), "\n[content]\n")
  
      def test_nesting(self):
          """ Nesting """
          # Prepare structure parent -> child -> grandchild
          grandchild = StructuredField()
          grandchild.set('name', "Grand Child\n")
          child = StructuredField()
          child.set('name', "Child Name\n")
          child.set("child", grandchild.save())
          parent = StructuredField()
          parent.set("name", "Parent Name\n")
          parent.set("child", child.save())
          # Reload back and check the names
          parent = StructuredField(parent.save())
          child = StructuredField(parent.get("child"))
          grandchild = StructuredField(child.get("child"))
          self.assertEqual(parent.get("name"), "Parent Name\n")
          self.assertEqual(child.get("name"), "Child Name\n")
          self.assertEqual(grandchild.get("name"), "Grand Child\n")
  
      def test_section_tags_in_header(self):
          """ Section tags in header """
          field = StructuredField("\n".join(
              ["[something]", self.start, self.one, self.end]))
          self.assertTrue("something" not in field)
          self.assertTrue("one" in field)
          self.assertEqual(field.get("one"), "1\n")
  
      def test_empty_section(self):
          """ Empty section """
          field = StructuredField()
          field.set("section", "")
          reloaded = StructuredField(field.save())
          self.assertEqual(reloaded.get("section"), "")
  
      def test_section_item_get(self):
          """ Get section item """
          text = "\n".join([self.start, "[section]\nx = 3\n", self.end])
          field = StructuredField(text)
          self.assertEqual(field.get("section", "x"), "3")
  
      def test_section_item_set(self):
          """ Set section item """
          text = "\n".join([self.start, "[section]\nx = 3\n", self.end])
          field = StructuredField()
          field.set("section", "3", "x")
          self.assertEqual(field.save(), text)
  
      def test_section_item_remove(self):
          """ Remove section item """
          text = "\n".join(
              [self.start, "[section]\nx = 3\ny = 7\n", self.end])
          field = StructuredField(text)
          field.remove("section", "x")
          self.assertEqual(field.save(), "\n".join(
              [self.start, "[section]\ny = 7\n", self.end]))
  
      def test_unicode_header(self):
          """ Unicode text in header """
          text = u"Už abychom měli unicode jako defaultní kódování!"
          field = StructuredField(text)
          field.set("section", "content")
          self.assertTrue(text in field.save())
  
      def test_unicode_section_content(self):
          """ Unicode in section content """
          chars = u"ěščřžýáíéů"
          text = "\n".join([self.start, "[section]", chars, self.end])
          field = StructuredField(text)
          self.assertEqual(field.get("section").strip(), chars)
  
      def test_unicode_section_name(self):
          """ Unicode in section name """
          chars = u"ěščřžýáíéů"
          text = "\n".join([self.start, u"[{0}]\nx".format(chars), self.end])
          field = StructuredField(text)
          self.assertEqual(field.get(chars).strip(), "x")
  
      def test_header_footer_modify(self):
          """ Modify header & footer """
          original = StructuredField()
          original.set("field", "field-content")
          original.header("header-content\n")
          original.footer("footer-content\n")
          copy = StructuredField(original.save())
          self.assertEqual(copy.header(), "header-content\n")
          self.assertEqual(copy.footer(), "footer-content\n")
  
      def test_trailing_whitespace(self):
          """ Trailing whitespace """
          original = StructuredField()
          original.set("name", "value")
          # Test with both space and tab appended after the section tag
          for char in [" ", "\t"]:
              spaced = re.sub(r"\]\n", "]{0}\n".format(char), original.save())
              copy = StructuredField(spaced)
              self.assertEqual(original.get("name"), copy.get("name"))
  
      def test_carriage_returns(self):
          """ Carriage returns """
          text1 = "\n".join([self.start, self.sections, self.end])
          text2 = re.sub(r"\n", "\r\n", text1)
          field1 = StructuredField(text1)
          field2 = StructuredField(text2)
          self.assertEqual(field1.save(), field2.save())
  
      def test_multiple_values(self):
          """ Multiple values """
          # Reading multiple values
          section = "[section]\nkey=val1 # comment\nkey = val2\n key = val3 "
          text = "\n".join([self.start, section, self.end])
          field = StructuredField(text, multi=True)
          self.assertEqual(
              field.get("section", "key"), ["val1", "val2", "val3"])
          # Writing multiple values
          values = ['1', '2', '3']
          field = StructuredField(multi=True)
          field.set("section", values, "key")
          self.assertEqual(field.get("section", "key"), values)
          self.assertTrue("key = 1\nkey = 2\nkey = 3" in field.save())
          # Remove multiple values
          field.remove("section", "key")
          self.assertTrue("key = 1\nkey = 2\nkey = 3" not in field.save())
          self.assertRaises(
              StructuredFieldError, field.get, "section", "key")
  
  
  class Run(unittest.TestCase):
  
      def test_interactive_not_joined(self):
          stdout, stderr = Common()._run(
              "echo abc; echo def >2", shell=True, interactive=True, cwd=".", env={}, log=None)
          self.assertEqual(stdout, None)
          self.assertEqual(stderr, None)
  
      def test_interactive_joined(self):
          stdout, _ = Common()._run(
              "echo abc; echo def >2",
              shell=True,
              interactive=True,
              cwd=".",
              env={},
              join=True,
              log=None)
          self.assertEqual(stdout, None)
  
      def test_not_joined_stdout(self):
          stdout, _ = Common()._run("ls /", shell=True, cwd=".", env={}, log=None)
          self.assertIn("sbin", stdout)
  
      def test_not_joined_stderr(self):
          _, stderr = Common()._run("ls non_existing || true", shell=True, cwd=".", env={}, log=None)
          self.assertIn("ls: cannot access", stderr)
  
      def test_joined(self):
          stdout, _ = Common()._run(
              "ls non_existing / || true",
              shell=True,
              cwd=".",
              env={},
              log=None,
              join=True)
          self.assertIn("ls: cannot access", stdout)
          self.assertIn("sbin", stdout)
  
      def test_big(self):
          stdout, _ = Common()._run(
              """for NUM in {1..100}; do LINE="$LINE n"; done; for NUM in {1..1000}; do echo $LINE; done""",  # noqa: E501
              shell=True,
              cwd=".",
              env={},
              log=None,
              join=True)
          self.assertIn("n n", stdout)
          self.assertEqual(len(stdout), 200000)
  
  
  def test_get_distgit_handler():
      for wrong_remotes in [[], ["blah"]]:
          with pytest.raises(tmt.utils.GeneralError):
              tmt.utils.get_distgit_handler([])
      # Fedora detection
      returned_object = tmt.utils.get_distgit_handler("""
          remote.origin.url ssh://lzachar@pkgs.fedoraproject.org/rpms/tmt
          remote.lzachar.url ssh://lzachar@pkgs.fedoraproject.org/forks/lzachar/rpms/tmt.git
          """.split('\n'))
      assert isinstance(returned_object, tmt.utils.FedoraDistGit)
      # CentOS detection
      returned_object = tmt.utils.get_distgit_handler("""
          remote.origin.url git+ssh://git@gitlab.com/redhat/centos-stream/rpms/ruby.git
          """.split('\n'))
      assert isinstance(returned_object, tmt.utils.CentOSDistGit)
  
  
  def test_FedoraDistGit(tmpdir):
      # Fake values, production hash is too long
      tmpdir.join('sources').write(
          'SHA512 (fn-1.tar.gz) = 09af\n')
      tmpdir.join('tmt.spec').write('')
      fedora_sources_obj = tmt.utils.FedoraDistGit()
      assert [("https://src.fedoraproject.org/repo/pkgs/rpms
  
  [... 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 = Common()
      common1._workdir_init()
      assert common1.workdir == possible_workdir
      # Second call has no id to try
      with pytest.raises(GeneralError):
          Common()._workdir_init()
      # Removed run-001 should be used again
      common1._workdir_cleanup(common1.workdir)
      assert not os.path.exists(possible_workdir)
      common2 = Common()
      common2._workdir_init()
      assert common2.workdir == possible_workdir
  
  
  def test_workdir_root_race(tmpdir, monkeypatch):
      """ Avoid race in workdir creation """
      monkeypatch.setattr(tmt.utils, 'WORKDIR_ROOT', str(tmpdir))
      results = queue.Queue()
      threads = []
  
      def create_workdir():
          try:
              common = Common()
              common._workdir_init()
              results.put(common.workdir)
          except Exception as err:
              results.put(err)
  
      total = 30
      for _ in range(total):
          threads.append(threading.Thread(target=create_workdir))
      for t in threads:
          t.start()
      for t in threads:
          t.join()
  
      all_good = True
      unique_workdirs = set()
      for t in threads:
          value = results.get()
          if isinstance(value, str):
              unique_workdirs.add(value)
          else:
              # None or Exception: Record to the log and fail test
              print(value)
              all_good = False
      assert all_good, "No exception raised"
      assert len(unique_workdirs) == total, "Each workdir is unique"
  
  
  def test_duration_to_seconds():
      """ Check conversion from sleep time format to seconds """
      assert duration_to_seconds(5) == 5
      assert duration_to_seconds('5') == 5
      assert duration_to_seconds('5s') == 5
      assert duration_to_seconds('5m') == 300
      assert duration_to_seconds('5h') == 18000
      assert duration_to_seconds('5d') == 432000
      with pytest.raises(tmt.utils.SpecificationError):
          duration_to_seconds('bad')
  
  
  class test_structured_field(unittest.TestCase):
      """ Self Test """
  
      def setUp(self):
          self.header = "This is a header.\n"
          self.footer = "This is a footer.\n"
          self.start = (
              "[structured-field-start]\n"
              "This is StructuredField version 1. "
              "Please, edit with care.\n")
          self.end = "[structured-field-end]\n"
          self.zeroend = "[end]\n"
          self.one = "[one]\n1\n"
          self.two = "[two]\n2\n"
          self.three = "[three]\n3\n"
          self.sections = "\n".join([self.one, self.two, self.three])
  
      def test_everything(self):
          """ Everything """
          # Version 0
          text0 = "\n".join([
                  self.header,
                  self.sections, self.zeroend,
                  self.footer])
          inited0 = StructuredField(text0, version=0)
          loaded0 = StructuredField()
          loaded0.load(text0, version=0)
          self.assertEqual(inited0.save(), text0)
          self.assertEqual(loaded0.save(), text0)
          # Version 1
          text1 = "\n".join([
                  self.header,
                  self.start, self.sections, self.end,
                  self.footer])
          inited1 = StructuredField(text1)
          loaded1 = StructuredField()
          loaded1.load(text1)
          self.assertEqual(inited1.save(), text1)
          self.assertEqual(loaded1.save(), text1)
          # Common checks
          for field in [inited0, loaded0, inited1, loaded1]:
              self.assertEqual(field.header(), self.header)
              self.assertEqual(field.footer(), self.footer)
              self.assertEqual(field.sections(), ["one", "two", "three"])
              self.assertEqual(field.get("one"), "1\n")
              self.assertEqual(field.get("two"), "2\n")
              self.assertEqual(field.get("three"), "3\n")
  
      def test_no_header(self):
          """ No header """
          # Version 0
          text0 = "\n".join([self.sections, self.zeroend, self.footer])
          field0 = StructuredField(text0, version=0)
          self.assertEqual(field0.save(), text0)
          # Version 1
          text1 = "\n".join(
                  [self.start, self.sections, self.end, self.footer])
          field1 = StructuredField(text1)
          self.assertEqual(field1.save(), text1)
          # Common checks
          for field in [field0, field1]:
              self.assertEqual(field.header(), "")
              self.assertEqual(field.footer(), self.footer)
              self.assertEqual(field.get("one"), "1\n")
              self.assertEqual(field.get("two"), "2\n")
              self.assertEqual(field.get("three"), "3\n")
  
      def test_no_footer(self):
          """ No footer """
          # Version 0
          text0 = "\n".join([self.header, self.sections, self.zeroend])
          field0 = StructuredField(text0, version=0)
          self.assertEqual(field0.save(), text0)
          # Version 1
          text1 = "\n".join(
                  [self.header, self.start, self.sections, self.end])
          field1 = StructuredField(text1)
          self.assertEqual(field1.save(), text1)
          # Common checks
          for field in [field0, field1]:
              self.assertEqual(field.header(), self.header)
              self.assertEqual(field.footer(), "")
              self.assertEqual(field.get("one"), "1\n")
              self.assertEqual(field.get("two"), "2\n")
              self.assertEqual(field.get("three"), "3\n")
  
      def test_just_sections(self):
          """ Just sections """
          # Version 0
          text0 = "\n".join([self.sections, self.zeroend])
          field0 = StructuredField(text0, version=0)
          self.assertEqual(field0.save(), text0)
          # Version 1
          text1 = "\n".join([self.start, self.sections, self.end])
          field1 = StructuredField(text1)
          self.assertEqual(field1.save(), text1)
          # Common checks
          for field in [field0, field1]:
              self.assertEqual(field.header(), "")
              self.assertEqual(field.footer(), "")
              self.assertEqual(field.get("one"), "1\n")
              self.assertEqual(field.get("two"), "2\n")
              self.assertEqual(field.get("three"), "3\n")
  
      def test_plain_text(self):
          """ Plain text """
          text = "Some plain text.\n"
          field0 = StructuredField(text, version=0)
          field1 = StructuredField(text)
          for field in [field0, field1]:
              self.assertEqual(field.header(), text)
              self.assertEqual(field.footer(), "")
              self.assertEqual(field.save(), text)
              self.assertEqual(list(field), [])
              self.assertEqual(bool(field), False)
  
      def test_missing_end_tag(self):
          """ Missing end tag """
          text = "\n".join([self.header, self.sections, self.footer])
          self.assertRaises(StructuredFieldError, StructuredField, text, 0)
  
      def test_broken_field(self):
          """ Broken field"""
          text = "[structured-field-start]"
          self.assertRaises(StructuredFieldError, StructuredField, text)
  
      def test_set_content(self):
          """ Set section content """
          field0 = StructuredField(version=0)
          field1 = StructuredField()
          for field in [field0, field1]:
              field.set("one", "1")
              self.assertEqual(field.get("one"), "1\n")
              field.set("two", "2")
              self.assertEqual(field.get("two"), "2\n")
              field.set("three", "3")
              self.assertEqual(field.get("three"), "3\n")
          self.assertEqual(field0.save(), "\n".join(
              [self.sections, self.zeroend]))
          self.assertEqual(field1.save(), "\n".join(
              [self.start, self.sections, self.end]))
  
      def test_remove_section(self):
          """ Remove section """
          field0 = StructuredField(
              "\n".join([self.sections, self.zeroend]), version=0)
          field1 = StructuredField(
              "\n".join([self.start, self.sections, self.end]))
          for field in [field0, field1]:
              field.remove("one")
              field.remove("two")
          self.assertEqual(
              field0.save(), "\n".join([self.three, self.zeroend]))
          self.assertEqual(
              field1.save(), "\n".join([self.start, self.three, self.end]))
  
      def test_section_tag_escaping(self):
          """ Section tag escaping """
          field = StructuredField()
          field.set("section", "\n[content]\n")
          reloaded = StructuredField(field.save())
          self.assertTrue("section" in reloaded)
          self.assertTrue("content" not in reloaded)
          self.assertEqual(reloaded.get("section"), "\n[content]\n")
  
      def test_nesting(self):
          """ Nesting """
          # Prepare structure parent -> child -> grandchild
          grandchild = StructuredField()
          grandchild.set('name', "Grand Child\n")
          child = StructuredField()
          child.set('name', "Child Name\n")
          child.set("child", grandchild.save())
          parent = StructuredField()
          parent.set("name", "Parent Name\n")
          parent.set("child", child.save())
          # Reload back and check the names
          parent = StructuredField(parent.save())
          child = StructuredField(parent.get("child"))
          grandchild = StructuredField(child.get("child"))
          self.assertEqual(parent.get("name"), "Parent Name\n")
          self.assertEqual(child.get("name"), "Child Name\n")
          self.assertEqual(grandchild.get("name"), "Grand Child\n")
  
      def test_section_tags_in_header(self):
          """ Section tags in header """
          field = StructuredField("\n".join(
              ["[something]", self.start, self.one, self.end]))
          self.assertTrue("something" not in field)
          self.assertTrue("one" in field)
          self.assertEqual(field.get("one"), "1\n")
  
      def test_empty_section(self):
          """ Empty section """
          field = StructuredField()
          field.set("section", "")
          reloaded = StructuredField(field.save())
          self.assertEqual(reloaded.get("section"), "")
  
      def test_section_item_get(self):
          """ Get section item """
          text = "\n".join([self.start, "[section]\nx = 3\n", self.end])
          field = StructuredField(text)
          self.assertEqual(field.get("section", "x"), "3")
  
      def test_section_item_set(self):
          """ Set section item """
          text = "\n".join([self.start, "[section]\nx = 3\n", self.end])
          field = StructuredField()
          field.set("section", "3", "x")
          self.assertEqual(field.save(), text)
  
      def test_section_item_remove(self):
          """ Remove section item """
          text = "\n".join(
              [self.start, "[section]\nx = 3\ny = 7\n", self.end])
          field = StructuredField(text)
          field.remove("section", "x")
          self.assertEqual(field.save(), "\n".join(
              [self.start, "[section]\ny = 7\n", self.end]))
  
      def test_unicode_header(self):
          """ Unicode text in header """
          text = u"Už abychom měli unicode jako defaultní kódování!"
          field = StructuredField(text)
          field.set("section", "content")
          self.assertTrue(text in field.save())
  
      def test_unicode_section_content(self):
          """ Unicode in section content """
          chars = u"ěščřžýáíéů"
          text = "\n".join([self.start, "[section]", chars, self.end])
          field = StructuredField(text)
          self.assertEqual(field.get("section").strip(), chars)
  
      def test_unicode_section_name(self):
          """ Unicode in section name """
          chars = u"ěščřžýáíéů"
          text = "\n".join([self.start, u"[{0}]\nx".format(chars), self.end])
          field = StructuredField(text)
          self.assertEqual(field.get(chars).strip(), "x")
  
      def test_header_footer_modify(self):
          """ Modify header & footer """
          original = StructuredField()
          original.set("field", "field-content")
          original.header("header-content\n")
          original.footer("footer-content\n")
          copy = StructuredField(original.save())
          self.assertEqual(copy.header(), "header-content\n")
          self.assertEqual(copy.footer(), "footer-content\n")
  
      def test_trailing_whitespace(self):
          """ Trailing whitespace """
          original = StructuredField()
          original.set("name", "value")
          # Test with both space and tab appended after the section tag
          for char in [" ", "\t"]:
              spaced = re.sub(r"\]\n", "]{0}\n".format(char), original.save())
              copy = StructuredField(spaced)
              self.assertEqual(original.get("name"), copy.get("name"))
  
      def test_carriage_returns(self):
          """ Carriage returns """
          text1 = "\n".join([self.start, self.sections, self.end])
          text2 = re.sub(r"\n", "\r\n", text1)
          field1 = StructuredField(text1)
          field2 = StructuredField(text2)
          self.assertEqual(field1.save(), field2.save())
  
      def test_multiple_values(self):
          """ Multiple values """
          # Reading multiple values
          section = "[section]\nkey=val1 # comment\nkey = val2\n key = val3 "
          text = "\n".join([self.start, section, self.end])
          field = StructuredField(text, multi=True)
          self.assertEqual(
              field.get("section", "key"), ["val1", "val2", "val3"])
          # Writing multiple values
          values = ['1', '2', '3']
          field = StructuredField(multi=True)
          field.set("section", values, "key")
          self.assertEqual(field.get("section", "key"), values)
          self.assertTrue("key = 1\nkey = 2\nkey = 3" in field.save())
          # Remove multiple values
          field.remove("section", "key")
          self.assertTrue("key = 1\nkey = 2\nkey = 3" not in field.save())
          self.assertRaises(
              StructuredFieldError, field.get, "section", "key")
  
  
  class Run(unittest.TestCase):
  
      def test_interactive_not_joined(self):
          stdout, stderr = Common()._run(
              "echo abc; echo def >2", shell=True, interactive=True, cwd=".", env={}, log=None)
          self.assertEqual(stdout, None)
          self.assertEqual(stderr, None)
  
      def test_interactive_joined(self):
          stdout, _ = Common()._run(
              "echo abc; echo def >2",
              shell=True,
              interactive=True,
              cwd=".",
              env={},
              join=True,
              log=None)
          self.assertEqual(stdout, None)
  
      def test_not_joined_stdout(self):
          stdout, _ = Common()._run("ls /", shell=True, cwd=".", env={}, log=None)
          self.assertIn("sbin", stdout)
  
      def test_not_joined_stderr(self):
          _, stderr = Common()._run("ls non_existing || true", shell=True, cwd=".", env={}, log=None)
          self.assertIn("ls: cannot access", stderr)
  
      def test_joined(self):
          stdout, _ = Common()._run(
              "ls non_existing / || true",
              shell=True,
              cwd=".",
              env={},
              log=None,
              join=True)
          self.assertIn("ls: cannot access", stdout)
          self.assertIn("sbin", stdout)
  
      def test_big(self):
          stdout, _ = Common()._run(
              """for NUM in {1..100}; do LINE="$LINE n"; done; for NUM in {1..1000}; do echo $LINE; done""",  # noqa: E501
              shell=True,
              cwd=".",
              env={},
              log=None,
              join=True)
          self.assertIn("n n", stdout)
          self.assertEqual(len(stdout), 200000)
  
  
  def test_get_distgit_handler():
      for wrong_remotes in [[], ["blah"]]:
          with pytest.raises(tmt.utils.GeneralError):
              tmt.utils.get_distgit_handler([])
      # Fedora detection
      returned_object = tmt.utils.get_distgit_handler("""
          remote.origin.url ssh://lzachar@pkgs.fedoraproject.org/rpms/tmt
          remote.lzachar.url ssh://lzachar@pkgs.fedoraproject.org/forks/lzachar/rpms/tmt.git
          """.split('\n'))
      assert isinstance(returned_object, tmt.utils.FedoraDistGit)
      # CentOS detection
      returned_object = tmt.utils.get_distgit_handler("""
          remote.origin.url git+ssh://git@gitlab.com/redhat/centos-stream/rpms/ruby.git
          """.split('\n'))
      assert isinstance(returned_object, tmt.utils.CentOSDistGit)
  
  
  def test_FedoraDistGit(tmpdir):
      # Fake values, production hash is too long
      tmpdir.join('sources').write(
          'SHA512 (fn-1.tar.gz) = 09af\n')
      tmpdir.join('tmt.spec').write('')
      fedora_sources_obj = tmt.utils.FedoraDistGit()
      assert [("https://src.fedoraproject.org/repo/pkgs/rpms/tmt/fn-1.tar.gz/sha512/09af/f
  
  [... 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 = f'{key}: {value}'
  
      return indent + message
  
  
  class Config:
      """ User configuration """
  
      def __init__(self) -> None:
          """ Initialize config directory path """
          self.path = os.path.expanduser(CONFIG_PATH)
          if not os.path.exists(self.path):
              try:
                  os.makedirs(self.path)
              except OSError as error:
                  raise GeneralError(
                      f"Failed to create config '{self.path}'.\n{error}")
  
      def last_run(self, run_id: Optional[str] = None) -> Optional[str]:
          """ Get and set last run id """
          symlink = os.path.join(self.path, 'last-run')
          if run_id:
              try:
                  os.remove(symlink)
              except OSError:
                  pass
              try:
                  os.symlink(run_id, symlink)
              except FileExistsError:
                  # Race when tmt runs in parallel
                  log.warning(f"Race condition, unable to save last run '{run_id}'.")
              except OSError as error:
                  raise GeneralError(
                      f"Unable to save last run '{self.path}'.\n{error}")
              return run_id
          if os.path.islink(symlink):
              return os.path.realpath(symlink)
          return None
  
  
  class StreamLogger(Thread):
      """
      Reading pipes of running process in threads.
  
      Code based on:
      https://github.com/packit/packit/blob/main/packit/utils/logging.py#L10
      """
  
      def __init__(self,
                   stream: Optional[IO[bytes]],
                   log_header: str,
                   logger: BaseLoggerFnType) -> None:
          super().__init__(daemon=True)
          self.stream = stream
          self.output: List[str] = []
          self.log_header = log_header
          self.logger = logger
  
      def run(self) -> None:
          if self.stream is None:
              return
  
          for _line in self.stream:
              line = _line.decode('utf-8', errors='replace')
              if line != '':
                  self.logger(
                      self.log_header,
                      line.rstrip('\n'),
                      'yellow',
                      level=3)
              self.output.append(line)
  
      def get_output(self) -> str:
          return "".join(self.output)
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Common
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  CommonDerivedType = TypeVar('CommonDerivedType', bound='Common')
  
  
  class CommandOutput(NamedTuple):
      stdout: Optional[str]
      stderr: Optional[str]
  
  
  class Common:
      """
      Common shared stuff
  
      Takes care of command line context, options and workdir handling.
      Provides logging functions info(), verbose() and debug().
      Implements read() and write() for comfortable file access.
      Provides the run() method for easy command execution.
      """
  
      # Command line context, options and workdir
      _context: Optional[click.Context] = None
      # When set to true, _opt will be ignored (default will be returned)
      ignore_class_options: bool = False
      _options: Dict[str, Any] = dict()
      _workdir: WorkdirType = None
  
      # TODO: must be declared outside of __init__(), because it must exist before
      # __init__() gets called to allow logging helpers work correctly when used
      # from mixins. But that's not very clean, is it? :( Maybe decoupling logging
      # from Common class would help, such a class would be able to initialize
      # itself without involving the rest of Common code. On the other hand,
      # Common owns workdir, for example, whose value affects logging too, so no
      # clear solution so far.
      #
      # Note: cannot use CommonDerivedType - it's a TypeVar filled in by the type
      # given to __init__() and therefore the type it's representing *now* is
      # unknown. but we know `parent` will be derived from `Common` class, so it's
      # mostly fine.
      parent: Optional['Common'] = None
  
      def __init__(
              self,
              parent: Optional[CommonDerivedType] = None,
              name: Optional[str] = None,
              workdir: WorkdirArgumentType = None,
              context: Optional[click.Context] = None,
              relative_indent: int = 1,
              **kwargs: Any) -> None:
          """
          Initialize name and relation with the parent object
  
          Prepare the workdir for provided id / directory path
          or generate a new workdir name if workdir=True given.
          Store command line context and options for future use
          if context is provided.
          """
          # Use lowercase class name as the default name
          self.name = name or self.__class__.__name__.lower()
          self.parent = parent
  
          # Relative log indent level shift against the parent
          self._relative_indent = relative_indent
  
          # Store command line context
          if context:
              self._save_context_to_instance(context)
  
          # Initialize the workdir if requested
          self._workdir_load(workdir)
  
      def __str__(self) -> str:
          """ Name is the default string representation """
          return self.name
  
      @classmethod
      def _save_context(cls, context: click.Context) -> None:
          """ Save provided command line context and options for future use """
          cls._context = context
          cls._options = context.params
  
      def _save_context_to_instance(self, context: click.Context) -> None:
          """ Save provided command line context and options to the instance """
          self._context = context
          self._options = context.params
  
      @overload
      @classmethod
      def _opt(cls, option: str) -> Any:
          pass
  
      @overload
      @classmethod
      def _opt(cls, option: str, default: T) -> T:
          pass
  
      @classmethod
      def _opt(cls, option: str, default: Any = None) -> Any:
          """ Get an option from the command line context (class version) """
          if cls.ignore_class_options:
              return default
          return cls._options.get(option, default)
  
      def _fmf_context(self) -> FmfContextType:
          """ Return the current fmf context """
          if self._context is None:
              return dict()
  
          try:
              return cast(FmfContextType, self._context.obj.fmf_context)
          except AttributeError:
              return dict()
  
      def opt(self, option: str, default: Optional[Any] = None) -> Any:
          """
          Get an option from the command line options
  
          Checks also parent options. For flags (boolean values) parent's
          True wins over child's False (e.g. run --quiet enables quiet
          mode for all included plans and steps).
  
          For options that can be used multiple times, the child overrides
          the parent if it was defined (e.g. run -av provision -vvv runs
          all steps except for provision in mildly verbose mode, provision
          is run with the most verbosity).
  
          Environment variables override command line options.
          """
          # Translate dashes to underscores to match click's conversion
          option = option.replace('-', '_')
          # Check the environment first
          if option == 'debug':
              try:
                  debug = os.environ['TMT_DEBUG']
                  return int(debug)
              except ValueError:
                  raise GeneralError(
                      f"Invalid debug level '{debug}', use an integer.")
              except KeyError:
                  pass
  
          # Get local option
          local = self._options.get(option, default)
          # Check parent option
          parent = None
          if self.parent:
              parent = self.parent.opt(option)
          # Special handling for special flags (parent's yes always wins)
          if option in ['quiet', 'force', 'dry']:
              return parent if parent else local
          # Special handling for counting options (child overrides the
          # parent if it was defined)
          elif option in ['debug', 'verbose']:
              winner = local if local else parent
              if winner is None:
                  winner = 0
              return winner
          else:
              return parent if parent is not None else local
  
      def _level(self) -> int:
          """ Hierarchy level """
          if self.parent is None:
              return -1
          else:
              return self.parent._level() + self._relative_indent
  
      def _indent(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0) -> str:
          """ Indent message according to the object hierarchy """
  
          return indent(
              key,
              value=value,
              color=color,
              level=self._level() + shift)
  
      def _log(self, message: str) -> None:
          """ Append provided message to the current log """
          # Nothing to do if there is no workdir
          if self.workdir is None:
              return
  
          # Store log only in the top parent
          if self.parent:
              self.parent._log(message)
          else:
              with open(os.path.join(self.workdir, LOG_FILENAME), 'a') as log:
                  log.write(datetime.datetime.utcnow().strftime('%H:%M:%S') + ' '
                            + remove_color(message) + '\n')
  
      def print(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          """ Print a message regardless the quiet mode """
          self._log(self._indent(key, value, color=None, shift=shift))
          echo(self._indent(key, value, color, shift), err=err)
  
      def info(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          """ Show a message unless in quiet mode """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet'):
              echo(self._indent(key, value, color, shift), err=err)
  
      def verbose(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          """
          Show message if in requested verbose mode level
  
          In quiet mode verbose messages are not displayed.
          """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet') and self.opt('verbose') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def debug(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          """
          Show message if in requested debug mode level
  
          In quiet mode debug messages are not displayed.
          """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet') and self.opt('debug') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def warn(self, message: str, shift: int = 0) -> None:
          """ Show a yellow warning message on info level, send to stderr """
          self.info('warn', message, color='yellow', shift=shift, err=True)
  
      def fail(self, message: str, shift: int = 0) -> None:
          """ Show a red failure message on info level, send to stderr """
          self.info('fail', message, color='red', shift=shift, err=True)
  
      def _run(self,
               command: Union[str, List[str]],
               cwd: Optional[str],
               shell: bool,
               env: Optional[EnvironmentType],
               log: Optional[BaseLoggerFnType],
               join: bool = False,
               interactive: bool = False,
               timeout: Optional[int] = None) -> CommandOutput:
          """
          Run command, capture the output
  
          By default stdout and stderr are captured separately.
          Use join=True to merge stderr into stdout.
          Use timeout=<seconds> to finish process after given time
          """
          # By default command ouput is logged using debug
          if not log:
              log = self.debug
          # Prepare the environment
          if env:
              if not isinstance(env, dict):
                  raise GeneralError(f"Invalid environment '{env}'.")
              # Do not modify current process environment
              environment = os.environ.copy()
              environment.update(env)
          else:
              environment = None
          self.debug('environment', pprint.pformat(environment), level=4)
  
          # Set only for shell=True as it would affect command
          executable = DEFAULT_SHELL if shell else None
  
          # Run the command in interactive mode if requested
          if interactive:
              try:
                  subprocess.run(
                      command, cwd=cwd, shell=shell, env=environment, check=True,
                      executable=executable)
              except subprocess.CalledProcessError:
                  # Interactive mode can return non-zero if the last command
                  # failed, ignore errors here
                  pass
              finally:
                  return CommandOutput(None, None)
  
          # Create the process
          try:
              process = subprocess.Popen(
                  command, cwd=cwd, shell=shell, env=environment,
                  stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
                  stderr=subprocess.STDOUT if join else subprocess.PIPE,
                  executable=executable)
          except FileNotFoundError as error:
              raise RunError(
                  f"File '{error.filename}' not found.", command, 127)
  
          stdout_thread = StreamLogger(
              process.stdout, log_header='out', logger=log)
          stderr_thread = stdout_thread
          if not join:
              stderr_thread = StreamLogger(
                  process.stderr, log_header='err', logger=log)
          stdout_thread.start()
          if not join:
              stderr_thread.start()
          try:
              process.wait(timeout=timeout)
          except subprocess.TimeoutExpired:
              process.kill()
              process.returncode = PROCESS_TIMEOUT
          stdout_thread.join()
          if not join:
              stderr_thread.join()
  
          # Handle the exit code, return output
          if process.returncode != 0:
              raise RunError(
                  message=f"Command returned '{process.returncode}'.",
                  command=command,
                  returncode=process.returncode,
                  stdout=stdout_thread.get_output(),
                  stderr=stderr_thread.get_output())
          if join:
              return CommandOutput(
                  stdout_thread.get_output(), None)
          else:
              return CommandOutput(
                  stdout_thread.get_output(), stderr_thread.get_output())
  
      def run(self,
              command: Union[str, List[str]],
              message: Optional[str] = None,
              cwd: Optional[str] = None,
              dry: bool = False,
              shell: bool = False,
              env: Optional[EnvironmentType] = None,
              interactive: bool = False,
              join: bool = False,
              log: Optional[BaseLoggerFnType] = None,
              timeout: Optional[int] = None) -> CommandOutput:
          """
          Run command, give message, handle errors
  
          Command is run in the workdir be default.
          In dry mode commands are not executed unless dry=True.
          Environment is updated with variables from the 'env' dictionary.
          Output is logged using self.debug() or custom 'log' function.
          Returns stdout if join=True, (stdout, stderr) tuple otherwise.
          """
  
          # A bit of logging - command, default message, error message for later...
          if isinstance(command, (list, tuple)):
              printable_command = ' '.join(shlex.quote(s) for s in command)
          else:
              printable_command = command
  
          if message:
              self.debug(message, level=2)
  
          self.debug(f'Run command: {printable_command}', level=2)
  
          # Nothing more to do in dry mode (unless requested)
          if self.opt('dry') and not dry:
              return CommandOutput(None, None)
  
          # Run the command, handle the exit code
          cwd = cwd or self.workdir
  
          # Fail nicely if the working directory does not exist
          if cwd and not os.path.exists(cwd):
              raise GeneralError(
                  f"The working directory '{cwd}' does not exist.")
  
          try:
              return self._run(
                  command, cwd, shell, env, log, join, interactive, timeout)
          except RunError as error:
              self.debug(error.message, level=3)
              message = f"Failed to run command: {printable_command} Reason: {error.message}"
              raise RunError(
                  message, error.command, error.returncode,
                  error.stdout, error.stderr)
  
      def read(self, path: str, level: int = 2) -> str:
          """ Read a file from the workdir """
          if self.workdir:
              path = os.path.join(self.workdir, path)
          self.debug(f"Read file '{path}'.", level=level)
          try:
              with open(path, encoding='utf-8', errors='replace') as data:
                  return data.read()
          except OSError as error:
              raise FileError(f"Failed to read '{path}'.\n{error}")
  
      def write(
              self,
              path: str,
              data: Any,
              mode: str = 'w',
              level: int = 2) -> None:
          """ Write a file to the workdir """
          if self.workdir:
              path = os.path.join(self.workdir, path)
          action = 'Append to' if mode == 'a' else 'Write'
          self.debug(f"{action} file '{path}'.", level=level)
          # Dry mode
          if self.opt('dry'):
              return
          try:
              with open(path, mode, encoding='utf-8', errors='replace')
  
  [... 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 = f'{key}: {value}'
  
      return indent + message
  
  
  class Config:
      """ User configuration """
  
      def __init__(self) -> None:
          """ Initialize config directory path """
          self.path = os.path.expanduser(CONFIG_PATH)
          if not os.path.exists(self.path):
              try:
                  os.makedirs(self.path)
              except OSError as error:
                  raise GeneralError(
                      f"Failed to create config '{self.path}'.\n{error}")
  
      def last_run(self, run_id: Optional[str] = None) -> Optional[str]:
          """ Get and set last run id """
          symlink = os.path.join(self.path, 'last-run')
          if run_id:
              try:
                  os.remove(symlink)
              except OSError:
                  pass
              try:
                  os.symlink(run_id, symlink)
              except FileExistsError:
                  # Race when tmt runs in parallel
                  log.warning(f"Race condition, unable to save last run '{run_id}'.")
              except OSError as error:
                  raise GeneralError(
                      f"Unable to save last run '{self.path}'.\n{error}")
              return run_id
          if os.path.islink(symlink):
              return os.path.realpath(symlink)
          return None
  
  
  class StreamLogger(Thread):
      """
      Reading pipes of running process in threads.
  
      Code based on:
      https://github.com/packit/packit/blob/main/packit/utils/logging.py#L10
      """
  
      def __init__(self,
                   stream: Optional[IO[bytes]],
                   log_header: str,
                   logger: BaseLoggerFnType) -> None:
          super().__init__(daemon=True)
          self.stream = stream
          self.output: List[str] = []
          self.log_header = log_header
          self.logger = logger
  
      def run(self) -> None:
          if self.stream is None:
              return
  
          for _line in self.stream:
              line = _line.decode('utf-8', errors='replace')
              if line != '':
                  self.logger(
                      self.log_header,
                      line.rstrip('\n'),
                      'yellow',
                      level=3)
              self.output.append(line)
  
      def get_output(self) -> str:
          return "".join(self.output)
  
  
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  #  Common
  # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  
  CommonDerivedType = TypeVar('CommonDerivedType', bound='Common')
  
  
  class CommandOutput(NamedTuple):
      stdout: Optional[str]
      stderr: Optional[str]
  
  
  class Common:
      """
      Common shared stuff
  
      Takes care of command line context, options and workdir handling.
      Provides logging functions info(), verbose() and debug().
      Implements read() and write() for comfortable file access.
      Provides the run() method for easy command execution.
      """
  
      # Command line context, options and workdir
      _context: Optional[click.Context] = None
      # When set to true, _opt will be ignored (default will be returned)
      ignore_class_options: bool = False
      _options: Dict[str, Any] = dict()
      _workdir: WorkdirType = None
  
      # TODO: must be declared outside of __init__(), because it must exist before
      # __init__() gets called to allow logging helpers work correctly when used
      # from mixins. But that's not very clean, is it? :( Maybe decoupling logging
      # from Common class would help, such a class would be able to initialize
      # itself without involving the rest of Common code. On the other hand,
      # Common owns workdir, for example, whose value affects logging too, so no
      # clear solution so far.
      #
      # Note: cannot use CommonDerivedType - it's a TypeVar filled in by the type
      # given to __init__() and therefore the type it's representing *now* is
      # unknown. but we know `parent` will be derived from `Common` class, so it's
      # mostly fine.
      parent: Optional['Common'] = None
  
      def __init__(
              self,
              parent: Optional[CommonDerivedType] = None,
              name: Optional[str] = None,
              workdir: WorkdirArgumentType = None,
              context: Optional[click.Context] = None,
              relative_indent: int = 1,
              **kwargs: Any) -> None:
          """
          Initialize name and relation with the parent object
  
          Prepare the workdir for provided id / directory path
          or generate a new workdir name if workdir=True given.
          Store command line context and options for future use
          if context is provided.
          """
          # Use lowercase class name as the default name
          self.name = name or self.__class__.__name__.lower()
          self.parent = parent
  
          # Relative log indent level shift against the parent
          self._relative_indent = relative_indent
  
          # Store command line context
          if context:
              self._save_context_to_instance(context)
  
          # Initialize the workdir if requested
          self._workdir_load(workdir)
  
      def __str__(self) -> str:
          """ Name is the default string representation """
          return self.name
  
      @classmethod
      def _save_context(cls, context: click.Context) -> None:
          """ Save provided command line context and options for future use """
          cls._context = context
          cls._options = context.params
  
      def _save_context_to_instance(self, context: click.Context) -> None:
          """ Save provided command line context and options to the instance """
          self._context = context
          self._options = context.params
  
      @overload
      @classmethod
      def _opt(cls, option: str) -> Any:
          pass
  
      @overload
      @classmethod
      def _opt(cls, option: str, default: T) -> T:
          pass
  
      @classmethod
      def _opt(cls, option: str, default: Any = None) -> Any:
          """ Get an option from the command line context (class version) """
          if cls.ignore_class_options:
              return default
          return cls._options.get(option, default)
  
      def _fmf_context(self) -> FmfContextType:
          """ Return the current fmf context """
          if self._context is None:
              return dict()
  
          try:
              return cast(FmfContextType, self._context.obj.fmf_context)
          except AttributeError:
              return dict()
  
      def opt(self, option: str, default: Optional[Any] = None) -> Any:
          """
          Get an option from the command line options
  
          Checks also parent options. For flags (boolean values) parent's
          True wins over child's False (e.g. run --quiet enables quiet
          mode for all included plans and steps).
  
          For options that can be used multiple times, the child overrides
          the parent if it was defined (e.g. run -av provision -vvv runs
          all steps except for provision in mildly verbose mode, provision
          is run with the most verbosity).
  
          Environment variables override command line options.
          """
          # Translate dashes to underscores to match click's conversion
          option = option.replace('-', '_')
          # Check the environment first
          if option == 'debug':
              try:
                  debug = os.environ['TMT_DEBUG']
                  return int(debug)
              except ValueError:
                  raise GeneralError(
                      f"Invalid debug level '{debug}', use an integer.")
              except KeyError:
                  pass
  
          # Get local option
          local = self._options.get(option, default)
          # Check parent option
          parent = None
          if self.parent:
              parent = self.parent.opt(option)
          # Special handling for special flags (parent's yes always wins)
          if option in ['quiet', 'force', 'dry']:
              return parent if parent else local
          # Special handling for counting options (child overrides the
          # parent if it was defined)
          elif option in ['debug', 'verbose']:
              winner = local if local else parent
              if winner is None:
                  winner = 0
              return winner
          else:
              return parent if parent is not None else local
  
      def _level(self) -> int:
          """ Hierarchy level """
          if self.parent is None:
              return -1
          else:
              return self.parent._level() + self._relative_indent
  
      def _indent(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0) -> str:
          """ Indent message according to the object hierarchy """
  
          return indent(
              key,
              value=value,
              color=color,
              level=self._level() + shift)
  
      def _log(self, message: str) -> None:
          """ Append provided message to the current log """
          # Nothing to do if there is no workdir
          if self.workdir is None:
              return
  
          # Store log only in the top parent
          if self.parent:
              self.parent._log(message)
          else:
              with open(os.path.join(self.workdir, LOG_FILENAME), 'a') as log:
                  log.write(datetime.datetime.utcnow().strftime('%H:%M:%S') + ' '
                            + remove_color(message) + '\n')
  
      def print(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          """ Print a message regardless the quiet mode """
          self._log(self._indent(key, value, color=None, shift=shift))
          echo(self._indent(key, value, color, shift), err=err)
  
      def info(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              err: bool = False) -> None:
          """ Show a message unless in quiet mode """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet'):
              echo(self._indent(key, value, color, shift), err=err)
  
      def verbose(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          """
          Show message if in requested verbose mode level
  
          In quiet mode verbose messages are not displayed.
          """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet') and self.opt('verbose') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def debug(
              self,
              key: str,
              value: Optional[str] = None,
              color: Optional[str] = None,
              shift: int = 0,
              level: int = 1,
              err: bool = False) -> None:
          """
          Show message if in requested debug mode level
  
          In quiet mode debug messages are not displayed.
          """
          self._log(self._indent(key, value, color=None, shift=shift))
          if not self.opt('quiet') and self.opt('debug') >= level:
              echo(self._indent(key, value, color, shift), err=err)
  
      def warn(self, message: str, shift: int = 0) -> None:
          """ Show a yellow warning message on info level, send to stderr """
          self.info('warn', message, color='yellow', shift=shift, err=True)
  
      def fail(self, message: str, shift: int = 0) -> None:
          """ Show a red failure message on info level, send to stderr """
          self.info('fail', message, color='red', shift=shift, err=True)
  
      def _run(self,
               command: Union[str, List[str]],
               cwd: Optional[str],
               shell: bool,
               env: Optional[EnvironmentType],
               log: Optional[BaseLoggerFnType],
               join: bool = False,
               interactive: bool = False,
               timeout: Optional[int] = None) -> CommandOutput:
          """
          Run command, capture the output
  
          By default stdout and stderr are captured separately.
          Use join=True to merge stderr into stdout.
          Use timeout=<seconds> to finish process after given time
          """
          # By default command ouput is logged using debug
          if not log:
              log = self.debug
          # Prepare the environment
          if env:
              if not isinstance(env, dict):
                  raise GeneralError(f"Invalid environment '{env}'.")
              # Do not modify current process environment
              environment = os.environ.copy()
              environment.update(env)
          else:
              environment = None
          self.debug('environment', pprint.pformat(environment), level=4)
  
          # Set only for shell=True as it would affect command
          executable = DEFAULT_SHELL if shell else None
  
          # Run the command in interactive mode if requested
          if interactive:
              try:
                  subprocess.run(
                      command, cwd=cwd, shell=shell, env=environment, check=True,
                      executable=executable)
              except subprocess.CalledProcessError:
                  # Interactive mode can return non-zero if the last command
                  # failed, ignore errors here
                  pass
              finally:
                  return CommandOutput(None, None)
  
          # Create the process
          try:
              process = subprocess.Popen(
                  command, cwd=cwd, shell=shell, env=environment,
                  stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
                  stderr=subprocess.STDOUT if join else subprocess.PIPE,
                  executable=executable)
          except FileNotFoundError as error:
              raise RunError(
                  f"File '{error.filename}' not found.", command, 127)
  
          stdout_thread = StreamLogger(
              process.stdout, log_header='out', logger=log)
          stderr_thread = stdout_thread
          if not join:
              stderr_thread = StreamLogger(
                  process.stderr, log_header='err', logger=log)
          stdout_thread.start()
          if not join:
              stderr_thread.start()
          try:
              process.wait(timeout=timeout)
          except subprocess.TimeoutExpired:
              process.kill()
              process.returncode = PROCESS_TIMEOUT
          stdout_thread.join()
          if not join:
              stderr_thread.join()
  
          # Handle the exit code, return output
          if process.returncode != 0:
              raise RunError(
                  message=f"Command returned '{process.returncode}'.",
                  command=command,
                  returncode=process.returncode,
                  stdout=stdout_thread.get_output(),
                  stderr=stderr_thread.get_output())
          if join:
              return CommandOutput(
                  stdout_thread.get_output(), None)
          else:
              return CommandOutput(
                  stdout_thread.get_output(), stderr_thread.get_output())
  
      def run(self,
              command: Union[str, List[str]],
              message: Optional[str] = None,
              cwd: Optional[str] = None,
              dry: bool = False,
              shell: bool = False,
              env: Optional[EnvironmentType] = None,
              interactive: bool = False,
              join: bool = False,
              log: Optional[BaseLoggerFnType] = None,
              timeout: Optional[int] = None) -> CommandOutput:
          """
          Run command, give message, handle errors
  
          Command is run in the workdir be default.
          In dry mode commands are not executed unless dry=True.
          Environment is updated with variables from the 'env' dictionary.
          Output is logged using self.debug() or custom 'log' function.
          Returns stdout if join=True, (stdout, stderr) tuple otherwise.
          """
  
          # A bit of logging - command, default message, error message for later...
          if isinstance(command, (list, tuple)):
              printable_command = ' '.join(shlex.quote(s) for s in command)
          else:
              printable_command = command
  
          if message:
              self.debug(message, level=2)
  
          self.debug(f'Run command: {printable_command}', level=2)
  
          # Nothing more to do in dry mode (unless requested)
          if self.opt('dry') and not dry:
              return CommandOutput(None, None)
  
          # Run the command, handle the exit code
          cwd = cwd or self.workdir
  
          # Fail nicely if the working directory does not exist
          if cwd and not os.path.exists(cwd):
              raise GeneralError(
                  f"The working directory '{cwd}' does not exist.")
  
          try:
              return self._run(
                  command, cwd, shell, env, log, join, interactive, timeout)
          except RunError as error:
              self.debug(error.message, level=3)
              message = f"Failed to run command: {printable_command} Reason: {error.message}"
              raise RunError(
                  message, error.command, error.returncode,
                  error.stdout, error.stderr)
  
      def read(self, path: str, level: int = 2) -> str:
          """ Read a file from the workdir """
          if self.workdir:
              path = os.path.join(self.workdir, path)
          self.debug(f"Read file '{path}'.", level=level)
          try:
              with open(path, encoding='utf-8', errors='replace') as data:
                  return data.read()
          except OSError as error:
              raise FileError(f"Failed to read '{path}'.\n{error}")
  
      def write(
              self,
              path: str,
              data: Any,
              mode: str = 'w',
              level: int = 2) -> None:
          """ Write a file to the workdir """
          if self.workdir:
              path = os.path.join(self.workdir, path)
          action = 'Append to' if mode == 'a' else 'Write'
          self.debug(f"{action} file '{path}'.", level=level)
          # Dry mode
          if self.opt('dry'):
              return
          try:
              with open(path, mode, encoding='utf-8', errors='replace')
  
  [... the rest is omitted ...]
files: tmt/utils.py