ceph/src/ceph-volume/ceph_volume/tests/util/test_system.py

   1 import os
   2 import pwd
   3 import getpass
   4 import pytest
   5 from textwrap import dedent
   6 from ceph_volume.util import system
   7 from mock.mock import patch
   8 from ceph_volume.tests.conftest import Factory
   9
  10
  11 @pytest.fixture
  12 def mock_find_executable_on_host(monkeypatch):
  13     """
  14     Monkeypatches util.system.find_executable_on_host, so that a caller can add behavior to the response
  15     """
  16     def apply(stdout=None, stderr=None, returncode=0):
  17         stdout_stream = Factory(read=lambda: stdout)
  18         stderr_stream = Factory(read=lambda: stderr)
  19         return_value = Factory(
  20             stdout=stdout_stream,
  21             stderr=stderr_stream,
  22             wait=lambda: returncode,
  23             communicate=lambda x: (stdout, stderr, returncode)
  24         )
  25
  26         monkeypatch.setattr(
  27             'ceph_volume.util.system.subprocess.Popen',
  28             lambda *a, **kw: return_value)
  29
  30     return apply
  31
  32 class TestMkdirP(object):
  33
  34     def test_existing_dir_does_not_raise_w_chown(self, monkeypatch, tmpdir):
  35         user = pwd.getpwnam(getpass.getuser())
  36         uid, gid = user[2], user[3]
  37         monkeypatch.setattr(system, 'get_ceph_user_ids', lambda: (uid, gid,))
  38         path = str(tmpdir)
  39         system.mkdir_p(path)
  40         assert os.path.isdir(path)
  41
  42     def test_new_dir_w_chown(self, monkeypatch, tmpdir):
  43         user = pwd.getpwnam(getpass.getuser())
  44         uid, gid = user[2], user[3]
  45         monkeypatch.setattr(system, 'get_ceph_user_ids', lambda: (uid, gid,))
  46         path = os.path.join(str(tmpdir), 'new')
  47         system.mkdir_p(path)
  48         assert os.path.isdir(path)
  49
  50     def test_existing_dir_does_not_raise_no_chown(self, tmpdir):
  51         path = str(tmpdir)
  52         system.mkdir_p(path, chown=False)
  53         assert os.path.isdir(path)
  54
  55     def test_new_dir_no_chown(self, tmpdir):
  56         path = os.path.join(str(tmpdir), 'new')
  57         system.mkdir_p(path, chown=False)
  58         assert os.path.isdir(path)
  59
  60
  61 @pytest.fixture
  62 def fake_proc(tmpdir, monkeypatch):
  63     PROCDIR = str(tmpdir)
  64     proc_path = os.path.join(PROCDIR, 'mounts')
  65     with open(proc_path, 'w') as f:
  66         f.write(dedent("""nfsd /proc/fs/nfsd nfsd rw,relatime 0 0
  67             rootfs / rootfs rw 0 0
  68             sysfs /sys sysfs rw,seclabel,nosuid,nodev,noexec,relatime 0 0
  69             proc /proc proc rw,nosuid,nodev,noexec,relatime 0 0
  70             devtmpfs /dev devtmpfs rw,seclabel,nosuid,size=238292k,nr_inodes=59573,mode=755 0 0
  71             securityfs /sys/kernel/security securityfs rw,nosuid,nodev,noexec,relatime 0 0
  72             tmpfs /dev/shm tmpfs rw,seclabel,nosuid,nodev 0 0
  73             devpts /dev/pts devpts rw,seclabel,nosuid,noexec,relatime,gid=5,mode=620,ptmxmode=000 0 0
  74             tmpfs /run tmpfs rw,seclabel,nosuid,nodev,mode=755 0 0
  75             tmpfs /sys/fs/cgroup tmpfs ro,seclabel,nosuid,nodev,noexec,mode=755 0 0
  76             cgroup /sys/fs/cgroup/systemd cgroup rw,nosuid,nodev,noexec,relatime,xattr,release_agent=/usr/lib/systemd/systemd-cgroups-agent,name=systemd 0 0
  77             cgroup /sys/fs/cgroup/freezer cgroup rw,nosuid,nodev,noexec,relatime,freezer 0 0
  78             configfs /sys/kernel/config configfs rw,relatime 0 0
  79             /dev/mapper/VolGroup00-LogVol00 / xfs rw,seclabel,relatime,attr2,inode64,noquota 0 0
  80             selinuxfs /sys/fs/selinux selinuxfs rw,relatime 0 0
  81             debugfs /sys/kernel/debug debugfs rw,relatime 0 0
  82             hugetlbfs /dev/hugepages hugetlbfs rw,seclabel,relatime 0 0
  83             mqueue /dev/mqueue mqueue rw,seclabel,relatime 0 0
  84             sunrpc /far/lib/nfs/rpc_pipefs rpc_pipefs rw,relatime 0 0
  85             /dev/sde4 /two/field/path
  86             nfsd /proc/fs/nfsd nfsd rw,relatime 0 0
  87             /dev/sde2 /boot xfs rw,seclabel,relatime,attr2,inode64,noquota 0 0
  88             tmpfs /far/lib/ceph/osd/ceph-5 tmpfs rw,seclabel,relatime 0 0
  89             tmpfs /far/lib/ceph/osd/ceph-7 tmpfs rw,seclabel,relatime 0 0
  90             /dev/sda1 /far/lib/ceph/osd/ceph-0 xfs rw,seclabel,noatime,attr2,inode64,noquota 0 0
  91             tmpfs /run/user/1000 tmpfs rw,seclabel,nosuid,nodev,relatime,size=50040k,mode=700,uid=1000,gid=1000 0 0
  92             /dev/sdc2 /boot xfs rw,seclabel,relatime,attr2,inode64,noquota 0 0
  93             tmpfs /run/user/1000 tmpfs rw,seclabel,mode=700,uid=1000,gid=1000 0 0"""))
  94     monkeypatch.setattr(system, 'PROCDIR', PROCDIR)
  95     monkeypatch.setattr(os.path, 'exists', lambda x: True)
  96
  97
  98 class TestPathIsMounted(object):
  99
 100     def test_is_mounted(self, fake_proc):
 101         assert system.path_is_mounted('/boot') is True
 102
 103     def test_is_not_mounted(self, fake_proc):
 104         assert system.path_is_mounted('/far/fib/feph') is False
 105
 106     def test_is_not_mounted_at_destination(self, fake_proc):
 107         assert system.path_is_mounted('/boot', destination='/dev/sda1') is False
 108
 109     def test_is_mounted_at_destination(self, fake_proc):
 110         assert system.path_is_mounted('/boot', destination='/dev/sdc2') is True
 111
 112
 113 class TestDeviceIsMounted(object):
 114
 115     def test_is_mounted(self, fake_proc):
 116         assert system.device_is_mounted('/dev/sda1') is True
 117
 118     def test_path_is_not_device(self, fake_proc):
 119         assert system.device_is_mounted('/far/lib/ceph/osd/ceph-7') is False
 120
 121     def test_is_not_mounted_at_destination(self, fake_proc):
 122         assert system.device_is_mounted('/dev/sda1', destination='/far/lib/ceph/osd/test-1') is False
 123
 124     def test_is_mounted_at_destination(self, fake_proc):
 125         assert system.device_is_mounted('/dev/sda1', destination='/far/lib/ceph/osd/ceph-7') is False
 126
 127     def test_is_realpath_dev_mounted_at_destination(self, fake_proc, monkeypatch):
 128         monkeypatch.setattr(system.os.path, 'realpath', lambda x: '/dev/sda1' if 'foo' in x else x)
 129         result = system.device_is_mounted('/dev/maper/foo', destination='/far/lib/ceph/osd/ceph-0')
 130         assert result is True
 131
 132     def test_is_realpath_path_mounted_at_destination(self, fake_proc, monkeypatch):
 133         monkeypatch.setattr(
 134             system.os.path, 'realpath',
 135             lambda x: '/far/lib/ceph/osd/ceph-0' if 'symlink' in x else x)
 136         result = system.device_is_mounted('/dev/sda1', destination='/symlink/lib/ceph/osd/ceph-0')
 137         assert result is True
 138
 139
 140 class TestGetMounts(object):
 141
 142     def test_not_mounted(self, tmpdir, monkeypatch):
 143         PROCDIR = str(tmpdir)
 144         proc_path = os.path.join(PROCDIR, 'mounts')
 145         with open(proc_path, 'w') as f:
 146             f.write('')
 147         monkeypatch.setattr(system, 'PROCDIR', PROCDIR)
 148         m = system.Mounts()
 149         assert m.get_mounts() == {}
 150
 151     def test_is_mounted_(self, fake_proc):
 152         m = system.Mounts()
 153         assert m.get_mounts()['/dev/sdc2'] == ['/boot']
 154
 155     def test_ignores_two_fields(self, fake_proc):
 156         m = system.Mounts()
 157         assert m.get_mounts().get('/dev/sde4') is None
 158
 159     def test_tmpfs_is_reported(self, fake_proc):
 160         m = system.Mounts()
 161         assert m.get_mounts()['tmpfs'][0] == '/dev/shm'
 162
 163     def test_non_skip_devs_arent_reported(self, fake_proc):
 164         m = system.Mounts()
 165         assert m.get_mounts().get('cgroup') is None
 166
 167     def test_multiple_mounts_are_appended(self, fake_proc):
 168         m = system.Mounts()
 169         assert len(m.get_mounts()['tmpfs']) == 7
 170
 171     def test_nonexistent_devices_are_skipped(self, tmpdir, monkeypatch):
 172         PROCDIR = str(tmpdir)
 173         proc_path = os.path.join(PROCDIR, 'mounts')
 174         with open(proc_path, 'w') as f:
 175             f.write(dedent("""nfsd /proc/fs/nfsd nfsd rw,relatime 0 0
 176                     /dev/sda1 /far/lib/ceph/osd/ceph-0 xfs rw,attr2,inode64,noquota 0 0
 177                     /dev/sda2 /far/lib/ceph/osd/ceph-1 xfs rw,attr2,inode64,noquota 0 0"""))
 178         monkeypatch.setattr(system, 'PROCDIR', PROCDIR)
 179         monkeypatch.setattr(os.path, 'exists', lambda x: False if x == '/dev/sda1' else True)
 180         m = system.Mounts()
 181         assert m.get_mounts().get('/dev/sda1') is None
 182
 183
 184 class TestIsBinary(object):
 185
 186     def test_is_binary(self, fake_filesystem):
 187         binary_path = fake_filesystem.create_file('/tmp/fake-file', contents='asd\n\nlkjh\x00')
 188         assert system.is_binary(binary_path.path)
 189
 190     def test_is_not_binary(self, fake_filesystem):
 191         binary_path = fake_filesystem.create_file('/tmp/fake-file', contents='asd\n\nlkjh0')
 192         assert system.is_binary(binary_path.path) is False
 193
 194
 195 class TestGetFileContents(object):
 196
 197     def test_path_does_not_exist(self, tmpdir):
 198         filepath = os.path.join(str(tmpdir), 'doesnotexist')
 199         assert system.get_file_contents(filepath, 'default') == 'default'
 200
 201     def test_path_has_contents(self, fake_filesystem):
 202         interesting_file = fake_filesystem.create_file('/tmp/fake-file', contents="1")
 203         result = system.get_file_contents(interesting_file.path)
 204         assert result == "1"
 205
 206     def test_path_has_multiline_contents(self, fake_filesystem):
 207         interesting_file = fake_filesystem.create_file('/tmp/fake-file', contents="0\n1")
 208         result = system.get_file_contents(interesting_file.path)
 209         assert result == "0\n1"
 210
 211     def test_exception_returns_default(self):
 212         with patch('builtins.open') as mocked_open:
 213             mocked_open.side_effect = Exception()
 214             result = system.get_file_contents('/tmp/fake-file')
 215         assert result == ''
 216
 217
 218 class TestWhich(object):
 219
 220     def test_executable_exists_but_is_not_file(self, monkeypatch):
 221         monkeypatch.setattr(system.os.path, 'isfile', lambda x: False)
 222         monkeypatch.setattr(system.os.path, 'exists', lambda x: True)
 223         assert system.which('exedir') == 'exedir'
 224
 225     def test_executable_does_not_exist(self, monkeypatch):
 226         monkeypatch.setattr(system.os.path, 'isfile', lambda x: False)
 227         monkeypatch.setattr(system.os.path, 'exists', lambda x: False)
 228         assert system.which('exedir') == 'exedir'
 229
 230     def test_executable_exists_as_file(self, monkeypatch):
 231         monkeypatch.setattr(system.os, 'getenv', lambda x, y: '')
 232         monkeypatch.setattr(system.os.path, 'isfile', lambda x: x != 'ceph')
 233         monkeypatch.setattr(system.os.path, 'exists', lambda x: x != 'ceph')
 234         assert system.which('ceph') == '/usr/local/bin/ceph'
 235
 236     def test_warnings_when_executable_isnt_matched(self, monkeypatch, capsys):
 237         monkeypatch.setattr(system.os.path, 'isfile', lambda x: True)
 238         monkeypatch.setattr(system.os.path, 'exists', lambda x: False)
 239         system.which('exedir')
 240         cap = capsys.readouterr()
 241         assert 'Executable exedir not in PATH' in cap.err
 242
 243     def test_run_on_host_found(self, mock_find_executable_on_host):
 244         mock_find_executable_on_host(stdout="/sbin/lvs\n", stderr="some stderr message\n")
 245         assert system.which('lvs', run_on_host=True) == '/sbin/lvs'
 246
 247     def test_run_on_host_not_found(self, mock_find_executable_on_host):
 248         mock_find_executable_on_host(stdout="", stderr="some stderr message\n")
 249         assert system.which('lvs', run_on_host=True) == 'lvs'
 250
 251 @pytest.fixture
 252 def stub_which(monkeypatch):
 253     def apply(value='/bin/restorecon'):
 254         monkeypatch.setattr(system, 'which', lambda x: value)
 255     return apply
 256
 257
 258 # python2 has no FileNotFoundError
 259 try:
 260     FileNotFoundError
 261 except NameError:
 262     FileNotFoundError = OSError
 263
 264
 265 class TestSetContext(object):
 266
 267     def setup(self):
 268         try:
 269             os.environ.pop('CEPH_VOLUME_SKIP_RESTORECON')
 270         except KeyError:
 271             pass
 272
 273     @pytest.mark.parametrize('value', ['1', 'True', 'true', 'TRUE', 'yes'])
 274     def test_set_context_skips(self, stub_call, fake_run, value):
 275         stub_call(('', '', 0))
 276         os.environ['CEPH_VOLUME_SKIP_RESTORECON'] = value
 277         system.set_context('/tmp/foo')
 278         assert fake_run.calls == []
 279
 280     @pytest.mark.parametrize('value', ['0', 'False', 'false', 'FALSE', 'no'])
 281     def test_set_context_doesnt_skip_with_env(self, stub_call, stub_which, fake_run, value):
 282         stub_call(('', '', 0))
 283         stub_which()
 284         os.environ['CEPH_VOLUME_SKIP_RESTORECON'] = value
 285         system.set_context('/tmp/foo')
 286         assert len(fake_run.calls)
 287
 288     def test_set_context_skips_on_executable(self, stub_call, stub_which, fake_run):
 289         stub_call(('', '', 0))
 290         stub_which('restorecon')
 291         system.set_context('/tmp/foo')
 292         assert fake_run.calls == []
 293
 294     def test_set_context_no_skip_on_executable(self, stub_call, stub_which, fake_run):
 295         stub_call(('', '', 0))
 296         stub_which('/bin/restorecon')
 297         system.set_context('/tmp/foo')
 298         assert len(fake_run.calls)
 299
 300     @patch('ceph_volume.process.call')
 301     def test_selinuxenabled_doesnt_exist(self, mocked_call, fake_run):
 302         mocked_call.side_effect = FileNotFoundError()
 303         system.set_context('/tmp/foo')
 304         assert fake_run.calls == []
 305
 306     def test_selinuxenabled_is_not_enabled(self, stub_call, fake_run):
 307         stub_call(('', '', 1))
 308         system.set_context('/tmp/foo')
 309         assert fake_run.calls == []