[Sqlalchemy-commits] sqlalchemy: move most tests that are specific to SessionTransact...
Brought to you by:
zzzeek
Menu
▾
▴
[Sqlalchemy-commits] sqlalchemy: move most tests that are specific to SessionTransact...
|
From: <co...@sq...> - 2012-01-28 01:50:24
|
details: http://hg.sqlalchemy.org/sqlalchemy/sqlalchemy/rev/1bccc8036d3c changeset: 8051:1bccc8036d3c user: Mike Bayer <mi...@zz...> date: Fri Jan 27 20:50:06 2012 -0500 description: move most tests that are specific to SessionTransaction over to test_transaction diffstat: test/orm/test_session.py | 382 ------------------------------------ test/orm/test_transaction.py | 452 ++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 430 insertions(+), 404 deletions(-) diffs (truncated from 982 to 300 lines): diff -r 5b231b7d7348 -r 1bccc8036d3c test/orm/test_session.py --- a/test/orm/test_session.py Fri Jan 27 20:32:52 2012 -0500 +++ b/test/orm/test_session.py Fri Jan 27 20:50:06 2012 -0500 @@ -50,29 +50,6 @@ s.close() c.execute("select * from users") - def test_no_close_transaction_on_flulsh(self): - User, users = self.classes.User, self.tables.users - - c = testing.db.connect() - try: - mapper(User, users) - s = create_session(bind=c) - s.begin() - tran = s.transaction - s.add(User(name='first')) - s.flush() - c.execute("select * from users") - u = User(name='two') - s.add(u) - s.flush() - u = User(name='third') - s.add(u) - s.flush() - assert s.transaction is tran - tran.close() - finally: - c.close() - def test_object_session_raises(self): User = self.classes.User @@ -455,365 +432,6 @@ {'id':7}), 7) - @engines.close_open_connections - def test_subtransaction_on_external(self): - users, User = self.tables.users, self.classes.User - - mapper(User, users) - conn = testing.db.connect() - trans = conn.begin() - sess = create_session(bind=conn, autocommit=False, autoflush=True) - sess.begin(subtransactions=True) - u = User(name='ed') - sess.add(u) - sess.flush() - sess.commit() # commit does nothing - trans.rollback() # rolls back - assert len(sess.query(User).all()) == 0 - sess.close() - - @testing.requires.savepoints - @engines.close_open_connections - def test_external_nested_transaction(self): - users, User = self.tables.users, self.classes.User - - mapper(User, users) - try: - conn = testing.db.connect() - trans = conn.begin() - sess = create_session(bind=conn, autocommit=False, autoflush=True) - u1 = User(name='u1') - sess.add(u1) - sess.flush() - - sess.begin_nested() - u2 = User(name='u2') - sess.add(u2) - sess.flush() - sess.rollback() - - trans.commit() - assert len(sess.query(User).all()) == 1 - except: - conn.close() - raise - - @testing.requires.savepoints - def test_heavy_nesting(self): - users = self.tables.users - - session = create_session(bind=testing.db) - session.begin() - session.connection().execute(users.insert().values(name='user1' - )) - session.begin(subtransactions=True) - session.begin_nested() - session.connection().execute(users.insert().values(name='user2' - )) - assert session.connection().execute('select count(1) from users' - ).scalar() == 2 - session.rollback() - assert session.connection().execute('select count(1) from users' - ).scalar() == 1 - session.connection().execute(users.insert().values(name='user3' - )) - session.commit() - assert session.connection().execute('select count(1) from users' - ).scalar() == 2 - - @testing.requires.independent_connections - def test_transactions_isolated(self): - User, users = self.classes.User, self.tables.users - - mapper(User, users) - users.delete().execute() - - s1 = create_session(bind=testing.db, autocommit=False) - s2 = create_session(bind=testing.db, autocommit=False) - u1 = User(name='u1') - s1.add(u1) - s1.flush() - - assert s2.query(User).all() == [] - - @testing.requires.two_phase_transactions - def test_twophase(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - # TODO: mock up a failure condition here - # to ensure a rollback succeeds - mapper(User, users) - mapper(Address, addresses) - - engine2 = engines.testing_engine() - sess = create_session(autocommit=True, autoflush=False, twophase=True) - sess.bind_mapper(User, testing.db) - sess.bind_mapper(Address, engine2) - sess.begin() - u1 = User(name='u1') - a1 = Address(email_address='u1@e') - sess.add_all((u1, a1)) - sess.commit() - sess.close() - engine2.dispose() - assert users.count().scalar() == 1 - assert addresses.count().scalar() == 1 - - def test_subtransaction_on_noautocommit(self): - User, users = self.classes.User, self.tables.users - - mapper(User, users) - sess = create_session(autocommit=False, autoflush=True) - sess.begin(subtransactions=True) - u = User(name='u1') - sess.add(u) - sess.flush() - sess.commit() # commit does nothing - sess.rollback() # rolls back - assert len(sess.query(User).all()) == 0 - sess.close() - - @testing.requires.savepoints - def test_nested_transaction(self): - User, users = self.classes.User, self.tables.users - - mapper(User, users) - sess = create_session() - sess.begin() - - u = User(name='u1') - sess.add(u) - sess.flush() - - sess.begin_nested() # nested transaction - - u2 = User(name='u2') - sess.add(u2) - sess.flush() - - sess.rollback() - - sess.commit() - assert len(sess.query(User).all()) == 1 - sess.close() - - @testing.requires.savepoints - def test_nested_autotrans(self): - User, users = self.classes.User, self.tables.users - - mapper(User, users) - sess = create_session(autocommit=False) - u = User(name='u1') - sess.add(u) - sess.flush() - - sess.begin_nested() # nested transaction - - u2 = User(name='u2') - sess.add(u2) - sess.flush() - - sess.rollback() - - sess.commit() - assert len(sess.query(User).all()) == 1 - sess.close() - - @testing.requires.savepoints - def test_nested_transaction_connection_add(self): - users, User = self.tables.users, self.classes.User - - mapper(User, users) - - sess = create_session(autocommit=True) - - sess.begin() - sess.begin_nested() - - u1 = User(name='u1') - sess.add(u1) - sess.flush() - - sess.rollback() - - u2 = User(name='u2') - sess.add(u2) - - sess.commit() - - eq_(set(sess.query(User).all()), set([u2])) - - sess.begin() - sess.begin_nested() - - u3 = User(name='u3') - sess.add(u3) - sess.commit() # commit the nested transaction - sess.rollback() - - eq_(set(sess.query(User).all()), set([u2])) - - sess.close() - - @testing.requires.savepoints - def test_mixed_transaction_control(self): - users, User = self.tables.users, self.classes.User - - mapper(User, users) - - sess = create_session(autocommit=True) - - sess.begin() - sess.begin_nested() - transaction = sess.begin(subtransactions=True) - - sess.add(User(name='u1')) - - transaction.commit() - sess.commit() - sess.commit() - - sess.close() - - eq_(len(sess.query(User).all()), 1) - - t1 = sess.begin() - t2 = sess.begin_nested() - - sess.add(User(name='u2')) - - t2.commit() - assert sess.transaction is t1 - - sess.close() - - @testing.requires.savepoints - def test_mixed_transaction_close(self): - users, User = self.tables.users, self.classes.User - - mapper(User, users) - - sess = create_session(autocommit=False) - - sess.begin_nested() - - sess.add(User(name='u1')) - sess.flush() - - sess.close() - - sess.add(User(name='u2')) - sess.commit() - - sess.close() - - eq_(len(sess.query(User).all()), 1) - - def test_error_on_using_inactive_session_commands(self): - users, User = self.tables.users, self.classes.User - - mapper(User, users) - sess = create_session(autocommit=True) - sess.begin() |