<s> import msgpack <EOL> from attic . testsuite import AtticTestCase <EOL> from attic . archive import CacheChunkBuffer , RobustUnpacker <EOL> from attic . key import PlaintextKey <EOL> class MockCache : <EOL> def __init__ ( self ) : <EOL> self . objects = { } <EOL> def add_chunk ( self , id , data , stats = None ) : <EOL> self . objects [ id ] = data <EOL> return id , len ( data ) , len ( data ) <EOL> class ChunkBufferTestCase ( AtticTestCase ) : <EOL> def test ( self ) : <EOL> data = [ { b'<STR_LIT:foo>' : <NUM_LIT:1> } , { b'<STR_LIT:bar>' : <NUM_LIT:2> } ] <EOL> cache = MockCache ( ) <EOL> key = PlaintextKey ( ) <EOL> chunks = CacheChunkBuffer ( cache , key , None ) <EOL> for d in data : <EOL> chunks . add ( d ) <EOL> chunks . flush ( ) <EOL> chunks . flush ( flush = True ) <EOL> self . assert_equal ( len ( chunks . chunks ) , <NUM_LIT:2> ) <EOL> unpacker = msgpack . Unpacker ( ) <EOL> for id in chunks . chunks : <EOL> unpacker . feed ( cache . objects [ id ] ) <EOL> self . assert_equal ( data , list ( unpacker ) ) <EOL> class RobustUnpackerTestCase ( AtticTestCase ) : <EOL> def make_chunks ( self , items ) : <EOL> return b'<STR_LIT>' . join ( msgpack . packb ( { '<STR_LIT:path>' : item } ) for item in items ) <EOL> def _validator ( self , value ) : <EOL> return isinstance ( value , dict ) and value . get ( b'<STR_LIT:path>' ) in ( b'<STR_LIT:foo>' , b'<STR_LIT:bar>' , b'<STR_LIT>' , b'<STR_LIT>' ) <EOL> def process ( self , input ) : <EOL> unpacker = RobustUnpacker ( validator = self . _validator ) <EOL> result = [ ] <EOL> for should_sync , chunks in input : <EOL> if should_sync : <EOL> unpacker . resync ( ) <EOL> for data in chunks : <EOL> unpacker . feed ( data ) <EOL> for item in unpacker : <EOL> result . append ( item ) <EOL> return result <EOL> def test_extra_garbage_no_sync ( self ) : <EOL> chunks = [ ( False , [ self . make_chunks ( [ b'<STR_LIT:foo>' , b'<STR_LIT:bar>' ] ) ] ) , <EOL> ( False , [ b'<STR_LIT>' ] + [ self . make_chunks ( [ b'<STR_LIT>' , b'<STR_LIT>' ] ) ] ) ] <EOL> result = self . process ( chunks ) <EOL> self . assert_equal ( result , [ <EOL> { b'<STR_LIT:path>' : b'<STR_LIT:foo>' } , { b'<STR_LIT:path>' : b'<STR_LIT:bar>' } , <EOL> <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <EOL> { b'<STR_LIT:path>' : b'<STR_LIT>' } , <EOL> { b'<STR_LIT:path>' : b'<STR_LIT>' } ] ) <EOL> def split ( self , left , length ) : <EOL> parts = [ ] <EOL> while left : <EOL> parts . append ( left [ : length ] ) <EOL> left = left [ length : ] <EOL> return parts <EOL> def test_correct_stream ( self ) : <EOL> chunks = self . split ( self . make_chunks ( [ b'<STR_LIT:foo>' , b'<STR_LIT:bar>' , b'<STR_LIT>' , b'<STR_LIT>' ] ) , <NUM_LIT:2> ) <EOL> input = [ ( False , chunks ) ] <EOL> result = self . process ( input ) <EOL> self . assert_equal ( result , [ { b'<STR_LIT:path>' : b'<STR_LIT:foo>' } , { b'<STR_LIT:path>' : b'<STR_LIT:bar>' } , { b'<STR_LIT:path>' : b'<STR_LIT>' } , { b'<STR_LIT:path>' : b'<STR_LIT>' } ] ) <EOL> def test_missing_chunk ( self ) : <EOL> chunks = self . split ( self . make_chunks ( [ b'<STR_LIT:foo>' , b'<STR_LIT:bar>' , b'<STR_LIT>' , b'<STR_LIT>' ] ) , <NUM_LIT:4> ) <EOL> input = [ ( False , chunks [ : <NUM_LIT:3> ] ) , ( True , chunks [ <NUM_LIT:4> : ] ) ] <EOL> result = self . process ( input ) <EOL> self . assert_equal ( result , [ { b'<STR_LIT:path>' : b'<STR_LIT:foo>' } , { b'<STR_LIT:path>' : b'<STR_LIT>' } , { b'<STR_LIT:path>' : b'<STR_LIT>' } ] ) <EOL> def test_corrupt_chunk ( self ) : <EOL> chunks = self . split ( self . make_chunks ( [ b'<STR_LIT:foo>' , b'<STR_LIT:bar>' , b'<STR_LIT>' , b'<STR_LIT>' ] ) , <NUM_LIT:4> ) <EOL> input = [ ( False , chunks [ : <NUM_LIT:3> ] ) , ( True , [ b'<STR_LIT>' , b'<STR_LIT>' ] + chunks [ <NUM_LIT:3> : ] ) ] <EOL> result = self . process ( input ) <EOL> self . assert_equal ( result , [ { b'<STR_LIT:path>' : b'<STR_LIT:foo>' } , { b'<STR_LIT:path>' : b'<STR_LIT>' } , { b'<STR_LIT:path>' : b'<STR_LIT>' } ] ) </s>
