# Copyright (C) 2020 Red Hat, Inc.
#
# This work is licensed under the GNU GPLv2 or later.
# See the COPYING file in the top-level directory.
# This file is a collection of code used for testing
# code paths primarily via our uitests/
import os
def fake_job_info():
import random
total = 1024 * 1024 * 1024
fakepcent = random.choice(range(1, 100))
remaining = ((total / 100) * fakepcent)
return [None, None, None, total, None, remaining]
def fake_interface_addresses(iface, source):
import libvirt
mac = iface.macaddr
if source == libvirt.VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_AGENT:
ret = {
'enp1s0': {'hwaddr': mac, 'addrs': [
{'addr': '10.0.0.1', 'prefix': 24, 'type': 0},
{'addr': 'fd00:beef::1', 'prefix': 128, 'type': 1},
{'addr': 'fe80::1', 'prefix': 64, 'type': 1}],
},
'lo': {'hwaddr': '00:00:00:00:00:00', 'addrs': [
{'addr': '127.0.0.1', 'prefix': 8, 'type': 0},
{'addr': '::1', 'prefix': 128, 'type': 1}],
},
}
else:
ret = {'vnet0': {'hwaddr': mac, 'addrs': [
{'addr': '10.0.0.3', 'prefix': 0, 'type': 0}],
}}
return ret
def fake_dhcp_leases():
ret = [{
'clientid': 'XXX',
'expirytime': 1598570993,
'hostname': None,
'iaid': '1448103320',
'iface': 'virbr1',
'ipaddr': 'fd00:beef::2',
'mac': 'BAD',
'prefix': 64,
'type': 1}, {
'clientid': 'YYY',
'expirytime': 1598570993,
'hostname': None,
'iaid': None,
'iface': 'virbr1',
'ipaddr': '10.0.0.2',
'mac': 'NOPE',
'prefix': 24,
'type': 0}]
return ret
def schedule_fake_agent_event(conn, cb):
import libvirt
vmname = conn.config.CLITestOptions.fake_agent_event
backend = conn.get_backend()
state = libvirt.VIR_CONNECT_DOMAIN_EVENT_AGENT_LIFECYCLE_STATE_CONNECTED
reason = libvirt.VIR_CONNECT_DOMAIN_EVENT_AGENT_LIFECYCLE_REASON_CHANNEL
def time_cb():
dom = backend.lookupByName(vmname)
cb(backend, dom, state, reason, None)
conn.timeout_add(500, time_cb)
def schedule_fake_nodedev_event(conn, lifecycle_cb, update_cb):
import libvirt
nodename = conn.config.CLITestOptions.fake_nodedev_event
backend = conn.get_backend()
def lifecycle_cb_wrapper():
nodedev = backend.nodeDeviceLookupByName(nodename)
state = libvirt.VIR_NODE_DEVICE_EVENT_CREATED
reason = 0
lifecycle_cb(backend, nodedev, state, reason, None)
def update_cb_wrapper():
nodedev = backend.nodeDeviceLookupByName(nodename)
update_cb(backend, nodedev, None)
conn.timeout_add(500, lifecycle_cb_wrapper)
conn.timeout_add(1000, update_cb_wrapper)
def fake_openauth(conn, cb, data):
ignore = conn
import libvirt
creds = [
[libvirt.VIR_CRED_USERNAME, "Username", None, None, None],
[libvirt.VIR_CRED_PASSPHRASE, "Password", None, None, None],
]
cb(creds, data)
assert all([bool(cred[4]) for cred in creds])
class fakeVirtBootstrap:
@staticmethod
def bootstrap(**kwargs):
import time
import logging
log = logging.getLogger("virtBootstrap")
log.info("mock virtBootstrap msg1")
kwargs["progress_cb"]({"status": "msg1"})
time.sleep(.5)
log.info("mock virtBootstrap msg2")
kwargs["progress_cb"]({"status": "msg2"})
time.sleep(.5)
log.info("mock virtBootstrap msg3")
kwargs["progress_cb"]({"status": "msg3"})
if "username" in kwargs:
raise RuntimeError("fakeVirtBootstrap mock auth failure!")
class CLITestOptionsClass:
"""
Helper class for parsing and tracking --test-* options.
The suboptions are:
* first-run: Run the app with fresh gsettings values saved to
a keyfile, mimicking a first app run. Also sets
GSETTINGS to use memory backend, in case any other app
preferences would be affected. The ui testsuite sets this
for most tests.
* xmleditor-enabled: Force the xmleditor gsettings preference on.
* gsettings-keyfile: Override the gsettings values with those
from the passed in keyfile, to test with different default
settings.
* leak-debug: Enabling this will tell us, at app exit time,
which vmmGObjects were not garbage collected. This is caused
by circular references to other objects, like a signal that
wasn't disconnected. It's not a big deal, but if we have objects
that can be created and destroyed a lot over the course of
the app lifecycle, every non-garbage collected class is a
memory leak. So it's nice to poke at this every now and then
and try to track down what we need to add to class _cleanup handling.
* no-events: Force disable libvirt event APIs for testing fallback
* break_setfacl: For setfacl calls to fail, for test scenarios.
This is hit via the directory search permissions checking
for disk image usage for qemu
* enable-libguestfs: Force enable the libguestfs gsetting
* disable-libguestfs: Force disable the libguestfs gsetting
* test-managed-save: Triggers a couple conditions for testing
managed save issues
* test-vm-run-fail: Make VM run fail, so we can test the error path
* spice-agent: Make spice-agent detection return true in viewer.py
* firstrun-uri: If set, use this as the initial connection URI
if we are doing firstrun testing
* fake-no-libvirtd: If doing firstrun testing, fake that
libvirtd is not installed
* fake-vnc-username: Fake VNC username auth request
* fake-console-resolution: Fake viewer console resolution response.
Spice doesn't return values here when we are just testing
against seabios in uitests, this fakes it to hit more code paths
* fake-systray: Enable the fake systray window
* fake-virtbootstrap: Mock the virtBootstrap module, since getting
it to actually work across fedora versions is hard
* object-denylist=NAME: Make object initialize for that name
fail to test some connection code paths
* conn-crash: Test connection abruptly closing like when
libvirtd is restarted.
* fake-agent-event: Fake a qemu guest agent API event
* fake-nodedev-event: Fake nodedev API events
* fake-openauth: Fake user+pass response from libvirt openauth,
for testing the TCP URI auth dialog
* fake-session-error: Fake a connection open error that
triggers logind session lookup
* short-poll: Use a polling interval of only .1 seconds to speed
up the uitests a bit
"""
def __init__(self, test_options_str):
optset = set()
for optstr in test_options_str:
optset.update(set(optstr.split(",")))
first_run = self._parse(optset)
self._process(first_run)
def _parse(self, optset):
def _get(optname):
if optname not in optset:
return False
optset.remove(optname)
return True
def _get_value(optname):
for opt in optset:
if opt.startswith(optname + "="):
optset.remove(opt)
return opt.split("=", 1)[1]
first_run = _get("first-run")
self.leak_debug = _get("leak-debug")
self.no_events = _get("no-events")
self.xmleditor_enabled = _get("xmleditor-enabled")
self.gsettings_keyfile = _get_value("gsettings-keyfile")
self.break_setfacl = _get("break-setfacl")
self.disable_libguestfs = _get("disable-libguestfs")
self.enable_libguestfs = _get("enable-libguestfs")
self.test_managed_save = _get("test-managed-save")
self.test_vm_run_fail = _get("test-vm-run-fail")
self.spice_agent = _get("spice-agent")
self.firstrun_uri = _get_value("firstrun-uri")
self.fake_no_libvirtd = _get("fake-no-libvirtd")
self.fake_vnc_username = _get("fake-vnc-username")
self.fake_console_resolution = _get("fake-console-resolution")
self.fake_systray = _get("fake-systray")
self.object_denylist = _get_value("object-denylist")
self.conn_crash = _get("conn-crash")
self.fake_agent_event = _get_value("fake-agent-event")
self.fake_nodedev_event = _get_value("fake-nodedev-event")
self.fake_openauth = _get("fake-openauth")
self.fake_session_error = _get("fake-session-error")
self.short_poll = _get("short-poll")
self.fake_virtbootstrap = _get("fake-virtbootstrap")
if optset: # pragma: no cover
raise RuntimeError("Unknown --test-options keys: %s" % optset)
return first_run
def _process(self, first_run):
if first_run:
# So other settings like gtk are reset and not affected
os.environ["GSETTINGS_BACKEND"] = "memory"
if first_run and not self.gsettings_keyfile:
import atexit
import tempfile
filename = tempfile.mktemp(prefix="virtmanager-firstrun-keyfile")
self.gsettings_keyfile = filename
atexit.register(lambda: os.unlink(filename))
if self.break_setfacl:
import virtinst.diskbackend
def fake_search(*args, **kwargs):
raise RuntimeError("Fake search fix fail from test suite")
virtinst.diskbackend.SETFACL = "getfacl"
# pylint: disable=protected-access
virtinst.diskbackend._fix_perms_chmod = fake_search