Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def tst_bug382(self):
dirname = self.newname()
fullname = self.mnt_dir + "/" + dirname
os.mkdir(fullname)
assert stat.S_ISDIR(os.stat(fullname).st_mode)
assert dirname in pyfuse3.listdir(self.mnt_dir)
cmd = ('(%d, %r)' % (pyfuse3.ROOT_INODE, path2bytes(dirname))).encode()
pyfuse3.setxattr('%s/%s' % (self.mnt_dir, CTRL_NAME), 'rmtree', cmd)
# Invalidation is asynchronous...
try:
retry(5, lambda: not os.path.exists(fullname))
except RetryTimeoutError:
pass # assert_raises should fail
assert_raises(FileNotFoundError, os.stat, fullname)
assert dirname not in pyfuse3.listdir(self.mnt_dir)
def main(args=None):
'''Control a mounted S3QL File System.'''
if args is None:
args = sys.argv[1:]
options = parse_args(args)
setup_logging(options)
path = options.mountpoint
ctrlfile = assert_fs_owner(path, mountpoint=True)
if options.action == 'flushcache':
pyfuse3.setxattr(ctrlfile, 's3ql_flushcache!', b'dummy')
elif options.action == 'dropcache':
pyfuse3.setxattr(ctrlfile, 's3ql_dropcache!', b'dummy')
elif options.action == 'upload-meta':
pyfuse3.setxattr(ctrlfile, 'upload-meta', b'dummy')
elif options.action == 'log':
level = getattr(logging, options.level.upper())
cmd = ('(%r, %r)' % (level, ','.join(options.modules))).encode()
pyfuse3.setxattr(ctrlfile, 'logging', cmd)
elif options.action == 'cachesize':
pyfuse3.setxattr(ctrlfile, 'cachesize', ('%d' % (options.cachesize * 1024,)).encode())
options = parse_args(args)
setup_logging(options)
path = options.mountpoint
ctrlfile = assert_fs_owner(path, mountpoint=True)
if options.action == 'flushcache':
pyfuse3.setxattr(ctrlfile, 's3ql_flushcache!', b'dummy')
elif options.action == 'dropcache':
pyfuse3.setxattr(ctrlfile, 's3ql_dropcache!', b'dummy')
elif options.action == 'upload-meta':
pyfuse3.setxattr(ctrlfile, 'upload-meta', b'dummy')
elif options.action == 'log':
level = getattr(logging, options.level.upper())
cmd = ('(%r, %r)' % (level, ','.join(options.modules))).encode()
pyfuse3.setxattr(ctrlfile, 'logging', cmd)
elif options.action == 'cachesize':
pyfuse3.setxattr(ctrlfile, 'cachesize', ('%d' % (options.cachesize * 1024,)).encode())
if options.action == 'flushcache':
pyfuse3.setxattr(ctrlfile, 's3ql_flushcache!', b'dummy')
elif options.action == 'dropcache':
pyfuse3.setxattr(ctrlfile, 's3ql_dropcache!', b'dummy')
elif options.action == 'upload-meta':
pyfuse3.setxattr(ctrlfile, 'upload-meta', b'dummy')
elif options.action == 'log':
level = getattr(logging, options.level.upper())
cmd = ('(%r, %r)' % (level, ','.join(options.modules))).encode()
pyfuse3.setxattr(ctrlfile, 'logging', cmd)
elif options.action == 'cachesize':
pyfuse3.setxattr(ctrlfile, 'cachesize', ('%d' % (options.cachesize * 1024,)).encode())
def blocking_umount(mountpoint):
'''Invoke fusermount and wait for daemon to terminate.'''
with open('/dev/null', 'wb') as devnull:
if subprocess.call(['fuser', '-m', mountpoint], stdout=devnull,
stderr=devnull) == 0:
raise MountInUseError(mountpoint)
ctrlfile = os.path.join(mountpoint, CTRL_NAME)
log.debug('Flushing cache...')
pyfuse3.setxattr(ctrlfile, 's3ql_flushcache!', b'dummy')
# Get pid
log.debug('Trying to get pid')
pid = parse_literal(pyfuse3.getxattr(ctrlfile, 's3ql_pid?'), int)
log.debug('PID is %d', pid)
# Get command line to make race conditions less-likely
cmdline = get_cmdline(pid)
# Unmount
log.debug('Unmounting...')
if os.getuid() == 0 or platform.system() == 'Darwin':
# MacOS X always uses umount rather than fusermount
umount_cmd = ['umount', mountpoint]
else:
def main(args=None):
'''Make directory tree immutable'''
if args is None:
args = sys.argv[1:]
options = parse_args(args)
setup_logging(options)
for name in options.path:
if os.path.ismount(name):
raise QuietError('%s is a mount point.' % name)
ctrlfile = assert_fs_owner(name)
fstat = os.stat(name)
pyfuse3.setxattr(ctrlfile, 'lock', ('%d' % fstat.st_ino).encode())