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