update gzip and tests to 3.12.7

RustPython · arihant2math · Jan 16, 2025 · Jan 16, 2025 · Jan 16, 2025 · Jan 16, 2025
commit 3368e4e6559cccfae75f5ddacc13ba549aeabe9f
diff --git a/Lib/gzip.py b/Lib/gzip.py
@@ -15,8 +15,7 @@
 
 FTEXT, FHCRC, FEXTRA, FNAME, FCOMMENT = 1, 2, 4, 8, 16
 
-READ = 'rb'
-WRITE = 'wb'
+READ, WRITE = 1, 2
 
 _COMPRESS_LEVEL_FAST = 1
 _COMPRESS_LEVEL_TRADEOFF = 6
@@ -179,10 +178,9 @@ def __init__(self, filename=None, mode=None,
         and 9 is slowest and produces the most compression. 0 is no compression
         at all. The default is 9.
 
-        The optional mtime argument is the timestamp requested by gzip. The time
-        is in Unix format, i.e., seconds since 00:00:00 UTC, January 1, 1970.
-        If mtime is omitted or None, the current time is used. Use mtime = 0
-        to generate a compressed stream that does not depend on creation time.
+        The mtime argument is an optional numeric timestamp to be written
+        to the last modification time field in the stream when compressing.
+        If omitted or None, the current time is used.
 
         """
 
@@ -351,7 +349,7 @@ def closed(self):
 
     def close(self):
         fileobj = self.fileobj
-        if fileobj is None or self._buffer.closed:
+        if fileobj is None:
             return
         try:
             if self.mode == WRITE:
@@ -580,21 +578,43 @@ def _rewind(self):
         self._new_member = True
 
 
-def compress(data, compresslevel=_COMPRESS_LEVEL_BEST, *, mtime=0):
+def _create_simple_gzip_header(compresslevel: int,
+                               mtime = None) -> bytes:
+    """
+    Write a simple gzip header with no extra fields.
+    :param compresslevel: Compresslevel used to determine the xfl bytes.
+    :param mtime: The mtime (must support conversion to a 32-bit integer).
+    :return: A bytes object representing the gzip header.
+    """
+    if mtime is None:
+        mtime = time.time()
+    if compresslevel == _COMPRESS_LEVEL_BEST:
+        xfl = 2
+    elif compresslevel == _COMPRESS_LEVEL_FAST:
+        xfl = 4
+    else:
+        xfl = 0
+    # Pack ID1 and ID2 magic bytes, method (8=deflate), header flags (no extra
+    # fields added to header), mtime, xfl and os (255 for unknown OS).
+    return struct.pack("<BBBBLBB", 0x1f, 0x8b, 8, 0, int(mtime), xfl, 255)
+
+
+def compress(data, compresslevel=_COMPRESS_LEVEL_BEST, *, mtime=None):
     """Compress data in one shot and return the compressed string.
 
     compresslevel sets the compression level in range of 0-9.
-    mtime can be used to set the modification time.
-    The modification time is set to 0 by default, for reproducibility.
+    mtime can be used to set the modification time. The modification time is
+    set to the current time by default.
     """
-    # Wbits=31 automatically includes a gzip header and trailer.
-    gzip_data = zlib.compress(data, level=compresslevel, wbits=31)
-    if mtime is None:
-        mtime = time.time()
-    # Reuse gzip header created by zlib, replace mtime and OS byte for
-    # consistency.
-    header = struct.pack("<4sLBB", gzip_data, int(mtime), gzip_data[8], 255)
-    return header + gzip_data[10:]
+    if mtime == 0:
+        # Use zlib as it creates the header with 0 mtime by default.
+        # This is faster and with less overhead.
+        return zlib.compress(data, level=compresslevel, wbits=31)
+    header = _create_simple_gzip_header(compresslevel, mtime)
+    trailer = struct.pack("<LL", zlib.crc32(data), (len(data) & 0xffffffff))
+    # Wbits=-15 creates a raw deflate block.
+    return (header + zlib.compress(data, level=compresslevel, wbits=-15) +
+            trailer)
 
 
 def decompress(data):

diff --git a/Lib/test/test_gzip.py b/Lib/test/test_gzip.py
@@ -293,8 +293,6 @@ def test_mode(self):
         with gzip.GzipFile(self.filename, 'x') as f:
             self.assertEqual(f.myfileobj.mode, 'xb')
 
-    # TODO: RUSTPYTHON
-    @unittest.skip("TODO: RUSTPYTHON")
     def test_1647484(self):
         for mode in ('wb', 'rb'):
             with gzip.GzipFile(self.filename, mode) as f:
@@ -589,8 +587,6 @@ def test_fileobj_from_fdopen(self):
             self.assertRaises(AttributeError, f.fileno)
 
     def test_fileobj_mode(self):
-        self.assertEqual(gzip.READ, 'rb')
-        self.assertEqual(gzip.WRITE, 'wb')
         gzip.GzipFile(self.filename, "wb").close()
         with open(self.filename, "r+b") as f:
             with gzip.GzipFile(fileobj=f, mode='r') as g:
@@ -715,36 +711,15 @@ def test_compress_mtime(self):
                         f.read(1) # to set mtime attribute
                         self.assertEqual(f.mtime, mtime)
 
-    def test_compress_mtime_default(self):
-        # test for gh-125260
-        datac = gzip.compress(data1, mtime=0)
-        datac2 = gzip.compress(data1)
-        self.assertEqual(datac, datac2)
-        datac3 = gzip.compress(data1, mtime=None)
-        self.assertNotEqual(datac, datac3)
-        with gzip.GzipFile(fileobj=io.BytesIO(datac3), mode="rb") as f:
-            f.read(1) # to set mtime attribute
-            self.assertGreater(f.mtime, 1)
-
     def test_compress_correct_level(self):
+        # gzip.compress calls with mtime == 0 take a different code path.
         for mtime in (0, 42):
             with self.subTest(mtime=mtime):
                 nocompress = gzip.compress(data1, compresslevel=0, mtime=mtime)
                 yescompress = gzip.compress(data1, compresslevel=1, mtime=mtime)
                 self.assertIn(data1, nocompress)
                 self.assertNotIn(data1, yescompress)
 
-    def test_issue112346(self):
-        # The OS byte should be 255, this should not change between Python versions.
-        for mtime in (0, 42):
-            with self.subTest(mtime=mtime):
-                compress = gzip.compress(data1, compresslevel=1, mtime=mtime)
-                self.assertEqual(
-                    struct.unpack("<IxB", compress[4:10]),
-                    (mtime, 255),
-                    "Gzip header does not properly set either mtime or OS byte."
-                )
-
     def test_decompress(self):
         for data in (data1, data2):
             buf = io.BytesIO()