<s> import uuid <EOL> import threading <EOL> from nose . tools import eq_ , ok_ <EOL> from kazoo . exceptions import CancelledError <EOL> from kazoo . exceptions import LockTimeout <EOL> from kazoo . testing import KazooTestCase <EOL> from kazoo . tests . util import wait <EOL> class KazooLockTests ( KazooTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( KazooLockTests , self ) . setUp ( ) <EOL> self . lockpath = "<STR_LIT:/>" + uuid . uuid4 ( ) . hex <EOL> self . condition = threading . Condition ( ) <EOL> self . released = threading . Event ( ) <EOL> self . active_thread = None <EOL> self . cancelled_threads = [ ] <EOL> def _thread_lock_acquire_til_event ( self , name , lock , event ) : <EOL> try : <EOL> with lock : <EOL> with self . condition : <EOL> eq_ ( self . active_thread , None ) <EOL> self . active_thread = name <EOL> self . condition . notify_all ( ) <EOL> event . wait ( ) <EOL> with self . condition : <EOL> eq_ ( self . active_thread , name ) <EOL> self . active_thread = None <EOL> self . condition . notify_all ( ) <EOL> self . released . set ( ) <EOL> except CancelledError : <EOL> with self . condition : <EOL> self . cancelled_threads . append ( name ) <EOL> self . condition . notify_all ( ) <EOL> def test_lock_one ( self ) : <EOL> lock_name = uuid . uuid4 ( ) . hex <EOL> lock = self . client . Lock ( self . lockpath , lock_name ) <EOL> event = threading . Event ( ) <EOL> thread = threading . Thread ( target = self . _thread_lock_acquire_til_event , <EOL> args = ( lock_name , lock , event ) ) <EOL> thread . start ( ) <EOL> lock2_name = uuid . uuid4 ( ) . hex <EOL> anotherlock = self . client . Lock ( self . lockpath , lock2_name ) <EOL> wait ( anotherlock . contenders ) <EOL> eq_ ( anotherlock . contenders ( ) , [ lock_name ] ) <EOL> with self . condition : <EOL> while self . active_thread != lock_name : <EOL> self . condition . wait ( ) <EOL> event . set ( ) <EOL> with self . condition : <EOL> while self . active_thread : <EOL> self . condition . wait ( ) <EOL> self . released . wait ( ) <EOL> thread . join ( ) <EOL> def test_lock ( self ) : <EOL> threads = [ ] <EOL> names = [ "<STR_LIT>" + str ( i ) for i in range ( <NUM_LIT:5> ) ] <EOL> contender_bits = { } <EOL> for name in names : <EOL> e = threading . Event ( ) <EOL> l = self . client . Lock ( self . lockpath , name ) <EOL> t = threading . Thread ( target = self . _thread_lock_acquire_til_event , <EOL> args = ( name , l , e ) ) <EOL> contender_bits [ name ] = ( t , e ) <EOL> threads . append ( t ) <EOL> lock = self . client . Lock ( self . lockpath , "<STR_LIT:test>" ) <EOL> lock . acquire ( ) <EOL> for t in threads : <EOL> t . start ( ) <EOL> wait ( lambda : len ( lock . contenders ( ) ) == <NUM_LIT:6> ) <EOL> contenders = lock . contenders ( ) <EOL> eq_ ( contenders [ <NUM_LIT:0> ] , "<STR_LIT:test>" ) <EOL> contenders = contenders [ <NUM_LIT:1> : ] <EOL> remaining = list ( contenders ) <EOL> lock . release ( ) <EOL> for contender in contenders : <EOL> thread , event = contender_bits [ contender ] <EOL> with self . condition : <EOL> while not self . active_thread : <EOL> self . condition . wait ( ) <EOL> eq_ ( self . active_thread , contender ) <EOL> eq_ ( lock . contenders ( ) , remaining ) <EOL> remaining = remaining [ <NUM_LIT:1> : ] <EOL> event . set ( ) <EOL> with self . condition : <EOL> while self . active_thread : <EOL> self . condition . wait ( ) <EOL> for thread in threads : <EOL> thread . join ( ) <EOL> def test_lock_reconnect ( self ) : <EOL> event = threading . Event ( ) <EOL> other_lock = self . client . Lock ( self . lockpath , '<STR_LIT>' ) <EOL> thread = threading . Thread ( target = self . _thread_lock_acquire_til_event , <EOL> args = ( '<STR_LIT>' , other_lock , event ) ) <EOL> lock = self . client . Lock ( self . lockpath , "<STR_LIT:test>" ) <EOL> lock . acquire ( ) <EOL> thread . start ( ) <EOL> wait ( lambda : len ( lock . contenders ( ) ) == <NUM_LIT:2> ) <EOL> eq_ ( lock . contenders ( ) , [ '<STR_LIT:test>' , '<STR_LIT>' ] ) <EOL> self . expire_session ( ) <EOL> lock . release ( ) <EOL> with self . condition : <EOL> while not self . active_thread : <EOL> self . condition . wait ( ) <EOL> eq_ ( self . active_thread , '<STR_LIT>' ) <EOL> event . set ( ) <EOL> thread . join ( ) <EOL> def test_lock_non_blocking ( self ) : <EOL> lock_name = uuid . uuid4 ( ) . hex <EOL> lock = self . client . Lock ( self . lockpath , lock_name ) <EOL> event = threading . Event ( ) <EOL> thread = threading . Thread ( target = self . _thread_lock_acquire_til_event , <EOL> args = ( lock_name , lock , event ) ) <EOL> thread . start ( ) <EOL> lock1 = self . client . Lock ( self . lockpath , lock_name ) <EOL> with self . condition : <EOL> if not self . active_thread : <EOL> self . condition . wait ( <NUM_LIT:5> ) <EOL> ok_ ( not lock1 . acquire ( blocking = False ) ) <EOL> eq_ ( lock . contenders ( ) , [ lock_name ] ) <EOL> event . set ( ) <EOL> thread . join ( ) <EOL> def test_lock_fail_first_call ( self ) : <EOL> event1 = threading . Event ( ) <EOL> lock1 = self . client . Lock ( self . lockpath , "<STR_LIT>" ) <EOL> thread1 = threading . Thread ( target = self . _thread_lock_acquire_til_event , <EOL> args = ( "<STR_LIT>" , lock1 , event1 ) ) <EOL> thread1 . start ( ) <EOL> with self . condition : <EOL> if not self . active_thread : <EOL> self . condition . wait ( <NUM_LIT:5> ) <EOL> eq_ ( self . active_thread , "<STR_LIT>" ) <EOL> eq_ ( lock1 . contenders ( ) , [ "<STR_LIT>" ] ) <EOL> event1 . set ( ) <EOL> thread1 . join ( ) <EOL> def test_lock_cancel ( self ) : <EOL> event1 = threading . Event ( ) <EOL> lock1 = self . client . Lock ( self . lockpath , "<STR_LIT>" ) <EOL> thread1 = threading . Thread ( target = self . _thread_lock_acquire_til_event , <EOL> args = ( "<STR_LIT>" , lock1 , event1 ) ) <EOL> thread1 . start ( ) <EOL> with self . condition : <EOL> if not self . active_thread : <EOL> self . condition . wait ( <NUM_LIT:5> ) <EOL> eq_ ( self . active_thread , "<STR_LIT>" ) <EOL> client2 = self . _get_client ( ) <EOL> client2 . start ( ) <EOL> event2 = threading . Event ( ) <EOL> lock2 = client2 . Lock ( self . lockpath , "<STR_LIT>" ) <EOL> thread2 = threading . Thread ( target = self . _thread_lock_acquire_til_event , <EOL> args = ( "<STR_LIT>" , lock2 , event2 ) ) <EOL> thread2 . start ( ) <EOL> wait ( lambda : len ( lock2 . contenders ( ) ) > <NUM_LIT:1> ) <EOL> eq_ ( lock2 . contenders ( ) , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> lock2 . cancel ( ) <EOL> with self . condition : <EOL> if not "<STR_LIT>" in self . cancelled_threads : <EOL> self . condition . wait ( ) <EOL> assert "<STR_LIT>" in self . cancelled_threads <EOL> eq_ ( lock2 . contenders ( ) , [ "<STR_LIT>" ] ) <EOL> thread2 . join ( ) <EOL> event1 . set ( ) <EOL> thread1 . join ( ) <EOL> client2 . stop ( ) <EOL> def test_lock_double_calls ( self ) : <EOL> lock1 = self . client . Lock ( self . lockpath , "<STR_LIT>" ) <EOL> lock1 . acquire ( ) <EOL> lock1 . acquire ( ) <EOL> lock1 . release ( ) <EOL> lock1 . release ( ) <EOL> def test_lock_reacquire ( self ) : <EOL> lock = self . client . Lock ( self . lockpath , "<STR_LIT>" ) <EOL> lock . acquire ( ) <EOL> lock . release ( ) <EOL> lock . acquire ( ) <EOL> lock . release ( ) <EOL> def test_lock_timeout ( self ) : <EOL> timeout = <NUM_LIT:3> <EOL> e = threading . Event ( ) <EOL> started = threading . Event ( ) <EOL> lock1 = self . client . Lock ( self . lockpath , "<STR_LIT>" ) <EOL> def _thread ( lock , event , timeout ) : <EOL> with lock : <EOL> started . set ( ) <EOL> event . wait ( timeout ) <EOL> if not event . isSet ( ) : <EOL> self . fail ( "<STR_LIT>" ) <EOL> t = threading . Thread ( target = _thread , args = ( lock1 , e , timeout * <NUM_LIT:3> ) ) <EOL> t . start ( ) <EOL> client2 = self . _get_client ( ) <EOL> client2 . start ( ) <EOL> started . wait ( <NUM_LIT:5> ) <EOL> self . assertTrue ( started . isSet ( ) ) <EOL> lock2 = client2 . Lock ( self . lockpath , "<STR_LIT>" ) <EOL> try : <EOL> lock2 . acquire ( timeout = timeout ) <EOL> except LockTimeout : <EOL> pass <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> finally : <EOL> e . set ( ) <EOL> t . join ( ) <EOL> client2 . stop ( ) <EOL> class TestSemaphore ( KazooTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestSemaphore , self ) . setUp ( ) <EOL> self . lockpath = "<STR_LIT:/>" + uuid . uuid4 ( ) . hex <EOL> self . condition = threading . Condition ( ) <EOL> self . released = threading . Event ( ) <EOL> self . active_thread = None <EOL> self . cancelled_threads = [ ] <EOL> def test_basic ( self ) : <EOL> sem1 = self . client . Semaphore ( self . lockpath ) <EOL> sem1 . acquire ( ) <EOL> sem1 . release ( ) <EOL> def test_lock_one ( self ) : <EOL> sem1 = self . client . Semaphore ( self . lockpath , max_leases = <NUM_LIT:1> ) <EOL> sem2 = self . client . Semaphore ( self . lockpath , max_leases = <NUM_LIT:1> ) <EOL> started = threading . Event ( ) <EOL> event = threading . Event ( ) <EOL> sem1 . acquire ( ) <EOL> def sema_one ( ) : <EOL> started . set ( ) <EOL> with sem2 : <EOL> event . set ( ) <EOL> thread = threading . Thread ( target = sema_one , args = ( ) ) <EOL> thread . start ( ) <EOL> started . wait ( <NUM_LIT:10> ) <EOL> self . assertFalse ( event . is_set ( ) ) <EOL> sem1 . release ( ) <EOL> event . wait ( <NUM_LIT:10> ) <EOL> self . assert_ ( event . is_set ( ) ) <EOL> thread . join ( ) <EOL> def test_non_blocking ( self ) : <EOL> sem1 = self . client . Semaphore ( <EOL> self . lockpath , identifier = '<STR_LIT>' , max_leases = <NUM_LIT:2> ) <EOL> sem2 = self . client . Semaphore ( <EOL> self . lockpath , identifier = '<STR_LIT>' , max_leases = <NUM_LIT:2> ) <EOL> sem3 = self . client . Semaphore ( <EOL> self . lockpath , identifier = '<STR_LIT>' , max_leases = <NUM_LIT:2> ) <EOL> sem1 . acquire ( ) <EOL> sem2 . acquire ( ) <EOL> ok_ ( not sem3 . acquire ( blocking = False ) ) <EOL> eq_ ( set ( sem1 . lease_holders ( ) ) , set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> sem2 . release ( ) <EOL> sem3 . acquire ( ) <EOL> eq_ ( set ( sem1 . lease_holders ( ) ) , set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> sem1 . release ( ) <EOL> sem3 . release ( ) <EOL> def test_non_blocking_release ( self ) : <EOL> sem1 = self . client . Semaphore ( <EOL> self . lockpath , identifier = '<STR_LIT>' , max_leases = <NUM_LIT:1> ) <EOL> sem2 = self . client . Semaphore ( <EOL> self . lockpath , identifier = '<STR_LIT>' , max_leases = <NUM_LIT:1> ) <EOL> sem1 . acquire ( ) <EOL> sem2 . acquire ( blocking = False ) <EOL> sem1 . release ( ) <EOL> sem2 . release ( ) <EOL> def test_holders ( self ) : <EOL> started = threading . Event ( ) <EOL> event = threading . Event ( ) <EOL> def sema_one ( ) : <EOL> with self . client . Semaphore ( self . lockpath , '<STR_LIT>' , max_leases = <NUM_LIT:1> ) : <EOL> started . set ( ) <EOL> event . wait ( ) <EOL> thread = threading . Thread ( target = sema_one , args = ( ) ) <EOL> thread . start ( ) <EOL> started . wait ( ) <EOL> sem1 = self . client . Semaphore ( self . lockpath ) <EOL> holders = sem1 . lease_holders ( ) <EOL> eq_ ( holders , [ '<STR_LIT>' ] ) <EOL> event . set ( ) <EOL> thread . join ( ) <EOL> def test_semaphore_cancel ( self ) : <EOL> sem1 = self . client . Semaphore ( self . lockpath , '<STR_LIT>' , max_leases = <NUM_LIT:1> ) <EOL> sem2 = self . client . Semaphore ( self . lockpath , '<STR_LIT>' , max_leases = <NUM_LIT:1> ) <EOL> sem1 . acquire ( ) <EOL> started = threading . Event ( ) <EOL> event = threading . Event ( ) <EOL> def sema_one ( ) : <EOL> started . set ( ) <EOL> try : <EOL> with sem2 : <EOL> started . set ( ) <EOL> except CancelledError : <EOL> event . set ( ) <EOL> thread = threading . Thread ( target = sema_one , args = ( ) ) <EOL> thread . start ( ) <EOL> started . wait ( ) <EOL> eq_ ( sem1 . lease_holders ( ) , [ '<STR_LIT>' ] ) <EOL> eq_ ( event . is_set ( ) , False ) <EOL> sem2 . cancel ( ) <EOL> event . wait ( ) <EOL> eq_ ( event . is_set ( ) , True ) <EOL> thread . join ( ) <EOL> def test_multiple_acquire_and_release ( self ) : <EOL> sem1 = self . client . Semaphore ( self . lockpath , '<STR_LIT>' , max_leases = <NUM_LIT:1> ) <EOL> sem1 . acquire ( ) <EOL> sem1 . acquire ( ) <EOL> eq_ ( True , sem1 . release ( ) ) <EOL> eq_ ( False , sem1 . release ( ) ) <EOL> def test_handle_session_loss ( self ) : <EOL> expire_semaphore = self . client . Semaphore ( self . lockpath , '<STR_LIT>' , <EOL> max_leases = <NUM_LIT:1> ) <EOL> client = self . _get_client ( ) <EOL> client . start ( ) <EOL> lh_semaphore = client . Semaphore ( self . lockpath , '<STR_LIT>' , max_leases = <NUM_LIT:1> ) <EOL> lh_semaphore . acquire ( ) <EOL> started = threading . Event ( ) <EOL> event = threading . Event ( ) <EOL> event2 = threading . Event ( ) <EOL> def sema_one ( ) : <EOL> started . set ( ) <EOL> with expire_semaphore : <EOL> event . set ( ) <EOL> event2 . wait ( ) <EOL> thread = threading . Thread ( target = sema_one , args = ( ) ) <EOL> thread . start ( ) <EOL> started . wait ( ) <EOL> eq_ ( lh_semaphore . lease_holders ( ) , [ '<STR_LIT>' ] ) <EOL> expired = threading . Event ( ) <EOL> def expire ( ) : <EOL> self . expire_session ( ) <EOL> expired . set ( ) <EOL> thread = threading . Thread ( target = expire , args = ( ) ) <EOL> thread . start ( ) <EOL> expire_semaphore . wake_event . wait ( ) <EOL> expired . wait ( ) <EOL> lh_semaphore . release ( ) <EOL> client . stop ( ) <EOL> event . wait ( <NUM_LIT:5> ) <EOL> eq_ ( expire_semaphore . lease_holders ( ) , [ '<STR_LIT>' ] ) <EOL> event2 . set ( ) <EOL> thread . join ( ) <EOL> def test_inconsistent_max_leases ( self ) : <EOL> sem1 = self . client . Semaphore ( self . lockpath , max_leases = <NUM_LIT:1> ) <EOL> sem2 = self . client . Semaphore ( self . lockpath , max_leases = <NUM_LIT:2> ) <EOL> sem1 . acquire ( ) <EOL> self . assertRaises ( ValueError , sem2 . acquire ) <EOL> def test_inconsistent_max_leases_other_data ( self ) : <EOL> sem1 = self . client . Semaphore ( self . lockpath , max_leases = <NUM_LIT:1> ) <EOL> sem2 = self . client . Semaphore ( self . lockpath , max_leases = <NUM_LIT:2> ) <EOL> self . client . ensure_path ( self . lockpath ) <EOL> self . client . set ( self . lockpath , b'<STR_LIT>' ) <EOL> sem1 . acquire ( ) <EOL> ok_ ( sem2 . acquire ( blocking = False ) ) <EOL> def test_reacquire ( self ) : <EOL> lock = self . client . Semaphore ( self . lockpath ) <EOL> lock . acquire ( ) <EOL> lock . release ( ) <EOL> lock . acquire ( ) <EOL> lock . release ( ) <EOL> def test_acquire_after_cancelled ( self ) : <EOL> lock = self . client . Semaphore ( self . lockpath ) <EOL> self . assertTrue ( lock . acquire ( ) ) <EOL> self . assertTrue ( lock . release ( ) ) <EOL> lock . cancel ( ) <EOL> self . assertTrue ( lock . cancelled ) <EOL> self . assertTrue ( lock . acquire ( ) ) <EOL> def test_timeout ( self ) : <EOL> timeout = <NUM_LIT:3> <EOL> e = threading . Event ( ) <EOL> started = threading . Event ( ) <EOL> sem1 = self . client . Semaphore ( self . lockpath , "<STR_LIT>" ) <EOL> def _thread ( sem , event , timeout ) : <EOL> with sem : <EOL> started . set ( ) <EOL> event . wait ( timeout ) <EOL> if not event . isSet ( ) : <EOL> self . fail ( "<STR_LIT>" ) <EOL> t = threading . Thread ( target = _thread , args = ( sem1 , e , timeout * <NUM_LIT:3> ) ) <EOL> t . start ( ) <EOL> client2 = self . _get_client ( ) <EOL> client2 . start ( ) <EOL> started . wait ( <NUM_LIT:5> ) <EOL> self . assertTrue ( started . isSet ( ) ) <EOL> sem2 = client2 . Semaphore ( self . lockpath , "<STR_LIT>" ) <EOL> try : <EOL> sem2 . acquire ( timeout = timeout ) <EOL> except LockTimeout : <EOL> e . set ( ) <EOL> finally : <EOL> t . join ( ) <EOL> client2 . stop ( ) </s>
