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
12 def mock_find_executable_on_host(monkeypatch
):
14 Monkeypatches util.system.find_executable_on_host, so that a caller can add behavior to the response
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(
22 wait
=lambda: returncode
,
23 communicate
=lambda x
: (stdout
, stderr
, returncode
)
27 'ceph_volume.util.system.subprocess.Popen',
28 lambda *a
, **kw
: return_value
)
32 class TestMkdirP(object):
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
,))
40 assert os
.path
.isdir(path
)
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')
48 assert os
.path
.isdir(path
)
50 def test_existing_dir_does_not_raise_no_chown(self
, tmpdir
):
52 system
.mkdir_p(path
, chown
=False)
53 assert os
.path
.isdir(path
)
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
)
62 def fake_proc(tmpdir
, monkeypatch
):
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)
98 class TestPathIsMounted(object):
100 def test_is_mounted(self
, fake_proc
):
101 assert system
.path_is_mounted('/boot') is True
103 def test_is_not_mounted(self
, fake_proc
):
104 assert system
.path_is_mounted('/far/fib/feph') is False
106 def test_is_not_mounted_at_destination(self
, fake_proc
):
107 assert system
.path_is_mounted('/boot', destination
='/dev/sda1') is False
109 def test_is_mounted_at_destination(self
, fake_proc
):
110 assert system
.path_is_mounted('/boot', destination
='/dev/sdc2') is True
113 class TestDeviceIsMounted(object):
115 def test_is_mounted(self
, fake_proc
):
116 assert system
.device_is_mounted('/dev/sda1') is True
118 def test_path_is_not_device(self
, fake_proc
):
119 assert system
.device_is_mounted('/far/lib/ceph/osd/ceph-7') is False
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
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
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
132 def test_is_realpath_path_mounted_at_destination(self
, fake_proc
, monkeypatch
):
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
140 class TestGetMounts(object):
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
:
147 monkeypatch
.setattr(system
, 'PROCDIR', PROCDIR
)
149 assert m
.get_mounts() == {}
151 def test_is_mounted_(self
, fake_proc
):
153 assert m
.get_mounts()['/dev/sdc2'] == ['/boot']
155 def test_ignores_two_fields(self
, fake_proc
):
157 assert m
.get_mounts().get('/dev/sde4') is None
159 def test_tmpfs_is_reported(self
, fake_proc
):
161 assert m
.get_mounts()['tmpfs'][0] == '/dev/shm'
163 def test_non_skip_devs_arent_reported(self
, fake_proc
):
165 assert m
.get_mounts().get('cgroup') is None
167 def test_multiple_mounts_are_appended(self
, fake_proc
):
169 assert len(m
.get_mounts()['tmpfs']) == 7
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)
181 assert m
.get_mounts().get('/dev/sda1') is None
184 class TestIsBinary(object):
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
)
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
195 class TestGetFileContents(object):
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'
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
)
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"
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')
218 class TestWhich(object):
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'
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'
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'
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
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'
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'
252 def stub_which(monkeypatch
):
253 def apply(value
='/bin/restorecon'):
254 monkeypatch
.setattr(system
, 'which', lambda x
: value
)
258 # python2 has no FileNotFoundError
262 FileNotFoundError
= OSError
265 class TestSetContext(object):
267 def setup_method(self
):
269 os
.environ
.pop('CEPH_VOLUME_SKIP_RESTORECON')
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
== []
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))
284 os
.environ
['CEPH_VOLUME_SKIP_RESTORECON'] = value
285 system
.set_context('/tmp/foo')
286 assert len(fake_run
.calls
)
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
== []
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
)
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
== []
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
== []