Merge remote-tracking branch 'origin/pr/170'

* origin/pr/170:
  qubes-vm-update: updates standalones too
  qubes-vm-update: add CLI options

Author: marmarek

qubes_config/tests/conftest.py

@@ -217,6 +217,10 @@ def add_feature_to_all(qapp, feature_name, enable_vm_names: List[str]):
 
 @pytest.fixture
 def test_qapp():
+    return test_qapp_impl()
+
+
+def test_qapp_impl():
     """Test QubesApp"""
     qapp = QubesTest()
     qapp._local_name = 'dom0'  # pylint: disable=protected-access

qui/updater/intro_page.py

@@ -175,6 +175,48 @@ def select_rows(self):
             row.selected = row.updates_available \
                            in self.head_checkbox.allowed
 
+    def select_rows_ignoring_conditions(self, cliargs):
+        cmd = ['qubes-vm-update', '--dry-run']
+
+        args = [a for a in dir(cliargs) if not a.startswith("_")]
+        for arg in args:
+            if arg in ("dom0", "no-restart", "restart", "max_concurrency",
+                       "log"):
+                continue
+            value = getattr(cliargs, arg)
+            if value:
+                if arg in ("skip", "targets"):
+                    vms = set(value.split(","))
+                    vms_without_dom0 = vms.difference({"dom0"})
+                    if not vms_without_dom0:
+                        continue
+                    value = ",".join(vms_without_dom0)
+                cmd.extend((f"--{arg.replace('_', '-')}", str(value)))
+
+        if not cmd[2:]:
+            to_update = set()
+        else:
+            self.log.debug("Run command %s", " ".join(cmd))
+            output = subprocess.check_output(cmd)
+            self.log.debug("Command returns: %s", output.decode())
+
+            to_update = {
+                vm_name.strip()
+                for line in output.decode().split("\n", maxsplit=1)
+                for vm_name in line.split(":", maxsplit=1)[1].split(",")
+            }
+
+        # handle dom0
+        if cliargs.dom0 or cliargs.all:
+            to_update.add("dom0")
+        if cliargs.targets and "dom0" in cliargs.targets.split(","):
+            to_update.add("dom0")
+        if cliargs.skip and "dom0" in cliargs.skip.split(","):
+            to_update = to_update.difference({"dom0"})
+
+        for row in self.list_store:
+            row.selected = row.name in to_update
+
 
 class UpdateRowWrapper(RowWrapper):
     COLUMN_NUM = 9

qui/updater/tests/test_intro_page.py

@@ -18,11 +18,15 @@
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301,
 # USA.
+import argparse
+
 import pytest
 from unittest.mock import patch
 from unittest.mock import Mock
 
+from qubes_config.tests.conftest import test_qapp_impl
 from qui.updater.intro_page import IntroPage, UpdateRowWrapper, UpdatesAvailable
+from qui.updater.updater import parse_args
 from qui.updater.utils import ListWrapper, HeaderCheckbox
 
 @patch('subprocess.check_output')
@@ -153,3 +157,54 @@ def test_on_checkbox_toggled(
     # no selected row
     assert not sut.checkbox_column_button.get_inconsistent()
     assert not sut.checkbox_column_button.get_active()
+
+all_domains = {vm.name for vm in test_qapp_impl().domains}
+all_templates = {vm.name for vm in test_qapp_impl().domains if vm.klass == "TemplateVM"}
+all_standalones = {vm.name for vm in test_qapp_impl().domains if vm.klass == "StandaloneVM"}
+
+@patch('subprocess.check_output')
+@pytest.mark.parametrize(
+    "args, templates, rest, selected",
+    (
+        pytest.param(('--all',), ",".join(all_templates).encode(), ",".join(all_domains.difference({"dom0"}).difference(all_templates)).encode(), all_domains),
+        pytest.param(('--update-if-stale', '10'), b'fedora-36', b'', {'fedora-36'}),
+        pytest.param(('--targets', 'dom0,fedora-36'), b'fedora-36', b'', {'dom0', 'fedora-36'}),
+        pytest.param(('--dom0', '--skip', 'dom0'), b'', b'', set()),
+        pytest.param(('--skip', 'dom0'), b'', b'', set()),
+        pytest.param(('--targets', 'dom0', '--skip', 'dom0'), b'', b'', set()),
+        pytest.param(('--dom0',), b'', b'', {'dom0'}),
+        pytest.param(('--standalones',), b'', ",".join(all_standalones).encode(), all_standalones),
+        pytest.param(('--templates', '--skip', 'fedora-36,garbage-name'),
+                     ",".join(all_templates.difference({"fedora-36"})).encode(),
+                     b'',
+                     all_templates.difference({"fedora-36"})),
+    ),
+)
+def test_select_rows_ignoring_conditions(
+        mock_subprocess, args, templates, rest, selected, real_builder, test_qapp,
+        mock_next_button, mock_settings, mock_list_store
+):
+    mock_log = Mock()
+    sut = IntroPage(real_builder, mock_log, mock_next_button)
+
+    # populate_vm_list
+    sut.list_store = ListWrapper(UpdateRowWrapper, mock_list_store)
+    for vm in test_qapp.domains:
+        sut.list_store.append_vm(vm)
+
+    assert len(sut.list_store) == 12
+
+    mock_subprocess.return_value = (
+        b'Following templates will be updated: ' + templates + b'\n'
+        b'Following qubes will be updated: ' + rest
+    )
+
+    cliargs = parse_args(args)
+    sut.select_rows_ignoring_conditions(cliargs)
+    to_update = {row.name for row in sut.list_store if row.selected}
+
+    assert to_update == selected
+    if not templates + rest:
+        mock_subprocess.assert_not_called()
+
+

qui/updater/tests/test_updater.py

@@ -20,14 +20,14 @@
 # USA.
 from unittest.mock import patch, call
 
-from qui.updater.updater import QubesUpdater
+from qui.updater.updater import QubesUpdater, parse_args
 
 
 @patch('logging.FileHandler')
 @patch('logging.getLogger')
 @patch('qui.updater.intro_page.IntroPage.populate_vm_list')
 def test_setup(populate_vm_list, _mock_logging, __mock_logging, test_qapp):
-    sut = QubesUpdater(test_qapp)
+    sut = QubesUpdater(test_qapp, parse_args(()))
     sut.perform_setup()
     calls = [call(sut.qapp, sut.settings)]
     populate_vm_list.assert_has_calls(calls)

qui/updater/updater.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 # pylint: disable=wrong-import-position,import-error
+import argparse
 import logging
 import time
 
@@ -10,7 +11,7 @@
 from qubes_config.widgets.gtk_utils import load_icon_at_gtk_size, load_theme, \
     show_dialog_with_icon, RESPONSES_OK
 from qui.updater.progress_page import ProgressPage
-from qui.updater.updater_settings import Settings
+from qui.updater.updater_settings import Settings, OverridenSettings
 from qui.updater.summary_page import SummaryPage
 from qui.updater.intro_page import IntroPage
 
@@ -33,14 +34,15 @@ class QubesUpdater(Gtk.Application):
     LOGPATH = '/var/log/qubes/qui.updater.log'
     LOG_FORMAT = '%(asctime)s %(message)s'
 
-    def __init__(self, qapp):
+    def __init__(self, qapp, cliargs):
         super().__init__(
             application_id="org.gnome.example",
             flags=Gio.ApplicationFlags.FLAGS_NONE
         )
         self.qapp = qapp
         self.primary = False
         self.connect("activate", self.do_activate)
+        self.cliargs = cliargs
 
         log_handler = logging.FileHandler(
             QubesUpdater.LOGPATH, encoding='utf-8')
@@ -49,13 +51,15 @@ def __init__(self, qapp):
 
         self.log = logging.getLogger('vm-update.agent.PackageManager')
         self.log.addHandler(log_handler)
-        self.log.setLevel("DEBUG")
+        self.log.setLevel(self.cliargs.log)
 
     def do_activate(self, *_args, **_kwargs):
         if not self.primary:
             self.perform_setup()
             self.primary = True
             self.hold()
+        elif len(self.intro_page.get_vms_to_update()) == 0:
+            self.exit_updater()
         else:
             self.main_window.present()
 
@@ -105,11 +109,25 @@ def perform_setup(self, *_args, **_kwargs):
             'qubes-customize', Gtk.IconSize.LARGE_TOOLBAR)
         settings_image = Gtk.Image.new_from_pixbuf(settings_pixbuf)
         self.button_settings.set_image(settings_image)
+
+        overriden_restart = None
+        if self.cliargs.restart:
+            overriden_restart = True
+        elif  self.cliargs.no_restart:
+            overriden_restart = False
+
+        overrides = OverridenSettings(
+            restart=overriden_restart,
+            max_concurrency=self.cliargs.max_concurrency,
+            update_if_stale=self.cliargs.update_if_stale,
+        )
+
         self.settings = Settings(
             self.main_window,
             self.qapp,
             self.log,
-            refresh_callback=self.intro_page.refresh_update_list
+            refresh_callback=self.intro_page.refresh_update_list,
+            overrides=overrides,
         )
 
         headers = [(3, "intro_name"), (3, "progress_name"), (3, "summary_name"),
@@ -137,6 +155,23 @@ def cell_data_func(_column, cell, model, it, data):
         self.main_window.connect("key-press-event", self.check_escape)
 
         self.intro_page.populate_vm_list(self.qapp, self.settings)
+
+        if skip_intro_if_args(self.cliargs):
+            self.log.info("Skipping intro page.")
+            self.intro_page.select_rows_ignoring_conditions(
+                cliargs=self.cliargs)
+            if len(self.intro_page.get_vms_to_update()) == 0:
+                show_dialog_with_icon(
+                    None, l("Quit"),
+                    l("Nothing to do."),
+                    buttons=RESPONSES_OK,
+                    icon_name="qubes-info"
+                )
+                self.main_window.close()
+                return
+            self.next_clicked(None, skip_intro=True)
+        else:
+            self.log.info("Show intro page.")
         self.main_window.show_all()
         width = self.intro_page.vm_list.get_preferred_width().natural_width
         self.main_window.resize(width + 50, int(width * 1.2))
@@ -145,9 +180,9 @@ def cell_data_func(_column, cell, model, it, data):
     def open_settings_window(self, _emitter):
         self.settings.show()
 
-    def next_clicked(self, _emitter):
+    def next_clicked(self, _emitter, skip_intro=False):
         self.log.debug("Next clicked")
-        if self.intro_page.is_visible:
+        if self.intro_page.is_visible or skip_intro:
             vms_to_update = self.intro_page.get_vms_to_update()
             self.intro_page.active = False
             self.progress_page.show()
@@ -213,9 +248,61 @@ def exit_updater(self, _emitter=None):
             self.release()
 
 
-def main():
+def parse_args(args):
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument('--log', action='store', default='WARNING',
+                        help='Provide logging level. Values: DEBUG, INFO, '
+                             'WARNING (default), ERROR, CRITICAL')
+    parser.add_argument('--max-concurrency', action='store',
+                        help='Maximum number of VMs configured simultaneously '
+                             '(default: number of cpus)',
+                        type=int)
+    restart_gr = parser.add_mutually_exclusive_group()
+    restart_gr.add_argument('--restart', action='store_true',
+                            help='Restart AppVMs whose template '
+                                 'has been updated.')
+    restart_gr.add_argument('--no-restart', action='store_true',
+                            help='Do not restart AppVMs whose template '
+                                 'has been updated.')
+
+    group = parser.add_mutually_exclusive_group()
+    group.add_argument('--targets', action='store',
+                       help='Comma separated list of VMs to target')
+    group.add_argument('--all', action='store_true',
+                       help='Target all non-disposable VMs (TemplateVMs and '
+                            'AppVMs)')
+    group.add_argument('--update-if-stale', action='store',
+                       help='Target all TemplateVMs with known updates or for '
+                            'which last update check was more than N days '
+                            'ago.',
+                       type=int)
+
+    parser.add_argument('--skip', action='store',
+                        help='Comma separated list of VMs to be skipped, '
+                             'works with all other options.', default="")
+    parser.add_argument('--templates', action='store_true',
+                        help='Target all TemplatesVMs')
+    parser.add_argument('--standalones', action='store_true',
+                        help='Target all StandaloneVMs')
+    parser.add_argument('--dom0', action='store_true',
+                        help='Target dom0')
+
+    args = parser.parse_args(args)
+
+    return args
+
+
+def skip_intro_if_args(args):
+    return args is not None and (args.templates or args.standalones or args.skip
+                                 or args.update_if_stale or args.all
+                                 or args.targets or args.dom0)
+
+
+def main(args=None):
+    cliargs = parse_args(args)
     qapp = Qubes()
-    app = QubesUpdater(qapp)
+    app = QubesUpdater(qapp, cliargs)
     app.run()