virt

Provides utilities for installing, snapshotting and managing Libvirt-based virtual machines.

Quick Terminology:

• host: system that runs VMs (a.k.a. hypervisor)

• guest: the VM itself

• domain: libvirt’s terminology for a VM

• “host”: guest translated to Czech :)

The host-related functionality is mainly installing libvirt + deps.

The guest-related functionality consists of:

1. Installing guests (virt-install from URL)

2. Preparing guests for snapshotting (booting up, taking RAM image)

3. Snapshotting guests (repeatedly restoring, using, throwing away)

There is a Guest() class, which represents a guest with a specific name (used for its domain name). Aside from doing (1) and (2), an instance of Guest can be used in two ways, both using a context manager (‘g’ is an instance):

• g.snapshotted() - Assumes (1) and (2) were done, creates a snapshot and restores the guest

  from its RAM image, waits for ssh.

  • Stops the guest and deletes the snapshot on __exit__

• g.booted() - Assumes (1) was done and just boots and waits for ssh.

Any host yum.repos.d repositories are given to Anaconda via kickstart ‘repo’ upon guest installation.

Step (1) can be replaced by importing a pre-existing ImageBuilder image, with (2) and (3), or Guest() usage, remaining unaffected / compatible.

Installation customization can be done via g.install() arguments, such as by instantiating Kickstart() in the test itself, modifying it, and passing the instance to g.install().

Example using snapshots:

import subprocess import virt

virt.Host.setup() g = virt.Guest(‘gui’)

# reuse if it already exists from previous tests, reinstall if not if not g.can_be_snapshotted():

g.install() g.prepare_for_snapshot()

with g.snapshotted():

state = g.ssh(‘ls’, ‘/root’, stdout=subprocess.PIPE) print(state.stdout) if state.returncode != 0:

report_failure()

with g.snapshotted():

out = g.ssh(…)

Example using plain one-time-use guest:

import virt

virt.Host.setup() g = virt.Guest()

ks = virt.Kickstart() ks.add_post(‘some test-specific stuff’) g.install(kickstart=ks)

with g.booted():

g.ssh( … ) g.ssh( … )

Module Contents

virt.GUEST_NAME = 'contest'

virt.GUEST_LOGIN_PASS = 'contest'

virt.GUEST_SSH_USER = 'root'

virt.GUEST_IMG_DIR = '/var/lib/libvirt/images'

virt.NETWORK_NETMASK = '255.255.252.0'

virt.NETWORK_HOST = '192.168.120.1'

virt.NETWORK_RANGE = ['192.168.120.2', '192.168.123.254']

virt.NETWORK_EXPIRY = 168

virt.INSTALL_FAILURES = [b'org\\.fedoraproject\\.Anaconda\\.Addons\\.OSCAP\\.*: The installation should be aborted',...

virt.PIPE

virt.DEVNULL

class virt.Host

Utilities for host system preparation.

static check_virt_capability()

Return True if the host has HW-accelerated virtualization support (HVM). Else return False.

static setup_network()

static create_sshvm(dest)

classmethod setup()

class virt.Kickstart(template=TEMPLATE, packages=PACKAGES, partitions=None)

TEMPLATE

PACKAGES = ['openscap-scanner']

ks

appends = []

packages

partitions

assemble()

to_tmpfile()

append(content)

Append arbitrary string content to the kickstart template.

add_pre(content)

add_post(content)

add_install_only_repo(name, baseurl)

add_host_repos()

add_oscap_addon(keyvals)

Append an OSCAP addon section, with key=value pairs from ‘keyvals’.

add_authorized_key(pubkey, homedir='/root', owner='root')

class virt.Guest(tag=None, *, name=GUEST_NAME)

When instantiated, represents a guest (VM).

Set a ‘tag’ (string) to a unique name you would like to share across tests that use snapshots - the .can_be_snapshotted() function will return True when it finds an already installed guest using the same tag. Tag-less guests can be used only for snapshotting within the same test and should not be shared across tests.

SETUP

SETUP_REQUIRES = ['openssh-server', 'qemu-guest-agent', 'policycoreutils-python-utils']

tag

name

ipaddr = None

ssh_keyfile_path

orig_disk_path = None

orig_disk_format = None

state_file_path

snapshot_path

install_ready_path

snapshot_ready_path

install(location=None, kickstart=None, rpmpack=None, secure_boot=False, disk_format='raw')

Install a new guest, to a shut down state.

If ‘location’ is given, it is passed to virt-install, otherwise a first host repo with an Anaconda stage2 image is used.

If custom ‘kickstart’ is used, it is passed to virt-install. It should be a Kickstart class instance. To customize the instance (ie. add code before/after code added by member functions), subclass Kickstart and set __init__() or assemble().

If custom ‘rpmpack’ is specified (RpmPack instance), it is used instead of a self-made instance.

start()

destroy()

shutdown()

soft_reboot()

Reboot by issuing ‘reboot’ via ssh.

reset()

undefine(incl_storage=False)

is_installed()

can_be_snapshotted()

prepare_for_snapshot()

snapshotted()

Create a snapshot, restore the guest, ready it for communication.

booted(*, safe_shutdown=False)

Just boot the guest, ready it for communication.

With ‘safe_shutdown’, guarantee that the guest shuts down cleanly. This is useful for setup-style use cases where the test wants to modify the guest before taking a snapshot.

ssh(*cmd, **kwargs)

Run a command via ssh(1) inside the guest.

ssh_stream(*cmd, **kwargs)

copy_from(remote_file, local_file='.')

copy_to(local_file, remote_file='.')

rsync_from(remote_path, local_path='.')

rsync_to(local_path, remote_path='.')

guest_agent_cmd(cmd, args=None, blind=False)

Execute an arbitrary qemu-guest-agent command.

If ‘blind’ is specified, the command is executed without waiting for completion and nothing is returned.

wipe()

Remove all previous data and metadata of a domain.

Useful to clean up an unclean state from a previous use of Guest(), or just to remove any leftovers after using a one-time guest.

virt.guest_domstate(name)

virt.wait_for_domstate(name, state, timeout=300, sleep=0.5)

Wait until the guest reaches a specified libvirt domain state (‘running’, ‘shut off’, etc.).

virt.domifaddr(name)

Return a guest’s IP address, queried from libvirt.

virt.wait_for_ifaddr(name, timeout=600, sleep=0.5)

virt.wait_for_ssh(ip, port=22, timeout=600, sleep=0.5, to_shutdown=False)

Attempt to repeatedly connect to a given ip address and port (both strings) and return when a connection has been established with a genuine sshd service (not just any TCP server).

If the attempts continue to fail for ‘timeout’ seconds, raise TimeoutError.

If ‘to_shutdown’ is true, wait for ssh to shut down, instead of to start. Useful for waiting until a guest reboots without changing domain state.

virt.virsh(*virsh_args, **run_args)

virt.translate_ssg_kickstart(ks_file)

Parse (and tweak) a kickstart shipped with the upstream content into class Kickstart instance.

virt.translate_oscap_kickstart(lines, datastream)

Parse (and tweak) a kickstart generated via ‘oscap xccdf generate fix’.

virt.domain_xml_diskinfo(xmlstr)

virt.get_state_image_disk(image)

Get path/format of the first <disk> definition in a RAM image file.

virt.set_state_image_disk(image, source_file, image_format)

Set a disk path inside a saved guest RAM image to ‘diskpath’.

virt.set_domain_memory(domain, amount, unit='MiB')

Set the amount of RAM allowed for a defined guest.