#NEWLINE# def test_threaded_execution( self ) : #NEWLINE# #INDENT# consumer = self . create_consumer ( ) #NEWLINE# r1 = modify_state ( 'k1' , 'v1' ) #NEWLINE# r2 = modify_state ( 'k2' , 'v2' ) #NEWLINE# r3 = modify_state ( 'k3' , 'v3' ) #NEWLINE# with CaptureLogs ( ) as capture : #NEWLINE# #INDENT# consumer . start ( ) #NEWLINE# r1 . get ( blocking = True ) #NEWLINE# r2 . get ( blocking = True ) #NEWLINE# r3 . get ( blocking = True ) #NEWLINE# consumer . stop ( ) #NEWLINE# for worker in r2 . worker_threads : #NEWLINE# #INDENT# worker . join ( ) #NEWLINE# #UNINDENT# #UNINDENT# self . assertEqual ( state , { #NEWLINE# #INDENT# 'k1' : 'v1' , #NEWLINE# 'k2' : 'v2' , #NEWLINE# 'k3' : 'v3' , #NEWLINE# #UNINDENT# } )