<s> """<STR_LIT>""" <EOL> import time <EOL> from collections import defaultdict <EOL> from hstestcase import HSTestCase <EOL> from hubstorage import ValueTooLarge <EOL> class BatchUploaderTest ( HSTestCase ) : <EOL> def _job_and_writer ( self , ** writerargs ) : <EOL> self . project . push_job ( self . spidername ) <EOL> job = self . start_job ( ) <EOL> bu = self . hsclient . batchuploader <EOL> w = bu . create_writer ( job . items . url , auth = self . auth , ** writerargs ) <EOL> return job , w <EOL> def test_writer_batchsize ( self ) : <EOL> job , w = self . _job_and_writer ( size = <NUM_LIT:10> ) <EOL> for x in xrange ( <NUM_LIT> ) : <EOL> w . write ( { '<STR_LIT:x>' : x } ) <EOL> w . close ( ) <EOL> groups = defaultdict ( int ) <EOL> for doc in job . items . list ( meta = [ '<STR_LIT>' ] ) : <EOL> groups [ doc [ '<STR_LIT>' ] ] += <NUM_LIT:1> <EOL> self . assertEqual ( len ( groups ) , <NUM_LIT:12> ) <EOL> def test_writer_maxitemsize ( self ) : <EOL> job , w = self . _job_and_writer ( ) <EOL> m = w . maxitemsize <EOL> self . assertRaisesRegexp ( <EOL> ValueTooLarge , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> w . write , { '<STR_LIT:b>' : '<STR_LIT:x>' * m } ) <EOL> self . assertRaisesRegexp ( <EOL> ValueTooLarge , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> w . write , { '<STR_LIT:b>' * m : '<STR_LIT:x>' } ) <EOL> self . assertRaisesRegexp ( <EOL> ValueTooLarge , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> w . write , { '<STR_LIT:b>' * ( m / <NUM_LIT:2> ) : '<STR_LIT:x>' * ( m / <NUM_LIT:2> ) } ) <EOL> def test_writer_contentencoding ( self ) : <EOL> for ce in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> job , w = self . _job_and_writer ( content_encoding = ce ) <EOL> for x in xrange ( <NUM_LIT> ) : <EOL> w . write ( { '<STR_LIT:x>' : x } ) <EOL> w . close ( ) <EOL> self . assertEqual ( job . items . stats ( ) [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> def test_writer_interval ( self ) : <EOL> job , w = self . _job_and_writer ( size = <NUM_LIT:1000> , interval = <NUM_LIT:1> ) <EOL> for x in xrange ( <NUM_LIT> ) : <EOL> w . write ( { '<STR_LIT:x>' : x } ) <EOL> if x == <NUM_LIT:50> : <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> w . close ( ) <EOL> groups = defaultdict ( int ) <EOL> for doc in job . items . list ( meta = [ '<STR_LIT>' ] ) : <EOL> groups [ doc [ '<STR_LIT>' ] ] += <NUM_LIT:1> <EOL> self . assertEqual ( len ( groups ) , <NUM_LIT:2> ) </s>
