<s> import time <EOL> import unittest <EOL> import eventlet <EOL> from eventlet import semaphore <EOL> from tests import LimitedTestCase <EOL> class TestSemaphore ( LimitedTestCase ) : <EOL> def test_bounded ( self ) : <EOL> sem = semaphore . CappedSemaphore ( <NUM_LIT:2> , limit = <NUM_LIT:3> ) <EOL> self . assertEqual ( sem . acquire ( ) , True ) <EOL> self . assertEqual ( sem . acquire ( ) , True ) <EOL> gt1 = eventlet . spawn ( sem . release ) <EOL> self . assertEqual ( sem . acquire ( ) , True ) <EOL> self . assertEqual ( - <NUM_LIT:3> , sem . balance ) <EOL> sem . release ( ) <EOL> sem . release ( ) <EOL> sem . release ( ) <EOL> gt2 = eventlet . spawn ( sem . acquire ) <EOL> sem . release ( ) <EOL> self . assertEqual ( <NUM_LIT:3> , sem . balance ) <EOL> gt1 . wait ( ) <EOL> gt2 . wait ( ) <EOL> def test_bounded_with_zero_limit ( self ) : <EOL> sem = semaphore . CappedSemaphore ( <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> gt = eventlet . spawn ( sem . acquire ) <EOL> sem . release ( ) <EOL> gt . wait ( ) <EOL> def test_non_blocking ( self ) : <EOL> sem = semaphore . Semaphore ( <NUM_LIT:0> ) <EOL> self . assertEqual ( sem . acquire ( blocking = False ) , False ) <EOL> def test_timeout ( self ) : <EOL> sem = semaphore . Semaphore ( <NUM_LIT:0> ) <EOL> start = time . time ( ) <EOL> self . assertEqual ( sem . acquire ( timeout = <NUM_LIT:0.1> ) , False ) <EOL> self . assertTrue ( time . time ( ) - start >= <NUM_LIT:0.1> ) <EOL> def test_timeout_non_blocking ( self ) : <EOL> sem = semaphore . Semaphore ( ) <EOL> self . assertRaises ( ValueError , sem . acquire , blocking = False , timeout = <NUM_LIT:1> ) <EOL> def test_semaphore_contention ( ) : <EOL> g_mutex = semaphore . Semaphore ( ) <EOL> counts = [ <NUM_LIT:0> , <NUM_LIT:0> ] <EOL> def worker ( no ) : <EOL> while min ( counts ) < <NUM_LIT:200> : <EOL> with g_mutex : <EOL> counts [ no - <NUM_LIT:1> ] += <NUM_LIT:1> <EOL> eventlet . sleep ( <NUM_LIT> ) <EOL> t1 = eventlet . spawn ( worker , no = <NUM_LIT:1> ) <EOL> t2 = eventlet . spawn ( worker , no = <NUM_LIT:2> ) <EOL> eventlet . sleep ( <NUM_LIT:0.5> ) <EOL> t1 . kill ( ) <EOL> t2 . kill ( ) <EOL> assert abs ( counts [ <NUM_LIT:0> ] - counts [ <NUM_LIT:1> ] ) < int ( min ( counts ) * <NUM_LIT:0.1> ) , counts <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
