# HG changeset patch # Parent 81c981ceb83ed7dc623213c23c4965219e26d240 Add read1() method to GzipFile. diff -r 81c981ceb83e Lib/gzip.py --- a/Lib/gzip.py Mon Apr 04 09:28:25 2011 -0700 +++ b/Lib/gzip.py Mon Apr 04 18:54:07 2011 +0200 @@ -348,6 +348,28 @@ self.offset += size return chunk + def read1(self, size=-1): + self._check_closed() + if self.mode != READ: + import errno + raise IOError(errno.EBADF, "read1() on write-only GzipFile object") + + if self.extrasize <= 0 and self.fileobj is None: + return b'' + + try: + self._read() + except EOFError: + pass + if size < 0 or size > self.extrasize: + size = self.extrasize + + offset = self.offset - self.extrastart + chunk = self.extrabuf[offset: offset + size] + self.extrasize -= size + self.offset += size + return chunk + def peek(self, n): if self.mode != READ: import errno diff -r 81c981ceb83e Lib/test/test_gzip.py --- a/Lib/test/test_gzip.py Mon Apr 04 09:28:25 2011 -0700 +++ b/Lib/test/test_gzip.py Mon Apr 04 18:54:07 2011 +0200 @@ -64,6 +64,21 @@ d = f.read() self.assertEqual(d, data1*50) + def test_read1(self): + self.test_write() + blocks = [] + nread = 0 + with gzip.GzipFile(self.filename, 'r') as f: + while True: + d = f.read1() + if not d: + break + blocks.append(d) + nread += len(d) + # Check that position was updated correctly (see issue10791). + self.assertEqual(f.tell(), nread) + self.assertEqual(b''.join(blocks), data1 * 50) + def test_io_on_closed_object(self): # Test that I/O operations on closed GzipFile objects raise a # ValueError, just like the corresponding functions on file objects. @@ -323,6 +338,14 @@ self.assertEqual(f.read(100), b'') self.assertEqual(nread, len(uncompressed)) + def test_textio_readlines(self): + # Issue #10791: TextIOWrapper.readlines() fails when wrapping GzipFile. + lines = (data1 * 50).decode("ascii").splitlines(True) + self.test_write() + with gzip.GzipFile(self.filename, 'r') as f: + with io.TextIOWrapper(f, encoding="ascii") as t: + self.assertEqual(t.readlines(), lines) + # Testing compress/decompress shortcut functions def test_compress(self):