files.new_contents: 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 TYPE_CHECKING, 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
  from tmt.utils import BaseLoggerFnType
  
  if TYPE_CHECKING:
      import tmt.cli
  
  
  # 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 = [
      "-s", "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  DEFAULT_RSYNC_PUSH_OPTIONS = ["-s", "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  DEFAULT_RSYNC_PULL_OPTIONS = ["-s", "-R", "-r", "-z", "--links", "--safe-links", "--protect-args"]
  
  
  class CheckRsyncOutcome(enum.Enum):
      ALREADY
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/__init__.py
files.old_contents: 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 TYPE_CHECKING, 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
  from tmt.utils import BaseLoggerFnType
  
  if TYPE_CHECKING:
      import tmt.cli
  
  
  # 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 = [
      "-s", "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  DEFAULT_RSYNC_PUSH_OPTIONS = ["-s", "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  DEFAULT_RSYNC_PULL_OPTIONS = ["-s", "-R", "-r", "-z", "--links", "--safe-links", "--protect-args"]
  
  
  class CheckRsyncOutcome(enum.Enum):
      ALREADY
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/__init__.py
files.new_contents: import dataclasses
  import datetime
  import sys
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import fmf.utils
  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 missing cpu.processors constraint
      '0.0.47',
      # NEW: added new CPU constraints
      '0.0.46',
      # 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'
  DEFAULT_USER = 'root'
  
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/artemis.py
files.old_contents: import dataclasses
  import datetime
  import sys
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import fmf.utils
  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 missing cpu.processors constraint
      '0.0.47',
      # NEW: added new CPU constraints
      '0.0.46',
      # 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'
  DEFAULT_USER = 'root'
  
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/artemis.py
files.new_contents: import dataclasses
  from typing import List, Optional, cast
  
  import click
  import fmf.utils
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  
  DEFAULT_USER = "root"
  
  
  @dataclasses.dataclass
  class ConnectGuestData(tmt.steps.provision.GuestSshData):
      user: str = DEFAULT_USER
  
  
  @dataclasses.dataclass
  class ProvisionConnectData(ConnectGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  @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
      """
  
      _data_class = ProvisionConnectData
      _guest_class = tmt.steps.provision.GuestSsh
  
      # Guest instance
      _guest = None
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
         
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/connect.py
files.old_contents: import dataclasses
  from typing import List, Optional, cast
  
  import click
  import fmf.utils
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.steps.provision import GuestSshData
  
  DEFAULT_USER = "root"
  
  
  @dataclasses.dataclass
  class ConnectGuestData(tmt.steps.provision.GuestSshData):
      user: str = DEFAULT_USER
  
  
  @dataclasses.dataclass
  class ProvisionConnectData(ConnectGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  @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
      """
  
      _data_class = ProvisionConnectData
      _guest_class = tmt.steps.provision.GuestSsh
  
      # Guest instance
      _guest = None
  
      @classmethod
      def options(cls, how: Optional[str] = None
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/connect.py
files.new_contents: import dataclasses
  import shlex
  from typing import Any, List, Optional, Union
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import BaseLoggerFnType
  
  
  @dataclasses.dataclass
  class ProvisionLocalData(tmt.steps.provision.GuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  class GuestLocal(tmt.Guest):
      """ Local Host """
  
      localhost = True
      parent: tmt.steps.Step
  
      @property
      def is_ready(self) -> bool:
          """ Local is always ready """
          return True
  
      def ansible(self, playbook: str, extra_args: Optional[str] = None) -> None:
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, _ = self.run(
              ['sudo', '-E', 'ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'local', '-i', 'localhost,', playbook],
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def execute(self,
                  command: Union[List[str], str],
                  friendly_command: Optional[str] = None,
                  test_session: bool = False,
                  silent: bool = False,
                  log: Optional[BaseLoggerFnType] = None,
                  **kwargs: Any
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/local.py
files.old_contents: import dataclasses
  import shlex
  from typing import Any, List, Optional, Union
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import BaseLoggerFnType
  
  
  @dataclasses.dataclass
  class ProvisionLocalData(tmt.steps.provision.GuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  class GuestLocal(tmt.Guest):
      """ Local Host """
  
      localhost = True
      parent: tmt.steps.Step
  
      @property
      def is_ready(self) -> bool:
          """ Local is always ready """
          return True
  
      def ansible(self, playbook: str, extra_args: Optional[str] = None) -> None:
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, _ = self.run(
              ['sudo', '-E', 'ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'local', '-i', 'localhost,', playbook],
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def execute(self,
                  command: Union[List[str], str],
                  friendly_command: Optional[str] = None,
                  test_session: bool = False,
                  silent: bool = False,
                  log: Optional[BaseLoggerFnType] = None,
                  **kwargs: Any
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/local.py
files.new_contents: import dataclasses
  import os
  from shlex import quote
  from typing import Any, List, Optional, Tuple, Union
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  from tmt.utils import BaseLoggerFnType
  
  # Timeout in seconds of waiting for a connection
  CONNECTION_TIMEOUT = 60
  
  # Defaults
  DEFAULT_IMAGE = "fedora"
  DEFAULT_USER = "root"
  
  
  @dataclasses.dataclass
  class PodmanGuestData(tmt.steps.provision.GuestData):
      image: str = DEFAULT_IMAGE
      user: str = DEFAULT_USER
      force_pull: bool = False
  
      container: Optional[str] = None
  
  
  @dataclasses.dataclass
  class ProvisionPodmanData(PodmanGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  class GuestContainer(tmt.Guest):
      """ Container Instance """
  
      _data_class = PodmanGuestData
  
      image: Optional[str]
      container: Optional[str]
      user: str
      force_pull: bool
      parent: tmt.steps.Step
  
      @property
      def is_ready(self) -> bool:
          """ Detect the guest is ready or not """
          # Check the container is running or not
          if self.container is None:
              return False
          cmd_output = self.podman(['container', 'inspect', '--format',
                                    '{{json .State.Running}}', self.container])
          cmd_stdout, cmd
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/podman.py
files.old_contents: import dataclasses
  import os
  from shlex import quote
  from typing import Any, List, Optional, Tuple, Union
  
  import click
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  from tmt.utils import BaseLoggerFnType
  
  # Timeout in seconds of waiting for a connection
  CONNECTION_TIMEOUT = 60
  
  # Defaults
  DEFAULT_IMAGE = "fedora"
  DEFAULT_USER = "root"
  
  
  @dataclasses.dataclass
  class PodmanGuestData(tmt.steps.provision.GuestData):
      image: str = DEFAULT_IMAGE
      user: str = DEFAULT_USER
      force_pull: bool = False
  
      container: Optional[str] = None
  
  
  @dataclasses.dataclass
  class ProvisionPodmanData(PodmanGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  class GuestContainer(tmt.Guest):
      """ Container Instance """
  
      _data_class = PodmanGuestData
  
      image: Optional[str]
      container: Optional[str]
      user: str
      force_pull: bool
      parent: tmt.steps.Step
  
      @property
      def is_ready(self) -> bool:
          """ Detect the guest is ready or not """
          # Check the container is running or not
          if self.container is None:
              return False
          cmd_output = self.podman(['container', 'inspect', '--format',
                                    '{{json .State.Running}}', self.container])
          cmd_stdout, cmd
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/podman.py
files.new_contents: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  import types
  from typing import TYPE_CHECKING, List, Optional
  
  import click
  import fmf
  import fmf.utils
  import requests
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  libvirt: Optional[types.ModuleType] = None
  testcloud: Optional[types.ModuleType] = None
  
  
  def import_testcloud() -> None:
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
  
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/testcloud.py
files.old_contents: # coding: utf-8
  
  import dataclasses
  import datetime
  import os
  import platform
  import re
  import time
  import types
  from typing import TYPE_CHECKING, List, Optional
  
  import click
  import fmf
  import fmf.utils
  import requests
  
  import tmt
  import tmt.steps
  import tmt.steps.provision
  import tmt.utils
  from tmt.utils import WORKDIR_ROOT, ProvisionError, retry_session
  
  if TYPE_CHECKING:
      import tmt.base
  
  
  libvirt: Optional[types.ModuleType] = None
  testcloud: Optional[types.ModuleType] = None
  
  
  def import_testcloud() -> None:
      """
      Import testcloud module only when needed
  
      Until we have a separate package for each plugin.
      """
      global testcloud
      global libvirt
      try:
          import libvirt
          import testcloud.image
          import testcloud.instance
          import testcloud.util
      except ImportError:
          raise ProvisionError(
              "Install 'testcloud' to provision using this method.")
  
  
  # Testcloud cache to our tmt's workdir root
  TESTCLOUD_DATA = os.path.join(WORKDIR_ROOT, 'testcloud')
  TESTCLOUD_IMAGES = os.path.join(TESTCLOUD_DATA, 'images')
  
  # Userdata for cloud-init
  USER_DATA = """#cloud-config
  chpasswd:
    list: |
      {user_name}:%s
    expire: false
  users:
    - default
    - name: {user_name}
  ssh_authorized_keys:
  
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/testcloud.py