ceph-volume: combine the LVPath arg validator with ValidDevice

This combines the two arg validators and makes both batch and
prepare/active both use ValidDevice.

This will also allow us to use lvs with a full path, like
/dev/vg/lv instead of always enforcing vg/lv.

Signed-off-by: Andrew Schoen <[email protected]>

Resolves: rm#27062

src/ceph-volume/ceph_volume/devices/lvm/common.py

@@ -52,7 +52,7 @@ def common_parser(prog, description):
     required_group.add_argument(
         '--data',
         required=True,
-        type=arg_validators.LVPath(),
+        type=arg_validators.ValidDevice(as_string=True),
         help='OSD data path. A physical device or logical volume',
     )
 

src/ceph-volume/ceph_volume/util/arg_validators.py

@@ -6,51 +6,24 @@
 from ceph_volume.util.device import Device
 
 
-class LVPath(object):
-    """
-    A simple validator to ensure that a logical volume is specified like::
-
-        <vg name>/<lv name>
-
-    Or a full path to a device, like ``/dev/sda``
+class ValidDevice(object):
 
-    Because for LVM it is better to be specific on what group does an lv
-    belongs to.
-    """
+    def __init__(self, as_string=False):
+        self.as_string = as_string
 
     def __call__(self, string):
+        device = Device(string)
         error = None
-        if string.startswith('/'):
-            if not os.path.exists(string):
-                error = "Argument (device) does not exist: %s" % string
-                raise argparse.ArgumentError(None, error)
-            else:
-                return string
-        try:
-            vg, lv = string.split('/')
-        except ValueError:
-            error = "Logical volume must be specified as 'volume_group/logical_volume' but got: %s" % string
-            raise argparse.ArgumentError(None, error)
-
-        if not vg:
-            error = "Didn't specify a volume group like 'volume_group/logical_volume', got: %s" % string
-        if not lv:
-            error = "Didn't specify a logical volume like 'volume_group/logical_volume', got: %s" % string
+        if not device.exists:
+            error = "Unable to proceed with non-existing device: %s" % string
+        elif device.has_gpt_headers:
+            error = "GPT headers found, they must be removed on: %s" % string
 
         if error:
             raise argparse.ArgumentError(None, error)
-        return string
-
-
-class ValidDevice(object):
-
-    def __call__(self, string):
-        device = Device(string)
-        if not device.exists:
-            raise argparse.ArgumentError(
-                None, "Unable to proceed with non-existing device: %s" % string
-            )
 
+        if self.as_string:
+            return string
         return device