<s> """<STR_LIT>""" <EOL> import pickle <EOL> import unittest <EOL> from google . appengine . datastore import datastore_stub_util <EOL> import cloudstorage <EOL> from google . appengine . ext import testbed <EOL> from mapreduce import output_writers <EOL> from mapreduce . tools import gcs_file_seg_reader <EOL> class GCSFileSegReaderTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( GCSFileSegReaderTest , self ) . setUp ( ) <EOL> self . testbed = testbed . Testbed ( ) <EOL> self . testbed . activate ( ) <EOL> self . testbed . init_app_identity_stub ( ) <EOL> self . testbed . init_blobstore_stub ( ) <EOL> policy = datastore_stub_util . PseudoRandomHRConsistencyPolicy ( probability = <NUM_LIT:1> ) <EOL> self . testbed . init_datastore_v3_stub ( consistency_policy = policy ) <EOL> self . testbed . init_memcache_stub ( ) <EOL> self . testbed . init_urlfetch_stub ( ) <EOL> self . writer_cls = output_writers . _GoogleCloudStorageOutputWriter <EOL> self . seg_prefix = "<STR_LIT>" <EOL> def tearDown ( self ) : <EOL> self . testbed . deactivate ( ) <EOL> super ( GCSFileSegReaderTest , self ) . tearDown ( ) <EOL> def testMissingMetadata ( self ) : <EOL> f = cloudstorage . open ( self . seg_prefix + "<STR_LIT:0>" , "<STR_LIT:w>" ) <EOL> f . write ( "<STR_LIT:abc>" ) <EOL> f . close ( ) <EOL> self . assertRaises ( ValueError , <EOL> gcs_file_seg_reader . _GCSFileSegReader , <EOL> self . seg_prefix , <NUM_LIT:0> ) <EOL> def testInvalidMetadata ( self ) : <EOL> f = cloudstorage . open ( self . seg_prefix + "<STR_LIT:0>" , "<STR_LIT:w>" , <EOL> options = { self . writer_cls . _VALID_LENGTH : "<STR_LIT>" } ) <EOL> f . write ( "<STR_LIT:abc>" ) <EOL> f . close ( ) <EOL> self . assertRaises ( ValueError , <EOL> gcs_file_seg_reader . _GCSFileSegReader , <EOL> self . seg_prefix , <NUM_LIT:0> ) <EOL> def ReadOneFileTest ( self , read_size ) : <EOL> f = cloudstorage . open ( self . seg_prefix + "<STR_LIT:0>" , "<STR_LIT:w>" , <EOL> options = { self . writer_cls . _VALID_LENGTH : "<STR_LIT:5>" } ) <EOL> f . write ( "<STR_LIT>" ) <EOL> f . close ( ) <EOL> r = gcs_file_seg_reader . _GCSFileSegReader ( self . seg_prefix , <NUM_LIT:0> ) <EOL> result = "<STR_LIT>" <EOL> while True : <EOL> tmp = r . read ( read_size ) <EOL> if not tmp : <EOL> break <EOL> result += tmp <EOL> self . assertEqual ( "<STR_LIT>" , result ) <EOL> self . assertEqual ( <NUM_LIT:5> , r . tell ( ) ) <EOL> def testReadBig ( self ) : <EOL> """<STR_LIT>""" <EOL> self . ReadOneFileTest ( <NUM_LIT:10> ) <EOL> def testReadSmall ( self ) : <EOL> """<STR_LIT>""" <EOL> self . ReadOneFileTest ( <NUM_LIT:1> ) <EOL> def testReadEqual ( self ) : <EOL> """<STR_LIT>""" <EOL> self . ReadOneFileTest ( <NUM_LIT:5> ) <EOL> def setUpMultipleFile ( self ) : <EOL> f = cloudstorage . open ( self . seg_prefix + "<STR_LIT:0>" , "<STR_LIT:w>" , <EOL> options = { self . writer_cls . _VALID_LENGTH : "<STR_LIT:5>" } ) <EOL> f . write ( "<STR_LIT>" ) <EOL> f . close ( ) <EOL> f = cloudstorage . open ( self . seg_prefix + "<STR_LIT:1>" , "<STR_LIT:w>" , <EOL> options = { self . writer_cls . _VALID_LENGTH : "<STR_LIT:5>" } ) <EOL> f . write ( "<STR_LIT>" ) <EOL> f . close ( ) <EOL> f = cloudstorage . open ( self . seg_prefix + "<STR_LIT:2>" , "<STR_LIT:w>" , <EOL> options = { self . writer_cls . _VALID_LENGTH : "<STR_LIT>" } ) <EOL> f . write ( "<STR_LIT>" ) <EOL> f . close ( ) <EOL> def testReadMultipleFiles ( self ) : <EOL> self . setUpMultipleFile ( ) <EOL> r = gcs_file_seg_reader . _GCSFileSegReader ( self . seg_prefix , <NUM_LIT:2> ) <EOL> result = "<STR_LIT>" <EOL> while True : <EOL> tmp = r . read ( <NUM_LIT:1> ) <EOL> if not tmp : <EOL> break <EOL> result += tmp <EOL> self . assertEqual ( "<STR_LIT>" , result ) <EOL> self . assertEqual ( len ( result ) , r . tell ( ) ) <EOL> def testPickle ( self ) : <EOL> self . setUpMultipleFile ( ) <EOL> r = gcs_file_seg_reader . _GCSFileSegReader ( self . seg_prefix , <NUM_LIT:2> ) <EOL> result = "<STR_LIT>" <EOL> while True : <EOL> r = pickle . loads ( pickle . dumps ( r ) ) <EOL> tmp = r . read ( <NUM_LIT:1> ) <EOL> if not tmp : <EOL> break <EOL> result += tmp <EOL> self . assertEqual ( len ( result ) , r . tell ( ) ) <EOL> self . assertEqual ( "<STR_LIT>" , result ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
