]> git.apps.os.sepia.ceph.com Git - ceph.git/commitdiff
ceph-volume/tests: migrate to pyfakefs 47170/head
authorGuillaume Abrioux <gabrioux@redhat.com>
Wed, 27 Jul 2022 11:41:41 +0000 (13:41 +0200)
committerGuillaume Abrioux <gabrioux@redhat.com>
Thu, 28 Jul 2022 18:14:06 +0000 (20:14 +0200)
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 <gabrioux@redhat.com>
src/ceph-volume/ceph_volume/tests/conftest.py
src/ceph-volume/ceph_volume/tests/devices/simple/test_activate.py
src/ceph-volume/ceph_volume/tests/devices/simple/test_scan.py
src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py
src/ceph-volume/ceph_volume/tests/util/test_disk.py
src/ceph-volume/ceph_volume/tests/util/test_encryption.py
src/ceph-volume/ceph_volume/tests/util/test_system.py
src/ceph-volume/tox.ini

index 4a832873fa7b189551f8976b15e574fab375c019..f060f78d44c57b5b645d020de4037337defcccfa 100644 (file)
@@ -311,3 +311,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
index ac2dd0e7b3c312a10c8a154679b135effda33041..5c7bd3117920ac4e19c6e37b2a799d13449f620e 100644 (file)
@@ -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')
index 118493625f37a7f69377474d12060cff05a55a4b..b5d12065578534159763c51ad3bf5a2658efe843 100644 (file)
@@ -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
 
index 45c4ed50ec3700571216c5d5d7395c6085c309ab..b0446a13b4685a6780cbb23a661b151d664a9d87 100644 (file)
@@ -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)
index ab652c22e6ddf5618d60ceaa580e692552cf91a1..299463504c6a6b40669bf30c99e6a4a92d18aa99 100644 (file)
@@ -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
 
 
index fc0991cf7b7406f6bd266728f0ac85b5f96a67c4..cd2ea8f187fc236435a73554d05da365c15209c4 100644 (file)
@@ -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'
index e7a124b8dd6ae340c0ffb2656b462061c1d60a97..763b18768b44b11209820b575abe0dc1b8d19ca2 100644 (file)
@@ -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 == ''
 
 
index c58951a9b00c384bf4010e1512850bb38bedb2f8..820cf6fb3475f624919b0cdcbf4f9f7916804113 100644 (file)
@@ -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