diff options
Diffstat (limited to 'Lib')
-rw-r--r-- | Lib/gzip.py | 21 | ||||
-rw-r--r-- | Lib/test/test_gzip.py | 62 |
2 files changed, 79 insertions, 4 deletions
diff --git a/Lib/gzip.py b/Lib/gzip.py index 7a3f813..931c9ef 100644 --- a/Lib/gzip.py +++ b/Lib/gzip.py @@ -42,7 +42,7 @@ class GzipFile: max_read_chunk = 10 * 1024 * 1024 # 10Mb def __init__(self, filename=None, mode=None, - compresslevel=9, fileobj=None): + compresslevel=9, fileobj=None, mtime=None): """Constructor for the GzipFile class. At least one of fileobj and filename must be given a @@ -69,6 +69,15 @@ class GzipFile: level of compression; 1 is fastest and produces the least compression, and 9 is slowest and produces the most compression. The default is 9. + The mtime argument is an optional numeric timestamp to be written + to the stream when compressing. All gzip compressed streams + are required to contain a timestamp. If omitted or None, the + current time is used. This module ignores the timestamp when + decompressing; however, some programs, such as gunzip, make use + of it. The format of the timestamp is the same as that of the + return value of time.time() and of the st_mtime member of the + object returned by os.stat(). + """ # guarantee the file is opened in binary mode on platforms @@ -107,6 +116,7 @@ class GzipFile: self.fileobj = fileobj self.offset = 0 + self.mtime = mtime if self.mode == WRITE: self._write_gzip_header() @@ -140,7 +150,10 @@ class GzipFile: if fname: flags = FNAME self.fileobj.write(chr(flags)) - write32u(self.fileobj, long(time.time())) + mtime = self.mtime + if mtime is None: + mtime = time.time() + write32u(self.fileobj, long(mtime)) self.fileobj.write('\002') self.fileobj.write('\377') if fname: @@ -158,10 +171,10 @@ class GzipFile: if method != 8: raise IOError, 'Unknown compression method' flag = ord( self.fileobj.read(1) ) - # modtime = self.fileobj.read(4) + self.mtime = read32(self.fileobj) # extraflag = self.fileobj.read(1) # os = self.fileobj.read(1) - self.fileobj.read(6) + self.fileobj.read(2) if flag & FEXTRA: # Read & discard the extra field, if present diff --git a/Lib/test/test_gzip.py b/Lib/test/test_gzip.py index 105882a..25f9ee6 100644 --- a/Lib/test/test_gzip.py +++ b/Lib/test/test_gzip.py @@ -6,6 +6,7 @@ import unittest from test import test_support import os import gzip +import struct data1 = """ int length=DEFAULTALLOC, err = Z_OK; @@ -160,6 +161,67 @@ class TestGzip(unittest.TestCase): self.assertEqual(f.name, self.filename) f.close() + def test_mtime(self): + mtime = 123456789 + fWrite = gzip.GzipFile(self.filename, 'w', mtime = mtime) + fWrite.write(data1) + fWrite.close() + + fRead = gzip.GzipFile(self.filename) + dataRead = fRead.read() + self.assertEqual(dataRead, data1) + self.assert_(hasattr(fRead, 'mtime')) + self.assertEqual(fRead.mtime, mtime) + fRead.close() + + def test_metadata(self): + mtime = 123456789 + + fWrite = gzip.GzipFile(self.filename, 'w', mtime = mtime) + fWrite.write(data1) + fWrite.close() + + fRead = open(self.filename, 'rb') + + # see RFC 1952: http://www.faqs.org/rfcs/rfc1952.html + + idBytes = fRead.read(2) + self.assertEqual(idBytes, '\x1f\x8b') # gzip ID + + cmByte = fRead.read(1) + self.assertEqual(cmByte, '\x08') # deflate + + flagsByte = fRead.read(1) + self.assertEqual(flagsByte, '\x08') # only the FNAME flag is set + + mtimeBytes = fRead.read(4) + self.assertEqual(mtimeBytes, struct.pack('<i', mtime)) # little-endian + + xflByte = fRead.read(1) + self.assertEqual(xflByte, '\x02') # maximum compression + + osByte = fRead.read(1) + self.assertEqual(osByte, '\xff') # OS "unknown" (OS-independent) + + # Since the FNAME flag is set, the zero-terminated filename follows. + # 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. + nameBytes = fRead.read(len(self.filename) + 1) + self.assertEqual(nameBytes, self.filename + '\x00') + + # Since no other flags were set, the header ends here. + # Rather than process the compressed data, let's seek to the trailer. + fRead.seek(os.stat(self.filename).st_size - 8) + + crc32Bytes = fRead.read(4) # CRC32 of uncompressed data [data1] + self.assertEqual(crc32Bytes, '\xaf\xd7d\x83') + + isizeBytes = fRead.read(4) + self.assertEqual(isizeBytes, struct.pack('<i', len(data1))) + + fRead.close() + def test_main(verbose=None): test_support.run_unittest(TestGzip) |