<s> from djangae . test import TestCase <EOL> from djangae . db import transaction <EOL> from djangae . contrib import sleuth <EOL> class TransactionTests ( TestCase ) : <EOL> def test_repeated_usage_in_a_loop ( self ) : <EOL> from . test_connector import TestUser <EOL> pk = TestUser . objects . create ( username = "<STR_LIT:foo>" ) . pk <EOL> for i in xrange ( <NUM_LIT:4> ) : <EOL> with transaction . atomic ( xg = True ) : <EOL> TestUser . objects . get ( pk = pk ) <EOL> continue <EOL> with transaction . atomic ( xg = True ) : <EOL> TestUser . objects . get ( pk = pk ) <EOL> def test_atomic_decorator ( self ) : <EOL> from . test_connector import TestUser <EOL> @ transaction . atomic <EOL> def txn ( ) : <EOL> TestUser . objects . create ( username = "<STR_LIT:foo>" , field2 = "<STR_LIT:bar>" ) <EOL> self . assertTrue ( transaction . in_atomic_block ( ) ) <EOL> raise ValueError ( ) <EOL> with self . assertRaises ( ValueError ) : <EOL> txn ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , TestUser . objects . count ( ) ) <EOL> def test_interaction_with_datastore_txn ( self ) : <EOL> from google . appengine . ext import db <EOL> from google . appengine . datastore . datastore_rpc import TransactionOptions <EOL> from . test_connector import TestUser <EOL> @ db . transactional ( propagation = TransactionOptions . INDEPENDENT ) <EOL> def some_indie_txn ( _username ) : <EOL> TestUser . objects . create ( username = _username ) <EOL> @ db . transactional ( ) <EOL> def some_non_indie_txn ( _username ) : <EOL> TestUser . objects . create ( username = _username ) <EOL> @ db . transactional ( ) <EOL> def double_nested_transactional ( ) : <EOL> @ db . transactional ( propagation = TransactionOptions . INDEPENDENT ) <EOL> def do_stuff ( ) : <EOL> TestUser . objects . create ( username = "<STR_LIT>" ) <EOL> raise ValueError ( ) <EOL> try : <EOL> return do_stuff <EOL> except : <EOL> return <EOL> with transaction . atomic ( ) : <EOL> double_nested_transactional ( ) <EOL> @ db . transactional ( ) <EOL> def something_containing_atomic ( ) : <EOL> with transaction . atomic ( ) : <EOL> TestUser . objects . create ( username = "<STR_LIT>" ) <EOL> something_containing_atomic ( ) <EOL> with transaction . atomic ( ) : <EOL> with transaction . atomic ( ) : <EOL> some_non_indie_txn ( "<STR_LIT>" ) <EOL> some_indie_txn ( "<STR_LIT>" ) <EOL> some_indie_txn ( "<STR_LIT>" ) <EOL> with transaction . atomic ( independent = True ) : <EOL> some_non_indie_txn ( "<STR_LIT>" ) <EOL> some_indie_txn ( "<STR_LIT>" ) <EOL> some_indie_txn ( "<STR_LIT>" ) <EOL> def test_atomic_context_manager ( self ) : <EOL> from . test_connector import TestUser <EOL> with self . assertRaises ( ValueError ) : <EOL> with transaction . atomic ( ) : <EOL> TestUser . objects . create ( username = "<STR_LIT:foo>" , field2 = "<STR_LIT:bar>" ) <EOL> raise ValueError ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , TestUser . objects . count ( ) ) <EOL> def test_non_atomic_context_manager ( self ) : <EOL> from . test_connector import TestUser <EOL> existing = TestUser . objects . create ( username = "<STR_LIT>" , field2 = "<STR_LIT>" ) <EOL> with transaction . atomic ( ) : <EOL> self . assertTrue ( transaction . in_atomic_block ( ) ) <EOL> user = TestUser . objects . create ( username = "<STR_LIT:foo>" , field2 = "<STR_LIT:bar>" ) <EOL> with transaction . non_atomic ( ) : <EOL> self . assertRaises ( TestUser . DoesNotExist , TestUser . objects . get , pk = user . pk ) <EOL> self . assertFalse ( transaction . in_atomic_block ( ) ) <EOL> with sleuth . watch ( "<STR_LIT>" ) as datastore_get : <EOL> TestUser . objects . get ( pk = existing . pk ) <EOL> self . assertFalse ( datastore_get . called ) <EOL> with transaction . atomic ( independent = True ) : <EOL> user2 = TestUser . objects . create ( username = "<STR_LIT>" , field2 = "<STR_LIT>" ) <EOL> self . assertTrue ( transaction . in_atomic_block ( ) ) <EOL> with transaction . non_atomic ( ) : <EOL> self . assertFalse ( transaction . in_atomic_block ( ) ) <EOL> self . assertRaises ( TestUser . DoesNotExist , TestUser . objects . get , pk = user2 . pk ) <EOL> with transaction . non_atomic ( ) : <EOL> self . assertFalse ( transaction . in_atomic_block ( ) ) <EOL> self . assertRaises ( TestUser . DoesNotExist , TestUser . objects . get , pk = user2 . pk ) <EOL> with sleuth . watch ( "<STR_LIT>" ) as datastore_get : <EOL> TestUser . objects . get ( pk = existing . pk ) <EOL> self . assertFalse ( transaction . in_atomic_block ( ) ) <EOL> self . assertRaises ( TestUser . DoesNotExist , TestUser . objects . get , pk = user2 . pk ) <EOL> self . assertTrue ( TestUser . objects . filter ( pk = user2 . pk ) . exists ( ) ) <EOL> self . assertTrue ( transaction . in_atomic_block ( ) ) <EOL> def test_xg_argument ( self ) : <EOL> from . test_connector import TestUser , TestFruit <EOL> @ transaction . atomic ( xg = True ) <EOL> def txn ( _username ) : <EOL> TestUser . objects . create ( username = _username , field2 = "<STR_LIT:bar>" ) <EOL> TestFruit . objects . create ( name = "<STR_LIT>" , color = "<STR_LIT>" ) <EOL> raise ValueError ( ) <EOL> with self . assertRaises ( ValueError ) : <EOL> txn ( "<STR_LIT:foo>" ) <EOL> self . assertEqual ( <NUM_LIT:0> , TestUser . objects . count ( ) ) <EOL> self . assertEqual ( <NUM_LIT:0> , TestFruit . objects . count ( ) ) <EOL> def test_independent_argument ( self ) : <EOL> """<STR_LIT>""" <EOL> from . test_connector import TestUser , TestFruit <EOL> @ transaction . atomic <EOL> def txn1 ( _username , _fruit ) : <EOL> @ transaction . atomic ( independent = True ) <EOL> def txn2 ( _fruit ) : <EOL> TestFruit . objects . create ( name = _fruit , color = "<STR_LIT>" ) <EOL> raise ValueError ( ) <EOL> TestUser . objects . create ( username = _username ) <EOL> txn2 ( _fruit ) <EOL> with self . assertRaises ( ValueError ) : <EOL> txn1 ( "<STR_LIT:test>" , "<STR_LIT>" ) <EOL> def test_nested_decorator ( self ) : <EOL> @ transaction . atomic <EOL> def inner_txn ( ) : <EOL> pass <EOL> @ transaction . atomic <EOL> def outer_txn ( ) : <EOL> inner_txn ( ) <EOL> inner_txn ( ) <EOL> outer_txn ( ) </s>
