<s> import unittest <EOL> from mapreduce import control <EOL> from mapreduce import input_readers <EOL> from mapreduce import model <EOL> from mapreduce import test_support <EOL> from testlib import testutil <EOL> try : <EOL> import cloudstorage <EOL> enable_cloudstorage_tests = True <EOL> except ImportError : <EOL> enable_cloudstorage_tests = False <EOL> _memory_mapper_data = [ ] <EOL> def _input_reader_memory_mapper ( data ) : <EOL> _memory_mapper_data . append ( data . read ( ) ) <EOL> class GoogleCloudStorageInputReaderEndToEndTest ( testutil . CloudStorageTestBase ) : <EOL> """<STR_LIT>""" <EOL> def _ClearMapperData ( self ) : <EOL> """<STR_LIT>""" <EOL> global _memory_mapper_data <EOL> _memory_mapper_data = [ ] <EOL> def setUp ( self ) : <EOL> super ( GoogleCloudStorageInputReaderEndToEndTest , self ) . setUp ( ) <EOL> self . _ClearMapperData ( ) <EOL> def create_test_content ( self , bucket_name , object_prefix , num_files ) : <EOL> """<STR_LIT>""" <EOL> created_content = [ ] <EOL> for file_num in range ( num_files ) : <EOL> content = "<STR_LIT>" % file_num <EOL> created_content . append ( content ) <EOL> test_file = cloudstorage . open ( <EOL> "<STR_LIT>" % ( bucket_name , object_prefix , file_num ) , <EOL> mode = "<STR_LIT:w>" ) <EOL> test_file . write ( content ) <EOL> test_file . close ( ) <EOL> return created_content <EOL> def _run_test ( self , num_shards , num_files ) : <EOL> bucket_name = "<STR_LIT>" <EOL> object_prefix = "<STR_LIT>" <EOL> job_name = "<STR_LIT>" <EOL> input_class = ( input_readers . __name__ + "<STR_LIT:.>" + <EOL> input_readers . _GoogleCloudStorageInputReader . __name__ ) <EOL> expected_content = self . create_test_content ( bucket_name , <EOL> object_prefix , <EOL> num_files ) <EOL> control . start_map ( <EOL> job_name , <EOL> __name__ + "<STR_LIT:.>" + "<STR_LIT>" , <EOL> input_class , <EOL> { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : bucket_name , <EOL> "<STR_LIT>" : [ object_prefix + "<STR_LIT:*>" ] <EOL> } , <EOL> } , <EOL> shard_count = num_shards ) <EOL> test_support . execute_until_empty ( self . taskqueue ) <EOL> self . assertEqual ( expected_content . sort ( ) , _memory_mapper_data . sort ( ) ) <EOL> def testSingleShard ( self ) : <EOL> self . _run_test ( num_shards = <NUM_LIT:1> , num_files = <NUM_LIT:10> ) <EOL> def testMultipleShards ( self ) : <EOL> self . _run_test ( num_shards = <NUM_LIT:4> , num_files = <NUM_LIT:10> ) <EOL> def testStrict ( self ) : <EOL> """<STR_LIT>""" <EOL> gcs_files = [ ] <EOL> for num in range ( <NUM_LIT:10> ) : <EOL> gcs_file = "<STR_LIT>" % num <EOL> with cloudstorage . open ( gcs_file , "<STR_LIT:w>" ) as buf : <EOL> buf . write ( str ( num + <NUM_LIT:100> ) ) <EOL> gcs_files . append ( "<STR_LIT>" % num ) <EOL> input_class = ( input_readers . __name__ + "<STR_LIT:.>" + <EOL> input_readers . _GoogleCloudStorageInputReader . __name__ ) <EOL> def _RunMR ( fail_on_missing_input = None ) : <EOL> """<STR_LIT>""" <EOL> self . _ClearMapperData ( ) <EOL> input_reader_dict = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : gcs_files , <EOL> } <EOL> if fail_on_missing_input is not None : <EOL> input_reader_dict [ "<STR_LIT>" ] = fail_on_missing_input <EOL> mr_id = control . start_map ( <EOL> "<STR_LIT>" , <EOL> __name__ + "<STR_LIT:.>" + "<STR_LIT>" , <EOL> input_class , <EOL> { <EOL> "<STR_LIT>" : input_reader_dict , <EOL> } , <EOL> shard_count = <NUM_LIT:10> ) <EOL> test_support . execute_until_empty ( self . taskqueue ) <EOL> return mr_id <EOL> _RunMR ( None ) <EOL> self . assertEqual ( [ str ( num + <NUM_LIT:100> ) for num in range ( <NUM_LIT:10> ) ] , <EOL> sorted ( _memory_mapper_data ) ) <EOL> _RunMR ( False ) <EOL> self . assertEqual ( [ str ( num + <NUM_LIT:100> ) for num in range ( <NUM_LIT:10> ) ] , <EOL> sorted ( _memory_mapper_data ) ) <EOL> _RunMR ( True ) <EOL> self . assertEqual ( [ str ( num + <NUM_LIT:100> ) for num in range ( <NUM_LIT:10> ) ] , <EOL> sorted ( _memory_mapper_data ) ) <EOL> cloudstorage . delete ( "<STR_LIT>" ) <EOL> mr_id = _RunMR ( False ) <EOL> self . assertEqual ( [ str ( num + <NUM_LIT:100> ) for num in [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:6> , <NUM_LIT:7> , <NUM_LIT:8> , <NUM_LIT:9> ] ] , <EOL> sorted ( _memory_mapper_data ) ) <EOL> self . assertEquals ( model . MapreduceState . get_by_job_id ( mr_id ) . result_status , <EOL> model . MapreduceState . RESULT_SUCCESS ) <EOL> mr_id = _RunMR ( True ) <EOL> self . assertEquals ( model . MapreduceState . get_by_job_id ( mr_id ) . result_status , <EOL> model . MapreduceState . RESULT_FAILED ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
