Lib/test/test_gzip.py

   1 #! /usr/bin/env python
   2 """Test script for the gzip module.
   3 """
   4
   5 import unittest
   6 from test import test_support
   7 import os
   8 import gzip
   9 import struct
  10
  11
  12 data1 = """  int length=DEFAULTALLOC, err = Z_OK;
  13   PyObject *RetVal;
  14   int flushmode = Z_FINISH;
  15   unsigned long start_total_out;
  16
  17 """
  18
  19 data2 = """/* zlibmodule.c -- gzip-compatible data compression */
  20 /* See http://www.gzip.org/zlib/
  21 /* See http://www.winimage.com/zLibDll for Windows */
  22 """
  23
  24
  25 class TestGzip(unittest.TestCase):
  26     filename = test_support.TESTFN
  27
  28     def setUp(self):
  29         test_support.unlink(self.filename)
  30
  31     def tearDown(self):
  32         test_support.unlink(self.filename)
  33
  34
  35     def test_write(self):
  36         f = gzip.GzipFile(self.filename, 'wb') ; f.write(data1 * 50)
  37
  38         # Try flush and fileno.
  39         f.flush()
  40         f.fileno()
  41         if hasattr(os, 'fsync'):
  42             os.fsync(f.fileno())
  43         f.close()
  44
  45         # Test multiple close() calls.
  46         f.close()
  47
  48     def test_read(self):
  49         self.test_write()
  50         # Try reading.
  51         f = gzip.GzipFile(self.filename, 'r') ; d = f.read() ; f.close()
  52         self.assertEqual(d, data1*50)
  53
  54     def test_append(self):
  55         self.test_write()
  56         # Append to the previous file
  57         f = gzip.GzipFile(self.filename, 'ab') ; f.write(data2 * 15) ; f.close()
  58
  59         f = gzip.GzipFile(self.filename, 'rb') ; d = f.read() ; f.close()
  60         self.assertEqual(d, (data1*50) + (data2*15))
  61
  62     def test_many_append(self):
  63         # Bug #1074261 was triggered when reading a file that contained
  64         # many, many members.  Create such a file and verify that reading it
  65         # works.
  66         f = gzip.open(self.filename, 'wb', 9)
  67         f.write('a')
  68         f.close()
  69         for i in range(0,200):
  70             f = gzip.open(self.filename, "ab", 9) # append
  71             f.write('a')
  72             f.close()
  73
  74         # Try reading the file
  75         zgfile = gzip.open(self.filename, "rb")
  76         contents = ""
  77         while 1:
  78             ztxt = zgfile.read(8192)
  79             contents += ztxt
  80             if not ztxt: break
  81         zgfile.close()
  82         self.assertEquals(contents, 'a'*201)
  83
  84
  85     def test_readline(self):
  86         self.test_write()
  87         # Try .readline() with varying line lengths
  88
  89         f = gzip.GzipFile(self.filename, 'rb')
  90         line_length = 0
  91         while 1:
  92             L = f.readline(line_length)
  93             if L == "" and line_length != 0: break
  94             self.assert_(len(L) <= line_length)
  95             line_length = (line_length + 1) % 50
  96         f.close()
  97
  98     def test_readlines(self):
  99         self.test_write()
 100         # Try .readlines()
 101
 102         f = gzip.GzipFile(self.filename, 'rb')
 103         L = f.readlines()
 104         f.close()
 105
 106         f = gzip.GzipFile(self.filename, 'rb')
 107         while 1:
 108             L = f.readlines(150)
 109             if L == []: break
 110         f.close()
 111
 112     def test_seek_read(self):
 113         self.test_write()
 114         # Try seek, read test
 115
 116         f = gzip.GzipFile(self.filename)
 117         while 1:
 118             oldpos = f.tell()
 119             line1 = f.readline()
 120             if not line1: break
 121             newpos = f.tell()
 122             f.seek(oldpos)  # negative seek
 123             if len(line1)>10:
 124                 amount = 10
 125             else:
 126                 amount = len(line1)
 127             line2 = f.read(amount)
 128             self.assertEqual(line1[:amount], line2)
 129             f.seek(newpos)  # positive seek
 130         f.close()
 131
 132     def test_seek_whence(self):
 133         self.test_write()
 134         # Try seek(whence=1), read test
 135
 136         f = gzip.GzipFile(self.filename)
 137         f.read(10)
 138         f.seek(10, whence=1)
 139         y = f.read(10)
 140         f.close()
 141         self.assertEquals(y, data1[20:30])
 142
 143     def test_seek_write(self):
 144         # Try seek, write test
 145         f = gzip.GzipFile(self.filename, 'w')
 146         for pos in range(0, 256, 16):
 147             f.seek(pos)
 148             f.write('GZ\n')
 149         f.close()
 150
 151     def test_mode(self):
 152         self.test_write()
 153         f = gzip.GzipFile(self.filename, 'r')
 154         self.assertEqual(f.myfileobj.mode, 'rb')
 155         f.close()
 156
 157     def test_1647484(self):
 158         for mode in ('wb', 'rb'):
 159             f = gzip.GzipFile(self.filename, mode)
 160             self.assert_(hasattr(f, "name"))
 161             self.assertEqual(f.name, self.filename)
 162             f.close()
 163
 164     def test_mtime(self):
 165         mtime = 123456789
 166         fWrite = gzip.GzipFile(self.filename, 'w', mtime = mtime)
 167         fWrite.write(data1)
 168         fWrite.close()
 169         fRead = gzip.GzipFile(self.filename)
 170         dataRead = fRead.read()
 171         self.assertEqual(dataRead, data1)
 172         self.assert_(hasattr(fRead, 'mtime'))
 173         self.assertEqual(fRead.mtime, mtime)
 174         fRead.close()
 175
 176     def test_metadata(self):
 177         mtime = 123456789
 178
 179         fWrite = gzip.GzipFile(self.filename, 'w', mtime = mtime)
 180         fWrite.write(data1)
 181         fWrite.close()
 182
 183         fRead = open(self.filename, 'rb')
 184
 185         # see RFC 1952: http://www.faqs.org/rfcs/rfc1952.html
 186
 187         idBytes = fRead.read(2)
 188         self.assertEqual(idBytes, '\x1f\x8b') # gzip ID
 189
 190         cmByte = fRead.read(1)
 191         self.assertEqual(cmByte, '\x08') # deflate
 192
 193         flagsByte = fRead.read(1)
 194         self.assertEqual(flagsByte, '\x08') # only the FNAME flag is set
 195
 196         mtimeBytes = fRead.read(4)
 197         self.assertEqual(mtimeBytes, struct.pack('<i', mtime)) # little-endian
 198
 199         xflByte = fRead.read(1)
 200         self.assertEqual(xflByte, '\x02') # maximum compression
 201
 202         osByte = fRead.read(1)
 203         self.assertEqual(osByte, '\xff') # OS "unknown" (OS-independent)
 204
 205         # Since the FNAME flag is set, the zero-terminated filename follows.
 206         # RFC 1952 specifies that this is the name of the input file, if any.
 207         # However, the gzip module defaults to storing the name of the output
 208         # file in this field.
 209         nameBytes = fRead.read(len(self.filename) + 1)
 210         self.assertEqual(nameBytes, self.filename + '\x00')
 211
 212         # Since no other flags were set, the header ends here.
 213         # Rather than process the compressed data, let's seek to the trailer.
 214         fRead.seek(os.stat(self.filename).st_size - 8)
 215
 216         crc32Bytes = fRead.read(4) # CRC32 of uncompressed data [data1]
 217         self.assertEqual(crc32Bytes, '\xaf\xd7d\x83')
 218
 219         isizeBytes = fRead.read(4)
 220         self.assertEqual(isizeBytes, struct.pack('<i', len(data1)))
 221
 222         fRead.close()
 223
 224     def test_with_open(self):
 225         # GzipFile supports the context management protocol
 226         with gzip.GzipFile(self.filename, "wb") as f:
 227             f.write(b"xxx")
 228         f = gzip.GzipFile(self.filename, "rb")
 229         f.close()
 230         try:
 231             with f:
 232                 pass
 233         except ValueError:
 234             pass
 235         else:
 236             self.fail("__enter__ on a closed file didn't raise an exception")
 237         try:
 238             with gzip.GzipFile(self.filename, "wb") as f:
 239                 1/0
 240         except ZeroDivisionError:
 241             pass
 242         else:
 243             self.fail("1/0 didn't raise an exception")
 244
 245 def test_main(verbose=None):
 246     test_support.run_unittest(TestGzip)
 247
 248 if __name__ == "__main__":
 249     test_main(verbose=True)