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)
350
374
def test_to_bytes(self):
351
375
content = ('this is some content\n'
352
376
'this content will be compressed\n')
389
413
z_content = zlib.compress(content)
390
414
self.assertEqual(57182, len(z_content))
391
415
block = groupcompress.GroupCompressBlock()
392
block._z_content = z_content
416
block._z_content_chunks = (z_content,)
393
417
block._z_content_length = len(z_content)
394
418
block._compressor_name = 'zlib'
395
419
block._content_length = 158634
434
458
z_content = zlib.compress(content)
435
459
self.assertEqual(57182, len(z_content))
436
460
block = groupcompress.GroupCompressBlock()
437
block._z_content = z_content
461
block._z_content_chunks = (z_content,)
438
462
block._z_content_length = len(z_content)
439
463
block._compressor_name = 'zlib'
440
464
block._content_length = 158634
1067
1091
self.add_key_to_manager(('key4',), locations, block, manager)
1068
1092
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',)))",