From 79b6d8f72c7f88de638dcac56c99c20391a6242f Mon Sep 17 00:00:00 2001 From: WillyPillow Date: Wed, 19 Aug 2020 02:02:05 +0800 Subject: [PATCH] qvm-template: Initial GUI implementation. --- qubesmanager/qvm_template_gui.py | 405 +++++++++++++++++++++++++++++++ 1 file changed, 405 insertions(+) create mode 100644 qubesmanager/qvm_template_gui.py diff --git a/qubesmanager/qvm_template_gui.py b/qubesmanager/qvm_template_gui.py new file mode 100644 index 0000000..4674832 --- /dev/null +++ b/qubesmanager/qvm_template_gui.py @@ -0,0 +1,405 @@ +import collections +import concurrent +import concurrent.futures +import itertools +import json +import os +import subprocess +import typing + +import gi +gi.require_version('Gtk', '3.0') + +#pylint: disable=wrong-import-position +from gi.repository import GLib +from gi.repository import Gtk +from gi.repository import Pango + +BASE_CMD = ['qvm-template', '--enablerepo=*', '--yes', '--quiet'] + +class Template(typing.NamedTuple): + status: str + name: str + evr: str + reponame: str + size: int + buildtime: str + installtime: str + licence: str + url: str + summary: str + # --- internal --- + description: str + default_status: str + weight: int + model: Gtk.TreeModel + # ---------------- + + # XXX: Is there a better way of doing this? + TYPES = [str, str, str, str, int, str, str, str, + str, str, str, str, int, Gtk.TreeModel] + + COL_NAMES = [ + 'Status', + 'Name', + 'Version', + 'Reponame', + 'Size (kB)', + 'Build Time', + 'Install Time', + 'License', + 'URL', + 'Summary'] + + @staticmethod + def build(status, entry, model): + return Template( + status, + entry['name'], + '%s:%s-%s' % (entry['epoch'], entry['version'], entry['release']), + entry['reponame'], + # XXX: This may overflow glib ints, though pretty unlikely in the + # foreseeable future + int(entry['size']) / 1000, + entry['buildtime'], + entry['installtime'], + entry['license'], + entry['url'], + entry['summary'], + entry['description'], + status, + Pango.Weight.BOOK, + model + ) + +class Action(typing.NamedTuple): + op: str + name: str + evr: str + + TYPES = [str, str, str] + COL_NAMES = ['Operation', 'Name', 'Version'] + +# TODO: Set default window sizes + +class ConfirmDialog(Gtk.Dialog): + def __init__(self, parent, actions): + super(ConfirmDialog, self).__init__( + title='Confirmation', transient_for=parent, modal=True) + self.add_buttons( + Gtk.STOCK_CANCEL, Gtk.ResponseType.CANCEL, + Gtk.STOCK_OK, Gtk.ResponseType.OK) + + box = self.get_content_area() + self.msg = Gtk.Label() + self.msg.set_markup(( + 'WARNING: Local changes made to the following' + ' templates will be overwritten! Continue?')) + box.add(self.msg) + + self.store = Gtk.ListStore(*Action.TYPES) + self.listing = Gtk.TreeView(model=self.store) + for idx, colname in enumerate(Action.COL_NAMES): + renderer = Gtk.CellRendererText() + col = Gtk.TreeViewColumn(colname, renderer, text=idx) + self.listing.append_column(col) + col.set_sort_column_id(idx) + + for row in actions: + self.store.append(row) + + self.scrollable_listing = Gtk.ScrolledWindow() + self.scrollable_listing.add(self.listing) + box.pack_start(self.scrollable_listing, True, True, 16) + + self.show_all() + +class ProgressDialog(Gtk.Dialog): + def __init__(self, parent): + super(ProgressDialog, self).__init__( + title='Processing...', transient_for=parent, modal=True) + box = self.get_content_area() + + self.spinner = Gtk.Spinner() + self.spinner.start() + box.add(self.spinner) + + self.msg = Gtk.Label() + self.msg.set_text('Processing...') + box.add(self.msg) + + self.infobox = Gtk.TextView() + self.scrollable = Gtk.ScrolledWindow() + self.scrollable.add(self.infobox) + + box.pack_start(self.scrollable, True, True, 16) + + self.show_all() + + def finish(self, success): + self.spinner.stop() + if success: + self.msg.set_text('Operations succeeded.') + else: + self.msg.set_markup('Error:') + self.add_button(Gtk.STOCK_OK, Gtk.ResponseType.OK) + self.run() + +class QubesTemplateApp(Gtk.Window): + def __init__(self): + super(QubesTemplateApp, self).__init__(title='Qubes Template Manager') + + self.iconsize = Gtk.IconSize.SMALL_TOOLBAR + + self.executor = concurrent.futures.ThreadPoolExecutor() + self.outerbox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL) + self.__build_action_models() + self.__build_toolbar() + self.__build_listing() + self.__build_infobox() + + self.add(self.outerbox) + + def __build_action_models(self): + #pylint: disable=invalid-name + OPS = [ + ['Installed', 'Reinstall', 'Remove'], + ['Extra', 'Remove'], + ['Upgradable', 'Upgrade', 'Remove'], + ['Downgradable', 'Downgrade', 'Remove'], + ['Available', 'Install'] + ] + self.action_models = {} + for ops in OPS: + # First element is the default status for the certain class of + # templates + self.action_models[ops[0]] = Gtk.ListStore(str) + for oper in ops: + self.action_models[ops[0]].append([oper]) + + def __build_toolbar(self): + self.toolbar = Gtk.Toolbar() + self.btn_refresh = Gtk.ToolButton( + icon_widget=Gtk.Image.new_from_icon_name( + 'view-refresh', self.iconsize), + label='Refresh') + self.btn_refresh.connect('clicked', self.refresh) + self.toolbar.insert(self.btn_refresh, 0) + + self.btn_install = Gtk.ToolButton( + icon_widget=Gtk.Image.new_from_icon_name('go-down', self.iconsize), + label='Apply') + self.btn_install.connect('clicked', self.show_confirm) + self.toolbar.insert(self.btn_install, 1) + + self.outerbox.pack_start(self.toolbar, False, True, 0) + + def __build_listing(self): + self.store = Gtk.ListStore(*Template.TYPES) + + self.listing = Gtk.TreeView(model=self.store) + self.cols = [] + for idx, colname in enumerate(Template.COL_NAMES): + if colname == 'Status': + renderer = Gtk.CellRendererCombo() + renderer.set_property('editable', True) + renderer.set_property('has-entry', False) + renderer.set_property('text-column', 0) + renderer.connect('edited', self.entry_edit) + col = Gtk.TreeViewColumn( + colname, + renderer, + text=idx, + weight=len(Template.TYPES) - 2, + model=len(Template.TYPES) - 1) + else: + renderer = Gtk.CellRendererText() + col = Gtk.TreeViewColumn( + colname, + renderer, + text=idx, + weight=len(Template.TYPES) - 2) + # Right-align for integers + if Template.TYPES[idx] is int: + renderer.set_property('xalign', 1.0) + self.cols.append(col) + self.listing.append_column(col) + col.set_sort_column_id(idx) + sel = self.listing.get_selection() + sel.set_mode(Gtk.SelectionMode.MULTIPLE) + sel.connect('changed', self.update_info) + + self.scrollable_listing = Gtk.ScrolledWindow() + self.scrollable_listing.add(self.listing) + self.scrollable_listing.set_visible(False) + + self.spinner = Gtk.Spinner() + + self.outerbox.pack_start(self.scrollable_listing, True, True, 0) + self.outerbox.pack_start(self.spinner, True, True, 0) + + def __build_infobox(self): + self.infobox = Gtk.TextView() + self.outerbox.pack_start(self.infobox, True, True, 16) + + def refresh(self, button=None): + # Ignore if we're already doing a refresh + #pylint: disable=no-member + if self.spinner.props.active: + return + self.scrollable_listing.set_visible(False) + self.spinner.start() + self.spinner.set_visible(True) + self.store.clear() + def worker(): + cmd = BASE_CMD[:] + if button is not None: + # Force refresh if triggered by button press + cmd.append('--refresh') + cmd.extend(['info', '--machine-readable-json', '--installed', + '--available', '--upgrades', '--extras']) + output = subprocess.check_output(cmd) + # Default type is dict as we're going to replace the lists with + # dicts shortly after + tpls = collections.defaultdict(dict, json.loads(output)) + # Remove duplicates + # Should this be done in qvm-template? + # TODO: Merge templates with same name? + # If so, we may need to have a separate UI to force versions. + local_names = set(x['name'] for x in tpls['installed']) + # Convert to dict for easier subtraction + for key in tpls: + tpls[key] = { + (x['name'], x['epoch'], x['version'], x['release']): x + for x in tpls[key]} + tpls['installed'] = { + k: v for k, v in tpls['installed'].items() + if k not in tpls['extra'] and k not in tpls['upgradable']} + tpls['available'] = { + k: v for k, v in tpls['available'].items() + if k not in tpls['installed'] + and k not in tpls['upgradable']} + # If the package name is installed but the specific version is + # neither installed or an upgrade, then it must be a downgrade + tpls['downgradable'] = { + k: v for k, v in tpls['available'].items() + if k[0] in local_names} + tpls['available'] = { + k: v for k, v in tpls['available'].items() + if k not in tpls['downgradable']} + # Convert back to list + for key in tpls: + tpls[key] = list(tpls[key].values()) + for status, seq in tpls.items(): + status_str = status.title() + for entry in seq: + self.store.append(Template.build( + status_str, entry, self.action_models[status_str])) + + def finish_cb(future): + def callback(): + if future.exception() is not None: + buf = self.infobox.get_buffer() + buf.set_text('Error:\n' + str(future.exception())) + self.spinner.set_visible(False) + self.spinner.stop() + self.scrollable_listing.set_visible(True) + GLib.idle_add(callback) + + future = self.executor.submit(worker) + future.add_done_callback(finish_cb) + + def show_confirm(self, button=None): + _ = button # unused + actions = [] + for row in self.store: + tpl = Template(*row) + if tpl.status != tpl.default_status: + actions.append(Action(tpl.status, tpl.name, tpl.evr)) + dialog = ConfirmDialog(self, actions) + resp = dialog.run() + dialog.destroy() + if resp == Gtk.ResponseType.OK: + self.do_install(actions) + + def do_install(self, actions): + dialog = ProgressDialog(self) + def worker(): + actions.sort() + for oper, grp in itertools.groupby(actions, lambda x: x[0]): + oper = oper.lower() + # No need to specify versions for local operations + if oper in ('remove', 'purge'): + specs = [x.name for x in grp] + else: + specs = [x.name + '-' + x.evr for x in grp] + # FIXME: (C)Python versions before 3.9 fully-buffers stderr in + # this context, cf. https://bugs.python.org/issue13601 + # Forcing it to be unbuffered for the time being so that + # the messages can be displayed in time. + envs = os.environ.copy() + envs['PYTHONUNBUFFERED'] = '1' + proc = subprocess.Popen( + BASE_CMD + [oper, '--'] + specs, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + env=envs) + #pylint: disable=cell-var-from-loop + for line in iter(proc.stdout.readline, ''): + # Need to modify the buffers in the main thread + def callback(): + buf = dialog.infobox.get_buffer() + end_iter = buf.get_end_iter() + buf.insert(end_iter, line) + GLib.idle_add(callback) + if proc.wait() != 0: + return False + return True + + def finish_cb(future): + def callback(): + dialog.finish(future.result()) + dialog.destroy() + self.refresh() + GLib.idle_add(callback) + + future = self.executor.submit(worker) + future.add_done_callback(finish_cb) + + def update_info(self, sel): + model, treeiters = sel.get_selected_rows() + if not treeiters: + return + buf = self.infobox.get_buffer() + if len(treeiters) > 1: + def row_to_spec(row): + tpl = Template(*row) + return tpl.name + '-' + tpl.evr + text = '\n'.join(row_to_spec(model[it]) for it in treeiters) + buf.set_text('Selected templates:\n' + text) + else: + itr = treeiters[0] + tpl = Template(*model[itr]) + text = 'Name: %s\n\nDescription:\n%s' % (tpl.name, tpl.description) + buf.set_text(text) + + def entry_edit(self, widget, path, text): + _ = widget # unused + #pylint: disable=unsubscriptable-object + tpl = Template(*self.store[path]) + tpl = tpl._replace(status=text) + if text == tpl.default_status: + tpl = tpl._replace(weight=Pango.Weight.BOOK) + else: + tpl = tpl._replace(weight=Pango.Weight.BOLD) + #pylint: disable=unsupported-assignment-operation + self.store[path] = tpl + +if __name__ == '__main__': + main = QubesTemplateApp() + main.connect('destroy', Gtk.main_quit) + main.show_all() + main.refresh() + Gtk.main()