<s> """<STR_LIT>""" <EOL> import os <EOL> import sys <EOL> import unittest <EOL> import kazoo . client <EOL> import kazoo . exceptions <EOL> import kazoo . protocol <EOL> import kazoo . protocol . states <EOL> sys . path . append ( os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) ) <EOL> from zkappscale import zktransaction as zk <EOL> PATH_SEPARATOR = "<STR_LIT:/>" <EOL> TOP_LEVEL = "<STR_LIT>" <EOL> class TestZookeeperTransaction ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . appid = '<STR_LIT>' <EOL> def test_successful_transaction ( self ) : <EOL> zoo = zk . ZKTransaction ( ) <EOL> txid = zoo . get_transaction_id ( self . appid , is_xg = False ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> self . assertEquals ( True , zoo . release_lock ( self . appid , txid ) ) <EOL> def test_two_locks_transaction ( self ) : <EOL> zoo = zk . ZKTransaction ( ) <EOL> txid = zoo . get_transaction_id ( self . appid , is_xg = False ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> self . assertRaises ( zk . ZKTransactionException , zoo . acquire_lock , <EOL> self . appid , txid , "<STR_LIT>" ) <EOL> def test_no_lock_transaction ( self ) : <EOL> zoo = zk . ZKTransaction ( ) <EOL> txid = zoo . get_transaction_id ( self . appid , is_xg = False ) <EOL> self . assertEquals ( True , zoo . release_lock ( self . appid , txid ) ) <EOL> def test_no_lock_with_notify_failed_transaction ( self ) : <EOL> zoo = zk . ZKTransaction ( ) <EOL> txid = zoo . get_transaction_id ( self . appid , is_xg = False ) <EOL> zoo . notify_failed_transaction ( self . appid , txid ) <EOL> self . assertRaises ( zk . ZKTransactionException , zoo . release_lock , self . appid , txid ) <EOL> def test_xg_successful_transaction ( self ) : <EOL> zoo = zk . ZKTransaction ( ) <EOL> txid = zoo . get_transaction_id ( self . appid , is_xg = True ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> self . assertEquals ( True , zoo . release_lock ( self . appid , txid ) ) <EOL> def test_xg_failed_transaction ( self ) : <EOL> zoo = zk . ZKTransaction ( ) <EOL> txid = zoo . get_transaction_id ( self . appid , is_xg = True ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> self . assertRaises ( zk . ZKTransactionException , zoo . acquire_lock , <EOL> self . appid , txid , "<STR_LIT>" ) <EOL> def test_versions_on_success ( self ) : <EOL> zoo = zk . ZKTransaction ( ) <EOL> txid = zoo . get_transaction_id ( self . appid , is_xg = True ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> current_id_1 = zoo . get_valid_transaction_id ( self . appid , txid , "<STR_LIT>" ) <EOL> current_id_2 = zoo . get_valid_transaction_id ( self . appid , txid , "<STR_LIT>" ) <EOL> zoo . register_updated_key ( self . appid , current_id_1 , txid , "<STR_LIT>" ) <EOL> zoo . register_updated_key ( self . appid , current_id_2 , txid , "<STR_LIT>" ) <EOL> self . assertEquals ( txid , zoo . get_valid_transaction_id ( self . appid , txid , "<STR_LIT>" ) ) <EOL> self . assertEquals ( txid , zoo . get_valid_transaction_id ( self . appid , txid , "<STR_LIT>" ) ) <EOL> self . assertEquals ( True , zoo . release_lock ( self . appid , txid ) ) <EOL> def test_versions_on_failure ( self ) : <EOL> zoo = zk . ZKTransaction ( ) <EOL> txid = zoo . get_transaction_id ( self . appid , is_xg = True ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> zoo . acquire_lock ( self . appid , txid , "<STR_LIT>" ) <EOL> current_id_1 = zoo . get_valid_transaction_id ( self . appid , txid , "<STR_LIT>" ) <EOL> current_id_2 = zoo . get_valid_transaction_id ( self . appid , txid , "<STR_LIT>" ) <EOL> zoo . register_updated_key ( self . appid , current_id_1 , txid , "<STR_LIT>" ) <EOL> zoo . register_updated_key ( self . appid , current_id_2 , txid , "<STR_LIT>" ) <EOL> zoo . notify_failed_transaction ( self . appid , txid ) <EOL> self . assertEquals ( current_id_1 , <EOL> zoo . get_valid_transaction_id ( self . appid , txid , "<STR_LIT>" ) ) <EOL> self . assertEquals ( current_id_2 , <EOL> zoo . get_valid_transaction_id ( self . appid , txid , "<STR_LIT>" ) ) <EOL> self . assertRaises ( zk . ZKTransactionException , zoo . release_lock , self . appid , txid ) <EOL> def tearDown ( self ) : <EOL> def delete_recursive ( handle , path ) : <EOL> try : <EOL> children = handle . get_children ( path ) <EOL> for child in children : <EOL> delete_recursive ( handle , PATH_SEPARATOR . join ( [ path , child ] ) ) <EOL> handle . delete ( path ) <EOL> except kazoo . exceptions . NoNodeError : <EOL> pass <EOL> handle = kazoo . client . KazooClient ( hosts = "<STR_LIT>" ) <EOL> delete_recursive ( handle , TOP_LEVEL ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
