<s> import random <EOL> from stdnet import odm , InvalidTransaction <EOL> from examples . models import SimpleModel , Dictionary <EOL> from stdnet . utils import test , populate <EOL> LEN = <NUM_LIT:100> <EOL> names = populate ( '<STR_LIT:string>' , LEN , min_len = <NUM_LIT:5> , max_len = <NUM_LIT:20> ) <EOL> class TransactionReceiver ( object ) : <EOL> def __init__ ( self ) : <EOL> self . transactions = [ ] <EOL> def __call__ ( self , signal , sender , instances = None , session = None , <EOL> ** kwargs ) : <EOL> self . transactions . append ( ( sender , instances ) ) <EOL> class TestTransactions ( test . TestWrite ) : <EOL> model = SimpleModel <EOL> def setUp ( self ) : <EOL> models = self . mapper <EOL> self . receiver = TransactionReceiver ( ) <EOL> models . post_commit . bind ( self . receiver , self . model ) <EOL> def testCreate ( self ) : <EOL> session = self . session ( ) <EOL> query = session . query ( self . model ) <EOL> with session . begin ( ) as t : <EOL> self . assertEqual ( t . session , session ) <EOL> s = t . add ( self . model ( code = '<STR_LIT:test>' , description = '<STR_LIT>' ) ) <EOL> self . assertFalse ( s . id ) <EOL> s2 = session . add ( self . model ( code = '<STR_LIT>' , <EOL> description = '<STR_LIT>' ) ) <EOL> yield t . on_result <EOL> all = yield query . filter ( code = ( '<STR_LIT:test>' , '<STR_LIT>' ) ) . all ( ) <EOL> self . assertEqual ( len ( all ) , <NUM_LIT:2> ) <EOL> receiver = self . receiver <EOL> self . assertTrue ( len ( receiver . transactions ) , <NUM_LIT:1> ) <EOL> sender , instances = receiver . transactions [ <NUM_LIT:0> ] <EOL> self . assertEqual ( sender , self . model ) <EOL> self . assertTrue ( instances ) <EOL> self . assertEqual ( instances , all ) <EOL> def testDelete ( self ) : <EOL> session = self . session ( ) <EOL> query = session . query ( self . model ) <EOL> with session . begin ( ) as t : <EOL> s = session . add ( self . model ( code = '<STR_LIT>' , <EOL> description = '<STR_LIT>' ) ) <EOL> yield t . on_result <EOL> yield self . async . assertEqual ( query . get ( id = s . id ) , s ) <EOL> yield session . delete ( s ) <EOL> yield self . async . assertRaises ( self . model . DoesNotExist , <EOL> query . get , id = s . id ) <EOL> def test_force_update ( self ) : <EOL> session = self . session ( ) <EOL> with session . begin ( ) as t : <EOL> s = session . add ( self . model ( code = '<STR_LIT>' , <EOL> description = '<STR_LIT>' ) ) <EOL> yield t . on_result <EOL> with session . begin ( ) as t : <EOL> s = t . add ( s , force_update = True ) <EOL> state = s . get_state ( ) <EOL> self . assertEqual ( state . action , '<STR_LIT>' ) <EOL> self . assertTrue ( state . persistent ) <EOL> yield t . on_result <EOL> class TestMultiFieldTransaction ( test . TestCase ) : <EOL> model = Dictionary <EOL> def make ( self ) : <EOL> with self . session ( ) . begin ( name = '<STR_LIT>' ) as t : <EOL> self . assertEqual ( t . name , '<STR_LIT>' ) <EOL> for name in names : <EOL> t . add ( self . model ( name = name ) ) <EOL> return t . on_result <EOL> def testHashField ( self ) : <EOL> yield self . make ( ) <EOL> session = self . session ( ) <EOL> query = session . query ( self . model ) <EOL> d1 , d2 = yield query . filter ( id__in = ( <NUM_LIT:1> , <NUM_LIT:2> ) ) . all ( ) <EOL> with session . begin ( ) as t : <EOL> d1 . data . add ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> d1 . data . add ( '<STR_LIT>' , <NUM_LIT> ) <EOL> d2 . data . add ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> d2 . data . add ( '<STR_LIT:foo>' , <NUM_LIT> ) <EOL> self . assertTrue ( d1 . data . cache . toadd ) <EOL> self . assertTrue ( d2 . data . cache . toadd ) <EOL> yield t . on_result <EOL> self . assertFalse ( d1 . data . cache . toadd ) <EOL> self . assertFalse ( d2 . data . cache . toadd ) <EOL> d1 , d2 = yield query . filter ( id__in = ( <NUM_LIT:1> , <NUM_LIT:2> ) ) . sort_by ( '<STR_LIT:id>' ) . load_related ( '<STR_LIT:data>' ) . all ( ) <EOL> self . assertEqual ( d1 . data [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( d2 . data [ '<STR_LIT>' ] , '<STR_LIT>' ) </s>
