diff -r 7757e98a9f3b Lib/gzip.py --- a/Lib/gzip.py Tue Mar 17 06:56:11 2015 +0200 +++ b/Lib/gzip.py Sun Mar 22 23:14:01 2015 +0100 @@ -334,17 +334,20 @@ if self.fileobj is None: raise ValueError("write() on closed GzipFile object") - # Convert data type if called by io.BufferedWriter. - if isinstance(data, memoryview): - data = data.tobytes() + if isinstance(data, bytes): + length = len(data) + else: + # accept any data that supports the buffer protocol + data = memoryview(data) + length = data.nbytes - if len(data) > 0: - self.size = self.size + len(data) + if length > 0: + self.fileobj.write(self.compress.compress(data)) + self.size += length self.crc = zlib.crc32(data, self.crc) & 0xffffffff - self.fileobj.write( self.compress.compress(data) ) - self.offset += len(data) + self.offset += length - return len(data) + return length def read(self, size=-1): self._check_closed() diff -r 7757e98a9f3b Lib/test/test_gzip.py --- a/Lib/test/test_gzip.py Tue Mar 17 06:56:11 2015 +0200 +++ b/Lib/test/test_gzip.py Sun Mar 22 23:14:01 2015 +0100 @@ -6,6 +6,7 @@ import os import io import struct +import array gzip = support.import_module('gzip') data1 = b""" int length=DEFAULTALLOC, err = Z_OK; @@ -43,6 +44,14 @@ class TestGzip(BaseTest): + def write_and_read_back(self, data, mode='b'): + b_data = bytes(data) + with gzip.GzipFile(self.filename, 'w'+mode) as f: + l = f.write(data) + self.assertEqual(l, len(b_data)) + with gzip.GzipFile(self.filename, 'r'+mode) as f: + self.assertEqual(f.read(), b_data) + def test_write(self): with gzip.GzipFile(self.filename, 'wb') as f: f.write(data1 * 50) @@ -57,13 +66,6 @@ # Test multiple close() calls. f.close() - def test_read(self): - self.test_write() - # Try reading. - with gzip.GzipFile(self.filename, 'r') as f: - d = f.read() - self.assertEqual(d, data1*50) - def test_read1(self): self.test_write() blocks = [] @@ -130,6 +132,39 @@ if not ztxt: break self.assertEqual(contents, b'a'*201) + # The following test_write_xy methods test that write accepts + # the corresponding bytes-like object type as input + # and that the data written equals bytes(xy) in all cases. + def test_write_memoryview(self): + data = memoryview(data1 * 50) + self.write_and_read_back(data) + m = memoryview(bytes(range(256))) + data = m.cast('B', shape=[8,8,4]) + self.write_and_read_back(data) + + def test_write_bytearray(self): + data = bytearray(data1 * 50) + self.write_and_read_back(data) + + def test_write_array(self): + for typecode in array.typecodes: + data = array.array(typecode, data1 * 80) + self.write_and_read_back(data) + + def test_write_incompatible_type(self): + # Test that non-bytes-like types raise TypeError. + for data in (data1.decode('utf-8'), list(data1)): + self.assertRaises(TypeError, self.write_and_read_back, data) + + def test_write_retry(self): + # Issue #21560: attempts to write incompatible types + # should not affect the state of the fileobject + data = data1.decode('utf-8') + try: + self.write_and_read_back(data * 50) + except TypeError: + self.write_and_read_back(data1 * 50) + def test_exclusive_write(self): with gzip.GzipFile(self.filename, 'xb') as f: f.write(data1 * 50)