Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

7 files changed, 2542 insertions, 0 deletions

diff --git a/src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py b/src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py
new file mode 100644
index 000000000..59ca12619
--- /dev/null
+++ b/src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py
@@ -0,0 +1,338 @@
+import argparse
+import pytest
+import os
+from ceph_volume import exceptions, process
+from ceph_volume.util import arg_validators
+from mock.mock import patch, MagicMock
+
+
+class TestOSDPath(object):
+
+    def setup(self):
+        self.validator = arg_validators.OSDPath()
+
+    def test_is_not_root(self, monkeypatch):
+        monkeypatch.setattr(os, 'getuid', lambda: 100)
+        with pytest.raises(exceptions.SuperUserError):
+            self.validator('')
+
+    def test_path_is_not_a_directory(self, is_root, monkeypatch, fake_filesystem):
+        fake_file = fake_filesystem.create_file('/tmp/foo')
+        monkeypatch.setattr(arg_validators.disk, 'is_partition', lambda x: False)
+        validator = arg_validators.OSDPath()
+        with pytest.raises(argparse.ArgumentError):
+            validator(fake_file.path)
+
+    def test_files_are_missing(self, is_root, tmpdir, monkeypatch):
+        tmppath = str(tmpdir)
+        monkeypatch.setattr(arg_validators.disk, 'is_partition', lambda x: False)
+        validator = arg_validators.OSDPath()
+        with pytest.raises(argparse.ArgumentError) as error:
+            validator(tmppath)
+        assert 'Required file (ceph_fsid) was not found in OSD' in str(error.value)
+
+
+class TestExcludeGroupOptions(object):
+
+    def setup(self):
+        self.parser = argparse.ArgumentParser()
+
+    def test_flags_in_one_group(self):
+        argv = ['<prog>', '--filestore', '--bar']
+        filestore_group = self.parser.add_argument_group('filestore')
+        bluestore_group = self.parser.add_argument_group('bluestore')
+        filestore_group.add_argument('--filestore')
+        bluestore_group.add_argument('--bluestore')
+        result = arg_validators.exclude_group_options(
+            self.parser,
+            ['filestore', 'bluestore'],
+            argv=argv
+        )
+        assert result is None
+
+    def test_flags_in_no_group(self):
+        argv = ['<prog>', '--foo', '--bar']
+        filestore_group = self.parser.add_argument_group('filestore')
+        bluestore_group = self.parser.add_argument_group('bluestore')
+        filestore_group.add_argument('--filestore')
+        bluestore_group.add_argument('--bluestore')
+        result = arg_validators.exclude_group_options(
+            self.parser,
+            ['filestore', 'bluestore'],
+            argv=argv
+        )
+        assert result is None
+
+    def test_flags_conflict(self, capsys):
+        argv = ['<prog>', '--filestore', '--bluestore']
+        filestore_group = self.parser.add_argument_group('filestore')
+        bluestore_group = self.parser.add_argument_group('bluestore')
+        filestore_group.add_argument('--filestore')
+        bluestore_group.add_argument('--bluestore')
+
+        arg_validators.exclude_group_options(
+            self.parser, ['filestore', 'bluestore'], argv=argv
+        )
+        stdout, stderr = capsys.readouterr()
+        assert 'Cannot use --filestore (filestore) with --bluestore (bluestore)' in stderr
+
+
+class TestValidDevice(object):
+
+    def setup(self, fake_filesystem):
+        self.validator = arg_validators.ValidDevice()
+
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    def test_path_is_valid(self, m_has_bs_label,
+                           fake_call, patch_bluestore_label,
+                           device_info, monkeypatch):
+        monkeypatch.setattr('ceph_volume.util.device.Device.exists', lambda: True)
+        lsblk = {"TYPE": "disk", "NAME": "sda"}
+        device_info(lsblk=lsblk)
+        result = self.validator('/dev/sda')
+        assert result.path == '/dev/sda'
+
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    def test_path_is_invalid(self, m_has_bs_label,
+                             fake_call, patch_bluestore_label,
+                             device_info):
+        lsblk = {"TYPE": "disk", "NAME": "sda"}
+        device_info(lsblk=lsblk)
+        with pytest.raises(argparse.ArgumentError):
+            self.validator('/device/does/not/exist')
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_dev_has_partitions(self, m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
+        mocked_device.return_value = MagicMock(
+            exists=True,
+            has_partitions=True,
+        )
+        with pytest.raises(RuntimeError):
+            self.validator('/dev/foo')
+
+class TestValidZapDevice(object):
+    def setup(self):
+        self.validator = arg_validators.ValidZapDevice()
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_device_has_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device):
+        mocked_device.return_value = MagicMock(
+            used_by_ceph=False,
+            exists=True,
+            has_partitions=True,
+            has_gpt_headers=False,
+            has_fs=False
+        )
+        self.validator.zap = False
+        with pytest.raises(RuntimeError):
+            assert self.validator('/dev/foo')
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_device_has_no_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device):
+        mocked_device.return_value = MagicMock(
+            used_by_ceph=False,
+            exists=True,
+            has_partitions=False,
+            has_gpt_headers=False,
+            has_fs=False
+        )
+        self.validator.zap = False
+        assert self.validator('/dev/foo')
+
+class TestValidDataDevice(object):
+    def setup(self):
+        self.validator = arg_validators.ValidDataDevice()
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_device_used_by_ceph(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
+        mocked_device.return_value = MagicMock(
+            used_by_ceph=True,
+            exists=True,
+            has_partitions=False,
+            has_gpt_headers=False
+        )
+        with pytest.raises(SystemExit):
+            self.validator.zap = False
+            self.validator('/dev/foo')
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_device_has_fs(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
+        mocked_device.return_value = MagicMock(
+            used_by_ceph=False,
+            exists=True,
+            has_partitions=False,
+            has_gpt_headers=False,
+            has_fs=True
+        )
+        with pytest.raises(RuntimeError):
+            self.validator.zap = False
+            self.validator('/dev/foo')
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=True)
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_device_has_bs_signature(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
+        mocked_device.return_value = MagicMock(
+            used_by_ceph=False,
+            exists=True,
+            has_partitions=False,
+            has_gpt_headers=False,
+            has_fs=False
+        )
+        with pytest.raises(RuntimeError):
+            self.validator.zap = False
+            self.validator('/dev/foo')
+
+class TestValidRawDevice(object):
+    def setup(self):
+        self.validator = arg_validators.ValidRawDevice()
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    @patch('ceph_volume.util.arg_validators.disk.blkid')
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_dmcrypt_device_already_prepared(self,  m_get_single_lv, m_blkid, m_has_bs_label, mocked_device, fake_call, monkeypatch):
+        def mock_call(cmd, **kw):
+            return ('', '', 1)
+        monkeypatch.setattr(process, 'call', mock_call)
+        m_blkid.return_value = {'UUID': '8fd92779-ad78-437c-a06f-275f7170fa74', 'TYPE': 'crypto_LUKS'}
+        mocked_device.return_value = MagicMock(
+            used_by_ceph=False,
+            exists=True,
+            has_partitions=False,
+            has_gpt_headers=False,
+            has_fs=False
+        )
+        with pytest.raises(SystemExit):
+            self.validator.zap = False
+            self.validator('/dev/foo')
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_device_already_prepared(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
+        mocked_device.return_value = MagicMock(
+            used_by_ceph=False,
+            exists=True,
+            has_partitions=False,
+            has_gpt_headers=False,
+            has_fs=False
+        )
+        with pytest.raises(SystemExit):
+            self.validator.zap = False
+            self.validator('/dev/foo')
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_device_not_prepared(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call, monkeypatch):
+        def mock_call(cmd, **kw):
+            return ('', '', 1)
+        monkeypatch.setattr(process, 'call', mock_call)
+        mocked_device.return_value = MagicMock(
+            used_by_ceph=False,
+            exists=True,
+            has_partitions=False,
+            has_gpt_headers=False,
+            has_fs=False
+        )
+        self.validator.zap = False
+        assert self.validator('/dev/foo')
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_device_has_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call, monkeypatch):
+        def mock_call(cmd, **kw):
+            return ('', '', 1)
+        monkeypatch.setattr(process, 'call', mock_call)
+        mocked_device.return_value = MagicMock(
+            used_by_ceph=False,
+            exists=True,
+            has_partitions=True,
+            has_gpt_headers=False,
+            has_fs=False
+        )
+        self.validator.zap = False
+        with pytest.raises(RuntimeError):
+            assert self.validator('/dev/foo')
+
+class TestValidBatchDevice(object):
+    def setup(self):
+        self.validator = arg_validators.ValidBatchDevice()
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_device_is_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
+        mocked_device.return_value = MagicMock(
+            used_by_ceph=False,
+            exists=True,
+            has_partitions=False,
+            has_gpt_headers=False,
+            has_fs=False,
+            is_partition=True
+        )
+        with pytest.raises(argparse.ArgumentError):
+            self.validator.zap = False
+            self.validator('/dev/foo')
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_device_is_not_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
+        mocked_device.return_value = MagicMock(
+            used_by_ceph=False,
+            exists=True,
+            has_partitions=False,
+            has_gpt_headers=False,
+            has_fs=False,
+            is_partition=False
+        )
+        self.validator.zap = False
+        assert self.validator('/dev/foo')
+
+class TestValidBatchDataDevice(object):
+    def setup(self):
+        self.validator = arg_validators.ValidBatchDataDevice()
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_device_is_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
+        mocked_device.return_value = MagicMock(
+            used_by_ceph=False,
+            exists=True,
+            has_partitions=False,
+            has_gpt_headers=False,
+            has_fs=False,
+            is_partition=True
+        )
+        with pytest.raises(argparse.ArgumentError):
+            self.validator.zap = False
+            assert self.validator('/dev/foo')
+
+    @patch('ceph_volume.util.arg_validators.Device')
+    @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
+    @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
+    def test_device_is_not_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
+        mocked_device.return_value = MagicMock(
+            used_by_ceph=False,
+            exists=True,
+            has_partitions=False,
+            has_gpt_headers=False,
+            has_fs=False,
+            is_partition=False
+        )
+        self.validator.zap = False
+        assert self.validator('/dev/foo')
diff --git a/src/ceph-volume/ceph_volume/tests/util/test_device.py b/src/ceph-volume/ceph_volume/tests/util/test_device.py
new file mode 100644
index 000000000..8eef3ff00
--- /dev/null
+++ b/src/ceph-volume/ceph_volume/tests/util/test_device.py
@@ -0,0 +1,704 @@
+import os
+import pytest
+from copy import deepcopy
+from ceph_volume.util import device
+from ceph_volume.api import lvm as api
+from mock.mock import patch, mock_open
+
+
+class TestDevice(object):
+
+    def test_sys_api(self, monkeypatch, device_info):
+        volume = api.Volume(lv_name='lv', lv_uuid='y', vg_name='vg',
+                            lv_tags={}, lv_path='/dev/VolGroup/lv')
+        volumes = []
+        volumes.append(volume)
+        monkeypatch.setattr(api, 'get_lvs', lambda **kwargs:
+                            deepcopy(volumes))
+
+        data = {"/dev/sda": {"foo": "bar"}}
+        lsblk = {"TYPE": "disk", "NAME": "sda"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.sys_api
+        assert "foo" in disk.sys_api
+
+    def test_lvm_size(self, monkeypatch, device_info):
+        volume = api.Volume(lv_name='lv', lv_uuid='y', vg_name='vg',
+                            lv_tags={}, lv_path='/dev/VolGroup/lv')
+        volumes = []
+        volumes.append(volume)
+        monkeypatch.setattr(api, 'get_lvs', lambda **kwargs:
+                            deepcopy(volumes))
+
+        # 5GB in size
+        data = {"/dev/sda": {"size": "5368709120"}}
+        lsblk = {"TYPE": "disk", "NAME": "sda"}
+        device_info(devices=data,lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.lvm_size.gb == 4
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_lvm_size_rounds_down(self, fake_call, device_info):
+        # 5.5GB in size
+        data = {"/dev/sda": {"size": "5905580032"}}
+        lsblk = {"TYPE": "disk", "NAME": "sda"}
+        device_info(devices=data,lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.lvm_size.gb == 4
+
+    def test_is_lv(self, fake_call, device_info):
+        data = {"lv_path": "vg/lv", "vg_name": "vg", "name": "lv"}
+        lsblk = {"TYPE": "lvm", "NAME": "vg-lv"}
+        device_info(lv=data,lsblk=lsblk)
+        disk = device.Device("vg/lv")
+        assert disk.is_lv
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_vgs_is_empty(self, fake_call, device_info, monkeypatch):
+        BarPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000",
+                                 pv_tags={})
+        pvolumes = []
+        pvolumes.append(BarPVolume)
+        lsblk = {"TYPE": "disk", "NAME": "sda"}
+        device_info(lsblk=lsblk)
+        monkeypatch.setattr(api, 'get_pvs', lambda **kwargs: {})
+
+        disk = device.Device("/dev/nvme0n1")
+        assert disk.vgs == []
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_vgs_is_not_empty(self, fake_call, device_info, monkeypatch):
+        vg = api.VolumeGroup(pv_name='/dev/nvme0n1', vg_name='foo/bar', vg_free_count=6,
+                             vg_extent_size=1073741824)
+        monkeypatch.setattr(api, 'get_all_devices_vgs', lambda : [vg])
+        lsblk = {"TYPE": "disk", "NAME": "nvme0n1"}
+        device_info(lsblk=lsblk)
+        disk = device.Device("/dev/nvme0n1")
+        assert len(disk.vgs) == 1
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_device_is_device(self, fake_call, device_info):
+        data = {"/dev/sda": {"foo": "bar"}}
+        lsblk = {"TYPE": "device", "NAME": "sda"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.is_device is True
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_loop_device_is_not_device(self, fake_call, device_info):
+        data = {"/dev/loop0": {"foo": "bar"}}
+        lsblk = {"TYPE": "loop"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/loop0")
+        assert disk.is_device is False
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_loop_device_is_device(self, fake_call, device_info):
+        data = {"/dev/loop0": {"foo": "bar"}}
+        lsblk = {"TYPE": "loop"}
+        os.environ["CEPH_VOLUME_ALLOW_LOOP_DEVICES"] = "1"
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/loop0")
+        assert disk.is_device is True
+        del os.environ["CEPH_VOLUME_ALLOW_LOOP_DEVICES"]
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_device_is_rotational(self, fake_call, device_info):
+        data = {"/dev/sda": {"rotational": "1"}}
+        lsblk = {"TYPE": "device", "NAME": "sda"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.rotational
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_device_is_not_rotational(self, fake_call, device_info):
+        data = {"/dev/sda": {"rotational": "0"}}
+        lsblk = {"TYPE": "device", "NAME": "sda"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert not disk.rotational
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_device_is_rotational_lsblk(self, fake_call, device_info):
+        data = {"/dev/sda": {"foo": "bar"}}
+        lsblk = {"TYPE": "device", "ROTA": "1", "NAME": "sda"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.rotational
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_device_is_not_rotational_lsblk(self, fake_call, device_info):
+        data = {"/dev/sda": {"rotational": "0"}}
+        lsblk = {"TYPE": "device", "ROTA": "0", "NAME": "sda"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert not disk.rotational
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_device_is_rotational_defaults_true(self, fake_call, device_info):
+        # rotational will default true if no info from sys_api or lsblk is found
+        data = {"/dev/sda": {"foo": "bar"}}
+        lsblk = {"TYPE": "device", "foo": "bar", "NAME": "sda"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.rotational
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_disk_is_device(self, fake_call, device_info):
+        data = {"/dev/sda": {"foo": "bar"}}
+        lsblk = {"TYPE": "disk", "NAME": "sda"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.is_device is True
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_is_partition(self, fake_call, device_info):
+        data = {"/dev/sda1": {"foo": "bar"}}
+        lsblk = {"TYPE": "part", "NAME": "sda1", "PKNAME": "sda"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda1")
+        assert disk.is_partition
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_mpath_device_is_device(self, fake_call, device_info):
+        data = {"/dev/foo": {"foo": "bar"}}
+        lsblk = {"TYPE": "mpath", "NAME": "foo"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/foo")
+        assert disk.is_device is True
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_is_not_lvm_member(self, fake_call, device_info):
+        data = {"/dev/sda1": {"foo": "bar"}}
+        lsblk = {"TYPE": "part", "NAME": "sda1", "PKNAME": "sda"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda1")
+        assert not disk.is_lvm_member
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_is_lvm_member(self, fake_call, device_info):
+        data = {"/dev/sda1": {"foo": "bar"}}
+        lsblk = {"TYPE": "part", "NAME": "sda1", "PKNAME": "sda"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda1")
+        assert not disk.is_lvm_member
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_is_mapper_device(self, fake_call, device_info):
+        lsblk = {"TYPE": "lvm", "NAME": "foo"}
+        device_info(lsblk=lsblk)
+        disk = device.Device("/dev/mapper/foo")
+        assert disk.is_mapper
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_dm_is_mapper_device(self, fake_call, device_info):
+        lsblk = {"TYPE": "lvm", "NAME": "dm-4"}
+        device_info(lsblk=lsblk)
+        disk = device.Device("/dev/dm-4")
+        assert disk.is_mapper
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_is_not_mapper_device(self, fake_call, device_info):
+        lsblk = {"TYPE": "disk", "NAME": "sda"}
+        device_info(lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert not disk.is_mapper
+
+    @pytest.mark.usefixtures("lsblk_ceph_disk_member",
+                             "disable_kernel_queries")
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_is_ceph_disk_lsblk(self, fake_call, monkeypatch, patch_bluestore_label):
+        disk = device.Device("/dev/sda")
+        assert disk.is_ceph_disk_member
+
+    @pytest.mark.usefixtures("blkid_ceph_disk_member",
+                             "lsblk_ceph_disk_member",
+                             "disable_kernel_queries")
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_is_ceph_disk_blkid(self, fake_call, monkeypatch, patch_bluestore_label):
+        disk = device.Device("/dev/sda")
+        assert disk.is_ceph_disk_member
+
+    @pytest.mark.usefixtures("lsblk_ceph_disk_member",
+                             "disable_kernel_queries")
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_is_ceph_disk_member_not_available_lsblk(self, fake_call, monkeypatch, patch_bluestore_label):
+        disk = device.Device("/dev/sda")
+        assert disk.is_ceph_disk_member
+        assert not disk.available
+        assert "Used by ceph-disk" in disk.rejected_reasons
+
+    @pytest.mark.usefixtures("blkid_ceph_disk_member",
+                             "lsblk_ceph_disk_member",
+                             "disable_kernel_queries")
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_is_ceph_disk_member_not_available_blkid(self, fake_call, monkeypatch, patch_bluestore_label):
+        disk = device.Device("/dev/sda")
+        assert disk.is_ceph_disk_member
+        assert not disk.available
+        assert "Used by ceph-disk" in disk.rejected_reasons
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_reject_removable_device(self, fake_call, device_info):
+        data = {"/dev/sdb": {"removable": 1}}
+        lsblk = {"TYPE": "disk", "NAME": "sdb"}
+        device_info(devices=data,lsblk=lsblk)
+        disk = device.Device("/dev/sdb")
+        assert not disk.available
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_reject_device_with_gpt_headers(self, fake_call, device_info):
+        data = {"/dev/sdb": {"removable": 0, "size": 5368709120}}
+        lsblk = {"TYPE": "disk", "NAME": "sdb"}
+        blkid= {"PTTYPE": "gpt"}
+        device_info(
+            devices=data,
+            blkid=blkid,
+            lsblk=lsblk,
+        )
+        disk = device.Device("/dev/sdb")
+        assert not disk.available
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_accept_non_removable_device(self, fake_call, device_info):
+        data = {"/dev/sdb": {"removable": 0, "size": 5368709120}}
+        lsblk = {"TYPE": "disk", "NAME": "sdb"}
+        device_info(devices=data,lsblk=lsblk)
+        disk = device.Device("/dev/sdb")
+        assert disk.available
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_reject_not_acceptable_device(self, fake_call, device_info):
+        data = {"/dev/dm-0": {"foo": "bar"}}
+        lsblk = {"TYPE": "mpath", "NAME": "dm-0"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/dm-0")
+        assert not disk.available
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    @patch('ceph_volume.util.device.os.path.realpath')
+    @patch('ceph_volume.util.device.os.path.islink')
+    def test_accept_symlink_to_device(self,
+                                      m_os_path_islink,
+                                      m_os_path_realpath,
+                                      device_info,
+                                      fake_call):
+        m_os_path_islink.return_value = True
+        m_os_path_realpath.return_value = '/dev/sdb'
+        data = {"/dev/sdb": {"ro": 0, "size": 5368709120}}
+        lsblk = {"TYPE": "disk"}
+        device_info(devices=data,lsblk=lsblk)
+        disk = device.Device("/dev/test_symlink")
+        print(disk)
+        print(disk.sys_api)
+        assert disk.available
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    @patch('ceph_volume.util.device.os.readlink')
+    @patch('ceph_volume.util.device.os.path.islink')
+    def test_reject_symlink_to_device_mapper(self,
+                                             m_os_path_islink,
+                                             m_os_readlink,
+                                             device_info,
+                                             fake_call):
+        m_os_path_islink.return_value = True
+        m_os_readlink.return_value = '/dev/dm-0'
+        data = {"/dev/mapper/mpatha": {"ro": 0, "size": 5368709120}}
+        lsblk = {"TYPE": "disk"}
+        device_info(devices=data,lsblk=lsblk)
+        disk = device.Device("/dev/mapper/mpatha")
+        assert disk.available
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_reject_readonly_device(self, fake_call, device_info):
+        data = {"/dev/cdrom": {"ro": 1}}
+        lsblk = {"TYPE": "disk", "NAME": "cdrom"}
+        device_info(devices=data,lsblk=lsblk)
+        disk = device.Device("/dev/cdrom")
+        assert not disk.available
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_reject_smaller_than_5gb(self, fake_call, device_info):
+        data = {"/dev/sda": {"size": 5368709119}}
+        lsblk = {"TYPE": "disk", "NAME": "sda"}
+        device_info(devices=data,lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert not disk.available, 'too small device is available'
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_accept_non_readonly_device(self, fake_call, device_info):
+        data = {"/dev/sda": {"ro": 0, "size": 5368709120}}
+        lsblk = {"TYPE": "disk", "NAME": "sda"}
+        device_info(devices=data,lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.available
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_reject_bluestore_device(self, fake_call, monkeypatch, patch_bluestore_label, device_info):
+        patch_bluestore_label.return_value = True
+        lsblk = {"TYPE": "disk", "NAME": "sda"}
+        device_info(lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert not disk.available
+        assert "Has BlueStore device label" in disk.rejected_reasons
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_reject_device_with_oserror(self, fake_call, monkeypatch, patch_bluestore_label, device_info):
+        patch_bluestore_label.side_effect = OSError('test failure')
+        lsblk = {"TYPE": "disk", "NAME": "sda"}
+        device_info(lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert not disk.available
+        assert "Failed to determine if device is BlueStore" in disk.rejected_reasons
+
+    @pytest.mark.usefixtures("lsblk_ceph_disk_member",
+                             "device_info_not_ceph_disk_member",
+                             "disable_kernel_queries")
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_is_not_ceph_disk_member_lsblk(self, fake_call, patch_bluestore_label):
+        disk = device.Device("/dev/sda")
+        assert disk.is_ceph_disk_member is False
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_existing_vg_available(self, fake_call, monkeypatch, device_info):
+        vg = api.VolumeGroup(pv_name='/dev/nvme0n1', vg_name='foo/bar', vg_free_count=1536,
+                             vg_extent_size=4194304)
+        monkeypatch.setattr(api, 'get_all_devices_vgs', lambda : [vg])
+        lsblk = {"TYPE": "disk", "NAME": "nvme0n1"}
+        data = {"/dev/nvme0n1": {"size": "6442450944"}}
+        lv = {"tags": {"ceph.osd_id": "1"}}
+        device_info(devices=data, lsblk=lsblk, lv=lv)
+        disk = device.Device("/dev/nvme0n1")
+        assert disk.available_lvm
+        assert not disk.available
+        assert not disk.available_raw
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_existing_vg_too_small(self, fake_call, monkeypatch, device_info):
+        vg = api.VolumeGroup(pv_name='/dev/nvme0n1', vg_name='foo/bar', vg_free_count=4,
+                             vg_extent_size=1073741824)
+        monkeypatch.setattr(api, 'get_all_devices_vgs', lambda : [vg])
+        lsblk = {"TYPE": "disk", "NAME": "nvme0n1"}
+        data = {"/dev/nvme0n1": {"size": "6442450944"}}
+        lv = {"tags": {"ceph.osd_id": "1"}}
+        device_info(devices=data, lsblk=lsblk, lv=lv)
+        disk = device.Device("/dev/nvme0n1")
+        assert not disk.available_lvm
+        assert not disk.available
+        assert not disk.available_raw
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_multiple_existing_vgs(self, fake_call, monkeypatch, device_info):
+        vg1 = api.VolumeGroup(pv_name='/dev/nvme0n1', vg_name='foo/bar', vg_free_count=1000,
+                             vg_extent_size=4194304)
+        vg2 = api.VolumeGroup(pv_name='/dev/nvme0n1', vg_name='foo/bar', vg_free_count=536,
+                             vg_extent_size=4194304)
+        monkeypatch.setattr(api, 'get_all_devices_vgs', lambda : [vg1, vg2])
+        lsblk = {"TYPE": "disk", "NAME": "nvme0n1"}
+        data = {"/dev/nvme0n1": {"size": "6442450944"}}
+        lv = {"tags": {"ceph.osd_id": "1"}}
+        device_info(devices=data, lsblk=lsblk, lv=lv)
+        disk = device.Device("/dev/nvme0n1")
+        assert disk.available_lvm
+        assert not disk.available
+        assert not disk.available_raw
+
+    @pytest.mark.parametrize("ceph_type", ["data", "block"])
+    def test_used_by_ceph(self, fake_call, device_info,
+                          monkeypatch, ceph_type):
+        data = {"/dev/sda": {"foo": "bar"}}
+        lsblk = {"TYPE": "part", "NAME": "sda", "PKNAME": "sda"}
+        FooPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000",
+                                 lv_uuid="0000", pv_tags={}, vg_name="vg")
+        pvolumes = []
+        pvolumes.append(FooPVolume)
+        lv_data = {"lv_name": "lv", "lv_path": "vg/lv", "vg_name": "vg",
+                   "lv_uuid": "0000", "lv_tags":
+                   "ceph.osd_id=0,ceph.type="+ceph_type}
+        volumes = []
+        lv = api.Volume(**lv_data)
+        volumes.append(lv)
+        monkeypatch.setattr(api, 'get_pvs', lambda **kwargs: pvolumes)
+        monkeypatch.setattr(api, 'get_lvs', lambda **kwargs:
+                            deepcopy(volumes))
+
+        device_info(devices=data, lsblk=lsblk, lv=lv_data)
+        vg = api.VolumeGroup(vg_name='foo/bar', vg_free_count=6,
+                             vg_extent_size=1073741824)
+        monkeypatch.setattr(api, 'get_device_vgs', lambda x: [vg])
+        disk = device.Device("/dev/sda")
+        assert disk.used_by_ceph
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_not_used_by_ceph(self, fake_call, device_info, monkeypatch):
+        FooPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", lv_uuid="0000", pv_tags={}, vg_name="vg")
+        pvolumes = []
+        pvolumes.append(FooPVolume)
+        data = {"/dev/sda": {"foo": "bar"}}
+        lsblk = {"TYPE": "part", "NAME": "sda", "PKNAME": "sda"}
+        lv_data = {"lv_path": "vg/lv", "vg_name": "vg", "lv_uuid": "0000", "tags": {"ceph.osd_id": 0, "ceph.type": "journal"}}
+        monkeypatch.setattr(api, 'get_pvs', lambda **kwargs: pvolumes)
+
+        device_info(devices=data, lsblk=lsblk, lv=lv_data)
+        disk = device.Device("/dev/sda")
+        assert not disk.used_by_ceph
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_get_device_id(self, fake_call, device_info):
+        udev = {k:k for k in ['ID_VENDOR', 'ID_MODEL', 'ID_SCSI_SERIAL']}
+        lsblk = {"TYPE": "disk", "NAME": "sda"}
+        device_info(udevadm=udev,lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk._get_device_id() == 'ID_VENDOR_ID_MODEL_ID_SCSI_SERIAL'
+
+    def test_has_bluestore_label(self):
+        # patch device.Device __init__ function to do nothing since we want to only test the
+        # low-level behavior of has_bluestore_label
+        with patch.object(device.Device, "__init__", lambda self, path, with_lsm=False: None):
+            disk = device.Device("/dev/sda")
+            disk.path = "/dev/sda"
+            with patch('builtins.open', mock_open(read_data=b'bluestore block device\n')):
+                assert disk.has_bluestore_label
+            with patch('builtins.open', mock_open(read_data=b'not a bluestore block device\n')):
+                assert not disk.has_bluestore_label
+
+
+class TestDeviceEncryption(object):
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_partition_is_not_encrypted_lsblk(self, fake_call, device_info):
+        lsblk = {'TYPE': 'part', 'FSTYPE': 'xfs', 'NAME': 'sda', 'PKNAME': 'sda'}
+        device_info(lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.is_encrypted is False
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_partition_is_encrypted_lsblk(self, fake_call, device_info):
+        lsblk = {'TYPE': 'part', 'FSTYPE': 'crypto_LUKS', 'NAME': 'sda', 'PKNAME': 'sda'}
+        device_info(lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.is_encrypted is True
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_partition_is_not_encrypted_blkid(self, fake_call, device_info):
+        lsblk = {'TYPE': 'part', 'NAME': 'sda', 'PKNAME': 'sda'}
+        blkid = {'TYPE': 'ceph data'}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/sda")
+        assert disk.is_encrypted is False
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_partition_is_encrypted_blkid(self, fake_call, device_info):
+        lsblk = {'TYPE': 'part', 'NAME': 'sda' ,'PKNAME': 'sda'}
+        blkid = {'TYPE': 'crypto_LUKS'}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/sda")
+        assert disk.is_encrypted is True
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_mapper_is_encrypted_luks1(self, fake_call, device_info, monkeypatch):
+        status = {'type': 'LUKS1'}
+        monkeypatch.setattr(device, 'encryption_status', lambda x: status)
+        lsblk = {'FSTYPE': 'xfs', 'NAME': 'uuid','TYPE': 'lvm'}
+        blkid = {'TYPE': 'mapper'}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/mapper/uuid")
+        assert disk.is_encrypted is True
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_mapper_is_encrypted_luks2(self, fake_call, device_info, monkeypatch):
+        status = {'type': 'LUKS2'}
+        monkeypatch.setattr(device, 'encryption_status', lambda x: status)
+        lsblk = {'FSTYPE': 'xfs', 'NAME': 'uuid', 'TYPE': 'lvm'}
+        blkid = {'TYPE': 'mapper'}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/mapper/uuid")
+        assert disk.is_encrypted is True
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_mapper_is_encrypted_plain(self, fake_call, device_info, monkeypatch):
+        status = {'type': 'PLAIN'}
+        monkeypatch.setattr(device, 'encryption_status', lambda x: status)
+        lsblk = {'FSTYPE': 'xfs', 'NAME': 'uuid', 'TYPE': 'lvm'}
+        blkid = {'TYPE': 'mapper'}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/mapper/uuid")
+        assert disk.is_encrypted is True
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_mapper_is_not_encrypted_plain(self, fake_call, device_info, monkeypatch):
+        monkeypatch.setattr(device, 'encryption_status', lambda x: {})
+        lsblk = {'FSTYPE': 'xfs', 'NAME': 'uuid', 'TYPE': 'lvm'}
+        blkid = {'TYPE': 'mapper'}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/mapper/uuid")
+        assert disk.is_encrypted is False
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_lv_is_encrypted_blkid(self, fake_call, device_info):
+        lsblk = {'TYPE': 'lvm', 'NAME': 'sda'}
+        blkid = {'TYPE': 'crypto_LUKS'}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/sda")
+        disk.lv_api = {}
+        assert disk.is_encrypted is True
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_lv_is_not_encrypted_blkid(self, fake_call, factory, device_info):
+        lsblk = {'TYPE': 'lvm', 'NAME': 'sda'}
+        blkid = {'TYPE': 'xfs'}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/sda")
+        disk.lv_api = factory(encrypted=None)
+        assert disk.is_encrypted is False
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_lv_is_encrypted_lsblk(self, fake_call, device_info):
+        lsblk = {'FSTYPE': 'crypto_LUKS', 'NAME': 'sda', 'TYPE': 'lvm'}
+        blkid = {'TYPE': 'mapper'}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/sda")
+        disk.lv_api = {}
+        assert disk.is_encrypted is True
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_lv_is_not_encrypted_lsblk(self, fake_call, factory, device_info):
+        lsblk = {'FSTYPE': 'xfs', 'NAME': 'sda', 'TYPE': 'lvm'}
+        blkid = {'TYPE': 'mapper'}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/sda")
+        disk.lv_api = factory(encrypted=None)
+        assert disk.is_encrypted is False
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_lv_is_encrypted_lvm_api(self, fake_call, factory, device_info):
+        lsblk = {'FSTYPE': 'xfs', 'NAME': 'sda', 'TYPE': 'lvm'}
+        blkid = {'TYPE': 'mapper'}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/sda")
+        disk.lv_api = factory(encrypted=True)
+        assert disk.is_encrypted is True
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_lv_is_not_encrypted_lvm_api(self, fake_call, factory, device_info):
+        lsblk = {'FSTYPE': 'xfs', 'NAME': 'sda', 'TYPE': 'lvm'}
+        blkid = {'TYPE': 'mapper'}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/sda")
+        disk.lv_api = factory(encrypted=False)
+        assert disk.is_encrypted is False
+
+
+class TestDeviceOrdering(object):
+
+    def setup(self):
+        self.data = {
+                "/dev/sda": {"removable": 0},
+                "/dev/sdb": {"removable": 1}, # invalid
+                "/dev/sdc": {"removable": 0},
+                "/dev/sdd": {"removable": 1}, # invalid
+        }
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_valid_before_invalid(self, fake_call, device_info):
+        lsblk_sda = {"NAME": "sda", "TYPE": "disk"}
+        lsblk_sdb = {"NAME": "sdb", "TYPE": "disk"}
+        device_info(devices=self.data,lsblk=lsblk_sda)
+        sda = device.Device("/dev/sda")
+        device_info(devices=self.data,lsblk=lsblk_sdb)
+        sdb = device.Device("/dev/sdb")
+
+        assert sda < sdb
+        assert sdb > sda
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_valid_alphabetical_ordering(self, fake_call, device_info):
+        lsblk_sda = {"NAME": "sda", "TYPE": "disk"}
+        lsblk_sdc = {"NAME": "sdc", "TYPE": "disk"}
+        device_info(devices=self.data,lsblk=lsblk_sda)
+        sda = device.Device("/dev/sda")
+        device_info(devices=self.data,lsblk=lsblk_sdc)
+        sdc = device.Device("/dev/sdc")
+
+        assert sda < sdc
+        assert sdc > sda
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_invalid_alphabetical_ordering(self, fake_call, device_info):
+        lsblk_sdb = {"NAME": "sdb", "TYPE": "disk"}
+        lsblk_sdd = {"NAME": "sdd", "TYPE": "disk"}
+        device_info(devices=self.data,lsblk=lsblk_sdb)
+        sdb = device.Device("/dev/sdb")
+        device_info(devices=self.data,lsblk=lsblk_sdd)
+        sdd = device.Device("/dev/sdd")
+
+        assert sdb < sdd
+        assert sdd > sdb
+
+
+class TestCephDiskDevice(object):
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_partlabel_lsblk(self, fake_call, device_info):
+        lsblk = {"TYPE": "disk", "NAME": "sda", "PARTLABEL": ""}
+        device_info(lsblk=lsblk)
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.partlabel == ''
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_partlabel_blkid(self, fake_call, device_info):
+        lsblk = {"TYPE": "disk", "NAME": "sda", "PARTLABEL": "ceph data"}
+        blkid = {"TYPE": "disk", "PARTLABEL": "ceph data"}
+        device_info(blkid=blkid, lsblk=lsblk)
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.partlabel == 'ceph data'
+
+    @pytest.mark.usefixtures("lsblk_ceph_disk_member",
+                             "blkid_ceph_disk_member",
+                             "disable_kernel_queries")
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_is_member_blkid(self, fake_call, monkeypatch):
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.is_member is True
+
+    @pytest.mark.usefixtures("lsblk_ceph_disk_member",
+                             "disable_kernel_queries")
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_is_member_lsblk(self, fake_call, patch_bluestore_label, device_info):
+        lsblk = {"TYPE": "disk", "NAME": "sda", "PARTLABEL": "ceph"}
+        device_info(lsblk=lsblk)
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.is_member is True
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_unknown_type(self, fake_call, device_info):
+        lsblk = {"TYPE": "disk", "NAME": "sda", "PARTLABEL": "gluster"}
+        device_info(lsblk=lsblk)
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.type == 'unknown'
+
+    ceph_types = ['data', 'wal', 'db', 'lockbox', 'journal', 'block']
+
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    @pytest.mark.usefixtures("lsblk_ceph_disk_member",
+                             "blkid_ceph_disk_member",
+                             "disable_kernel_queries")
+    def test_type_blkid(self, monkeypatch, fake_call, device_info, ceph_partlabel):
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.type in self.ceph_types
+
+    @pytest.mark.usefixtures("blkid_ceph_disk_member",
+                             "lsblk_ceph_disk_member",
+                             "disable_kernel_queries")
+    @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)
+    def test_type_lsblk(self, fake_call, device_info, ceph_partlabel):
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.type in self.ceph_types
diff --git a/src/ceph-volume/ceph_volume/tests/util/test_disk.py b/src/ceph-volume/ceph_volume/tests/util/test_disk.py
new file mode 100644
index 000000000..fcd644a86
--- /dev/null
+++ b/src/ceph-volume/ceph_volume/tests/util/test_disk.py
@@ -0,0 +1,524 @@
+import os
+import pytest
+from ceph_volume.util import disk
+from mock.mock import patch
+
+
+class TestLsblkParser(object):
+
+    def test_parses_whitespace_values(self):
+        output = 'NAME="sdaa5" PARTLABEL="ceph data" RM="0" SIZE="10M" RO="0" TYPE="part"'
+        result = disk._lsblk_parser(output)
+        assert result['PARTLABEL'] == 'ceph data'
+
+    def test_ignores_bogus_pairs(self):
+        output = 'NAME="sdaa5" PARTLABEL RM="0" SIZE="10M" RO="0" TYPE="part" MOUNTPOINT=""'
+        result = disk._lsblk_parser(output)
+        assert result['SIZE'] == '10M'
+
+
+class TestBlkidParser(object):
+
+    def test_parses_whitespace_values(self):
+        output = '''/dev/sdb1: UUID="62416664-cbaf-40bd-9689-10bd337379c3" TYPE="xfs" PART_ENTRY_SCHEME="gpt" PART_ENTRY_NAME="ceph data" PART_ENTRY_UUID="b89c03bc-bf58-4338-a8f8-a2f484852b4f"'''  # noqa
+        result = disk._blkid_parser(output)
+        assert result['PARTLABEL'] == 'ceph data'
+
+    def test_ignores_unmapped(self):
+        output = '''/dev/sdb1: UUID="62416664-cbaf-40bd-9689-10bd337379c3" TYPE="xfs" PART_ENTRY_SCHEME="gpt" PART_ENTRY_NAME="ceph data" PART_ENTRY_UUID="b89c03bc-bf58-4338-a8f8-a2f484852b4f"'''  # noqa
+        result = disk._blkid_parser(output)
+        assert len(result.keys()) == 4
+
+    def test_translates_to_partuuid(self):
+        output = '''/dev/sdb1: UUID="62416664-cbaf-40bd-9689-10bd337379c3" TYPE="xfs" PART_ENTRY_SCHEME="gpt" PART_ENTRY_NAME="ceph data" PART_ENTRY_UUID="b89c03bc-bf58-4338-a8f8-a2f484852b4f"'''  # noqa
+        result = disk._blkid_parser(output)
+        assert result['PARTUUID'] == 'b89c03bc-bf58-4338-a8f8-a2f484852b4f'
+
+
+class TestBlkid(object):
+
+    def test_parses_translated(self, stub_call):
+        output = '''/dev/sdb1: UUID="62416664-cbaf-40bd-9689-10bd337379c3" TYPE="xfs" PART_ENTRY_SCHEME="gpt" PART_ENTRY_NAME="ceph data" PART_ENTRY_UUID="b89c03bc-bf58-4338-a8f8-a2f484852b4f"'''  # noqa
+        stub_call((output.split(), [], 0))
+        result = disk.blkid('/dev/sdb1')
+        assert result['PARTUUID'] == 'b89c03bc-bf58-4338-a8f8-a2f484852b4f'
+        assert result['PARTLABEL'] == 'ceph data'
+        assert result['UUID'] == '62416664-cbaf-40bd-9689-10bd337379c3'
+        assert result['TYPE'] == 'xfs'
+
+class TestUdevadmProperty(object):
+
+    def test_good_output(self, stub_call):
+        output = """ID_MODEL=SK_hynix_SC311_SATA_512GB
+ID_PART_TABLE_TYPE=gpt
+ID_SERIAL_SHORT=MS83N71801150416A""".split()
+        stub_call((output, [], 0))
+        result = disk.udevadm_property('dev/sda')
+        assert result['ID_MODEL'] == 'SK_hynix_SC311_SATA_512GB'
+        assert result['ID_PART_TABLE_TYPE'] == 'gpt'
+        assert result['ID_SERIAL_SHORT'] == 'MS83N71801150416A'
+
+    def test_property_filter(self, stub_call):
+        output = """ID_MODEL=SK_hynix_SC311_SATA_512GB
+ID_PART_TABLE_TYPE=gpt
+ID_SERIAL_SHORT=MS83N71801150416A""".split()
+        stub_call((output, [], 0))
+        result = disk.udevadm_property('dev/sda', ['ID_MODEL',
+                                                   'ID_SERIAL_SHORT'])
+        assert result['ID_MODEL'] == 'SK_hynix_SC311_SATA_512GB'
+        assert 'ID_PART_TABLE_TYPE' not in result
+
+    def test_fail_on_broken_output(self, stub_call):
+        output = ["ID_MODEL:SK_hynix_SC311_SATA_512GB"]
+        stub_call((output, [], 0))
+        with pytest.raises(ValueError):
+            disk.udevadm_property('dev/sda')
+
+
+class TestDeviceFamily(object):
+
+    def test_groups_multiple_devices(self, stub_call):
+        out = [
+            'NAME="sdaa5" PARLABEL="ceph lockbox"',
+            'NAME="sdaa" RO="0"',
+            'NAME="sdaa1" PARLABEL="ceph data"',
+            'NAME="sdaa2" PARLABEL="ceph journal"',
+        ]
+        stub_call((out, '', 0))
+        result = disk.device_family('sdaa5')
+        assert len(result) == 4
+
+    def test_parses_output_correctly(self, stub_call):
+        names = ['sdaa', 'sdaa5', 'sdaa1', 'sdaa2']
+        out = [
+            'NAME="sdaa5" PARLABEL="ceph lockbox"',
+            'NAME="sdaa" RO="0"',
+            'NAME="sdaa1" PARLABEL="ceph data"',
+            'NAME="sdaa2" PARLABEL="ceph journal"',
+        ]
+        stub_call((out, '', 0))
+        result = disk.device_family('sdaa5')
+        for parsed in result:
+            assert parsed['NAME'] in names
+
+
+class TestHumanReadableSize(object):
+
+    def test_bytes(self):
+        result = disk.human_readable_size(800)
+        assert result == '800.00 B'
+
+    def test_kilobytes(self):
+        result = disk.human_readable_size(800*1024)
+        assert result == '800.00 KB'
+
+    def test_megabytes(self):
+        result = disk.human_readable_size(800*1024*1024)
+        assert result == '800.00 MB'
+
+    def test_gigabytes(self):
+        result = disk.human_readable_size(8.19*1024*1024*1024)
+        assert result == '8.19 GB'
+
+    def test_terabytes(self):
+        result = disk.human_readable_size(81.2*1024*1024*1024*1024)
+        assert result == '81.20 TB'
+
+    def test_petabytes(self):
+        result = disk.human_readable_size(9.23*1024*1024*1024*1024*1024)
+        assert result == '9.23 PB'
+
+class TestSizeFromHumanReadable(object):
+
+    def test_bytes(self):
+        result = disk.size_from_human_readable('2')
+        assert result == disk.Size(b=2)
+
+    def test_kilobytes(self):
+        result = disk.size_from_human_readable('2 K')
+        assert result == disk.Size(kb=2)
+
+    def test_megabytes(self):
+        result = disk.size_from_human_readable('2 M')
+        assert result == disk.Size(mb=2)
+
+    def test_gigabytes(self):
+        result = disk.size_from_human_readable('2 G')
+        assert result == disk.Size(gb=2)
+
+    def test_terabytes(self):
+        result = disk.size_from_human_readable('2 T')
+        assert result == disk.Size(tb=2)
+
+    def test_petabytes(self):
+        result = disk.size_from_human_readable('2 P')
+        assert result == disk.Size(pb=2)
+
+    def test_case(self):
+        result = disk.size_from_human_readable('2 t')
+        assert result == disk.Size(tb=2)
+
+    def test_space(self):
+        result = disk.size_from_human_readable('2T')
+        assert result == disk.Size(tb=2)
+
+    def test_float(self):
+        result = disk.size_from_human_readable('2.0')
+        assert result == disk.Size(b=2)
+        result = disk.size_from_human_readable('2.0T')
+        assert result == disk.Size(tb=2)
+        result = disk.size_from_human_readable('1.8T')
+        assert result == disk.Size(tb=1.8)
+
+
+class TestSizeParse(object):
+
+    def test_bytes(self):
+        result = disk.Size.parse('2')
+        assert result == disk.Size(b=2)
+
+    def test_kilobytes(self):
+        result = disk.Size.parse('2K')
+        assert result == disk.Size(kb=2)
+
+    def test_megabytes(self):
+        result = disk.Size.parse('2M')
+        assert result == disk.Size(mb=2)
+
+    def test_gigabytes(self):
+        result = disk.Size.parse('2G')
+        assert result == disk.Size(gb=2)
+
+    def test_terabytes(self):
+        result = disk.Size.parse('2T')
+        assert result == disk.Size(tb=2)
+
+    def test_petabytes(self):
+        result = disk.Size.parse('2P')
+        assert result == disk.Size(pb=2)
+
+    def test_tb(self):
+        result = disk.Size.parse('2Tb')
+        assert result == disk.Size(tb=2)
+
+    def test_case(self):
+        result = disk.Size.parse('2t')
+        assert result == disk.Size(tb=2)
+
+    def test_space(self):
+        result = disk.Size.parse('2T')
+        assert result == disk.Size(tb=2)
+
+    def test_float(self):
+        result = disk.Size.parse('2.0')
+        assert result == disk.Size(b=2)
+        result = disk.Size.parse('2.0T')
+        assert result == disk.Size(tb=2)
+        result = disk.Size.parse('1.8T')
+        assert result == disk.Size(tb=1.8)
+
+
+class TestGetDevices(object):
+
+    def test_no_devices_are_found(self, tmpdir, patched_get_block_devs_sysfs):
+        patched_get_block_devs_sysfs.return_value = []
+        result = disk.get_devices(_sys_block_path=str(tmpdir))
+        assert result == {}
+
+    @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False)
+    def test_sda_block_is_found(self, patched_get_block_devs_sysfs, fake_filesystem):
+        sda_path = '/dev/sda'
+        patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']]
+        result = disk.get_devices()
+        assert len(result.keys()) == 1
+        assert result[sda_path]['human_readable_size'] == '0.00 B'
+        assert result[sda_path]['model'] == ''
+        assert result[sda_path]['partitions'] == {}
+
+    @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False)
+    def test_sda_size(self, patched_get_block_devs_sysfs, fake_filesystem):
+        sda_path = '/dev/sda'
+        patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']]
+        fake_filesystem.create_file('/sys/block/sda/size', contents = '1024')
+        result = disk.get_devices()
+        assert list(result.keys()) == [sda_path]
+        assert result[sda_path]['human_readable_size'] == '512.00 KB'
+
+    @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False)
+    def test_sda_sectorsize_fallsback(self, patched_get_block_devs_sysfs, fake_filesystem):
+        # if no sectorsize, it will use queue/hw_sector_size
+        sda_path = '/dev/sda'
+        patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']]
+        fake_filesystem.create_file('/sys/block/sda/queue/hw_sector_size', contents = '1024')
+        result = disk.get_devices()
+        assert list(result.keys()) == [sda_path]
+        assert result[sda_path]['sectorsize'] == '1024'
+
+    @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False)
+    def test_sda_sectorsize_from_logical_block(self, patched_get_block_devs_sysfs, fake_filesystem):
+        sda_path = '/dev/sda'
+        patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']]
+        fake_filesystem.create_file('/sys/block/sda/queue/logical_block_size', contents = '99')
+        result = disk.get_devices()
+        assert result[sda_path]['sectorsize'] == '99'
+
+    @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False)
+    def test_sda_sectorsize_does_not_fallback(self, patched_get_block_devs_sysfs, fake_filesystem):
+        sda_path = '/dev/sda'
+        patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']]
+        fake_filesystem.create_file('/sys/block/sda/queue/logical_block_size', contents = '99')
+        fake_filesystem.create_file('/sys/block/sda/queue/hw_sector_size', contents = '1024')
+        result = disk.get_devices()
+        assert result[sda_path]['sectorsize'] == '99'
+
+    @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False)
+    def test_is_rotational(self, patched_get_block_devs_sysfs, fake_filesystem):
+        sda_path = '/dev/sda'
+        patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']]
+        fake_filesystem.create_file('/sys/block/sda/queue/rotational', contents = '1')
+        result = disk.get_devices()
+        assert result[sda_path]['rotational'] == '1'
+
+    @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False)
+    def test_is_ceph_rbd(self, patched_get_block_devs_sysfs, fake_filesystem):
+        rbd_path = '/dev/rbd0'
+        patched_get_block_devs_sysfs.return_value = [[rbd_path, rbd_path, 'disk']]
+        result = disk.get_devices()
+        assert rbd_path not in result
+
+
+class TestSizeCalculations(object):
+
+    @pytest.mark.parametrize('aliases', [
+        ('b', 'bytes'),
+        ('kb', 'kilobytes'),
+        ('mb', 'megabytes'),
+        ('gb', 'gigabytes'),
+        ('tb', 'terabytes'),
+    ])
+    def test_aliases(self, aliases):
+        short_alias, long_alias = aliases
+        s = disk.Size(b=1)
+        short_alias = getattr(s, short_alias)
+        long_alias = getattr(s, long_alias)
+        assert short_alias == long_alias
+
+    @pytest.mark.parametrize('values', [
+        ('b', 857619069665.28),
+        ('kb', 837518622.72),
+        ('mb', 817889.28),
+        ('gb', 798.72),
+        ('tb', 0.78),
+    ])
+    def test_terabytes(self, values):
+        # regardless of the input value, all the other values correlate to each
+        # other the same, every time
+        unit, value = values
+        s = disk.Size(**{unit: value})
+        assert s.b == 857619069665.28
+        assert s.kb == 837518622.72
+        assert s.mb == 817889.28
+        assert s.gb == 798.72
+        assert s.tb == 0.78
+
+
+class TestSizeOperators(object):
+
+    @pytest.mark.parametrize('larger', [1025, 1024.1, 1024.001])
+    def test_gigabytes_is_smaller(self, larger):
+        assert disk.Size(gb=1) < disk.Size(mb=larger)
+
+    @pytest.mark.parametrize('smaller', [1023, 1023.9, 1023.001])
+    def test_gigabytes_is_larger(self, smaller):
+        assert disk.Size(gb=1) > disk.Size(mb=smaller)
+
+    @pytest.mark.parametrize('larger', [1025, 1024.1, 1024.001, 1024])
+    def test_gigabytes_is_smaller_or_equal(self, larger):
+        assert disk.Size(gb=1) <= disk.Size(mb=larger)
+
+    @pytest.mark.parametrize('smaller', [1023, 1023.9, 1023.001, 1024])
+    def test_gigabytes_is_larger_or_equal(self, smaller):
+        assert disk.Size(gb=1) >= disk.Size(mb=smaller)
+
+    @pytest.mark.parametrize('values', [
+        ('b', 857619069665.28),
+        ('kb', 837518622.72),
+        ('mb', 817889.28),
+        ('gb', 798.72),
+        ('tb', 0.78),
+    ])
+    def test_equality(self, values):
+        unit, value = values
+        s = disk.Size(**{unit: value})
+        # both tb and b, since b is always calculated regardless, and is useful
+        # when testing tb
+        assert disk.Size(tb=0.78) == s
+        assert disk.Size(b=857619069665.28) == s
+
+    @pytest.mark.parametrize('values', [
+        ('b', 857619069665.28),
+        ('kb', 837518622.72),
+        ('mb', 817889.28),
+        ('gb', 798.72),
+        ('tb', 0.78),
+    ])
+    def test_inequality(self, values):
+        unit, value = values
+        s = disk.Size(**{unit: value})
+        # both tb and b, since b is always calculated regardless, and is useful
+        # when testing tb
+        assert disk.Size(tb=1) != s
+        assert disk.Size(b=100) != s
+
+
+class TestSizeOperations(object):
+
+    def test_assignment_addition_with_size_objects(self):
+        result = disk.Size(mb=256) + disk.Size(gb=1)
+        assert result.gb == 1.25
+        assert result.gb.as_int() == 1
+        assert result.gb.as_float() == 1.25
+
+    def test_self_addition_with_size_objects(self):
+        base = disk.Size(mb=256)
+        base += disk.Size(gb=1)
+        assert base.gb == 1.25
+
+    def test_self_addition_does_not_alter_state(self):
+        base = disk.Size(mb=256)
+        base + disk.Size(gb=1)
+        assert base.mb == 256
+
+    def test_addition_with_non_size_objects(self):
+        with pytest.raises(TypeError):
+            disk.Size(mb=100) + 4
+
+    def test_assignment_subtraction_with_size_objects(self):
+        base = disk.Size(gb=1)
+        base -= disk.Size(mb=256)
+        assert base.mb == 768
+
+    def test_self_subtraction_does_not_alter_state(self):
+        base = disk.Size(gb=1)
+        base - disk.Size(mb=256)
+        assert base.gb == 1
+
+    def test_subtraction_with_size_objects(self):
+        result = disk.Size(gb=1) - disk.Size(mb=256)
+        assert result.mb == 768
+
+    def test_subtraction_with_non_size_objects(self):
+        with pytest.raises(TypeError):
+            disk.Size(mb=100) - 4
+
+    def test_multiplication_with_size_objects(self):
+        with pytest.raises(TypeError):
+            disk.Size(mb=100) * disk.Size(mb=1)
+
+    def test_multiplication_with_non_size_objects(self):
+        base = disk.Size(gb=1)
+        result = base * 2
+        assert result.gb == 2
+        assert result.gb.as_int() == 2
+
+    def test_division_with_size_objects(self):
+        result = disk.Size(gb=1) / disk.Size(mb=1)
+        assert int(result) == 1024
+
+    def test_division_with_non_size_objects(self):
+        base = disk.Size(gb=1)
+        result = base / 2
+        assert result.mb == 512
+        assert result.mb.as_int() == 512
+
+    def test_division_with_non_size_objects_without_state(self):
+        base = disk.Size(gb=1)
+        base / 2
+        assert base.gb == 1
+        assert base.gb.as_int() == 1
+
+
+class TestSizeAttributes(object):
+
+    def test_attribute_does_not_exist(self):
+        with pytest.raises(AttributeError):
+            disk.Size(mb=1).exabytes
+
+
+class TestSizeFormatting(object):
+
+    def test_default_formatting_tb_to_b(self):
+        size = disk.Size(tb=0.0000000001)
+        result = "%s" % size
+        assert result == "109.95 B"
+
+    def test_default_formatting_tb_to_kb(self):
+        size = disk.Size(tb=0.00000001)
+        result = "%s" % size
+        assert result == "10.74 KB"
+
+    def test_default_formatting_tb_to_mb(self):
+        size = disk.Size(tb=0.000001)
+        result = "%s" % size
+        assert result == "1.05 MB"
+
+    def test_default_formatting_tb_to_gb(self):
+        size = disk.Size(tb=0.001)
+        result = "%s" % size
+        assert result == "1.02 GB"
+
+    def test_default_formatting_tb_to_tb(self):
+        size = disk.Size(tb=10)
+        result = "%s" % size
+        assert result == "10.00 TB"
+
+
+class TestSizeSpecificFormatting(object):
+
+    def test_formatting_b(self):
+        size = disk.Size(b=2048)
+        result = "%s" % size.b
+        assert "%s" % size.b == "%s" % size.bytes
+        assert result == "2048.00 B"
+
+    def test_formatting_kb(self):
+        size = disk.Size(kb=5700)
+        result = "%s" % size.kb
+        assert "%s" % size.kb == "%s" % size.kilobytes
+        assert result == "5700.00 KB"
+
+    def test_formatting_mb(self):
+        size = disk.Size(mb=4000)
+        result = "%s" % size.mb
+        assert "%s" % size.mb == "%s" % size.megabytes
+        assert result == "4000.00 MB"
+
+    def test_formatting_gb(self):
+        size = disk.Size(gb=77777)
+        result = "%s" % size.gb
+        assert "%s" % size.gb == "%s" % size.gigabytes
+        assert result == "77777.00 GB"
+
+    def test_formatting_tb(self):
+        size = disk.Size(tb=1027)
+        result = "%s" % size.tb
+        assert "%s" % size.tb == "%s" % size.terabytes
+        assert result == "1027.00 TB"
+
+
+class TestAllowLoopDevsWarning(object):
+    def test_loop_dev_warning(self, fake_call, caplog):
+        assert disk.allow_loop_devices() is False
+        assert not caplog.records
+        os.environ['CEPH_VOLUME_ALLOW_LOOP_DEVICES'] = "y"
+        assert disk.allow_loop_devices() is True
+        log = caplog.records[0]
+        assert log.levelname == "WARNING"
+        assert "will never be supported in production" in log.message
+
+
+class TestHasBlueStoreLabel(object):
+    def test_device_path_is_a_path(self, fake_filesystem):
+        device_path = '/var/lib/ceph/osd/ceph-0'
+        fake_filesystem.create_dir(device_path)
+        assert not disk.has_bluestore_label(device_path)
\ No newline at end of file
diff --git a/src/ceph-volume/ceph_volume/tests/util/test_encryption.py b/src/ceph-volume/ceph_volume/tests/util/test_encryption.py
new file mode 100644
index 000000000..cd2ea8f18
--- /dev/null
+++ b/src/ceph-volume/ceph_volume/tests/util/test_encryption.py
@@ -0,0 +1,138 @@
+from ceph_volume.util import encryption
+from mock.mock import patch
+import base64
+
+class TestGetKeySize(object):
+    def test_get_size_from_conf_default(self, conf_ceph_stub):
+        conf_ceph_stub('''
+        [global]
+        fsid=asdf
+        ''')
+        assert encryption.get_key_size_from_conf() == '512'
+
+    def test_get_size_from_conf_custom(self, conf_ceph_stub):
+        conf_ceph_stub('''
+        [global]
+        fsid=asdf
+        [osd]
+        osd_dmcrypt_key_size=256
+        ''')
+        assert encryption.get_key_size_from_conf() == '256'
+
+    def test_get_size_from_conf_custom_invalid(self, conf_ceph_stub):
+        conf_ceph_stub('''
+        [global]
+        fsid=asdf
+        [osd]
+        osd_dmcrypt_key_size=1024
+        ''')
+        assert encryption.get_key_size_from_conf() == '512'
+
+class TestStatus(object):
+
+    def test_skips_unuseful_lines(self, stub_call):
+        out = ['some line here', '  device: /dev/sdc1']
+        stub_call((out, '', 0))
+        assert encryption.status('/dev/sdc1') == {'device': '/dev/sdc1'}
+
+    def test_removes_extra_quotes(self, stub_call):
+        out = ['some line here', '  device: "/dev/sdc1"']
+        stub_call((out, '', 0))
+        assert encryption.status('/dev/sdc1') == {'device': '/dev/sdc1'}
+
+    def test_ignores_bogus_lines(self, stub_call):
+        out = ['some line here', '  ']
+        stub_call((out, '', 0))
+        assert encryption.status('/dev/sdc1') == {}
+
+
+class TestDmcryptClose(object):
+
+    def test_mapper_exists(self, fake_run, fake_filesystem):
+        file_name = fake_filesystem.create_file('mapper-device')
+        encryption.dmcrypt_close(file_name.path)
+        arguments = fake_run.calls[0]['args'][0]
+        assert arguments[0] == 'cryptsetup'
+        assert arguments[1] == 'remove'
+        assert arguments[2].startswith('/')
+
+    def test_mapper_does_not_exist(self, fake_run):
+        file_name = '/path/does/not/exist'
+        encryption.dmcrypt_close(file_name)
+        assert fake_run.calls == []
+
+
+class TestDmcryptKey(object):
+
+    def test_dmcrypt(self):
+        result = encryption.create_dmcrypt_key()
+        assert len(base64.b64decode(result)) == 128
+
+class TestLuksFormat(object):
+    @patch('ceph_volume.util.encryption.process.call')
+    def test_luks_format_command_with_default_size(self, m_call, conf_ceph_stub):
+        conf_ceph_stub('[global]\nfsid=abcd')
+        expected = [
+            'cryptsetup',
+            '--batch-mode',
+            '--key-size',
+            '512',
+            '--key-file',
+            '-',
+            'luksFormat',
+            '/dev/foo'
+        ]
+        encryption.luks_format('abcd', '/dev/foo')
+        assert m_call.call_args[0][0] == expected
+
+    @patch('ceph_volume.util.encryption.process.call')
+    def test_luks_format_command_with_custom_size(self, m_call, conf_ceph_stub):
+        conf_ceph_stub('[global]\nfsid=abcd\n[osd]\nosd_dmcrypt_key_size=256')
+        expected = [
+            'cryptsetup',
+            '--batch-mode',
+            '--key-size',
+            '256',
+            '--key-file',
+            '-',
+            'luksFormat',
+            '/dev/foo'
+        ]
+        encryption.luks_format('abcd', '/dev/foo')
+        assert m_call.call_args[0][0] == expected
+
+
+class TestLuksOpen(object):
+    @patch('ceph_volume.util.encryption.process.call')
+    def test_luks_open_command_with_default_size(self, m_call, conf_ceph_stub):
+        conf_ceph_stub('[global]\nfsid=abcd')
+        expected = [
+            'cryptsetup',
+            '--key-size',
+            '512',
+            '--key-file',
+            '-',
+            '--allow-discards',
+            'luksOpen',
+            '/dev/foo',
+            '/dev/bar'
+        ]
+        encryption.luks_open('abcd', '/dev/foo', '/dev/bar')
+        assert m_call.call_args[0][0] == expected
+
+    @patch('ceph_volume.util.encryption.process.call')
+    def test_luks_open_command_with_custom_size(self, m_call, conf_ceph_stub):
+        conf_ceph_stub('[global]\nfsid=abcd\n[osd]\nosd_dmcrypt_key_size=256')
+        expected = [
+            'cryptsetup',
+            '--key-size',
+            '256',
+            '--key-file',
+            '-',
+            '--allow-discards',
+            'luksOpen',
+            '/dev/foo',
+            '/dev/bar'
+        ]
+        encryption.luks_open('abcd', '/dev/foo', '/dev/bar')
+        assert m_call.call_args[0][0] == expected
diff --git a/src/ceph-volume/ceph_volume/tests/util/test_prepare.py b/src/ceph-volume/ceph_volume/tests/util/test_prepare.py
new file mode 100644
index 000000000..080823307
--- /dev/null
+++ b/src/ceph-volume/ceph_volume/tests/util/test_prepare.py
@@ -0,0 +1,413 @@
+import pytest
+from textwrap import dedent
+import json
+from ceph_volume.util import prepare
+from ceph_volume.util.prepare import system
+from ceph_volume import conf
+from ceph_volume.tests.conftest import Factory
+
+
+class TestOSDIDAvailable(object):
+
+    def test_false_if_id_is_none(self):
+        assert not prepare.osd_id_available(None)
+
+    def test_returncode_is_not_zero(self, monkeypatch):
+        monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: ('', '', 1))
+        with pytest.raises(RuntimeError):
+            prepare.osd_id_available(1)
+
+    def test_id_does_exist_but_not_available(self, monkeypatch):
+        stdout = dict(nodes=[
+            dict(id=0, status="up"),
+        ])
+        stdout = ['', json.dumps(stdout)]
+        monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: (stdout, '', 0))
+        result = prepare.osd_id_available(0)
+        assert not result
+
+    def test_id_does_not_exist(self, monkeypatch):
+        stdout = dict(nodes=[
+            dict(id=0),
+        ])
+        stdout = ['', json.dumps(stdout)]
+        monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: (stdout, '', 0))
+        result = prepare.osd_id_available(1)
+        assert result
+
+    def test_returns_true_when_id_is_destroyed(self, monkeypatch):
+        stdout = dict(nodes=[
+            dict(id=0, status="destroyed"),
+        ])
+        stdout = ['', json.dumps(stdout)]
+        monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: (stdout, '', 0))
+        result = prepare.osd_id_available(0)
+        assert result
+
+
+class TestFormatDevice(object):
+
+    def test_include_force(self, fake_run, monkeypatch):
+        monkeypatch.setattr(conf, 'ceph', Factory(get_list=lambda *a, **kw: []))
+        prepare.format_device('/dev/sxx')
+        flags = fake_run.calls[0]['args'][0]
+        assert '-f' in flags
+
+    def test_device_is_always_appended(self, fake_run, conf_ceph):
+        conf_ceph(get_list=lambda *a, **kw: [])
+        prepare.format_device('/dev/sxx')
+        flags = fake_run.calls[0]['args'][0]
+        assert flags[-1] == '/dev/sxx'
+
+    def test_extra_flags_are_added(self, fake_run, conf_ceph):
+        conf_ceph(get_list=lambda *a, **kw: ['--why-yes'])
+        prepare.format_device('/dev/sxx')
+        flags = fake_run.calls[0]['args'][0]
+        assert '--why-yes' in flags
+
+    def test_default_options(self, conf_ceph_stub, fake_run):
+        conf_ceph_stub(dedent("""[global]
+        fsid = 1234lkjh1234"""))
+        conf.cluster = 'ceph'
+        prepare.format_device('/dev/sda1')
+        expected = [
+            'mkfs', '-t', 'xfs',
+            '-f', '-i', 'size=2048',  # default flags
+            '/dev/sda1']
+        assert expected == fake_run.calls[0]['args'][0]
+
+    def test_multiple_options_are_used(self, conf_ceph_stub, fake_run):
+        conf_ceph_stub(dedent("""[global]
+        fsid = 1234lkjh1234
+        [osd]
+        osd mkfs options xfs = -f -i size=1024"""))
+        conf.cluster = 'ceph'
+        prepare.format_device('/dev/sda1')
+        expected = [
+            'mkfs', '-t', 'xfs',
+            '-f', '-i', 'size=1024',
+            '/dev/sda1']
+        assert expected == fake_run.calls[0]['args'][0]
+
+    def test_multiple_options_will_get_the_force_flag(self, conf_ceph_stub, fake_run):
+        conf_ceph_stub(dedent("""[global]
+        fsid = 1234lkjh1234
+        [osd]
+        osd mkfs options xfs = -i size=1024"""))
+        conf.cluster = 'ceph'
+        prepare.format_device('/dev/sda1')
+        expected = [
+            'mkfs', '-t', 'xfs',
+            '-f', '-i', 'size=1024',
+            '/dev/sda1']
+        assert expected == fake_run.calls[0]['args'][0]
+
+    def test_underscore_options_are_used(self, conf_ceph_stub, fake_run):
+        conf_ceph_stub(dedent("""[global]
+        fsid = 1234lkjh1234
+        [osd]
+        osd_mkfs_options_xfs = -i size=128"""))
+        conf.cluster = 'ceph'
+        prepare.format_device('/dev/sda1')
+        expected = [
+            'mkfs', '-t', 'xfs',
+            '-f', '-i', 'size=128',
+            '/dev/sda1']
+        assert expected == fake_run.calls[0]['args'][0]
+
+
+mkfs_filestore_flags = [
+    'ceph-osd',
+    '--cluster',
+    '--osd-objectstore', 'filestore',
+    '--mkfs',
+    '-i',
+    '--monmap',
+    '--keyfile', '-', # goes through stdin
+    '--osd-data',
+    '--osd-journal',
+    '--osd-uuid',
+    '--setuser', 'ceph',
+    '--setgroup', 'ceph'
+]
+
+
+class TestOsdMkfsFilestore(object):
+
+    @pytest.mark.parametrize('flag', mkfs_filestore_flags)
+    def test_keyring_is_used(self, fake_call, monkeypatch, flag):
+        monkeypatch.setattr(prepare, '__release__', 'mimic')
+        monkeypatch.setattr(system, 'chown', lambda path: True)
+        prepare.osd_mkfs_filestore(1, 'asdf', keyring='secret')
+        assert flag in fake_call.calls[0]['args'][0]
+
+    def test_keyring_is_used_luminous(self, fake_call, monkeypatch):
+        monkeypatch.setattr(prepare, '__release__', 'luminous')
+        monkeypatch.setattr(system, 'chown', lambda path: True)
+        prepare.osd_mkfs_filestore(1, 'asdf', keyring='secret')
+        assert '--keyfile' not in fake_call.calls[0]['args'][0]
+
+
+class TestOsdMkfsBluestore(object):
+
+    def test_keyring_is_added(self, fake_call, monkeypatch):
+        monkeypatch.setattr(system, 'chown', lambda path: True)
+        prepare.osd_mkfs_bluestore(1, 'asdf', keyring='secret')
+        assert '--keyfile' in fake_call.calls[0]['args'][0]
+
+    def test_keyring_is_not_added(self, fake_call, monkeypatch):
+        monkeypatch.setattr(system, 'chown', lambda path: True)
+        prepare.osd_mkfs_bluestore(1, 'asdf')
+        assert '--keyfile' not in fake_call.calls[0]['args'][0]
+
+    def test_keyring_is_not_added_luminous(self, fake_call, monkeypatch):
+        monkeypatch.setattr(system, 'chown', lambda path: True)
+        prepare.osd_mkfs_bluestore(1, 'asdf')
+        monkeypatch.setattr(prepare, '__release__', 'luminous')
+        assert '--keyfile' not in fake_call.calls[0]['args'][0]
+
+    def test_wal_is_added(self, fake_call, monkeypatch):
+        monkeypatch.setattr(system, 'chown', lambda path: True)
+        prepare.osd_mkfs_bluestore(1, 'asdf', wal='/dev/smm1')
+        assert '--bluestore-block-wal-path' in fake_call.calls[0]['args'][0]
+        assert '/dev/smm1' in fake_call.calls[0]['args'][0]
+
+    def test_db_is_added(self, fake_call, monkeypatch):
+        monkeypatch.setattr(system, 'chown', lambda path: True)
+        prepare.osd_mkfs_bluestore(1, 'asdf', db='/dev/smm2')
+        assert '--bluestore-block-db-path' in fake_call.calls[0]['args'][0]
+        assert '/dev/smm2' in fake_call.calls[0]['args'][0]
+
+
+class TestMountOSD(object):
+
+    def test_default_options(self, conf_ceph_stub, fake_run):
+        conf_ceph_stub(dedent("""[global]
+        fsid = 1234lkjh1234"""))
+        conf.cluster = 'ceph'
+        prepare.mount_osd('/dev/sda1', 1)
+        expected = [
+            'mount', '-t', 'xfs', '-o',
+            'rw,noatime,inode64', # default flags
+            '/dev/sda1', '/var/lib/ceph/osd/ceph-1']
+        assert expected == fake_run.calls[0]['args'][0]
+
+    def test_mount_options_are_used(self, conf_ceph_stub, fake_run):
+        conf_ceph_stub(dedent("""[global]
+        fsid = 1234lkjh1234
+        [osd]
+        osd mount options xfs = rw"""))
+        conf.cluster = 'ceph'
+        prepare.mount_osd('/dev/sda1', 1)
+        expected = [
+            'mount', '-t', 'xfs', '-o',
+            'rw',
+            '/dev/sda1', '/var/lib/ceph/osd/ceph-1']
+        assert expected == fake_run.calls[0]['args'][0]
+
+    def test_multiple_whitespace_options_are_used(self, conf_ceph_stub, fake_run):
+        conf_ceph_stub(dedent("""[global]
+        fsid = 1234lkjh1234
+        [osd]
+        osd mount options xfs = rw auto exec"""))
+        conf.cluster = 'ceph'
+        prepare.mount_osd('/dev/sda1', 1)
+        expected = [
+            'mount', '-t', 'xfs', '-o',
+            'rw,auto,exec',
+            '/dev/sda1', '/var/lib/ceph/osd/ceph-1']
+        assert expected == fake_run.calls[0]['args'][0]
+
+    def test_multiple_comma_whitespace_options_are_used(self, conf_ceph_stub, fake_run):
+        conf_ceph_stub(dedent("""[global]
+        fsid = 1234lkjh1234
+        [osd]
+        osd mount options xfs = rw, auto, exec"""))
+        conf.cluster = 'ceph'
+        prepare.mount_osd('/dev/sda1', 1)
+        expected = [
+            'mount', '-t', 'xfs', '-o',
+            'rw,auto,exec',
+            '/dev/sda1', '/var/lib/ceph/osd/ceph-1']
+        assert expected == fake_run.calls[0]['args'][0]
+
+    def test_underscore_mount_options_are_used(self, conf_ceph_stub, fake_run):
+        conf_ceph_stub(dedent("""[global]
+        fsid = 1234lkjh1234
+        [osd]
+        osd mount options xfs = rw"""))
+        conf.cluster = 'ceph'
+        prepare.mount_osd('/dev/sda1', 1)
+        expected = [
+            'mount', '-t', 'xfs', '-o',
+            'rw',
+            '/dev/sda1', '/var/lib/ceph/osd/ceph-1']
+        assert expected == fake_run.calls[0]['args'][0]
+
+
+ceph_conf_mount_values = [
+    ['rw,', 'auto,' 'exec'],
+    ['rw', 'auto', 'exec'],
+    [' rw ', ' auto ', ' exec '],
+    ['rw,', 'auto,', 'exec,'],
+    [',rw ', ',auto ', ',exec,'],
+    [',rw,', ',auto,', ',exec,'],
+]
+
+string_mount_values = [
+    'rw, auto exec ',
+    'rw  auto exec',
+    ',rw, auto, exec,',
+    ' rw  auto exec ',
+    ' rw,auto,exec ',
+    'rw,auto,exec',
+    ',rw,auto,exec,',
+    'rw,auto,exec ',
+    'rw, auto, exec ',
+]
+
+
+class TestNormalizeFlags(object):
+    # a bit overkill since most of this is already tested in prepare.mount_osd
+    # tests
+
+    @pytest.mark.parametrize("flags", ceph_conf_mount_values)
+    def test_normalize_lists(self, flags):
+        result = sorted(prepare._normalize_mount_flags(flags).split(','))
+        assert ','.join(result) == 'auto,exec,rw'
+
+    @pytest.mark.parametrize("flags", string_mount_values)
+    def test_normalize_strings(self, flags):
+        result = sorted(prepare._normalize_mount_flags(flags).split(','))
+        assert ','.join(result) == 'auto,exec,rw'
+
+    @pytest.mark.parametrize("flags", ceph_conf_mount_values)
+    def test_normalize_extra_flags(self, flags):
+        result = prepare._normalize_mount_flags(flags, extras=['discard'])
+        assert sorted(result.split(',')) == ['auto', 'discard', 'exec', 'rw']
+
+    @pytest.mark.parametrize("flags", ceph_conf_mount_values)
+    def test_normalize_duplicate_extra_flags(self, flags):
+        result = prepare._normalize_mount_flags(flags, extras=['rw', 'discard'])
+        assert sorted(result.split(',')) == ['auto', 'discard', 'exec', 'rw']
+
+    @pytest.mark.parametrize("flags", string_mount_values)
+    def test_normalize_strings_flags(self, flags):
+        result = sorted(prepare._normalize_mount_flags(flags, extras=['discard']).split(','))
+        assert ','.join(result) == 'auto,discard,exec,rw'
+
+    @pytest.mark.parametrize("flags", string_mount_values)
+    def test_normalize_strings_duplicate_flags(self, flags):
+        result = sorted(prepare._normalize_mount_flags(flags, extras=['discard','rw']).split(','))
+        assert ','.join(result) == 'auto,discard,exec,rw'
+
+
+class TestMkfsFilestore(object):
+
+    def test_non_zero_exit_status(self, stub_call, monkeypatch):
+        conf.cluster = 'ceph'
+        monkeypatch.setattr('ceph_volume.util.prepare.system.chown', lambda x: True)
+        stub_call(([], [], 1))
+        with pytest.raises(RuntimeError) as error:
+            prepare.osd_mkfs_filestore('1', 'asdf-1234', 'keyring')
+        assert "Command failed with exit code 1" in str(error.value)
+
+    def test_non_zero_exit_formats_command_correctly(self, stub_call, monkeypatch):
+        conf.cluster = 'ceph'
+        monkeypatch.setattr('ceph_volume.util.prepare.system.chown', lambda x: True)
+        stub_call(([], [], 1))
+        with pytest.raises(RuntimeError) as error:
+            prepare.osd_mkfs_filestore('1', 'asdf-1234', 'keyring')
+        expected = ' '.join([
+            'ceph-osd',
+            '--cluster',
+            'ceph',
+            '--osd-objectstore', 'filestore', '--mkfs',
+            '-i', '1', '--monmap', '/var/lib/ceph/osd/ceph-1/activate.monmap',
+            '--keyfile', '-', '--osd-data', '/var/lib/ceph/osd/ceph-1/',
+            '--osd-journal', '/var/lib/ceph/osd/ceph-1/journal',
+            '--osd-uuid', 'asdf-1234',
+            '--setuser', 'ceph', '--setgroup', 'ceph'])
+        assert expected in str(error.value)
+
+
+class TestMkfsBluestore(object):
+
+    def test_non_zero_exit_status(self, stub_call, monkeypatch):
+        conf.cluster = 'ceph'
+        monkeypatch.setattr('ceph_volume.util.prepare.system.chown', lambda x: True)
+        stub_call(([], [], 1))
+        with pytest.raises(RuntimeError) as error:
+            prepare.osd_mkfs_bluestore('1', 'asdf-1234', keyring='keyring')
+        assert "Command failed with exit code 1" in str(error.value)
+
+    def test_non_zero_exit_formats_command_correctly(self, stub_call, monkeypatch):
+        conf.cluster = 'ceph'
+        monkeypatch.setattr('ceph_volume.util.prepare.system.chown', lambda x: True)
+        stub_call(([], [], 1))
+        with pytest.raises(RuntimeError) as error:
+            prepare.osd_mkfs_bluestore('1', 'asdf-1234', keyring='keyring')
+        expected = ' '.join([
+            'ceph-osd',
+            '--cluster',
+            'ceph',
+            '--osd-objectstore', 'bluestore', '--mkfs',
+            '-i', '1', '--monmap', '/var/lib/ceph/osd/ceph-1/activate.monmap',
+            '--keyfile', '-', '--osd-data', '/var/lib/ceph/osd/ceph-1/',
+            '--osd-uuid', 'asdf-1234',
+            '--setuser', 'ceph', '--setgroup', 'ceph'])
+        assert expected in str(error.value)
+
+
+class TestGetJournalSize(object):
+
+    def test_undefined_size_fallbacks_formatted(self, conf_ceph_stub):
+        conf_ceph_stub(dedent("""
+        [global]
+        fsid = a25d19a6-7d57-4eda-b006-78e35d2c4d9f
+        """))
+        result = prepare.get_journal_size()
+        assert result == '5G'
+
+    def test_undefined_size_fallbacks_unformatted(self, conf_ceph_stub):
+        conf_ceph_stub(dedent("""
+        [global]
+        fsid = a25d19a6-7d57-4eda-b006-78e35d2c4d9f
+        """))
+        result = prepare.get_journal_size(lv_format=False)
+        assert result.gb.as_int() == 5
+
+    def test_defined_size_unformatted(self, conf_ceph_stub):
+        conf_ceph_stub(dedent("""
+        [global]
+        fsid = a25d19a6-7d57-4eda-b006-78e35d2c4d9f
+
+        [osd]
+        osd journal size = 10240
+        """))
+        result = prepare.get_journal_size(lv_format=False)
+        assert result.gb.as_int() == 10
+
+    def test_defined_size_formatted(self, conf_ceph_stub):
+        conf_ceph_stub(dedent("""
+        [global]
+        fsid = a25d19a6-7d57-4eda-b006-78e35d2c4d9f
+
+        [osd]
+        osd journal size = 10240
+        """))
+        result = prepare.get_journal_size()
+        assert result == '10G'
+
+    def test_refuse_tiny_journals(self, conf_ceph_stub):
+        conf_ceph_stub(dedent("""
+        [global]
+        fsid = a25d19a6-7d57-4eda-b006-78e35d2c4d9f
+
+        [osd]
+        osd journal size = 1024
+        """))
+        with pytest.raises(RuntimeError) as error:
+            prepare.get_journal_size()
+        assert 'journal sizes must be larger' in str(error.value)
+        assert 'detected: 1024.00 MB' in str(error.value)
diff --git a/src/ceph-volume/ceph_volume/tests/util/test_system.py b/src/ceph-volume/ceph_volume/tests/util/test_system.py
new file mode 100644
index 000000000..5746f7023
--- /dev/null
+++ b/src/ceph-volume/ceph_volume/tests/util/test_system.py
@@ -0,0 +1,309 @@
+import os
+import pwd
+import getpass
+import pytest
+from textwrap import dedent
+from ceph_volume.util import system
+from mock.mock import patch
+from ceph_volume.tests.conftest import Factory
+
+
+@pytest.fixture
+def mock_find_executable_on_host(monkeypatch):
+    """
+    Monkeypatches util.system.find_executable_on_host, so that a caller can add behavior to the response
+    """
+    def apply(stdout=None, stderr=None, returncode=0):
+        stdout_stream = Factory(read=lambda: stdout)
+        stderr_stream = Factory(read=lambda: stderr)
+        return_value = Factory(
+            stdout=stdout_stream,
+            stderr=stderr_stream,
+            wait=lambda: returncode,
+            communicate=lambda x: (stdout, stderr, returncode)
+        )
+
+        monkeypatch.setattr(
+            'ceph_volume.util.system.subprocess.Popen',
+            lambda *a, **kw: return_value)
+
+    return apply
+
+class TestMkdirP(object):
+
+    def test_existing_dir_does_not_raise_w_chown(self, monkeypatch, tmpdir):
+        user = pwd.getpwnam(getpass.getuser())
+        uid, gid = user[2], user[3]
+        monkeypatch.setattr(system, 'get_ceph_user_ids', lambda: (uid, gid,))
+        path = str(tmpdir)
+        system.mkdir_p(path)
+        assert os.path.isdir(path)
+
+    def test_new_dir_w_chown(self, monkeypatch, tmpdir):
+        user = pwd.getpwnam(getpass.getuser())
+        uid, gid = user[2], user[3]
+        monkeypatch.setattr(system, 'get_ceph_user_ids', lambda: (uid, gid,))
+        path = os.path.join(str(tmpdir), 'new')
+        system.mkdir_p(path)
+        assert os.path.isdir(path)
+
+    def test_existing_dir_does_not_raise_no_chown(self, tmpdir):
+        path = str(tmpdir)
+        system.mkdir_p(path, chown=False)
+        assert os.path.isdir(path)
+
+    def test_new_dir_no_chown(self, tmpdir):
+        path = os.path.join(str(tmpdir), 'new')
+        system.mkdir_p(path, chown=False)
+        assert os.path.isdir(path)
+
+
+@pytest.fixture
+def fake_proc(tmpdir, monkeypatch):
+    PROCDIR = str(tmpdir)
+    proc_path = os.path.join(PROCDIR, 'mounts')
+    with open(proc_path, 'w') as f:
+        f.write(dedent("""nfsd /proc/fs/nfsd nfsd rw,relatime 0 0
+            rootfs / rootfs rw 0 0
+            sysfs /sys sysfs rw,seclabel,nosuid,nodev,noexec,relatime 0 0
+            proc /proc proc rw,nosuid,nodev,noexec,relatime 0 0
+            devtmpfs /dev devtmpfs rw,seclabel,nosuid,size=238292k,nr_inodes=59573,mode=755 0 0
+            securityfs /sys/kernel/security securityfs rw,nosuid,nodev,noexec,relatime 0 0
+            tmpfs /dev/shm tmpfs rw,seclabel,nosuid,nodev 0 0
+            devpts /dev/pts devpts rw,seclabel,nosuid,noexec,relatime,gid=5,mode=620,ptmxmode=000 0 0
+            tmpfs /run tmpfs rw,seclabel,nosuid,nodev,mode=755 0 0
+            tmpfs /sys/fs/cgroup tmpfs ro,seclabel,nosuid,nodev,noexec,mode=755 0 0
+            cgroup /sys/fs/cgroup/systemd cgroup rw,nosuid,nodev,noexec,relatime,xattr,release_agent=/usr/lib/systemd/systemd-cgroups-agent,name=systemd 0 0
+            cgroup /sys/fs/cgroup/freezer cgroup rw,nosuid,nodev,noexec,relatime,freezer 0 0
+            configfs /sys/kernel/config configfs rw,relatime 0 0
+            /dev/mapper/VolGroup00-LogVol00 / xfs rw,seclabel,relatime,attr2,inode64,noquota 0 0
+            selinuxfs /sys/fs/selinux selinuxfs rw,relatime 0 0
+            debugfs /sys/kernel/debug debugfs rw,relatime 0 0
+            hugetlbfs /dev/hugepages hugetlbfs rw,seclabel,relatime 0 0
+            mqueue /dev/mqueue mqueue rw,seclabel,relatime 0 0
+            sunrpc /far/lib/nfs/rpc_pipefs rpc_pipefs rw,relatime 0 0
+            /dev/sde4 /two/field/path
+            nfsd /proc/fs/nfsd nfsd rw,relatime 0 0
+            /dev/sde2 /boot xfs rw,seclabel,relatime,attr2,inode64,noquota 0 0
+            tmpfs /far/lib/ceph/osd/ceph-5 tmpfs rw,seclabel,relatime 0 0
+            tmpfs /far/lib/ceph/osd/ceph-7 tmpfs rw,seclabel,relatime 0 0
+            /dev/sda1 /far/lib/ceph/osd/ceph-0 xfs rw,seclabel,noatime,attr2,inode64,noquota 0 0
+            tmpfs /run/user/1000 tmpfs rw,seclabel,nosuid,nodev,relatime,size=50040k,mode=700,uid=1000,gid=1000 0 0
+            /dev/sdc2 /boot xfs rw,seclabel,relatime,attr2,inode64,noquota 0 0
+            tmpfs /run/user/1000 tmpfs rw,seclabel,mode=700,uid=1000,gid=1000 0 0"""))
+    monkeypatch.setattr(system, 'PROCDIR', PROCDIR)
+    monkeypatch.setattr(os.path, 'exists', lambda x: True)
+
+
+class TestPathIsMounted(object):
+
+    def test_is_mounted(self, fake_proc):
+        assert system.path_is_mounted('/boot') is True
+
+    def test_is_not_mounted(self, fake_proc):
+        assert system.path_is_mounted('/far/fib/feph') is False
+
+    def test_is_not_mounted_at_destination(self, fake_proc):
+        assert system.path_is_mounted('/boot', destination='/dev/sda1') is False
+
+    def test_is_mounted_at_destination(self, fake_proc):
+        assert system.path_is_mounted('/boot', destination='/dev/sdc2') is True
+
+
+class TestDeviceIsMounted(object):
+
+    def test_is_mounted(self, fake_proc):
+        assert system.device_is_mounted('/dev/sda1') is True
+
+    def test_path_is_not_device(self, fake_proc):
+        assert system.device_is_mounted('/far/lib/ceph/osd/ceph-7') is False
+
+    def test_is_not_mounted_at_destination(self, fake_proc):
+        assert system.device_is_mounted('/dev/sda1', destination='/far/lib/ceph/osd/test-1') is False
+
+    def test_is_mounted_at_destination(self, fake_proc):
+        assert system.device_is_mounted('/dev/sda1', destination='/far/lib/ceph/osd/ceph-7') is False
+
+    def test_is_realpath_dev_mounted_at_destination(self, fake_proc, monkeypatch):
+        monkeypatch.setattr(system.os.path, 'realpath', lambda x: '/dev/sda1' if 'foo' in x else x)
+        result = system.device_is_mounted('/dev/maper/foo', destination='/far/lib/ceph/osd/ceph-0')
+        assert result is True
+
+    def test_is_realpath_path_mounted_at_destination(self, fake_proc, monkeypatch):
+        monkeypatch.setattr(
+            system.os.path, 'realpath',
+            lambda x: '/far/lib/ceph/osd/ceph-0' if 'symlink' in x else x)
+        result = system.device_is_mounted('/dev/sda1', destination='/symlink/lib/ceph/osd/ceph-0')
+        assert result is True
+
+
+class TestGetMounts(object):
+
+    def test_not_mounted(self, tmpdir, monkeypatch):
+        PROCDIR = str(tmpdir)
+        proc_path = os.path.join(PROCDIR, 'mounts')
+        with open(proc_path, 'w') as f:
+            f.write('')
+        monkeypatch.setattr(system, 'PROCDIR', PROCDIR)
+        m = system.Mounts()
+        assert m.get_mounts() == {}
+
+    def test_is_mounted_(self, fake_proc):
+        m = system.Mounts()
+        assert m.get_mounts()['/dev/sdc2'] == ['/boot']
+
+    def test_ignores_two_fields(self, fake_proc):
+        m = system.Mounts()
+        assert m.get_mounts().get('/dev/sde4') is None
+
+    def test_tmpfs_is_reported(self, fake_proc):
+        m = system.Mounts()
+        assert m.get_mounts()['tmpfs'][0] == '/dev/shm'
+
+    def test_non_skip_devs_arent_reported(self, fake_proc):
+        m = system.Mounts()
+        assert m.get_mounts().get('cgroup') is None
+
+    def test_multiple_mounts_are_appended(self, fake_proc):
+        m = system.Mounts()
+        assert len(m.get_mounts()['tmpfs']) == 7
+
+    def test_nonexistent_devices_are_skipped(self, tmpdir, monkeypatch):
+        PROCDIR = str(tmpdir)
+        proc_path = os.path.join(PROCDIR, 'mounts')
+        with open(proc_path, 'w') as f:
+            f.write(dedent("""nfsd /proc/fs/nfsd nfsd rw,relatime 0 0
+                    /dev/sda1 /far/lib/ceph/osd/ceph-0 xfs rw,attr2,inode64,noquota 0 0
+                    /dev/sda2 /far/lib/ceph/osd/ceph-1 xfs rw,attr2,inode64,noquota 0 0"""))
+        monkeypatch.setattr(system, 'PROCDIR', PROCDIR)
+        monkeypatch.setattr(os.path, 'exists', lambda x: False if x == '/dev/sda1' else True)
+        m = system.Mounts()
+        assert m.get_mounts().get('/dev/sda1') is None
+
+
+class TestIsBinary(object):
+
+    def test_is_binary(self, fake_filesystem):
+        binary_path = fake_filesystem.create_file('/tmp/fake-file', contents='asd\n\nlkjh\x00')
+        assert system.is_binary(binary_path.path)
+
+    def test_is_not_binary(self, fake_filesystem):
+        binary_path = fake_filesystem.create_file('/tmp/fake-file', contents='asd\n\nlkjh0')
+        assert system.is_binary(binary_path.path) is False
+
+
+class TestGetFileContents(object):
+
+    def test_path_does_not_exist(self, tmpdir):
+        filepath = os.path.join(str(tmpdir), 'doesnotexist')
+        assert system.get_file_contents(filepath, 'default') == 'default'
+
+    def test_path_has_contents(self, fake_filesystem):
+        interesting_file = fake_filesystem.create_file('/tmp/fake-file', contents="1")
+        result = system.get_file_contents(interesting_file.path)
+        assert result == "1"
+
+    def test_path_has_multiline_contents(self, fake_filesystem):
+        interesting_file = fake_filesystem.create_file('/tmp/fake-file', contents="0\n1")
+        result = system.get_file_contents(interesting_file.path)
+        assert result == "0\n1"
+
+    def test_exception_returns_default(self):
+        with patch('builtins.open') as mocked_open:
+            mocked_open.side_effect = Exception()
+            result = system.get_file_contents('/tmp/fake-file')
+        assert result == ''
+
+
+class TestWhich(object):
+
+    def test_executable_exists_but_is_not_file(self, monkeypatch):
+        monkeypatch.setattr(system.os.path, 'isfile', lambda x: False)
+        monkeypatch.setattr(system.os.path, 'exists', lambda x: True)
+        assert system.which('exedir') == 'exedir'
+
+    def test_executable_does_not_exist(self, monkeypatch):
+        monkeypatch.setattr(system.os.path, 'isfile', lambda x: False)
+        monkeypatch.setattr(system.os.path, 'exists', lambda x: False)
+        assert system.which('exedir') == 'exedir'
+
+    def test_executable_exists_as_file(self, monkeypatch):
+        monkeypatch.setattr(system.os, 'getenv', lambda x, y: '')
+        monkeypatch.setattr(system.os.path, 'isfile', lambda x: x != 'ceph')
+        monkeypatch.setattr(system.os.path, 'exists', lambda x: x != 'ceph')
+        assert system.which('ceph') == '/usr/local/bin/ceph'
+
+    def test_warnings_when_executable_isnt_matched(self, monkeypatch, capsys):
+        monkeypatch.setattr(system.os.path, 'isfile', lambda x: True)
+        monkeypatch.setattr(system.os.path, 'exists', lambda x: False)
+        system.which('exedir')
+        cap = capsys.readouterr()
+        assert 'Executable exedir not in PATH' in cap.err
+
+    def test_run_on_host_found(self, mock_find_executable_on_host):
+        mock_find_executable_on_host(stdout="/sbin/lvs\n", stderr="some stderr message\n")
+        assert system.which('lvs', run_on_host=True) == '/sbin/lvs'
+
+    def test_run_on_host_not_found(self, mock_find_executable_on_host):
+        mock_find_executable_on_host(stdout="", stderr="some stderr message\n")
+        assert system.which('lvs', run_on_host=True) == 'lvs'
+
+@pytest.fixture
+def stub_which(monkeypatch):
+    def apply(value='/bin/restorecon'):
+        monkeypatch.setattr(system, 'which', lambda x: value)
+    return apply
+
+
+# python2 has no FileNotFoundError
+try:
+    FileNotFoundError
+except NameError:
+    FileNotFoundError = OSError
+
+
+class TestSetContext(object):
+
+    def setup(self):
+        try:
+            os.environ.pop('CEPH_VOLUME_SKIP_RESTORECON')
+        except KeyError:
+            pass
+
+    @pytest.mark.parametrize('value', ['1', 'True', 'true', 'TRUE', 'yes'])
+    def test_set_context_skips(self, stub_call, fake_run, value):
+        stub_call(('', '', 0))
+        os.environ['CEPH_VOLUME_SKIP_RESTORECON'] = value
+        system.set_context('/tmp/foo')
+        assert fake_run.calls == []
+
+    @pytest.mark.parametrize('value', ['0', 'False', 'false', 'FALSE', 'no'])
+    def test_set_context_doesnt_skip_with_env(self, stub_call, stub_which, fake_run, value):
+        stub_call(('', '', 0))
+        stub_which()
+        os.environ['CEPH_VOLUME_SKIP_RESTORECON'] = value
+        system.set_context('/tmp/foo')
+        assert len(fake_run.calls)
+
+    def test_set_context_skips_on_executable(self, stub_call, stub_which, fake_run):
+        stub_call(('', '', 0))
+        stub_which('restorecon')
+        system.set_context('/tmp/foo')
+        assert fake_run.calls == []
+
+    def test_set_context_no_skip_on_executable(self, stub_call, stub_which, fake_run):
+        stub_call(('', '', 0))
+        stub_which('/bin/restorecon')
+        system.set_context('/tmp/foo')
+        assert len(fake_run.calls)
+
+    @patch('ceph_volume.process.call')
+    def test_selinuxenabled_doesnt_exist(self, mocked_call, fake_run):
+        mocked_call.side_effect = FileNotFoundError()
+        system.set_context('/tmp/foo')
+        assert fake_run.calls == []
+
+    def test_selinuxenabled_is_not_enabled(self, stub_call, fake_run):
+        stub_call(('', '', 1))
+        system.set_context('/tmp/foo')
+        assert fake_run.calls == []
diff --git a/src/ceph-volume/ceph_volume/tests/util/test_util.py b/src/ceph-volume/ceph_volume/tests/util/test_util.py
new file mode 100644
index 000000000..1a094d33f
--- /dev/null
+++ b/src/ceph-volume/ceph_volume/tests/util/test_util.py
@@ -0,0 +1,116 @@
+import pytest
+from ceph_volume import util
+
+
+class TestAsBytes(object):
+
+    def test_bytes_just_gets_returned(self):
+        bytes_string = "contents".encode('utf-8')
+        assert util.as_bytes(bytes_string) == bytes_string
+
+    def test_string_gets_converted_to_bytes(self):
+        result = util.as_bytes('contents')
+        assert isinstance(result, bytes)
+
+
+class TestStrToInt(object):
+
+    def test_passing_a_float_str_comma(self):
+        result = util.str_to_int("1,99")
+        assert result == 1
+
+    def test_passing_a_float_does_not_round_comma(self):
+        result = util.str_to_int("1,99", round_down=False)
+        assert result == 2
+
+    @pytest.mark.parametrize("value", ['2', 2])
+    def test_passing_an_int(self, value):
+        result = util.str_to_int(value)
+        assert result == 2
+
+    @pytest.mark.parametrize("value", ['1.99', 1.99])
+    def test_passing_a_float(self, value):
+        result = util.str_to_int(value)
+        assert result == 1
+
+    @pytest.mark.parametrize("value", ['1.99', 1.99])
+    def test_passing_a_float_does_not_round(self, value):
+        result = util.str_to_int(value, round_down=False)
+        assert result == 2
+
+    def test_text_is_not_an_integer_like(self):
+        with pytest.raises(RuntimeError) as error:
+            util.str_to_int("1.4GB")
+        assert str(error.value) == "Unable to convert to integer: '1.4GB'"
+
+    def test_input_is_not_string(self):
+        with pytest.raises(RuntimeError) as error:
+            util.str_to_int(None)
+        assert str(error.value) == "Unable to convert to integer: 'None'"
+
+
+def true_responses(upper_casing=False):
+    if upper_casing:
+        return ['Y', 'YES', '']
+    return ['y', 'yes', '']
+
+
+def false_responses(upper_casing=False):
+    if upper_casing:
+        return ['N', 'NO']
+    return ['n', 'no']
+
+
+def invalid_responses():
+    return [9, 0.1, 'h', [], {}, None]
+
+
+class TestStrToBool(object):
+
+    @pytest.mark.parametrize('response', true_responses())
+    def test_trueish(self, response):
+        assert util.str_to_bool(response) is True
+
+    @pytest.mark.parametrize('response', false_responses())
+    def test_falseish(self, response):
+        assert util.str_to_bool(response) is False
+
+    @pytest.mark.parametrize('response', true_responses(True))
+    def test_trueish_upper(self, response):
+        assert util.str_to_bool(response) is True
+
+    @pytest.mark.parametrize('response', false_responses(True))
+    def test_falseish_upper(self, response):
+        assert util.str_to_bool(response) is False
+
+    @pytest.mark.parametrize('response', invalid_responses())
+    def test_invalid(self, response):
+        with pytest.raises(ValueError):
+            util.str_to_bool(response)
+
+
+class TestPromptBool(object):
+
+    @pytest.mark.parametrize('response', true_responses())
+    def test_trueish(self, response):
+        fake_input = lambda x: response
+        qx = 'what the what?'
+        assert util.prompt_bool(qx, input_=fake_input) is True
+
+    @pytest.mark.parametrize('response', false_responses())
+    def test_falseish(self, response):
+        fake_input = lambda x: response
+        qx = 'what the what?'
+        assert util.prompt_bool(qx, input_=fake_input) is False
+
+    def test_try_again_true(self):
+        responses = ['g', 'h', 'y']
+        fake_input = lambda x: responses.pop(0)
+        qx = 'what the what?'
+        assert util.prompt_bool(qx, input_=fake_input) is True
+
+    def test_try_again_false(self):
+        responses = ['g', 'h', 'n']
+        fake_input = lambda x: responses.pop(0)
+        qx = 'what the what?'
+        assert util.prompt_bool(qx, input_=fake_input) is False