files: 
  """ Step Classes """
  
  import re
  import sys
  import textwrap
  import warnings
  from typing import (TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Type,
                      TypeVar, Union, cast, overload)
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  import click
  from click import echo
  
  import tmt.options
  import tmt.utils
  from tmt.options import show_step_method_hints
  
  if TYPE_CHECKING:
      import tmt.base
      from tmt.base import Plan
      from tmt.steps.provision import Guest
  
  
  DEFAULT_PLUGIN_METHOD_ORDER: int = 50
  
  
  # Supported steps and actions
  STEPS = ['discover', 'provision', 'prepare', 'execute', 'report', 'finish']
  ACTIONS = ['login', 'reboot']
  
  # Step phase order
  PHASE_START = 10
  PHASE_BASE = 50
  PHASE_END = 90
  
  
  class Phase(tmt.utils.Common):
      """ A phase of a step """
  
      def __init__(
              self,
              order: int = tmt.utils.DEFAULT_PLUGIN_ORDER,
              *args: Any,
              **kwargs: Any):
          super().__init__(*args,
  
  [... the rest is omitted ...]
files: tmt/steps/__init__.py
files: 
  """ Step Classes """
  
  import re
  import sys
  import textwrap
  import warnings
  from typing import (TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Type,
                      TypeVar, Union, cast, overload)
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  import click
  from click import echo
  
  import tmt.options
  import tmt.utils
  from tmt.options import show_step_method_hints
  
  if TYPE_CHECKING:
      import tmt.base
      from tmt.base import Plan
      from tmt.steps.provision import Guest
  
  
  DEFAULT_PLUGIN_METHOD_ORDER: int = 50
  
  
  # Supported steps and actions
  STEPS = ['discover', 'provision', 'prepare', 'execute', 'report', 'finish']
  ACTIONS = ['login', 'reboot']
  
  # Step phase order
  PHASE_START = 10
  PHASE_BASE = 50
  PHASE_END = 90
  
  
  class Phase(tmt.utils.Common):
      """ A phase of a step """
  
      def __init__(
              self,
              order: int = tmt.utils.DEFAULT_PLUGIN_ORDER,
              *args: Any,
              **kwargs: Any):
          super().__init__(*args,
  
  [... the rest is omitted ...]
files: tmt/steps/__init__.py
files: import glob
  import os
  import re
  import shutil
  import subprocess
  
  import click
  import fmf
  
  import tmt
  import tmt.beakerlib
  import tmt.steps
  import tmt.steps.discover
  from tmt.utils import git_clone
  
  
  @tmt.steps.provides_method('fmf')
  class DiscoverFmf(tmt.steps.discover.DiscoverPlugin):
      """
      Discover available tests from fmf metadata
  
      By default all available tests from the current repository are used
      so the minimal configuration looks like this:
  
          discover:
              how: fmf
  
      Full config example:
  
          discover:
              how: fmf
              url: https://github.com/teemtee/tmt
              ref: main
              path: /fmf/root
              test: /tests/basic
              filter: 'tier: 1'
  
      If no 'ref' is provided, the default branch from the origin is used.
  
      For DistGit repo one can extract source tarball first and discover
      tests from it by using 'distgit-source: true'. It can be used
      together with 'ref', 'path' and 'url', However 'ref' is not possible
      without using 'url'.
  
          discover:
              how: fmf
              dist-git-source: true
  
     
  
  [... the rest is omitted ...]
files: tmt/steps/discover/fmf.py
files: import glob
  import os
  import re
  import shutil
  import subprocess
  
  import click
  import fmf
  
  import tmt
  import tmt.beakerlib
  import tmt.steps
  import tmt.steps.discover
  from tmt.utils import git_clone
  
  
  @tmt.steps.provides_method('fmf')
  class DiscoverFmf(tmt.steps.discover.DiscoverPlugin):
      """
      Discover available tests from fmf metadata
  
      By default all available tests from the current repository are used
      so the minimal configuration looks like this:
  
          discover:
              how: fmf
  
      Full config example:
  
          discover:
              how: fmf
              url: https://github.com/teemtee/tmt
              ref: main
              path: /fmf/root
              test: /tests/basic
              filter: 'tier: 1'
  
      If no 'ref' is provided, the default branch from the origin is used.
  
      For DistGit repo one can extract source tarball first and discover
      tests from it by using 'distgit-source: true'. It can be used
      together with 'ref', 'path' and 'url', However 'ref' is not possible
      without using 'url'.
  
          discover:
              how: fmf
              dist-git-source: true
  
     
  
  [... the rest is omitted ...]
files: tmt/steps/discover/fmf.py
files: import copy
  import os
  from typing import TYPE_CHECKING, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.discover
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  @tmt.steps.provides_method('shell')
  class DiscoverShell(tmt.steps.discover.DiscoverPlugin):
      """
      Use provided list of shell script tests
  
      List of test cases to be executed can be defined manually directly
      in the plan as a list of dictionaries containing test name, actual
      test script and optionally a path to the test. Example config:
  
      discover:
          how: shell
          tests:
          - name: /help/main
            test: tmt --help
          - name: /help/test
            test: tmt test --help
          - name: /help/smoke
            test: ./smoke.sh
            path: /tests/shell
  
      For DistGit repo one can extract source tarball and use its code.
      It is extracted to TMT_SOURCE_DIR however no patches are applied
      (only source tarball is extracted).
  
      discover:
          how: shell
          dist-git-source: true
          tests:
          - name: /upstream
            test: cd $TMT_SOURCE_DIR/*/
  
  [... the rest is omitted ...]
files: tmt/steps/discover/shell.py
files: import copy
  import os
  from typing import TYPE_CHECKING, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.discover
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  @tmt.steps.provides_method('shell')
  class DiscoverShell(tmt.steps.discover.DiscoverPlugin):
      """
      Use provided list of shell script tests
  
      List of test cases to be executed can be defined manually directly
      in the plan as a list of dictionaries containing test name, actual
      test script and optionally a path to the test. Example config:
  
      discover:
          how: shell
          tests:
          - name: /help/main
            test: tmt --help
          - name: /help/test
            test: tmt test --help
          - name: /help/smoke
            test: ./smoke.sh
            path: /tests/shell
  
      For DistGit repo one can extract source tarball and use its code.
      It is extracted to TMT_SOURCE_DIR however no patches are applied
      (only source tarball is extracted).
  
      discover:
          how: shell
          dist-git-source: true
          tests:
          - name: /upstream
            test: cd $TMT_SOURCE_DIR/*/
  
  [... the rest is omitted ...]
files: tmt/steps/discover/shell.py
files: import json
  import os
  import sys
  import time
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.execute
  import tmt.utils
  from tmt.steps.execute import TEST_OUTPUT_FILENAME, Script
  
  # Script handling reboots, in restraint compatible fashion
  TMT_REBOOT_SCRIPT = Script("/usr/local/bin/tmt-reboot",
                             aliases=[
                                 "/usr/local/bin/rstrnt-reboot",
                                 "/usr/local/bin/rhts-reboot"],
                             related_variables=[
                                 "TMT_REBOOT_COUNT",
                                 "REBOOTCOUNT",
                                 "RSTRNT_REBOOTCOUNT"]
                             )
  
  # Script handling result reporting, in restraint compatible fashion
  TMT_REPORT_RESULT_SCRIPT = Script("/usr/local/bin/tmt-report-result",
                                    aliases=[
                                        "/usr/local/bin/rstrnt-report-result",
                                        "/usr/local/bin/rhts-report-result"],
                                    related_variables=[]
                                    )
  
  # Script for archiving a file, usable for BEAKERLIB_COMMAND_SUBMIT_LOG
  TMT_FILE_SUBMIT_SCRIPT = Script("/usr/local/bin/tmt-file-submit",
                                  aliases=[
                                      "/usr/local/bin/rstrnt-report-log",
                                      "/usr/local/bin/rhts-submit-log",
                                      "/usr/local/bin/rhts_submit
  
  [... the rest is omitted ...]
files: tmt/steps/execute/internal.py
files: import json
  import os
  import sys
  import time
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.execute
  import tmt.utils
  from tmt.steps.execute import TEST_OUTPUT_FILENAME, Script
  
  # Script handling reboots, in restraint compatible fashion
  TMT_REBOOT_SCRIPT = Script("/usr/local/bin/tmt-reboot",
                             aliases=[
                                 "/usr/local/bin/rstrnt-reboot",
                                 "/usr/local/bin/rhts-reboot"],
                             related_variables=[
                                 "TMT_REBOOT_COUNT",
                                 "REBOOTCOUNT",
                                 "RSTRNT_REBOOTCOUNT"]
                             )
  
  # Script handling result reporting, in restraint compatible fashion
  TMT_REPORT_RESULT_SCRIPT = Script("/usr/local/bin/tmt-report-result",
                                    aliases=[
                                        "/usr/local/bin/rstrnt-report-result",
                                        "/usr/local/bin/rhts-report-result"],
                                    related_variables=[]
                                    )
  
  # Script for archiving a file, usable for BEAKERLIB_COMMAND_SUBMIT_LOG
  TMT_FILE_SUBMIT_SCRIPT = Script("/usr/local/bin/tmt-file-submit",
                                  aliases=[
                                      "/usr/local/bin/rstrnt-report-log",
                                      "/usr/local/bin/rhts-submit-log",
                                      "/usr/local/bin/rhts_submit
  
  [... the rest is omitted ...]
files: tmt/steps/execute/internal.py
files: from typing import Any, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.finish
  from tmt.steps.provision import Guest
  
  
  @tmt.steps.provides_method('shell')
  class FinishShell(tmt.steps.finish.FinishPlugin):
      """
      Perform finishing tasks using shell scripts
  
      Example config:
  
      finish:
          how: shell
          script:
              - upload-logs.sh || true
              - rm -rf /tmp/temporary-files
  
      Use the 'order' attribute to select in which order finishing tasks
      should happen if there are multiple configs. Default order is '50'.
      """
  
      # Supported keys
      _keys = ["script"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Finish command line options """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '-s', '--script', metavar='SCRIPT',
                  help='Shell script to be executed.')
              ]
          return options
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'script
  
  [... the rest is omitted ...]
files: tmt/steps/finish/shell.py
files: from typing import Any, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.finish
  from tmt.steps.provision import Guest
  
  
  @tmt.steps.provides_method('shell')
  class FinishShell(tmt.steps.finish.FinishPlugin):
      """
      Perform finishing tasks using shell scripts
  
      Example config:
  
      finish:
          how: shell
          script:
              - upload-logs.sh || true
              - rm -rf /tmp/temporary-files
  
      Use the 'order' attribute to select in which order finishing tasks
      should happen if there are multiple configs. Default order is '50'.
      """
  
      # Supported keys
      _keys = ["script"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Finish command line options """
          options = super().options(how)
          options[:0] = [
              click.option(
                  '-s', '--script', metavar='SCRIPT',
                  help='Shell script to be executed.')
              ]
          return options
  
      def default(self, option: str, default: Optional[Any] = None) -> Any:
          """ Return default data for given option """
          if option == 'script
  
  [... the rest is omitted ...]
files: tmt/steps/finish/shell.py
files: import sys
  from typing import Any, List, Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  import tmt.utils
  from tmt.steps import Step
  from tmt.steps.provision import Guest
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  StepDataType = TypedDict(
      'StepDataType',
      {
          'playbook': List[str],
          'playbooks': List[str]
          }
      )
  
  
  @tmt.steps.provides_method('ansible')
  class PrepareAnsible(tmt.steps.prepare.PreparePlugin):  # type: ignore[misc]
      """
      Prepare guest using ansible
  
      Single playbook config:
  
          prepare:
              how: ansible
              playbook: ansible/packages.yml
  
      Multiple playbooks config:
  
          prepare:
              how: ansible
              playbook:
                - playbook/one.yml
                - playbook/two.yml
                - playbook/three.yml
              extra-args: '-vvv'
  
      The playbook path should be relative to the metadata tree root.
      Use 'order' attribute to select in which order preparation should
      happen if there are multiple configs. Default
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/ansible.py
files: import sys
  from typing import Any, List, Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  import tmt.utils
  from tmt.steps import Step
  from tmt.steps.provision import Guest
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  StepDataType = TypedDict(
      'StepDataType',
      {
          'playbook': List[str],
          'playbooks': List[str]
          }
      )
  
  
  @tmt.steps.provides_method('ansible')
  class PrepareAnsible(tmt.steps.prepare.PreparePlugin):  # type: ignore[misc]
      """
      Prepare guest using ansible
  
      Single playbook config:
  
          prepare:
              how: ansible
              playbook: ansible/packages.yml
  
      Multiple playbooks config:
  
          prepare:
              how: ansible
              playbook:
                - playbook/one.yml
                - playbook/two.yml
                - playbook/three.yml
              extra-args: '-vvv'
  
      The playbook path should be relative to the metadata tree root.
      Use 'order' attribute to select in which order preparation should
      happen if there are multiple configs. Default
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/ansible.py
files: import os
  import re
  import shutil
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  
  COPR_URL = 'https://copr.fedorainfracloud.org/coprs'
  
  
  class InstallBase(tmt.utils.Common):
      """ Base class for installation implementations """
  
      # Each installer knows its package manager and copr plugin
      package_manager = None
      copr_plugin = None
  
      def __init__(self, parent, guest):
          """ Initialize installation data """
          super().__init__(parent=parent, relative_indent=0)
          self.guest = guest
  
          # Get package related data from the plugin
          self.packages = self.parent.get("package", [])
          self.directories = self.parent.get("directory", [])
          self.exclude = self.parent.get("exclude", [])
          if not self.packages and not self.directories:
              self.debug("No packages for installation found.", level=3)
  
          # Prepare package lists and installation command
          self.prepare_packages()
          self.prepare_sudo()
          self.prepare_command()
  
      def prepare_packages(self):
          """ Process package names and directories """
          self.local_packages = []
          self.debuginfo_packages = []
          self.repository_packages = []
  
  
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/install.py
files: import os
  import re
  import shutil
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  
  COPR_URL = 'https://copr.fedorainfracloud.org/coprs'
  
  
  class InstallBase(tmt.utils.Common):
      """ Base class for installation implementations """
  
      # Each installer knows its package manager and copr plugin
      package_manager = None
      copr_plugin = None
  
      def __init__(self, parent, guest):
          """ Initialize installation data """
          super().__init__(parent=parent, relative_indent=0)
          self.guest = guest
  
          # Get package related data from the plugin
          self.packages = self.parent.get("package", [])
          self.directories = self.parent.get("directory", [])
          self.exclude = self.parent.get("exclude", [])
          if not self.packages and not self.directories:
              self.debug("No packages for installation found.", level=3)
  
          # Prepare package lists and installation command
          self.prepare_packages()
          self.prepare_sudo()
          self.prepare_command()
  
      def prepare_packages(self):
          """ Process package names and directories """
          self.local_packages = []
          self.debuginfo_packages = []
          self.repository_packages = []
  
  
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/install.py
files: from typing import Any, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  import tmt.utils
  from tmt.steps.provision import Guest
  
  
  # TODO: drop ignore once type annotations between modules enabled
  @tmt.steps.provides_method('shell')
  class PrepareShell(tmt.steps.prepare.PreparePlugin):  # type: ignore[misc]
      """
      Prepare guest using shell scripts
  
      Example config:
  
      prepare:
          how: shell
          script:
          - sudo dnf install -y 'dnf-command(copr)'
          - sudo dnf copr enable -y psss/tmt
          - sudo dnf install -y tmt
  
      Use 'order' attribute to select in which order preparation should
      happen if there are multiple configs. Default order is '50'.
      Default order of required packages installation is '70'.
      """
  
      # Supported keys
      _keys = ["script"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> Any:
          """ Prepare command line options """
          return [
              click.option(
                  '-s', '--script', metavar='SCRIPT',
                  help='Shell script to be executed.')
              ] + super().options(how
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/shell.py
files: from typing import Any, List, Optional
  
  import click
  import fmf
  
  import tmt
  import tmt.steps
  import tmt.steps.prepare
  import tmt.utils
  from tmt.steps.provision import Guest
  
  
  # TODO: drop ignore once type annotations between modules enabled
  @tmt.steps.provides_method('shell')
  class PrepareShell(tmt.steps.prepare.PreparePlugin):  # type: ignore[misc]
      """
      Prepare guest using shell scripts
  
      Example config:
  
      prepare:
          how: shell
          script:
          - sudo dnf install -y 'dnf-command(copr)'
          - sudo dnf copr enable -y psss/tmt
          - sudo dnf install -y tmt
  
      Use 'order' attribute to select in which order preparation should
      happen if there are multiple configs. Default order is '50'.
      Default order of required packages installation is '70'.
      """
  
      # Supported keys
      _keys = ["script"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> Any:
          """ Prepare command line options """
          return [
              click.option(
                  '-s', '--script', metavar='SCRIPT',
                  help='Shell script to be executed.')
              ] + super().options
  
  [... the rest is omitted ...]
files: tmt/steps/prepare/shell.py
files: import collections
  import dataclasses
  import datetime
  import enum
  import os
  import random
  import re
  import shlex
  import string
  import subprocess
  import tempfile
  from shlex import quote
  from typing import Any, Dict, List, Optional, Type, Union, cast
  
  import click
  import fmf
  
  import tmt
  import tmt.plugins
  import tmt.steps
  import tmt.utils
  from tmt.steps import Action
  
  # Timeout in seconds of waiting for a connection after reboot
  CONNECTION_TIMEOUT = 5 * 60
  
  # When waiting for guest to recover from reboot, try re-connecting every
  # this many seconds.
  RECONNECT_WAIT_TICK = 5
  RECONNECT_WAIT_TICK_INCREASE = 1.0
  
  # Default rsync options
  DEFAULT_RSYNC_OPTIONS = [
      "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  DEFAULT_RSYNC_PUSH_OPTIONS = ["-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  DEFAULT_RSYNC_PULL_OPTIONS = ["-R", "-r", "-z", "--links", "--safe-links", "--protect-args"]
  
  
  class Check
  
  [... the rest is omitted ...]
files: tmt/steps/provision/__init__.py
files: import collections
  import dataclasses
  import datetime
  import enum
  import os
  import random
  import re
  import shlex
  import string
  import subprocess
  import tempfile
  from shlex import quote
  from typing import Any, Dict, List, Optional, Type, Union, cast
  
  import click
  import fmf
  
  import tmt
  import tmt.plugins
  import tmt.steps
  import tmt.utils
  from tmt.steps import Action
  
  # Timeout in seconds of waiting for a connection after reboot
  CONNECTION_TIMEOUT = 5 * 60
  
  # When waiting for guest to recover from reboot, try re-connecting every
  # this many seconds.
  RECONNECT_WAIT_TICK = 5
  RECONNECT_WAIT_TICK_INCREASE = 1.0
  
  # Default rsync options
  DEFAULT_RSYNC_OPTIONS = [
      "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  DEFAULT_RSYNC_PUSH_OPTIONS = ["-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  DEFAULT_RSYNC_PULL_OPTIONS = ["-R", "-r", "-z", "--links", "--safe-links", "--protect-args"]
  
  
  class Check
  
  [... the rest is omitted ...]
files: tmt/steps/provision/__init__.py
files: import dataclasses
  import datetime
  import sys
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import requests
  
  import tmt
  import tmt.options
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import ProvisionError, retry_session, updatable_message
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  
  # List of Artemis API versions supported and understood by this plugin.
  # Since API gains support for new features over time, it is important to
  # know when particular feature became available, and avoid using it with
  # older APIs.
  SUPPORTED_API_VERSIONS = (
      # NEW: added hostname HW constraint
      '0.0.38',
      # NEW: virtualization HW constraint
      '0.0.37',
      # NEW: boot.method HW constraint
      '0.0.32',
      # NEW: network HW constraint
      '0.0.28'
      )
  
  
  # The default Artemis API version - the most recent supported versions
  # should be perfectly fine.
  DEFAULT_API_VERSION = SUPPORTED_API_VERSIONS[0]
  
  DEFAULT_API_URL = 'http://127.0.0.1:8001'
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/artemis.py
files: import dataclasses
  import datetime
  import sys
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import requests
  
  import tmt
  import tmt.options
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import ProvisionError, retry_session, updatable_message
  
  if sys.version_info >= (3, 8):
      from typing import TypedDict
  else:
      from typing_extensions import TypedDict
  
  
  # List of Artemis API versions supported and understood by this plugin.
  # Since API gains support for new features over time, it is important to
  # know when particular feature became available, and avoid using it with
  # older APIs.
  SUPPORTED_API_VERSIONS = (
      # NEW: added hostname HW constraint
      '0.0.38',
      # NEW: virtualization HW constraint
      '0.0.37',
      # NEW: boot.method HW constraint
      '0.0.32',
      # NEW: network HW constraint
      '0.0.28'
      )
  
  
  # The default Artemis API version - the most recent supported versions
  # should be perfectly fine.
  DEFAULT_API_VERSION = SUPPORTED_API_VERSIONS[0]
  
  DEFAULT_API_URL = 'http://127.0.0.1:8001'
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/artemis.py
files: from typing import Any, List, Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.steps.provision import GuestSshData
  
  
  @tmt.steps.provides_method('connect')
  class ProvisionConnect(tmt.steps.provision.ProvisionPlugin):
      """
      Connect to a provisioned guest using ssh
  
      Private key authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              key: /home/psss/.ssh/example_rsa
  
      Password authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              password: secret
  
      User defaults to 'root', so if you have private key correctly set
      the minimal configuration can look like this:
  
          provision:
              how: connect
              guest: host.example.org
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["guest", "key", "user", "password", "port"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for connect """
          options = super().options(how)
          options[:0
  
  [... the rest is omitted ...]
files: tmt/steps/provision/connect.py
files: from typing import Any, List, Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.steps.provision import GuestSshData
  
  
  @tmt.steps.provides_method('connect')
  class ProvisionConnect(tmt.steps.provision.ProvisionPlugin):
      """
      Connect to a provisioned guest using ssh
  
      Private key authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              key: /home/psss/.ssh/example_rsa
  
      Password authentication:
  
          provision:
              how: connect
              guest: host.example.org
              user: root
              password: secret
  
      User defaults to 'root', so if you have private key correctly set
      the minimal configuration can look like this:
  
          provision:
              how: connect
              guest: host.example.org
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys = ["guest", "key", "user", "password", "port"]
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for connect """
          options = super().options(how)
          options[:0
  
  [... the rest is omitted ...]
files: tmt/steps/provision/connect.py
files: import tmt
  import tmt.steps
  import tmt.steps.provision
  
  
  @tmt.steps.provides_method('local')
  class ProvisionLocal(tmt.steps.provision.ProvisionPlugin):
      """
      Use local host for test execution
  
      In general it is not recommended to run tests on your local machine
      as there might be security risks. Run only those tests which you
      know are safe so that you don't destroy your laptop ;-)
  
      Example config:
  
          provision:
              how: local
  
      Note that 'tmt run' is expected to be executed under a regular user.
      If there are admin rights required (for example in the prepare step)
      you might be asked for a sudo password.
      """
  
      # Guest instance
      _guest = None
  
      def wake(self, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
          if data:
              self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Create a GuestLocal instance
          data = tmt.steps.provision.GuestSshData(
              guest='localhost',
              role=self.get('role')
              )
          self._guest = GuestLocal(data, name=self.name,
  
  [... the rest is omitted ...]
files: tmt/steps/provision/local.py
files: import tmt
  import tmt.steps
  import tmt.steps.provision
  
  
  @tmt.steps.provides_method('local')
  class ProvisionLocal(tmt.steps.provision.ProvisionPlugin):
      """
      Use local host for test execution
  
      In general it is not recommended to run tests on your local machine
      as there might be security risks. Run only those tests which you
      know are safe so that you don't destroy your laptop ;-)
  
      Example config:
  
          provision:
              how: local
  
      Note that 'tmt run' is expected to be executed under a regular user.
      If there are admin rights required (for example in the prepare step)
      you might be asked for a sudo password.
      """
  
      # Guest instance
      _guest = None
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          if data:
              self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Create a GuestLocal instance
          data = tmt.steps.provision.GuestSshData(
              guest='localhost',
              role=self.get('role')
              )
          self._guest =
  
  [... the rest is omitted ...]
files: tmt/steps/provision/local.py
files: import dataclasses
  import os
  from shlex import quote
  from typing import Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  
  # Timeout in seconds of waiting for a connection
  CONNECTION_TIMEOUT = 60
  
  # Defaults
  DEFAULT_IMAGE = "fedora"
  DEFAULT_USER = "root"
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class PodmanGuestData(tmt.steps.provision.GuestData):  # type: ignore[misc]
      image: str = DEFAULT_IMAGE
      user: str = DEFAULT_USER
      force_pull: bool = False
  
      container: Optional[str] = None
  
  
  @tmt.steps.provides_method('container')
  class ProvisionPodman(tmt.steps.provision.ProvisionPlugin):
      """
      Create a new container using podman
  
      Example config:
  
          provision:
              how: container
              image: fedora:latest
  
      In order to always pull the fresh container image use 'pull: true'.
  
      In order to run the container with different user as the default 'root',
      use 'user: USER'.
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys
  
  [... the rest is omitted ...]
files: tmt/steps/provision/podman.py
files: import dataclasses
  import os
  from shlex import quote
  from typing import Optional
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  
  # Timeout in seconds of waiting for a connection
  CONNECTION_TIMEOUT = 60
  
  # Defaults
  DEFAULT_IMAGE = "fedora"
  DEFAULT_USER = "root"
  
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class PodmanGuestData(tmt.steps.provision.GuestData):  # type: ignore[misc]
      image: str = DEFAULT_IMAGE
      user: str = DEFAULT_USER
      force_pull: bool = False
  
      container: Optional[str] = None
  
  
  @tmt.steps.provides_method('container')
  class ProvisionPodman(tmt.steps.provision.ProvisionPlugin):
      """
      Create a new container using podman
  
      Example config:
  
          provision:
              how: container
              image: fedora:latest
  
      In order to always pull the fresh container image use 'pull: true'.
  
      In order to run the container with different user as the default 'root',
      use 'user: USER'.
      """
  
      # Guest instance
      _guest = None
  
      # Supported keys
      _keys
  
  [... the rest is omitted ...]
files: tmt/steps/provision/podman.py
files: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  from typing import Optional
  
  import click
  import requests
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  
  def import_testcloud():
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
    - {public_key}
  ssh_pwauth
  
  [... the rest is omitted ...]
files: tmt/steps/provision/testcloud.py
files: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  from typing import Optional
  
  import click
  import requests
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  
  def import_testcloud():
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
    - {public_key}
  ssh_pwauth
  
  [... the rest is omitted ...]
files: tmt/steps/provision/testcloud.py