lvm/deactivate: add unit tests, remove --all

Remove the --all flag until its actually implemented.

Fixes: https://tracker.ceph.com/issues/43330
Signed-off-by: Jan Fajerski <jfajerski@suse.com>
(cherry picked from commit c13901f963d515fd34bd0ff9df67bcf214507b70)

 Conflicts:
	src/ceph-volume/ceph_volume/devices/lvm/deactivate.py

diff --git a/src/ceph-volume/ceph_volume/devices/lvm/deactivate.py b/src/ceph-volume/ceph_volume/devices/lvm/deactivate.py
index b3f42faa94d010a7f94db925560fe56204afc645..5de6dbe36b5a15e3a90bac0d4a99d097496eccf1 100644 (file)
--- a/src/ceph-volume/ceph_volume/devices/lvm/deactivate.py
+++ b/src/ceph-volume/ceph_volume/devices/lvm/deactivate.py
@@ -12,16 +12,14 @@ logger = logging.getLogger(__name__)
 def deactivate_osd(osd_id=None, osd_uuid=None):
 
     lvs = []
-    if osd_uuid:
+    if osd_uuid is not None:
         lvs = get_lvs_by_tag('ceph.osd_fsid={}'.format(osd_uuid))
+        osd_id = next(lv.tags['ceph.osd_id'] for lv in lvs)
     else:
         lvs = get_lvs_by_tag('ceph.osd_id={}'.format(osd_id))
 
-    try:
-        data_lv = next(lv for lv in lvs if lv.tags['ceph.type'] in ['data', 'block'])
-    except StopIteration:
-        logger.error('No data or block LV found for OSD {}'.format(osd_id))
-        sys.exit(1)
+    data_lv = next(lv for lv in lvs if lv.tags['ceph.type'] in ['data', 'block'])
+
     conf.cluster = data_lv.tags['ceph.cluster_name']
     logger.debug('Found cluster name {}'.format(conf.cluster))
 
@@ -40,7 +38,12 @@ class Deactivate(object):
     def deactivate(self, args=None):
         if args:
             self.args = args
-        deactivate_osd(self.args.osd_id, self.args.osd_uuid)
+        try:
+            deactivate_osd(self.args.osd_id, self.args.osd_uuid)
+        except StopIteration:
+            logger.error(('No data or block LV found for OSD'
+                          '{}').format(self.args.osd_id))
+            sys.exit(1)
 
     def __init__(self, argv):
         self.argv = argv
@@ -70,21 +73,18 @@ class Deactivate(object):
             nargs='?',
             help='The UUID of the OSD, similar to a SHA1, takes precedence over osd_id'
         )
-        parser.add_argument(
-            '--all',
-            action='store_true',
-            help='Deactivate all OSD volumes found in the system',
-        )
+        # parser.add_argument(
+        #     '--all',
+        #     action='store_true',
+        #     help='Deactivate all OSD volumes found in the system',
+        # )
         if len(self.argv) == 0:
             print(sub_command_help)
             return
         args = parser.parse_args(self.argv)
         # Default to bluestore here since defaulting it in add_argument may
         # cause both to be True
-        if args.all:
-            self.deactivate_all(args)
-        else:
-            if not args.osd_id and not args.osd_uuid:
-                raise ValueError(('Can not identify OSD, pass either all or'
-                                 'osd_id or osd_uuid'))
-            self.deactivate(args)
+        if not args.osd_id and not args.osd_uuid:
+            raise ValueError(('Can not identify OSD, pass either all or'
+                             'osd_id or osd_uuid'))
+        self.deactivate(args)

diff --git a/src/ceph-volume/ceph_volume/tests/devices/lvm/test_deactivate.py b/src/ceph-volume/ceph_volume/tests/devices/lvm/test_deactivate.py
new file mode 100644 (file)
index 0000000..4b8304c
--- /dev/null
+++ b/src/ceph-volume/ceph_volume/tests/devices/lvm/test_deactivate.py
@@ -0,0 +1,59 @@
+import pytest
+from mock.mock import patch
+from ceph_volume.api import lvm
+from ceph_volume.devices.lvm import deactivate
+
+class TestDeactivate(object):
+
+    @patch("ceph_volume.devices.lvm.deactivate.get_lvs_by_tag")
+    def test_no_osd(self, p_get_lvs):
+        p_get_lvs.return_value = []
+        with pytest.raises(StopIteration):
+            deactivate.deactivate_osd(0)
+
+    @patch("ceph_volume.devices.lvm.deactivate.get_lvs_by_tag")
+    @patch("ceph_volume.util.system.unmount_tmpfs")
+    def test_unmount_tmpfs_called_osd_id(self, p_u_tmpfs, p_get_lvs):
+        FooVolume = lvm.Volume(
+            lv_name='foo', lv_path='/dev/vg/foo',
+            lv_tags="ceph.osd_id=0,ceph.cluster_name=foo,ceph.type=data")
+        p_get_lvs.return_value = [FooVolume]
+
+        deactivate.deactivate_osd(0)
+        p_u_tmpfs.assert_called_with(
+            '/var/lib/ceph/osd/{}-{}'.format('foo', 0))
+
+    @patch("ceph_volume.devices.lvm.deactivate.get_lvs_by_tag")
+    @patch("ceph_volume.util.system.unmount_tmpfs")
+    def test_unmount_tmpfs_called_osd_uuid(self, p_u_tmpfs, p_get_lvs):
+        FooVolume = lvm.Volume(
+            lv_name='foo', lv_path='/dev/vg/foo',
+            lv_tags="ceph.osd_fsid=0,ceph.osd_id=1,ceph.cluster_name=foo,ceph.type=data")
+        p_get_lvs.return_value = [FooVolume]
+
+        deactivate.deactivate_osd(None, 0)
+        p_u_tmpfs.assert_called_with(
+            '/var/lib/ceph/osd/{}-{}'.format('foo', 1))
+
+    @patch("ceph_volume.devices.lvm.deactivate.get_lvs_by_tag")
+    @patch("ceph_volume.util.system.unmount_tmpfs")
+    @patch("ceph_volume.util.encryption.dmcrypt_close")
+    def test_no_crypt_no_dmclose(self, p_dm_close, p_u_tmpfs, p_get_lvs):
+        FooVolume = lvm.Volume(
+            lv_name='foo', lv_path='/dev/vg/foo',
+            lv_tags="ceph.osd_id=0,ceph.cluster_name=foo,ceph.type=data")
+        p_get_lvs.return_value = [FooVolume]
+
+        deactivate.deactivate_osd(0)
+
+    @patch("ceph_volume.devices.lvm.deactivate.get_lvs_by_tag")
+    @patch("ceph_volume.util.system.unmount_tmpfs")
+    @patch("ceph_volume.util.encryption.dmcrypt_close")
+    def test_crypt_dmclose(self, p_dm_close, p_u_tmpfs, p_get_lvs):
+        FooVolume = lvm.Volume(
+            lv_name='foo', lv_path='/dev/vg/foo', lv_uuid='123',
+            lv_tags="ceph.osd_id=0,ceph.encrypted=1,ceph.cluster_name=foo,ceph.type=data")
+        p_get_lvs.return_value = [FooVolume]
+
+        deactivate.deactivate_osd(0)
+        p_dm_close.assert_called_with('123')