# Copyright (C) 2009, 2013 Red Hat, Inc. # Copyright (C) 2009 Cole Robinson # # This work is licensed under the GNU GPLv2 or later. # See the COPYING file in the top-level directory. import collections from gi.repository import Gtk from gi.repository import Pango import virtinst from virtinst import Cloner from virtinst import log from virtinst import xmlutil from .lib import uiutil from .baseclass import vmmGObjectUI from .asyncjob import vmmAsyncJob from .storagebrowse import vmmStorageBrowser def _get_cloneable_msg(diskinfo): """Is the passed path even clone-able""" if diskinfo.get_cloneable_msg(): return diskinfo.get_cloneable_msg() if diskinfo.disk.is_empty(): return _("No storage to clone.") class _StorageInfo: """ Our wrapper class that is close to Cloner _DiskInfo content. We track all user choices explicitly in this class, and then serialize them into the actual cloner diskinfo once the user clicks 'Clone' to complete the process. """ def __init__(self, vm, cloner, diskinfo): self._diskinfo = diskinfo self._orig_disk = diskinfo.disk self._orig_vm_name = vm.get_name() self._new_vm_name = cloner.new_guest.name self.share_msg = diskinfo.get_share_msg() self.cloneable_msg = _get_cloneable_msg(diskinfo) self._manual_path = None self._generated_path = (diskinfo.new_disk and diskinfo.new_disk.get_source_path()) if not self._generated_path: self._reset_generated_path() self._is_clone_requested = diskinfo.is_clone_requested() if self.share_msg or not self.is_cloneable(): self._is_clone_requested = False def is_cloneable(self): return not bool(self.cloneable_msg) def is_clone_requested(self): return self._is_clone_requested def is_share_requested(self): return not self._is_clone_requested def warn_about_sharing(self): return not self.share_msg def get_target(self): return self._orig_disk.target def get_orig_disk_path(self): return self._orig_disk.get_source_path() def get_new_disk_path(self): if self._manual_path: return self._manual_path return self._generated_path def set_clone_requested(self, val): self._is_clone_requested = bool(val) def set_manual_path(self, val): self._manual_path = val def set_new_vm_name(self, val): if not val or val == self._new_vm_name: return self._new_vm_name = val self._reset_generated_path() def _reset_generated_path(self): self._generated_path = Cloner.generate_clone_disk_path( self._orig_disk.conn, self._orig_vm_name, self._new_vm_name, self._orig_disk.get_source_path()) def set_values_on_diskinfo(self, diskinfo): if not self._is_clone_requested: diskinfo.set_share_requested() diskinfo.new_disk = None return diskinfo.set_clone_requested() sparse = True newpath = self.get_new_disk_path() diskinfo.set_new_path(newpath, sparse) ################### # UI info helpers # ################### def get_tooltip(self): lines = [] lines.append(_("Disk target: %s") % self.get_target()) lines.append(_("Original path: %s") % self.get_orig_disk_path()) if self.get_new_disk_path(): lines.append(_("New path: %s") % self.get_new_disk_path()) lines.append("\n") if self.share_msg: lines.append(_("Storage is safe to share: %(reason)s") % { "reason": self.share_msg}) else: lines.append( _("Sharing this storage is potentially dangerous.")) if self.cloneable_msg: lines.append(_("Storage is not cloneable: %(reason)s") % { "reason": self.cloneable_msg}) return "\n".join(lines) def get_markup(self, vm): lines = [] line = "" path = self.get_orig_disk_path() if path: line += path else: line += _("No storage.") lines.append(line) line = "" if self.is_share_requested(): line += _("Share disk with %s") % vm.get_name() if self.is_clone_requested(): line += _("Clone this disk") sizelabel = self.get_size_label() if sizelabel: line += " (%s)" % sizelabel if line: lines.append(line) label = "\n".join(lines) markup = "%s" % xmlutil.xml_escape(label) return markup def get_icon_name(self): if self._orig_disk.is_floppy(): return "media-floppy" if self._orig_disk.is_cdrom(): return "media-optical" return "drive-harddisk" def get_size_label(self): size = self._orig_disk.get_size() if not size: return "" # pragma: no cover return "%.1f GiB" % float(size) 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._storage_list = None self._storage_browser = None self._storage_dialog = self.widget("vmm-change-storage") self._storage_dialog.set_transient_for(self.topwin) self.builder.connect_signals({ "on_clone_delete_event": self._close_cb, "on_clone_cancel_clicked": self._close_cb, "on_clone_ok_clicked": self._finish_clicked_cb, "on_storage_selection_changed": self._storage_selection_changed_cb, "on_storage_details_clicked": self._storage_details_clicked_cb, # Storage subdialog signals "on_vmm_change_storage_delete_event": self._storage_dialog_close_cb, "on_change_storage_cancel_clicked": self._storage_dialog_close_cb, "on_change_storage_ok_clicked": self._storage_dialog_finish_cb, "on_change_storage_doclone_toggled": self._storage_dialog_doclone_toggled_cb, "on_change_storage_browse_clicked": self._storage_dialog_browse_cb, }) self.bind_escape_key_close() self._cleanup_on_app_close() self._init_ui() ####################### # Standard UI methods # ####################### @property def conn(self): return self.vm and self.vm.conn or 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 _storage_dialog_close(self): self._storage_dialog.hide() return 1 def close(self, ignore1=None, ignore2=None): log.debug("Closing clone wizard") self._storage_dialog_close() self.topwin.hide() self._set_vm(None) self._storage_list = None return 1 def _vm_removed_cb(self, _conn, vm): if self.vm == vm: 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_cb) self.vm = newvm def _cleanup(self): self._storage_dialog.destroy() self._storage_dialog = None if self._storage_browser: self._storage_browser.cleanup() self._storage_browser = None ########### # UI init # ########### def _init_ui(self): storage_list = self.widget("storage-list") # [disk target, tooltip] model = Gtk.ListStore(str, str) storage_list.set_model(model) storage_list.set_tooltip_column(1) cloneCol = Gtk.TreeViewColumn(_("Clone")) pathCol = Gtk.TreeViewColumn(_("Storage")) pathCol.set_sizing(Gtk.TreeViewColumnSizing.AUTOSIZE) pathCol.set_expand(True) storage_list.append_column(cloneCol) storage_list.append_column(pathCol) def separator_cb(_model, _iter): return _model[_iter][0] == "separator" storage_list.set_row_separator_func(separator_cb) chkbox = Gtk.CellRendererToggle() chkbox.connect('toggled', self._storage_clone_toggled_cb) chkimg = Gtk.CellRendererPixbuf() chkimg.set_property('stock-size', Gtk.IconSize.MENU) cloneCol.pack_start(chkimg, False) cloneCol.pack_start(chkbox, False) def chk_cell_data_cb(column, cell, model, _iter, data): target = model[_iter][0] if target == "separator": return sinfo = self._storage_list[target] visible = sinfo.is_cloneable() active = sinfo.is_clone_requested() _chkimg = column.get_cells()[0] _chkbox = column.get_cells()[1] _chkbox.set_property('active', active) _chkbox.set_property('visible', visible) _chkimg.set_property('visible', not visible) icon = Gtk.STOCK_INFO if sinfo.warn_about_sharing(): icon = Gtk.STOCK_DIALOG_WARNING _chkimg.set_property('stock-id', icon) tooltip = sinfo.get_tooltip() if tooltip != model[_iter][1]: model[_iter][1] = tooltip cloneCol.set_cell_data_func(chkbox, chk_cell_data_cb) cloneCol.set_cell_data_func(chkimg, chk_cell_data_cb) pathtxt = Gtk.CellRendererText() pathCol.set_sort_column_id(0) pathtxt.set_property("width-chars", 30) pathtxt.set_property("ellipsize", Pango.EllipsizeMode.MIDDLE) pathimg = Gtk.CellRendererPixbuf() pathimg.set_property('stock-size', Gtk.IconSize.MENU) pathimg.set_padding(3, 0) pathCol.pack_start(pathimg, False) pathCol.pack_start(pathtxt, True) def path_cb(column, cell, model, _iter, data): target = model[_iter][0] if target == "separator": return sinfo = self._storage_list[target] _pathimg = column.get_cells()[0] _pathtxt = column.get_cells()[1] markup = sinfo.get_markup(self.vm) _pathtxt.set_property('markup', markup) _pathimg.set_property('icon-name', sinfo.get_icon_name()) pathCol.set_cell_data_func(pathtxt, path_cb) pathCol.set_cell_data_func(pathimg, path_cb) def _reset_state(self): self.widget("clone-cancel").grab_focus() self.widget("clone-new-name").set_text("") # Populate default clone values cloner = self._build_cloner() cloner.prepare() self.widget("clone-orig-name").set_text(cloner.src_name) self.widget("clone-new-name").set_text(cloner.new_guest.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()) # Collect info about all the VMs disks self._storage_list = collections.OrderedDict() for diskinfo in cloner.get_diskinfos(): sinfo = _StorageInfo(self.vm, cloner, diskinfo) self._storage_list[sinfo.get_target()] = sinfo self._populate_storage_ui() ####################### # Functional routines # ####################### def _build_cloner(self): conn = self.conn.get_backend() orig_name = self.vm.get_name() new_name = self.widget("clone-new-name").get_text() cloner = Cloner(conn, src_name=orig_name) if new_name: cloner.set_clone_name(new_name) return cloner ####################### # Storage UI building # ####################### def _set_paths_from_clone_name(self): newname = self.widget("clone-new-name").get_text() for sinfo in list(self._storage_list.values()): sinfo.set_new_vm_name(newname) def _populate_storage_ui(self): """ Fill in the storage UI from the collected StorageInfo classes """ model = self.widget("storage-list").get_model() model.clear() for sinfo in self._storage_list.values(): model.append([sinfo.get_target(), sinfo.get_tooltip()]) model.append(["separator", None]) def _storage_clone_toggled_cb(self, src, treepath): model = self.widget("storage-list").get_model() row = model[treepath] target = row[0] sinfo = self._storage_list[target] active = not src.get_property("active") sinfo.set_clone_requested(active) model.emit("row-changed", row.path, row.iter) ########################### # Storage window handling # ########################### def _show_storage_window(self): tgt = uiutil.get_list_selection(self.widget("storage-list")) sinfo = self._storage_list[tgt] # If storage paths are dependent on manually entered clone name, # make sure they are up to date self._set_paths_from_clone_name() orig = sinfo.get_orig_disk_path() new = sinfo.get_new_disk_path() size = sinfo.get_size_label() can_clone = sinfo.is_cloneable() do_clone = sinfo.is_clone_requested() 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) self.widget("change-storage-new").set_text(new or "") self.widget("change-storage-doclone").set_sensitive(can_clone) self.widget("vmm-change-storage").show_all() def _storage_dialog_finish(self): target = self.widget("change-storage-target").get_text() sinfo = self._storage_list[target] # Sync 'do clone' checkbox, and main dialog combo do_clone = self.widget("change-storage-doclone").get_active() sinfo.set_clone_requested(do_clone) if not do_clone: self._storage_dialog_close() return new_path = self.widget("change-storage-new").get_text() if virtinst.DeviceDisk.path_definitely_exists( self.vm.conn.get_backend(), new_path): text1 = _("Cloning will overwrite the existing file") text2 = _("Using an existing image will overwrite " "the path during the clone process. Are " "you sure you want to use this path?") res = self.err.yes_no(text1, text2) if not res: return sinfo.set_manual_path(new_path) self._storage_dialog_close() ################### # Finish routines # ################### def _validate(self, cloner): new_paths = [] warn_str = "" for sinfo in self._storage_list.values(): if sinfo.is_clone_requested(): new_paths.append(sinfo.get_new_disk_path()) continue if not sinfo.warn_about_sharing(): continue warn_str += "%s: %s\n" % ( sinfo.get_target(), sinfo.get_orig_disk_path()) if warn_str: res = self.err.ok_cancel( _("Sharing storage may cause data to be overwritten."), _("The following disk devices will be shared with %(vmname)s:" "\n\n%(pathlist)s\n" "Running the new guest could overwrite data in these " "disk images.") % { "vmname": self.vm.get_name(), "pathlist": warn_str}) if not res: return False for diskinfo in cloner.get_diskinfos(): diskinfo.raise_error() def _finish_cb(self, error, details, conn, cloner): self.reset_finish_cursor() if error is not None: error = (_("Error creating virtual machine clone '%(vm)s': " "%(error)s") % { "vm": cloner.new_guest.name, "error": error, }) self.err.show_err(error, details=details) return conn.schedule_priority_tick(pollvm=True) self.close() def _async_clone(self, asyncjob, cloner): meter = asyncjob.get_meter() refresh_pools = [] for diskinfo in cloner.get_nonshare_diskinfos(): disk = diskinfo.new_disk pool = disk.get_parent_pool() if not pool: continue poolname = pool.name() if poolname not in refresh_pools: refresh_pools.append(poolname) cloner.start_duplicate(meter) for poolname in refresh_pools: try: pool = self.conn.get_pool_by_name(poolname) self.idle_add(pool.refresh) except Exception: # pragma: no cover log.debug("Error looking up pool=%s for refresh after " "VM clone.", poolname, exc_info=True) def _build_final_cloner(self): self._set_paths_from_clone_name() cloner = self._build_cloner() for diskinfo in cloner.get_diskinfos(): target = diskinfo.disk.target sinfo = self._storage_list[target] sinfo.set_values_on_diskinfo(diskinfo) cloner.prepare() for diskinfo in cloner.get_diskinfos(): diskinfo.raise_error() if self._validate(cloner) is False: return return cloner def _finish(self): try: cloner = self._build_final_cloner() if not cloner: return except Exception as e: msg = _("Error with clone settings: %s") % str(e) return self.err.show_err(msg) self.set_finish_cursor() title = (_("Creating virtual machine clone '%s'") % cloner.new_guest.name) text = title if cloner.get_nonshare_diskinfos(): text = (_("Creating virtual machine clone '%s' and selected " "storage (this may take a while)") % cloner.new_guest.name) progWin = vmmAsyncJob(self._async_clone, [cloner], self._finish_cb, [self.conn, cloner], title, text, self.topwin) progWin.run() ################ # UI listeners # ################ def _close_cb(self, src, event=None): return self.close() def _finish_clicked_cb(self, src): return self._finish() def _storage_selection_changed_cb(self, src): row = uiutil.get_list_selected_row(self.widget("storage-list")) self.widget("storage-details-button").set_sensitive(bool(row)) def _storage_details_clicked_cb(self, src): self._show_storage_window() def _storage_dialog_doclone_toggled_cb(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_dialog_close_cb(self, src, event=None): return self._storage_dialog_close() def _storage_dialog_finish_cb(self, src): self._storage_dialog_finish() def _storage_dialog_browse_cb(self, ignore): def callback(src_ignore, txt): self.widget("change-storage-new").set_text(txt) 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)