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:

Installing guests (virt-install from URL)
Preparing guests for snapshotting (booting up, taking RAM image)
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 = -1

virt.DEVNULL = -3

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=None, partitions=None)

TEMPLATE

ks

appends = []

packages = []

partitions = None

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 = 'contest'

ipaddr = None

ssh_keyfile_path = '/var/lib/libvirt/images/contest.sshkey'

ssh_pubkey = None

disk_path = None

disk_format = None

state_file_path = '/var/lib/libvirt/images/contest.state'

snapshot_path = '/var/lib/libvirt/images/contest-snap.qcow2'

install_ready_path

snapshot_ready_path

install_basic(location=None, kickstart=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().

install(*, kickstart=None, rpmpack=None, **kwargs)

Install a new guest, to a shut down state.

This is a more user-friendly wrapper for install_basic(), providing extra herbs and spices useful for a default typical Anaconda based installation.

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

import_image(disk_path, disk_format='raw', *, secure_boot=False)

Import an existing disk image, creating a new guest domain from it.

The image is used as-is, in place. No copy or move is performed. Ideally, the image should be located in GUEST_IMG_DIR and named after the ‘.name’ attribute of the guest 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='.')

generate_ssh_keypair()

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(host, port=22, *, to_shutdown=False)

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.