1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
#!/usr/bin/env python3
'''
util.py - Utility functions for Python-LLFUSE unit tests.
Copyright © 2015 Nikolaus Rath <Nikolaus.org>
This file is part of Python-LLFUSE. This work may be distributed under
the terms of the GNU LGPL.
'''
import platform
import subprocess
import pytest
import os
import stat
import time
def fuse_test_marker():
'''Return a pytest.marker that indicates FUSE availability
If system/user/environment does not support FUSE, return
a `pytest.mark.skip` object with more details. If FUSE is
supported, return `pytest.mark.uses_fuse()`.
'''
if platform.system() == 'Darwin':
# No working autodetection, just assume it will work.
return pytest.mark.uses_fuse()
skip = lambda x: pytest.mark.skip(reason=x)
# Python 2.x: Popen is not a context manager...
which = subprocess.Popen(['which', 'fusermount'], stdout=subprocess.PIPE,
universal_newlines=True)
try:
fusermount_path = which.communicate()[0].strip()
finally:
which.wait()
if not fusermount_path or which.returncode != 0:
return skip("Can't find fusermount executable")
if not os.path.exists('/dev/fuse'):
return skip("FUSE kernel module does not seem to be loaded")
if os.getuid() == 0:
return pytest.mark.uses_fuse()
mode = os.stat(fusermount_path).st_mode
if mode & stat.S_ISUID == 0:
return skip('fusermount executable not setuid, and we are not root.')
try:
fd = os.open('/dev/fuse', os.O_RDWR)
except OSError as exc:
return skip('Unable to open /dev/fuse: %s' % exc.strerror)
else:
os.close(fd)
return pytest.mark.uses_fuse()
def exitcode(process):
if isinstance(process, subprocess.Popen):
return process.poll()
else:
if process.is_alive():
return None
else:
return process.exitcode
def wait_for(callable, timeout=10, interval=0.1):
'''Wait until *callable* returns something True and return it
If *timeout* expires, return None
'''
waited = 0
while True:
ret = callable()
if ret:
return ret
if waited > timeout:
return None
waited += interval
time.sleep(interval)
def wait_for_mount(mount_process, mnt_dir):
elapsed = 0
while elapsed < 30:
if os.path.ismount(mnt_dir):
return True
if exitcode(mount_process) is not None:
pytest.fail('file system process terminated prematurely')
time.sleep(0.1)
elapsed += 0.1
pytest.fail("mountpoint failed to come up")
def cleanup(mnt_dir):
if platform.system() == 'Darwin':
subprocess.call(['umount', '-l', mnt_dir], stdout=subprocess.DEVNULL,
stderr=subprocess.STDOUT)
else:
subprocess.call(['fusermount', '-z', '-u', mnt_dir], stdout=subprocess.DEVNULL,
stderr=subprocess.STDOUT)
def umount(mount_process, mnt_dir):
if platform.system() == 'Darwin':
subprocess.check_call(['umount', '-l', mnt_dir])
else:
subprocess.check_call(['fusermount', '-z', '-u', mnt_dir])
assert not os.path.ismount(mnt_dir)
# Give mount process a little while to terminate. Popen.wait(timeout)
# was only added in 3.3...
elapsed = 0
while elapsed < 30:
code = exitcode(mount_process)
if code is not None:
if code == 0:
return
pytest.fail('file system process terminated with code %s' % (code,))
time.sleep(0.1)
elapsed += 0.1
pytest.fail('mount process did not terminate')
|
