qubesos: add force parameter to shutdown and restarted states

ansible_collections/qubesos/core/plugins/module_utils/qubes_helper.py

@@ -129,14 +129,19 @@ def info(self):
             }
         return info
 
-    def shutdown(self, vmname, wait=False):
+    def shutdown(self, vmname, wait=False, force=False):
         """
         Shutdown the specified qube via the given id or name,
         optionally waiting until it halts.
+
+        If ``force`` is True, passes ``force=True`` to
+        ``qubesadmin.vm.QubesVM.shutdown`` so the shutdown proceeds
+        regardless of connected domains (equivalent to
+        ``qvm-shutdown --force``).
         """
         vm = self.get_vm(vmname)
         with suppress(QubesVMNotStartedError):
-            vm.shutdown()
+            vm.shutdown(force=force)
 
         if wait:
             try:
@@ -154,13 +159,13 @@ def shutdown(self, vmname, wait=False):
                 )
         return 0
 
-    def restart(self, vmname, wait=False):
+    def restart(self, vmname, wait=False, force=False):
         """
         Restart the specified qube via the given id or name
         by shutting it down (with optional wait) and then starting it.
         """
         try:
-            self.shutdown(vmname, wait=wait)
+            self.shutdown(vmname, wait=wait, force=force)
         except RuntimeError:
             raise
         vm = self.get_vm(vmname)

ansible_collections/qubesos/core/plugins/module_utils/qubes_module_qube.py

@@ -77,6 +77,7 @@ def __init__(self, module: AnsibleModule):
         self.created = False
         self.deleted = False
         self.diff = {"before": {}, "after": {}}
+        self.force_shutdown = module.params.get("force")
         self.helper = QubesHelper(module)
         self.qube: QubesVM = self.helper.app.domains.get(self.qube_name)
         self.shutdown_if_required = module.params.get("shutdown_if_required")
@@ -131,7 +132,9 @@ def _shutdown_for_template_update(self):
                 # `shutdown_if_required` module params allows us stop it
                 # if necessary
                 if self.shutdown_if_required:
-                    self.helper.shutdown(self.qube_name, wait=True)
+                    self.helper.shutdown(
+                        self.qube_name, wait=True, force=self.force_shutdown
+                    )
                 else:
                     self.module.fail_json(
                         msg="Cannot change the template while the qube is running"
@@ -577,7 +580,9 @@ def run(self):
             if self.wants.state in ["halted", "restarted"]:
                 if not self.qube.is_halted():
                     self.changed = True
-                    self.helper.shutdown(self.qube_name, wait=True)
+                    self.helper.shutdown(
+                        self.qube_name, wait=True, force=self.force_shutdown
+                    )
 
             self.enforce_all()
 
@@ -622,6 +627,7 @@ def main():
             clone_src=dict(type="str", default=None),
             devices=dict(type="raw", default=None),
             features=dict(type="dict", default=None),
+            force=dict(type="bool", default=False),
             notes=dict(type="str", default=None),
             properties=dict(type="dict", default=None),
             services=dict(type="list", default=None),

ansible_collections/qubesos/core/plugins/modules/qube.py

@@ -95,6 +95,20 @@
             - You have to provide a dict with features name as dict key and the desired value as dict value.
         type: dict
 
+    force:
+        description:
+            - If C(true), shut down the target VM regardless of whether other
+              VMs are connected to it (e.g. AppVMs using it as a netvm).
+            - Equivalent to C(qvm-shutdown --force) on the CLI. The target still
+              halts gracefully; only the "connected domains" precondition is
+              skipped. Dependent VMs keep running but lose their uplink until
+              the netvm is started again.
+            - For a hard-kill (equivalent to C(qvm-kill) / SIGKILL of the Xen
+              domain, no graceful shutdown), use C(state=destroyed) instead.
+            - Only applies to C(shutdown) and C(restarted) states.
+        type: bool
+        default: false
+
     notes:
         description:    
             - Set the qube notes.

plugins/modules/qubesos.py

@@ -60,6 +60,19 @@
       - Only applies to C(shutdown) and C(restarted) states.
     type: bool
     default: false
+  force:
+    description:
+      - If C(true), shut down the target VM regardless of whether other
+        VMs are connected to it (e.g. AppVMs using it as a netvm).
+      - Equivalent to C(qvm-shutdown --force) on the CLI. The target still
+        halts gracefully; only the "connected domains" precondition is
+        skipped. Dependent VMs keep running but lose their uplink until
+        the netvm is started again.
+      - For a hard-kill (equivalent to C(qvm-kill) / SIGKILL of the Xen
+        domain, no graceful shutdown), use C(state=destroyed) instead.
+      - Only applies to C(shutdown) and C(restarted) states.
+    type: bool
+    default: false
   command:
     description:
       - Non-idempotent command to execute on the VM.
@@ -438,6 +451,7 @@ def core(module):
     tags = module.params.get("tags", [])
     devices = module.params.get("devices", None)
     notes = module.params.get("notes", None)
+    force = module.params.get("force", False)
 
     v = QubesHelper(module)
 
@@ -533,6 +547,7 @@ def core(module):
                     "clone_src": clone_src,
                     "devices": devices,
                     "features": features,
+                    "force": force,
                     "klass": vmtype,
                     "name": guest,
                     "notes": notes,
@@ -634,6 +649,7 @@ def core(module):
                 {
                     "name": guest,
                     "state": state,
+                    "force": force,
                 }
             )
             return VIRT_SUCCESS, fake_module.returned_data
@@ -662,6 +678,7 @@ def main():
                 ],
             ),
             wait=dict(type="bool", default=True),
+            force=dict(type="bool", default=False),
             command=dict(type="str", choices=ALL_COMMANDS),
             label=dict(type="str", default=None),
             vmtype=dict(type="str", default=None),

tests/qubes/test_legacy_qubesos_module.py

@@ -1316,3 +1316,50 @@ def test_no_vm_klass_should_no_try_to_convert_to_appvm(request, vmname, qubes):
     assert rc == VIRT_SUCCESS
     assert qubes.domains[vmname].get_power_state() == "Running"
     assert qubes.domains[vmname].klass == "StandaloneVM"
+
+
+def test_lifecycle_shutdown_force_with_dependent(qubes, vm, netvm):
+    """force=True shuts down a netvm that still has a running dependent.
+
+    Without force, the underlying vm.shutdown() raises QubesVMInUseError,
+    which the module surfaces as a task failure. Mirrors the CLI's
+    qvm-shutdown --force semantics: the target halts gracefully; only
+    the "connected domains" precondition is skipped. Dependents keep
+    running (without uplink).
+
+    Refs: QubesOS/qubes-issues#10856
+    """
+    # Start the netvm
+    rc, _ = run_module({"command": "start", "name": netvm.name})
+    assert rc == VIRT_SUCCESS
+    assert netvm.is_running()
+
+    # Point the dependent AppVM at this netvm and start it
+    rc, _ = run_module(
+        {
+            "state": "present",
+            "name": vm.name,
+            "properties": {"netvm": netvm.name},
+        }
+    )
+
+    assert rc == VIRT_SUCCESS
+    rc, _ = run_module({"command": "start", "name": vm.name})
+    assert rc == VIRT_SUCCESS
+    qubes.domains.refresh_cache(force=True)
+    assert qubes.domains[vm.name].is_running()
+
+    # Force-shutdown the netvm despite the running dependent
+    rc, _ = run_module(
+        {
+            "state": "shutdown",
+            "name": netvm.name,
+            "wait": True,
+            "force": True,
+        }
+    )
+
+    assert rc == VIRT_SUCCESS
+    assert netvm.is_halted()
+    # Dependent is still running; it has merely lost its uplink.
+    assert qubes.domains[vm.name].is_running()

tests/qubes/test_module_qube.py

@@ -1145,3 +1145,48 @@ def test_no_vm_klass_should_no_try_to_convert_to_appvm(request, vmname, qubes):
 
     assert qubes.domains[vmname].get_power_state() == "Running"
     assert qubes.domains[vmname].klass == "StandaloneVM"
+
+
+def test_lifecycle_shutdown_force_with_dependent(qubes, vm, netvm):
+    """force=True shuts down a netvm that still has a running dependent.
+
+    Without force, the underlying vm.shutdown() raises QubesVMInUseError,
+    which the module surfaces as a task failure. Mirrors the CLI's
+    qvm-shutdown --force semantics: the target halts gracefully; only
+    the "connected domains" precondition is skipped. Dependents keep
+    running (without uplink).
+
+    Refs: QubesOS/qubes-issues#10856
+    """
+    # Start the netvm
+
+    run_module(
+        {
+            "state": "running",
+            "name": netvm.name,
+        }
+    )
+    assert netvm.is_running()
+
+    # Point the dependent AppVM at this netvm and start it
+    run_module(
+        {
+            "state": "running",
+            "name": vm.name,
+            "properties": {"netvm": netvm.name},
+        }
+    )
+    qubes.domains.refresh_cache(force=True)
+    assert qubes.domains[vm.name].is_running()
+
+    # Force-shutdown the netvm despite the running dependent
+    run_module(
+        {
+            "state": "halted",
+            "name": netvm.name,
+            "force": True,
+        }
+    )
+    assert netvm.is_halted()
+    # Dependent is still running; it has merely lost its uplink.
+    assert qubes.domains[vm.name].is_running()