move sections
[python/dscho.git] / Lib / test / test_gzip.py
blob87131d0c7448d82d922ed0e90d09a5b48bc25b96
1 #! /usr/bin/env python
2 """Test script for the gzip module.
3 """
5 import unittest
6 from test import test_support
7 import os
8 import io
9 import struct
10 gzip = test_support.import_module('gzip')
12 data1 = """ int length=DEFAULTALLOC, err = Z_OK;
13 PyObject *RetVal;
14 int flushmode = Z_FINISH;
15 unsigned long start_total_out;
17 """
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 """
25 class TestGzip(unittest.TestCase):
26 filename = test_support.TESTFN
28 def setUp(self):
29 test_support.unlink(self.filename)
31 def tearDown(self):
32 test_support.unlink(self.filename)
35 def test_write(self):
36 f = gzip.GzipFile(self.filename, 'wb') ; f.write(data1 * 50)
38 # Try flush and fileno.
39 f.flush()
40 f.fileno()
41 if hasattr(os, 'fsync'):
42 os.fsync(f.fileno())
43 f.close()
45 # Test multiple close() calls.
46 f.close()
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)
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()
59 f = gzip.GzipFile(self.filename, 'rb') ; d = f.read() ; f.close()
60 self.assertEqual(d, (data1*50) + (data2*15))
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()
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)
84 def test_buffered_reader(self):
85 # Issue #7471: a GzipFile can be wrapped in a BufferedReader for
86 # performance.
87 self.test_write()
89 f = gzip.GzipFile(self.filename, 'rb')
90 with io.BufferedReader(f) as r:
91 lines = [line for line in r]
93 self.assertEqual(lines, 50 * data1.splitlines(True))
95 def test_readline(self):
96 self.test_write()
97 # Try .readline() with varying line lengths
99 f = gzip.GzipFile(self.filename, 'rb')
100 line_length = 0
101 while 1:
102 L = f.readline(line_length)
103 if L == "" and line_length != 0: break
104 self.assertTrue(len(L) <= line_length)
105 line_length = (line_length + 1) % 50
106 f.close()
108 def test_readlines(self):
109 self.test_write()
110 # Try .readlines()
112 f = gzip.GzipFile(self.filename, 'rb')
113 L = f.readlines()
114 f.close()
116 f = gzip.GzipFile(self.filename, 'rb')
117 while 1:
118 L = f.readlines(150)
119 if L == []: break
120 f.close()
122 def test_seek_read(self):
123 self.test_write()
124 # Try seek, read test
126 f = gzip.GzipFile(self.filename)
127 while 1:
128 oldpos = f.tell()
129 line1 = f.readline()
130 if not line1: break
131 newpos = f.tell()
132 f.seek(oldpos) # negative seek
133 if len(line1)>10:
134 amount = 10
135 else:
136 amount = len(line1)
137 line2 = f.read(amount)
138 self.assertEqual(line1[:amount], line2)
139 f.seek(newpos) # positive seek
140 f.close()
142 def test_seek_whence(self):
143 self.test_write()
144 # Try seek(whence=1), read test
146 f = gzip.GzipFile(self.filename)
147 f.read(10)
148 f.seek(10, whence=1)
149 y = f.read(10)
150 f.close()
151 self.assertEquals(y, data1[20:30])
153 def test_seek_write(self):
154 # Try seek, write test
155 f = gzip.GzipFile(self.filename, 'w')
156 for pos in range(0, 256, 16):
157 f.seek(pos)
158 f.write('GZ\n')
159 f.close()
161 def test_mode(self):
162 self.test_write()
163 f = gzip.GzipFile(self.filename, 'r')
164 self.assertEqual(f.myfileobj.mode, 'rb')
165 f.close()
167 def test_1647484(self):
168 for mode in ('wb', 'rb'):
169 f = gzip.GzipFile(self.filename, mode)
170 self.assertTrue(hasattr(f, "name"))
171 self.assertEqual(f.name, self.filename)
172 f.close()
174 def test_mtime(self):
175 mtime = 123456789
176 fWrite = gzip.GzipFile(self.filename, 'w', mtime = mtime)
177 fWrite.write(data1)
178 fWrite.close()
179 fRead = gzip.GzipFile(self.filename)
180 dataRead = fRead.read()
181 self.assertEqual(dataRead, data1)
182 self.assertTrue(hasattr(fRead, 'mtime'))
183 self.assertEqual(fRead.mtime, mtime)
184 fRead.close()
186 def test_metadata(self):
187 mtime = 123456789
189 fWrite = gzip.GzipFile(self.filename, 'w', mtime = mtime)
190 fWrite.write(data1)
191 fWrite.close()
193 fRead = open(self.filename, 'rb')
195 # see RFC 1952: http://www.faqs.org/rfcs/rfc1952.html
197 idBytes = fRead.read(2)
198 self.assertEqual(idBytes, '\x1f\x8b') # gzip ID
200 cmByte = fRead.read(1)
201 self.assertEqual(cmByte, '\x08') # deflate
203 flagsByte = fRead.read(1)
204 self.assertEqual(flagsByte, '\x08') # only the FNAME flag is set
206 mtimeBytes = fRead.read(4)
207 self.assertEqual(mtimeBytes, struct.pack('<i', mtime)) # little-endian
209 xflByte = fRead.read(1)
210 self.assertEqual(xflByte, '\x02') # maximum compression
212 osByte = fRead.read(1)
213 self.assertEqual(osByte, '\xff') # OS "unknown" (OS-independent)
215 # Since the FNAME flag is set, the zero-terminated filename follows.
216 # RFC 1952 specifies that this is the name of the input file, if any.
217 # However, the gzip module defaults to storing the name of the output
218 # file in this field.
219 nameBytes = fRead.read(len(self.filename) + 1)
220 self.assertEqual(nameBytes, self.filename + '\x00')
222 # Since no other flags were set, the header ends here.
223 # Rather than process the compressed data, let's seek to the trailer.
224 fRead.seek(os.stat(self.filename).st_size - 8)
226 crc32Bytes = fRead.read(4) # CRC32 of uncompressed data [data1]
227 self.assertEqual(crc32Bytes, '\xaf\xd7d\x83')
229 isizeBytes = fRead.read(4)
230 self.assertEqual(isizeBytes, struct.pack('<i', len(data1)))
232 fRead.close()
234 def test_with_open(self):
235 # GzipFile supports the context management protocol
236 with gzip.GzipFile(self.filename, "wb") as f:
237 f.write(b"xxx")
238 f = gzip.GzipFile(self.filename, "rb")
239 f.close()
240 try:
241 with f:
242 pass
243 except ValueError:
244 pass
245 else:
246 self.fail("__enter__ on a closed file didn't raise an exception")
247 try:
248 with gzip.GzipFile(self.filename, "wb") as f:
249 1 // 0
250 except ZeroDivisionError:
251 pass
252 else:
253 self.fail("1 // 0 didn't raise an exception")
255 def test_zero_padded_file(self):
256 with gzip.GzipFile(self.filename, "wb") as f:
257 f.write(data1 * 50)
259 # Pad the file with zeroes
260 with open(self.filename, "ab") as f:
261 f.write("\x00" * 50)
263 with gzip.GzipFile(self.filename, "rb") as f:
264 d = f.read()
265 self.assertEqual(d, data1 * 50, "Incorrect data in file")
267 def test_main(verbose=None):
268 test_support.run_unittest(TestGzip)
270 if __name__ == "__main__":
271 test_main(verbose=True)