#NEWLINE# def simple_queue_test( self , q ) : #NEWLINE# #INDENT# if q . qsize ( ) : #NEWLINE# #INDENT# raise RuntimeError ( 'Call this function with an empty queue' ) #NEWLINE# #UNINDENT# self . assertTrue ( q . empty ( ) ) #NEWLINE# self . assertFalse ( q . full ( ) ) #NEWLINE# q . put ( 111 ) #NEWLINE# q . put ( 333 ) #NEWLINE# q . put ( 222 ) #NEWLINE# target_order = dict ( Queue = [ 111 , 333 , 222 ] , LifoQueue = [ 222 , 333 , 111 ] , PriorityQueue = [ 111 , 222 , 333 ] ) #NEWLINE# actual_order = [ q . get ( ) , q . get ( ) , q . get ( ) ] #NEWLINE# self . assertEqual ( actual_order , target_order [ q . __class__ . __name__ ] , "Didn't seem to queue the correct data!" ) #NEWLINE# for i in range ( ( QUEUE_SIZE - 1 ) ) : #NEWLINE# #INDENT# q . put ( i ) #NEWLINE# self . assertTrue ( q . qsize ( ) , 'Queue should not be empty' ) #NEWLINE# #UNINDENT# self . assertTrue ( ( not qfull ( q ) ) , 'Queue should not be full' ) #NEWLINE# last = ( 2 * QUEUE_SIZE ) #NEWLINE# full = ( ( 3 * 2 ) * QUEUE_SIZE ) #NEWLINE# q . put ( last ) #NEWLINE# self . assertTrue ( qfull ( q ) , 'Queue should be full' ) #NEWLINE# self . assertFalse ( q . empty ( ) ) #NEWLINE# self . assertTrue ( q . full ( ) ) #NEWLINE# try : #NEWLINE# #INDENT# q . put ( full , block = 0 ) #NEWLINE# self . fail ( "Didn't appear to block with a full queue" ) #NEWLINE# #UNINDENT# except queue . Full : #NEWLINE# #INDENT# pass #NEWLINE# #UNINDENT# try : #NEWLINE# #INDENT# q . put ( full , timeout = 0.01 ) #NEWLINE# self . fail ( "Didn't appear to time-out with a full queue" ) #NEWLINE# #UNINDENT# except queue . Full : #NEWLINE# #INDENT# pass #NEWLINE# #UNINDENT# self . do_blocking_test ( q . put , ( full , ) , q . get , ( ) ) #NEWLINE# self . do_blocking_test ( q . put , ( full , True , 10 ) , q . get , ( ) ) #NEWLINE# for i in range ( QUEUE_SIZE ) : #NEWLINE# #INDENT# q . get ( ) #NEWLINE# #UNINDENT# self . assertTrue ( ( not q . qsize ( ) ) , 'Queue should be empty' ) #NEWLINE# try : #NEWLINE# #INDENT# q . get ( block = 0 ) #NEWLINE# self . fail ( "Didn't appear to block with an empty queue" ) #NEWLINE# #UNINDENT# except queue . Empty : #NEWLINE# #INDENT# pass #NEWLINE# #UNINDENT# try : #NEWLINE# #INDENT# q . get ( timeout = 0.01 ) #NEWLINE# self . fail ( "Didn't appear to time-out with an empty queue" ) #NEWLINE# #UNINDENT# except queue . Empty : #NEWLINE# #INDENT# pass #NEWLINE# #UNINDENT# self . do_blocking_test ( q . get , ( ) , q . put , ( 'empty' , ) ) #NEWLINE# self . do_blocking_test ( q . get , ( True , 10 ) , q . put , ( 'empty' , ) )