Lib/test/test_zlib.py

   1 import unittest
   2 from test import test_support
   3 import zlib
   4 import random
   5
   6 # print test_support.TESTFN
   7
   8 def 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
  23 class 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
  59 class 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
  76 class 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
  91 class 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
 306 def genblock(seed, length, step=1024, generator=random):
 307     """length-byte stream of random data from a seed (in step-byte blocks)."""
 308     if seed is not None:
 309         generator.seed(seed)
 310     randint = generator.randint
 311     if length < step or step < 2:
 312         step = length
 313     blocks = []
 314     for i in range(0, length, step):
 315         blocks.append(''.join([chr(randint(0,255))
 316                                for x in range(step)]))
 317     return ''.join(blocks)[:length]
 318
 319
 320
 321 def choose_lines(source, number, seed=None, generator=random):
 322     """Return a list of number lines randomly chosen from the source"""
 323     if seed is not None:
 324         generator.seed(seed)
 325     sources = source.split('\n')
 326     return [generator.choice(sources) for n in range(number)]
 327
 328
 329
 330 HAMLET_SCENE = """
 331 LAERTES
 332
 333        O, fear me not.
 334        I stay too long: but here my father comes.
 335
 336        Enter POLONIUS
 337
 338        A double blessing is a double grace,
 339        Occasion smiles upon a second leave.
 340
 341 LORD POLONIUS
 342
 343        Yet here, Laertes! aboard, aboard, for shame!
 344        The wind sits in the shoulder of your sail,
 345        And you are stay'd for. There; my blessing with thee!
 346        And these few precepts in thy memory
 347        See thou character. Give thy thoughts no tongue,
 348        Nor any unproportioned thought his act.
 349        Be thou familiar, but by no means vulgar.
 350        Those friends thou hast, and their adoption tried,
 351        Grapple them to thy soul with hoops of steel;
 352        But do not dull thy palm with entertainment
 353        Of each new-hatch'd, unfledged comrade. Beware
 354        Of entrance to a quarrel, but being in,
 355        Bear't that the opposed may beware of thee.
 356        Give every man thy ear, but few thy voice;
 357        Take each man's censure, but reserve thy judgment.
 358        Costly thy habit as thy purse can buy,
 359        But not express'd in fancy; rich, not gaudy;
 360        For the apparel oft proclaims the man,
 361        And they in France of the best rank and station
 362        Are of a most select and generous chief in that.
 363        Neither a borrower nor a lender be;
 364        For loan oft loses both itself and friend,
 365        And borrowing dulls the edge of husbandry.
 366        This above all: to thine ownself be true,
 367        And it must follow, as the night the day,
 368        Thou canst not then be false to any man.
 369        Farewell: my blessing season this in thee!
 370
 371 LAERTES
 372
 373        Most humbly do I take my leave, my lord.
 374
 375 LORD POLONIUS
 376
 377        The time invites you; go; your servants tend.
 378
 379 LAERTES
 380
 381        Farewell, Ophelia; and remember well
 382        What I have said to you.
 383
 384 OPHELIA
 385
 386        'Tis in my memory lock'd,
 387        And you yourself shall keep the key of it.
 388
 389 LAERTES
 390
 391        Farewell.
 392 """
 393
 394
 395 def test_main():
 396     test_support.run_unittest(
 397         ChecksumTestCase,
 398         ExceptionTestCase,
 399         CompressTestCase,
 400         CompressObjectTestCase
 401     )
 402
 403 if __name__ == "__main__":
 404     test_main()
 405
 406 def test(tests=''):
 407     if not tests: tests = 'o'
 408     testcases = []
 409     if 'k' in tests: testcases.append(ChecksumTestCase)
 410     if 'x' in tests: testcases.append(ExceptionTestCase)
 411     if 'c' in tests: testcases.append(CompressTestCase)
 412     if 'o' in tests: testcases.append(CompressObjectTestCase)
 413     test_support.run_unittest(*testcases)
 414
 415 if False:
 416     import sys
 417     sys.path.insert(1, '/Py23Src/python/dist/src/Lib/test')
 418     import test_zlib as tz
 419     ts, ut = tz.test_support, tz.unittest
 420     su = ut.TestSuite()
 421     su.addTest(ut.makeSuite(tz.CompressTestCase))
 422     ts.run_suite(su)