#NEWLINE# def test_multithreaded_unpinning( self ) : #NEWLINE# #INDENT# thread1_lock = Lock ( ) #NEWLINE# thread2_lock = Lock ( ) #NEWLINE# thread1_lock . acquire ( ) #NEWLINE# thread2_lock . acquire ( ) #NEWLINE# pinned = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# def thread1_worker( ) : #NEWLINE# #INDENT# with use_master : #NEWLINE# #INDENT# thread1_lock . acquire ( ) #NEWLINE# #UNINDENT# pinned [ 1 ] = this_thread_is_pinned ( ) #NEWLINE# #UNINDENT# def thread2_worker( ) : #NEWLINE# #INDENT# pin_this_thread ( ) #NEWLINE# with use_master : #NEWLINE# #INDENT# thread2_lock . acquire ( ) #NEWLINE# #UNINDENT# pinned [ 2 ] = this_thread_is_pinned ( ) #NEWLINE# #UNINDENT# thread1 = Thread ( target = thread1_worker ) #NEWLINE# thread2 = Thread ( target = thread2_worker ) #NEWLINE# thread1 . start ( ) #NEWLINE# thread2 . start ( ) #NEWLINE# thread2_lock . release ( ) #NEWLINE# thread2 . join ( ) #NEWLINE# self . assertEqual ( pinned [ 2 ] , True ) #NEWLINE# thread1_lock . release ( ) #NEWLINE# thread1 . join ( ) #NEWLINE# self . assertEqual ( pinned [ 1 ] , False )