#NEWLINE# def _runTest( self , num_shards ) : #NEWLINE# #INDENT# entity_count = 1000 #NEWLINE# bucket_name = 'bucket' #NEWLINE# job_name = 'test_map' #NEWLINE# for _ in range ( entity_count ) : #NEWLINE# #INDENT# TestEntity ( ) . put ( ) #NEWLINE# #UNINDENT# mapreduce_id = control . start_map ( job_name , ( __name__ + '.test_handler_yield_key_str' ) , DATASTORE_READER_NAME , { #NEWLINE# #INDENT# 'entity_kind' : ( ( __name__ + '.' ) + TestEntity . __name__ ) , #NEWLINE# 'output_writer' : { #NEWLINE# #INDENT# 'bucket_name' : bucket_name , #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } , shard_count = num_shards , output_writer_spec = self . WRITER_NAME ) #NEWLINE# test_support . execute_until_empty ( self . taskqueue ) #NEWLINE# mapreduce_state = model . MapreduceState . get_by_job_id ( mapreduce_id ) #NEWLINE# filenames = self . WRITER_CLS . get_filenames ( mapreduce_state ) #NEWLINE# self . assertEqual ( num_shards , len ( set ( filenames ) ) ) #NEWLINE# total_entries = 0 #NEWLINE# for shard in range ( num_shards ) : #NEWLINE# #INDENT# self . assertTrue ( filenames [ shard ] . startswith ( ( '/%s/%s' % ( bucket_name , job_name ) ) ) ) #NEWLINE# data = '' . join ( [ _ for _ in records . RecordsReader ( cloudstorage . open ( filenames [ shard ] ) ) ] ) #NEWLINE# total_entries += len ( data . strip ( ) . split ( '\n' ) ) #NEWLINE# #UNINDENT# self . assertEqual ( entity_count , total_entries )