mirror of
https://github.com/virt-manager/virt-manager.git
synced 2024-12-23 17:34:21 +03:00
domain: Add test mocking for IP address APIs
Signed-off-by: Cole Robinson <crobinso@redhat.com>
This commit is contained in:
Cole Robinson
parent
f06cc11546
commit
a2ad7b4f2f
@ -734,6 +734,16 @@ test-many-devices, like an alternate RNG, EOL OS ID, title field
|
||||
</model>
|
||||
</video>
|
||||
|
||||
<!-- mock connected qemu agent and NIC for IP UI testing -->
|
||||
<interface type='bridge'>
|
||||
<source bridge='blahbr0'/>
|
||||
<mac address='22:11:11:11:72:72'/>
|
||||
</interface>
|
||||
<channel type='unix'>
|
||||
<source mode='bind' path='/tmp/qemuga.sock'/>
|
||||
<target type='virtio' name='org.qemu.guest_agent.0' state='connected'/>
|
||||
</channel>
|
||||
|
||||
<!-- To test enable USB in addhw -->
|
||||
<controller type='usb' model='none'/>
|
||||
</devices>
|
||||
|
||||
@ -328,7 +328,6 @@ class Details(uiutils.UITestCase):
|
||||
appl.click()
|
||||
uiutils.check(lambda: not appl.sensitive)
|
||||
|
||||
|
||||
# Network values w/ macvtap manual
|
||||
tab = self._select_hw(win, "NIC :54:32:10", "network-tab")
|
||||
tab.find("IP address", "push button").click()
|
||||
@ -355,6 +354,36 @@ class Details(uiutils.UITestCase):
|
||||
appl.click()
|
||||
uiutils.check(lambda: not appl.sensitive)
|
||||
|
||||
def testDetailsNetIPAddress(self):
|
||||
"""
|
||||
Test all the IP code paths with a few mock cases
|
||||
"""
|
||||
win = self._open_details_window(vmname="test-many-devices")
|
||||
def check_ip(*args):
|
||||
for ip in args:
|
||||
tab.find_fuzzy(ip, "label")
|
||||
|
||||
# First case has a virtual network, so hits the leases path
|
||||
tab = self._select_hw(win, "NIC :54:32:10", "network-tab")
|
||||
check_ip("10.0.0.2", "fd00:beef::2")
|
||||
tab.find("IP address:", "push button").click()
|
||||
check_ip("10.0.0.2", "fd00:beef::2")
|
||||
|
||||
# Next case has a missing virtual network, so hits the arp path
|
||||
tab = self._select_hw(win, "NIC :11:11:11", "network-tab")
|
||||
check_ip("Unknown")
|
||||
tab.find("IP address:", "push button").click()
|
||||
check_ip("10.0.0.3")
|
||||
|
||||
win.keyCombo("<alt>F4")
|
||||
uiutils.check(lambda: not win.showing)
|
||||
self.app.topwin.click_title()
|
||||
|
||||
# Tests the fake qemu guest agent path
|
||||
win = self._open_details_window(vmname="test alternate devs title")
|
||||
tab = self._select_hw(win, "NIC :11:72:72", "network-tab")
|
||||
check_ip("10.0.0.1", "fd00:beef::1/128")
|
||||
|
||||
|
||||
def testDetailsEditDevices1(self):
|
||||
"""
|
||||
@ -364,6 +393,22 @@ class Details(uiutils.UITestCase):
|
||||
shutdown=True)
|
||||
appl = win.find("config-apply", "push button")
|
||||
|
||||
# Fail to hotremove
|
||||
tab = self._select_hw(win, "Floppy 1", "disk-tab")
|
||||
share = tab.find("Shareable", "check box")
|
||||
share.click()
|
||||
uiutils.check(lambda: appl.sensitive)
|
||||
win.find("config-remove").click()
|
||||
delete = self.app.root.find_fuzzy("Remove Disk", "frame")
|
||||
delete.find_fuzzy("Delete", "button").click()
|
||||
self._click_alert_button("change will take effect", "OK")
|
||||
uiutils.check(lambda: not delete.showing)
|
||||
uiutils.check(lambda: appl.sensitive)
|
||||
uiutils.check(lambda: share.checked)
|
||||
win.find("config-cancel").click()
|
||||
|
||||
self._stop_vm(win)
|
||||
|
||||
# Graphics simple VNC -> SPICE
|
||||
tab = self._select_hw(win, "Display VNC", "graphics-tab")
|
||||
tab.combo_select("Type:", "Spice")
|
||||
|
||||
@ -1265,7 +1265,7 @@ class vmmDetails(vmmGObjectUI):
|
||||
self._browse_file(cb, reason=reason)
|
||||
|
||||
def _set_network_ip_details(self, net):
|
||||
ipv4, ipv6 = self.vm.get_interface_addresses(net)
|
||||
ipv4, ipv6 = self.vm.get_ips(net)
|
||||
label = ipv4 or ""
|
||||
if ipv6:
|
||||
if label:
|
||||
@ -1275,7 +1275,7 @@ class vmmDetails(vmmGObjectUI):
|
||||
|
||||
def _refresh_ip(self):
|
||||
net = self._get_hw_row()[HW_LIST_COL_DEVICE]
|
||||
self.vm.refresh_interface_addresses(net)
|
||||
self.vm.refresh_ips(net)
|
||||
self._set_network_ip_details(net)
|
||||
|
||||
|
||||
|
||||
@ -4,6 +4,59 @@
|
||||
# See the COPYING file in the top-level directory.
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
class CLITestOptionsClass:
|
||||
"""
|
||||
Helper class for parsing and tracking --test-* options.
|
||||
|
||||
@ -19,20 +19,13 @@ from virtinst import log
|
||||
from .libvirtobject import vmmLibvirtObject
|
||||
from ..baseclass import vmmGObject
|
||||
from ..lib.libvirtenummap import LibvirtEnumMap
|
||||
from ..lib import testmock
|
||||
|
||||
|
||||
class _SENTINEL(object):
|
||||
pass
|
||||
|
||||
|
||||
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 start_job_progress_thread(vm, meter, progtext):
|
||||
current_thread = threading.currentThread()
|
||||
|
||||
@ -71,6 +64,87 @@ def start_job_progress_thread(vm, meter, progtext):
|
||||
t.start()
|
||||
|
||||
|
||||
class _IPFetcher:
|
||||
"""
|
||||
Helper class to contain all IP fetching and processing logic
|
||||
"""
|
||||
def __init__(self):
|
||||
self._cache = None
|
||||
|
||||
def refresh(self, vm, iface):
|
||||
self._cache = {"qemuga": {}, "arp": {}}
|
||||
|
||||
if iface.type == "network":
|
||||
net = vm.conn.get_net(iface.source)
|
||||
if net:
|
||||
net.get_dhcp_leases(refresh=True)
|
||||
|
||||
if not vm.is_active():
|
||||
return
|
||||
|
||||
if vm.agent_ready():
|
||||
self._cache["qemuga"] = vm.get_interface_addresses(
|
||||
iface,
|
||||
libvirt.VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_AGENT)
|
||||
|
||||
arp_flag = getattr(libvirt,
|
||||
"VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_ARP", 3)
|
||||
self._cache["arp"] = vm.get_interface_addresses(iface, arp_flag)
|
||||
|
||||
def get(self, vm, iface):
|
||||
if self._cache is None:
|
||||
self.refresh(vm, iface)
|
||||
|
||||
qemuga = self._cache["qemuga"]
|
||||
arp = self._cache["arp"]
|
||||
leases = []
|
||||
if iface.type == "network":
|
||||
net = vm.conn.get_net(iface.source)
|
||||
if net:
|
||||
leases = net.get_dhcp_leases()
|
||||
|
||||
def extract_dom(addrs):
|
||||
ipv4 = None
|
||||
ipv6 = None
|
||||
if addrs["hwaddr"] == iface.macaddr:
|
||||
for addr in (addrs["addrs"] or []):
|
||||
if addr["type"] == 0:
|
||||
ipv4 = addr["addr"]
|
||||
elif (addr["type"] == 1 and
|
||||
not str(addr["addr"]).startswith("fe80")):
|
||||
ipv6 = addr["addr"] + "/" + str(addr["prefix"])
|
||||
return ipv4, ipv6
|
||||
|
||||
def extract_lease(lease):
|
||||
ipv4 = None
|
||||
ipv6 = None
|
||||
mac = lease["mac"]
|
||||
if vm.conn.is_test():
|
||||
# Hack it to match our interface for UI testing
|
||||
mac = iface.macaddr
|
||||
if mac == iface.macaddr:
|
||||
if lease["type"] == 0:
|
||||
ipv4 = lease["ipaddr"]
|
||||
elif lease["type"] == 1:
|
||||
ipv6 = lease["ipaddr"]
|
||||
return ipv4, ipv6
|
||||
|
||||
|
||||
for datalist in [list(qemuga.values()), leases, list(arp.values())]:
|
||||
ipv4 = None
|
||||
ipv6 = None
|
||||
for data in datalist:
|
||||
if "expirytime" in data:
|
||||
tmpipv4, tmpipv6 = extract_lease(data)
|
||||
else:
|
||||
tmpipv4, tmpipv6 = extract_dom(data)
|
||||
ipv4 = tmpipv4 or ipv4
|
||||
ipv6 = tmpipv6 or ipv6
|
||||
if ipv4 or ipv6:
|
||||
return ipv4, ipv6
|
||||
return None, None
|
||||
|
||||
|
||||
class vmmInspectionApplication(object):
|
||||
def __init__(self):
|
||||
self.name = None
|
||||
@ -283,7 +357,7 @@ class vmmDomain(vmmLibvirtObject):
|
||||
self._autostart = None
|
||||
self._domain_caps = None
|
||||
self._status_reason = None
|
||||
self._ip_cache = None
|
||||
self._ipfetcher = _IPFetcher()
|
||||
|
||||
self.managedsave_supported = False
|
||||
self._domain_state_supported = False
|
||||
@ -984,7 +1058,7 @@ class vmmDomain(vmmLibvirtObject):
|
||||
|
||||
def job_info(self):
|
||||
if self.conn.is_test():
|
||||
return _fake_job_info()
|
||||
return testmock.fake_job_info()
|
||||
return self._backend.jobInfo()
|
||||
def abort_job(self):
|
||||
self._backend.abortJob()
|
||||
@ -1045,86 +1119,36 @@ class vmmDomain(vmmLibvirtObject):
|
||||
"""
|
||||
Return connected state of an agent.
|
||||
"""
|
||||
# we need to get a fresh agent channel object on each call so it
|
||||
# reflects the current state
|
||||
dev = self._get_agent()
|
||||
return dev and dev.target_state == "connected"
|
||||
if not dev:
|
||||
return False
|
||||
|
||||
def refresh_interface_addresses(self, iface):
|
||||
self._ip_cache = {"qemuga": {}, "arp": {}}
|
||||
if iface.type == "network":
|
||||
net = self.conn.get_net(iface.source)
|
||||
if net:
|
||||
net.get_dhcp_leases(refresh=True)
|
||||
if not self.is_active():
|
||||
return
|
||||
|
||||
if self.agent_ready():
|
||||
self._ip_cache["qemuga"] = self._get_interface_addresses(
|
||||
libvirt.VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_AGENT)
|
||||
|
||||
arp_flag = getattr(libvirt,
|
||||
"VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_ARP", 3)
|
||||
self._ip_cache["arp"] = self._get_interface_addresses(arp_flag)
|
||||
|
||||
def _get_interface_addresses(self, source):
|
||||
log.debug("Calling interfaceAddresses source=%s", source)
|
||||
try:
|
||||
return self._backend.interfaceAddresses(source)
|
||||
except Exception as e:
|
||||
log.debug("interfaceAddresses failed: %s", str(e))
|
||||
return {}
|
||||
|
||||
def get_interface_addresses(self, iface):
|
||||
if self._ip_cache is None:
|
||||
self.refresh_interface_addresses(iface)
|
||||
|
||||
qemuga = self._ip_cache["qemuga"]
|
||||
arp = self._ip_cache["arp"]
|
||||
leases = []
|
||||
if iface.type == "network":
|
||||
net = self.conn.get_net(iface.source)
|
||||
if net:
|
||||
leases = net.get_dhcp_leases()
|
||||
|
||||
def extract_dom(info):
|
||||
ipv4 = None
|
||||
ipv6 = None
|
||||
for addrs in info.values():
|
||||
if addrs["hwaddr"] != iface.macaddr:
|
||||
continue
|
||||
if not addrs["addrs"]:
|
||||
continue
|
||||
for addr in addrs["addrs"]:
|
||||
if addr["type"] == 0:
|
||||
ipv4 = addr["addr"]
|
||||
elif (addr["type"] == 1 and
|
||||
not str(addr["addr"]).startswith("fe80")):
|
||||
ipv6 = addr["addr"] + "/" + str(addr["prefix"])
|
||||
return ipv4, ipv6
|
||||
|
||||
def extract_lease(info):
|
||||
ipv4 = None
|
||||
ipv6 = None
|
||||
if info["mac"] == iface.macaddr:
|
||||
if info["type"] == 0:
|
||||
ipv4 = info["ipaddr"]
|
||||
elif info["type"] == 1:
|
||||
ipv6 = info["ipaddr"]
|
||||
return ipv4, ipv6
|
||||
|
||||
for ips in ([qemuga] + leases + [arp]):
|
||||
if "expirytime" in ips:
|
||||
ipv4, ipv6 = extract_lease(ips)
|
||||
else:
|
||||
ipv4, ipv6 = extract_dom(ips)
|
||||
if ipv4 or ipv6:
|
||||
return ipv4, ipv6
|
||||
return None, None
|
||||
target_state = dev.target_state
|
||||
if self.conn.is_test():
|
||||
# test driver doesn't report 'connected' state so hack it here
|
||||
target_state = "connected"
|
||||
return target_state == "connected"
|
||||
|
||||
def refresh_snapshots(self):
|
||||
self._snapshot_list = None
|
||||
|
||||
def get_interface_addresses(self, iface, source):
|
||||
ret = {}
|
||||
log.debug("Calling interfaceAddresses source=%s", source)
|
||||
try:
|
||||
ret = self._backend.interfaceAddresses(source)
|
||||
except Exception as e:
|
||||
log.debug("interfaceAddresses failed: %s", str(e))
|
||||
if self.conn.is_test():
|
||||
ret = testmock.fake_interface_addresses(iface, source)
|
||||
return ret
|
||||
|
||||
def get_ips(self, iface):
|
||||
return self._ipfetcher.get(self, iface)
|
||||
|
||||
def refresh_ips(self, iface):
|
||||
return self._ipfetcher.refresh(self, iface)
|
||||
|
||||
def set_time(self):
|
||||
"""
|
||||
Try to set VM time to the current value. This is typically useful when
|
||||
|
||||
@ -10,6 +10,7 @@ from virtinst import log
|
||||
from virtinst import Network
|
||||
|
||||
from .libvirtobject import vmmLibvirtObject
|
||||
from ..lib import testmock
|
||||
|
||||
|
||||
def _make_addr_str(addrStr, prefix, netmaskStr):
|
||||
@ -88,11 +89,14 @@ class vmmNetwork(vmmLibvirtObject):
|
||||
self._backend.setAutostart(value)
|
||||
|
||||
def _refresh_dhcp_leases(self):
|
||||
ret = []
|
||||
try:
|
||||
self._leases = self._backend.DHCPLeases()
|
||||
ret = self._backend.DHCPLeases()
|
||||
except Exception as e:
|
||||
log.debug("Error getting %s DHCP leases: %s", self, str(e))
|
||||
self._leases = []
|
||||
if self.conn.is_test():
|
||||
ret = testmock.fake_dhcp_leases()
|
||||
self._leases = ret
|
||||
|
||||
def get_dhcp_leases(self, refresh=False):
|
||||
if self._leases is None or refresh:
|
||||
|
||||
Loading…
Reference in New Issue
Block a user