CA-393194: Fix pvremove failure

drivers/lvutil.py

@@ -529,6 +529,27 @@ def createVG(root, vgname):
 
     # End block
 
+def getPVsInVG(vgname):
+    # Get PVs in a specific VG
+    pvs_ret = cmd_lvm([CMD_PVS, '--separator', ' ', '--noheadings', '-o', 'pv_name,vg_name'])
+
+    # Parse each line to extract PV and VG information
+    # No need to handle exceptions here, return empty list if any error
+    pvs_in_vg = []
+    lines = pvs_ret.strip().split('\n')
+    for line in lines:
+        # To avoid invalid return format
+        parts = line.split()
+        if len(parts) != 2:
+            util.SMlog("Warning: Invalid or empty line in pvs output: %s" % line)
+            continue
+        pv, vg = parts
+        if vg == vgname:
+            pvs_in_vg.append(pv)
+
+    util.SMlog("PVs in VG %s: %s" % (vgname, pvs_in_vg))
+    return pvs_in_vg
+
 def removeVG(root, vgname):
     # Check PVs match VG
     try:
@@ -542,9 +563,11 @@ def removeVG(root, vgname):
               opterr='error is %d' % inst.code)
 
     try:
+        # Get PVs in VG before removing the VG
+        devs_in_vg = getPVsInVG(vgname)
         cmd_lvm([CMD_VGREMOVE, vgname])
 
-        for dev in root.split(','):
+        for dev in devs_in_vg:
             cmd_lvm([CMD_PVREMOVE, dev])
     except util.CommandException as inst:
         raise xs_errors.XenError('LVMDelete', \

tests/test_lvutil.py

@@ -355,3 +355,41 @@ def test_warning_if_cmd_takes_too_long(self, m_smlog, m_time, _1, m_pread):
         self.assertIn("Long LVM call", m_smlog.call_args[0][0])
         self.assertIn(f"took {lvutil.MAX_OPERATION_DURATION*2}", m_smlog.call_args[0][0])
 
+@mock.patch('lvutil.cmd_lvm')
+@mock.patch('util.SMlog', autospec=True)
+class TestGetPVsInVG(unittest.TestCase):
+
+    def test_pvs_in_vg(self, mock_smlog, mock_cmd_lvm):
+        # Normal case
+        mock_cmd_lvm.return_value = "pv1 vg1\npv2 vg1\npv3 vg2"
+        result = lvutil.getPVsInVG("vg1")
+        self.assertEqual(result, ["pv1", "pv2"])
+        mock_smlog.assert_called_once_with("PVs in VG vg1: ['pv1', 'pv2']")
+
+    def test_no_pvs(self, mock_smlog, mock_cmd_lvm):
+        # Test when no PVs are returned
+        mock_cmd_lvm.return_value = ""
+        result = lvutil.getPVsInVG("vg1")
+        self.assertEqual(result, [])
+        mock_smlog.assert_has_calls([
+            mock.call("Warning: Invalid or empty line in pvs output: "),
+            mock.call("PVs in VG vg1: []")
+        ])
+
+    def test_no_pvs_in_vg(self, mock_smlog, mock_cmd_lvm):
+        # Test when no PVs belong to the specified VG
+        mock_cmd_lvm.return_value = "pv1 vg2\npv2 vg2"
+        result = lvutil.getPVsInVG("vg1")
+        self.assertEqual(result, [])
+        mock_smlog.assert_called_once_with("PVs in VG vg1: []")
+
+    def test_command_error(self, mock_smlog, mock_cmd_lvm):
+        # Test invalid return value from cmd_lvm
+        mock_cmd_lvm.return_value = "Invalid retrun value."
+        result = lvutil.getPVsInVG("vg1")
+        self.assertEqual(result, [])
+        mock_smlog.assert_has_calls([
+            mock.call("Warning: Invalid or empty line in pvs output: Invalid retrun value."),
+            mock.call("PVs in VG vg1: []")
+        ])
+        mock_smlog.assert_called_with("PVs in VG vg1: []")