"""
monkeypatch.setattr('os.getuid', lambda: 0)
+@pytest.fixture
+def is_non_root(monkeypatch):
+ """
+ Patch ``os.getuid()`` so that ceph-volume's decorators that ensure a user
+ is not root.
+ """
+ monkeypatch.setattr('os.getuid', lambda: 100)
@pytest.fixture
def tmpfile(tmpdir):
fs.create_dir('/sys/block/sda/slaves')
fs.create_dir('/sys/block/sda/queue')
fs.create_dir('/sys/block/rbd0')
+ fs.create_dir('/var/log/ceph')
+ fs.create_dir('/tmp/osdpath')
yield fs
@pytest.fixture
a.objectstore.activate()
assert result.value.args[0] == 'Please provide both osd_id and osd_fsid'
- def test_bluestore_no_systemd(self, is_root, monkeypatch, capture):
+ def test_bluestore_no_systemd(self, m_create_key, is_root, monkeypatch, capture):
monkeypatch.setattr('ceph_volume.configuration.load', lambda: None)
fake_enable = Capture()
fake_start_osd = Capture()
assert fake_enable.calls == []
assert fake_start_osd.calls == []
- def test_bluestore_systemd(self, is_root, monkeypatch, capture):
+ def test_bluestore_systemd(self, m_create_key, is_root, monkeypatch, capture):
monkeypatch.setattr('ceph_volume.configuration.load', lambda: None)
fake_enable = Capture()
fake_start_osd = Capture()
assert fake_enable.calls != []
assert fake_start_osd.calls != []
- def test_bluestore_no_systemd_autodetect(self, is_root, monkeypatch, capture):
+ def test_bluestore_no_systemd_autodetect(self, m_create_key, is_root, monkeypatch, capture):
monkeypatch.setattr('ceph_volume.configuration.load', lambda: None)
fake_enable = Capture()
fake_start_osd = Capture()
assert fake_enable.calls == []
assert fake_start_osd.calls == []
- def test_bluestore_systemd_autodetect(self, is_root, monkeypatch, capture):
+ def test_bluestore_systemd_autodetect(self, m_create_key, is_root, monkeypatch, capture):
monkeypatch.setattr('ceph_volume.configuration.load', lambda: None)
fake_enable = Capture()
fake_start_osd = Capture()
m_activate.assert_has_calls(calls)
@patch('ceph_volume.objectstore.lvmbluestore.LvmBlueStore.activate')
- def test_detects_osds_to_activate_no_systemd(self, m_activate, is_root, monkeypatch):
+ def test_detects_osds_to_activate_no_systemd(self, m_activate, m_create_key, is_root, monkeypatch):
monkeypatch.setattr('ceph_volume.objectstore.lvmbluestore.direct_report', lambda: direct_report)
args = ['--all', '--no-systemd', '--bluestore']
a = activate.Activate(args)
def mock_prepare_dmcrypt(self, *args, **kwargs):
return '/dev/mapper/' + kwargs['mapping']
- @patch('os.getuid', return_value=1)
- def test_newdb_non_root(self, m_getuid):
+ def test_newdb_non_root(self, is_non_root):
with pytest.raises(Exception) as error:
migrate.NewDB(argv=[
'--osd-id', '1',
expected = 'This command needs to be executed with sudo or as root'
assert expected in str(error.value)
- @patch('os.getuid')
- def test_newdb_not_target_lvm(self, m_getuid, capsys):
- m_getuid.return_value = 0
+ def test_newdb_not_target_lvm(self, is_root, capsys):
with pytest.raises(SystemExit) as error:
migrate.NewDB(argv=[
'--osd-id', '1',
assert expected in stderr
- @patch('os.getuid')
- def test_newdb_already_in_use(self, m_getuid, monkeypatch, capsys):
- m_getuid.return_value = 0
-
+ def test_newdb_already_in_use(self, is_root, monkeypatch, capsys):
self.mock_volume = api.Volume(lv_name='volume1',
lv_uuid='y',
vg_name='vg',
expected = 'Target Logical Volume is already used by ceph: vgname/new_db'
assert expected in stderr
- @patch('os.getuid')
- def test_newdb(self, m_getuid, monkeypatch, capsys):
- m_getuid.return_value = 0
-
+ def test_newdb(self, is_root, monkeypatch, capsys):
source_tags = \
'ceph.osd_id=0,ceph.type=data,ceph.osd_fsid=1234,'\
'ceph.wal_uuid=wal_uuid,ceph.db_device=/dbdevice'
'--dev-target', '/dev/VolGroup/target_volume',
'--command', 'bluefs-bdev-new-db']
- @patch('os.getuid')
- def test_newwal(self, m_getuid, monkeypatch, capsys):
- m_getuid.return_value = 0
-
+ def test_newwal(self, is_root, monkeypatch, capsys):
source_tags = \
'ceph.osd_id=0,ceph.type=data,ceph.osd_fsid=1234'
assert not stderr
- @patch('os.getuid')
- def test_migrate_data_db_to_new_db(self, m_getuid, monkeypatch):
- m_getuid.return_value = 0
+ def test_migrate_data_db_to_new_db(self, is_root, monkeypatch):
source_tags = 'ceph.osd_id=2,ceph.type=data,ceph.osd_fsid=1234,' \
'ceph.cluster_name=ceph,ceph.db_uuid=dbuuid,ceph.db_device=db_dev'
'--devs-source', '/var/lib/ceph/osd/ceph-2/block',
'--devs-source', '/var/lib/ceph/osd/ceph-2/block.db']
- @patch('os.getuid')
- def test_migrate_data_db_to_new_db_skip_wal(self, m_getuid, monkeypatch):
- m_getuid.return_value = 0
-
+ def test_migrate_data_db_to_new_db_skip_wal(self, is_root, monkeypatch):
source_tags = 'ceph.osd_id=2,ceph.type=data,ceph.osd_fsid=1234,' \
'ceph.cluster_name=ceph,ceph.db_uuid=dbuuid,ceph.db_device=db_dev'
source_db_tags = 'ceph.osd_id=2,ceph.type=db,ceph.osd_fsid=1234,' \
'--devs-source', '/var/lib/ceph/osd/ceph-2/block',
'--devs-source', '/var/lib/ceph/osd/ceph-2/block.db']
- @patch('os.getuid')
- def test_migrate_data_db_wal_to_new_db(self, m_getuid, monkeypatch):
- m_getuid.return_value = 0
-
+ def test_migrate_data_db_wal_to_new_db(self, is_root, monkeypatch):
source_tags = 'ceph.osd_id=2,ceph.type=data,ceph.osd_fsid=1234,' \
'ceph.cluster_name=ceph,ceph.db_uuid=dbuuid,ceph.db_device=db_dev,' \
'ceph.wal_uuid=waluuid,ceph.wal_device=wal_dev'
monkeypatch,
capsys):
m_getuid.return_value = 0
-
source_tags = 'ceph.osd_id=2,ceph.type=data,ceph.osd_fsid=1234,' \
'ceph.cluster_name=ceph,ceph.db_uuid=dbuuid,ceph.db_device=db_dev'
source_db_tags = 'ceph.osd_id=2,ceph.type=db,ceph.osd_fsid=1234,' \
' please use new-db or new-wal command before.'
assert expected in stderr
- @patch('os.getuid')
def test_dont_migrate_db_to_wal(self,
- m_getuid,
+ is_root,
monkeypatch,
capsys):
- m_getuid.return_value = 0
-
source_tags = 'ceph.osd_id=2,ceph.type=data,ceph.osd_fsid=1234,' \
'ceph.cluster_name=ceph,ceph.db_uuid=dbuuid,ceph.db_device=db_dev,' \
'ceph.wal_uuid=waluuid,ceph.wal_device=wal_dev'
expected = 'Migrate to WAL is not supported'
assert expected in stderr
- @patch('os.getuid')
def test_migrate_data_db_to_db(self,
- m_getuid,
+ is_root,
monkeypatch,
capsys):
- m_getuid.return_value = 0
-
source_tags = 'ceph.osd_id=2,ceph.type=data,ceph.osd_fsid=1234,' \
'ceph.cluster_name=ceph,ceph.db_uuid=dbuuid,ceph.db_device=db_dev,' \
'ceph.wal_uuid=waluuid,ceph.wal_device=wal_dev'
'--command', 'bluefs-bdev-migrate',
'--devs-source', '/var/lib/ceph/osd/ceph-2/block']
- @patch('os.getuid')
def test_migrate_data_wal_to_db(self,
- m_getuid,
+ is_root,
monkeypatch,
capsys):
- m_getuid.return_value = 0
-
source_tags = 'ceph.osd_id=2,ceph.type=data,ceph.osd_fsid=1234,' \
'ceph.cluster_name=ceph,ceph.db_uuid=dbuuid,ceph.db_device=db_dev,' \
'ceph.wal_uuid=waluuid,ceph.wal_device=wal_dev'
@patch('ceph_volume.util.prepare.create_key', return_value='fake-secret')
class TestGetClusterFsid(object):
- def setup(self):
+ def setup_method(self):
self.p = lvm.prepare.Prepare([])
def test_fsid_is_passed_in(self, m_create_key, factory):
@patch('ceph_volume.util.prepare.create_key', return_value='fake-secret')
class TestPrepare(object):
- def setup(self):
+ def setup_method(self):
self.p = lvm.prepare.Prepare([])
def test_main_spits_help_with_no_arguments(self, m_create_key, capsys):
"/dev/sdb3": {},
"/dev/sdc": {},
"/dev/sdd": {},
+ "/dev/sde": {},
+ "/dev/sde1": {},
"/dev/mapper/ceph--osd--block--1": {},
"/dev/mapper/ceph--osd--block--2": {},
}
def _lsblk_all_devices(abspath=True):
return [
- {"NAME": "/dev/sda", "KNAME": "/dev/sda", "PKNAME": ""},
- {"NAME": "/dev/sda1", "KNAME": "/dev/sda1", "PKNAME": "/dev/sda"},
- {"NAME": "/dev/sda2", "KNAME": "/dev/sda2", "PKNAME": "/dev/sda"},
- {"NAME": "/dev/sda3", "KNAME": "/dev/sda3", "PKNAME": "/dev/sda"},
- {"NAME": "/dev/sdb", "KNAME": "/dev/sdb", "PKNAME": ""},
- {"NAME": "/dev/sdb2", "KNAME": "/dev/sdb2", "PKNAME": "/dev/sdb"},
- {"NAME": "/dev/sdb3", "KNAME": "/dev/sdb3", "PKNAME": "/dev/sdb"},
- {"NAME": "/dev/sdc", "KNAME": "/dev/sdc", "PKNAME": ""},
- {"NAME": "/dev/sdd", "KNAME": "/dev/sdd", "PKNAME": ""},
- {"NAME": "/dev/mapper/ceph--osd--block--1", "KNAME": "/dev/mapper/ceph--osd--block--1", "PKNAME": "/dev/sdd"},
- {"NAME": "/dev/mapper/ceph--osd--block--2", "KNAME": "/dev/mapper/ceph--osd--block--2", "PKNAME": "/dev/sdd"},
+ {"NAME": "/dev/sda", "KNAME": "/dev/sda", "PKNAME": "", "TYPE": "disk"},
+ {"NAME": "/dev/sda1", "KNAME": "/dev/sda1", "PKNAME": "/dev/sda", "TYPE": "part"},
+ {"NAME": "/dev/sda2", "KNAME": "/dev/sda2", "PKNAME": "/dev/sda", "TYPE": "part"},
+ {"NAME": "/dev/sda3", "KNAME": "/dev/sda3", "PKNAME": "/dev/sda", "TYPE": "part"},
+ {"NAME": "/dev/sdb", "KNAME": "/dev/sdb", "PKNAME": "", "TYPE": "disk"},
+ {"NAME": "/dev/sdb2", "KNAME": "/dev/sdb2", "PKNAME": "/dev/sdb", "TYPE": "part"},
+ {"NAME": "/dev/sdb3", "KNAME": "/dev/sdb3", "PKNAME": "/dev/sdb", "TYPE": "part"},
+ {"NAME": "/dev/sdc", "KNAME": "/dev/sdc", "PKNAME": "", "TYPE": "disk"},
+ {"NAME": "/dev/sdd", "KNAME": "/dev/sdd", "PKNAME": "", "TYPE": "disk"},
+ {"NAME": "/dev/sde", "KNAME": "/dev/sde", "PKNAME": "", "TYPE": "disk"},
+ {"NAME": "/dev/sde1", "KNAME": "/dev/sde1", "PKNAME": "/dev/sde", "TYPE": "part"},
+ {"NAME": "/dev/mapper/ceph--osd--block--1", "KNAME": "/dev/mapper/ceph--osd--block--1", "PKNAME": "/dev/sdd", "TYPE": "lvm"},
+ {"NAME": "/dev/mapper/ceph--osd--block--2", "KNAME": "/dev/mapper/ceph--osd--block--2", "PKNAME": "/dev/sdd", "TYPE": "lvm"},
]
# dummy lsblk output for device with optional parent output
}
}'''
+def _bluestore_tool_label_output_sde1():
+ return '''{
+ "/dev/sde1": {
+ "osd_uuid": "sde1-uuid",
+ "size": 214747316224,
+ "btime": "2023-07-26T13:20:19.509457+0000",
+ "description": "main",
+ "bfm_blocks": "268435456",
+ "bfm_blocks_per_key": "128",
+ "bfm_bytes_per_block": "4096",
+ "bfm_size": "214747316224",
+ "bluefs": "1",
+ "ceph_fsid": "sde1-fsid",
+ "kv_backend": "rocksdb",
+ "magic": "ceph osd volume v026",
+ "mkfs_done": "yes",
+ "osd_key": "AQCSHcFkUeLIMBAAjKqANkXafjvVISkXt6FGCA==",
+ "ready": "ready",
+ "require_osd_release": "16",
+ "whoami": "1"
+ }
+}'''
+
def _bluestore_tool_label_output_dm_okay():
return '''{
"/dev/mapper/ceph--osd--block--1": {
return _lsblk_output(dev, parent="/dev/sdb"), '', 0
if dev == "/dev/sda" or dev == "/dev/sdb" or dev == "/dev/sdc" or dev == "/dev/sdd":
return _lsblk_output(dev), '', 0
+ if dev == "/dev/sde1":
+ return _lsblk_output(dev, parent="/dev/sde"), '', 0
if "mapper" in dev:
return _lsblk_output(dev, parent="/dev/sdd"), '', 0
pytest.fail('dev {} needs behavior specified for it'.format(dev))
if "/dev/sdb2" in command:
# sdb2 is a phantom atari partition that appears to have some valid bluestore info
return _bluestore_tool_label_output_sdb2(), '', 0
+ if "/dev/sde1" in command:
+ return _bluestore_tool_label_output_sde1(), '', 0
if "/dev/mapper/ceph--osd--block--1" in command:
# dm device 1 is a valid bluestore OSD (the other is corrupted/invalid)
return _bluestore_tool_label_output_dm_okay(), '', 0
return False # empty disk
if disk_path == "/dev/sdd":
return False # has LVM subdevices
+ if disk_path == "/dev/sde":
+ return False # has partitions, it means it shouldn't be an OSD
+ if disk_path == "/dev/sde1":
+ return True # is a valid OSD
if disk_path == "/dev/mapper/ceph--osd--block--1":
return True # good OSD
if disk_path == "/dev/mapper/ceph--osd--block--2":
assert sdb['device'] == '/dev/sdb'
assert sdb['ceph_fsid'] == 'sdb-fsid'
assert sdb['type'] == 'bluestore'
-
lvm1 = result['lvm-1-uuid']
assert lvm1['osd_uuid'] == 'lvm-1-uuid'
assert lvm1['osd_id'] == 2
assert lvm1['device'] == '/dev/mapper/ceph--osd--block--1'
assert lvm1['ceph_fsid'] == 'lvm-1-fsid'
assert lvm1['type'] == 'bluestore'
+ sde1 = result['sde1-uuid']
+ assert sde1['osd_uuid'] == 'sde1-uuid'
+ assert sde1['osd_id'] == 1
+ assert sde1['device'] == '/dev/sde1'
+ assert sde1['ceph_fsid'] == 'sde1-fsid'
+ assert sde1['type'] == 'bluestore'
@patch('ceph_volume.util.device.disk.get_devices')
@patch('ceph_volume.util.disk.has_bluestore_label')
patched_get_devices.side_effect = _devices_side_effect
result = raw.list.List([]).generate()
- assert len(result) == 3
+ assert len(result) == 2
assert 'sdb-uuid' in result
import pytest
from ceph_volume.devices import raw
-from mock.mock import patch
+from mock.mock import patch, MagicMock
from ceph_volume import objectstore
class TestRaw(object):
assert 'Path to bluestore block.wal block device' in stdout
assert 'Enable device encryption via dm-crypt' in stdout
+ @patch('ceph_volume.util.arg_validators.set_dmcrypt_no_workqueue', return_value=MagicMock())
@patch('ceph_volume.util.arg_validators.ValidRawDevice.__call__')
- def test_prepare_dmcrypt_no_secret_passed(self, m_valid_device, m_create_key, capsys):
+ def test_prepare_dmcrypt_no_secret_passed(self,
+ m_valid_device,
+ m_set_dmcrypt_no_workqueue,
+ m_create_key,
+ capsys):
m_valid_device.return_value = '/dev/foo'
with pytest.raises(SystemExit):
raw.prepare.Prepare(argv=['--bluestore', '--data', '/dev/foo', '--dmcrypt']).main()
import os
import pytest
from ceph_volume.devices.simple import activate
+from mock.mock import patch
class TestActivate(object):
- def test_no_data_uuid(self, factory, is_root, monkeypatch, capture, fake_filesystem):
+ @patch('ceph_volume.decorators.os.getuid', return_value=0)
+ def test_no_data_uuid(self, m_getuid, factory, capture, fake_filesystem):
fake_filesystem.create_file('/tmp/json-config', contents='{}')
args = factory(osd_id='0', osd_fsid='1234', json_config='/tmp/json-config')
with pytest.raises(RuntimeError):
stdout, stderr = capsys.readouterr()
assert 'Activate OSDs by mounting devices previously configured' in stdout
- def test_activate_all(self, is_root, monkeypatch):
+ def test_activate_all(self, monkeypatch):
'''
make sure Activate calls activate for each file returned by glob
'''
@patch('ceph_volume.objectstore.baseobjectstore.BaseObjectStore.get_ptuuid', Mock(return_value='c6798f59-01'))
@patch('ceph_volume.api.lvm.Volume.set_tags', MagicMock())
@patch('ceph_volume.api.lvm.get_single_lv')
- def test_prepare(self, m_get_single_lv, factory):
+ def test_prepare(self, m_get_single_lv, is_root, factory):
m_get_single_lv.return_value = Volume(lv_name='lv_foo',
lv_path='/fake-path',
vg_name='vg_foo',
@patch('ceph_volume.systemd.systemctl.osd_is_active', return_value=False)
def test_activate_all(self,
+ m_create_key,
mock_lvm_direct_report,
+ is_root,
factory,
fake_run):
args = factory(no_systemd=True)
@patch('ceph_volume.systemd.systemctl.osd_is_active', return_value=False)
def test_activate_all_no_osd_found(self,
+ m_create_key,
+ is_root,
factory,
fake_run,
monkeypatch,
@patch('ceph_volume.systemd.systemctl.osd_is_active', return_value=True)
def test_activate_all_osd_is_active(self,
mock_lvm_direct_report,
+ is_root,
factory,
fake_run):
args = factory(no_systemd=False)
@patch('ceph_volume.api.lvm.get_lvs')
def test_activate_osd_id_and_fsid(self,
m_get_lvs,
+ is_root,
factory):
args = factory(osd_id='1',
osd_fsid='824f7edf',
@patch('ceph_volume.api.lvm.get_lvs')
def test_activate_not_osd_id_and_fsid(self,
m_get_lvs,
+ is_root,
factory):
args = factory(no_systemd=True,
osd_id=None,
assert m_get_lvs.mock_calls == [call(tags={'ceph.osd_fsid': '824f7edf'})]
def test_activate_osd_id_and_not_fsid(self,
+ is_root,
factory):
args = factory(no_systemd=True,
osd_id='1',
assert str(error.value) == 'could not activate osd.1, please provide the osd_fsid too'
def test_activate_not_osd_id_and_not_fsid(self,
+ is_root,
factory):
args = factory(no_systemd=True,
osd_id=None,
@patch('ceph_volume.api.lvm.get_lvs')
def test_activate_couldnt_find_osd(self,
m_get_lvs,
+ is_root,
factory):
args = factory(osd_id='1',
osd_fsid='824f7edf',
@patch.dict('os.environ', {'CEPH_VOLUME_DMCRYPT_SECRET': 'dmcrypt-key'})
@patch('ceph_volume.objectstore.rawbluestore.prepare_utils.create_id')
@patch('ceph_volume.objectstore.rawbluestore.system.generate_uuid')
- def test_prepare(self, m_generate_uuid, m_create_id, factory):
+ def test_prepare(self, m_generate_uuid, m_create_id, is_root, factory):
m_generate_uuid.return_value = 'fake-uuid'
m_create_id.return_value = MagicMock()
self.raw_bs.prepare_dmcrypt = MagicMock()
assert m_create_osd_path.mock_calls == [call('1', tmpfs=True)]
def test_activate_raises_exception(self,
+ is_root,
mock_raw_direct_report):
with pytest.raises(RuntimeError) as error:
self.raw_bs.activate([],
assert str(error.value) == 'did not find any matching OSD to activate'
def test_activate_osd_id(self,
+ is_root,
mock_raw_direct_report):
self.raw_bs._activate = MagicMock()
self.raw_bs.activate([],
tmpfs=True)]
def test_activate_osd_fsid(self,
+ is_root,
mock_raw_direct_report):
self.raw_bs._activate = MagicMock()
with pytest.raises(RuntimeError):
def setup_method(self):
conf.log_path = '/tmp/'
- def test_no_arguments_parsing_error(self):
+ def test_no_arguments_parsing_error(self, fake_filesystem):
with pytest.raises(RuntimeError):
main(args=[])
- def test_parsing_suffix_error(self):
+ def test_parsing_suffix_error(self, fake_filesystem):
with pytest.raises(exceptions.SuffixParsingError):
main(args=['asdf'])
- def test_correct_command(self, monkeypatch):
+ def test_correct_command(self, monkeypatch, fake_filesystem):
run = Capture()
monkeypatch.setattr(process, 'run', run)
main(args=['ceph-volume-systemd', 'lvm-8715BEB4-15C5-49DE-BA6F-401086EC7B41-0' ])
assert '--cluster' in stdout
assert '--log-path' in stdout
- def test_log_ignoring_missing_ceph_conf(self, caplog):
+ def test_log_ignoring_missing_ceph_conf(self, caplog, fake_filesystem):
with pytest.raises(SystemExit) as error:
main.Volume(argv=['ceph-volume', '--cluster', 'barnacle', 'lvm', '--help'])
# make sure we aren't causing an actual error
assert log.message == 'ignoring inability to load ceph.conf'
assert log.levelname == 'WARNING'
- def test_logs_current_command(self, caplog):
+ def test_logs_current_command(self, caplog, fake_filesystem):
with pytest.raises(SystemExit) as error:
main.Volume(argv=['ceph-volume', '--cluster', 'barnacle', 'lvm', '--help'])
# make sure we aren't causing an actual error
assert log.message == 'Running command: ceph-volume --cluster barnacle lvm --help'
assert log.levelname == 'INFO'
- def test_logs_set_level_warning(self, caplog):
+ def test_logs_set_level_warning(self, caplog, fake_filesystem):
with pytest.raises(SystemExit) as error:
main.Volume(argv=['ceph-volume', '--log-level', 'warning', '--cluster', 'barnacle', 'lvm', '--help'])
# make sure we aren't causing an actual error
writer.seek(0)
val = buffer.getvalue()
assert self.octpus_and_squid_en.encode(encoding) in val
-
- def test_writer_uses_log_on_unicodeerror(self, stream, monkeypatch, capture):
-
- if sys.version_info > (3,):
- pytest.skip("Something breaks inside of pytest's capsys")
- monkeypatch.setattr(terminal.terminal_logger, 'info', capture)
- buffer = io.BytesIO()
- writer = stream(buffer, 'ascii')
- terminal._Write(_writer=writer).raw(self.message)
- assert self.octpus_and_squid_en in capture.calls[0]['args'][0]
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
self.validator = arg_validators.OSDPath()
def test_is_not_root(self, monkeypatch):
- monkeypatch.setattr(os, 'getuid', lambda: 100)
+ monkeypatch.setattr('ceph_volume.decorators.os.getuid', lambda : 100)
with pytest.raises(exceptions.SuperUserError):
self.validator('')
- def test_path_is_not_a_directory(self, is_root, monkeypatch, fake_filesystem):
+ def test_path_is_not_a_directory(self, monkeypatch, fake_filesystem):
fake_file = fake_filesystem.create_file('/tmp/foo')
+ monkeypatch.setattr('ceph_volume.decorators.os.getuid', lambda : 0)
monkeypatch.setattr(arg_validators.disk, 'is_partition', lambda x: False)
- validator = arg_validators.OSDPath()
with pytest.raises(argparse.ArgumentError):
- validator(fake_file.path)
+ self.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()
+ @patch('ceph_volume.decorators.os.getuid', return_value=0)
+ @patch('ceph_volume.util.arg_validators.disk.is_partition', return_value=False)
+ def test_files_are_missing(self, m_is_partition, m_getuid, fake_filesystem):
with pytest.raises(argparse.ArgumentError) as error:
- validator(tmppath)
+ self.validator('/tmp/osdpath')
assert 'Required file (ceph_fsid) was not found in OSD' in str(error.value)
result = disk.get_devices(_sys_block_path=str(tmpdir))
assert result == {}
- def test_sda_block_is_found(self, patched_get_block_devs_sysfs, fake_filesystem):
+ @patch('ceph_volume.util.disk.udevadm_property')
+ def test_sda_block_is_found(self, m_udev_adm_property, patched_get_block_devs_sysfs, fake_filesystem):
sda_path = '/dev/sda'
- patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']]
+ patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk', sda_path]]
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'] == {}
- def test_sda_size(self, patched_get_block_devs_sysfs, fake_filesystem):
+ @patch('ceph_volume.util.disk.udevadm_property')
+ def test_sda_size(self, m_udev_adm_property, patched_get_block_devs_sysfs, fake_filesystem):
sda_path = '/dev/sda'
- patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']]
+ patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk', sda_path]]
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'
- def test_sda_sectorsize_fallsback(self, patched_get_block_devs_sysfs, fake_filesystem):
+ @patch('ceph_volume.util.disk.udevadm_property')
+ def test_sda_sectorsize_fallsback(self, m_udev_adm_property, 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']]
+ patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk', sda_path]]
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'
- def test_sda_sectorsize_from_logical_block(self, patched_get_block_devs_sysfs, fake_filesystem):
+ @patch('ceph_volume.util.disk.udevadm_property')
+ def test_sda_sectorsize_from_logical_block(self, m_udev_adm_property, patched_get_block_devs_sysfs, fake_filesystem):
sda_path = '/dev/sda'
- patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']]
+ patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk', sda_path]]
fake_filesystem.create_file('/sys/block/sda/queue/logical_block_size', contents = '99')
result = disk.get_devices()
assert result[sda_path]['sectorsize'] == '99'
- def test_sda_sectorsize_does_not_fallback(self, patched_get_block_devs_sysfs, fake_filesystem):
+ @patch('ceph_volume.util.disk.udevadm_property')
+ def test_sda_sectorsize_does_not_fallback(self, m_udev_adm_property, patched_get_block_devs_sysfs, fake_filesystem):
sda_path = '/dev/sda'
- patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']]
+ patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk', sda_path]]
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'
- def test_is_rotational(self, patched_get_block_devs_sysfs, fake_filesystem):
+ @patch('ceph_volume.util.disk.udevadm_property')
+ def test_is_rotational(self, m_udev_adm_property, patched_get_block_devs_sysfs, fake_filesystem):
sda_path = '/dev/sda'
- patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']]
+ patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk', sda_path]]
fake_filesystem.create_file('/sys/block/sda/queue/rotational', contents = '1')
result = disk.get_devices()
assert result[sda_path]['rotational'] == '1'
- def test_is_ceph_rbd(self, patched_get_block_devs_sysfs, fake_filesystem):
+ @patch('ceph_volume.util.disk.udevadm_property')
+ def test_is_ceph_rbd(self, m_udev_adm_property, patched_get_block_devs_sysfs, fake_filesystem):
rbd_path = '/dev/rbd0'
- patched_get_block_devs_sysfs.return_value = [[rbd_path, rbd_path, 'disk']]
+ patched_get_block_devs_sysfs.return_value = [[rbd_path, rbd_path, 'disk', rbd_path]]
result = disk.get_devices()
assert rbd_path not in result
- def test_actuator_device(self, patched_get_block_devs_sysfs, fake_filesystem):
+ @patch('ceph_volume.util.disk.udevadm_property')
+ def test_actuator_device(self, m_udev_adm_property, patched_get_block_devs_sysfs, fake_filesystem):
sda_path = '/dev/sda'
fake_actuator_nb = 2
- patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']]
+ patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk', sda_path]]
for actuator in range(0, fake_actuator_nb):
fake_filesystem.create_dir(f'/sys/block/sda/queue/independent_access_ranges/{actuator}')
result = disk.get_devices()
class TestAllowLoopDevsWarning(object):
- def setup(self):
+ def setup_method(self):
disk.AllowLoopDevices.allow = False
disk.AllowLoopDevices.warned = False
if os.environ.get('CEPH_VOLUME_ALLOW_LOOP_DEVICES'):
class TestLuksOpen(object):
+ @patch('ceph_volume.util.encryption.bypass_workqueue', return_value=False)
@patch('ceph_volume.util.encryption.process.call')
- def test_luks_open_command_with_default_size(self, m_call, conf_ceph_stub):
+ def test_luks_open_command_with_default_size(self, m_call, m_bypass_workqueue, conf_ceph_stub):
conf_ceph_stub('[global]\nfsid=abcd')
expected = [
'cryptsetup',
encryption.luks_open('abcd', '/dev/foo', '/dev/bar')
assert m_call.call_args[0][0] == expected
+ @patch('ceph_volume.util.encryption.bypass_workqueue', return_value=False)
@patch('ceph_volume.util.encryption.process.call')
- def test_luks_open_command_with_custom_size(self, m_call, conf_ceph_stub):
+ def test_luks_open_command_with_custom_size(self, m_call, m_bypass_workqueue, conf_ceph_stub):
conf_ceph_stub('[global]\nfsid=abcd\n[osd]\nosd_dmcrypt_key_size=256')
expected = [
'cryptsetup',
class TestOsdMkfsBluestore(object):
- def setup(self):
+ def setup_method(self):
conf.cluster = 'ceph'
def test_keyring_is_added(self, fake_call, monkeypatch):
projects / ceph-ci.git / commitdiff