#NEWLINE# def test_for_update( self ) : #NEWLINE# #INDENT# u1 = User . create ( username = 'u1' ) #NEWLINE# u2 = User . create ( username = 'u2' ) #NEWLINE# u3 = User . create ( username = 'u3' ) #NEWLINE# test_db . set_autocommit ( False ) #NEWLINE# users = User . select ( ) . where ( ( User . username == 'u1' ) ) . for_update ( ) #NEWLINE# updated = User . update ( username = 'u1_edited' ) . where ( ( User . username == 'u1' ) ) . execute ( ) #NEWLINE# self . assertEqual ( updated , 1 ) #NEWLINE# new_db = self . new_connection ( ) #NEWLINE# res = new_db . execute_sql ( ( 'select username from users where id = %s;' % u1 . id ) ) #NEWLINE# username = res . fetchone ( ) [ 0 ] #NEWLINE# self . assertEqual ( username , 'u1' ) #NEWLINE# test_db . commit ( ) #NEWLINE# res = new_db . execute_sql ( ( 'select username from users where id = %s;' % u3 . id ) ) #NEWLINE# username = res . fetchone ( ) [ 0 ] #NEWLINE# self . assertEqual ( username , 'u1_edited' )