<s> import zookeeper , zktestbase , unittest , threading <EOL> ZOO_OPEN_ACL_UNSAFE = { "<STR_LIT>" : <NUM_LIT> , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:id>" : "<STR_LIT>" } <EOL> class CreationTest ( zktestbase . TestBase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> zktestbase . TestBase . setUp ( self ) <EOL> try : <EOL> zookeeper . delete ( self . handle , "<STR_LIT>" ) <EOL> zookeeper . delete ( self . handle , "<STR_LIT>" ) <EOL> except : <EOL> pass <EOL> def test_sync_create ( self ) : <EOL> self . assertEqual ( self . connected , True ) <EOL> ret = zookeeper . create ( self . handle , "<STR_LIT>" , "<STR_LIT>" , [ ZOO_OPEN_ACL_UNSAFE ] , zookeeper . EPHEMERAL ) <EOL> self . assertEqual ( ret , "<STR_LIT>" ) <EOL> self . assertRaises ( zookeeper . NoChildrenForEphemeralsException , <EOL> zookeeper . create , <EOL> self . handle , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> [ ZOO_OPEN_ACL_UNSAFE ] , <EOL> zookeeper . EPHEMERAL ) <EOL> def test_sync_create_existing ( self ) : <EOL> self . assertEqual ( self . connected , True ) <EOL> ret = zookeeper . create ( self . handle , "<STR_LIT>" , "<STR_LIT>" , [ ZOO_OPEN_ACL_UNSAFE ] , zookeeper . EPHEMERAL ) <EOL> self . assertEqual ( ret , "<STR_LIT>" ) <EOL> self . assertRaises ( zookeeper . NodeExistsException , <EOL> zookeeper . create , <EOL> self . handle , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> [ ZOO_OPEN_ACL_UNSAFE ] , <EOL> zookeeper . EPHEMERAL ) <EOL> def test_exception_paths ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertRaises ( zookeeper . BadArgumentsException , <EOL> zookeeper . create , <EOL> self . handle , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> [ ZOO_OPEN_ACL_UNSAFE ] , <EOL> - <NUM_LIT:1> ) <EOL> self . assertRaises ( zookeeper . InvalidACLException , <EOL> zookeeper . create , <EOL> self . handle , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ZOO_OPEN_ACL_UNSAFE ) <EOL> def test_async_create ( self ) : <EOL> self . cv = threading . Condition ( ) <EOL> def callback ( handle , rc , value ) : <EOL> self . cv . acquire ( ) <EOL> self . callback_flag = True <EOL> self . rc = rc <EOL> self . cv . notify ( ) <EOL> self . cv . release ( ) <EOL> self . assertEqual ( self . connected , True , "<STR_LIT>" ) <EOL> self . cv . acquire ( ) <EOL> ret = zookeeper . acreate ( self . handle , "<STR_LIT>" , "<STR_LIT>" , <EOL> [ ZOO_OPEN_ACL_UNSAFE ] , zookeeper . EPHEMERAL , <EOL> callback ) <EOL> self . assertEqual ( ret , zookeeper . OK , "<STR_LIT>" ) <EOL> while not self . callback_flag : <EOL> self . cv . wait ( <NUM_LIT:15> ) <EOL> self . cv . release ( ) <EOL> self . assertEqual ( self . callback_flag , True , "<STR_LIT>" ) <EOL> self . assertEqual ( self . rc , zookeeper . OK ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
