Merged revisions 68484-68485 via svnmerge from
svn+ssh://pythondev@svn.python.org/python/trunk
........
r68484 | antoine.pitrou | 2009-01-10 17:13:45 +0100 (sam., 10 janv. 2009) | 3 lines
Issue #3860: GzipFile and BZ2File now support the context manager protocol.
........
r68485 | antoine.pitrou | 2009-01-10 17:15:24 +0100 (sam., 10 janv. 2009) | 1 line
Add NEWS entry for r68484.
........
diff --git a/Lib/gzip.py b/Lib/gzip.py
index 560a722..019c3e2 100644
--- a/Lib/gzip.py
+++ b/Lib/gzip.py
@@ -472,6 +472,14 @@
else:
raise StopIteration
+ def __enter__(self):
+ if self.fileobj is None:
+ raise ValueError("I/O operation on closed GzipFile object")
+ return self
+
+ def __exit__(self, *args):
+ self.close()
+
def _test():
# Act like gzip; with -d, act like gunzip.
diff --git a/Lib/test/test_bz2.py b/Lib/test/test_bz2.py
index 366ab7a..18d84bb 100644
--- a/Lib/test/test_bz2.py
+++ b/Lib/test/test_bz2.py
@@ -258,6 +258,28 @@
bz2f.close()
self.assertEqual(xlines, [b'Test'])
+ def testContextProtocol(self):
+ # BZ2File supports the context management protocol
+ f = None
+ with BZ2File(self.filename, "wb") as f:
+ f.write(b"xxx")
+ f = BZ2File(self.filename, "rb")
+ f.close()
+ try:
+ with f:
+ pass
+ except ValueError:
+ pass
+ else:
+ self.fail("__enter__ on a closed file didn't raise an exception")
+ try:
+ with BZ2File(self.filename, "wb") as f:
+ 1/0
+ except ZeroDivisionError:
+ pass
+ else:
+ self.fail("1/0 didn't raise an exception")
+
class BZ2CompressorTest(BaseTest):
def testCompress(self):
diff --git a/Lib/test/test_gzip.py b/Lib/test/test_gzip.py
index e758826..baf3a21 100644
--- a/Lib/test/test_gzip.py
+++ b/Lib/test/test_gzip.py
@@ -166,7 +166,6 @@
fWrite = gzip.GzipFile(self.filename, 'w', mtime = mtime)
fWrite.write(data1)
fWrite.close()
-
fRead = gzip.GzipFile(self.filename)
dataRead = fRead.read()
self.assertEqual(dataRead, data1)
@@ -223,6 +222,27 @@
fRead.close()
+ def test_with_open(self):
+ # GzipFile supports the context management protocol
+ with gzip.GzipFile(self.filename, "wb") as f:
+ f.write(b"xxx")
+ f = gzip.GzipFile(self.filename, "rb")
+ f.close()
+ try:
+ with f:
+ pass
+ except ValueError:
+ pass
+ else:
+ self.fail("__enter__ on a closed file didn't raise an exception")
+ try:
+ with gzip.GzipFile(self.filename, "wb") as f:
+ 1/0
+ except ZeroDivisionError:
+ pass
+ else:
+ self.fail("1/0 didn't raise an exception")
+
def test_main(verbose=None):
support.run_unittest(TestGzip)