@pytest.fixture(params=[0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 0.95, 0.999, 1.0])
def data_allocate_fraction(request):
return request.param
+
+@pytest.fixture
+def fake_filesystem(fs):
+
+ fs.create_dir('/sys/block/sda/slaves')
+ fs.create_dir('/sys/block/sda/queue')
+ fs.create_dir('/sys/block/rbd0')
+ yield fs
class TestActivate(object):
- def test_no_data_uuid(self, factory, tmpfile, is_root, monkeypatch, capture):
- json_config = tmpfile(contents='{}')
- args = factory(osd_id='0', osd_fsid='1234', json_config=json_config)
+ def test_no_data_uuid(self, factory, is_root, monkeypatch, 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):
activate.Activate([]).activate(args)
class TestEnableSystemdUnits(object):
- def test_nothing_is_activated(self, tmpfile, is_root, capsys):
- json_config = tmpfile(contents='{}')
- activation = activate.Activate(['--no-systemd', '--file', json_config, '0', '1234'], from_trigger=True)
+ def test_nothing_is_activated(self, is_root, capsys, fake_filesystem):
+ fake_filesystem.create_file('/tmp/json-config', contents='{}')
+ activation = activate.Activate(['--no-systemd', '--file', '/tmp/json-config', '0', '1234'], from_trigger=True)
activation.activate = lambda x: True
activation.main()
activation.enable_systemd_units('0', '1234')
assert 'Skipping masking of ceph-disk' in stderr
assert 'Skipping enabling and starting OSD simple' in stderr
- def test_no_systemd_flag_is_true(self, tmpfile, is_root):
- json_config = tmpfile(contents='{}')
- activation = activate.Activate(['--no-systemd', '--file', json_config, '0', '1234'], from_trigger=True)
+ def test_no_systemd_flag_is_true(self, is_root, fake_filesystem):
+ fake_filesystem.create_file('/tmp/json-config', contents='{}')
+ activation = activate.Activate(['--no-systemd', '--file', '/tmp/json-config', '0', '1234'], from_trigger=True)
activation.activate = lambda x: True
activation.main()
assert activation.skip_systemd is True
- def test_no_systemd_flag_is_false(self, tmpfile, is_root):
- json_config = tmpfile(contents='{}')
- activation = activate.Activate(['--file', json_config, '0', '1234'], from_trigger=True)
+ def test_no_systemd_flag_is_false(self, is_root, fake_filesystem):
+ fake_filesystem.create_file('/tmp/json-config', contents='{}')
+ activation = activate.Activate(['--file', '/tmp/json-config', '0', '1234'], from_trigger=True)
activation.activate = lambda x: True
activation.main()
assert activation.skip_systemd is False
- def test_masks_ceph_disk(self, tmpfile, is_root, monkeypatch, capture):
+ def test_masks_ceph_disk(self, is_root, monkeypatch, capture, fake_filesystem):
monkeypatch.setattr('ceph_volume.systemd.systemctl.mask_ceph_disk', capture)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_volume', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_osd', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.start_osd', lambda *a: True)
- json_config = tmpfile(contents='{}')
- activation = activate.Activate(['--file', json_config, '0', '1234'], from_trigger=False)
+ fake_filesystem.create_file('/tmp/json-config', contents='{}')
+ activation = activate.Activate(['--file', '/tmp/json-config', '0', '1234'], from_trigger=False)
activation.activate = lambda x: True
activation.main()
activation.enable_systemd_units('0', '1234')
assert len(capture.calls) == 1
- def test_enables_simple_unit(self, tmpfile, is_root, monkeypatch, capture):
+ def test_enables_simple_unit(self, is_root, monkeypatch, capture, fake_filesystem):
monkeypatch.setattr('ceph_volume.systemd.systemctl.mask_ceph_disk', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_volume', capture)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_osd', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.start_osd', lambda *a: True)
- json_config = tmpfile(contents='{}')
- activation = activate.Activate(['--file', json_config, '0', '1234'], from_trigger=False)
+ fake_filesystem.create_file('/tmp/json-config', contents='{}')
+ activation = activate.Activate(['--file', '/tmp/json-config', '0', '1234'], from_trigger=False)
activation.activate = lambda x: True
activation.main()
activation.enable_systemd_units('0', '1234')
assert len(capture.calls) == 1
assert capture.calls[0]['args'] == ('0', '1234', 'simple')
- def test_enables_osd_unit(self, tmpfile, is_root, monkeypatch, capture):
+ def test_enables_osd_unit(self, is_root, monkeypatch, capture, fake_filesystem):
monkeypatch.setattr('ceph_volume.systemd.systemctl.mask_ceph_disk', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_volume', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_osd', capture)
monkeypatch.setattr('ceph_volume.systemd.systemctl.start_osd', lambda *a: True)
- json_config = tmpfile(contents='{}')
- activation = activate.Activate(['--file', json_config, '0', '1234'], from_trigger=False)
+ fake_filesystem.create_file('/tmp/json-config', contents='{}')
+ activation = activate.Activate(['--file', '/tmp/json-config', '0', '1234'], from_trigger=False)
activation.activate = lambda x: True
activation.main()
activation.enable_systemd_units('0', '1234')
assert len(capture.calls) == 1
assert capture.calls[0]['args'] == ('0',)
- def test_starts_osd_unit(self, tmpfile, is_root, monkeypatch, capture):
+ def test_starts_osd_unit(self, is_root, monkeypatch, capture, fake_filesystem):
monkeypatch.setattr('ceph_volume.systemd.systemctl.mask_ceph_disk', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_volume', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_osd', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.start_osd', capture)
- json_config = tmpfile(contents='{}')
- activation = activate.Activate(['--file', json_config, '0', '1234'], from_trigger=False)
+ fake_filesystem.create_file('/tmp/json-config', contents='{}')
+ activation = activate.Activate(['--file', '/tmp/json-config', '0', '1234'], from_trigger=False)
activation.activate = lambda x: True
activation.main()
activation.enable_systemd_units('0', '1234')
class TestGetContents(object):
- def test_multiple_lines_are_left_as_is(self, tmpfile):
- magic_file = tmpfile(contents='first\nsecond\n')
+ def setup(self):
+ self.magic_file_name = '/tmp/magic-file'
+
+ def test_multiple_lines_are_left_as_is(self, fake_filesystem):
+ magic_file = fake_filesystem.create_file(self.magic_file_name, contents='first\nsecond\n')
scanner = scan.Scan([])
- assert scanner.get_contents(magic_file) == 'first\nsecond\n'
+ assert scanner.get_contents(magic_file.path) == 'first\nsecond\n'
- def test_extra_whitespace_gets_removed(self, tmpfile):
- magic_file = tmpfile(contents='first ')
+ def test_extra_whitespace_gets_removed(self, fake_filesystem):
+ magic_file = fake_filesystem.create_file(self.magic_file_name, contents='first ')
scanner = scan.Scan([])
- assert scanner.get_contents(magic_file) == 'first'
+ assert scanner.get_contents(magic_file.path) == 'first'
- def test_single_newline_values_are_trimmed(self, tmpfile):
- magic_file = tmpfile(contents='first\n')
+ def test_single_newline_values_are_trimmed(self, fake_filesystem):
+ magic_file = fake_filesystem.create_file(self.magic_file_name, contents='first\n')
scanner = scan.Scan([])
- assert scanner.get_contents(magic_file) == 'first'
+ assert scanner.get_contents(magic_file.path) == 'first'
class TestEtcPath(object):
assert scanner.etc_path == path
assert os.path.isdir(path)
- def test_complains_when_file(self, tmpfile):
- path = tmpfile()
+ def test_complains_when_file(self, fake_filesystem):
+ etc_dir = fake_filesystem.create_file('/etc/ceph/osd')
scanner = scan.Scan([])
- scanner._etc_path = path
+ scanner._etc_path = etc_dir.path
with pytest.raises(RuntimeError):
scanner.etc_path
with pytest.raises(exceptions.SuperUserError):
self.validator('')
- def test_path_is_not_a_directory(self, is_root, tmpfile, monkeypatch):
+ 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(tmpfile())
+ validator(fake_file.path)
def test_files_are_missing(self, is_root, tmpdir, monkeypatch):
tmppath = str(tmpdir)
import os
import pytest
from ceph_volume.util import disk
+from mock.mock import patch
class TestLsblkParser(object):
class TestGetDevices(object):
- def setup_path(self, tmpdir):
- path = os.path.join(str(tmpdir), 'block')
- os.makedirs(path)
- return path
-
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 == {}
- def test_sda_block_is_found(self, tmpdir, patched_get_block_devs_sysfs):
+ @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']]
- block_path = self.setup_path(tmpdir)
- os.makedirs(os.path.join(block_path, 'sda'))
- result = disk.get_devices(_sys_block_path=block_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, tmpfile, tmpdir, patched_get_block_devs_sysfs):
+ @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']]
- block_path = self.setup_path(tmpdir)
- block_sda_path = os.path.join(block_path, 'sda')
- os.makedirs(block_sda_path)
- tmpfile('size', '1024', directory=block_sda_path)
- result = disk.get_devices(_sys_block_path=block_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, tmpfile, tmpdir, patched_get_block_devs_sysfs):
+ @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']]
- block_path = self.setup_path(tmpdir)
- block_sda_path = os.path.join(block_path, 'sda')
- sda_queue_path = os.path.join(block_sda_path, 'queue')
- os.makedirs(block_sda_path)
- os.makedirs(sda_queue_path)
- tmpfile('hw_sector_size', contents='1024', directory=sda_queue_path)
- result = disk.get_devices(_sys_block_path=block_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, tmpfile, tmpdir, patched_get_block_devs_sysfs):
+ @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']]
- block_path = self.setup_path(tmpdir)
- block_sda_path = os.path.join(block_path, 'sda')
- sda_queue_path = os.path.join(block_sda_path, 'queue')
- os.makedirs(block_sda_path)
- os.makedirs(sda_queue_path)
- tmpfile('logical_block_size', contents='99', directory=sda_queue_path)
- result = disk.get_devices(_sys_block_path=block_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, tmpfile, tmpdir, patched_get_block_devs_sysfs):
+ @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']]
- block_path = self.setup_path(tmpdir)
- block_sda_path = os.path.join(block_path, 'sda')
- sda_queue_path = os.path.join(block_sda_path, 'queue')
- os.makedirs(block_sda_path)
- os.makedirs(sda_queue_path)
- tmpfile('logical_block_size', contents='99', directory=sda_queue_path)
- tmpfile('hw_sector_size', contents='1024', directory=sda_queue_path)
- result = disk.get_devices(_sys_block_path=block_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, tmpfile, tmpdir, patched_get_block_devs_sysfs):
+ @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']]
- block_path = self.setup_path(tmpdir)
- block_sda_path = os.path.join(block_path, 'sda')
- sda_queue_path = os.path.join(block_sda_path, 'queue')
- os.makedirs(block_sda_path)
- os.makedirs(sda_queue_path)
- tmpfile('rotational', contents='1', directory=sda_queue_path)
- result = disk.get_devices(_sys_block_path=block_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, tmpfile, tmpdir, patched_get_block_devs_sysfs):
+ @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']]
- block_path = self.setup_path(tmpdir)
- block_rbd_path = os.path.join(block_path, 'rbd0')
- os.makedirs(block_rbd_path)
- result = disk.get_devices(_sys_block_path=block_path)
+ result = disk.get_devices()
assert rbd_path not in result
class TestDmcryptClose(object):
- def test_mapper_exists(self, fake_run, tmpfile):
- file_name = tmpfile(name='mapper-device')
- encryption.dmcrypt_close(file_name)
+ 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'
class TestIsBinary(object):
- def test_is_binary(self, tmpfile):
- binary_path = tmpfile(contents='asd\n\nlkjh\x00')
- assert system.is_binary(binary_path)
+ 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, tmpfile):
- binary_path = tmpfile(contents='asd\n\nlkjh0')
- assert system.is_binary(binary_path) is False
+ 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):
filepath = os.path.join(str(tmpdir), 'doesnotexist')
assert system.get_file_contents(filepath, 'default') == 'default'
- def test_path_has_contents(self, tmpfile):
- interesting_file = tmpfile(contents="1")
- result = system.get_file_contents(interesting_file)
+ 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, tmpfile):
- interesting_file = tmpfile(contents="0\n1")
- result = system.get_file_contents(interesting_file)
+ 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, tmpfile):
- interesting_file = tmpfile(contents="0")
+ def test_exception_returns_default(self, fake_filesystem):
+ interesting_file = fake_filesystem.create_file('/tmp/fake-file', contents="0")
# remove read, causes IOError
- os.chmod(interesting_file, 0o000)
- result = system.get_file_contents(interesting_file)
+ os.chmod(interesting_file.path, 0o000)
+ result = system.get_file_contents(interesting_file.path)
assert result == ''
pytest
pytest-xdist
mock
+ pyfakefs
install_command=./tox_install_command.sh {opts} {packages}
commands=py.test --numprocesses=auto -vv {posargs:ceph_volume/tests} --ignore=ceph_volume/tests/functional