ENH: Add DeterministicGzipFile wrapper class

Author: effigies

nibabel/openers.py

@@ -40,6 +40,21 @@
     HAVE_INDEXED_GZIP = False
 
 
+class DeterministicGzipFile(gzip.GzipFile):
+    """ Deterministic variant of GzipFile
+
+    This writer does not add filename information to the header, and defaults
+    to a modification time (``mtime``) of 0 seconds.
+    """
+    def __init__(self, filename=None, mode=None, compresslevel=9, fileobj=None, mtime=0):
+        if mode and 'b' not in mode:
+            mode += 'b'
+        if filename:
+            fileobj = self.myfileobj = open(filename, mode or 'rb')
+        return super().__init__(filename="", mode=mode, compresslevel=compresslevel,
+                                fileobj=fileobj, mtime=mtime)
+
+
 def _gzip_open(filename, mode='rb', compresslevel=9, keep_open=False):
 
     # use indexed_gzip if possible for faster read access.  If keep_open ==

nibabel/tests/test_openers.py

@@ -16,7 +16,7 @@
 import time
 
 from numpy.compat.py3k import asstr, asbytes
-from ..openers import Opener, ImageOpener, HAVE_INDEXED_GZIP, BZ2File
+from ..openers import Opener, ImageOpener, HAVE_INDEXED_GZIP, BZ2File, DeterministicGzipFile
 from ..tmpdirs import InTemporaryDirectory
 from ..volumeutils import BinOpener
 
@@ -356,6 +356,66 @@ def md5sum(fname):
         return hashlib.md5(fobj.read()).hexdigest()
 
 
+def test_DeterministicGzipFile():
+    with InTemporaryDirectory():
+        msg = b"Hello, I'd like to have an argument."
+
+        # No filename, no mtime
+        with open("ref.gz", "wb") as fobj:
+            with GzipFile(filename="", mode="wb", fileobj=fobj, mtime=0) as gzobj:
+                gzobj.write(msg)
+        anon_chksum = md5sum("ref.gz")
+
+        with DeterministicGzipFile("default.gz", "wb") as fobj:
+            internal_fobj = fobj.myfileobj
+            fobj.write(msg)
+        # Check that myfileobj is being closed by GzipFile.close()
+        # This is in case GzipFile changes its internal implementation
+        assert internal_fobj.closed
+
+        assert md5sum("default.gz") == anon_chksum
+
+        # No filename, current mtime
+        now = time.time()
+        with open("ref.gz", "wb") as fobj:
+            with GzipFile(filename="", mode="wb", fileobj=fobj, mtime=now) as gzobj:
+                gzobj.write(msg)
+        now_chksum = md5sum("ref.gz")
+
+        with DeterministicGzipFile("now.gz", "wb", mtime=now) as fobj:
+            fobj.write(msg)
+
+        assert md5sum("now.gz") == now_chksum
+
+        # Change in default behavior
+        with mock.patch("time.time") as t:
+            t.return_value = now
+
+            # GzipFile will use time.time()
+            with open("ref.gz", "wb") as fobj:
+                with GzipFile(filename="", mode="wb", fileobj=fobj) as gzobj:
+                    gzobj.write(msg)
+            assert md5sum("ref.gz") == now_chksum
+
+            # DeterministicGzipFile will use 0
+            with DeterministicGzipFile("now.gz", "wb") as fobj:
+                fobj.write(msg)
+            assert md5sum("now.gz") == anon_chksum
+
+        # GzipFile is filename dependent, DeterministicGzipFile is independent
+        with GzipFile("filenameA.gz", mode="wb", mtime=0) as gzobj:
+            gzobj.write(msg)
+        fnameA_chksum = md5sum("filenameA.gz")
+        assert fnameA_chksum != anon_chksum
+
+        with DeterministicGzipFile("filenameA.gz", "wb") as fobj:
+            fobj.write(msg)
+
+        # But the contents are the same with different filenames
+        assert md5sum("filenameA.gz") == anon_chksum
+
+
+
 def test_bitwise_determinism():
     with InTemporaryDirectory():
         msg = b"Hello, I'd like to have an argument."