1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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(mapping='123', skip_path_check=True)
|
