def set_mount_points(self, points, read_access=True):
"""
This method prepares a fake mounts file containing the
mount points specified in the C{points} list of strings. This file
can then be used by referencing C{self.mount_file}.
If C{read_access} is set to C{False}, then all mount points will
yield a permission denied error when inspected.
"""
self.read_access = read_access
content = "\n".join("/dev/sda%d %s ext4 rw 0 0" % (i, point)
for i, point in enumerate(points))
f = open(self.mount_file, "w")
f.write(content)
f.close()
for point in points:
self.stat_results[point] = os.statvfs_result(
(4096, 0, 1000, 500, 0, 0, 0, 0, 0, 0))
python类statvfs_result()的实例源码
def test_ignore_unmounted_and_virtual_mountpoints(self):
"""
Make sure autofs and virtual mountpoints are ignored. This is to
ensure non-regression on bug #1045374.
"""
self.read_access = True
content = "\n".join(["auto_direct /opt/whatever autofs",
"none /run/lock tmpfs",
"proc /proc proc",
"/dev/sda1 /home ext4"])
f = open(self.mount_file, "w")
f.write(content)
f.close()
self.stat_results["/home"] = os.statvfs_result(
(4096, 0, 1000, 500, 0, 0, 0, 0, 0, 0))
result = [x for x in get_mount_info(self.mount_file, self.statvfs)]
expected = {"device": "/dev/sda1", "mount-point": "/home",
"filesystem": "ext4", "total-space": 3, "free-space": 1}
self.assertEqual([expected], result)
def test_sample_free_space(self):
"""Test collecting information about free space."""
counter = mock_counter(1)
def statvfs(path, multiplier=lambda: next(counter)):
return os.statvfs_result(
(4096, 0, mb(1000), mb(multiplier() * 100), 0, 0, 0, 0, 0, 0))
plugin = self.get_mount_info(
statvfs=statvfs, create_time=self.reactor.time)
step_size = self.monitor.step_size
self.monitor.add(plugin)
self.reactor.advance(step_size)
message = plugin.create_free_space_message()
self.assertTrue(message)
self.assertEqual(message.get("type"), "free-space")
free_space = message.get("free-space", ())
self.assertEqual(len(free_space), 1)
self.assertEqual(free_space[0], (step_size, "/", 409600))
def mock_os_statvfs(self):
f_bsize = 4096
f_frsize = 4096
f_blocks = 12868767
f_bfree = 11503139
f_bavail = 10847779
f_files = 3276800
f_ffree = 3220784
f_favail = 3220784
f_flag = 4096
f_namemax = 255
return os.statvfs_result((f_bsize, f_frsize,
f_blocks, f_bfree,
f_bavail, f_files,
f_ffree, f_favail,
f_flag, f_namemax))
def statvfs(volume, host='localhost', port=24007, protocal='tcp'):
_lazy_init_libgfapi()
data = StatVfsData()
_api.glfs_statvfs.restype = ctypes.c_int
_api.glfs_statvfs.argtypes = [ctypes.c_void_p,
ctypes.c_char_p,
ctypes.POINTER(StatVfsData)]
fs = _mount_gluster_volume(volume, host, port, protocal)
rc = _api.glfs_statvfs(fs, "/", ctypes.byref(data))
# To convert to os.statvfs_result we need to pass tuple/list in
# following order: bsize, frsize, blocks, bfree, bavail, files
# ffree, favail, flag, namemax
return os.statvfs_result((data.f_bsize,
data.f_frsize,
data.f_blocks,
data.f_bfree,
data.f_bavail,
data.f_files,
data.f_ffree,
data.f_favail,
data.f_flag,
data.f_namemax))
def statvfs(self, path):
"""Get attributes of the file system containing a file
:param bytes path:
The path of the file system to get attributes for
:returns: An :class:`SFTPVFSAttrs` or an os.statvfs_result
containing the file system attributes
:raises: :exc:`SFTPError` to return an error to the client
"""
raise SFTPError(FX_OP_UNSUPPORTED, 'Not implemented')
def fstatvfs(self, file_obj):
"""Return attributes of the file system containing an open file
:param file file_obj:
The open file to get file system attributes for
:returns: An :class:`SFTPVFSAttrs` or an os.statvfs_result
containing the file system attributes
:raises: :exc:`SFTPError` to return an error to the client
"""
raise SFTPError(FX_OP_UNSUPPORTED, 'Not implemented')
def test_get_filesystem_subpath(self):
self.set_mount_points(["/"])
self.stat_results["/"] = os.statvfs_result(
(4096, 0, 1000, 500, 0, 0, 0, 0, 0, 0))
info = get_filesystem_for_path("/home", self.mount_file, self.statvfs)
self.assertEqual(info["mount-point"], "/")
def statvfs_result_fixture(path):
"""Fixture for a dummy statvfs_result."""
return os.statvfs_result((4096, 0, mb(1000), mb(100), 0, 0, 0, 0, 0, 0))
def get_mount_info(self, *args, **kwargs):
if "mounts_file" not in kwargs:
kwargs["mounts_file"] = self.makeFile("/dev/hda1 / ext3 rw 0 0\n")
if "mtab_file" not in kwargs:
kwargs["mtab_file"] = self.makeFile("/dev/hda1 / ext3 rw 0 0\n")
if "statvfs" not in kwargs:
kwargs["statvfs"] = lambda path: os.statvfs_result(
(0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
plugin = MountInfo(*args, **kwargs)
# To make sure tests are isolated from the real system by default.
plugin.is_device_removable = lambda x: False
return plugin
def add_mount(self, point, block_size=4096, capacity=1000, unused=1000,
fs="ext3", device=None):
if device is None:
device = "/dev/" + point.replace("/", "_")
self.stat_results[point] = os.statvfs_result(
(block_size, 0, capacity, unused, 0, 0, 0, 0, 0, 0))
f = open(self.mount_file, "a")
f.write("/dev/%s %s %s rw 0 0\n" % (device, point, fs))
f.close()
def test_statvfs_result_pickle(self):
try:
result = os.statvfs(self.fname)
except OSError as e:
# On AtheOS, glibc always returns ENOSYS
if e.errno == errno.ENOSYS:
self.skipTest('os.statvfs() failed with ENOSYS')
for proto in range(pickle.HIGHEST_PROTOCOL + 1):
p = pickle.dumps(result, proto)
self.assertIn(b'statvfs_result', p)
if proto < 4:
self.assertIn(b'cos\nstatvfs_result\n', p)
unpickled = pickle.loads(p)
self.assertEqual(result, unpickled)
def getVolumeStatvfs(volumeId, host=GLUSTER_VOL_HOST,
port=GLUSTER_VOL_PORT,
protocol=GLUSTER_VOL_PROTOCAL):
statvfsdata = StatVfsStruct()
fs = glfsInit(volumeId, host, port, protocol)
rc = _glfs_statvfs(fs, GLUSTER_VOL_PATH, ctypes.byref(statvfsdata))
if rc != 0:
raise GlfsStatvfsException(rc=rc)
glfsFini(fs, volumeId)
# To convert to os.statvfs_result we need to pass tuple/list in
# following order: bsize, frsize, blocks, bfree, bavail, files,
# ffree, favail, flag, namemax
return os.statvfs_result((statvfsdata.f_bsize,
statvfsdata.f_frsize,
statvfsdata.f_blocks,
statvfsdata.f_bfree,
statvfsdata.f_bavail,
statvfsdata.f_files,
statvfsdata.f_ffree,
statvfsdata.f_favail,
statvfsdata.f_flag,
statvfsdata.f_namemax))
# C function prototypes for using the library gfapi
def getVolumeStatvfs(volumeId, host=GLUSTER_VOL_HOST,
port=GLUSTER_VOL_PORT,
protocol=GLUSTER_VOL_PROTOCAL):
statvfsdata = StatVfsStruct()
fs = glfsInit(volumeId, host, port, protocol)
rc = _glfs_statvfs(fs, GLUSTER_VOL_PATH, ctypes.byref(statvfsdata))
if rc != 0:
raise GlfsStatvfsException(rc=rc)
glfsFini(fs, volumeId)
# To convert to os.statvfs_result we need to pass tuple/list in
# following order: bsize, frsize, blocks, bfree, bavail, files,
# ffree, favail, flag, namemax
return os.statvfs_result((statvfsdata.f_bsize,
statvfsdata.f_frsize,
statvfsdata.f_blocks,
statvfsdata.f_bfree,
statvfsdata.f_bavail,
statvfsdata.f_files,
statvfsdata.f_ffree,
statvfsdata.f_favail,
statvfsdata.f_flag,
statvfsdata.f_namemax))
# C function prototypes for using the library gfapi
def statvfs(self):
"""
See :meth:`FileJuggler.do_statvfs`.
"""
return os.statvfs_result(self._execute(['STATVFS']))
def test_statvfs_attributes(self):
if not hasattr(os, "statvfs"):
return
try:
result = os.statvfs(self.fname)
except OSError as e:
# On AtheOS, glibc always returns ENOSYS
if e.errno == errno.ENOSYS:
return
# Make sure direct access works
self.assertEqual(result.f_bfree, result[3])
# Make sure all the attributes are there.
members = ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
'ffree', 'favail', 'flag', 'namemax')
for value, member in enumerate(members):
self.assertEqual(getattr(result, 'f_' + member), result[value])
# Make sure that assignment really fails
try:
result.f_bfree = 1
self.fail("No exception thrown")
except AttributeError:
pass
try:
result.parrot = 1
self.fail("No exception thrown")
except AttributeError:
pass
# Use the constructor with a too-short tuple.
try:
result2 = os.statvfs_result((10,))
self.fail("No exception thrown")
except TypeError:
pass
# Use the constructor with a too-long tuple.
try:
result2 = os.statvfs_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
except TypeError:
pass
def test_read_disjointed_changing_total_space(self):
"""
Total space measurements are only sent when (a) none have ever
been sent, or (b) the value has changed since the last time
data was collected. This test ensures that the (b) criteria
is checked per-mount point. The sample statvfs() function
only provides changing total space for /; therefore, new
messages should only be queued for / after the first message
is created.
"""
counter = mock_counter(1)
def statvfs(path, multiplier=lambda: next(counter)):
if path == "/":
return os.statvfs_result(
(4096, 0, mb(1000), mb(100), 0, 0, 0, 0, 0, 0))
return os.statvfs_result(
(4096, 0, mb(multiplier() * 1000), mb(100), 0, 0, 0, 0, 0, 0))
filename = self.makeFile("""\
/dev/hda1 / ext3 rw 0 0
/dev/hde1 /mnt/hde1 ext3 rw 0 0
""")
plugin = self.get_mount_info(mounts_file=filename, statvfs=statvfs,
create_time=self.reactor.time,
interval=self.monitor.step_size,
mtab_file=filename)
self.monitor.add(plugin)
self.reactor.advance(self.monitor.step_size * 2)
message = plugin.create_mount_info_message()
self.assertTrue(message)
mount_info = message.get("mount-info", ())
self.assertEqual(len(mount_info), 3)
self.assertEqual(mount_info[0][0], self.monitor.step_size)
self.assertEqual(mount_info[0][1],
{"device": "/dev/hda1", "mount-point": "/",
"filesystem": "ext3", "total-space": 4096000})
self.assertEqual(mount_info[1][0], self.monitor.step_size)
self.assertEqual(mount_info[1][1],
{"device": "/dev/hde1", "mount-point": "/mnt/hde1",
"filesystem": "ext3", "total-space": 4096000})
self.assertEqual(mount_info[2][0], self.monitor.step_size * 2)
self.assertEqual(mount_info[2][1],
{"device": "/dev/hde1", "mount-point": "/mnt/hde1",
"filesystem": "ext3", "total-space": 8192000})
def test_statvfs_attributes(self):
try:
result = os.statvfs(self.fname)
except OSError, e:
# On AtheOS, glibc always returns ENOSYS
if e.errno == errno.ENOSYS:
self.skipTest('glibc always returns ENOSYS on AtheOS')
# Make sure direct access works
self.assertEqual(result.f_bfree, result[3])
# Make sure all the attributes are there.
members = ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
'ffree', 'favail', 'flag', 'namemax')
for value, member in enumerate(members):
self.assertEqual(getattr(result, 'f_' + member), result[value])
# Make sure that assignment really fails
try:
result.f_bfree = 1
self.fail("No exception raised")
except TypeError:
pass
try:
result.parrot = 1
self.fail("No exception raised")
except AttributeError:
pass
# Use the constructor with a too-short tuple.
try:
result2 = os.statvfs_result((10,))
self.fail("No exception raised")
except TypeError:
pass
# Use the constructor with a too-long tuple.
try:
result2 = os.statvfs_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
except TypeError:
pass
def test_statvfs_attributes(self):
try:
result = os.statvfs(self.fname)
except OSError, e:
# On AtheOS, glibc always returns ENOSYS
if e.errno == errno.ENOSYS:
self.skipTest('glibc always returns ENOSYS on AtheOS')
# Make sure direct access works
self.assertEqual(result.f_bfree, result[3])
# Make sure all the attributes are there.
members = ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
'ffree', 'favail', 'flag', 'namemax')
for value, member in enumerate(members):
self.assertEqual(getattr(result, 'f_' + member), result[value])
# Make sure that assignment really fails
try:
result.f_bfree = 1
self.fail("No exception raised")
except TypeError:
pass
try:
result.parrot = 1
self.fail("No exception raised")
except AttributeError:
pass
# Use the constructor with a too-short tuple.
try:
result2 = os.statvfs_result((10,))
self.fail("No exception raised")
except TypeError:
pass
# Use the constructor with a too-long tuple.
try:
result2 = os.statvfs_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
except TypeError:
pass
def test_statvfs_attributes(self):
if not hasattr(os, "statvfs"):
return
try:
result = os.statvfs(self.fname)
except OSError as e:
# On AtheOS, glibc always returns ENOSYS
if e.errno == errno.ENOSYS:
return
# Make sure direct access works
self.assertEqual(result.f_bfree, result[3])
# Make sure all the attributes are there.
members = ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
'ffree', 'favail', 'flag', 'namemax')
for value, member in enumerate(members):
self.assertEqual(getattr(result, 'f_' + member), result[value])
# Make sure that assignment really fails
try:
result.f_bfree = 1
self.fail("No exception raised")
except AttributeError:
pass
try:
result.parrot = 1
self.fail("No exception raised")
except AttributeError:
pass
# Use the constructor with a too-short tuple.
try:
result2 = os.statvfs_result((10,))
self.fail("No exception raised")
except TypeError:
pass
# Use the constructor with a too-long tuple.
try:
result2 = os.statvfs_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
except TypeError:
pass
def test_statvfs_attributes(self):
try:
result = os.statvfs(self.fname)
except OSError, e:
# On AtheOS, glibc always returns ENOSYS
if e.errno == errno.ENOSYS:
self.skipTest('glibc always returns ENOSYS on AtheOS')
# Make sure direct access works
self.assertEqual(result.f_bfree, result[3])
# Make sure all the attributes are there.
members = ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
'ffree', 'favail', 'flag', 'namemax')
for value, member in enumerate(members):
self.assertEqual(getattr(result, 'f_' + member), result[value])
# Make sure that assignment really fails
try:
result.f_bfree = 1
self.fail("No exception raised")
except (TypeError, AttributeError):
pass
try:
result.parrot = 1
self.fail("No exception raised")
except AttributeError:
pass
# Use the constructor with a too-short tuple.
try:
result2 = os.statvfs_result((10,))
self.fail("No exception raised")
except TypeError:
pass
# Use the constructor with a too-long tuple.
try:
result2 = os.statvfs_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
except TypeError:
pass
def test_statvfs_attributes(self):
try:
result = os.statvfs(self.fname)
except OSError as e:
# On AtheOS, glibc always returns ENOSYS
if e.errno == errno.ENOSYS:
self.skipTest('os.statvfs() failed with ENOSYS')
# Make sure direct access works
self.assertEqual(result.f_bfree, result[3])
# Make sure all the attributes are there.
members = ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
'ffree', 'favail', 'flag', 'namemax')
for value, member in enumerate(members):
self.assertEqual(getattr(result, 'f_' + member), result[value])
# Make sure that assignment really fails
try:
result.f_bfree = 1
self.fail("No exception raised")
except AttributeError:
pass
try:
result.parrot = 1
self.fail("No exception raised")
except AttributeError:
pass
# Use the constructor with a too-short tuple.
try:
result2 = os.statvfs_result((10,))
self.fail("No exception raised")
except TypeError:
pass
# Use the constructor with a too-long tuple.
try:
result2 = os.statvfs_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
except TypeError:
pass
def test_statvfs_attributes(self):
if not hasattr(os, "statvfs"):
return
try:
result = os.statvfs(self.fname)
except OSError, e:
# On AtheOS, glibc always returns ENOSYS
if e.errno == errno.ENOSYS:
return
# Make sure direct access works
self.assertEqual(result.f_bfree, result[3])
# Make sure all the attributes are there.
members = ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
'ffree', 'favail', 'flag', 'namemax')
for value, member in enumerate(members):
self.assertEqual(getattr(result, 'f_' + member), result[value])
# Make sure that assignment really fails
try:
result.f_bfree = 1
self.fail("No exception raised")
except TypeError:
pass
try:
result.parrot = 1
self.fail("No exception raised")
except AttributeError:
pass
# Use the constructor with a too-short tuple.
try:
result2 = os.statvfs_result((10,))
self.fail("No exception raised")
except TypeError:
pass
# Use the constructor with a too-long tuple.
try:
result2 = os.statvfs_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
except TypeError:
pass
def test_statvfs_attributes(self):
try:
result = os.statvfs(self.fname)
except OSError as e:
# On AtheOS, glibc always returns ENOSYS
if e.errno == errno.ENOSYS:
self.skipTest('os.statvfs() failed with ENOSYS')
# Make sure direct access works
self.assertEqual(result.f_bfree, result[3])
# Make sure all the attributes are there.
members = ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
'ffree', 'favail', 'flag', 'namemax')
for value, member in enumerate(members):
self.assertEqual(getattr(result, 'f_' + member), result[value])
# Make sure that assignment really fails
try:
result.f_bfree = 1
self.fail("No exception raised")
except AttributeError:
pass
try:
result.parrot = 1
self.fail("No exception raised")
except AttributeError:
pass
# Use the constructor with a too-short tuple.
try:
result2 = os.statvfs_result((10,))
self.fail("No exception raised")
except TypeError:
pass
# Use the constructor with a too-long tuple.
try:
result2 = os.statvfs_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
except TypeError:
pass
