diff --git a/UpdateManager/backend/__init__.py b/UpdateManager/backend/__init__.py index 72a60318..1cb398fc 100644 --- a/UpdateManager/backend/__init__.py +++ b/UpdateManager/backend/__init__.py @@ -7,8 +7,7 @@ import gi gi.require_version("Gtk", "3.0") -gi.require_version("Snapd", "2") -from gi.repository import GLib, Gtk, Snapd +from gi.repository import GLib, Gtk from apt import Cache import json @@ -109,314 +108,6 @@ class InstallBackend(Dialog): """Commit the cache changes""" raise NotImplementedError - def get_snap_seeds(self): - seeded_snaps = {} - unseeded_snaps = {} - - curr_channel = "stable/ubuntu-" + get_dist_version() - cache = self.window_main.cache - - try: - d2s_file = open( - "/usr/share/ubuntu-release-upgrader/deb2snap.json", "r" - ) - d2s = json.load(d2s_file) - d2s_file.close() - - for snap in d2s["seeded"]: - seed = d2s["seeded"][snap] - deb = seed.get("deb", None) - to_channel = seed.get("to_channel", curr_channel) - metapkg = seed.get("metapkg", None) - if metapkg not in cache: - continue - if metapkg and cache[metapkg].is_installed is False: - continue - seeded_snaps[snap] = (deb, to_channel) - - for snap in d2s["unseeded"]: - unseed = d2s["unseeded"][snap] - from_channel = unseed.get("from_channel", curr_channel) - metapkg = seed.get("metapkg", None) - if metapkg not in cache: - continue - if metapkg and cache[metapkg].is_installed is False: - continue - unseeded_snaps[snap] = from_channel - except Exception as e: - logging.debug("error reading deb2snap.json file (%s)" % e) - - return seeded_snaps, unseeded_snaps - - def get_deb2snap_dups(self): - # update and grab the latest cache - try: - if self.window_main.cache is None: - self.window_main.cache = MyCache(None) - else: - self.window_main.cache.open(None) - self.window_main.cache._initDepCache() - cache = self.window_main.cache - except Exception as e: - # just return an empty array for now, it's perfectly safe to - # postpone this duplicates check to a later update. - logging.debug("error reading cache (%s)" % e) - return [] - - duplicates = [] - seeded_snaps, _ = self.get_snap_seeds() - - for snap, (deb, _) in seeded_snaps.items(): - # if the deb is installed and was not manually installed, - # replace it - if deb in cache and cache[deb].is_installed: - deb_is_auto = True - cache[deb].mark_delete() - - for pkg in cache.get_changes(): - if ( - pkg.is_installed - and pkg.marked_delete - and not pkg.is_auto_installed - ): - deb_is_auto = False - break - - cache.clear() - - if deb_is_auto: - duplicates.append(deb) - - return duplicates - - def get_snap_transitions(self): - # populate snap_list with deb2snap transitions - snap_list = {} - seeded_snaps, unseeded_snaps = self.get_snap_seeds() - - for snap, (deb, to_channel) in seeded_snaps.items(): - snap_object = {} - # check if the snap is already installed - snap_info = subprocess.Popen( - ["snap", "info", snap], - universal_newlines=True, - stdout=subprocess.PIPE, - ).communicate() - if re.search("^installed: ", snap_info[0], re.MULTILINE): - logging.debug("Snap %s is installed" % snap) - continue - elif deb in self.window_main.duplicate_packages: - # install the snap if the deb was just marked delete - snap_object["command"] = "install" - snap_object["channel"] = to_channel - snap_list[snap] = snap_object - - for snap, (from_channel) in unseeded_snaps.items(): - snap_object = {} - # check if the snap is already installed - snap_info = subprocess.Popen( - ["snap", "info", snap], - universal_newlines=True, - stdout=subprocess.PIPE, - ).communicate() - if re.search("^installed: ", snap_info[0], re.MULTILINE): - logging.debug("Snap %s is installed" % snap) - # its not tracking the release channel so don't remove - re_channel = "stable/ubuntu-[0-9][0-9].[0-9][0-9]" - if not re.search( - r"^tracking:.*%s" % re_channel, snap_info[0], re.MULTILINE - ): - logging.debug( - "Snap %s is not tracking the release channel" % snap - ) - continue - - snap_object["command"] = "remove" - - # check if this snap is being used by any other snaps - conns = subprocess.Popen( - ["snap", "connections", snap], - universal_newlines=True, - stdout=subprocess.PIPE, - ).communicate() - - for conn in conns[0].split("\n"): - conn_cols = conn.split() - if len(conn_cols) != 4: - continue - plug = conn_cols[1] - slot = conn_cols[2] - - if slot.startswith(snap + ":"): - plug_snap = plug.split(":")[0] - if ( - plug_snap != "-" - and plug_snap not in unseeded_snaps - ): - logging.debug( - "Snap %s is being used by %s. " - "Switching it to stable track" - % (snap, plug_snap) - ) - snap_object["command"] = "refresh" - snap_object["channel"] = "stable" - break - - snap_list[snap] = snap_object - - return snap_list - - def update_snap_cb(self, client, change, _, user_data): - index, count, progress_bar = user_data - if not progress_bar: - return - - # determine how much of this change has been done - task_total = 0 - task_done = 0 - for task in change.get_tasks(): - task_total += task.get_progress_total() - task_done += task.get_progress_done() - - task_fraction = task_done / task_total - - # determine how much total progress has been made - total_fraction = (task_fraction / count) + (index / count) - - # change.get_tasks() can increase between callbacks so we must - # avoid jumping backward in progress here - if total_fraction > progress_bar.get_fraction(): - GLib.idle_add(progress_bar.set_fraction, total_fraction) - - def update_snaps(self): - # update status and progress bar - def update_status(status): - GLib.idle_add(self.label_details.set_label, status) - - def update_progress(progress_bar): - progress_bar.pulse() - return True - - update_status(_("Updating snaps")) - - progress_bar = None - progress_timer = None - - progress_bars = self.progressbar_slot.get_children() - if progress_bars and isinstance(progress_bars[0], Gtk.ProgressBar): - progress_bar = progress_bars[0] - progress_timer = GLib.timeout_add( - 100, update_progress, progress_bar - ) - - # populate snap_list with deb2snap transitions - snap_list = self.get_snap_transitions() - - if progress_timer: - GLib.source_remove(progress_timer) - progress_bar.set_fraction(0) - - # (un)install (un)seeded snap(s) - try: - client = Snapd.Client() - client.connect_sync() - index = 0 - count = len(snap_list) - for snap, snap_object in snap_list.items(): - command = snap_object["command"] - if command == "refresh": - update_status(_("Refreshing %s snap" % snap)) - client.refresh_sync( - snap, - snap_object["channel"], - self.update_snap_cb, - progress_callback_data=(index, count, progress_bar), - ) - elif command == "remove": - update_status(_("Removing %s snap" % snap)) - client.remove_sync( - snap, - self.update_snap_cb, - progress_callback_data=(index, count, progress_bar), - ) - else: - update_status(_("Installing %s snap" % snap)) - client.install_sync( - snap, - snap_object["channel"], - self.update_snap_cb, - progress_callback_data=(index, count, progress_bar), - ) - index += 1 - except GLib.Error as e: - logging.debug("error updating snaps (%s)" % e) - GLib.idle_add( - self.window_main.start_error, - False, - _("Upgrade only partially completed."), - _( - "An error occurred while updating snaps. " - "Please check your network connection." - ), - ) - return - - # continue with the rest of the updates - GLib.idle_add(self.window_main.start_available) - - def _action_done( - self, - action, - authorized, - success, - error_string, - error_desc, - trans_failed=False, - ): - - # If the progress dialog should be closed automatically afterwards - # settings = Gio.Settings.new("com.ubuntu.update-manager") - # close_after_install = settings.get_boolean( - # "autoclose-install-window") - # FIXME: confirm with mpt whether this should still be a setting - # close_after_install = False - - if action == self.ACTION_PRE_INSTALL and success: - # Now do the regular updates - self.action = self.ACTION_INSTALL - self.start() - elif action == self.ACTION_INSTALL: - if ( - success - and os.path.exists("/usr/bin/snap") - and hasattr(self, "pane_update_progress") - ): - Thread(target=self.update_snaps).start() - elif success: - self.window_main.start_available() - elif error_string: - self.window_main.start_error( - trans_failed, error_string, error_desc - ) - else: - # exit gracefuly, we can't just exit as this will trigger - # a crash if system.exit() is called in a exception handler - GLib.timeout_add(1, self.window_main.exit) - else: - if error_string: - self.window_main.start_error(True, error_string, error_desc) - elif ( - success - and os.path.exists("/usr/bin/snap") - and hasattr(self, "pane_update_progress") - ): - self.window_main.duplicate_packages = self.get_deb2snap_dups() - self.window_main.start_available() - else: - is_cancelled_update = not success - self.window_main.start_available(is_cancelled_update) - - # try aptdaemon if ( os.path.exists("/usr/sbin/aptd")