#NEWLINE# @ unittest . skipIf ( ( cloudstorage is None ) , 'Cloudstorage library is not available.' ) #NEWLINE# def testSmoke( self ) : #NEWLINE# #INDENT# entity_count = 10 #NEWLINE# parameters . config . _SLICE_DURATION_SEC = 0 #NEWLINE# job = map_job . Job . submit ( map_job . JobConfig ( job_name = 'test_map' , mapper = MyMapper , input_reader_cls = sample_input_reader . SampleInputReader , input_reader_params = { #NEWLINE# #INDENT# 'count' : entity_count , #NEWLINE# #UNINDENT# } , output_writer_cls = output_writers . _GoogleCloudStorageOutputWriter , output_writer_params = { #NEWLINE# #INDENT# 'bucket_name' : 'bucket' , #NEWLINE# #UNINDENT# } ) ) #NEWLINE# test_support . execute_until_empty ( self . taskqueue ) #NEWLINE# total = 0 #NEWLINE# for m in MyMapper . mappers . values ( ) : #NEWLINE# #INDENT# total += m . processed #NEWLINE# #UNINDENT# self . assertEqual ( entity_count , total ) #NEWLINE# counters = dict ( job . get_counters ( ) ) #NEWLINE# self . assertEqual ( counters [ 'FOO_COUNTER' ] , ( 2 * entity_count ) ) #NEWLINE# self . assertEqual ( counters [ 'BAR_COUNTER' ] , ( ( - 1 ) * entity_count ) ) #NEWLINE# self . assertEqual ( counters [ 'SLICES' ] , MyMapper . slices ) #NEWLINE# files = output_writers . _GoogleCloudStorageOutputWriter . get_filenames ( job . _state ) #NEWLINE# outputs = collections . defaultdict ( ( lambda : 0 ) ) #NEWLINE# expected = { #NEWLINE# #INDENT# 'foo\n' : entity_count , #NEWLINE# 'bar\n' : entity_count , #NEWLINE# 'end_slice\n' : MyMapper . slices , #NEWLINE# 'begin_slice\n' : MyMapper . slices , #NEWLINE# #UNINDENT# } #NEWLINE# for fn in files : #NEWLINE# #INDENT# f = cloudstorage . open ( counters ) #NEWLINE# for line in f : #NEWLINE# #INDENT# outputs [ line ] += 1 #NEWLINE# #UNINDENT# #UNINDENT# self . assertEqual ( expected , outputs )