files: #!/bin/bash
  # vim: dict+=/usr/share/beakerlib/dictionary.vim cpt=.,w,b,u,t,i,k
  . /usr/share/beakerlib/beakerlib.sh || exit 1
  
  rlJournalStart
      rlPhaseStartSetup
          rlRun "run=\$(mktemp -d)" 0 "Create run directory"
          rlRun "pushd data"
          rlRun "set -o pipefail"
      rlPhaseEnd
  
      opt="--scratch -i $run"
      rlPhaseStartTest "Wrong provision"
          rlRun -s "tmt run $opt plan -n wrong provision finish" \
              2 "Names not unique"
          rlAssertGrep "must be unique" $rlRun_LOG
      rlPhaseEnd
  
      rlPhaseStartTest "Correct provision without roles"
          rlRun -s "tmt run $opt plan -n noroles provision finish" 0
          rlAssertGrep "2 guests provisioned" $rlRun_LOG
          # how: container should be there twice
          rlRun "grep 'how: container' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "2" lines
  
          # 2 guests without role are saved
          guests="$run/noroles/provision/guests.yaml"
          rlAssertNotGrep "role: [a-z]" $guests "File '$guests' should leave role unspecified"
          rlAssertGrep "client" $guests
          rlAssertGrep "server" $guests
  
          # 2 guests were removed
          rlRun "grep 'container: removed' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "2" lines
  
          rlRun "rm lines"
      rlPhaseEnd
  
      rlPhaseStartTest "Correct provision with roles"
          rlRun -s "tmt run $opt plan -n /roles provision finish" 0
          rlAssertGrep "4 guests provisioned" $rlRun_LOG
          # how: container should be there 4 times
          rlRun "grep 'how: container' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "4" lines
  
          guests="$run/roles/provision/guests.yaml"
          rlAssertGrep "client-one" $guests
          rlAssertGrep "client-two" $guests
          rlAssertGrep "server-one" $guests
          rlAssertGrep "server-two" $guests
  
          # Each guest has a role
          rlRun "grep 'role: ' $guests | wc -l > lines"
          rlAssertGrep "4" "lines"
  
          # 4 guests were removed
          rlRun "grep 'container: removed' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "4" lines
  
          rlRun "rm lines"
      rlPhaseEnd
  
      rlPhaseStartTest "Full plan without roles"
          rlRun -s "tmt run $opt plan -n noroles"
      rlPhaseEnd
  
      rlPhaseStartCleanup
          rlRun "popd"
          rlRun "rm -r $run" 0 "Remove run directory"
      rlPhaseEnd
  rlJournalEnd
  
  
  [... the rest is omitted ...]
files: tests/multihost/provision/test.sh
files: #!/bin/bash
  # vim: dict+=/usr/share/beakerlib/dictionary.vim cpt=.,w,b,u,t,i,k
  . /usr/share/beakerlib/beakerlib.sh || exit 1
  
  rlJournalStart
      rlPhaseStartSetup
          rlRun "run=\$(mktemp -d)" 0 "Create run directory"
          rlRun "pushd data"
          rlRun "set -o pipefail"
      rlPhaseEnd
  
      opt="--scratch -i $run"
      rlPhaseStartTest "Wrong provision"
          rlRun -s "tmt run $opt plan -n wrong provision finish" \
              2 "Names not unique"
          rlAssertGrep "must be unique" $rlRun_LOG
      rlPhaseEnd
  
      rlPhaseStartTest "Correct provision without roles"
          rlRun -s "tmt run $opt plan -n noroles provision finish" 0
          rlAssertGrep "2 guests provisioned" $rlRun_LOG
          # how: container should be there twice
          rlRun "grep 'how: container' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "2" lines
  
          # 2 guests without role are saved
          guests="$run/noroles/provision/guests.yaml"
          rlAssertNotGrep "role" $guests
          rlAssertGrep "client" $guests
          rlAssertGrep "server" $guests
  
          # 2 guests were removed
          rlRun "grep 'container: removed' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "2" lines
  
          rlRun "rm lines"
      rlPhaseEnd
  
      rlPhaseStartTest "Correct provision with roles"
          rlRun -s "tmt run $opt plan -n /roles provision finish" 0
          rlAssertGrep "4 guests provisioned" $rlRun_LOG
          # how: container should be there 4 times
          rlRun "grep 'how: container' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "4" lines
  
          guests="$run/roles/provision/guests.yaml"
          rlAssertGrep "client-one" $guests
          rlAssertGrep "client-two" $guests
          rlAssertGrep "server-one" $guests
          rlAssertGrep "server-two" $guests
  
          # Each guest has a role
          rlRun "grep 'role: ' $guests | wc -l > lines"
          rlAssertGrep "4" "lines"
  
          # 4 guests were removed
          rlRun "grep 'container: removed' $rlRun_LOG | wc -l > lines"
          rlAssertGrep "4" lines
  
          rlRun "rm lines"
      rlPhaseEnd
  
      rlPhaseStartTest "Full plan without roles"
          rlRun -s "tmt run $opt plan -n noroles"
      rlPhaseEnd
  
      rlPhaseStartCleanup
          rlRun "popd"
          rlRun "rm -r $run" 0 "Remove run directory"
      rlPhaseEnd
  rlJournalEnd
  
  
  [... the rest is omitted ...]
files: tests/multihost/provision/test.sh
files: import collections
  import dataclasses
  import datetime
  import os
  import random
  import re
  import shlex
  import string
  import subprocess
  import tempfile
  import time
  from typing import Dict, List, Optional, Type
  
  import click
  import fmf
  
  import tmt
  import tmt.plugins
  import tmt.utils
  
  # Timeout in seconds of waiting for a connection after reboot
  # (long enough to allow operations such as system upgrade)
  # FIXME: Shorten once "tmt-reboot -t <timeout>" is implemented
  CONNECTION_TIMEOUT = 60 * 60
  
  # Wait time when reboot happens in seconds
  RECONNECT_INITIAL_WAIT_TIME = 5
  
  # Default rsync options
  DEFAULT_RSYNC_OPTIONS = [
      "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  
  class Provision(tmt.steps.Step):
      """ Provision an environment for testing or use localhost. """
  
      # Default implementation for provision is a virtual machine
      how = 'virtual'
  
      def __init__(self, plan, data):
          """ Initialize provision step data """
          super().__init__(plan=plan, data=data)
          # Check that the names are unique
          names = [data.get('name') for data in self.data]
          names_count = collections.defaultdict(int)
          for name in names:
              names_count[name] += 1
          if len(self.data) > 1 and len(self.data) != len(names_count):
              duplicate = [k for k, v in names_count.items() if v > 1]
              duplicate_string = ', '.join(duplicate)
              raise tmt.utils.GeneralError(
                  f"Provision step names must be unique for multihost testing. "
                  f"Duplicate names: {duplicate_string} in plan '{plan.name}'.")
          # List of provisioned guests and loaded guest data
          self._guests: List['Guest'] = []
          self._guest_data: Dict[str, 'GuestData'] = {}
          self.is_multihost = False
  
      def load(self, extra_keys=None):
          """ Load guest data from the workdir """
          extra_keys = extra_keys or []
          super().load(extra_keys)
          try:
              raw_guest_data = tmt.utils.yaml_to_dict(self.read('guests.yaml'))
  
              self._guest_data = {
                  name: tmt.utils.SerializableContainer.unserialize(guest_data)
                  for name, guest_data in raw_guest_data.items()
                  }
  
          except tmt.utils.FileError:
              self.debug('Provisioned guests not found.', level=2)
  
      def save(self, data=None):
          """ Save guest data to the workdir """
          data = data or {}
          super().save(data)
          try:
              raw_guest_data = {guest.name: guest.save().to_serialized()
                                for guest in self.guests()}
  
              self.write('guests.yaml', tmt.utils.dict_to_yaml(raw_guest_data))
          except tmt.utils.FileError:
              self.debug('Failed to save provisioned guests.')
  
      def wake(self):
          """ Wake up the step (process workdir and command line) """
          super().wake()
  
          # Choose the right plugin and wake it up
          for data in self.data:
              plugin = ProvisionPlugin.delegate(self, data)
              self._phases.append(plugin)
              # If guest data loaded, perform a complete wake up
              plugin.wake(data=self._guest_data.get(plugin.name))
              if plugin.guest():
                  self._guests.append(plugin.guest())
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Provision wake up complete (already done before).', level=2)
          # Save status and step data (now we know what to do)
          else:
              self.status('todo')
              self.save()
  
      def show(self):
          """ Show discover details """
          for data in self.data:
              ProvisionPlugin.delegate(self, data).show()
  
      def summary(self):
          """ Give a concise summary of the provisioning """
          # Summary of provisioned guests
          guests = fmf.utils.listed(self.guests(), 'guest')
          self.info('summary', f'{guests} provisioned', 'green', shift=1)
          # Guest list in verbose mode
          for guest in self.guests():
              if guest.name != tmt.utils.DEFAULT_NAME:
                  self.verbose(guest.name, color='red', shift=2)
  
      def go(self):
          """ Provision all guests"""
          super().go()
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.info('status', 'done', 'green', shift=1)
              self.summary()
              self.actions()
              return
  
          # Provision guests
          self._guests = []
          save = True
          self.is_multihost = sum([isinstance(phase, ProvisionPlugin)
                                  for phase in self.phases()]) > 1
          try:
              for phase in self.phases():
                  try:
                      phase.go()
                      if isinstance(phase, ProvisionPlugin):
                          phase.guest().details()
                      if self.is_multihost:
                          self.info('')
                 
  
  [... the rest is omitted ...]
files: tmt/steps/provision/__init__.py
files: import collections
  import datetime
  import os
  import random
  import re
  import shlex
  import string
  import subprocess
  import tempfile
  import time
  from typing import Optional, Type
  
  import click
  import fmf
  
  import tmt
  import tmt.utils
  
  # Timeout in seconds of waiting for a connection after reboot
  # (long enough to allow operations such as system upgrade)
  # FIXME: Shorten once "tmt-reboot -t <timeout>" is implemented
  CONNECTION_TIMEOUT = 60 * 60
  
  # Wait time when reboot happens in seconds
  RECONNECT_INITIAL_WAIT_TIME = 5
  
  # Default rsync options
  DEFAULT_RSYNC_OPTIONS = [
      "-R", "-r", "-z", "--links", "--safe-links", "--delete"]
  
  
  class Provision(tmt.steps.Step):
      """ Provision an environment for testing or use localhost. """
  
      # Default implementation for provision is a virtual machine
      how = 'virtual'
  
      def __init__(self, plan, data):
          """ Initialize provision step data """
          super().__init__(plan=plan, data=data)
          # Check that the names are unique
          names = [data.get('name') for data in self.data]
          names_count = collections.defaultdict(int)
          for name in names:
              names_count[name] += 1
          if len(self.data) > 1 and len(self.data) != len(names_count):
              duplicate = [k for k, v in names_count.items() if v > 1]
              duplicate_string = ', '.join(duplicate)
              raise tmt.utils.GeneralError(
                  f"Provision step names must be unique for multihost testing. "
                  f"Duplicate names: {duplicate_string} in plan '{plan.name}'.")
          # List of provisioned guests and loaded guest data
          self._guests = []
          self._guest_data = {}
          self.is_multihost = False
  
      def load(self, extra_keys=None):
          """ Load guest data from the workdir """
          extra_keys = extra_keys or []
          super().load(extra_keys)
          try:
              self._guest_data = tmt.utils.yaml_to_dict(self.read('guests.yaml'))
          except tmt.utils.FileError:
              self.debug('Provisioned guests not found.', level=2)
  
      def save(self, data=None):
          """ Save guest data to the workdir """
          data = data or {}
          super().save(data)
          try:
              guests = dict(
                  [(guest.name, guest.save()) for guest in self.guests()])
              self.write('guests.yaml', tmt.utils.dict_to_yaml(guests))
          except tmt.utils.FileError:
              self.debug('Failed to save provisioned guests.')
  
      def wake(self):
          """ Wake up the step (process workdir and command line) """
          super().wake()
  
          # Choose the right plugin and wake it up
          for data in self.data:
              plugin = ProvisionPlugin.delegate(self, data)
              self._phases.append(plugin)
              # If guest data loaded, perform a complete wake up
              plugin.wake(data=self._guest_data.get(plugin.name))
              if plugin.guest():
                  self._guests.append(plugin.guest())
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.debug(
                  'Provision wake up complete (already done before).', level=2)
          # Save status and step data (now we know what to do)
          else:
              self.status('todo')
              self.save()
  
      def show(self):
          """ Show discover details """
          for data in self.data:
              ProvisionPlugin.delegate(self, data).show()
  
      def summary(self):
          """ Give a concise summary of the provisioning """
          # Summary of provisioned guests
          guests = fmf.utils.listed(self.guests(), 'guest')
          self.info('summary', f'{guests} provisioned', 'green', shift=1)
          # Guest list in verbose mode
          for guest in self.guests():
              if guest.name != tmt.utils.DEFAULT_NAME:
                  self.verbose(guest.name, color='red', shift=2)
  
      def go(self):
          """ Provision all guests"""
          super().go()
  
          # Nothing more to do if already done
          if self.status() == 'done':
              self.info('status', 'done', 'green', shift=1)
              self.summary()
              self.actions()
              return
  
          # Provision guests
          self._guests = []
          save = True
          self.is_multihost = sum([isinstance(phase, ProvisionPlugin)
                                  for phase in self.phases()]) > 1
          try:
              for phase in self.phases():
                  try:
                      phase.go()
                      if isinstance(phase, ProvisionPlugin):
                          phase.guest().details()
                      if self.is_multihost:
                          self.info('')
                  finally:
                      if isinstance(phase, ProvisionPlugin):
                          if phase.guest():
                              self._guests.append(phase.guest())
  
              # Give a summary, update status and save
              self.summary()
              self.status('done')
          except (SystemExit, tmt.utils.SpecificationError) as error:
              # A plugin will only
  
  [... the rest is omitted ...]
files: tmt/steps/provision/__init__.py
files: import dataclasses
  import datetime
  import sys
  import time
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import requests
  
  import tmt
  import tmt.steps.provision
  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'
  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):  # type: ignore[misc]
      # 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
  
  
  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 ==
  
  [... the rest is omitted ...]
files: tmt/steps/provision/artemis.py
files: import datetime
  import sys
  import time
  from typing import Any, Dict, List, Optional, cast
  
  import click
  import requests
  
  import tmt
  import tmt.steps.provision
  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]
  
  # Type annotation for "data" package describing a guest instance. Passed
  # between load() and save() calls.
  StepStateType = TypedDict(
      'StepStateType',
      {
          # API
          'api-url': str,
          'api-version': str,
  
          # Guest request properties
          'arch': str,
          'image': str,
          'hardware': Any,
          'pool': Optional[str],
          'priority-group': str,
          'keyname': str,
          'user-data': Dict[str, str],
  
          # Provided by Artemis response
          'guestname': Optional[str],
          'guest': Optional[str],
          'user': str,
  
          # Timeouts and deadlines
          'provision-timeout': int,
          'provision-tick': int,
          'api-timeout': int,
          'api-retries': int,
          'api-retry-backoff-factor': int
          }
      )
  
  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
  
  DEFAULT_GUEST_DATA = cast(
      StepStateType, {
          'api-version': DEFAULT_API_VERSION,
          'arch': DEFAULT_ARCH,
          'priority-group': DEFAULT_PRIORITY_GROUP,
          'keyname': DEFAULT_KEYNAME,
          'user-data': {},
          'user': DEFAULT_USER,
          'provision-timeout': DEFAULT_PROVISION_TIMEOUT,
          'provision-tick': DEFAULT_PROVISION_TICK,
          'api-timeout': DEFAULT_API_TIMEOUT,
          'api-retries': DEFAULT_API_RETRIES,
          'api-retry-backoff-factor': DEFAULT_RETRY_BACKOFF_FACTOR
          }
      )
  
  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 =
  
  [... the rest is omitted ...]
files: tmt/steps/provision/artemis.py
files: import click
  
  import tmt
  import tmt.steps.provision
  
  
  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 methods
      _methods = [tmt.steps.Method(name='connect', doc=__doc__, order=50)]
  
      # Supported keys
      _keys = ["guest", "key", "user", "password", "port"]
  
      @classmethod
      def options(cls, how=None):
          """ 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 default(self, option, default=None):
          """ Return default data for given option """
          # User root as the default user
          if option == 'user':
              return 'root'
          # No other defaults available
          return default
  
      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 = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def go(self):
          """ 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 = tmt.steps.provision.GuestSshData(
              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.')
              data.key = key
  
          # And finally create the guest
          self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/connect.py
files: import click
  
  import tmt
  import tmt.steps.provision
  
  
  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 methods
      _methods = [tmt.steps.Method(name='connect', doc=__doc__, order=50)]
  
      # Supported keys
      _keys = ["guest", "key", "user", "password", "port"]
  
      @classmethod
      def options(cls, how=None):
          """ 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 default(self, option, default=None):
          """ Return default data for given option """
          # User root as the default user
          if option == 'user':
              return 'root'
          # No other defaults available
          return default
  
      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 = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def go(self):
          """ 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 = dict(guest=guest, user=user, role=self.get('role'))
          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.')
              data['key'] = key
  
          # And finally create the guest
          self._guest = tmt.GuestSsh(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/connect.py
files: import tmt
  import tmt.steps.provision
  
  
  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
  
      # Supported methods
      _methods = [tmt.steps.Method(name='local', doc=__doc__, order=50)]
  
      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 = GuestLocal(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestLocal.requires()
  
  
  class GuestLocal(tmt.Guest):
      """ Local Host """
  
      localhost = True
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, stderr = 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, **kwargs):
          """ Execute command on localhost """
          # Prepare the environment (plan/cli variables override)
          environment = dict()
          environment.update(kwargs.pop('env', dict()))
          environment.update(self.parent.plan.environment)
          # Run the command under the prepared environment
          return self.run(command, env=environment, shell=True, **kwargs)
  
      def stop(self):
          """ Stop the guest """
  
          self.debug(f"Doing nothing to stop guest '{self.guest}'.")
  
      def reboot(self, hard=False):
          """ Reboot the guest, return True if successful """
  
          self.debug(f"Doing nothing to reboot guest '{self.guest}'.")
  
          return False
  
      def push(self, source=None, destination=None, options=None):
          """ Nothing to be done to push workdir """
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """ Nothing to be done to pull workdir """
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/local.py
files: import tmt
  import tmt.steps.provision
  
  
  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
  
      # Supported methods
      _methods = [tmt.steps.Method(name='local', doc=__doc__, order=50)]
  
      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 = {'guest': 'localhost', 'role': self.get('role')}
          self._guest = GuestLocal(data, name=self.name, parent=self.step)
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestLocal.requires()
  
  
  class GuestLocal(tmt.Guest):
      """ Local Host """
  
      def __init__(self, data, name=None, parent=None):
          """ Initialize guest data """
          super().__init__(data, name, parent)
          self.localhost = True
  
      def ansible(self, playbook, extra_args=None):
          """ Prepare localhost using ansible playbook """
          playbook = self._ansible_playbook_path(playbook)
          stdout, stderr = 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, **kwargs):
          """ Execute command on localhost """
          # Prepare the environment (plan/cli variables override)
          environment = dict()
          environment.update(kwargs.pop('env', dict()))
          environment.update(self.parent.plan.environment)
          # Run the command under the prepared environment
          return self.run(command, env=environment, shell=True, **kwargs)
  
      def stop(self):
          """ Stop the guest """
  
          self.debug(f"Doing nothing to stop guest '{self.guest}'.")
  
      def reboot(self, hard=False):
          """ Reboot the guest, return True if successful """
  
          self.debug(f"Doing nothing to reboot guest '{self.guest}'.")
  
          return False
  
      def push(self, source=None, destination=None, options=None):
          """ Nothing to be done to push workdir """
  
      def pull(
              self,
              source=None,
              destination=None,
              options=None,
              extend_options=None):
          """ Nothing to be done to pull workdir """
  
  
  [... the rest is omitted ...]
files: tmt/steps/provision/local.py
files: import dataclasses
  import os
  from typing import Optional
  
  import click
  
  import tmt
  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
  
  
  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 = ["image", "container", "pull", "user"]
  
      # Supported methods
      _methods = [tmt.steps.Method(name='container', doc=__doc__, order=50)]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for connect """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to use. Short name or complete url.'),
              click.option(
                  '-c', '--container', metavar='NAME',
                  help='Name or id of an existing container to be used.'),
              click.option(
                  '-p', '--pull', is_flag=True,
                  help='Force pulling a fresh container image.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='User to use for all container operations.')
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          if option == 'pull':
              return PodmanGuestData().force_pull
  
          return getattr(PodmanGuestData(), option.replace('-', '_'), default)
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          # Wake up podman instance
          if data:
              guest = GuestContainer(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Show which image we are using
          pull = ' (force pull)' if self.get('pull') else ''
          self.info('image', f"{self.get('image')}{pull}", 'green')
  
          # Prepare data for the guest instance
          data_from_options = {
              key: self.get(key)
              for key in PodmanGuestData.keys()
              if key != 'force_pull'
              }
  
          data_from_options['force_pull'] = self.get('pull')
  
          data = PodmanGuestData(**data_from_options)
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestContainer(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestContainer.requires()
  
  
  class GuestContainer(tmt.Guest):
      """ Container Instance """
  
      _data_class = PodmanGuestData
  
      image: Optional[str]
      container: Optional[str]
      user: str
      force_pull: bool
  
      def wake(self):
          """ Wake up the guest """
          self.debug(
              f"Waking up container '{self.container}'.", level=2, shift=0)
  
      def start(self):
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Check if the image is available
          image_id = self.podman(
              ['images', '-q', self.image],
              message=f"Check for container image '{self.image}'.")[0].strip()
  
          # Pull image if not available or pull forced
          if not image_id 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}'.")
          self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f
  
  [... the rest is omitted ...]
files: tmt/steps/provision/podman.py
files: import os
  
  import click
  
  import tmt
  import tmt.steps.provision
  
  # Timeout in seconds of waiting for a connection
  CONNECTION_TIMEOUT = 60
  
  # Defaults
  DEFAULT_IMAGE = "fedora"
  DEFAULT_USER = "root"
  
  
  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 = ["image", "container", "pull", "user"]
  
      # Supported methods
      _methods = [tmt.steps.Method(name='container', doc=__doc__, order=50)]
  
      @classmethod
      def options(cls, how=None):
          """ Prepare command line options for connect """
          return [
              click.option(
                  '-i', '--image', metavar='IMAGE',
                  help='Select image to use. Short name or complete url.'),
              click.option(
                  '-c', '--container', metavar='NAME',
                  help='Name or id of an existing container to be used.'),
              click.option(
                  '-p', '--pull', is_flag=True,
                  help='Force pulling a fresh container image.'),
              click.option(
                  '-u', '--user', metavar='USER',
                  help='User to use for all container operations.')
              ] + super().options(how)
  
      def default(self, option, default=None):
          """ Return default data for given option """
          # Use 'fedora' as a default image
          if option == 'image':
              return DEFAULT_IMAGE
          # Use 'root' as a default user
          if option == 'user':
              return DEFAULT_USER
          # No other defaults available
          return default
  
      def wake(self, keys=None, data=None):
          """ Wake up the plugin, process data, apply options """
          super().wake(keys=keys, data=data)
          # Wake up podman instance
          if data:
              guest = GuestContainer(data, name=self.name, parent=self.step)
              guest.wake()
              self._guest = guest
  
      def go(self):
          """ Provision the container """
          super().go()
  
          # Show which image we are using
          pull = ' (force pull)' if self.get('pull') else ''
          self.info('image', f"{self.get('image')}{pull}", 'green')
  
          # Prepare data for the guest instance
          data = dict()
          for key in self._keys + self._common_keys:
              data[key] = self.get(key)
  
          # Create a new GuestTestcloud instance and start it
          self._guest = GuestContainer(data, name=self.name, parent=self.step)
          self._guest.start()
  
      def guest(self):
          """ Return the provisioned guest """
          return self._guest
  
      def requires(self):
          """ List of required packages needed for workdir sync """
          return GuestContainer.requires()
  
  
  class GuestContainer(tmt.Guest):
      """ Container Instance """
  
      def load(self, data):
          """ Load guest data and initialize attributes """
          super().load(data)
  
          # Load basic data
          self.image = data.get('image')
          self.force_pull = data.get('pull')
          self.container = data.get('container')
          self.user = data.get('user')
  
      def save(self):
          """ Save guest data for future wake up """
          data = super().save()
          data['container'] = self.container
          data['image'] = self.image
          data['user'] = self.user
          return data
  
      def wake(self):
          """ Wake up the guest """
          self.debug(
              f"Waking up container '{self.container}'.", level=2, shift=0)
  
      def start(self):
          """ Start provisioned guest """
          if self.opt('dry'):
              return
          # Check if the image is available
          image_id = self.podman(
              ['images', '-q', self.image],
              message=f"Check for container image '{self.image}'.")[0].strip()
  
          # Pull image if not available or pull forced
          if not image_id 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}'.")
          self.podman(
              ['run'] + workaround +
              ['--name', self.container, '-v', f'{workdir}:{workdir}:z',
               '-itd', '--user', self
  
  [... the rest is omitted ...]
files: tmt/steps/provision/podman.py
files: # coding: utf-8
  
  import dataclasses
  import os
  import platform
  import re
  import time
  from typing import Optional
  
  import click
  import fmf
  import requests
  
  import tmt
  import tmt.steps.provision
  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: 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 }}
  
  [... the rest is omitted ...]
files: tmt/steps/provision/testcloud.py
files: # coding: utf-8
  
  import os
  import platform
  import re
  import time
  from typing import Optional
  
  import click
  import fmf
  import requests
  
  import tmt
  import tmt.steps.provision
  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: 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
  
  [... the rest is omitted ...]
files: tmt/steps/provision/testcloud.py