ceph-volume: extracts batch.filter_devices from Batch._get_strategy

This allows us to easily provide tests for that method.

Signed-off-by: Andrew Schoen <aschoen@redhat.com>

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

@@ -94,6 +94,28 @@ def get_strategy(args, devices):
             return backend
 
 
+def filter_devices(args):
+    unused_devices = [device for device in args.devices if not device.used_by_ceph]
+    # only data devices, journals can be reused
+    used_devices = [device.abspath for device in args.devices if device.used_by_ceph]
+    args.filtered_devices = {}
+    if used_devices:
+        for device in used_devices:
+            args.filtered_devices[device] = {"reasons": ["Used by ceph as a data device already"]}
+        logger.info("Ignoring devices already used by ceph: %s" % ", ".join(used_devices))
+    if len(unused_devices) == 1:
+        last_device = unused_devices[0]
+        if not last_device.rotational and last_device.is_lvm_member:
+            reason = "Used by ceph as a %s already and there are no devices left for data/block" % (
+                last_device.lvs[0].tags.get("ceph.type"),
+            )
+            args.filtered_devices[last_device.abspath] = {"reasons": [reason]}
+            logger.info(reason + ": %s" % last_device.abspath)
+            unused_devices = []
+
+    return unused_devices
+
+
 class Batch(object):
 
     help = 'Automatically size devices for multi-OSD provisioning with minimal interaction'
@@ -154,23 +176,7 @@ class Batch(object):
 
     def _get_strategy(self, args):
         strategy = get_strategy(args, args.devices)
-        unused_devices = [device for device in args.devices if not device.used_by_ceph]
-        # only data devices, journals can be reused
-        used_devices = [device.abspath for device in args.devices if device.used_by_ceph]
-        args.filtered_devices = {}
-        if used_devices:
-            for device in used_devices:
-                args.filtered_devices[device] = {"reasons": ["Used by ceph as a data device already"]}
-            logger.info("Ignoring devices already used by ceph: %s" % ", ".join(used_devices))
-        if len(unused_devices) == 1:
-            last_device = unused_devices[0]
-            if not last_device.rotational and last_device.is_lvm_member:
-                reason = "Used by ceph as a %s already and there are no devices left for data/block" % (
-                    last_device.lvs[0].tags.get("ceph.type"),
-                )
-                args.filtered_devices[last_device.abspath] = {"reasons": [reason]}
-                logger.info(reason + ": %s" % last_device.abspath)
-                unused_devices = []
+        unused_devices = filter_devices(args)
         if not unused_devices and not args.format == 'json':
             # report nothing changed
             mlogger.info("All devices are already used by ceph. No OSDs will be created.")

src/ceph-volume/ceph_volume/tests/devices/lvm/test_batch.py

@@ -0,0 +1,61 @@
+from ceph_volume.devices.lvm import batch
+
+
+class TestFilterDevices(object):
+
+    def test_filter_used_device(self, factory):
+        device1 = factory(used_by_ceph=True, abspath="/dev/sda")
+        args = factory(devices=[device1], filtered_devices={})
+        result = batch.filter_devices(args)
+        assert not result
+        assert device1.abspath in args.filtered_devices
+
+    def test_has_unused_devices(self, factory):
+        device1 = factory(
+            used_by_ceph=False,
+            abspath="/dev/sda",
+            rotational=False,
+            is_lvm_member=False
+        )
+        args = factory(devices=[device1], filtered_devices={})
+        result = batch.filter_devices(args)
+        assert device1 in result
+        assert not args.filtered_devices
+
+    def test_filter_device_used_as_a_journal(self, factory):
+        hdd1 = factory(
+            used_by_ceph=True,
+            abspath="/dev/sda",
+            rotational=True,
+            is_lvm_member=True,
+        )
+        lv = factory(tags={"ceph.type": "journal"})
+        ssd1 = factory(
+            used_by_ceph=False,
+            abspath="/dev/nvme0n1",
+            rotational=False,
+            is_lvm_member=True,
+            lvs=[lv],
+        )
+        args = factory(devices=[hdd1, ssd1], filtered_devices={})
+        result = batch.filter_devices(args)
+        assert not result
+        assert ssd1.abspath in args.filtered_devices
+
+    def test_last_device_is_not_filtered(self, factory):
+        hdd1 = factory(
+            used_by_ceph=True,
+            abspath="/dev/sda",
+            rotational=True,
+            is_lvm_member=True,
+        )
+        ssd1 = factory(
+            used_by_ceph=False,
+            abspath="/dev/nvme0n1",
+            rotational=False,
+            is_lvm_member=False,
+        )
+        args = factory(devices=[hdd1, ssd1], filtered_devices={})
+        result = batch.filter_devices(args)
+        assert result
+        assert len(args.filtered_devices) == 1