CA-393194: Fix pvremove failure

In lvmohba test, the test failed at
`SR.destroy(sr_ref) -> removeVG -> pvremove ` with the
lvm error message "No devices to process".
The root cause is that the PV which was to be removed
was not really used as PV when it comes to multipathed devices.
The LVM official document says:
"Since each LUN has multiple device nodes in /dev 
that point to the same underlying data,
they all contain the same LVM metadata
and thus LVM commands will find the same metadata
multiple times and report them as duplicates.
These duplicate messages are only warnings
and do not mean the LVM operation has failed.
Rather, they are alerting the user that only one of the devices
has been used as a physical volume and the others are being ignored." 
Please refer to https://docs.redhat.com/en/documentation/
red_hat_enterprise_linux/6/html/logical_volume_manager_administration
/duplicate_pv_multipath#duplicate_pv_multipath

This fix is to find the real PV in a VG before removing the VG.

Signed-off-by: Stephen Cheng <stephen.cheng@cloud.com>

Author: stephenchengCloud

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: []")