bpo-30017: Allowed calling the close() method of the zip entry writer object (#1041) (#1092)
multiple times. Writing to closed zip entry writer object now always produce
a ValueError.
(cherry picked from commit 4c0d9ea995da595e90e08813b89510de59907802)
diff --git a/Lib/test/test_zipfile.py b/Lib/test/test_zipfile.py
index d39f05f..d09ad96 100644
--- a/Lib/test/test_zipfile.py
+++ b/Lib/test/test_zipfile.py
@@ -734,6 +734,48 @@
compression = zipfile.ZIP_LZMA
+class AbstractWriterTests:
+
+ def tearDown(self):
+ unlink(TESTFN2)
+
+ def test_close_after_close(self):
+ data = b'content'
+ with zipfile.ZipFile(TESTFN2, "w", self.compression) as zipf:
+ w = zipf.open('test', 'w')
+ w.write(data)
+ w.close()
+ self.assertTrue(w.closed)
+ w.close()
+ self.assertTrue(w.closed)
+ self.assertEqual(zipf.read('test'), data)
+
+ def test_write_after_close(self):
+ data = b'content'
+ with zipfile.ZipFile(TESTFN2, "w", self.compression) as zipf:
+ w = zipf.open('test', 'w')
+ w.write(data)
+ w.close()
+ self.assertTrue(w.closed)
+ self.assertRaises(ValueError, w.write, b'')
+ self.assertEqual(zipf.read('test'), data)
+
+class StoredWriterTests(AbstractWriterTests, unittest.TestCase):
+ compression = zipfile.ZIP_STORED
+
+@requires_zlib
+class DeflateWriterTests(AbstractWriterTests, unittest.TestCase):
+ compression = zipfile.ZIP_DEFLATED
+
+@requires_bz2
+class Bzip2WriterTests(AbstractWriterTests, unittest.TestCase):
+ compression = zipfile.ZIP_BZIP2
+
+@requires_lzma
+class LzmaWriterTests(AbstractWriterTests, unittest.TestCase):
+ compression = zipfile.ZIP_LZMA
+
+
class PyZipFileTests(unittest.TestCase):
def assertCompiledIn(self, name, namelist):
if name + 'o' not in namelist:
diff --git a/Lib/zipfile.py b/Lib/zipfile.py
index d7f5beb..06eedec 100644
--- a/Lib/zipfile.py
+++ b/Lib/zipfile.py
@@ -981,6 +981,8 @@
return True
def write(self, data):
+ if self.closed:
+ raise ValueError('I/O operation on closed file.')
nbytes = len(data)
self._file_size += nbytes
self._crc = crc32(data, self._crc)
@@ -991,6 +993,8 @@
return nbytes
def close(self):
+ if self.closed:
+ return
super().close()
# Flush any data from the compressor, and update header info
if self._compressor: