source: trunk/essentials/dev-lang/python/Lib/test/test_zlib.py@ 3951

Last change on this file since 3951 was 3225, checked in by bird, 19 years ago

Python 2.5

File size: 16.0 KB
Line 
1import unittest
2from test import test_support
3import zlib
4import random
5
6# print test_support.TESTFN
7
8def getbuf():
9 # This was in the original. Avoid non-repeatable sources.
10 # Left here (unused) in case something wants to be done with it.
11 import imp
12 try:
13 t = imp.find_module('test_zlib')
14 file = t[0]
15 except ImportError:
16 file = open(__file__)
17 buf = file.read() * 8
18 file.close()
19 return buf
20
21
22
23class ChecksumTestCase(unittest.TestCase):
24 # checksum test cases
25 def test_crc32start(self):
26 self.assertEqual(zlib.crc32(""), zlib.crc32("", 0))
27 self.assert_(zlib.crc32("abc", 0xffffffff))
28
29 def test_crc32empty(self):
30 self.assertEqual(zlib.crc32("", 0), 0)
31 self.assertEqual(zlib.crc32("", 1), 1)
32 self.assertEqual(zlib.crc32("", 432), 432)
33
34 def test_adler32start(self):
35 self.assertEqual(zlib.adler32(""), zlib.adler32("", 1))
36 self.assert_(zlib.adler32("abc", 0xffffffff))
37
38 def test_adler32empty(self):
39 self.assertEqual(zlib.adler32("", 0), 0)
40 self.assertEqual(zlib.adler32("", 1), 1)
41 self.assertEqual(zlib.adler32("", 432), 432)
42
43 def assertEqual32(self, seen, expected):
44 # 32-bit values masked -- checksums on 32- vs 64- bit machines
45 # This is important if bit 31 (0x08000000L) is set.
46 self.assertEqual(seen & 0x0FFFFFFFFL, expected & 0x0FFFFFFFFL)
47
48 def test_penguins(self):
49 self.assertEqual32(zlib.crc32("penguin", 0), 0x0e5c1a120L)
50 self.assertEqual32(zlib.crc32("penguin", 1), 0x43b6aa94)
51 self.assertEqual32(zlib.adler32("penguin", 0), 0x0bcf02f6)
52 self.assertEqual32(zlib.adler32("penguin", 1), 0x0bd602f7)
53
54 self.assertEqual(zlib.crc32("penguin"), zlib.crc32("penguin", 0))
55 self.assertEqual(zlib.adler32("penguin"),zlib.adler32("penguin",1))
56
57
58
59class ExceptionTestCase(unittest.TestCase):
60 # make sure we generate some expected errors
61 def test_bigbits(self):
62 # specifying total bits too large causes an error
63 self.assertRaises(zlib.error,
64 zlib.compress, 'ERROR', zlib.MAX_WBITS + 1)
65
66 def test_badcompressobj(self):
67 # verify failure on building compress object with bad params
68 self.assertRaises(ValueError, zlib.compressobj, 1, zlib.DEFLATED, 0)
69
70 def test_baddecompressobj(self):
71 # verify failure on building decompress object with bad params
72 self.assertRaises(ValueError, zlib.decompressobj, 0)
73
74
75
76class CompressTestCase(unittest.TestCase):
77 # Test compression in one go (whole message compression)
78 def test_speech(self):
79 x = zlib.compress(HAMLET_SCENE)
80 self.assertEqual(zlib.decompress(x), HAMLET_SCENE)
81
82 def test_speech128(self):
83 # compress more data
84 data = HAMLET_SCENE * 128
85 x = zlib.compress(data)
86 self.assertEqual(zlib.decompress(x), data)
87
88
89
90
91class CompressObjectTestCase(unittest.TestCase):
92 # Test compression object
93 def test_pair(self):
94 # straightforward compress/decompress objects
95 data = HAMLET_SCENE * 128
96 co = zlib.compressobj()
97 x1 = co.compress(data)
98 x2 = co.flush()
99 self.assertRaises(zlib.error, co.flush) # second flush should not work
100 dco = zlib.decompressobj()
101 y1 = dco.decompress(x1 + x2)
102 y2 = dco.flush()
103 self.assertEqual(data, y1 + y2)
104
105 def test_compressoptions(self):
106 # specify lots of options to compressobj()
107 level = 2
108 method = zlib.DEFLATED
109 wbits = -12
110 memlevel = 9
111 strategy = zlib.Z_FILTERED
112 co = zlib.compressobj(level, method, wbits, memlevel, strategy)
113 x1 = co.compress(HAMLET_SCENE)
114 x2 = co.flush()
115 dco = zlib.decompressobj(wbits)
116 y1 = dco.decompress(x1 + x2)
117 y2 = dco.flush()
118 self.assertEqual(HAMLET_SCENE, y1 + y2)
119
120 def test_compressincremental(self):
121 # compress object in steps, decompress object as one-shot
122 data = HAMLET_SCENE * 128
123 co = zlib.compressobj()
124 bufs = []
125 for i in range(0, len(data), 256):
126 bufs.append(co.compress(data[i:i+256]))
127 bufs.append(co.flush())
128 combuf = ''.join(bufs)
129
130 dco = zlib.decompressobj()
131 y1 = dco.decompress(''.join(bufs))
132 y2 = dco.flush()
133 self.assertEqual(data, y1 + y2)
134
135 def test_decompinc(self, flush=False, source=None, cx=256, dcx=64):
136 # compress object in steps, decompress object in steps
137 source = source or HAMLET_SCENE
138 data = source * 128
139 co = zlib.compressobj()
140 bufs = []
141 for i in range(0, len(data), cx):
142 bufs.append(co.compress(data[i:i+cx]))
143 bufs.append(co.flush())
144 combuf = ''.join(bufs)
145
146 self.assertEqual(data, zlib.decompress(combuf))
147
148 dco = zlib.decompressobj()
149 bufs = []
150 for i in range(0, len(combuf), dcx):
151 bufs.append(dco.decompress(combuf[i:i+dcx]))
152 self.assertEqual('', dco.unconsumed_tail, ########
153 "(A) uct should be '': not %d long" %
154 len(dco.unconsumed_tail))
155 if flush:
156 bufs.append(dco.flush())
157 else:
158 while True:
159 chunk = dco.decompress('')
160 if chunk:
161 bufs.append(chunk)
162 else:
163 break
164 self.assertEqual('', dco.unconsumed_tail, ########
165 "(B) uct should be '': not %d long" %
166 len(dco.unconsumed_tail))
167 self.assertEqual(data, ''.join(bufs))
168 # Failure means: "decompressobj with init options failed"
169
170 def test_decompincflush(self):
171 self.test_decompinc(flush=True)
172
173 def test_decompimax(self, source=None, cx=256, dcx=64):
174 # compress in steps, decompress in length-restricted steps
175 source = source or HAMLET_SCENE
176 # Check a decompression object with max_length specified
177 data = source * 128
178 co = zlib.compressobj()
179 bufs = []
180 for i in range(0, len(data), cx):
181 bufs.append(co.compress(data[i:i+cx]))
182 bufs.append(co.flush())
183 combuf = ''.join(bufs)
184 self.assertEqual(data, zlib.decompress(combuf),
185 'compressed data failure')
186
187 dco = zlib.decompressobj()
188 bufs = []
189 cb = combuf
190 while cb:
191 #max_length = 1 + len(cb)//10
192 chunk = dco.decompress(cb, dcx)
193 self.failIf(len(chunk) > dcx,
194 'chunk too big (%d>%d)' % (len(chunk), dcx))
195 bufs.append(chunk)
196 cb = dco.unconsumed_tail
197 bufs.append(dco.flush())
198 self.assertEqual(data, ''.join(bufs), 'Wrong data retrieved')
199
200 def test_decompressmaxlen(self, flush=False):
201 # Check a decompression object with max_length specified
202 data = HAMLET_SCENE * 128
203 co = zlib.compressobj()
204 bufs = []
205 for i in range(0, len(data), 256):
206 bufs.append(co.compress(data[i:i+256]))
207 bufs.append(co.flush())
208 combuf = ''.join(bufs)
209 self.assertEqual(data, zlib.decompress(combuf),
210 'compressed data failure')
211
212 dco = zlib.decompressobj()
213 bufs = []
214 cb = combuf
215 while cb:
216 max_length = 1 + len(cb)//10
217 chunk = dco.decompress(cb, max_length)
218 self.failIf(len(chunk) > max_length,
219 'chunk too big (%d>%d)' % (len(chunk),max_length))
220 bufs.append(chunk)
221 cb = dco.unconsumed_tail
222 if flush:
223 bufs.append(dco.flush())
224 else:
225 while chunk:
226 chunk = dco.decompress('', max_length)
227 self.failIf(len(chunk) > max_length,
228 'chunk too big (%d>%d)' % (len(chunk),max_length))
229 bufs.append(chunk)
230 self.assertEqual(data, ''.join(bufs), 'Wrong data retrieved')
231
232 def test_decompressmaxlenflush(self):
233 self.test_decompressmaxlen(flush=True)
234
235 def test_maxlenmisc(self):
236 # Misc tests of max_length
237 dco = zlib.decompressobj()
238 self.assertRaises(ValueError, dco.decompress, "", -1)
239 self.assertEqual('', dco.unconsumed_tail)
240
241 def test_flushes(self):
242 # Test flush() with the various options, using all the
243 # different levels in order to provide more variations.
244 sync_opt = ['Z_NO_FLUSH', 'Z_SYNC_FLUSH', 'Z_FULL_FLUSH']
245 sync_opt = [getattr(zlib, opt) for opt in sync_opt
246 if hasattr(zlib, opt)]
247 data = HAMLET_SCENE * 8
248
249 for sync in sync_opt:
250 for level in range(10):
251 obj = zlib.compressobj( level )
252 a = obj.compress( data[:3000] )
253 b = obj.flush( sync )
254 c = obj.compress( data[3000:] )
255 d = obj.flush()
256 self.assertEqual(zlib.decompress(''.join([a,b,c,d])),
257 data, ("Decompress failed: flush "
258 "mode=%i, level=%i") % (sync, level))
259 del obj
260
261 def test_odd_flush(self):
262 # Test for odd flushing bugs noted in 2.0, and hopefully fixed in 2.1
263 import random
264
265 if hasattr(zlib, 'Z_SYNC_FLUSH'):
266 # Testing on 17K of "random" data
267
268 # Create compressor and decompressor objects
269 co = zlib.compressobj(zlib.Z_BEST_COMPRESSION)
270 dco = zlib.decompressobj()
271
272 # Try 17K of data
273 # generate random data stream
274 try:
275 # In 2.3 and later, WichmannHill is the RNG of the bug report
276 gen = random.WichmannHill()
277 except AttributeError:
278 try:
279 # 2.2 called it Random
280 gen = random.Random()
281 except AttributeError:
282 # others might simply have a single RNG
283 gen = random
284 gen.seed(1)
285 data = genblock(1, 17 * 1024, generator=gen)
286
287 # compress, sync-flush, and decompress
288 first = co.compress(data)
289 second = co.flush(zlib.Z_SYNC_FLUSH)
290 expanded = dco.decompress(first + second)
291
292 # if decompressed data is different from the input data, choke.
293 self.assertEqual(expanded, data, "17K random source doesn't match")
294
295 def test_empty_flush(self):
296 # Test that calling .flush() on unused objects works.
297 # (Bug #1083110 -- calling .flush() on decompress objects
298 # caused a core dump.)
299
300 co = zlib.compressobj(zlib.Z_BEST_COMPRESSION)
301 self.failUnless(co.flush()) # Returns a zlib header
302 dco = zlib.decompressobj()
303 self.assertEqual(dco.flush(), "") # Returns nothing
304
305 if hasattr(zlib.compressobj(), "copy"):
306 def test_compresscopy(self):
307 # Test copying a compression object
308 data0 = HAMLET_SCENE
309 data1 = HAMLET_SCENE.swapcase()
310 c0 = zlib.compressobj(zlib.Z_BEST_COMPRESSION)
311 bufs0 = []
312 bufs0.append(c0.compress(data0))
313
314 c1 = c0.copy()
315 bufs1 = bufs0[:]
316
317 bufs0.append(c0.compress(data0))
318 bufs0.append(c0.flush())
319 s0 = ''.join(bufs0)
320
321 bufs1.append(c1.compress(data1))
322 bufs1.append(c1.flush())
323 s1 = ''.join(bufs1)
324
325 self.assertEqual(zlib.decompress(s0),data0+data0)
326 self.assertEqual(zlib.decompress(s1),data0+data1)
327
328 def test_badcompresscopy(self):
329 # Test copying a compression object in an inconsistent state
330 c = zlib.compressobj()
331 c.compress(HAMLET_SCENE)
332 c.flush()
333 self.assertRaises(ValueError, c.copy)
334
335 if hasattr(zlib.decompressobj(), "copy"):
336 def test_decompresscopy(self):
337 # Test copying a decompression object
338 data = HAMLET_SCENE
339 comp = zlib.compress(data)
340
341 d0 = zlib.decompressobj()
342 bufs0 = []
343 bufs0.append(d0.decompress(comp[:32]))
344
345 d1 = d0.copy()
346 bufs1 = bufs0[:]
347
348 bufs0.append(d0.decompress(comp[32:]))
349 s0 = ''.join(bufs0)
350
351 bufs1.append(d1.decompress(comp[32:]))
352 s1 = ''.join(bufs1)
353
354 self.assertEqual(s0,s1)
355 self.assertEqual(s0,data)
356
357 def test_baddecompresscopy(self):
358 # Test copying a compression object in an inconsistent state
359 data = zlib.compress(HAMLET_SCENE)
360 d = zlib.decompressobj()
361 d.decompress(data)
362 d.flush()
363 self.assertRaises(ValueError, d.copy)
364
365def genblock(seed, length, step=1024, generator=random):
366 """length-byte stream of random data from a seed (in step-byte blocks)."""
367 if seed is not None:
368 generator.seed(seed)
369 randint = generator.randint
370 if length < step or step < 2:
371 step = length
372 blocks = []
373 for i in range(0, length, step):
374 blocks.append(''.join([chr(randint(0,255))
375 for x in range(step)]))
376 return ''.join(blocks)[:length]
377
378
379
380def choose_lines(source, number, seed=None, generator=random):
381 """Return a list of number lines randomly chosen from the source"""
382 if seed is not None:
383 generator.seed(seed)
384 sources = source.split('\n')
385 return [generator.choice(sources) for n in range(number)]
386
387
388
389HAMLET_SCENE = """
390LAERTES
391
392 O, fear me not.
393 I stay too long: but here my father comes.
394
395 Enter POLONIUS
396
397 A double blessing is a double grace,
398 Occasion smiles upon a second leave.
399
400LORD POLONIUS
401
402 Yet here, Laertes! aboard, aboard, for shame!
403 The wind sits in the shoulder of your sail,
404 And you are stay'd for. There; my blessing with thee!
405 And these few precepts in thy memory
406 See thou character. Give thy thoughts no tongue,
407 Nor any unproportioned thought his act.
408 Be thou familiar, but by no means vulgar.
409 Those friends thou hast, and their adoption tried,
410 Grapple them to thy soul with hoops of steel;
411 But do not dull thy palm with entertainment
412 Of each new-hatch'd, unfledged comrade. Beware
413 Of entrance to a quarrel, but being in,
414 Bear't that the opposed may beware of thee.
415 Give every man thy ear, but few thy voice;
416 Take each man's censure, but reserve thy judgment.
417 Costly thy habit as thy purse can buy,
418 But not express'd in fancy; rich, not gaudy;
419 For the apparel oft proclaims the man,
420 And they in France of the best rank and station
421 Are of a most select and generous chief in that.
422 Neither a borrower nor a lender be;
423 For loan oft loses both itself and friend,
424 And borrowing dulls the edge of husbandry.
425 This above all: to thine ownself be true,
426 And it must follow, as the night the day,
427 Thou canst not then be false to any man.
428 Farewell: my blessing season this in thee!
429
430LAERTES
431
432 Most humbly do I take my leave, my lord.
433
434LORD POLONIUS
435
436 The time invites you; go; your servants tend.
437
438LAERTES
439
440 Farewell, Ophelia; and remember well
441 What I have said to you.
442
443OPHELIA
444
445 'Tis in my memory lock'd,
446 And you yourself shall keep the key of it.
447
448LAERTES
449
450 Farewell.
451"""
452
453
454def test_main():
455 test_support.run_unittest(
456 ChecksumTestCase,
457 ExceptionTestCase,
458 CompressTestCase,
459 CompressObjectTestCase
460 )
461
462if __name__ == "__main__":
463 test_main()
464
465def test(tests=''):
466 if not tests: tests = 'o'
467 testcases = []
468 if 'k' in tests: testcases.append(ChecksumTestCase)
469 if 'x' in tests: testcases.append(ExceptionTestCase)
470 if 'c' in tests: testcases.append(CompressTestCase)
471 if 'o' in tests: testcases.append(CompressObjectTestCase)
472 test_support.run_unittest(*testcases)
473
474if False:
475 import sys
476 sys.path.insert(1, '/Py23Src/python/dist/src/Lib/test')
477 import test_zlib as tz
478 ts, ut = tz.test_support, tz.unittest
479 su = ut.TestSuite()
480 su.addTest(ut.makeSuite(tz.CompressTestCase))
481 ts.run_suite(su)
Note: See TracBrowser for help on using the repository browser.