347
347
self.assertEqual(z_content, block._z_content)
348
348
self.assertEqual(content, block._content)
350
def test_to_chunks(self):
351
content_chunks = ['this is some content\n',
352
'this content will be compressed\n']
353
content_len = sum(map(len, content_chunks))
354
content = ''.join(content_chunks)
355
gcb = groupcompress.GroupCompressBlock()
356
gcb.set_chunked_content(content_chunks, content_len)
357
total_len, block_chunks = gcb.to_chunks()
358
block_bytes = ''.join(block_chunks)
359
self.assertEqual(gcb._z_content_length, len(gcb._z_content))
360
self.assertEqual(total_len, len(block_bytes))
361
self.assertEqual(gcb._content_length, content_len)
362
expected_header =('gcb1z\n' # group compress block v1 zlib
363
'%d\n' # Length of compressed content
364
'%d\n' # Length of uncompressed content
365
) % (gcb._z_content_length, gcb._content_length)
366
# The first chunk should be the header chunk. It is small, fixed size,
367
# and there is no compelling reason to split it up
368
self.assertEqual(expected_header, block_chunks[0])
369
self.assertStartsWith(block_bytes, expected_header)
370
remaining_bytes = block_bytes[len(expected_header):]
371
raw_bytes = zlib.decompress(remaining_bytes)
372
self.assertEqual(content, raw_bytes)
374
350
def test_to_bytes(self):
375
351
content = ('this is some content\n'
376
352
'this content will be compressed\n')
413
389
z_content = zlib.compress(content)
414
390
self.assertEqual(57182, len(z_content))
415
391
block = groupcompress.GroupCompressBlock()
416
block._z_content_chunks = (z_content,)
392
block._z_content = z_content
417
393
block._z_content_length = len(z_content)
418
394
block._compressor_name = 'zlib'
419
395
block._content_length = 158634
458
434
z_content = zlib.compress(content)
459
435
self.assertEqual(57182, len(z_content))
460
436
block = groupcompress.GroupCompressBlock()
461
block._z_content_chunks = (z_content,)
437
block._z_content = z_content
462
438
block._z_content_length = len(z_content)
463
439
block._compressor_name = 'zlib'
464
440
block._content_length = 158634
1091
1067
self.add_key_to_manager(('key4',), locations, block, manager)
1092
1068
self.assertTrue(manager.check_is_well_utilized())
1095
class Test_GCBuildDetails(tests.TestCase):
1097
def test_acts_like_tuple(self):
1098
# _GCBuildDetails inlines some of the data that used to be spread out
1099
# across a bunch of tuples
1100
bd = groupcompress._GCBuildDetails((('parent1',), ('parent2',)),
1101
('INDEX', 10, 20, 0, 5))
1102
self.assertEqual(4, len(bd))
1103
self.assertEqual(('INDEX', 10, 20, 0, 5), bd[0])
1104
self.assertEqual(None, bd[1]) # Compression Parent is always None
1105
self.assertEqual((('parent1',), ('parent2',)), bd[2])
1106
self.assertEqual(('group', None), bd[3]) # Record details
1108
def test__repr__(self):
1109
bd = groupcompress._GCBuildDetails((('parent1',), ('parent2',)),
1110
('INDEX', 10, 20, 0, 5))
1111
self.assertEqual("_GCBuildDetails(('INDEX', 10, 20, 0, 5),"
1112
" (('parent1',), ('parent2',)))",