fix: confirm VM control tasks safely

tests/lxc_updates/test_check_only_job.py

@@ -290,8 +290,13 @@ class CheckOnlyTests(unittest.TestCase):
 
     def test_control_route_uses_same_lock_and_preserves_qemu(self):
         app = Flask(__name__)
-        command = Mock(return_value=subprocess.CompletedProcess(
-            [], 0, '"UPID:fixture:000A:000B:000C:vzstart:100:root@pam:"\n', ''))
+        def control_result(argv, **_):
+            path = argv[2]
+            vm_type, vmid, action = path.split('/')[3], path.split('/')[4], path.rsplit('/', 1)[1]
+            task_type = ('vz' if vm_type == 'lxc' else 'qm') + action
+            return subprocess.CompletedProcess(
+                [], 0, f'"UPID:fixture:000A:000B:000C:{task_type}:{vmid}:root@pam:"\n', '')
+        command = Mock(side_effect=control_result)
         scope = {'app': app, 'jsonify': jsonify, 'request': request,
                  'require_auth': lambda fn: fn,
                  'get_cached_pvesh_cluster_resources_vm': lambda: [

tests/lxc_updates/test_control_task_reservation.py

@@ -109,18 +109,38 @@ class ControlReservationTests(unittest.TestCase):
             job.acquire(100).close()
             probe.assert_not_called()
 
-    def test_definite_failure_clears_marker_but_timeout_or_missing_upid_cannot_confirm_completion(self):
+    def test_definite_failure_or_unidentified_submission_never_leaves_a_permanent_reservation(self):
         self.route.run.return_value = self.result({}, code=1)
         self.assertEqual(self.route.client.post('/api/vms/100/control', json={'action': 'start'}).status_code, 500)
         self.assertEqual(self.lock_path.read_text(), '')
         for response in [subprocess.TimeoutExpired('pvesh', 30), self.result('')]:
             self.route.run.side_effect = response if isinstance(response, Exception) else None
             self.route.run.return_value = response
-            self.route.client.post('/api/vms/100/control', json={'action': 'start'})
-            self.assertIsNone(json.loads(self.lock_path.read_text())['task'])
-            with self.assertRaises(BlockingIOError):
-                job.acquire(100)
-            self.lock_path.write_text('')  # fixture reset after explicit assertions
+            response = self.route.client.post('/api/vms/100/control', json={'action': 'start'})
+            self.assertIn(response.status_code, (500, 502), response.json)
+            self.assertEqual(self.lock_path.read_text(), '')
+            job.acquire(100).close()
+
+    def test_ha_task_recovers_like_a_native_lxc_task(self):
+        task = TASK.replace('vzreboot', 'hastop')
+        self.lock_path.write_text(json.dumps({'node': 'fixture', 'task': task}))
+        with patch.object(job.subprocess, 'run', return_value=self.result(
+                {'status': 'stopped', 'exitstatus': 'OK'})) as probe:
+            job.acquire(100).close()
+        self.assertEqual(probe.call_count, 1)
+        self.assertEqual(self.lock_path.read_text(), '')
+
+    def test_wait_deadline_releases_fd_but_keeps_known_task_for_recovery(self):
+        handle = self.accepted()
+        with patch.object(job, 'CONTROL_TASK_WAIT_SECONDS', 0):
+            job._wait_control(100, handle)
+        self.assertTrue(handle.closed)
+        record = json.loads(self.lock_path.read_text())
+        self.assertEqual(record['task'], TASK)
+        with patch.object(job.subprocess, 'run', return_value=self.result(
+                {'status': 'stopped', 'exitstatus': 'OK'})):
+            job.acquire(100).close()
+        self.assertEqual(self.lock_path.read_text(), '')
 
     def test_thread_launch_failure_retains_marker_for_recovery(self):
         self.thread.start.side_effect = RuntimeError('thread failed')

tests/lxc_updates/test_vm_control_task.py

@@ -57,12 +57,33 @@ class ControlTaskTests(unittest.TestCase):
             self.assertEqual(retained.call_count, 4)
             lock.close.assert_not_called()
 
-    def test_legacy_acceptance_without_task_and_invalid_action(self):
-        self.run.return_value = subprocess.CompletedProcess([], 0, '', '')
-        response = self.client.post('/api/vms/200/control', json={'action': 'start'})
-        self.assertEqual(response.status_code, 200)
-        self.assertTrue(response.json['success'])
-        self.assertIsNone(response.json['task'])
+    def test_control_post_reads_final_upid_after_pvesh_log_output(self):
+        lock = Mock()
+        retained = Mock()
+        module = SimpleNamespace(acquire=lambda _: lock, begin_control=Mock(),
+                                 retain_control=retained, cancel_control=Mock())
+        task = TASK.replace('vzreboot', 'vzstart')
+        self.run.return_value = subprocess.CompletedProcess(
+            [], 0, f'waiting for server lock\n"{task}"\n', '')
+        with patch.dict('sys.modules', {'lxc_check_job': module}):
+            response = self.client.post('/api/vms/100/control', json={'action': 'start'})
+        self.assertEqual(response.status_code, 200, response.json)
+        self.assertEqual(response.json['task'], task)
+        retained.assert_called_once_with(100, task, lock)
+
+    def test_control_post_rejects_success_without_a_confirmable_task(self):
+        lock = Mock()
+        module = SimpleNamespace(acquire=lambda _: lock, begin_control=Mock(),
+                                 retain_control=Mock(), cancel_control=Mock())
+        self.run.return_value = subprocess.CompletedProcess([], 0, 'waiting for server lock\n', '')
+        with patch.dict('sys.modules', {'lxc_check_job': module}):
+            response = self.client.post('/api/vms/100/control', json={'action': 'start'})
+        self.assertEqual(response.status_code, 502, response.json)
+        self.assertFalse(response.json['success'])
+        self.assertIn('task', response.json['error'].lower())
+        module.retain_control.assert_not_called()
+        module.cancel_control.assert_called_once_with(lock)
+        lock.close.assert_called_once()
         self.run.reset_mock()
         self.assertEqual(self.client.post('/api/vms/100/control', json={'action': 'destroy'}).status_code, 400)
         self.run.assert_not_called()
@@ -91,6 +112,18 @@ class ControlTaskTests(unittest.TestCase):
                     self.assertEqual(self.run.call_count, 2)
                     self.assertTrue(all(call.args[0][:2] == ['pvesh', 'get'] for call in self.run.call_args_list))
 
+    def test_ha_start_and_stop_tasks_belong_to_the_guest(self):
+        for action, expected in [('hastart', 'running'), ('hastop', 'stopped')]:
+            with self.subTest(action=action):
+                self.run.reset_mock()
+                self.run.side_effect = [self.result(dict(status='stopped', exitstatus='OK')),
+                                        self.result(dict(status=expected))]
+                task = TASK.replace('vzreboot', action)
+                response = self.get(task)
+                self.assertEqual(response.status_code, 200, response.json)
+                self.assertEqual(response.json['guest_status'], expected)
+                self.assertEqual(self.run.call_count, 2)
+
     def test_wrong_guest_node_type_and_path_make_zero_probes(self):
         for task in [TASK.replace(':100:', ':200:'), TASK.replace('fixture', 'other'),
                      TASK.replace('vzreboot', 'qmreboot'), TASK.replace('vzreboot', 'vzdump'),