qubesos: add force parameter to shutdown and restarted states

plugins/modules/qubesos.py

@@ -59,6 +59,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.
@@ -373,14 +386,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:
@@ -398,13 +416,14 @@ 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.
+        ``force`` is passed through to :meth:`shutdown`.
         """
         try:
-            self.shutdown(vmname, wait=wait)
+            self.shutdown(vmname, wait=wait, force=force)
         except RuntimeError:
             raise
         vm = self.get_vm(vmname)
@@ -992,13 +1011,21 @@ def apply_devices(vmname):
             if current != "shutdown":
                 res["changed"] = True
                 try:
-                    v.shutdown(guest, wait=module.params.get("wait", False))
+                    v.shutdown(
+                        guest,
+                        wait=module.params.get("wait", False),
+                        force=module.params.get("force", False),
+                    )
                 except RuntimeError as e:
                     module.fail_json(msg=str(e))
         elif state == "restarted":
             res["changed"] = True
             try:
-                v.restart(guest, wait=module.params.get("wait", False))
+                v.restart(
+                    guest,
+                    wait=module.params.get("wait", False),
+                    force=module.params.get("force", False),
+                )
                 res["msg"] = "restarted"
             except RuntimeError as e:
                 module.fail_json(msg=str(e))
@@ -1041,6 +1068,7 @@ def main():
                 ],
             ),
             wait=dict(type="bool", default=False),
+            force=dict(type="bool", default=False),
             command=dict(type="str", choices=ALL_COMMANDS),
             label=dict(type="str", default="red"),
             vmtype=dict(type="str", default="AppVM"),

tests/qubes/test_module.py

@@ -1105,3 +1105,52 @@ def test_properties_invalid_type_for_new_properties(qubes, vmname, request):
     )
     assert rc2 == VIRT_FAILED
     assert "Invalid property value type" in res2
+
+
+def test_lifecycle_shutdown_force_with_dependent(qubes, vm, netvm, request):
+    """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, _ = core(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, _ = core(
+        Module(
+            {
+                "state": "present",
+                "name": vm.name,
+                "properties": {"netvm": netvm.name},
+            }
+        )
+    )
+    assert rc == VIRT_SUCCESS
+    rc, _ = core(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, _ = core(
+        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()