Fix several test issues

Author: rhpvorderman

src/isal/igzip.py

@@ -26,10 +26,9 @@
 import gzip
 import io
 import os
-
 import _compression
 import sys
-
+from gzip import READ, WRITE
 from . import isal_zlib
 
 __all__ = ["IGzipFile", "open", "compress", "decompress", "BadGzipFile"]
@@ -41,10 +40,12 @@
 
 BUFFER_SIZE = _compression.BUFFER_SIZE
 
-class BadGzipFile(OSError):
+try:
+    class BadGzipFile(gzip.BadGzipFile):
+        pass
+except AttributeError:  # Versions lower than 3.8 do not have BadGzipFile
     pass
 
-
 # The open method was copied from the python source with minor adjustments.
 def open(filename, mode="rb", compresslevel=_COMPRESS_LEVEL_TRADEOFF,
          encoding=None, errors=None, newline=None):
@@ -104,7 +105,7 @@ def __init__(self, filename=None, mode=None,
                     isal_zlib.ISAL_BEST_SPEED, isal_zlib.ISAL_BEST_COMPRESSION
                 ))
         super().__init__(filename, mode, compresslevel, fileobj, mtime)
-        if hasattr(self, "compress"):
+        if self.mode == gzip.WRITE:
             self.compress = isal_zlib.compressobj(compresslevel,
                                                   isal_zlib.DEFLATED,
                                                   -isal_zlib.MAX_WBITS,
@@ -169,6 +170,12 @@ def _read_eof(self):
         # Gzip files can be padded with zeroes and still have archives.
         # Consume all zero bytes and set the file position to the first
         # non-zero byte. See http://www.gzip.org/#faq8
+        # Also the isal_zlib.decompressobj does not consume the last two bytes
+        # when using ISAL_GZIP_NO_HDR.
+        for i in range(2):
+            c = self._fp.read(1)
+            if c == b"\x00":
+                break
         c = b"\x00"
         while c == b"\x00":
             c = self._fp.read(1)
@@ -187,12 +194,12 @@ def compress(data, compresslevel=_COMPRESS_LEVEL_BEST, *, mtime=None):
     return buf.getvalue()
 
 
-# Unlike stdlib, do not use the roundabout way of doing this via a file.
 def decompress(data):
     """Decompress a gzip compressed string in one shot.
     Return the decompressed string.
     """
-    return isal_zlib.decompress(data, wbits=16 + isal_zlib.MAX_WBITS)
+    with IGzipFile(fileobj=io.BytesIO(data)) as f:
+        return f.read()
 
 
 def main():

tests/test_gzip_compliance.py

@@ -391,13 +391,17 @@ def test_compresslevel_metadata(self):
         # see RFC 1952: http://www.faqs.org/rfcs/rfc1952.html
         # specifically, discussion of XFL in section 2.3.1
         cases = [
-            ('fast', 1, b'\x04'),
-            ('best', 9, b'\x02'),
-            ('tradeoff', 6, b'\x00'),
+            ('fast', 0, b'\x04'),
+            ('best', 3, b'\x02'),
+            ('tradeoff', 2, b'\x00'),
         ]
         xflOffset = 8
 
         for (name, level, expectedXflByte) in cases:
+            major, minor, _, _, _ = sys.version_info
+            if major == 3 and minor <=7 or major < 3:
+                # Specific xfl bytes introduced in 3.7
+                expectedXflByte = b'\x02'
             with self.subTest(name):
                 fWrite = igzip.IGzipFile(self.filename, 'w', compresslevel=level)
                 with fWrite:
@@ -440,14 +444,24 @@ def test_zero_padded_file(self):
             d = f.read()
             self.assertEqual(d, data1 * 50, "Incorrect data in file")
 
+    @unittest.skipIf(sys.version_info[0] == 3 and sys.version_info[1] < 8
+                     or sys.version_info[0] < 3,
+                     reason="BadGzipFile exception only in version 3.8 or "
+                            "higher")
     def test_igzip_BadGzipFile_exception(self):
         self.assertTrue(issubclass(igzip.BadGzipFile, OSError))
 
     def test_bad_gzip_file(self):
+        major, minor, _, _, _ = sys.version_info
+        if major == 3 and minor >= 8 or major > 3:
+            error = igzip.BadGzipFile
+        else:
+            error = OSError
+
         with open(self.filename, 'wb') as file:
             file.write(data1 * 50)
         with igzip.IGzipFile(self.filename, 'r') as file:
-            self.assertRaises(igzip.BadGzipFile, file.readlines)
+            self.assertRaises(error, file.readlines)
 
     def test_non_seekable_file(self):
         uncompressed = data1 * 50
@@ -518,7 +532,11 @@ def test_fileobj_mode(self):
             if "x" in mode:
                 os.unlink(self.filename)
             with open(self.filename, mode) as f:
-                with self.assertWarns(FutureWarning):
+                major, minor, _, _, _ = sys.version_info
+                if major == 3 and minor >= 9 or major > 3:
+                    with self.assertWarns(FutureWarning):
+                        g = igzip.IGzipFile(fileobj=f)
+                else:
                     g = igzip.IGzipFile(fileobj=f)
                 with g:
                     self.assertEqual(g.mode, igzip.WRITE)
@@ -562,7 +580,7 @@ def test_compress(self):
     def test_compress_mtime(self):
         mtime = 123456789
         for data in [data1, data2]:
-            for args in [(), (1,), (6,), (9,)]:
+            for args in [(), (0,), (1,), (2,), (3,)]:
                 with self.subTest(data=data, args=args):
                     datac = igzip.compress(data, *args, mtime=mtime)
                     self.assertEqual(type(datac), bytes)