From: Guillaume Abrioux <gabrioux@redhat.com>
Date: Wed, 4 Nov 2020 14:11:58 +0000 (+0100)
Subject: ceph-volume: add a unit tests to lvm batch
X-Git-Tag: v14.2.15~4^2
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=f0297fa9a57e31a955e60c17243385079f8960f2;p=ceph.git

ceph-volume: add a unit tests to lvm batch

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>
(cherry picked from commit 13514a24cfdc32d67cfbc1201aa427168a926978)
---

diff --git a/src/ceph-volume/ceph_volume/tests/devices/lvm/test_batch.py b/src/ceph-volume/ceph_volume/tests/devices/lvm/test_batch.py
index 4987a85f5eae..44a05bfa8287 100644
--- a/src/ceph-volume/ceph_volume/tests/devices/lvm/test_batch.py
+++ b/src/ceph-volume/ceph_volume/tests/devices/lvm/test_batch.py
@@ -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,