Merge pull request RustPython#3757 from youknowone/update-gzip

update gzip and test_gzip from CPython 3.10.4

Author: youknowone

Lib/gzip.py

@@ -62,6 +62,7 @@ def open(filename, mode="rb", compresslevel=_COMPRESS_LEVEL_BEST,
         raise TypeError("filename must be a str or bytes object, or a file")
 
     if "t" in mode:
+        encoding = io.text_encoding(encoding)
         return io.TextIOWrapper(binary_file, encoding, errors, newline)
     else:
         return binary_file
@@ -277,7 +278,7 @@ def write(self,data):
         if self.fileobj is None:
             raise ValueError("write() on closed GzipFile object")
 
-        if isinstance(data, bytes):
+        if isinstance(data, (bytes, bytearray)):
             length = len(data)
         else:
             # accept any data that supports the buffer protocol
@@ -516,7 +517,7 @@ def _add_read_data(self, data):
 
     def _read_eof(self):
         # We've read to the end of the file
-        # We check the that the computed CRC and size of the
+        # We check that the computed CRC and size of the
         # uncompressed data matches the stored values.  Note that the size
         # stored is the true file size mod 2**32.
         crc32, isize = struct.unpack("<II", self._read_exact(8))
@@ -583,8 +584,7 @@ def main():
                 g = sys.stdout.buffer
             else:
                 if arg[-3:] != ".gz":
-                    print("filename doesn't end in .gz:", repr(arg))
-                    continue
+                    sys.exit(f"filename doesn't end in .gz: {arg!r}")
                 f = open(arg, "rb")
                 g = builtins.open(arg[:-3], "wb")
         else:
@@ -596,7 +596,7 @@ def main():
                 f = builtins.open(arg, "rb")
                 g = open(arg + ".gz", "wb")
         while True:
-            chunk = f.read(1024)
+            chunk = f.read(io.DEFAULT_BUFFER_SIZE)
             if not chunk:
                 break
             g.write(chunk)

Lib/test/test_gzip.py

@@ -10,8 +10,9 @@
 import sys
 import unittest
 from subprocess import PIPE, Popen
-from test import support
-from test.support import _4G, bigmemtest, os_helper, import_helper
+from test.support import import_helper
+from test.support import os_helper
+from test.support import _4G, bigmemtest
 from test.support.script_helper import assert_python_ok, assert_python_failure
 
 gzip = import_helper.import_module('gzip')
@@ -328,8 +329,15 @@ def test_metadata(self):
             cmByte = fRead.read(1)
             self.assertEqual(cmByte, b'\x08') # deflate
 
+            try:
+                expectedname = self.filename.encode('Latin-1') + b'\x00'
+                expectedflags = b'\x08' # only the FNAME flag is set
+            except UnicodeEncodeError:
+                expectedname = b''
+                expectedflags = b'\x00'
+
             flagsByte = fRead.read(1)
-            self.assertEqual(flagsByte, b'\x08') # only the FNAME flag is set
+            self.assertEqual(flagsByte, expectedflags)
 
             mtimeBytes = fRead.read(4)
             self.assertEqual(mtimeBytes, struct.pack('<i', mtime)) # little-endian
@@ -344,9 +352,8 @@ def test_metadata(self):
             # RFC 1952 specifies that this is the name of the input file, if any.
             # However, the gzip module defaults to storing the name of the output
             # file in this field.
-            expected = self.filename.encode('Latin-1') + b'\x00'
-            nameBytes = fRead.read(len(expected))
-            self.assertEqual(nameBytes, expected)
+            nameBytes = fRead.read(len(expectedname))
+            self.assertEqual(nameBytes, expectedname)
 
             # Since no other flags were set, the header ends here.
             # Rather than process the compressed data, let's seek to the trailer.
@@ -358,6 +365,10 @@ def test_metadata(self):
             isizeBytes = fRead.read(4)
             self.assertEqual(isizeBytes, struct.pack('<i', len(data1)))
 
+    def test_metadata_ascii_name(self):
+        self.filename = os_helper.TESTFN_ASCII
+        self.test_metadata()
+
     def test_compresslevel_metadata(self):
         # see RFC 1952: http://www.faqs.org/rfcs/rfc1952.html
         # specifically, discussion of XFL in section 2.3.1
@@ -489,7 +500,9 @@ def test_fileobj_mode(self):
             if "x" in mode:
                 os_helper.unlink(self.filename)
             with open(self.filename, mode) as f:
-                with gzip.GzipFile(fileobj=f) as g:
+                with self.assertWarns(FutureWarning):
+                    g = gzip.GzipFile(fileobj=f)
+                with g:
                     self.assertEqual(g.mode, gzip.WRITE)
 
     def test_bytes_filename(self):
@@ -578,6 +591,15 @@ def test_prepend_error(self):
         with gzip.open(self.filename, "rb") as f:
             f._buffer.raw._fp.prepend()
 
+    def test_issue44439(self):
+        q = array.array('Q', [1, 2, 3, 4, 5])
+        LENGTH = len(q) * q.itemsize
+
+        with gzip.GzipFile(fileobj=io.BytesIO(), mode='w') as f:
+            self.assertEqual(f.write(q), LENGTH)
+            self.assertEqual(f.tell(), LENGTH)
+
+
 class TestOpen(BaseTest):
     def test_binary_modes(self):
         uncompressed = data1 * 50
@@ -647,14 +669,14 @@ def test_implicit_binary_modes(self):
     def test_text_modes(self):
         uncompressed = data1.decode("ascii") * 50
         uncompressed_raw = uncompressed.replace("\n", os.linesep)
-        with gzip.open(self.filename, "wt") as f:
+        with gzip.open(self.filename, "wt", encoding="ascii") as f:
             f.write(uncompressed)
         with open(self.filename, "rb") as f:
             file_data = gzip.decompress(f.read()).decode("ascii")
             self.assertEqual(file_data, uncompressed_raw)
-        with gzip.open(self.filename, "rt") as f:
+        with gzip.open(self.filename, "rt", encoding="ascii") as f:
             self.assertEqual(f.read(), uncompressed)
-        with gzip.open(self.filename, "at") as f:
+        with gzip.open(self.filename, "at", encoding="ascii") as f:
             f.write(uncompressed)
         with open(self.filename, "rb") as f:
             file_data = gzip.decompress(f.read()).decode("ascii")
@@ -668,7 +690,7 @@ def test_fileobj(self):
             self.assertEqual(f.read(), uncompressed_bytes)
         with gzip.open(io.BytesIO(compressed), "rb") as f:
             self.assertEqual(f.read(), uncompressed_bytes)
-        with gzip.open(io.BytesIO(compressed), "rt") as f:
+        with gzip.open(io.BytesIO(compressed), "rt", encoding="ascii") as f:
             self.assertEqual(f.read(), uncompressed_str)
 
     def test_bad_params(self):
@@ -716,9 +738,9 @@ def test_encoding_error_handler(self):
     def test_newline(self):
         # Test with explicit newline (universal newline mode disabled).
         uncompressed = data1.decode("ascii") * 50
-        with gzip.open(self.filename, "wt", newline="\n") as f:
+        with gzip.open(self.filename, "wt", encoding="ascii", newline="\n") as f:
             f.write(uncompressed)
-        with gzip.open(self.filename, "rt", newline="\r") as f:
+        with gzip.open(self.filename, "rt", encoding="ascii", newline="\r") as f:
             self.assertEqual(f.readlines(), [uncompressed])
 
 
@@ -768,10 +790,10 @@ def test_decompress_infile_outfile(self):
         self.assertEqual(err, b'')
 
     def test_decompress_infile_outfile_error(self):
-        rc, out, err = assert_python_ok('-m', 'gzip', '-d', 'thisisatest.out')
-        self.assertIn(b"filename doesn't end in .gz:", out)
-        self.assertEqual(rc, 0)
-        self.assertEqual(err, b'')
+        rc, out, err = assert_python_failure('-m', 'gzip', '-d', 'thisisatest.out')
+        self.assertEqual(b"filename doesn't end in .gz: 'thisisatest.out'", err.strip())
+        self.assertEqual(rc, 1)
+        self.assertEqual(out, b'')
 
     @create_and_remove_directory(TEMPDIR)
     def test_compress_stdin_outfile(self):
@@ -827,9 +849,5 @@ def test_decompress_cannot_have_flags_compression(self):
         self.assertEqual(out, b'')
 
 
-def test_main(verbose=None):
-    support.run_unittest(TestGzip, TestOpen, TestCommandLine)
-
-
 if __name__ == "__main__":
-    test_main(verbose=True)
+    unittest.main()