]> git.apps.os.sepia.ceph.com Git - ceph.git/commitdiff
ceph-volume: add a unit tests to lvm batch
authorGuillaume Abrioux <gabrioux@redhat.com>
Wed, 4 Nov 2020 14:11:58 +0000 (15:11 +0100)
committerDimitri Savineau <dsavinea@redhat.com>
Mon, 9 Nov 2020 15:54:50 +0000 (10:54 -0500)
This commit adds unit tests in order to cover `_sort_rotational_disks()`
call when deploying with full hdd/ssd or mixed hdd/sdd scenarios.

Fixes: https://tracker.ceph.com/issues/48150
Signed-off-by: Guillaume Abrioux <gabrioux@redhat.com>
Co-authored-by: Dimitri Savineau <dsavinea@redhat.com>
src/ceph-volume/ceph_volume/tests/devices/lvm/test_batch.py

index 4987a85f5eae99009c259f85c740080f2818104f..44a05bfa82872d0e0d263829ff31e3aca5d873b5 100644 (file)
@@ -116,6 +116,40 @@ class TestBatch(object):
         report = b._create_report(plan)
         json.loads(report)
 
+    @pytest.mark.parametrize('rota', [0, 1])
+    def test_batch_sort_full(self, factory, rota):
+        device1 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sda")
+        device2 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sdb")
+        device3 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sdc")
+        devices = [device1, device2, device3]
+        args = factory(report=True,
+                       devices=devices,
+                       filestore=False,
+                      )
+        b = batch.Batch([])
+        b.args = args
+        b._sort_rotational_disks()
+        assert len(b.args.devices) == 3
+
+    @pytest.mark.parametrize('objectstore', ['bluestore', 'filestore'])
+    def test_batch_sort_mixed(self, factory, objectstore):
+        device1 = factory(used_by_ceph=False, available=True, rotational=1, abspath="/dev/sda")
+        device2 = factory(used_by_ceph=False, available=True, rotational=1, abspath="/dev/sdb")
+        device3 = factory(used_by_ceph=False, available=True, rotational=0, abspath="/dev/sdc")
+        devices = [device1, device2, device3]
+        args = factory(report=True,
+                       devices=devices,
+                       filestore=False if objectstore == 'bluestore' else True,
+                      )
+        b = batch.Batch([])
+        b.args = args
+        b._sort_rotational_disks()
+        assert len(b.args.devices) == 2
+        if objectstore == 'bluestore':
+            assert len(b.args.db_devices) == 1
+        else:
+            assert len(b.args.journal_devices) == 1
+
     def test_get_physical_osds_return_len(self, factory,
                                           mock_devices_available,
                                           conf_ceph_stub,