files.new_contents: #!/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
  
  [... the rest is omitted ...]
files.new_path: tests/multihost/provision/test.sh
files.old_contents: #!/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
  
  [... the rest is omitted ...]
files.old_path: tests/multihost/provision/test.sh
files.new_contents: 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
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/__init__.py
files.old_contents: 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
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/__init__.py
files.new_contents: 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 = '
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/artemis.py
files.old_contents: 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(
      'StepState
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/artemis.py
files.new_contents: 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
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/connect.py
files.old_contents: 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
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/connect.py
files.new_contents: 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
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/local.py
files.old_contents: 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 = Guest
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/local.py
files.new_contents: 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 = [t
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/podman.py
files.old_contents: 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
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/podman.py
files.new_contents: # 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
  runc
  
  [... the rest is omitted ...]
files.new_path: tmt/steps/provision/testcloud.py
files.old_contents: # 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
  
  [... the rest is omitted ...]
files.old_path: tmt/steps/provision/testcloud.py