test_connect.py 8.88 KB
Newer Older
1
#!/usr/bin/env python3
Lars Karlitski's avatar
Lars Karlitski committed
2 3 4

import dbus
import libvirttest
5
import pytest
6
import xmldata
Lars Karlitski's avatar
Lars Karlitski committed
7

8 9

class TestConnect(libvirttest.BaseTestClass):
10
    def test_connect_domain_create_xml(self):
11
        def domain_started(path, event, detail):
12 13
            if event != libvirttest.DomainEvent.STARTED:
                return
14
            assert detail == libvirttest.DomainEventStartedDetailType.BOOTED
15
            assert isinstance(path, dbus.ObjectPath)
Lars Karlitski's avatar
Lars Karlitski committed
16 17
            self.loop.quit()

18
        self.connect.connect_to_signal('DomainEvent', domain_started)
Lars Karlitski's avatar
Lars Karlitski committed
19

20
        path = self.connect.DomainCreateXML(xmldata.minimal_domain_xml, 0)
21
        assert isinstance(path, dbus.ObjectPath)
Lars Karlitski's avatar
Lars Karlitski committed
22 23 24

        self.main_loop()

25
    def test_comnect_domain_define_xml(self):
26
        def domain_defined(path, event, detail):
27 28
            if event != libvirttest.DomainEvent.DEFINED:
                return
29
            assert detail == libvirttest.DomainEventDefinedDetailType.ADDED
30
            assert isinstance(path, dbus.ObjectPath)
Lars Karlitski's avatar
Lars Karlitski committed
31 32
            self.loop.quit()

33
        self.connect.connect_to_signal('DomainEvent', domain_defined)
Lars Karlitski's avatar
Lars Karlitski committed
34

35
        path = self.connect.DomainDefineXML(xmldata.minimal_domain_xml)
36
        assert isinstance(path, dbus.ObjectPath)
Lars Karlitski's avatar
Lars Karlitski committed
37 38 39

        self.main_loop()

40 41
    @pytest.mark.parametrize("lookup_method_name,lookup_item", [
        ("DomainLookupByID", 'Id'),
42
        ("DomainLookupByName", 'Name'),
43
        ("DomainLookupByUUID", 'UUID'),
44 45 46 47 48
    ])
    def test_connect_domain_lookup_by_id(self, lookup_method_name, lookup_item):
        """Parameterized test for all DomainLookupBy* API calls of Connect interface
        """
        original_path = self.connect.ListDomains(0)[0]
49
        obj, _ = self.get_test_domain()
50 51 52 53
        props = obj.GetAll('org.libvirt.Domain', dbus_interface=dbus.PROPERTIES_IFACE)
        path = getattr(self.connect, lookup_method_name)(props[lookup_item])
        assert original_path == path

54 55 56 57 58
    def test_connect_find_storage_pool_sources(self):
        storageType = "logical"
        sources = self.connect.FindStoragePoolSources(storageType, "", 0)
        assert isinstance(sources, dbus.String)

59 60 61 62 63 64 65 66 67 68 69 70
    def test_connect_list_domains(self):
        domains = self.connect.ListDomains(0)
        assert isinstance(domains, dbus.Array)
        assert len(domains) == 1

        for path in domains:
            assert isinstance(path, dbus.ObjectPath)
            domain = self.bus.get_object('org.libvirt', path)

            # ensure the path exists by calling Introspect on it
            domain.Introspect(dbus_interface=dbus.INTROSPECTABLE_IFACE)

71
    @pytest.mark.parametrize("property_name,expected_type", [
72
        ("Encrypted", dbus.Boolean),
73
        ("Hostname", dbus.String),
74
        ("LibVersion", dbus.UInt64),
75
        ("Secure", dbus.Boolean),
76 77 78 79 80 81 82
        ("Version", dbus.UInt64),
    ])
    def test_connect_properties_return_type(self, property_name, expected_type):
        obj = self.bus.get_object('org.libvirt', '/org/libvirt/Test')
        props = obj.GetAll('org.libvirt.Connect', dbus_interface=dbus.PROPERTIES_IFACE)
        assert isinstance(props[property_name], expected_type)

83 84 85 86
    def test_connect_get_sysinfo(self):
        sysinfo = self.connect.GetSysinfo(0)
        assert isinstance(sysinfo, dbus.String)

87 88 89 90 91 92 93 94 95 96 97 98
    def test_list_networks(self):
        networks = self.connect.ListNetworks(0)
        assert isinstance(networks, dbus.Array)
        assert len(networks) == 1

        for path in networks:
            assert isinstance(path, dbus.ObjectPath)
            network = self.bus.get_object('org.libvirt', path)

            # ensure the path exists by calling Introspect on it
            network.Introspect(dbus_interface=dbus.INTROSPECTABLE_IFACE)

99 100 101
    def test_connect_get_capabilities(self):
        assert isinstance(self.connect.GetCapabilities(), dbus.String)

102 103 104 105
    def test_connect_get_cpu_model_names(self):
        arch = "x86_64"
        assert isinstance(self.connect.GetCPUModelNames(arch, 0), dbus.Array)

106 107 108 109 110 111 112 113 114 115 116 117
    def test_connect_list_storage_pools(self):
        storage_pools = self.connect.ListStoragePools(0)
        assert isinstance(storage_pools, dbus.Array)
        assert len(storage_pools) == 1

        for path in storage_pools:
            assert isinstance(path, dbus.ObjectPath)
            storage_pool = self.bus.get_object('org.libvirt', path)

            # ensure the path exists by calling Introspect on it
            storage_pool.Introspect(dbus_interface=dbus.INTROSPECTABLE_IFACE)

118
    def test_connect_network_create_xml(self):
119 120 121
        def network_started(path, event):
            if event != libvirttest.NetworkEvent.STARTED:
                return
122 123 124
            assert isinstance(path, dbus.ObjectPath)
            self.loop.quit()

125
        self.connect.connect_to_signal('NetworkEvent', network_started)
126

127
        path = self.connect.NetworkCreateXML(xmldata.minimal_network_xml)
128 129 130 131
        assert isinstance(path, dbus.ObjectPath)

        self.main_loop()

132
    def test_connect_network_define_xml(self):
133 134 135
        def network_defined(path, event):
            if event != libvirttest.NetworkEvent.DEFINED:
                return
136 137 138
            assert isinstance(path, dbus.ObjectPath)
            self.loop.quit()

139
        self.connect.connect_to_signal('NetworkEvent', network_defined)
140

141
        path = self.connect.NetworkDefineXML(xmldata.minimal_network_xml)
142 143 144 145
        assert isinstance(path, dbus.ObjectPath)

        self.main_loop()

146 147
    @pytest.mark.parametrize("lookup_method_name,lookup_item", [
        ("NetworkLookupByName", 'Name'),
148
        ("NetworkLookupByUUID", 'UUID'),
149 150 151 152
    ])
    def test_connect_network_lookup_by_property(self, lookup_method_name, lookup_item):
        """Parameterized test for all NetworkLookupBy* API calls of Connect interface
        """
153
        original_path, obj = self.get_test_network()
154 155 156 157
        prop = obj.Get('org.libvirt.Network', lookup_item, dbus_interface=dbus.PROPERTIES_IFACE)
        path = getattr(self.connect, lookup_method_name)(prop)
        assert original_path == path

158 159 160 161
    def test_connect_node_get_cpu_stats(self):
        stats = self.connect.NodeGetCPUStats(0, 0)
        assert isinstance(stats, dbus.Dictionary)

162 163 164 165
    def test_connect_node_get_free_memory(self):
        free_mem = self.connect.NodeGetFreeMemory()
        assert isinstance(free_mem, dbus.UInt64)

166 167 168 169
    def test_connect_node_get_cpumap(self):
        info = self.connect.NodeGetCPUMap(0)
        assert isinstance(info, dbus.Array)

170 171 172 173 174 175 176 177 178 179
    def test_connect_storage_pool_create_xml(self):
        def storage_pool_started(path, event, _detail):
            if event != libvirttest.StoragePoolEvent.STARTED:
                return
            assert isinstance(path, dbus.ObjectPath)
            self.loop.quit()

        self.connect.connect_to_signal('StoragePoolEvent', storage_pool_started)

        path = self.connect.StoragePoolCreateXML(
180
            xmldata.minimal_storage_pool_xml, 0)
181 182 183 184
        assert isinstance(path, dbus.ObjectPath)

        self.main_loop()

185 186 187 188 189 190 191 192 193 194
    def test_connect_storage_pool_define_xml(self):
        def storage_pool_defined(path, event, _detail):
            if event != libvirttest.StoragePoolEvent.DEFINED:
                return
            assert isinstance(path, dbus.ObjectPath)
            self.loop.quit()

        self.connect.connect_to_signal('StoragePoolEvent', storage_pool_defined)

        path = self.connect.StoragePoolDefineXML(
195
            xmldata.minimal_storage_pool_xml, 0)
196 197 198 199
        assert isinstance(path, dbus.ObjectPath)

        self.main_loop()

200 201
    @pytest.mark.parametrize("lookup_method_name,lookup_item", [
        ("StoragePoolLookupByName", 'Name'),
202
        ("StoragePoolLookupByUUID", 'UUID'),
203 204 205 206 207 208
    ])
    def test_connect_storage_pool_lookup_by_property(self,
                                                     lookup_method_name,
                                                     lookup_item):
        """Parameterized test for all StoragePoolLookupBy* API calls of Connect interface
        """
209
        original_path, obj = self.get_test_storage_pool()
210 211 212 213 214
        prop = obj.Get('org.libvirt.StoragePool', lookup_item,
                       dbus_interface=dbus.PROPERTIES_IFACE)
        path = getattr(self.connect, lookup_method_name)(prop)
        assert original_path == path

215 216 217
    @pytest.mark.usefixtures("storage_volume_create")
    @pytest.mark.parametrize("lookup_method_name,lookup_item", [
        ("StorageVolLookupByKey", 'Key'),
218
        ("StorageVolLookupByPath", 'Path'),
219 220 221 222 223 224 225 226 227 228 229 230
    ])
    def test_connect_storage_vol_lookup_by_property(self,
                                                    lookup_method_name,
                                                    lookup_item):
        """Parameterized test for all StorageVolLookupBy* API calls of Connect interface
        """
        original_path, obj = self.get_test_storage_volume()
        prop = obj.Get('org.libvirt.StorageVol', lookup_item,
                       dbus_interface=dbus.PROPERTIES_IFACE)
        path = getattr(self.connect, lookup_method_name)(prop)
        assert original_path == path

231

Lars Karlitski's avatar
Lars Karlitski committed
232
if __name__ == '__main__':
233
    libvirttest.run()