shaba/virt-manager
1
0
Fork 0
mirror of https://github.com/virt-manager/virt-manager.git synced 2024-12-23 17:34:21 +03:00
Code Issues Projects Releases Wiki Activity
70a5cb7b55
virt-manager/virtManager/clone.py
Cole Robinson 1684049fa3 virtManager: Misc coverage annotations 
Signed-off-by: Cole Robinson <crobinso@redhat.com>
2020-08-26 15:20:06 -04:00

894 lines
30 KiB
Python
Raw Blame History

# Copyright (C) 2009, 2013 Red Hat, Inc.
# Copyright (C) 2009 Cole Robinson <crobinso@redhat.com>
#
# This work is licensed under the GNU GPLv2 or later.
# See the COPYING file in the top-level directory.
import os
from gi.repository import Gtk
import virtinst
from virtinst import Cloner
from virtinst import DeviceInterface
from virtinst import log
from .lib import uiutil
from .baseclass import vmmGObjectUI
from .asyncjob import vmmAsyncJob
from .storagebrowse import vmmStorageBrowser
from .object.storagepool import vmmStoragePool
STORAGE_COMBO_CLONE = 0
STORAGE_COMBO_SHARE = 1
STORAGE_COMBO_SEP = 2
STORAGE_COMBO_DETAILS = 3
STORAGE_INFO_ORIG_PATH = 0
STORAGE_INFO_NEW_PATH = 1
STORAGE_INFO_TARGET = 2
STORAGE_INFO_SIZE = 3
STORAGE_INFO_DEVTYPE = 4
STORAGE_INFO_DO_CLONE = 5
STORAGE_INFO_CAN_CLONE = 6
STORAGE_INFO_CAN_SHARE = 7
STORAGE_INFO_DO_DEFAULT = 8
STORAGE_INFO_DEFINFO = 9
STORAGE_INFO_FAILINFO = 10
STORAGE_INFO_COMBO = 11
STORAGE_INFO_MANUAL_PATH = 12
NETWORK_INFO_LABEL = 0
NETWORK_INFO_ORIG_MAC = 1
NETWORK_INFO_NEW_MAC = 2
def can_we_clone(conn, vol, path):
"""Is the passed path even clone-able"""
ret = True
msg = None
if not path:
msg = _("No storage to clone.")
elif not vol:
is_dev = path.startswith("/dev")
if conn.is_remote():
msg = _("Cannot clone unmanaged remote storage.")
elif not os.access(path, os.R_OK):
if is_dev:
msg = _("Block devices to clone must be libvirt\n"
"managed storage volumes.")
else:
msg = _("No write access to parent directory.")
elif not os.path.exists(path):
msg = _("Path does not exist.")
else:
pool = vol.get_parent_pool()
if not vmmStoragePool.supports_volume_creation(
pool.get_type(), clone=True):
msg = _("Cannot clone %s storage pool.") % pool.get_type()
if msg:
ret = False
return (ret, msg)
def do_we_default(conn, vol, path, ro, shared, devtype):
""" Returns (do we clone by default?, info string if not)"""
ignore = conn
info = ""
can_default = True
def append_str(str1, str2, delim=", "):
if not str2:
return str1
if str1:
str1 += delim
str1 += str2
return str1
if (devtype == virtinst.DeviceDisk.DEVICE_CDROM or
devtype == virtinst.DeviceDisk.DEVICE_FLOPPY):
info = append_str(info, _("Removable"))
if ro:
info = append_str(info, _("Read Only"))
elif not vol and path and not os.access(path, os.W_OK):
info = append_str(info, _("No write access"))
if vol:
pool_type = vol.get_parent_pool().get_type()
if pool_type == virtinst.StoragePool.TYPE_DISK:
info = append_str(info, _("Disk device"))
can_default = False
if shared:
info = append_str(info, _("Shareable"))
return (not info, info, can_default)
class vmmCloneVM(vmmGObjectUI):
@classmethod
def show_instance(cls, parentobj, vm):
try:
# Maintain one dialog per connection
uri = vm.conn.get_uri()
if cls._instances is None:
cls._instances = {}
if uri not in cls._instances:
cls._instances[uri] = vmmCloneVM()
cls._instances[uri].show(parentobj.topwin, vm)
except Exception as e: # pragma: no cover
parentobj.err.show_err(
_("Error launching clone dialog: %s") % str(e))
def __init__(self):
vmmGObjectUI.__init__(self, "clone.ui", "vmm-clone")
self.vm = None
self.clone_design = None
self.storage_list = {}
self.target_list = []
self.net_list = {}
self.mac_list = []
self.storage_browser = None
self.change_mac = self.widget("vmm-change-mac")
self.change_mac.set_transient_for(self.topwin)
self.change_storage = self.widget("vmm-change-storage")
self.change_storage.set_transient_for(self.topwin)
self.builder.connect_signals({
"on_clone_delete_event": self.close,
"on_clone_cancel_clicked": self.close,
"on_clone_ok_clicked": self.finish,
# Change mac dialog
"on_vmm_change_mac_delete_event": self.change_mac_close,
"on_change_mac_cancel_clicked": self.change_mac_close,
"on_change_mac_ok_clicked": self.change_mac_finish,
# Change storage dialog
"on_vmm_change_storage_delete_event": self.change_storage_close,
"on_change_storage_cancel_clicked": self.change_storage_close,
"on_change_storage_ok_clicked": self.change_storage_finish,
"on_change_storage_doclone_toggled": self.change_storage_doclone_toggled,
"on_change_storage_browse_clicked": self.change_storage_browse,
})
self.bind_escape_key_close()
self._cleanup_on_app_close()
self._init_ui()
@property
def conn(self):
if self.vm:
return self.vm.conn
return None
def show(self, parent, vm):
log.debug("Showing clone wizard")
self._set_vm(vm)
self.reset_state()
self.topwin.set_transient_for(parent)
self.topwin.resize(1, 1)
self.topwin.present()
def close(self, ignore1=None, ignore2=None):
log.debug("Closing clone wizard")
self.change_mac_close()
self.change_storage_close()
self.topwin.hide()
self._set_vm(None)
self.clone_design = None
self.storage_list = {}
self.target_list = []
self.net_list = {}
self.mac_list = []
return 1
def _vm_removed(self, _conn, connkey):
if self.vm.get_connkey() == connkey:
self.close()
def _set_vm(self, newvm):
oldvm = self.vm
if oldvm:
oldvm.conn.disconnect_by_obj(self)
if newvm:
newvm.conn.connect("vm-removed", self._vm_removed)
self.vm = newvm
def _cleanup(self):
self.change_mac.destroy()
self.change_mac = None
self.change_storage.destroy()
self.change_storage = None
if self.storage_browser:
self.storage_browser.cleanup()
self.storage_browser = None
def change_mac_close(self, ignore1=None, ignore2=None):
self.change_mac.hide()
return 1
def change_storage_close(self, ignore1=None, ignore2=None):
self.change_storage.hide()
return 1
# First time setup
def _init_ui(self):
context = self.topwin.get_style_context()
defcolor = context.get_background_color(Gtk.StateType.NORMAL)
self.widget("storage-viewport").override_background_color(
Gtk.StateType.NORMAL,
defcolor)
# Populate state
def reset_state(self):
self.widget("clone-cancel").grab_focus()
# Populate default clone values
self.setup_clone_info()
cd = self.clone_design
self.widget("clone-orig-name").set_text(cd.original_guest)
self.widget("clone-new-name").set_text(cd.clone_name)
uiutil.set_grid_row_visible(
self.widget("clone-dest-host"), self.conn.is_remote())
self.widget("clone-dest-host").set_text(self.conn.get_pretty_desc())
# We need to determine which disks fail (and why).
self.storage_list, self.target_list = self.check_all_storage()
self.populate_storage_lists()
self.populate_network_list()
return
def setup_clone_info(self):
self.clone_design = self.build_new_clone_design()
def build_new_clone_design(self, new_name=None):
design = Cloner(self.conn.get_backend())
design.original_guest = self.vm.get_name()
if not new_name:
new_name = design.generate_clone_name()
design.clone_name = new_name
# Erase any clone_policy from the original design, so that we
# get the entire device list.
design.clone_policy = []
return design
def populate_network_list(self):
net_box = self.widget("clone-network-box")
for c in net_box.get_children():
net_box.remove(c)
c.destroy()
self.net_list = {}
self.mac_list = []
def build_net_row(labelstr, origmac, newmac):
label = Gtk.Label(label=labelstr)
label.set_alignment(0, .5)
button = Gtk.Button(_("Details..."))
button.connect("clicked", self.net_change_mac, origmac)
hbox = Gtk.HBox()
hbox.set_spacing(12)
hbox.pack_start(label, True, True, 0)
hbox.pack_end(button, False, False, False)
hbox.show_all()
net_box.pack_start(hbox, False, False, False)
net_row = []
net_row.insert(NETWORK_INFO_LABEL, labelstr)
net_row.insert(NETWORK_INFO_ORIG_MAC, origmac)
net_row.insert(NETWORK_INFO_NEW_MAC, newmac)
self.net_list[origmac] = net_row
self.mac_list.append(origmac)
for net in self.vm.xmlobj.devices.interface:
mac = net.macaddr
net_dev = net.source
net_type = net.type
# Generate a new MAC
newmac = DeviceInterface.generate_mac(
self.conn.get_backend())
# [ interface type, device name, origmac, newmac, label ]
if net_type == DeviceInterface.TYPE_USER:
label = _("Usermode (%(mac)s)") % {"mac": mac}
elif net_type == DeviceInterface.TYPE_VIRTUAL:
net = None
for netobj in self.vm.conn.list_nets():
if netobj.get_name() == net_dev:
net = netobj
break
if net:
label = _("%(netmode)s (%(mac)s)") % {
"netmode": net.pretty_forward_mode(),
"mac": mac,
}
elif net_dev:
label = _("Virtual Network %(netdevice)s (%(mac)s)") % {
"netdevice": net_dev,
"mac": mac,
}
else:
label = _("Virtual Network (%(mac)s)") % {"mac": mac}
else:
# 'bridge' or anything else
pretty_net_type = net_type.capitalize()
if net_dev:
label = _("%(nettype)s %(netdevice)s (%(mac)s)") % {
"nettype": pretty_net_type,
"netdevice": net_dev,
"mac": mac,
}
else:
label = _("%(nettype)s (%(mac)s)") % {
"nettype": pretty_net_type,
"mac": mac,
}
build_net_row(label, mac, newmac)
no_net = (not list(self.net_list.keys()))
self.widget("clone-network-box").set_visible(not no_net)
self.widget("clone-no-net").set_visible(no_net)
def check_all_storage(self):
"""
Determine which storage is clonable, and which isn't
"""
diskinfos = self.vm.xmlobj.devices.disk
cd = self.clone_design
storage_list = {}
# We need to determine which disks fail (and why).
all_targets = [d.target for d in diskinfos]
for disk in diskinfos:
force_target = disk.target
path = disk.path
ro = disk.read_only
shared = disk.shareable
devtype = disk.device
size = None
clone_path = None
failinfo = ""
definfo = ""
storage_row = []
storage_row.insert(STORAGE_INFO_ORIG_PATH, path or "-")
storage_row.insert(STORAGE_INFO_NEW_PATH, clone_path)
storage_row.insert(STORAGE_INFO_TARGET, force_target)
storage_row.insert(STORAGE_INFO_SIZE, size)
storage_row.insert(STORAGE_INFO_DEVTYPE, devtype)
storage_row.insert(STORAGE_INFO_DO_CLONE, False)
storage_row.insert(STORAGE_INFO_CAN_CLONE, False)
storage_row.insert(STORAGE_INFO_CAN_SHARE, False)
storage_row.insert(STORAGE_INFO_DO_DEFAULT, False)
storage_row.insert(STORAGE_INFO_DEFINFO, definfo)
storage_row.insert(STORAGE_INFO_FAILINFO, failinfo)
storage_row.insert(STORAGE_INFO_COMBO, None)
storage_row.insert(STORAGE_INFO_MANUAL_PATH, False)
skip_targets = all_targets[:]
skip_targets.remove(force_target)
vol = self.conn.get_vol_by_path(path)
default, definfo, can_default = do_we_default(self.conn, vol, path,
ro, shared, devtype)
def storage_add(failinfo=None):
# pylint: disable=cell-var-from-loop
storage_row[STORAGE_INFO_DEFINFO] = definfo
storage_row[STORAGE_INFO_DO_DEFAULT] = default
storage_row[STORAGE_INFO_CAN_SHARE] = bool(definfo)
if failinfo:
storage_row[STORAGE_INFO_FAILINFO] = failinfo
storage_row[STORAGE_INFO_DO_CLONE] = False
storage_list[force_target] = storage_row
# If origdisk is empty, deliberately make it fail
if not path:
storage_add(_("Nothing to clone."))
continue
try:
cd.skip_target = skip_targets
cd.setup_original()
except Exception as e:
log.exception("Disk target '%s' caused clone error",
force_target)
storage_add(str(e))
continue
can_clone, cloneinfo = can_we_clone(self.conn, vol, path)
if not can_clone:
storage_add(cloneinfo)
continue
storage_row[STORAGE_INFO_CAN_CLONE] = True
# If we cannot create default clone_path don't even try to do that
if not can_default:
storage_add()
continue
try:
# Generate disk path, make sure that works
clone_path = self.generate_clone_path_name(path)
log.debug("Original path: %s\nGenerated clone path: %s",
path, clone_path)
cd.clone_paths = clone_path
size = cd.original_disks[0].get_size()
except Exception as e:
log.exception("Error setting generated path '%s'",
clone_path)
storage_add(str(e))
storage_row[STORAGE_INFO_NEW_PATH] = clone_path
storage_row[STORAGE_INFO_SIZE] = self.pretty_storage(size)
storage_add()
return storage_list, all_targets
def generate_clone_path_name(self, origpath, newname=None):
cd = self.clone_design
if not newname:
newname = cd.clone_name
clone_path = cd.generate_clone_disk_path(origpath,
newname=newname)
return clone_path
def set_paths_from_clone_name(self):
cd = self.clone_design
newname = self.widget("clone-new-name").get_text()
if not newname:
return
if cd.clone_name == newname:
return
for row in list(self.storage_list.values()):
origpath = row[STORAGE_INFO_ORIG_PATH]
if row[STORAGE_INFO_MANUAL_PATH]:
continue
if not row[STORAGE_INFO_DO_CLONE]:
return
try:
newpath = self.generate_clone_path_name(origpath, newname)
row[STORAGE_INFO_NEW_PATH] = newpath
except Exception as e:
log.debug("Generating new path from clone name failed: %s",
str(e))
def build_storage_entry(self, disk, storage_box):
origpath = disk[STORAGE_INFO_ORIG_PATH]
devtype = disk[STORAGE_INFO_DEVTYPE]
size = disk[STORAGE_INFO_SIZE]
can_clone = disk[STORAGE_INFO_CAN_CLONE]
do_clone = disk[STORAGE_INFO_DO_CLONE]
can_share = disk[STORAGE_INFO_CAN_SHARE]
is_default = disk[STORAGE_INFO_DO_DEFAULT]
definfo = disk[STORAGE_INFO_DEFINFO]
failinfo = disk[STORAGE_INFO_FAILINFO]
target = disk[STORAGE_INFO_TARGET]
orig_name = self.vm.get_name()
disk_label = os.path.basename(origpath)
info_label = None
if not can_clone:
info_label = Gtk.Label()
info_label.set_alignment(0, .5)
info_label.set_markup("<span size='small'>%s</span>" % failinfo)
info_label.set_line_wrap(True)
if not is_default:
disk_label += (definfo and " (%s)" % definfo or "")
# Build icon
icon = Gtk.Image()
if devtype == virtinst.DeviceDisk.DEVICE_FLOPPY:
iconname = "media-floppy"
elif devtype == virtinst.DeviceDisk.DEVICE_CDROM:
iconname = "media-optical"
else:
iconname = "drive-harddisk"
icon.set_from_icon_name(iconname, Gtk.IconSize.MENU)
disk_name_label = Gtk.Label(label=disk_label)
disk_name_label.set_alignment(0, .5)
disk_name_box = Gtk.HBox(spacing=9)
disk_name_box.pack_start(icon, False, False, 0)
disk_name_box.pack_start(disk_name_label, True, True, 0)
def sep_func(model, it, combo):
ignore = combo
return model[it][2]
# [String, sensitive, is sep]
model = Gtk.ListStore(str, bool, bool)
option_combo = Gtk.ComboBox()
option_combo.set_model(model)
text = Gtk.CellRendererText()
option_combo.pack_start(text, True)
option_combo.add_attribute(text, "text", 0)
option_combo.add_attribute(text, "sensitive", 1)
option_combo.set_row_separator_func(sep_func, option_combo)
option_combo.connect("changed", self.storage_combo_changed, target)
vbox = Gtk.VBox(spacing=1)
if can_clone or can_share:
model.insert(STORAGE_COMBO_CLONE,
[(_("Clone this disk") +
(size and " (%s)" % size or "")),
can_clone, False])
model.insert(STORAGE_COMBO_SHARE,
[_("Share disk with %s") % orig_name, can_share,
False])
model.insert(STORAGE_COMBO_SEP, ["", False, True])
model.insert(STORAGE_COMBO_DETAILS,
[_("Details..."), True, False])
if (can_clone and is_default) or do_clone:
option_combo.set_active(STORAGE_COMBO_CLONE)
else:
option_combo.set_active(STORAGE_COMBO_SHARE)
else:
model.insert(STORAGE_COMBO_CLONE,
[_("Storage cannot be shared or cloned."),
False, False])
option_combo.set_active(STORAGE_COMBO_CLONE)
vbox.pack_start(disk_name_box, False, False, 0)
vbox.pack_start(option_combo, False, False, 0)
if info_label:
vbox.pack_start(info_label, False, False, 0)
storage_box.pack_start(vbox, False, False, 0)
disk[STORAGE_INFO_COMBO] = option_combo
def populate_storage_lists(self):
storage_box = self.widget("clone-storage-box")
for c in storage_box.get_children():
storage_box.remove(c)
c.destroy()
for target in self.target_list:
disk = self.storage_list[target]
self.build_storage_entry(disk, storage_box)
num_c = min(len(self.target_list), 3)
if num_c:
scroll = self.widget("clone-storage-scroll")
scroll.set_size_request(-1, 80 * num_c)
storage_box.show_all()
no_storage = not bool(len(self.target_list))
self.widget("clone-storage-box").set_visible(not no_storage)
self.widget("clone-no-storage-pass").set_visible(no_storage)
skip_targets = []
new_disks = []
for target in self.target_list:
do_clone = self.storage_list[target][STORAGE_INFO_DO_CLONE]
new_path = self.storage_list[target][STORAGE_INFO_NEW_PATH]
if do_clone:
new_disks.append(new_path)
else:
skip_targets.append(target)
self.clone_design.skip_target = skip_targets
try:
self.clone_design.clone_paths = new_disks
except Exception as e:
# Just log the error and go on. The UI will fail later if needed
log.debug("Error setting clone_paths: %s", str(e))
# If any storage cannot be cloned or shared, don't allow cloning
clone = True
tooltip = ""
for row in list(self.storage_list.values()):
can_clone = row[STORAGE_INFO_CAN_CLONE]
can_share = row[STORAGE_INFO_CAN_SHARE]
if not (can_clone or can_share):
clone = False
tooltip = _("One or more disks cannot be cloned or shared.")
break
ok_button = self.widget("clone-ok")
ok_button.set_sensitive(clone)
ok_button.set_tooltip_text(tooltip)
def net_change_mac(self, ignore, origmac):
row = self.net_list[origmac]
orig_mac = row[NETWORK_INFO_ORIG_MAC]
new_mac = row[NETWORK_INFO_NEW_MAC]
typ = row[NETWORK_INFO_LABEL]
self.widget("change-mac-orig").set_text(orig_mac)
self.widget("change-mac-type").set_text(typ)
self.widget("change-mac-new").set_text(new_mac)
self.change_mac.show_all()
def storage_combo_changed(self, src, target):
idx = src.get_active()
row = self.storage_list[target]
if idx == STORAGE_COMBO_CLONE:
row[STORAGE_INFO_DO_CLONE] = True
return
elif idx == STORAGE_COMBO_SHARE:
row[STORAGE_INFO_DO_CLONE] = False
return
elif idx != STORAGE_COMBO_DETAILS:
return
do_clone = row[STORAGE_INFO_DO_CLONE]
if do_clone:
src.set_active(STORAGE_COMBO_CLONE)
else:
src.set_active(STORAGE_COMBO_SHARE)
# Show storage
self.storage_change_path(row)
def change_storage_doclone_toggled(self, src):
do_clone = src.get_active()
self.widget("change-storage-new").set_sensitive(do_clone)
self.widget("change-storage-browse").set_sensitive(do_clone)
def storage_change_path(self, row):
# If storage paths are dependent on manually entered clone name,
# make sure they are up to date
self.set_paths_from_clone_name()
orig = row[STORAGE_INFO_ORIG_PATH]
new = row[STORAGE_INFO_NEW_PATH]
tgt = row[STORAGE_INFO_TARGET]
size = row[STORAGE_INFO_SIZE]
can_clone = row[STORAGE_INFO_CAN_CLONE]
can_share = row[STORAGE_INFO_CAN_SHARE]
do_clone = row[STORAGE_INFO_DO_CLONE]
self.widget("change-storage-doclone").set_active(True)
self.widget("change-storage-doclone").toggled()
self.widget("change-storage-orig").set_text(orig)
self.widget("change-storage-target").set_text(tgt)
self.widget("change-storage-size").set_text(size or "-")
self.widget("change-storage-doclone").set_active(do_clone)
if can_clone:
self.widget("change-storage-new").set_text(new or "")
else:
self.widget("change-storage-new").set_text("")
self.widget("change-storage-doclone").set_sensitive(can_clone and
can_share)
self.widget("vmm-change-storage").show_all()
def change_mac_finish(self, ignore):
orig = self.widget("change-mac-orig").get_text()
new = self.widget("change-mac-new").get_text()
row = self.net_list[orig]
try:
DeviceInterface.check_mac_in_use(self.conn.get_backend(), new)
row[NETWORK_INFO_NEW_MAC] = new
except Exception as e:
self.err.show_err(_("Error changing MAC address: %s") % str(e))
return
self.change_mac_close()
def change_storage_finish(self, ignore):
target = self.widget("change-storage-target").get_text()
row = self.storage_list[target]
# Sync 'do clone' checkbox, and main dialog combo
combo = row[STORAGE_INFO_COMBO]
do_clone = self.widget("change-storage-doclone").get_active()
if do_clone:
combo.set_active(STORAGE_COMBO_CLONE)
else:
combo.set_active(STORAGE_COMBO_SHARE)
row[STORAGE_INFO_DO_CLONE] = do_clone
if not do_clone:
self.change_storage_close()
return
new_path = self.widget("change-storage-new").get_text()
if virtinst.DeviceDisk.path_definitely_exists(self.clone_design.conn,
new_path):
res = self.err.yes_no(_("Cloning will overwrite the existing "
"file"),
_("Using an existing image will overwrite "
"the path during the clone process. Are "
"you sure you want to use this path?"))
if not res:
return
try:
self.clone_design.clone_paths = new_path
row[STORAGE_INFO_NEW_PATH] = new_path
row[STORAGE_INFO_MANUAL_PATH] = True
self.populate_storage_lists()
except Exception as e:
self.err.show_err(_("Error changing storage path: %s") % str(e))
return
self.change_storage_close()
def pretty_storage(self, size):
if not size:
return ""
return "%.1f GiB" % float(size)
# Listeners
def validate(self):
self.set_paths_from_clone_name()
name = self.widget("clone-new-name").get_text()
# Make another clone_design
cd = self.build_new_clone_design(name)
# Set MAC addresses
clonemacs = []
for mac in self.mac_list:
row = self.net_list[mac]
clonemacs.append(row[NETWORK_INFO_NEW_MAC])
cd.clone_macs = clonemacs
skip_targets = []
new_paths = []
warn_str = ""
for target in self.target_list:
path = self.storage_list[target][STORAGE_INFO_ORIG_PATH]
new_path = self.storage_list[target][STORAGE_INFO_NEW_PATH]
do_clone = self.storage_list[target][STORAGE_INFO_DO_CLONE]
do_default = self.storage_list[target][STORAGE_INFO_DO_DEFAULT]
if do_clone:
new_paths.append(new_path)
else:
skip_targets.append(target)
if not path or path == '-':
continue
if not do_default:
continue
warn_str += "%s: %s\n" % (target, path)
cd.skip_target = skip_targets
cd.setup_original()
cd.clone_paths = new_paths
if warn_str:
res = self.err.ok_cancel(
_("Skipping disks may cause data to be overwritten."),
_("The following disk devices will not be cloned:\n\n%s\n"
"Running the new guest could overwrite data in these "
"disk images.")
% warn_str)
if not res:
return False
cd.setup_clone()
self.clone_design = cd
return True
def _finish_cb(self, error, details, conn):
self.reset_finish_cursor()
if error is not None:
error = (_("Error creating virtual machine clone '%(vm)s': "
"%(error)s") % {
"vm": self.clone_design.clone_name,
"error": error,
})
self.err.show_err(error, details=details)
return
conn.schedule_priority_tick(pollvm=True)
self.close()
def finish(self, src_ignore):
try:
if not self.validate():
return
except Exception as e:
self.err.show_err(_("Uncaught error validating input: %s") % str(e))
return
self.set_finish_cursor()
title = (_("Creating virtual machine clone '%s'") %
self.clone_design.clone_name)
text = title
if self.clone_design.clone_disks:
text = (_("Creating virtual machine clone '%s' and selected "
"storage (this may take a while)") %
self.clone_design.clone_name)
progWin = vmmAsyncJob(self._async_clone, [],
self._finish_cb, [self.conn],
title, text, self.topwin)
progWin.run()
def _async_clone(self, asyncjob):
meter = asyncjob.get_meter()
refresh_pools = []
for disk in self.clone_design.clone_disks:
if not disk.wants_storage_creation():
continue
pool = disk.get_parent_pool()
if not pool:
continue
poolname = pool.name()
if poolname not in refresh_pools:
refresh_pools.append(poolname)
self.clone_design.start_duplicate(meter)
for poolname in refresh_pools:
try:
pool = self.conn.get_pool(poolname)
self.idle_add(pool.refresh)
except Exception:
log.debug("Error looking up pool=%s for refresh after "
"VM clone.", poolname, exc_info=True)
def change_storage_browse(self, ignore):
def callback(src_ignore, txt):
self.widget("change-storage-new").set_text(txt)
if self.storage_browser and self.storage_browser.conn != self.conn:
self.storage_browser.cleanup()
self.storage_browser = None
if self.storage_browser is None:
self.storage_browser = vmmStorageBrowser(self.conn)
self.storage_browser.set_finish_cb(callback)
self.storage_browser.show(self.topwin)
Reference in New Issue View Git Blame Copy Permalink