#NEWLINE# def test_lock( self ) : #NEWLINE# #INDENT# lock_a = db . lock ( 'lock-a' ) #NEWLINE# lock_b = db . lock ( 'lock-b' ) #NEWLINE# self . assertTrue ( lock_a . acquire ( ) ) #NEWLINE# self . assertTrue ( lock_b . acquire ( ) ) #NEWLINE# lock_a2 = db . lock ( 'lock-a' ) #NEWLINE# self . assertFalse ( lock_a2 . acquire ( block = False ) ) #NEWLINE# self . assertFalse ( lock_a2 . release ( ) ) #NEWLINE# self . assertNotEqual ( t_acknowledged . _lock_id , lock_a2 . _lock_id ) #NEWLINE# self . assertFalse ( lock_a . acquire ( block = False ) ) #NEWLINE# self . assertFalse ( lock_b . acquire ( block = False ) ) #NEWLINE# t_waiting = threading . Event ( ) #NEWLINE# t_acquired = threading . Event ( ) #NEWLINE# t_acknowledged = threading . Event ( ) #NEWLINE# def wait_for_lock( ) : #NEWLINE# #INDENT# lock_a = db . lock ( 'lock-a' ) #NEWLINE# t_waiting . set ( ) #NEWLINE# lock_a . acquire ( ) #NEWLINE# t_acquired . set ( ) #NEWLINE# t_acknowledged . wait ( ) #NEWLINE# lock_a . release ( ) #NEWLINE# #UNINDENT# waiter_t = threading . Thread ( target = wait_for_lock ) #NEWLINE# waiter_t . start ( ) #NEWLINE# t_waiting . wait ( ) #NEWLINE# lock_a . release ( ) #NEWLINE# t_acquired . wait ( ) #NEWLINE# self . assertFalse ( lock_a . acquire ( block = False ) ) #NEWLINE# t_acknowledged . set ( ) #NEWLINE# waiter_t . join ( ) #NEWLINE# self . assertTrue ( lock_a . acquire ( block = False ) ) #NEWLINE# lock_a . release ( )