libvirttest.py 7.15 KB
Newer Older
1
from enum import IntEnum
Lars Karlitski's avatar
Lars Karlitski committed
2 3 4 5
from dbus.mainloop.glib import DBusGMainLoop
from gi.repository import GLib
import dbus
import os
6
import pytest
Lars Karlitski's avatar
Lars Karlitski committed
7
import subprocess
8
import sys
Lars Karlitski's avatar
Lars Karlitski committed
9
import time
10
import xmldata
11

Lars Karlitski's avatar
Lars Karlitski committed
12

13
root = os.environ.get('abs_top_builddir')
Lars Karlitski's avatar
Lars Karlitski committed
14 15 16 17 18
exe = os.path.join(root, 'src', 'libvirt-dbus')

DBusGMainLoop(set_as_default=True)


19 20
def run():
    exit(pytest.main(sys.argv))
Lars Karlitski's avatar
Lars Karlitski committed
21 22


23 24 25 26 27 28 29 30 31 32 33 34 35
class BaseTestClass():
    """ Base test class for whole test suite
    """
    connect = None
    bus = None
    libvirt_dbus = None
    loop = False

    @pytest.fixture(autouse=True)
    def libvirt_dbus_setup(self, request):
        """Start libvirt-dbus for each test function
        """
        os.environ['LIBVIRT_DEBUG'] = '3'
36
        self.libvirt_dbus = subprocess.Popen([exe, '--session'])
Lars Karlitski's avatar
Lars Karlitski committed
37 38 39 40 41 42 43 44 45
        self.bus = dbus.SessionBus()

        for i in range(10):
            if self.bus.name_has_owner('org.libvirt'):
                break
            time.sleep(0.1)
        else:
            raise TimeoutError('error starting libvirt-dbus')

46
        obj = self.bus.get_object('org.libvirt', '/org/libvirt/Test')
47
        self.connect = dbus.Interface(obj, 'org.libvirt.Connect')
Lars Karlitski's avatar
Lars Karlitski committed
48

49 50 51 52 53 54 55
    @pytest.fixture(autouse=True)
    def libvirt_dbus_teardown(self):
        """Terminate libvirt-dbus at the teardown of each test
        """
        yield
        self.libvirt_dbus.terminate()
        self.libvirt_dbus.wait(timeout=10)
Lars Karlitski's avatar
Lars Karlitski committed
56 57

    def main_loop(self):
58 59 60
        """Initializes the mainloop
        """
        assert getattr(self, 'loop', False) is False
Lars Karlitski's avatar
Lars Karlitski committed
61 62 63 64

        def timeout():
            self.loop.quit()
            del self.loop
65
            self.timeout = True
Lars Karlitski's avatar
Lars Karlitski committed
66

67
        self.timeout = False
Lars Karlitski's avatar
Lars Karlitski committed
68
        self.loop = GLib.MainLoop()
69
        GLib.timeout_add(2000, timeout)
Lars Karlitski's avatar
Lars Karlitski committed
70
        self.loop.run()
71 72
        if self.timeout:
            raise TimeoutError()
73

74
    @pytest.fixture
75 76 77 78 79 80 81 82 83 84 85 86
    def interface_create(self):
        """ Fixture to define dummy interface on the test driver

        This fixture should be used in the setup of every test manipulating
        with interfaces.
        """
        path = self.connect.InterfaceDefineXML(xmldata.minimal_interface_xml, 0)
        obj = self.bus.get_object('org.libvirt', path)
        interface_obj = dbus.Interface(obj, 'org.libvirt.Interface')
        interface_obj.Create(0)
        return path, interface_obj

87
    def node_device_create(self):
88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103
        # We need a usable parent nodedev: possible candidates are
        # scsi_host2 (available since libvirt 3.1.0) and
        # test-scsi-host-vport (available until libvirt 3.0.0).
        #
        # Looking up a non-existing nodedev raises an exception, so
        # we need to ignore them here: that's okay, because if we
        # can't find any then NodeDeviceCreateXML() itself will fail
        xml = xmldata.minimal_node_device_xml
        for parent in ['scsi_host2', 'test-scsi-host-vport']:
            try:
                if self.connect.NodeDeviceLookupByName(parent):
                    xml = xml.replace('@parent@', parent)
                    break
            except dbus.exceptions.DBusException:
                pass
        path = self.connect.NodeDeviceCreateXML(xml, 0)
104 105
        return path

106 107 108 109 110 111 112 113 114
    @pytest.fixture(name="node_device_create")
    def fixture_node_device_create(self):
        """ Fixture to create dummy node device on the test driver

        This fixture should be used in the setup of every test manipulating
        with node devices.
        """
        return self.node_device_create()

115 116 117 118 119 120 121 122 123 124 125 126 127
    @pytest.fixture
    def storage_volume_create(self):
        """ Fixture to create dummy storage volume on the test driver

        This fixture should be used in the setup of every test manipulating
        with test volume.
        """
        _, test_storage_pool = self.get_test_storage_pool()
        interface_obj = dbus.Interface(test_storage_pool,
                                       'org.libvirt.StoragePool')
        path = interface_obj.StorageVolCreateXML(xmldata.minimal_storage_vol_xml, 0)
        yield path

128
    def get_test_domain(self):
129 130 131
        path = self.connect.ListDomains(0)[0]
        obj = self.bus.get_object('org.libvirt', path)
        return obj, dbus.Interface(obj, 'org.libvirt.Domain')
132

133
    def get_test_network(self):
134 135 136 137 138 139 140 141 142 143
        """Fetch information for the test network from test driver

        Returns:
            (dbus.proxies.ProxyObject, dbus.proxies.ProxyObject):
            Test Network Object, Local proxy for the test Network Object.

        """
        path = self.connect.ListNetworks(0)[0]
        obj = self.bus.get_object('org.libvirt', path)
        return path, obj
144

145
    def get_test_storage_pool(self):
146 147 148 149 150 151 152 153 154 155 156 157
        """Fetch information for the test storage pool from test driver

        Returns:
            (dbus.proxies.ProxyObject, dbus.proxies.ProxyObject):
            Test StoragePool Object, Local proxy for the test StoragePool
            Object.

        """
        path = self.connect.ListStoragePools(0)[0]
        obj = self.bus.get_object('org.libvirt', path)
        return path, obj

158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173
    def get_test_storage_volume(self):
        """Fetch information for the test storage vol from test driver

        Returns:
            (dbus.proxies.ProxyObject, dbus.proxies.ProxyObject):
            Test StorageVol Object, Local proxy for the test StorageVol
            Object.

        """
        _, test_storage_pool = self.get_test_storage_pool()
        pool_iface = dbus.Interface(test_storage_pool,
                                    'org.libvirt.StoragePool')
        path = pool_iface.ListStorageVolumes(0)[0]
        obj = self.bus.get_object('org.libvirt', path)
        return path, obj

174 175 176 177 178 179 180 181 182 183 184

class DomainEvent(IntEnum):
    DEFINED = 0
    UNDEFINED = 1
    STARTED = 2
    SUSPENDED = 3
    RESUMED = 4
    STOPPED = 5
    SHUTDOWN = 6
    PMSUSPENDED = 7
    CRASHED = 8
185 186


187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215
class DomainEventDefinedDetailType(IntEnum):
    ADDED = 0
    UPDATED = 1
    RENAMED = 2
    FROM_SNAPSHOT = 3
    LAST = 4


class DomainEventResumedDetailType(IntEnum):
    UNPAUSED = 0
    MIGRATED = 1
    FROM_SNAPSHOT = 2
    POSTCOPY = 3
    LAST = 4


class DomainEventStartedDetailType(IntEnum):
    BOOTED = 0
    MIGRATED = 1
    RESTORED = 2
    FROM_SNAPSHOT = 3
    WAKEUP = 4
    LAST = 5


class DomainEventStoppedDetailType(IntEnum):
    SHUTDOWN = 0
    DESTROYED = 1
    CRASHED = 2
216
    MIGRATED = 3
217 218 219 220 221 222 223 224
    SAVED = 4
    FAILED = 5
    FROM_SNAPSHOT = 6
    LAST = 7


class DomainEventSuspendedDetailType(IntEnum):
    PAUSED = 0
225
    MIGRATED = 1
226
    IOERROR = 2
227
    WATCHDOG = 3
228 229 230 231 232 233 234 235 236 237 238 239 240 241
    RESTORED = 4
    FROM_SNAPSHOT = 5
    API_ERROR = 6
    POSTCOPY = 7
    POSTCOPY_FAILED = 8
    LAST = 9


class DomainEventUndefinedDetailType(IntEnum):
    REMOVED = 0
    RENAMED = 1
    LAST = 2


242 243 244 245 246 247 248 249 250 251 252 253
class DomainState(IntEnum):
    NOSTATE = 0
    RUNNING = 1
    BLOCKED = 2
    PAUSED = 3
    SHUTDOWN = 4
    SHUTOFF = 5
    CRASHED = 6
    PMSUSPENDED = 7
    LAST = 8


254 255 256 257 258
class NetworkEvent(IntEnum):
    DEFINED = 0
    UNDEFINED = 1
    STARTED = 2
    STOPPED = 3
259

260

261 262 263
class NodeDeviceEvent(IntEnum):
    CREATED = 0
    DELETED = 1
264

265

266 267 268 269 270 271 272 273
class StoragePoolBuildFlags(IntEnum):
    NEW = 0
    REPAIR = 1
    RESIZE = 2
    NO_OVERWRITE = 4
    OVERWRITE = 8


274 275 276 277 278 279
class StoragePoolEvent(IntEnum):
    DEFINED = 0
    UNDEFINED = 1
    STARTED = 2
    STOPPED = 3
    LAST = 4