fix: accept HA shutdown tasks

AppImage/scripts/flask_server.py

@@ -15924,7 +15924,9 @@ def api_vm_control(vmid):
                 task_types = [f"{'vz' if vm_type == 'lxc' else 'qm'}{action}"]
                 if action == 'start':
                     task_types.append('hastart')
-                elif action == 'stop':
+                elif action in ('stop', 'shutdown'):
+                    # Proxmox 9 routes HA shutdown through hastop for both
+                    # LXC and QEMU guests.
                     task_types.append('hastop')
                 task_pattern = (r'UPID:[\w.-]+:[0-9A-Fa-f]+:[0-9A-Fa-f]+:'
                                 r'[0-9A-Fa-f]+:(?:' + '|'.join(task_types) +

AppImage/scripts/tests/test_lxc_vm_ip_inventory.py

@@ -36,6 +36,7 @@ class LxcVmIpInventoryTests(unittest.TestCase):
             "_get_lxc_app_watch_map": lambda: {},
             "_get_lxc_docker_inventory_map": lambda: {},
             "get_proxmox_node_name": lambda: "pve",
+            "_fast_guest_status": lambda _vmid, _vm_type: "running",
             "_get_cached_vm_list_search_notes": lambda _resources, _node: {},
             "get_cached_vm_disk": lambda _vmid: None,
             "_guest_lifecycle_lock": threading.RLock(),

tests/lxc_updates/test_vm_control_task.py

@@ -71,6 +71,29 @@ class ControlTaskTests(unittest.TestCase):
         self.assertEqual(response.json['task'], task)
         retained.assert_called_once_with(100, task, lock)
 
+    def test_control_post_accepts_ha_shutdown_task_for_lxc_and_qemu(self):
+        lock = Mock()
+        retained = Mock()
+        module = SimpleNamespace(acquire=lambda _: lock, begin_control=Mock(),
+                                 retain_control=retained, cancel_control=Mock())
+        with patch.dict('sys.modules', {'lxc_check_job': module}):
+            for vmid in (100, 200):
+                with self.subTest(vmid=vmid):
+                    self.run.reset_mock()
+                    task = TASK.replace('vzreboot:100', f'hastop:{vmid}')
+                    self.run.return_value = self.result(task)
+                    response = self.client.post(
+                        f'/api/vms/{vmid}/control', json={'action': 'shutdown'})
+                    self.assertEqual(response.status_code, 200, response.json)
+                    self.assertTrue(response.json['success'])
+                    self.assertEqual(response.json['task'], task)
+                    self.assertEqual(self.run.call_count, 1)
+                    self.assertEqual(self.run.call_args.args[0], [
+                        'pvesh', 'create',
+                        f'/nodes/fixture/{"lxc" if vmid == 100 else "qemu"}/{vmid}/status/shutdown',
+                    ])
+        retained.assert_called_once_with(100, TASK.replace('vzreboot:100', 'hastop:100'), lock)
+
     def test_control_post_rejects_success_without_a_confirmable_task(self):
         lock = Mock()
         module = SimpleNamespace(acquire=lambda _: lock, begin_control=Mock(),