vmupdate: move shutdown_domains to shutdown_utils to avoid circular import

Extract shutdown_domains() into a new vmupdate/shutdown_utils.py module
so that both vmupdate.py and qube_connection.py can import it cleanly,
without any circular dependency or late imports.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Jayant-kernel

vmupdate/qube_connection.py

@@ -33,6 +33,7 @@
 from vmupdate.agent.source.log_config import LOGPATH, LOG_FILE
 from vmupdate.agent.source.status import StatusInfo, FinalStatus
 from vmupdate.agent.source.common.process_result import ProcessResult
+from vmupdate.shutdown_utils import shutdown_domains
 
 
 class QubeConnection:
@@ -94,9 +95,6 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
         if self.qube.is_running() and not self._initially_running:
             if self._has_assigned_pci_devices(self.qube):
-                # Late import to avoid circular dependency:
-                # qube_connection <- vmupdate <- update_manager <- qube_connection
-                from vmupdate.vmupdate import shutdown_domains  # pylint: disable=import-outside-toplevel
                 self.logger.info(
                     'Waiting for full shutdown %s (PCI devices assigned)',
                     self.qube.name)

vmupdate/shutdown_utils.py

@@ -0,0 +1,44 @@
+# coding=utf-8
+#
+# The Qubes OS Project, http://www.qubes-os.org
+#
+# Copyright (C) 2022  Piotr Bartman <prbartman@invisiblethingslab.com>
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301,
+# USA.
+import asyncio
+
+import qubesadmin.exc
+from qubesadmin.events.utils import wait_for_domain_shutdown
+from vmupdate.agent.source.common.exit_codes import EXIT
+
+
+def shutdown_domains(to_shutdown, log):
+    """
+    Try to shut down vms and wait to finish.
+    """
+    ret_code = EXIT.OK
+    wait_for = []
+    for vm in to_shutdown:
+        try:
+            vm.shutdown(force=True)
+            wait_for.append(vm)
+        except qubesadmin.exc.QubesVMError as exc:
+            log.error(str(exc))
+            ret_code = EXIT.ERR_SHUTDOWN_APP
+
+    asyncio.run(wait_for_domain_shutdown(wait_for))
+
+    return ret_code, wait_for

vmupdate/vmupdate.py

@@ -3,7 +3,6 @@
 Update qubes.
 """
 import argparse
-import asyncio
 import logging
 import sys
 import os
@@ -13,9 +12,9 @@
 
 import qubesadmin
 import qubesadmin.exc
-from qubesadmin.events.utils import wait_for_domain_shutdown
 from vmupdate.agent.source.status import FinalStatus
 from vmupdate.agent.source.common.exit_codes import EXIT
+from vmupdate.shutdown_utils import shutdown_domains
 from . import update_manager
 from .agent.source.args import AgentArgs
 
@@ -394,24 +393,6 @@ def get_derived_vm_to_apply(templates, derived_statuses):
     return to_restart, to_shutdown
 
 
-def shutdown_domains(to_shutdown, log):
-    """
-    Try to shut down vms and wait to finish.
-    """
-    ret_code = EXIT.OK
-    wait_for = []
-    for vm in to_shutdown:
-        try:
-            vm.shutdown(force=True)
-            wait_for.append(vm)
-        except qubesadmin.exc.QubesVMError as exc:
-            log.error(str(exc))
-            ret_code = EXIT.ERR_SHUTDOWN_APP
-
-    asyncio.run(wait_for_domain_shutdown(wait_for))
-
-    return ret_code, wait_for
-
 
 def restart_vms(to_restart, log):
     """