我们从Python开源项目中，提取了以下24个代码示例，用于说明如何使用os.statvfs_result()。
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))
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 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): 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
