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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
|
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
util.py - Utility functions for pyfuse3 unit tests.
Copyright © 2015 Nikolaus Rath <Nikolaus.org>
This file is part of pyfuse3. This work may be distributed under
the terms of the GNU LGPL.
'''
import os
import platform
import pytest
import shutil
import stat
import subprocess
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
skip = lambda x: pytest.mark.skip(reason=x)
fusermount_path = shutil.which('fusermount')
if fusermount_path is None:
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(mount_process, 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)
mount_process.terminate()
if isinstance(mount_process, subprocess.Popen):
try:
mount_process.wait(1)
except subprocess.TimeoutExpired:
mount_process.kill()
else:
mount_process.join(5)
if mount_process.exitcode is None:
mount_process.kill()
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)
if isinstance(mount_process, subprocess.Popen):
try:
code = mount_process.wait(5)
if code == 0:
return
pytest.fail('file system process terminated with code %s' % (code,))
except subprocess.TimeoutExpired:
mount_process.terminate()
try:
mount_process.wait(1)
except subprocess.TimeoutExpired:
mount_process.kill()
else:
mount_process.join(5)
code = mount_process.exitcode
if code == 0:
return
elif code is None:
mount_process.terminate()
mount_process.join(1)
else:
pytest.fail('file system process terminated with code %s' % (code,))
pytest.fail('mount process did not terminate')
|
