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 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_INSTALLED = 'already-installed'
      INSTALLED = 'installed'
  
  
  @dataclasses.dataclass
  class GuestData(tmt.utils.SerializableContainer):
      """
      Keys necessary to describe, create, save and restore a guest.
  
      Very basic set of keys shared across all known guest classes.
      """
  
      # guest role in the multihost scenario
      role: Optional[str] = None
      # hostname or ip address
      guest: Optional[str] = None
  
  
  class Guest(tmt.utils.Common):
      """
      Guest provisioned for test execution
  
      A base class for guest-like classes. Provides some of the basic methods
      and functionality, but note some of the methods are left intentionally
      empty. These do not have valid implementation on this level, and it's up
      to Guest subclasses to provide one working in their respective
      infrastructure.
  
      The following keys are expected in the 'data' container::
  
          role ....... guest role in the multihost scenario
          guest ...... name, hostname or ip address
  
      These are by default imported into instance attributes.
      """
  
      # Used by save() to construct the correct container for keys.
      _data_class: Type[GuestData] = GuestData
  
      role: Optional[str]
      guest: Optional[str]
  
      # Flag to indicate localhost guest, requires special handling
      localhost = False
  
      # TODO: do we need this list? Can whatever code is using it use _data_class directly?
      # List of supported keys
      # (used for import/export to/from attributes during load and save)
      @property
      def _keys(self) -> List[str]:
          return list(self._data_class.keys())
  
      def __init__(self,
                   *,
                   data: GuestData,
                   name: Optional[str] = None,
                   parent: Optional[tmt.utils.Common] = None) -> None:
          """ Initialize guest data """
          super().__init__(parent=parent, name=name)
  
          self.load(data)
  
      def _random_name(self, prefix: str = '', length: int = 16) -> str:
          """ Generate a random name """
          # Append at least 5 random characters
          min_random_part = max(5, length - len(prefix))
          name = prefix + ''.join(
              random.choices(string.ascii_letters, k=min_random_part))
          # Return tail (containing random characters) of name
          return name[-length:]
  
      def _tmt_name(self) -> str:
          """ Generate a name prefixed with tmt run id """
          # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
          parent = cast(Provision, self.parent)
  
          assert parent.plan.my_run is not None  # narrow type
          assert parent.plan.my_run.workdir is not None  # narrow type
          _, run_id = os.path.split(parent.plan.my_run.workdir)
          return self._random_name(prefix="tmt-{0}-".format(run_id[-3:]))
  
      @property
      def is_ready(self) -> bool:
          """ Detect guest is ready or not """
  
          raise NotImplementedError()
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options related to guests """
          return list()
  
      def load(self, data: GuestData) -> None:
          """
          Load guest data into object attributes for easy access
  
          Called during guest object initialization. Takes care of storing
          all supported keys (see class attribute _keys for the list) from
          provided data to the guest object attributes. Child classes can
          extend it to make additional guest attributes easily available.
  
          Data dictionary can contain guest information from both command
          line options / L2 metadata / user configuration and wake up data
          stored by the save() method below.
          """
          data.inject_to(self)
  
      def save(self) -> GuestData:
          """
          Save guest data for future wake up
  
          Export all essential guest data into a dictionary which will be
          stored in the `guests.yaml` file for possible future wake up of
          the guest. Everything needed to attach to a running instance
          should be added into the data dictionary by child classes.
          """
          return self._data_class.extract_from(self)
  
      def wake(self) -> None:
          """
          Wake up the guest
  
          Perform any actions necessary after step wake up to be able to
          attach to a running guest instance and execute commands. Called
          after load() is completed so all guest data should be prepared.
          """
          self
  
  [... 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 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_INSTALLED = 'already-installed'
      INSTALLED = 'installed'
  
  
  @dataclasses.dataclass
  class GuestData(tmt.utils.SerializableContainer):
      """
      Keys necessary to describe, create, save and restore a guest.
  
      Very basic set of keys shared across all known guest classes.
      """
  
      # guest role in the multihost scenario
      role: Optional[str] = None
      # hostname or ip address
      guest: Optional[str] = None
  
  
  class Guest(tmt.utils.Common):
      """
      Guest provisioned for test execution
  
      A base class for guest-like classes. Provides some of the basic methods
      and functionality, but note some of the methods are left intentionally
      empty. These do not have valid implementation on this level, and it's up
      to Guest subclasses to provide one working in their respective
      infrastructure.
  
      The following keys are expected in the 'data' container::
  
          role ....... guest role in the multihost scenario
          guest ...... name, hostname or ip address
  
      These are by default imported into instance attributes.
      """
  
      # Used by save() to construct the correct container for keys.
      _data_class: Type[GuestData] = GuestData
  
      role: Optional[str]
      guest: Optional[str]
  
      # Flag to indicate localhost guest, requires special handling
      localhost = False
  
      # TODO: do we need this list? Can whatever code is using it use _data_class directly?
      # List of supported keys
      # (used for import/export to/from attributes during load and save)
      @property
      def _keys(self) -> List[str]:
          return list(self._data_class.keys())
  
      def __init__(self,
                   *,
                   data: GuestData,
                   name: Optional[str] = None,
                   parent: Optional[tmt.utils.Common] = None) -> None:
          """ Initialize guest data """
          super().__init__(parent=parent, name=name)
  
          self.load(data)
  
      def _random_name(self, prefix: str = '', length: int = 16) -> str:
          """ Generate a random name """
          # Append at least 5 random characters
          min_random_part = max(5, length - len(prefix))
          name = prefix + ''.join(
              random.choices(string.ascii_letters, k=min_random_part))
          # Return tail (containing random characters) of name
          return name[-length:]
  
      def _tmt_name(self) -> str:
          """ Generate a name prefixed with tmt run id """
          # FIXME: cast() - https://github.com/teemtee/tmt/issues/1372
          parent = cast(Provision, self.parent)
  
          assert parent.plan.my_run is not None  # narrow type
          assert parent.plan.my_run.workdir is not None  # narrow type
          _, run_id = os.path.split(parent.plan.my_run.workdir)
          return self._random_name(prefix="tmt-{0}-".format(run_id[-3:]))
  
      @property
      def is_ready(self) -> bool:
          """ Detect guest is ready or not """
  
          raise NotImplementedError()
  
      @classmethod
      def options(cls, how: Optional[str] = None) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options related to guests """
          return list()
  
      def load(self, data: GuestData) -> None:
          """
          Load guest data into object attributes for easy access
  
          Called during guest object initialization. Takes care of storing
          all supported keys (see class attribute _keys for the list) from
          provided data to the guest object attributes. Child classes can
          extend it to make additional guest attributes easily available.
  
          Data dictionary can contain guest information from both command
          line options / L2 metadata / user configuration and wake up data
          stored by the save() method below.
          """
          data.inject_to(self)
  
      def save(self) -> GuestData:
          """
          Save guest data for future wake up
  
          Export all essential guest data into a dictionary which will be
          stored in the `guests.yaml` file for possible future wake up of
          the guest. Everything needed to attach to a running instance
          should be added into the data dictionary by child classes.
          """
          return self._data_class.extract_from(self)
  
      def wake(self) -> None:
          """
          Wake up the guest
  
          Perform any actions necessary after step wake up to be able to
          attach to a running guest instance and execute commands. Called
          after load() is completed so all guest data should be prepared.
          """
          self
  
  [... 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 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'
  DEFAULT_ARCH = 'x86_64'
  DEFAULT_PRIORITY_GROUP = 'default-priority'
  DEFAULT_KEYNAME = 'default'
  DEFAULT_PROVISION_TIMEOUT = 600
  DEFAULT_PROVISION_TICK = 60
  DEFAULT_API_TIMEOUT = 10
  DEFAULT_API_RETRIES = 10
  # Should lead to delays of 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256 seconds
  DEFAULT_RETRY_BACKOFF_FACTOR = 1
  
  # Type annotation for "data" package describing a guest instance. Passed
  # between load() and save() calls
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class ArtemisGuestData(tmt.steps.provision.GuestSshData):
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      # API
      api_url: str = DEFAULT_API_URL
      api_version: str = DEFAULT_API_VERSION
  
      # Guest request properties
      arch: str = DEFAULT_ARCH
      image: Optional[str] = None
      hardware: Optional[Any] = None
      pool: Optional[str] = None
      priority_group: str = DEFAULT_PRIORITY_GROUP
      keyname: str = DEFAULT_KEYNAME
      user_data: Dict[str, str] = dataclasses.field(default_factory=dict)
  
      # Provided by Artemis response
      guestname: Optional[str] = None
  
      # Timeouts and deadlines
      provision_timeout: int = DEFAULT_PROVISION_TIMEOUT
      provision_tick: int = DEFAULT_PROVISION_TICK
      api_timeout: int = DEFAULT_API_TIMEOUT
      api_retries: int = DEFAULT_API_RETRIES
      api_retry_backoff_factor: int = DEFAULT_RETRY_BACKOFF_FACTOR
  
  
  @dataclasses.dataclass
  class ProvisionArtemisData(ArtemisGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  GUEST_STATE_COLOR_DEFAULT = 'green'
  
  GUEST_STATE_COLORS = {
      'routing': 'yellow',
      'provisioning': 'magenta',
      'promised': 'blue',
      'preparing': 'cyan',
      'cancelled': 'red',
      'error': 'red'
      }
  
  
  # Type annotation for Artemis API `GET /guests/$guestname` response.
  # Partial, not all fields necessary since plugin ignores most of them.
  GuestInspectType = TypedDict(
      'GuestInspectType', {
          'state': str,
          'address': Optional[str]
          }
      )
  
  
  class ArtemisAPI:
      def __init__(self, guest: 'GuestArtemis') -> None:
          self._guest = guest
  
          self.http_session = retry_session.create(
              retries=guest.api_retries,
              backoff_factor=guest.api_retry_backoff_factor,
              allowed_methods=('HEAD', 'GET', 'POST', 'DELETE', 'PUT'),
              status_forcelist=(
                  429,  # Too Many Requests
                  500,  # Internal Server Error
                  502,  # Bad Gateway
                  503,  # Service Unavailable
                  504   # Gateway Timeout
                  ),
              timeout=guest.api_timeout
              )
  
      def query(
              self,
              path: str,
              method: str = 'get',
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Base helper for Artemis API queries.
  
          Trivial dispatcher per method, returning retrieved response.
  
          :param path: API path to contact.
          :param method: HTTP method to use.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          url = f'{self._guest.api_url}{path}'
  
          if method == 'get':
              return self.http_session.get(url, **request_kwargs)
  
          if method == 'post':
              return self.http_session.post(url, **request_kwargs)
  
          if method == 'delete':
              return self.http_session.delete(url, **request_kwargs)
  
          if method == 'put':
              return self.http_session.put(url, **request_kwargs)
  
          raise tmt.utils.GeneralError(
              f'Unsupported Artemis API method.\n{method}')
  
      def create(
              self,
  
  
  [... 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 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'
  DEFAULT_ARCH = 'x86_64'
  DEFAULT_PRIORITY_GROUP = 'default-priority'
  DEFAULT_KEYNAME = 'default'
  DEFAULT_PROVISION_TIMEOUT = 600
  DEFAULT_PROVISION_TICK = 60
  DEFAULT_API_TIMEOUT = 10
  DEFAULT_API_RETRIES = 10
  # Should lead to delays of 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256 seconds
  DEFAULT_RETRY_BACKOFF_FACTOR = 1
  
  # Type annotation for "data" package describing a guest instance. Passed
  # between load() and save() calls
  # TODO: get rid of `ignore` once superclass is no longer `Any`
  
  
  @dataclasses.dataclass
  class ArtemisGuestData(tmt.steps.provision.GuestSshData):
      # Override parent class with our defaults
      user: str = DEFAULT_USER
  
      # API
      api_url: str = DEFAULT_API_URL
      api_version: str = DEFAULT_API_VERSION
  
      # Guest request properties
      arch: str = DEFAULT_ARCH
      image: Optional[str] = None
      hardware: Optional[Any] = None
      pool: Optional[str] = None
      priority_group: str = DEFAULT_PRIORITY_GROUP
      keyname: str = DEFAULT_KEYNAME
      user_data: Dict[str, str] = dataclasses.field(default_factory=dict)
  
      # Provided by Artemis response
      guestname: Optional[str] = None
  
      # Timeouts and deadlines
      provision_timeout: int = DEFAULT_PROVISION_TIMEOUT
      provision_tick: int = DEFAULT_PROVISION_TICK
      api_timeout: int = DEFAULT_API_TIMEOUT
      api_retries: int = DEFAULT_API_RETRIES
      api_retry_backoff_factor: int = DEFAULT_RETRY_BACKOFF_FACTOR
  
  
  @dataclasses.dataclass
  class ProvisionArtemisData(ArtemisGuestData, tmt.steps.provision.ProvisionStepData):
      pass
  
  
  GUEST_STATE_COLOR_DEFAULT = 'green'
  
  GUEST_STATE_COLORS = {
      'routing': 'yellow',
      'provisioning': 'magenta',
      'promised': 'blue',
      'preparing': 'cyan',
      'cancelled': 'red',
      'error': 'red'
      }
  
  
  # Type annotation for Artemis API `GET /guests/$guestname` response.
  # Partial, not all fields necessary since plugin ignores most of them.
  GuestInspectType = TypedDict(
      'GuestInspectType', {
          'state': str,
          'address': Optional[str]
          }
      )
  
  
  class ArtemisAPI:
      def __init__(self, guest: 'GuestArtemis') -> None:
          self._guest = guest
  
          self.http_session = retry_session.create(
              retries=guest.api_retries,
              backoff_factor=guest.api_retry_backoff_factor,
              allowed_methods=('HEAD', 'GET', 'POST', 'DELETE', 'PUT'),
              status_forcelist=(
                  429,  # Too Many Requests
                  500,  # Internal Server Error
                  502,  # Bad Gateway
                  503,  # Service Unavailable
                  504   # Gateway Timeout
                  ),
              timeout=guest.api_timeout
              )
  
      def query(
              self,
              path: str,
              method: str = 'get',
              request_kwargs: Optional[Dict[str, Any]] = None
              ) -> requests.Response:
          """
          Base helper for Artemis API queries.
  
          Trivial dispatcher per method, returning retrieved response.
  
          :param path: API path to contact.
          :param method: HTTP method to use.
          :param request_kwargs: optional request options, as supported by
              :py:mod:`requests` library.
          """
  
          request_kwargs = request_kwargs or {}
  
          url = f'{self._guest.api_url}{path}'
  
          if method == 'get':
              return self.http_session.get(url, **request_kwargs)
  
          if method == 'post':
              return self.http_session.post(url, **request_kwargs)
  
          if method == 'delete':
              return self.http_session.delete(url, **request_kwargs)
  
          if method == 'put':
              return self.http_session.put(url, **request_kwargs)
  
          raise tmt.utils.GeneralError(
              f'Unsupported Artemis API method.\n{method}')
  
      def create(
              self,
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/artemis.py
files: 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]:
          """ Prepare command line options for connect """
          return [
              click.option(
                  '-g', '--guest', metavar='GUEST',
                  help='Select remote host to connect to (hostname or ip).'),
              click.option(
                  '-P', '--port', metavar='PORT',
                  help='Use specific port to connect to.'),
              click.option(
                  '-k', '--key', metavar='PRIVATE_KEY',
                  help='Private key for login into the guest system.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-p', '--password', metavar='PASSWORD',
                  help='Password for login into the guest system.'),
              ] + super().options(how)
  
      def go(self) -> None:
          """ Prepare the connection """
          super().go()
  
          # Check connection details
          guest = self.get('guest')
          user = self.get('user')
          key = self.get('key')
          password = self.get('password')
          port = self.get('port')
  
          # Check guest and auth info
          if not guest:
              raise tmt.utils.SpecificationError(
                  'Provide a host name or an ip address to connect.')
          data = ConnectGuestData(
              role=self.get('role'),
              guest=guest,
              user=user
              )
          self.info('guest', guest, 'green')
          self.info('user', user, 'green')
          if port:
              self.info('port', port, 'green')
              data.port = port
  
          # Use provided password for authentication
          if password:
              self.info('password', password, 'green')
              self.debug('Using password authentication.')
              data.password = password
          # Default to using a private key (user can have configured one)
          else:
              self.info('key', key or 'not provided', 'green')
              self.debug('Using private key authentication.')
              # TODO: something to fix in the future, multiple --key options would help.
              # Right now, the default value is List[str], while the option is just str.
              if isinstance(key, list):
                  data.key = key
              else:
                  data.key = [key]
  
          # FIXME: cast() - typeless "dispatcher" method
          data.ssh_option = cast(List[str], self.get('ssh-option'))
          if data.ssh_option:
              self.info('ssh options', fmf.utils.listed(data.ssh_option), 'green')
  
          # And finally create the guest
          self._guest = tmt.GuestSsh(data=data, name=self.name, parent=self.step)
  
      def guest(self) -> Optional[tmt.GuestSsh]:
          """ Return the provisioned guest """
          return self._guest
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/connect.py
files: 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) -> List[tmt.options.ClickOptionDecoratorType]:
          """ Prepare command line options for connect """
          return [
              click.option(
                  '-g', '--guest', metavar='GUEST',
                  help='Select remote host to connect to (hostname or ip).'),
              click.option(
                  '-P', '--port', metavar='PORT',
                  help='Use specific port to connect to.'),
              click.option(
                  '-k', '--key', metavar='PRIVATE_KEY',
                  help='Private key for login into the guest system.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='Username to use for all guest operations.'),
              click.option(
                  '-p', '--password', metavar='PASSWORD',
                  help='Password for login into the guest system.'),
              ] + super().options(how)
  
      # FIXME: ignore - https://github.com/teemtee/tmt/issues/1437
      def wake(self, data: Optional[GuestSshData] = None) -> None:  # type: ignore[override]
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
          if data:
              self._guest = tmt.GuestSsh(data=data, name=self.name, parent=self.step)
  
      def go(self) -> None:
          """ Prepare the connection """
          super().go()
  
          # Check connection details
          guest = self.get('guest')
          user = self.get('user')
          key = self.get('key')
          password = self.get('password')
          port = self.get('port')
  
          # Check guest and auth info
          if not guest:
              raise tmt.utils.SpecificationError(
                  'Provide a host name or an ip address to connect.')
          data = ConnectGuestData(
              role=self.get('role'),
              guest=guest,
              user=user
              )
          self.info('guest', guest, 'green')
          self.info('user', user, 'green')
          if port:
              self.info('port', port, 'green')
              data.port = port
  
          # Use provided password for authentication
          if password:
              self.info('password', password, 'green')
              self.debug('Using password authentication.')
              data.password = password
          # Default to using a private key (user can have configured one)
          else:
              self.info('key', key or 'not provided', 'green')
              self.debug('Using private key authentication.')
              # TODO: something to fix in the future, multiple --key options would help.
              # Right now, the default value is List[str], while the option is just str.
              if isinstance(key, list):
                  data.key = key
              else:
                  data.key = [key]
  
          # FIXME: cast() - typeless "dispatcher" method
          data.ssh_option = cast(List[str], self.get('ssh-option'))
          if data.ssh_option:
              self.info('ssh options', fmf.utils.listed(data.ssh_option), 'green')
  
          # And finally create the guest
          self._guest = tmt.GuestSsh(data=data, name=self.name, parent=self.step)
  
      def guest(self) -> Optional[tmt.GuestSsh]:
          """ Return the provisioned guest """
          return self._guest
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/connect.py
files: 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) -> tmt.utils.CommandOutput:
          """ Execute command on localhost """
          # Prepare the environment (plan/cli variables override)
          environment: tmt.utils.EnvironmentType = dict()
          # Beware of using dict() as default: `env` is `Optional[EnvironmentType]`, and may
          # be set by a caller by perfectly acceptable `None`. In such a case, `pop()` with
          # a default would return not the default, but existing key, i.e. `None`.
          environment.update(kwargs.pop('env', None) or {})
          environment.update(self.parent.plan.environment)
          if friendly_command is None:
              if isinstance(command, (list, tuple)):
                  friendly_command = ' '.join(shlex.quote(s) for s in command)
              else:
                  friendly_command = command
          # Run the command under the prepared environment
          return self.run(command,
                          env=environment,
                          shell=True,
                          log=log if log else self._command_verbose_logger,
                          friendly_command=friendly_command,
                          silent=silent,
                          **kwargs)
  
      def stop(self) -> None:
          """ Stop the guest """
  
          self.debug(f"Doing nothing to stop guest '{self.guest}'.")
  
      def reboot(self,
                 hard: bool = False,
                 command: Optional[str] = None,
                 timeout: Optional[int] = None) -> bool:
          """ Reboot the guest, return True if successful """
  
          self.debug(f"Doing nothing to reboot guest '{self.guest}'.")
  
          return False
  
      def push(
              self,
              source: Optional[str] = None,
              destination: Optional[str] = None,
              options: Optional[List[str]] = None) -> None:
          """ Nothing to be done to push workdir """
  
      def pull(
              self,
              source: Optional[str] = None,
              destination: Optional[str] = None,
              options: Optional[List[str]] = None,
              extend_options: Optional[List[str]] = None) -> None:
          """ Nothing to be done to pull workdir """
  
  
  @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.
      """
  
      _data_class = ProvisionLocalData
      _guest_class = GuestLocal
  
      # Guest instance
      _guest = None
  
      def go(self) -> None:
          """ Provision the container """
          super().go()
  
          # Create a GuestLocal instance
          data = tmt.steps.provision.GuestData(
              guest='localhost',
              role=self.get('role')
              )
          self._guest = GuestLocal(data=data, name=self.name, parent=self.step)
  
      def guest(self) -> Optional[GuestLocal]:
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self) -> List[str]:
          """ List of required packages needed for workdir sync """
          return GuestLocal.requires()
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/local.py
files: 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) -> tmt.utils.CommandOutput:
          """ Execute command on localhost """
          # Prepare the environment (plan/cli variables override)
          environment: tmt.utils.EnvironmentType = dict()
          # Beware of using dict() as default: `env` is `Optional[EnvironmentType]`, and may
          # be set by a caller by perfectly acceptable `None`. In such a case, `pop()` with
          # a default would return not the default, but existing key, i.e. `None`.
          environment.update(kwargs.pop('env', None) or {})
          environment.update(self.parent.plan.environment)
          if friendly_command is None:
              if isinstance(command, (list, tuple)):
                  friendly_command = ' '.join(shlex.quote(s) for s in command)
              else:
                  friendly_command = command
          # Run the command under the prepared environment
          return self.run(command,
                          env=environment,
                          shell=True,
                          log=log if log else self._command_verbose_logger,
                          friendly_command=friendly_command,
                          silent=silent,
                          **kwargs)
  
      def stop(self) -> None:
          """ Stop the guest """
  
          self.debug(f"Doing nothing to stop guest '{self.guest}'.")
  
      def reboot(self,
                 hard: bool = False,
                 command: Optional[str] = None,
                 timeout: Optional[int] = None) -> bool:
          """ Reboot the guest, return True if successful """
  
          self.debug(f"Doing nothing to reboot guest '{self.guest}'.")
  
          return False
  
      def push(
              self,
              source: Optional[str] = None,
              destination: Optional[str] = None,
              options: Optional[List[str]] = None) -> None:
          """ Nothing to be done to push workdir """
  
      def pull(
              self,
              source: Optional[str] = None,
              destination: Optional[str] = None,
              options: Optional[List[str]] = None,
              extend_options: Optional[List[str]] = None) -> None:
          """ Nothing to be done to pull workdir """
  
  
  @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.
      """
  
      _data_class = ProvisionLocalData
      _guest_class = GuestLocal
  
      # Guest instance
      _guest = None
  
      def wake(self, data: Optional[tmt.steps.provision.GuestData] = None) -> None:
          """ Wake up the plugin, process data, apply options """
          super().wake(data=data)
          if data:
              self._guest = GuestLocal(data=data, name=self.name, parent=self.step)
  
      def go(self) -> None:
          """ Provision the container """
          super().go()
  
          # Create a GuestLocal instance
          data = tmt.steps.provision.GuestData(
              guest='localhost',
              role=self.get('role')
              )
          self._guest = GuestLocal(data=data, name=self.name, parent=self.step)
  
      def guest(self) -> Optional[GuestLocal]:
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self) -> List[str]:
          """ List of required packages needed for workdir sync """
          return GuestLocal.requires()
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/local.py
files: 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_stderr = cmd_output
          return str(cmd_stdout).strip() == 'true'
  
      def wake(self) -> None:
          """ Wake up the guest """
          self.debug(
              f"Waking up container '{self.container}'.", level=2, shift=0)
  
      def start(self) -> None:
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Check if the image is available
          assert self.image is not None
          command = ['image', 'exists', self.image]
          try:
              self.podman(command, message=f"Check for container image '{self.image}'.")
              needs_pull = False
          except tmt.utils.RunError:
              needs_pull = True
  
          # Pull image if not available or pull forced
          if needs_pull or self.force_pull:
              self.podman(
                  ['pull', '-q', self.image],
                  message=f"Pull image '{self.image}'.")
  
          # Mount the whole plan directory in the container
          workdir = self.parent.plan.workdir
  
          self.container = self.guest = self._tmt_name()
          self.verbose('name', self.container, 'green')
  
          # FIXME: Workaround for BZ#1900021 (f34 container on centos-8)
          workaround = ['--security-opt', 'seccomp=unconfined']
  
          # Run the container
          self.debug(f"Start container '{self.image}'.")
          assert self.container is not None
          self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f'{workdir}:{workdir}:z',
               '-itd', '--user', self.user, self.image])
  
      def reboot(self, hard: bool = False,
                 command: Optional[Union[str, List[str], Tuple[str, ...]]] = None,
                 timeout: Optional[int] = None) -> bool:
          """ Restart the container, return True if successful  """
          if command:
              raise tmt.utils.ProvisionError(
                  "Custom reboot command not supported in podman provision.")
          if not hard:
              raise tmt.utils.ProvisionError(
                  "Containers do not support soft reboot, they can only be "
                  "stopped and started again (hard reboot).")
          assert self.container is not None
          self.podman(['container', 'restart', self.container])
          return self.reconnect(timeout=timeout or CONNECTION_TIMEOUT)
  
      def ansible(self, playbook: str, extra_args: Optional[str] = None) -> None:
          """ Prepare container using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          # As non-root we must run with podman unshare
          podman_unshare = ['podman', 'unshare'] if os.geteuid() != 0 else []
          stdout, _ = self.run(
              podman_unshare + ['ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'podman', '-i', f'{self.container},', playbook],
              cwd=self.parent.plan.worktree,
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def podman(self, command: List[str], **kwargs: Any) -> tmt.utils.CommandOutput:
          """ Run given command via podman """
          return self.run(['podman'] + command, **kwargs)
  
      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) -> tmt.utils.CommandOutput:
          """ Execute given commands in podman via shell """
          if not self.container and not self.opt('dry'):
              raise tmt.utils.ProvisionError(
                  'Could not execute without provisioned container.')
  
          # Change to given directory on guest if cwd provided
          directory = kwargs.get('cwd', '')
          if directory:
              directory = f"cd {quote(directory)}; "
  
          # Prepare the environment variables export
          environment = self._export_environment(
              self._prepare_environment(kwargs.get('env', dict())))
  
          # Run in interactive mode if requested
          interactive = ['-it'] if kwargs.get('interactive') else []
  
          # Note that we MUST run commands via bash, so variables
          #
  
  [... the rest is omitted ...]
files: tmt/steps/provision/podman.py
files: 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_stderr = cmd_output
          return str(cmd_stdout).strip() == 'true'
  
      def wake(self) -> None:
          """ Wake up the guest """
          self.debug(
              f"Waking up container '{self.container}'.", level=2, shift=0)
  
      def start(self) -> None:
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Check if the image is available
          assert self.image is not None
          command = ['image', 'exists', self.image]
          try:
              self.podman(command, message=f"Check for container image '{self.image}'.")
              needs_pull = False
          except tmt.utils.RunError:
              needs_pull = True
  
          # Pull image if not available or pull forced
          if needs_pull or self.force_pull:
              self.podman(
                  ['pull', '-q', self.image],
                  message=f"Pull image '{self.image}'.")
  
          # Mount the whole plan directory in the container
          workdir = self.parent.plan.workdir
  
          self.container = self.guest = self._tmt_name()
          self.verbose('name', self.container, 'green')
  
          # FIXME: Workaround for BZ#1900021 (f34 container on centos-8)
          workaround = ['--security-opt', 'seccomp=unconfined']
  
          # Run the container
          self.debug(f"Start container '{self.image}'.")
          assert self.container is not None
          self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f'{workdir}:{workdir}:z',
               '-itd', '--user', self.user, self.image])
  
      def reboot(self, hard: bool = False,
                 command: Optional[Union[str, List[str], Tuple[str, ...]]] = None,
                 timeout: Optional[int] = None) -> bool:
          """ Restart the container, return True if successful  """
          if command:
              raise tmt.utils.ProvisionError(
                  "Custom reboot command not supported in podman provision.")
          if not hard:
              raise tmt.utils.ProvisionError(
                  "Containers do not support soft reboot, they can only be "
                  "stopped and started again (hard reboot).")
          assert self.container is not None
          self.podman(['container', 'restart', self.container])
          return self.reconnect(timeout=timeout or CONNECTION_TIMEOUT)
  
      def ansible(self, playbook: str, extra_args: Optional[str] = None) -> None:
          """ Prepare container using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          # As non-root we must run with podman unshare
          podman_unshare = ['podman', 'unshare'] if os.geteuid() != 0 else []
          stdout, _ = self.run(
              podman_unshare + ['ansible-playbook'] +
              self._ansible_verbosity() +
              self._ansible_extra_args(extra_args) +
              ['-c', 'podman', '-i', f'{self.container},', playbook],
              cwd=self.parent.plan.worktree,
              env=self._prepare_environment())
          self._ansible_summary(stdout)
  
      def podman(self, command: List[str], **kwargs: Any) -> tmt.utils.CommandOutput:
          """ Run given command via podman """
          return self.run(['podman'] + command, **kwargs)
  
      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) -> tmt.utils.CommandOutput:
          """ Execute given commands in podman via shell """
          if not self.container and not self.opt('dry'):
              raise tmt.utils.ProvisionError(
                  'Could not execute without provisioned container.')
  
          # Change to given directory on guest if cwd provided
          directory = kwargs.get('cwd', '')
          if directory:
              directory = f"cd {quote(directory)}; "
  
          # Prepare the environment variables export
          environment = self._export_environment(
              self._prepare_environment(kwargs.get('env', dict())))
  
          # Run in interactive mode if requested
          interactive = ['-it'] if kwargs.get('interactive') else []
  
          # Note that we MUST run commands via bash, so variables
          #
  
  [... 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
  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:
    - {public_key}
  ssh_pwauth: true
  disable_root: false
  runcmd:
    - sed -i -e '/^.*PermitRootLogin/s/^.*$/PermitRootLogin yes/'
      -e '/^.*UseDNS/s/^.*$/UseDNS no/'
      -e '/^.*GSSAPIAuthentication/s/^.*$/GSSAPIAuthentication no/'
      /etc/ssh/sshd_config
    - systemctl reload sshd
    - [sh, -c, 'if [ ! -f /etc/systemd/network/20-tc-usernet.network ] &&
    systemctl status systemd-networkd | grep -q "enabled;\\svendor\\spreset:\\senabled";
    then mkdir -p /etc/systemd/network/ &&
    echo "[Match]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "Name=en*" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "[Network]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "DHCP=yes" >> /etc/systemd/network/20-tc-usernet.network; fi']
    - [sh, -c, 'if systemctl status systemd-networkd |
    grep -q "enabled;\\svendor\\spreset:\\senabled"; then
    systemctl restart systemd-networkd; fi']
    - [sh, -c, 'if cat /etc/os-release |
    grep -q platform:el8; then systemctl restart sshd; fi']
  """
  
  COREOS_DATA = """variant: fcos
  version: 1.4.0
  passwd:
    users:
      - name: {user_name}
        ssh_authorized_keys:
          - {public_key}
  """
  
  # Libvirt domain XML template related variables
  DOMAIN_TEMPLATE_NAME = 'domain-template.jinja'
  DOMAIN_TEMPLATE_FILE = os.path.join(TESTCLOUD_DATA, DOMAIN_TEMPLATE_NAME)
  DOMAIN_TEMPLATE = """<domain type='{{ virt_type }}' xmlns:qemu='http://libvirt.org/schemas/domain/qemu/1.0'>
    <name>{{ domain_name }}</name>
    <uuid>{{ uuid }}</uuid>
    <memory unit='KiB'>{{ memory }}</memory>
    <currentMemory unit='KiB'>{{ memory }}</currentMemory>
    <vcpu placement='static'>2</vcpu>
    <os>
      <type arch='{{ arch }}' machine='{{ model }}'>hvm</type>
      {{ uefi_loader }}
      <boot dev='hd'/>
    </os>
    {{ cpu }}
    {{ extra_specs }}
    <clock offset='utc'>
      <timer name='rtc' tickpolicy='catchup'/>
      <timer name='pit' tickpolicy='delay'/>
      <timer name='hpet' present='no'/>
    </clock>
    <on_poweroff>destroy</on_poweroff>
    <on_reboot>restart</on_reboot>
    <on_crash>restart</on_crash>
    <devices>
      <emulator>{{ emulator_path }}</emulator>
      <disk type='file' device='disk'>
        <driver name='qemu' type='qcow2' cache='unsafe'/>
        <source file="{{ disk }}"/>
        <target dev='vda' bus='virtio'/>
        {{ boot_drive_address }}
      </disk>
      <disk type='file' device='disk'>
        <driver name='qemu' type='raw'/>
        <source file="{{ seed }}"/>
        <target dev='vdb' bus='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x08' function='0x0'/>
      </disk>
      <interface type='{{ network_type }}'>
        <mac address="{{ mac_address }}"/>
        {{ network_source }}
        {{ ip_setup }}
        <model type='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>
      </interface>
      <serial type='pty'>
        <target port='0'/>
      </serial>
      <console type='pty'>
        <target type='serial' port='0'/>
      </console>
      <input type="keyboard" bus="virtio"/>
      <rng model='virtio'>
  
  
  [... 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
  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:
    - {public_key}
  ssh_pwauth: true
  disable_root: false
  runcmd:
    - sed -i -e '/^.*PermitRootLogin/s/^.*$/PermitRootLogin yes/'
      -e '/^.*UseDNS/s/^.*$/UseDNS no/'
      -e '/^.*GSSAPIAuthentication/s/^.*$/GSSAPIAuthentication no/'
      /etc/ssh/sshd_config
    - systemctl reload sshd
    - [sh, -c, 'if [ ! -f /etc/systemd/network/20-tc-usernet.network ] &&
    systemctl status systemd-networkd | grep -q "enabled;\\svendor\\spreset:\\senabled";
    then mkdir -p /etc/systemd/network/ &&
    echo "[Match]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "Name=en*" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "[Network]" >> /etc/systemd/network/20-tc-usernet.network &&
    echo "DHCP=yes" >> /etc/systemd/network/20-tc-usernet.network; fi']
    - [sh, -c, 'if systemctl status systemd-networkd |
    grep -q "enabled;\\svendor\\spreset:\\senabled"; then
    systemctl restart systemd-networkd; fi']
    - [sh, -c, 'if cat /etc/os-release |
    grep -q platform:el8; then systemctl restart sshd; fi']
  """
  
  COREOS_DATA = """variant: fcos
  version: 1.4.0
  passwd:
    users:
      - name: {user_name}
        ssh_authorized_keys:
          - {public_key}
  """
  
  # Libvirt domain XML template related variables
  DOMAIN_TEMPLATE_NAME = 'domain-template.jinja'
  DOMAIN_TEMPLATE_FILE = os.path.join(TESTCLOUD_DATA, DOMAIN_TEMPLATE_NAME)
  DOMAIN_TEMPLATE = """<domain type='{{ virt_type }}' xmlns:qemu='http://libvirt.org/schemas/domain/qemu/1.0'>
    <name>{{ domain_name }}</name>
    <uuid>{{ uuid }}</uuid>
    <memory unit='KiB'>{{ memory }}</memory>
    <currentMemory unit='KiB'>{{ memory }}</currentMemory>
    <vcpu placement='static'>2</vcpu>
    <os>
      <type arch='{{ arch }}' machine='{{ model }}'>hvm</type>
      {{ uefi_loader }}
      <boot dev='hd'/>
    </os>
    {{ cpu }}
    {{ extra_specs }}
    <clock offset='utc'>
      <timer name='rtc' tickpolicy='catchup'/>
      <timer name='pit' tickpolicy='delay'/>
      <timer name='hpet' present='no'/>
    </clock>
    <on_poweroff>destroy</on_poweroff>
    <on_reboot>restart</on_reboot>
    <on_crash>restart</on_crash>
    <devices>
      <emulator>{{ emulator_path }}</emulator>
      <disk type='file' device='disk'>
        <driver name='qemu' type='qcow2' cache='unsafe'/>
        <source file="{{ disk }}"/>
        <target dev='vda' bus='virtio'/>
        {{ boot_drive_address }}
      </disk>
      <disk type='file' device='disk'>
        <driver name='qemu' type='raw'/>
        <source file="{{ seed }}"/>
        <target dev='vdb' bus='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x08' function='0x0'/>
      </disk>
      <interface type='{{ network_type }}'>
        <mac address="{{ mac_address }}"/>
        {{ network_source }}
        {{ ip_setup }}
        <model type='virtio'/>
        <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>
      </interface>
      <serial type='pty'>
        <target port='0'/>
      </serial>
      <console type='pty'>
        <target type='serial' port='0'/>
      </console>
      <input type="keyboard" bus="virtio"/>
      <rng model='virtio'>
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/testcloud.py