From 2817f5df20b184621646fec4a2135f49a81e5741 Mon Sep 17 00:00:00 2001 From: Guillaume Abrioux Date: Wed, 27 Jul 2022 13:41:41 +0200 Subject: [PATCH] ceph-volume/tests: migrate to pyfakefs ceph-volume unit tests shouldn't actually create contents on the filesystem from where it runs (even though they are written in a tmp dir), let's use pyfakefs. Signed-off-by: Guillaume Abrioux (cherry picked from commit 1274ba34b4d21c5abf2cd2da035d6d25ae8d03da) --- src/ceph-volume/ceph_volume/tests/conftest.py | 8 ++ .../tests/devices/simple/test_activate.py | 48 +++++------ .../tests/devices/simple/test_scan.py | 27 +++--- .../tests/util/test_arg_validators.py | 5 +- .../ceph_volume/tests/util/test_disk.py | 82 +++++++------------ .../ceph_volume/tests/util/test_encryption.py | 6 +- .../ceph_volume/tests/util/test_system.py | 32 ++++---- src/ceph-volume/tox.ini | 1 + 8 files changed, 98 insertions(+), 111 deletions(-) diff --git a/src/ceph-volume/ceph_volume/tests/conftest.py b/src/ceph-volume/ceph_volume/tests/conftest.py index 8393b2e5d1b74..4a37490fff16c 100644 --- a/src/ceph-volume/ceph_volume/tests/conftest.py +++ b/src/ceph-volume/ceph_volume/tests/conftest.py @@ -303,3 +303,11 @@ def device_info(monkeypatch, patch_bluestore_label): @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 diff --git a/src/ceph-volume/ceph_volume/tests/devices/simple/test_activate.py b/src/ceph-volume/ceph_volume/tests/devices/simple/test_activate.py index ac2dd0e7b3c31..5c7bd3117920a 100644 --- a/src/ceph-volume/ceph_volume/tests/devices/simple/test_activate.py +++ b/src/ceph-volume/ceph_volume/tests/devices/simple/test_activate.py @@ -5,9 +5,9 @@ from ceph_volume.devices.simple import activate 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) @@ -45,9 +45,9 @@ class TestActivate(object): 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') @@ -56,69 +56,69 @@ class TestEnableSystemdUnits(object): 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') diff --git a/src/ceph-volume/ceph_volume/tests/devices/simple/test_scan.py b/src/ceph-volume/ceph_volume/tests/devices/simple/test_scan.py index 118493625f37a..b5d1206557853 100644 --- a/src/ceph-volume/ceph_volume/tests/devices/simple/test_scan.py +++ b/src/ceph-volume/ceph_volume/tests/devices/simple/test_scan.py @@ -5,20 +5,23 @@ from ceph_volume.devices.simple import scan 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): @@ -36,10 +39,10 @@ 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 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 index 45c4ed50ec370..b0446a13b4685 100644 --- a/src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py +++ b/src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py @@ -16,11 +16,12 @@ class TestOSDPath(object): 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) diff --git a/src/ceph-volume/ceph_volume/tests/util/test_disk.py b/src/ceph-volume/ceph_volume/tests/util/test_disk.py index ab652c22e6ddf..299463504c6a6 100644 --- a/src/ceph-volume/ceph_volume/tests/util/test_disk.py +++ b/src/ceph-volume/ceph_volume/tests/util/test_disk.py @@ -1,6 +1,7 @@ import os import pytest from ceph_volume.util import disk +from mock.mock import patch class TestLsblkParser(object): @@ -219,97 +220,70 @@ class TestSizeParse(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 diff --git a/src/ceph-volume/ceph_volume/tests/util/test_encryption.py b/src/ceph-volume/ceph_volume/tests/util/test_encryption.py index fc0991cf7b740..cd2ea8f187fc2 100644 --- a/src/ceph-volume/ceph_volume/tests/util/test_encryption.py +++ b/src/ceph-volume/ceph_volume/tests/util/test_encryption.py @@ -48,9 +48,9 @@ class TestStatus(object): 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' diff --git a/src/ceph-volume/ceph_volume/tests/util/test_system.py b/src/ceph-volume/ceph_volume/tests/util/test_system.py index e7a124b8dd6ae..763b18768b44b 100644 --- a/src/ceph-volume/ceph_volume/tests/util/test_system.py +++ b/src/ceph-volume/ceph_volume/tests/util/test_system.py @@ -182,13 +182,13 @@ class TestGetMounts(object): 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): @@ -197,21 +197,21 @@ 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 == '' diff --git a/src/ceph-volume/tox.ini b/src/ceph-volume/tox.ini index c58951a9b00c3..820cf6fb3475f 100644 --- a/src/ceph-volume/tox.ini +++ b/src/ceph-volume/tox.ini @@ -7,6 +7,7 @@ deps= 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 -- 2.39.5