1 import unittest

     2 from test import test_support

     3 import zlib

     4 import binascii

     5 import random

     6 

     7 

     8 class ChecksumTestCase(unittest.TestCase):

     9     # checksum test cases

    10     def test_crc32start(self):

    11         self.assertEqual(zlib.crc32(""), zlib.crc32("", 0))

    12         self.assert_(zlib.crc32("abc", 0xffffffff))

    13 

    14     def test_crc32empty(self):

    15         self.assertEqual(zlib.crc32("", 0), 0)

    16         self.assertEqual(zlib.crc32("", 1), 1)

    17         self.assertEqual(zlib.crc32("", 432), 432)

    18 

    19     def test_adler32start(self):

    20         self.assertEqual(zlib.adler32(""), zlib.adler32("", 1))

    21         self.assert_(zlib.adler32("abc", 0xffffffff))

    22 

    23     def test_adler32empty(self):

    24         self.assertEqual(zlib.adler32("", 0), 0)

    25         self.assertEqual(zlib.adler32("", 1), 1)

    26         self.assertEqual(zlib.adler32("", 432), 432)

    27 

    28     def assertEqual32(self, seen, expected):

    29         # 32-bit values masked -- checksums on 32- vs 64- bit machines

    30         # This is important if bit 31 (0x08000000L) is set.

    31         self.assertEqual(seen & 0x0FFFFFFFFL, expected & 0x0FFFFFFFFL)

    32 

    33     def test_penguins(self):

    34         self.assertEqual32(zlib.crc32("penguin", 0), 0x0e5c1a120L)

    35         self.assertEqual32(zlib.crc32("penguin", 1), 0x43b6aa94)

    36         self.assertEqual32(zlib.adler32("penguin", 0), 0x0bcf02f6)

    37         self.assertEqual32(zlib.adler32("penguin", 1), 0x0bd602f7)

    38 

    39         self.assertEqual(zlib.crc32("penguin"), zlib.crc32("penguin", 0))

    40         self.assertEqual(zlib.adler32("penguin"),zlib.adler32("penguin",1))

    41 

    42     def test_abcdefghijklmnop(self):

    43         """test issue1202 compliance: signed crc32, adler32 in 2.x"""

    44         foo = 'abcdefghijklmnop'

    45         # explicitly test signed behavior

    46         self.assertEqual(zlib.crc32(foo), -1808088941)

    47         self.assertEqual(zlib.crc32('spam'), 1138425661)

    48         self.assertEqual(zlib.adler32(foo+foo), -721416943)

    49         self.assertEqual(zlib.adler32('spam'), 72286642)

    50 

    51     def test_same_as_binascii_crc32(self):

    52         foo = 'abcdefghijklmnop'

    53         self.assertEqual(binascii.crc32(foo), zlib.crc32(foo))

    54         self.assertEqual(binascii.crc32('spam'), zlib.crc32('spam'))

    55 

    56     def test_negative_crc_iv_input(self):

    57         # The range of valid input values for the crc state should be

    58         # -2**31 through 2**32-1 to allow inputs artifically constrained

    59         # to a signed 32-bit integer.

    60         self.assertEqual(zlib.crc32('ham', -1), zlib.crc32('ham', 0xffffffffL))

    61         self.assertEqual(zlib.crc32('spam', -3141593),

    62                          zlib.crc32('spam',  0xffd01027L))

    63         self.assertEqual(zlib.crc32('spam', -(2**31)),

    64                          zlib.crc32('spam',  (2**31)))

    65 

    66 

    67 class ExceptionTestCase(unittest.TestCase):

    68     # make sure we generate some expected errors

    69     def test_badlevel(self):

    70         # specifying compression level out of range causes an error

    71         # (but -1 is Z_DEFAULT_COMPRESSION and apparently the zlib

    72         # accepts 0 too)

    73         self.assertRaises(zlib.error, zlib.compress, 'ERROR', 10)

    74 

    75     def test_badcompressobj(self):

    76         # verify failure on building compress object with bad params

    77         self.assertRaises(ValueError, zlib.compressobj, 1, zlib.DEFLATED, 0)

    78         # specifying total bits too large causes an error

    79         self.assertRaises(ValueError,

    80                 zlib.compressobj, 1, zlib.DEFLATED, zlib.MAX_WBITS + 1)

    81 

    82     def test_baddecompressobj(self):

    83         # verify failure on building decompress object with bad params

    84         self.assertRaises(ValueError, zlib.decompressobj, 0)

    85 

    86     def test_decompressobj_badflush(self):

    87         # verify failure on calling decompressobj.flush with bad params

    88         self.assertRaises(ValueError, zlib.decompressobj().flush, 0)

    89         self.assertRaises(ValueError, zlib.decompressobj().flush, -1)

    90 

    91 

    92 

    93 class CompressTestCase(unittest.TestCase):

    94     # Test compression in one go (whole message compression)

    95     def test_speech(self):

    96         x = zlib.compress(HAMLET_SCENE)

    97         self.assertEqual(zlib.decompress(x), HAMLET_SCENE)

    98 

    99     def test_speech128(self):

   100         # compress more data

   101         data = HAMLET_SCENE * 128

   102         x = zlib.compress(data)

   103         self.assertEqual(zlib.decompress(x), data)

   104 

   105 

   106 

   107 

   108 class CompressObjectTestCase(unittest.TestCase):

   109     # Test compression object

   110     def test_pair(self):

   111         # straightforward compress/decompress objects

   112         data = HAMLET_SCENE * 128

   113         co = zlib.compressobj()

   114         x1 = co.compress(data)

   115         x2 = co.flush()

   116         self.assertRaises(zlib.error, co.flush) # second flush should not work

   117         dco = zlib.decompressobj()

   118         y1 = dco.decompress(x1 + x2)

   119         y2 = dco.flush()

   120         self.assertEqual(data, y1 + y2)

   121 

   122     def test_compressoptions(self):

   123         # specify lots of options to compressobj()

   124         level = 2

   125         method = zlib.DEFLATED

   126         wbits = -12

   127         memlevel = 9

   128         strategy = zlib.Z_FILTERED

   129         co = zlib.compressobj(level, method, wbits, memlevel, strategy)

   130         x1 = co.compress(HAMLET_SCENE)

   131         x2 = co.flush()

   132         dco = zlib.decompressobj(wbits)

   133         y1 = dco.decompress(x1 + x2)

   134         y2 = dco.flush()

   135         self.assertEqual(HAMLET_SCENE, y1 + y2)

   136 

   137     def test_compressincremental(self):

   138         # compress object in steps, decompress object as one-shot

   139         data = HAMLET_SCENE * 128

   140         co = zlib.compressobj()

   141         bufs = []

   142         for i in range(0, len(data), 256):

   143             bufs.append(co.compress(data[i:i+256]))

   144         bufs.append(co.flush())

   145         combuf = ''.join(bufs)

   146 

   147         dco = zlib.decompressobj()

   148         y1 = dco.decompress(''.join(bufs))

   149         y2 = dco.flush()

   150         self.assertEqual(data, y1 + y2)

   151 

   152     def test_decompinc(self, flush=False, source=None, cx=256, dcx=64):

   153         # compress object in steps, decompress object in steps

   154         source = source or HAMLET_SCENE

   155         data = source * 128

   156         co = zlib.compressobj()

   157         bufs = []

   158         for i in range(0, len(data), cx):

   159             bufs.append(co.compress(data[i:i+cx]))

   160         bufs.append(co.flush())

   161         combuf = ''.join(bufs)

   162 

   163         self.assertEqual(data, zlib.decompress(combuf))

   164 

   165         dco = zlib.decompressobj()

   166         bufs = []

   167         for i in range(0, len(combuf), dcx):

   168             bufs.append(dco.decompress(combuf[i:i+dcx]))

   169             self.assertEqual('', dco.unconsumed_tail, ########

   170                              "(A) uct should be '': not %d long" %

   171                                        len(dco.unconsumed_tail))

   172         if flush:

   173             bufs.append(dco.flush())

   174         else:

   175             while True:

   176                 chunk = dco.decompress('')

   177                 if chunk:

   178                     bufs.append(chunk)

   179                 else:

   180                     break

   181         self.assertEqual('', dco.unconsumed_tail, ########

   182                          "(B) uct should be '': not %d long" %

   183                                        len(dco.unconsumed_tail))

   184         self.assertEqual(data, ''.join(bufs))

   185         # Failure means: "decompressobj with init options failed"

   186 

   187     def test_decompincflush(self):

   188         self.test_decompinc(flush=True)

   189 

   190     def test_decompimax(self, source=None, cx=256, dcx=64):

   191         # compress in steps, decompress in length-restricted steps

   192         source = source or HAMLET_SCENE

   193         # Check a decompression object with max_length specified

   194         data = source * 128

   195         co = zlib.compressobj()

   196         bufs = []

   197         for i in range(0, len(data), cx):

   198             bufs.append(co.compress(data[i:i+cx]))

   199         bufs.append(co.flush())

   200         combuf = ''.join(bufs)

   201         self.assertEqual(data, zlib.decompress(combuf),

   202                          'compressed data failure')

   203 

   204         dco = zlib.decompressobj()

   205         bufs = []

   206         cb = combuf

   207         while cb:

   208             #max_length = 1 + len(cb)//10

   209             chunk = dco.decompress(cb, dcx)

   210             self.failIf(len(chunk) > dcx,

   211                     'chunk too big (%d>%d)' % (len(chunk), dcx))

   212             bufs.append(chunk)

   213             cb = dco.unconsumed_tail

   214         bufs.append(dco.flush())

   215         self.assertEqual(data, ''.join(bufs), 'Wrong data retrieved')

   216 

   217     def test_decompressmaxlen(self, flush=False):

   218         # Check a decompression object with max_length specified

   219         data = HAMLET_SCENE * 128

   220         co = zlib.compressobj()

   221         bufs = []

   222         for i in range(0, len(data), 256):

   223             bufs.append(co.compress(data[i:i+256]))

   224         bufs.append(co.flush())

   225         combuf = ''.join(bufs)

   226         self.assertEqual(data, zlib.decompress(combuf),

   227                          'compressed data failure')

   228 

   229         dco = zlib.decompressobj()

   230         bufs = []

   231         cb = combuf

   232         while cb:

   233             max_length = 1 + len(cb)//10

   234             chunk = dco.decompress(cb, max_length)

   235             self.failIf(len(chunk) > max_length,

   236                         'chunk too big (%d>%d)' % (len(chunk),max_length))

   237             bufs.append(chunk)

   238             cb = dco.unconsumed_tail

   239         if flush:

   240             bufs.append(dco.flush())

   241         else:

   242             while chunk:

   243                 chunk = dco.decompress('', max_length)

   244                 self.failIf(len(chunk) > max_length,

   245                             'chunk too big (%d>%d)' % (len(chunk),max_length))

   246                 bufs.append(chunk)

   247         self.assertEqual(data, ''.join(bufs), 'Wrong data retrieved')

   248 

   249     def test_decompressmaxlenflush(self):

   250         self.test_decompressmaxlen(flush=True)

   251 

   252     def test_maxlenmisc(self):

   253         # Misc tests of max_length

   254         dco = zlib.decompressobj()

   255         self.assertRaises(ValueError, dco.decompress, "", -1)

   256         self.assertEqual('', dco.unconsumed_tail)

   257 

   258     def test_flushes(self):

   259         # Test flush() with the various options, using all the

   260         # different levels in order to provide more variations.

   261         sync_opt = ['Z_NO_FLUSH', 'Z_SYNC_FLUSH', 'Z_FULL_FLUSH']

   262         sync_opt = [getattr(zlib, opt) for opt in sync_opt

   263                     if hasattr(zlib, opt)]

   264         data = HAMLET_SCENE * 8

   265 

   266         for sync in sync_opt:

   267             for level in range(10):

   268                 obj = zlib.compressobj( level )

   269                 a = obj.compress( data[:3000] )

   270                 b = obj.flush( sync )

   271                 c = obj.compress( data[3000:] )

   272                 d = obj.flush()

   273                 self.assertEqual(zlib.decompress(''.join([a,b,c,d])),

   274                                  data, ("Decompress failed: flush "

   275                                         "mode=%i, level=%i") % (sync, level))

   276                 del obj

   277 

   278     def test_odd_flush(self):

   279         # Test for odd flushing bugs noted in 2.0, and hopefully fixed in 2.1

   280         import random

   281 

   282         if hasattr(zlib, 'Z_SYNC_FLUSH'):

   283             # Testing on 17K of "random" data

   284 

   285             # Create compressor and decompressor objects

   286             co = zlib.compressobj(zlib.Z_BEST_COMPRESSION)

   287             dco = zlib.decompressobj()

   288 

   289             # Try 17K of data

   290             # generate random data stream

   291             try:

   292                 # In 2.3 and later, WichmannHill is the RNG of the bug report

   293                 gen = random.WichmannHill()

   294             except AttributeError:

   295                 try:

   296                     # 2.2 called it Random

   297                     gen = random.Random()

   298                 except AttributeError:

   299                     # others might simply have a single RNG

   300                     gen = random

   301             gen.seed(1)

   302             data = genblock(1, 17 * 1024, generator=gen)

   303 

   304             # compress, sync-flush, and decompress

   305             first = co.compress(data)

   306             second = co.flush(zlib.Z_SYNC_FLUSH)

   307             expanded = dco.decompress(first + second)

   308 

   309             # if decompressed data is different from the input data, choke.

   310             self.assertEqual(expanded, data, "17K random source doesn't match")

   311 

   312     def test_empty_flush(self):

   313         # Test that calling .flush() on unused objects works.

   314         # (Bug #1083110 -- calling .flush() on decompress objects

   315         # caused a core dump.)

   316 

   317         co = zlib.compressobj(zlib.Z_BEST_COMPRESSION)

   318         self.failUnless(co.flush())  # Returns a zlib header

   319         dco = zlib.decompressobj()

   320         self.assertEqual(dco.flush(), "") # Returns nothing

   321 

   322     if hasattr(zlib.compressobj(), "copy"):

   323         def test_compresscopy(self):

   324             # Test copying a compression object

   325             data0 = HAMLET_SCENE

   326             data1 = HAMLET_SCENE.swapcase()

   327             c0 = zlib.compressobj(zlib.Z_BEST_COMPRESSION)

   328             bufs0 = []

   329             bufs0.append(c0.compress(data0))

   330 

   331             c1 = c0.copy()

   332             bufs1 = bufs0[:]

   333 

   334             bufs0.append(c0.compress(data0))

   335             bufs0.append(c0.flush())

   336             s0 = ''.join(bufs0)

   337 

   338             bufs1.append(c1.compress(data1))

   339             bufs1.append(c1.flush())

   340             s1 = ''.join(bufs1)

   341 

   342             self.assertEqual(zlib.decompress(s0),data0+data0)

   343             self.assertEqual(zlib.decompress(s1),data0+data1)

   344 

   345         def test_badcompresscopy(self):

   346             # Test copying a compression object in an inconsistent state

   347             c = zlib.compressobj()

   348             c.compress(HAMLET_SCENE)

   349             c.flush()

   350             self.assertRaises(ValueError, c.copy)

   351 

   352     if hasattr(zlib.decompressobj(), "copy"):

   353         def test_decompresscopy(self):

   354             # Test copying a decompression object

   355             data = HAMLET_SCENE

   356             comp = zlib.compress(data)

   357 

   358             d0 = zlib.decompressobj()

   359             bufs0 = []

   360             bufs0.append(d0.decompress(comp[:32]))

   361 

   362             d1 = d0.copy()

   363             bufs1 = bufs0[:]

   364 

   365             bufs0.append(d0.decompress(comp[32:]))

   366             s0 = ''.join(bufs0)

   367 

   368             bufs1.append(d1.decompress(comp[32:]))

   369             s1 = ''.join(bufs1)

   370 

   371             self.assertEqual(s0,s1)

   372             self.assertEqual(s0,data)

   373 

   374         def test_baddecompresscopy(self):

   375             # Test copying a compression object in an inconsistent state

   376             data = zlib.compress(HAMLET_SCENE)

   377             d = zlib.decompressobj()

   378             d.decompress(data)

   379             d.flush()

   380             self.assertRaises(ValueError, d.copy)

   381 

   382 def genblock(seed, length, step=1024, generator=random):

   383     """length-byte stream of random data from a seed (in step-byte blocks)."""

   384     if seed is not None:

   385         generator.seed(seed)

   386     randint = generator.randint

   387     if length < step or step < 2:

   388         step = length

   389     blocks = []

   390     for i in range(0, length, step):

   391         blocks.append(''.join([chr(randint(0,255))

   392                                for x in range(step)]))

   393     return ''.join(blocks)[:length]

   394 

   395 

   396 

   397 def choose_lines(source, number, seed=None, generator=random):

   398     """Return a list of number lines randomly chosen from the source"""

   399     if seed is not None:

   400         generator.seed(seed)

   401     sources = source.split('\n')

   402     return [generator.choice(sources) for n in range(number)]

   403 

   404 

   405 

   406 HAMLET_SCENE = """

   407 LAERTES

   408 

   409        O, fear me not.

   410        I stay too long: but here my father comes.

   411 

   412        Enter POLONIUS

   413 

   414        A double blessing is a double grace,

   415        Occasion smiles upon a second leave.

   416 

   417 LORD POLONIUS

   418 

   419        Yet here, Laertes! aboard, aboard, for shame!

   420        The wind sits in the shoulder of your sail,

   421        And you are stay'd for. There; my blessing with thee!

   422        And these few precepts in thy memory

   423        See thou character. Give thy thoughts no tongue,

   424        Nor any unproportioned thought his act.

   425        Be thou familiar, but by no means vulgar.

   426        Those friends thou hast, and their adoption tried,

   427        Grapple them to thy soul with hoops of steel;

   428        But do not dull thy palm with entertainment

   429        Of each new-hatch'd, unfledged comrade. Beware

   430        Of entrance to a quarrel, but being in,

   431        Bear't that the opposed may beware of thee.

   432        Give every man thy ear, but few thy voice;

   433        Take each man's censure, but reserve thy judgment.

   434        Costly thy habit as thy purse can buy,

   435        But not express'd in fancy; rich, not gaudy;

   436        For the apparel oft proclaims the man,

   437        And they in France of the best rank and station

   438        Are of a most select and generous chief in that.

   439        Neither a borrower nor a lender be;

   440        For loan oft loses both itself and friend,

   441        And borrowing dulls the edge of husbandry.

   442        This above all: to thine ownself be true,

   443        And it must follow, as the night the day,

   444        Thou canst not then be false to any man.

   445        Farewell: my blessing season this in thee!

   446 

   447 LAERTES

   448 

   449        Most humbly do I take my leave, my lord.

   450 

   451 LORD POLONIUS

   452 

   453        The time invites you; go; your servants tend.

   454 

   455 LAERTES

   456 

   457        Farewell, Ophelia; and remember well

   458        What I have said to you.

   459 

   460 OPHELIA

   461 

   462        'Tis in my memory lock'd,

   463        And you yourself shall keep the key of it.

   464 

   465 LAERTES

   466 

   467        Farewell.

   468 """

   469 

   470 

   471 def test_main():

   472     test_support.run_unittest(

   473         ChecksumTestCase,

   474         ExceptionTestCase,

   475         CompressTestCase,

   476         CompressObjectTestCase

   477     )

   478 

   479 if __name__ == "__main__":

   480     test_main()