def __hashEntry(self, prefix, entry, s):
if stat.S_ISREG(s.st_mode):
digest = self.__index.check(prefix, entry, s, hashFile)
elif stat.S_ISDIR(s.st_mode):
digest = self.__hashDir(prefix, entry)
elif stat.S_ISLNK(s.st_mode):
digest = self.__index.check(prefix, entry, s, DirHasher.__hashLink)
elif stat.S_ISBLK(s.st_mode) or stat.S_ISCHR(s.st_mode):
digest = struct.pack("<L", s.st_rdev)
elif stat.S_ISFIFO(s.st_mode):
digest = b''
else:
digest = b''
logging.getLogger(__name__).warning("Unknown file: %s", entry)
return digest
python类S_ISCHR的实例源码
def __init__(self, devname=None):
if devname is None:
self.name = "/dev/urandom"
else:
self.name = devname
# Test that /dev/urandom is a character special device
f = open(self.name, "rb", 0)
fmode = os.fstat(f.fileno())[stat.ST_MODE]
if not stat.S_ISCHR(fmode):
f.close()
raise TypeError("%r is not a character special device" % (self.name,))
self.__file = f
BaseRNG.__init__(self)
def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
super().__init__(extra)
self._extra['pipe'] = pipe
self._loop = loop
self._pipe = pipe
self._fileno = pipe.fileno()
mode = os.fstat(self._fileno).st_mode
if not (stat.S_ISFIFO(mode) or
stat.S_ISSOCK(mode) or
stat.S_ISCHR(mode)):
raise ValueError("Pipe transport is for pipes/sockets only.")
_set_nonblocking(self._fileno)
self._protocol = protocol
self._closing = False
self._loop.call_soon(self._protocol.connection_made, self)
# only start reading when connection_made() has been called
self._loop.call_soon(self._loop.add_reader,
self._fileno, self._read_ready)
if waiter is not None:
# only wake up the waiter when connection_made() has been called
self._loop.call_soon(waiter._set_result_unless_cancelled, None)
def __init__(self, devname=None):
if devname is None:
self.name = "/dev/urandom"
else:
self.name = devname
# Test that /dev/urandom is a character special device
f = open(self.name, "rb", 0)
fmode = os.fstat(f.fileno())[stat.ST_MODE]
if not stat.S_ISCHR(fmode):
f.close()
raise TypeError("%r is not a character special device" % (self.name,))
self.__file = f
BaseRNG.__init__(self)
def __init__(self, devname=None):
if devname is None:
self.name = "/dev/urandom"
else:
self.name = devname
# Test that /dev/urandom is a character special device
f = open(self.name, "rb", 0)
fmode = os.fstat(f.fileno())[stat.ST_MODE]
if not stat.S_ISCHR(fmode):
f.close()
raise TypeError("%r is not a character special device" % (self.name,))
self.__file = f
BaseRNG.__init__(self)
def is_special_file(path):
"""
This function checks to see if a special file. It checks if the
file is a character special device, block special device, FIFO, or
socket.
"""
mode = os.stat(path).st_mode
# Character special device.
if stat.S_ISCHR(mode):
return True
# Block special device
if stat.S_ISBLK(mode):
return True
# FIFO.
if stat.S_ISFIFO(mode):
return True
# Socket.
if stat.S_ISSOCK(mode):
return True
return False
def __init__(self, devname=None):
if devname is None:
self.name = "/dev/urandom"
else:
self.name = devname
# Test that /dev/urandom is a character special device
f = open(self.name, "rb", 0)
fmode = os.fstat(f.fileno())[stat.ST_MODE]
if not stat.S_ISCHR(fmode):
f.close()
raise TypeError("%r is not a character special device" % (self.name,))
self.__file = f
BaseRNG.__init__(self)
def __init__(self, devname=None):
if devname is None:
self.name = "/dev/urandom"
else:
self.name = devname
# Test that /dev/urandom is a character special device
f = open(self.name, "rb", 0)
fmode = os.fstat(f.fileno())[stat.ST_MODE]
if not stat.S_ISCHR(fmode):
f.close()
raise TypeError("%r is not a character special device" % (self.name,))
self.__file = f
BaseRNG.__init__(self)
def __init__(self, devname=None):
if devname is None:
self.name = "/dev/urandom"
else:
self.name = devname
# Test that /dev/urandom is a character special device
f = open(self.name, "rb", 0)
fmode = os.fstat(f.fileno())[stat.ST_MODE]
if not stat.S_ISCHR(fmode):
f.close()
raise TypeError("%r is not a character special device" % (self.name,))
self.__file = f
BaseRNG.__init__(self)
def __init__(self, devname=None):
if devname is None:
self.name = "/dev/urandom"
else:
self.name = devname
# Test that /dev/urandom is a character special device
f = open(self.name, "rb", 0)
fmode = os.fstat(f.fileno())[stat.ST_MODE]
if not stat.S_ISCHR(fmode):
f.close()
raise TypeError("%r is not a character special device" % (self.name,))
self.__file = f
BaseRNG.__init__(self)
def test_write_full(self):
devfull = '/dev/full'
if not (os.path.exists(devfull) and
stat.S_ISCHR(os.stat(devfull).st_mode)):
# Issue #21934: OpenBSD does not have a /dev/full character device
self.skipTest('requires %r' % devfull)
with open(devfull, 'wb', 1) as f:
with self.assertRaises(IOError):
f.write('hello\n')
with open(devfull, 'wb', 1) as f:
with self.assertRaises(IOError):
# Issue #17976
f.write('hello')
f.write('\n')
with open(devfull, 'wb', 0) as f:
with self.assertRaises(IOError):
f.write('h')
def test_write_full(self):
devfull = '/dev/full'
if not (os.path.exists(devfull) and
stat.S_ISCHR(os.stat(devfull).st_mode)):
# Issue #21934: OpenBSD does not have a /dev/full character device
self.skipTest('requires %r' % devfull)
with open(devfull, 'wb', 1) as f:
with self.assertRaises(IOError):
f.write('hello\n')
with open(devfull, 'wb', 1) as f:
with self.assertRaises(IOError):
# Issue #17976
f.write('hello')
f.write('\n')
with open(devfull, 'wb', 0) as f:
with self.assertRaises(IOError):
f.write('h')
def __init__(self, devname=None):
if devname is None:
self.name = "/dev/urandom"
else:
self.name = devname
# Test that /dev/urandom is a character special device
f = open(self.name, "rb", 0)
fmode = os.fstat(f.fileno())[stat.ST_MODE]
if not stat.S_ISCHR(fmode):
f.close()
raise TypeError("%r is not a character special device" % (self.name,))
self.__file = f
BaseRNG.__init__(self)
def __init__(self, devname=None):
if devname is None:
self.name = "/dev/urandom"
else:
self.name = devname
# Test that /dev/urandom is a character special device
f = open(self.name, "rb", 0)
fmode = os.fstat(f.fileno())[stat.ST_MODE]
if not stat.S_ISCHR(fmode):
f.close()
raise TypeError("%r is not a character special device" % (self.name,))
self.__file = f
BaseRNG.__init__(self)
def __init__(self, devname=None):
if devname is None:
self.name = "/dev/urandom"
else:
self.name = devname
# Test that /dev/urandom is a character special device
f = open(self.name, "rb", 0)
fmode = os.fstat(f.fileno())[stat.ST_MODE]
if not stat.S_ISCHR(fmode):
f.close()
raise TypeError("%r is not a character special device" % (self.name,))
self.__file = f
BaseRNG.__init__(self)
def __init__(self, devname=None):
if devname is None:
self.name = "/dev/urandom"
else:
self.name = devname
# Test that /dev/urandom is a character special device
f = open(self.name, "rb", 0)
fmode = os.fstat(f.fileno())[stat.ST_MODE]
if not stat.S_ISCHR(fmode):
f.close()
raise TypeError("%r is not a character special device" % (self.name,))
self.__file = f
BaseRNG.__init__(self)
def __init__(self, devname=None):
if devname is None:
self.name = "/dev/urandom"
else:
self.name = devname
# Test that /dev/urandom is a character special device
f = open(self.name, "rb", 0)
fmode = os.fstat(f.fileno())[stat.ST_MODE]
if not stat.S_ISCHR(fmode):
f.close()
raise TypeError("%r is not a character special device" % (self.name,))
self.__file = f
BaseRNG.__init__(self)
def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
super().__init__(extra)
self._extra['pipe'] = pipe
self._loop = loop
self._pipe = pipe
self._fileno = pipe.fileno()
mode = os.fstat(self._fileno).st_mode
if not (stat.S_ISFIFO(mode) or
stat.S_ISSOCK(mode) or
stat.S_ISCHR(mode)):
raise ValueError("Pipe transport is for pipes/sockets only.")
_set_nonblocking(self._fileno)
self._protocol = protocol
self._closing = False
self._loop.call_soon(self._protocol.connection_made, self)
# only start reading when connection_made() has been called
self._loop.call_soon(self._loop.add_reader,
self._fileno, self._read_ready)
if waiter is not None:
# only wake up the waiter when connection_made() has been called
self._loop.call_soon(futures._set_result_unless_cancelled,
waiter, None)
def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
super().__init__(extra)
self._extra['pipe'] = pipe
self._loop = loop
self._pipe = pipe
self._fileno = pipe.fileno()
mode = os.fstat(self._fileno).st_mode
if not (stat.S_ISFIFO(mode) or
stat.S_ISSOCK(mode) or
stat.S_ISCHR(mode)):
raise ValueError("Pipe transport is for pipes/sockets only.")
_set_nonblocking(self._fileno)
self._protocol = protocol
self._closing = False
self._loop.add_reader(self._fileno, self._read_ready)
self._loop.call_soon(self._protocol.connection_made, self)
if waiter is not None:
# wait until protocol.connection_made() has been called
self._loop.call_soon(waiter._set_result_unless_cancelled, None)
def _ftypelet(mode):
if stat.S_ISREG(mode) or not stat.S_IFMT(mode):
return '-'
if stat.S_ISBLK(mode):
return 'b'
if stat.S_ISCHR(mode):
return 'c'
if stat.S_ISDIR(mode):
return 'd'
if stat.S_ISFIFO(mode):
return 'p'
if stat.S_ISLNK(mode):
return 'l'
if stat.S_ISSOCK(mode):
return 's'
return '?'
def __init__(self, devname=None):
if devname is None:
self.name = "/dev/urandom"
else:
self.name = devname
# Test that /dev/urandom is a character special device
f = open(self.name, "rb", 0)
fmode = os.fstat(f.fileno())[stat.ST_MODE]
if not stat.S_ISCHR(fmode):
f.close()
raise TypeError("%r is not a character special device" % (self.name,))
self.__file = f
BaseRNG.__init__(self)
def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
super().__init__(extra, loop)
self._extra['pipe'] = pipe
self._pipe = pipe
self._fileno = pipe.fileno()
mode = os.fstat(self._fileno).st_mode
is_socket = stat.S_ISSOCK(mode)
if not (is_socket or
stat.S_ISFIFO(mode) or
stat.S_ISCHR(mode)):
raise ValueError("Pipe transport is only for "
"pipes, sockets and character devices")
_set_nonblocking(self._fileno)
self._protocol = protocol
self._buffer = []
self._conn_lost = 0
self._closing = False # Set when close() or write_eof() called.
self._loop.call_soon(self._protocol.connection_made, self)
# On AIX, the reader trick (to be notified when the read end of the
# socket is closed) only works for sockets. On other platforms it
# works for pipes and sockets. (Exception: OS X 10.4? Issue #19294.)
if is_socket or not sys.platform.startswith("aix"):
# only start reading when connection_made() has been called
self._loop.call_soon(self._loop.add_reader,
self._fileno, self._read_ready)
if waiter is not None:
# only wake up the waiter when connection_made() has been called
self._loop.call_soon(waiter._set_result_unless_cancelled, None)
def is_special_file(cls, filename):
"""Checks to see if a file is a special UNIX file.
It checks if the file is a character special device, block special
device, FIFO, or socket.
:param filename: Name of the file
:returns: True if the file is a special file. False, if is not.
"""
# If it does not exist, it must be a new file so it cannot be
# a special file.
if not os.path.exists(filename):
return False
mode = os.stat(filename).st_mode
# Character special device.
if stat.S_ISCHR(mode):
return True
# Block special device
if stat.S_ISBLK(mode):
return True
# Named pipe / FIFO
if stat.S_ISFIFO(mode):
return True
# Socket.
if stat.S_ISSOCK(mode):
return True
return False
def ischr(self):
return stat.S_ISCHR(self.mode)
def isdev(self):
return (stat.S_ISCHR(self.mode) or stat.S_ISBLK(self.mode))
# class CpioInfo
def is_special_file(cls, filename):
"""Checks to see if a file is a special UNIX file.
It checks if the file is a character special device, block special
device, FIFO, or socket.
:param filename: Name of the file
:returns: True if the file is a special file. False, if is not.
"""
# If it does not exist, it must be a new file so it cannot be
# a special file.
if not os.path.exists(filename):
return False
mode = os.stat(filename).st_mode
# Character special device.
if stat.S_ISCHR(mode):
return True
# Block special device
if stat.S_ISBLK(mode):
return True
# Named pipe / FIFO
if stat.S_ISFIFO(mode):
return True
# Socket.
if stat.S_ISSOCK(mode):
return True
return False
def statinfo(st):
return {
'mode' : "%04o" % stat.S_IMODE(st.st_mode),
'isdir' : stat.S_ISDIR(st.st_mode),
'ischr' : stat.S_ISCHR(st.st_mode),
'isblk' : stat.S_ISBLK(st.st_mode),
'isreg' : stat.S_ISREG(st.st_mode),
'isfifo' : stat.S_ISFIFO(st.st_mode),
'islnk' : stat.S_ISLNK(st.st_mode),
'issock' : stat.S_ISSOCK(st.st_mode),
'uid' : st.st_uid,
'gid' : st.st_gid,
'size' : st.st_size,
'inode' : st.st_ino,
'dev' : st.st_dev,
'nlink' : st.st_nlink,
'atime' : st.st_atime,
'mtime' : st.st_mtime,
'ctime' : st.st_ctime,
'wusr' : bool(st.st_mode & stat.S_IWUSR),
'rusr' : bool(st.st_mode & stat.S_IRUSR),
'xusr' : bool(st.st_mode & stat.S_IXUSR),
'wgrp' : bool(st.st_mode & stat.S_IWGRP),
'rgrp' : bool(st.st_mode & stat.S_IRGRP),
'xgrp' : bool(st.st_mode & stat.S_IXGRP),
'woth' : bool(st.st_mode & stat.S_IWOTH),
'roth' : bool(st.st_mode & stat.S_IROTH),
'xoth' : bool(st.st_mode & stat.S_IXOTH),
'isuid' : bool(st.st_mode & stat.S_ISUID),
'isgid' : bool(st.st_mode & stat.S_ISGID),
}
def is_special_file(cls, filename):
"""Checks to see if a file is a special UNIX file.
It checks if the file is a character special device, block special
device, FIFO, or socket.
:param filename: Name of the file
:returns: True if the file is a special file. False, if is not.
"""
# If it does not exist, it must be a new file so it cannot be
# a special file.
if not os.path.exists(filename):
return False
mode = os.stat(filename).st_mode
# Character special device.
if stat.S_ISCHR(mode):
return True
# Block special device
if stat.S_ISBLK(mode):
return True
# Named pipe / FIFO
if stat.S_ISFIFO(mode):
return True
# Socket.
if stat.S_ISSOCK(mode):
return True
return False
def is_char_device(self):
"""
Whether this path is a character device.
"""
try:
return S_ISCHR(self.stat().st_mode)
except OSError as e:
if e.errno != ENOENT:
raise
# Path doesn't exist or is a broken symlink
# (see https://bitbucket.org/pitrou/pathlib/issue/12/)
return False
def open(self, target):
# target is a character device
if os.path.exists(target) \
and stat.S_ISCHR(os.stat(target).st_mode):
self._file = os.open(target, os.O_RDWR)
# treat target as ipv4 socket
else: self._sock.connect((target, 9100))
# close connection
