def test_closerange(self):
first = os.open(support.TESTFN, os.O_CREAT|os.O_RDWR)
# We must allocate two consecutive file descriptors, otherwise
# it will mess up other file descriptors (perhaps even the three
# standard ones).
second = os.dup(first)
try:
retries = 0
while second != first + 1:
os.close(first)
retries += 1
if retries > 10:
# XXX test skipped
self.skipTest("couldn't allocate two consecutive fds")
first, second = second, os.dup(second)
finally:
os.close(second)
# close a fd that is open, and one that isn't
os.closerange(first, first + 2)
self.assertRaises(OSError, os.write, first, b"a")
python类closerange()的实例源码
def test_closerange(self):
first = os.open(test_support.TESTFN, os.O_CREAT|os.O_RDWR)
# We must allocate two consecutive file descriptors, otherwise
# it will mess up other file descriptors (perhaps even the three
# standard ones).
second = os.dup(first)
try:
retries = 0
while second != first + 1:
os.close(first)
retries += 1
if retries > 10:
# XXX test skipped
self.skipTest("couldn't allocate two consecutive fds")
first, second = second, os.dup(second)
finally:
os.close(second)
# close a fd that is open, and one that isn't
os.closerange(first, first + 2)
self.assertRaises(OSError, os.write, first, "a")
def test_closerange(self):
first = os.open(test_support.TESTFN, os.O_CREAT|os.O_RDWR)
# We must allocate two consecutive file descriptors, otherwise
# it will mess up other file descriptors (perhaps even the three
# standard ones).
second = os.dup(first)
try:
retries = 0
while second != first + 1:
os.close(first)
retries += 1
if retries > 10:
# XXX test skipped
self.skipTest("couldn't allocate two consecutive fds")
first, second = second, os.dup(second)
finally:
os.close(second)
# close a fd that is open, and one that isn't
os.closerange(first, first + 2)
self.assertRaises(OSError, os.write, first, "a")
def test_closerange(self):
first = os.open(support.TESTFN, os.O_CREAT|os.O_RDWR)
# We must allocate two consecutive file descriptors, otherwise
# it will mess up other file descriptors (perhaps even the three
# standard ones).
second = os.dup(first)
try:
retries = 0
while second != first + 1:
os.close(first)
retries += 1
if retries > 10:
# XXX test skipped
self.skipTest("couldn't allocate two consecutive fds")
first, second = second, os.dup(second)
finally:
os.close(second)
# close a fd that is open, and one that isn't
os.closerange(first, first + 2)
self.assertRaises(OSError, os.write, first, b"a")
def sane_execvp(filename, args, close_fds=True, restore_signals=True):
"""Execute a new program with sanitized environment.
"""
def _restore_signals():
"""Reset the default behavior to all signals.
"""
for i in _SIGNALS:
signal.signal(i, signal.SIG_DFL)
def _close_fds():
"""Close all file descriptors except 0, 1, 2.
"""
os.closerange(3, subprocess.MAXFD)
if close_fds:
_close_fds()
if restore_signals:
_restore_signals()
os.execvp(filename, args)
def test_closerange(self):
first = os.open(test_support.TESTFN, os.O_CREAT|os.O_RDWR)
# We must allocate two consecutive file descriptors, otherwise
# it will mess up other file descriptors (perhaps even the three
# standard ones).
second = os.dup(first)
try:
retries = 0
while second != first + 1:
os.close(first)
retries += 1
if retries > 10:
# XXX test skipped
self.skipTest("couldn't allocate two consecutive fds")
first, second = second, os.dup(second)
finally:
os.close(second)
# close a fd that is open, and one that isn't
os.closerange(first, first + 2)
self.assertRaises(OSError, os.write, first, "a")
def test_closerange(self):
first = os.open(support.TESTFN, os.O_CREAT|os.O_RDWR)
# We must allocate two consecutive file descriptors, otherwise
# it will mess up other file descriptors (perhaps even the three
# standard ones).
second = os.dup(first)
try:
retries = 0
while second != first + 1:
os.close(first)
retries += 1
if retries > 10:
# XXX test skipped
self.skipTest("couldn't allocate two consecutive fds")
first, second = second, os.dup(second)
finally:
os.close(second)
# close a fd that is open, and one that isn't
os.closerange(first, first + 2)
self.assertRaises(OSError, os.write, first, b"a")
def test_closerange(self):
first = os.open(test_support.TESTFN, os.O_CREAT|os.O_RDWR)
# We must allocate two consecutive file descriptors, otherwise
# it will mess up other file descriptors (perhaps even the three
# standard ones).
second = os.dup(first)
try:
retries = 0
while second != first + 1:
os.close(first)
retries += 1
if retries > 10:
# XXX test skipped
self.skipTest("couldn't allocate two consecutive fds")
first, second = second, os.dup(second)
finally:
os.close(second)
# close a fd that is open, and one that isn't
os.closerange(first, first + 2)
self.assertRaises(OSError, os.write, first, "a")
def test_closerange(self):
first = os.open(support.TESTFN, os.O_CREAT|os.O_RDWR)
# We must allocate two consecutive file descriptors, otherwise
# it will mess up other file descriptors (perhaps even the three
# standard ones).
second = os.dup(first)
try:
retries = 0
while second != first + 1:
os.close(first)
retries += 1
if retries > 10:
# XXX test skipped
self.skipTest("couldn't allocate two consecutive fds")
first, second = second, os.dup(second)
finally:
os.close(second)
# close a fd that is open, and one that isn't
os.closerange(first, first + 2)
self.assertRaises(OSError, os.write, first, b"a")
def _close_fds(self, but):
if hasattr(os, 'closerange'):
os.closerange(3, but)
os.closerange(but + 1, MAXFD)
else:
for i in xrange(3, MAXFD):
if i == but:
continue
try:
os.close(i)
except:
pass
def _run_child(self, cmd):
if isinstance(cmd, basestring):
cmd = ['/bin/sh', '-c', cmd]
os.closerange(3, MAXFD)
try:
os.execvp(cmd[0], cmd)
finally:
os._exit(1)
def exit():
"""
Causes python to exit without garbage-collecting any objects, and thus avoids
calling object destructor methods. This is a sledgehammer workaround for
a variety of bugs in PyQt and Pyside that cause crashes on exit.
This function does the following in an attempt to 'safely' terminate
the process:
* Invoke atexit callbacks
* Close all open file handles
* os._exit()
Note: there is some potential for causing damage with this function if you
are using objects that _require_ their destructors to be called (for example,
to properly terminate log files, disconnect from devices, etc). Situations
like this are probably quite rare, but use at your own risk.
"""
## first disable our own cleanup function; won't be needing it.
setConfigOptions(exitCleanup=False)
## invoke atexit callbacks
atexit._run_exitfuncs()
## close file handles
if sys.platform == 'darwin':
for fd in range(3, 4096):
if fd not in [7]: # trying to close 7 produces an illegal instruction on the Mac.
os.close(fd)
else:
os.closerange(3, 4096) ## just guessing on the maximum descriptor count..
os._exit(0)
## Convenience functions for command-line use
def exit():
"""
Causes python to exit without garbage-collecting any objects, and thus avoids
calling object destructor methods. This is a sledgehammer workaround for
a variety of bugs in PyQt and Pyside that cause crashes on exit.
This function does the following in an attempt to 'safely' terminate
the process:
* Invoke atexit callbacks
* Close all open file handles
* os._exit()
Note: there is some potential for causing damage with this function if you
are using objects that _require_ their destructors to be called (for example,
to properly terminate log files, disconnect from devices, etc). Situations
like this are probably quite rare, but use at your own risk.
"""
## first disable our own cleanup function; won't be needing it.
setConfigOptions(exitCleanup=False)
## invoke atexit callbacks
atexit._run_exitfuncs()
## close file handles
if sys.platform == 'darwin':
for fd in range(3, 4096):
if fd not in [7]: # trying to close 7 produces an illegal instruction on the Mac.
os.close(fd)
else:
os.closerange(3, 4096) ## just guessing on the maximum descriptor count..
os._exit(0)
## Convenience functions for command-line use
def _close_fds(self, but):
if hasattr(os, 'closerange'):
os.closerange(3, but)
os.closerange(but + 1, MAXFD)
else:
for i in xrange(3, MAXFD):
if i == but:
continue
try:
os.close(i)
except:
pass
def _run_child(self, cmd):
if isinstance(cmd, basestring):
cmd = ['/bin/sh', '-c', cmd]
os.closerange(3, MAXFD)
try:
os.execvp(cmd[0], cmd)
finally:
os._exit(1)
def _close_fds(self, but):
if hasattr(os, 'closerange'):
os.closerange(3, but)
os.closerange(but + 1, MAXFD)
else:
for i in xrange(3, MAXFD):
if i == but:
continue
try:
os.close(i)
except:
pass
def _close_fds(self, but):
if hasattr(os, 'closerange'):
os.closerange(3, but)
os.closerange(but + 1, MAXFD)
else:
for i in xrange(3, MAXFD):
if i == but:
continue
try:
os.close(i)
except:
pass
def _close_fds(self, but):
if hasattr(os, 'closerange'):
os.closerange(3, but)
os.closerange(but + 1, MAXFD)
else:
for i in xrange(3, MAXFD):
if i == but:
continue
try:
os.close(i)
except:
pass
def closerange(fd_low, fd_high):
# Iterate through and close all file descriptors.
for fd in range(fd_low, fd_high):
try:
os.close(fd)
except OSError: # ERROR, fd wasn't open to begin with (ignored)
pass
def _close_fds(self, but):
if hasattr(os, 'closerange'):
os.closerange(3, but)
os.closerange(but + 1, MAXFD)
else:
for i in xrange(3, MAXFD):
if i == but:
continue
try:
os.close(i)
except:
pass
def _close_fds(self, but):
if hasattr(os, 'closerange'):
os.closerange(3, but)
os.closerange(but + 1, MAXFD)
else:
for i in xrange(3, MAXFD):
if i == but:
continue
try:
os.close(i)
except:
pass
def _close_fds(self, but):
if hasattr(os, 'closerange'):
os.closerange(3, but)
os.closerange(but + 1, MAXFD)
else:
for i in xrange(3, MAXFD):
if i == but:
continue
try:
os.close(i)
except:
pass
def _close_fds(self, keep):
# `keep` is a set of fds, so we
# use os.closerange from 3 to min(keep)
# and then from max(keep + 1) to MAXFD and
# loop through filling in the gaps.
# Under new python versions, we need to explicitly set
# passed fds to be inheritable or they will go away on exec
if hasattr(os, 'set_inheritable'):
set_inheritable = os.set_inheritable
else:
set_inheritable = lambda i, v: True
if hasattr(os, 'closerange'):
keep = sorted(keep)
min_keep = min(keep)
max_keep = max(keep)
os.closerange(3, min_keep)
os.closerange(max_keep + 1, MAXFD)
for i in xrange(min_keep, max_keep):
if i in keep:
set_inheritable(i, True)
continue
try:
os.close(i)
except:
pass
else:
for i in xrange(3, MAXFD):
if i in keep:
set_inheritable(i, True)
continue
try:
os.close(i)
except:
pass
def _close_fds(self, keep):
# `keep` is a set of fds, so we
# use os.closerange from 3 to min(keep)
# and then from max(keep + 1) to MAXFD and
# loop through filling in the gaps.
# Under new python versions, we need to explicitly set
# passed fds to be inheritable or they will go away on exec
if hasattr(os, 'set_inheritable'):
set_inheritable = os.set_inheritable
else:
set_inheritable = lambda i, v: True
if hasattr(os, 'closerange'):
keep = sorted(keep)
min_keep = min(keep)
max_keep = max(keep)
os.closerange(3, min_keep)
os.closerange(max_keep + 1, MAXFD)
for i in xrange(min_keep, max_keep):
if i in keep:
set_inheritable(i, True)
continue
try:
os.close(i)
except:
pass
else:
for i in xrange(3, MAXFD):
if i in keep:
set_inheritable(i, True)
continue
try:
os.close(i)
except:
pass
def _close_fds(self, fds_to_keep):
start_fd = 3
for fd in sorted(fds_to_keep):
if fd >= start_fd:
os.closerange(start_fd, fd)
start_fd = fd + 1
if start_fd <= MAXFD:
os.closerange(start_fd, MAXFD)
def test_closerange(self):
if hasattr(os, "closerange"):
fd = support.make_bad_fd()
# Make sure none of the descriptors we are about to close are
# currently valid (issue 6542).
for i in range(10):
try: os.fstat(fd+i)
except OSError:
pass
else:
break
if i < 2:
raise unittest.SkipTest(
"Unable to acquire a range of invalid file descriptors")
self.assertEqual(os.closerange(fd, fd + i-1), None)
def closerange(fd_low, fd_high):
# Iterate through and close all file descriptors.
for fd in range(fd_low, fd_high):
try:
os.close(fd)
except OSError: # ERROR, fd wasn't open to begin with (ignored)
pass
def closerange(fd_low, fd_high):
# Iterate through and close all file descriptors.
for fd in range(fd_low, fd_high):
try:
os.close(fd)
except OSError: # ERROR, fd wasn't open to begin with (ignored)
pass
def closerange(fd_low, fd_high):
# Iterate through and close all file descriptors.
for fd in range(fd_low, fd_high):
try:
os.close(fd)
except OSError: # ERROR, fd wasn't open to begin with (ignored)
pass
def _close_fds(self, but):
if hasattr(os, 'closerange'):
os.closerange(3, but)
os.closerange(but + 1, MAXFD)
else:
for i in xrange(3, MAXFD):
if i == but:
continue
try:
os.close(i)
except:
pass
