1
import testenv; testenv.configure_for_tests()
3
from sqlalchemy import *
4
from sqlalchemy.orm import attributes
5
from sqlalchemy import exc as sa_exc
6
from sqlalchemy.orm import *
8
from testlib.fixtures import *
11
class TransactionTest(FixtureTest):
13
session = sessionmaker()
15
def setup_mappers(self):
16
mapper(User, users, properties={
17
'addresses':relation(Address, backref='user',
18
cascade="all, delete-orphan"),
20
mapper(Address, addresses)
23
class FixtureDataTest(TransactionTest):
26
def test_attrs_on_rollback(self):
28
u1 = sess.query(User).get(7)
31
self.assertEquals(u1.name, 'jack')
33
def test_commit_persistent(self):
35
u1 = sess.query(User).get(7)
39
self.assertEquals(u1.name, 'ed')
41
def test_concurrent_commit_persistent(self):
43
u1 = s1.query(User).get(7)
48
u2 = s2.query(User).get(7)
49
assert u2.name == 'ed'
53
assert u1.name == 'will'
55
class AutoExpireTest(TransactionTest):
58
def test_expunge_pending_on_rollback(self):
60
u2= User(name='newuser')
66
def test_trans_pending_cleared_on_commit(self):
68
u2= User(name='newuser')
73
u3 = User(name='anotheruser')
79
def test_update_deleted_on_rollback(self):
85
# this actually tests that the delete() operation,
86
# when cascaded to the "addresses" collection, does not
87
# trigger a flush (via lazyload) before the cascade is complete.
89
assert u1 in s.deleted
92
assert u1 not in s.deleted
94
def test_gced_delete_on_rollback(self):
101
u1_state = attributes.instance_state(u1)
102
assert u1_state in s.identity_map.all_states()
103
assert u1_state in s._deleted
105
assert u1_state not in s.identity_map.all_states()
106
assert u1_state not in s._deleted
109
assert u1_state.obj() is None
112
assert u1_state in s.identity_map.all_states()
113
u1 = s.query(User).filter_by(name='ed').one()
114
assert u1_state not in s.identity_map.all_states()
115
assert s.scalar(users.count()) == 1
118
assert s.scalar(users.count()) == 0
121
def test_trans_deleted_cleared_on_rollback(self):
133
def test_update_deleted_on_rollback_cascade(self):
135
u1 = User(name='ed', addresses=[Address(email_address='foo')])
140
assert u1 in s.deleted
141
assert u1.addresses[0] in s.deleted
144
assert u1 not in s.deleted
145
assert u1.addresses[0] not in s.deleted
147
def test_update_deleted_on_rollback_orphan(self):
149
u1 = User(name='ed', addresses=[Address(email_address='foo')])
154
u1.addresses.remove(a1)
157
self.assertEquals(s.query(Address).filter(Address.email_address=='foo').all(), [])
159
assert a1 not in s.deleted
160
assert u1.addresses == [a1]
162
def test_commit_pending(self):
163
sess = self.session()
164
u1 = User(name='newuser')
168
self.assertEquals(u1.name, 'newuser')
171
def test_concurrent_commit_pending(self):
173
u1 = User(name='edward')
178
u2 = s2.query(User).filter(User.name=='edward').one()
182
assert u1.name == 'will'
184
class TwoPhaseTest(TransactionTest):
187
@testing.requires.two_phase_transactions
188
def test_rollback_on_prepare(self):
189
s = self.session(twophase=True)
198
class RollbackRecoverTest(TransactionTest):
201
def test_pk_violation(self):
203
a1 = Address(email_address='foo')
204
u1 = User(id=1, name='ed', addresses=[a1])
208
a2 = Address(email_address='bar')
209
u2 = User(id=1, name='jack', addresses=[a2])
212
a1.email_address = 'foober'
214
self.assertRaises(sa_exc.FlushError, s.commit)
215
self.assertRaises(sa_exc.InvalidRequestError, s.commit)
221
assert u1.name == 'ed'
222
assert a1.email_address == 'foo'
224
a1.email_address = 'foober'
228
[User(id=1, name='edward', addresses=[Address(email_address='foober')])]
231
@testing.requires.savepoints
232
def test_pk_violation_with_savepoint(self):
234
a1 = Address(email_address='foo')
235
u1 = User(id=1, name='ed', addresses=[a1])
239
a2 = Address(email_address='bar')
240
u2 = User(id=1, name='jack', addresses=[a2])
243
a1.email_address = 'foober'
246
self.assertRaises(sa_exc.FlushError, s.commit)
247
self.assertRaises(sa_exc.InvalidRequestError, s.commit)
255
assert s.query(User).all() == [User(id=1, name='edward', addresses=[Address(email_address='foober')])]
258
class SavepointTest(TransactionTest):
262
@testing.requires.savepoints
263
def test_savepoint_rollback(self):
266
u2 = User(name='jack')
270
u3 = User(name='wendy')
271
u4 = User(name='foo')
275
self.assertEquals(s.query(User.name).order_by(User.id).all(), [('edward',), ('jackward',), ('wendy',), ('foo',)])
277
assert u1.name == 'ed'
278
assert u2.name == 'jack'
279
self.assertEquals(s.query(User.name).order_by(User.id).all(), [('ed',), ('jack',)])
281
assert u1.name == 'ed'
282
assert u2.name == 'jack'
283
self.assertEquals(s.query(User.name).order_by(User.id).all(), [('ed',), ('jack',)])
285
@testing.requires.savepoints
286
def test_savepoint_delete(self):
291
self.assertEquals(s.query(User).filter_by(name='ed').count(), 1)
295
self.assertEquals(s.query(User).filter_by(name='ed').count(), 0)
298
@testing.requires.savepoints
299
def test_savepoint_commit(self):
302
u2 = User(name='jack')
306
u3 = User(name='wendy')
307
u4 = User(name='foo')
311
self.assertEquals(s.query(User.name).order_by(User.id).all(), [('edward',), ('jackward',), ('wendy',), ('foo',)])
314
assert u1.name == 'edward'
315
assert u2.name == 'jackward'
316
self.assertEquals(s.query(User.name).order_by(User.id).all(), [('edward',), ('jackward',), ('wendy',), ('foo',)])
317
self.assert_sql_count(testing.db, go, 1)
320
self.assertEquals(s.query(User.name).order_by(User.id).all(), [('edward',), ('jackward',), ('wendy',), ('foo',)])
322
@testing.requires.savepoints
323
def test_savepoint_rollback_collections(self):
325
u1 = User(name='ed', addresses=[Address(email_address='foo')])
330
u1.addresses.append(Address(email_address='bar'))
332
u2 = User(name='jack', addresses=[Address(email_address='bat')])
334
self.assertEquals(s.query(User).order_by(User.id).all(),
336
User(name='edward', addresses=[Address(email_address='foo'), Address(email_address='bar')]),
337
User(name='jack', addresses=[Address(email_address='bat')])
341
self.assertEquals(s.query(User).order_by(User.id).all(),
343
User(name='edward', addresses=[Address(email_address='foo'), Address(email_address='bar')]),
347
self.assertEquals(s.query(User).order_by(User.id).all(),
349
User(name='edward', addresses=[Address(email_address='foo'), Address(email_address='bar')]),
353
@testing.requires.savepoints
354
def test_savepoint_commit_collections(self):
356
u1 = User(name='ed', addresses=[Address(email_address='foo')])
361
u1.addresses.append(Address(email_address='bar'))
363
u2 = User(name='jack', addresses=[Address(email_address='bat')])
365
self.assertEquals(s.query(User).order_by(User.id).all(),
367
User(name='edward', addresses=[Address(email_address='foo'), Address(email_address='bar')]),
368
User(name='jack', addresses=[Address(email_address='bat')])
372
self.assertEquals(s.query(User).order_by(User.id).all(),
374
User(name='edward', addresses=[Address(email_address='foo'), Address(email_address='bar')]),
375
User(name='jack', addresses=[Address(email_address='bat')])
379
self.assertEquals(s.query(User).order_by(User.id).all(),
381
User(name='edward', addresses=[Address(email_address='foo'), Address(email_address='bar')]),
382
User(name='jack', addresses=[Address(email_address='bat')])
386
@testing.requires.savepoints
387
def test_expunge_pending_on_rollback(self):
388
sess = self.session()
391
u2= User(name='newuser')
395
assert u2 not in sess
397
@testing.requires.savepoints
398
def test_update_deleted_on_rollback(self):
406
assert u1 in s.deleted
409
assert u1 not in s.deleted
412
class AccountingFlagsTest(TransactionTest):
414
def test_no_expire_on_commit(self):
415
sess = sessionmaker(expire_on_commit=False)()
420
testing.db.execute(users.update(users.c.name=='ed').values(name='edward'))
422
assert u1.name == 'ed'
424
assert u1.name == 'edward'
426
def test_rollback_no_accounting(self):
427
sess = sessionmaker(_enable_transaction_accounting=False)()
435
testing.db.execute(users.update(users.c.name=='ed').values(name='edward'))
437
assert u1.name == 'edwardo'
439
assert u1.name == 'edward'
441
def test_commit_no_accounting(self):
442
sess = sessionmaker(_enable_transaction_accounting=False)()
450
testing.db.execute(users.update(users.c.name=='ed').values(name='edward'))
452
assert u1.name == 'edwardo'
455
assert testing.db.execute(select([users.c.name])).fetchall() == [('edwardo',)]
456
assert u1.name == 'edwardo'
461
def test_preflush_no_accounting(self):
462
sess = sessionmaker(_enable_transaction_accounting=False, autocommit=True)()
469
u2 = User(name="some other user")
475
assert testing.db.execute(select([users.c.name])).fetchall() == [('ed',)]
478
class AutocommitTest(TransactionTest):
479
def test_begin_nested_requires_trans(self):
480
sess = create_session(autocommit=True)
481
self.assertRaises(sa_exc.InvalidRequestError, sess.begin_nested)
483
def test_begin_preflush(self):
484
sess = create_session(autocommit=True)
490
u2 = User(name='some other user')
493
assert u2 not in sess
495
assert sess.query(User).filter_by(name='ed').one() is u1
500
if __name__ == '__main__':