Use new form of with-statement instead of contextlib.nested().

This commit is contained in:
Raymond Hettinger 2009-06-04 00:11:54 +00:00
parent a678d94d58
commit 686057b8fa
4 changed files with 11 additions and 12 deletions

View File

@ -11,7 +11,6 @@ Functions:
import os import os
import stat import stat
import contextlib
from itertools import filterfalse from itertools import filterfalse
__all__ = ["cmp", "dircmp", "cmpfiles"] __all__ = ["cmp", "dircmp", "cmpfiles"]
@ -63,7 +62,7 @@ def _sig(st):
def _do_cmp(f1, f2): def _do_cmp(f1, f2):
bufsize = BUFSIZE bufsize = BUFSIZE
with contextlib.nested(open(f1, 'rb'), open(f2, 'rb')) as (fp1, fp2): with open(f1, 'rb') as fp1, open(f2, 'rb') as fp2:
while True: while True:
b1 = fp1.read(bufsize) b1 = fp1.read(bufsize)
b2 = fp2.read(bufsize) b2 = fp2.read(bufsize)

View File

@ -591,13 +591,11 @@ class TransientResource(object):
raise ResourceDenied("an optional resource is not available") raise ResourceDenied("an optional resource is not available")
def transient_internet(): # Context managers that raise ResourceDenied when various issues
"""Return a context manager that raises ResourceDenied when various issues # with the Internet connection manifest themselves as exceptions.
with the Internet connection manifest themselves as exceptions.""" time_out = TransientResource(IOError, errno=errno.ETIMEDOUT)
time_out = TransientResource(IOError, errno=errno.ETIMEDOUT) socket_peer_reset = TransientResource(socket.error, errno=errno.ECONNRESET)
socket_peer_reset = TransientResource(socket.error, errno=errno.ECONNRESET) ioerror_peer_reset = TransientResource(IOError, errno=errno.ECONNRESET)
ioerror_peer_reset = TransientResource(IOError, errno=errno.ECONNRESET)
return contextlib.nested(time_out, socket_peer_reset, ioerror_peer_reset)
@contextlib.contextmanager @contextlib.contextmanager

View File

@ -146,8 +146,8 @@ class InterProcessSignalTests(unittest.TestCase):
# re-raises information about any exceptions the child # re-raises information about any exceptions the child
# throws. The real work happens in self.run_test(). # throws. The real work happens in self.run_test().
os_done_r, os_done_w = os.pipe() os_done_r, os_done_w = os.pipe()
with nested(closing(os.fdopen(os_done_r, 'rb')), with closing(os.fdopen(os_done_r, 'rb')) as done_r, \
closing(os.fdopen(os_done_w, 'wb'))) as (done_r, done_w): closing(os.fdopen(os_done_w, 'wb')) as done_w:
child = os.fork() child = os.fork()
if child == 0: if child == 0:
# In the child process; run the test and report results # In the child process; run the test and report results

View File

@ -174,7 +174,9 @@ class OtherNetworkTests(unittest.TestCase):
(expected_err, url, req, type(err), err)) (expected_err, url, req, type(err), err))
self.assert_(isinstance(err, expected_err), msg) self.assert_(isinstance(err, expected_err), msg)
else: else:
with support.transient_internet(): with support.time_out, \
support.socket_peer_reset, \
support.ioerror_peer_reset:
buf = f.read() buf = f.read()
f.close() f.close()
debug("read %d bytes" % len(buf)) debug("read %d bytes" % len(buf))