from __future__ import with_statement
-from sqlalchemy.testing import eq_, assert_raises, \
- assert_raises_message, assert_warnings
-from sqlalchemy import *
-from sqlalchemy.orm import attributes
-from sqlalchemy import exc as sa_exc, event
-from sqlalchemy.orm import exc as orm_exc
-from sqlalchemy.orm import *
+from sqlalchemy import (
+ testing, exc as sa_exc, event, String, Column, Table, select, func)
+from sqlalchemy.testing import (
+ fixtures, engines, eq_, assert_raises, assert_raises_message,
+ assert_warnings)
+from sqlalchemy.orm import (
+ exc as orm_exc, Session, mapper, sessionmaker, create_session,
+ relationship, attributes)
from sqlalchemy.testing.util import gc_collect
-from sqlalchemy import testing
-from sqlalchemy.testing import fixtures
-from sqlalchemy.testing import engines
from test.orm._fixtures import FixtureTest
u = User(name='ed')
sess.add(u)
sess.flush()
- sess.commit() # commit does nothing
- trans.rollback() # rolls back
+ sess.commit() # commit does nothing
+ trans.rollback() # rolls back
assert len(sess.query(User).all()) == 0
sess.close()
conn.close()
raise
-
@testing.requires.savepoints
def test_nested_accounting_new_items_removed(self):
User, users = self.classes.User, self.tables.users
session = create_session(bind=testing.db)
session.begin()
- session.connection().execute(users.insert().values(name='user1'
- ))
+ 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.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'
- ))
+ 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
+ assert session.connection().execute(
+ 'select count(1) from users').scalar() == 2
@testing.requires.independent_connections
def test_transactions_isolated(self):
u = User(name='u1')
sess.add(u)
sess.flush()
- sess.commit() # commit does nothing
- sess.rollback() # rolls back
+ sess.commit() # commit does nothing
+ sess.rollback() # rolls back
assert len(sess.query(User).all()) == 0
sess.close()
u3 = User(name='u3')
sess.add(u3)
- sess.commit() # commit the nested transaction
+ sess.commit() # commit the nested transaction
sess.rollback()
eq_(set(sess.query(User).all()), set([u2]))
sess = Session()
to_flush = [User(name='ed'), User(name='jack'), User(name='wendy')]
+
@event.listens_for(sess, "after_flush_postexec")
def add_another_user(session, ctx):
if to_flush:
session.add(to_flush.pop(0))
x = [1]
- @event.listens_for(sess, "after_commit")
+
+ @event.listens_for(sess, "after_commit") # noqa
def add_another_user(session):
x[0] += 1
sess.commit
)
-
def test_error_on_using_inactive_session_commands(self):
users, User = self.tables.users, self.classes.User
trans = sess.begin()
trans.rollback()
assert_raises_message(
- sa_exc.ResourceClosedError,
- "This transaction is closed",
- trans.rollback
- )
+ sa_exc.ResourceClosedError, "This transaction is closed",
+ trans.rollback)
assert_raises_message(
- sa_exc.ResourceClosedError,
- "This transaction is closed",
- trans.commit
- )
+ sa_exc.ResourceClosedError, "This transaction is closed",
+ trans.commit)
def test_deactive_status_check(self):
sess = create_session()
sess, u1 = self._inactive_flushed_session_fixture()
u2 = User(name='u2')
sess.add(u2)
+
def go():
sess.rollback()
assert_warnings(go,
def test_warning_on_using_inactive_session_dirty(self):
sess, u1 = self._inactive_flushed_session_fixture()
u1.name = 'newname'
+
def go():
sess.rollback()
assert_warnings(go,
def test_warning_on_using_inactive_session_delete(self):
sess, u1 = self._inactive_flushed_session_fixture()
sess.delete(u1)
+
def go():
sess.rollback()
assert_warnings(go,
assert session.transaction is not None, \
'autocommit=False should start a new transaction'
+
class _LocalFixture(FixtureTest):
run_setup_mappers = 'once'
run_inserts = None
def setup_mappers(cls):
User, Address = cls.classes.User, cls.classes.Address
users, addresses = cls.tables.users, cls.tables.addresses
- mapper(User, users, properties={
- 'addresses':relationship(Address, backref='user',
- cascade="all, delete-orphan",
- order_by=addresses.c.id),
+ mapper(
+ User, users, properties={
+ 'addresses': relationship(
+ Address, backref='user', cascade="all, delete-orphan",
+ order_by=addresses.c.id),
})
mapper(Address, addresses)
assert u1.name == 'will'
+
class CleanSavepointTest(FixtureTest):
"""test the behavior for [ticket:2452] - rollback on begin_nested()
only expires objects tracked as being modified in that transaction.
@testing.requires.savepoints
def test_rollback_ignores_clean_on_savepoint(self):
- User, users = self.classes.User, self.tables.users
+
def update_fn(s, u2):
u2.name = 'u2modified'
self._run_test(update_fn)
@testing.requires.savepoints
def test_rollback_ignores_clean_on_savepoint_agg_upd_eval(self):
- User, users = self.classes.User, self.tables.users
+ User = self.classes.User
+
def update_fn(s, u2):
- s.query(User).filter_by(name='u2').update(dict(name='u2modified'),
- synchronize_session='evaluate')
+ s.query(User).filter_by(name='u2').update(
+ dict(name='u2modified'), synchronize_session='evaluate')
self._run_test(update_fn)
@testing.requires.savepoints
def test_rollback_ignores_clean_on_savepoint_agg_upd_fetch(self):
- User, users = self.classes.User, self.tables.users
+ User = self.classes.User
+
def update_fn(s, u2):
s.query(User).filter_by(name='u2').update(dict(name='u2modified'),
synchronize_session='fetch')
self._run_test(update_fn)
+
class ContextManagerTest(FixtureTest):
run_inserts = None
__backend__ = True
mapper(User, users)
sess = Session()
+
def go():
with sess.begin_nested():
sess.add(User()) # name can't be null
mapper(User, users)
sess = Session(autocommit=True)
+
def go():
with sess.begin():
sess.add(User()) # name can't be null
def test_expunge_pending_on_rollback(self):
User = self.classes.User
sess = self.session()
- u2= User(name='newuser')
+ u2 = User(name='newuser')
sess.add(u2)
assert u2 in sess
sess.rollback()
def test_trans_pending_cleared_on_commit(self):
User = self.classes.User
sess = self.session()
- u2= User(name='newuser')
+ u2 = User(name='newuser')
sess.add(u2)
assert u2 in sess
sess.commit()
u1.addresses.remove(a1)
s.flush()
- eq_(s.query(Address).filter(Address.email_address=='foo').all(),
- [])
+ eq_(s.query(Address).filter(Address.email_address == 'foo').all(), [])
s.rollback()
assert a1 not in s.deleted
assert u1.addresses == [a1]
sess.commit()
eq_(u1.name, 'newuser')
-
def test_concurrent_commit_pending(self):
User = self.classes.User
s1 = self.session()
s1.commit()
s2 = self.session()
- u2 = s2.query(User).filter(User.name=='edward').one()
+ u2 = s2.query(User).filter(User.name == 'edward').one()
u2.name = 'will'
s2.commit()
assert u1.name == 'will'
+
class TwoPhaseTest(_LocalFixture):
__backend__ = True
assert u not in s
+
class RollbackRecoverTest(_LocalFixture):
__backend__ = True
s.commit()
eq_(
s.query(User).all(),
- [User(id=1, name='edward',
- addresses=[Address(email_address='foober')])]
- )
+ [
+ User(
+ id=1, name='edward',
+ addresses=[Address(email_address='foober')])])
class SavepointTest(_LocalFixture):
u1.name = 'edward'
u2.name = 'jackward'
s.add_all([u3, u4])
- eq_(s.query(User.name).order_by(User.id).all(),
- [('edward',), ('jackward',), ('wendy',), ('foo',)])
+ eq_(
+ s.query(User.name).order_by(User.id).all(),
+ [('edward',), ('jackward',), ('wendy',), ('foo',)])
s.rollback()
assert u1.name == 'ed'
assert u2.name == 'jack'
- eq_(s.query(User.name).order_by(User.id).all(),
- [('ed',), ('jack',)])
+ eq_(
+ s.query(User.name).order_by(User.id).all(),
+ [('ed',), ('jack',)])
s.commit()
assert u1.name == 'ed'
assert u2.name == 'jack'
- eq_(s.query(User.name).order_by(User.id).all(),
- [('ed',), ('jack',)])
+ eq_(s.query(User.name).order_by(User.id).all(), [('ed',), ('jack',)])
@testing.requires.savepoints
def test_savepoint_delete(self):
u1.name = 'edward'
u2.name = 'jackward'
s.add_all([u3, u4])
- eq_(s.query(User.name).order_by(User.id).all(),
- [('edward',), ('jackward',), ('wendy',), ('foo',)])
+ eq_(
+ s.query(User.name).order_by(User.id).all(),
+ [('edward',), ('jackward',), ('wendy',), ('foo',)])
s.commit()
+
def go():
assert u1.name == 'edward'
assert u2.name == 'jackward'
- eq_(s.query(User.name).order_by(User.id).all(),
- [('edward',), ('jackward',), ('wendy',), ('foo',)])
+ eq_(
+ s.query(User.name).order_by(User.id).all(),
+ [('edward',), ('jackward',), ('wendy',), ('foo',)])
self.assert_sql_count(testing.db, go, 1)
s.commit()
- eq_(s.query(User.name).order_by(User.id).all(),
- [('edward',), ('jackward',), ('wendy',), ('foo',)])
+ eq_(
+ s.query(User.name).order_by(User.id).all(),
+ [('edward',), ('jackward',), ('wendy',), ('foo',)])
@testing.requires.savepoints
def test_savepoint_rollback_collections(self):
s.add(u1)
s.commit()
- u1.name='edward'
+ u1.name = 'edward'
u1.addresses.append(Address(email_address='bar'))
s.begin_nested()
u2 = User(name='jack', addresses=[Address(email_address='bat')])
s.add(u2)
- eq_(s.query(User).order_by(User.id).all(),
+ eq_(
+ s.query(User).order_by(User.id).all(),
[
- User(name='edward', addresses=[Address(email_address='foo'),
- Address(email_address='bar')]),
+ User(
+ name='edward',
+ addresses=[
+ Address(email_address='foo'),
+ Address(email_address='bar')]),
User(name='jack', addresses=[Address(email_address='bat')])
- ]
- )
+ ])
s.rollback()
- eq_(s.query(User).order_by(User.id).all(),
+ eq_(
+ s.query(User).order_by(User.id).all(),
[
- User(name='edward', addresses=[Address(email_address='foo'),
- Address(email_address='bar')]),
- ]
- )
+ User(
+ name='edward',
+ addresses=[
+ Address(email_address='foo'),
+ Address(email_address='bar')]),
+ ])
s.commit()
- eq_(s.query(User).order_by(User.id).all(),
+ eq_(
+ s.query(User).order_by(User.id).all(),
[
- User(name='edward', addresses=[Address(email_address='foo'),
- Address(email_address='bar')]),
+ User(
+ name='edward',
+ addresses=[
+ Address(email_address='foo'),
+ Address(email_address='bar')]),
]
)
s.add(u1)
s.commit()
- u1.name='edward'
+ u1.name = 'edward'
u1.addresses.append(Address(email_address='bar'))
s.begin_nested()
u2 = User(name='jack', addresses=[Address(email_address='bat')])
s.add(u2)
- eq_(s.query(User).order_by(User.id).all(),
+ eq_(
+ s.query(User).order_by(User.id).all(),
[
- User(name='edward', addresses=[Address(email_address='foo'), Address(email_address='bar')]),
+ User(
+ name='edward',
+ addresses=[
+ Address(email_address='foo'),
+ Address(email_address='bar')]),
User(name='jack', addresses=[Address(email_address='bat')])
]
)
s.commit()
- eq_(s.query(User).order_by(User.id).all(),
+ eq_(
+ s.query(User).order_by(User.id).all(),
[
- User(name='edward', addresses=[Address(email_address='foo'), Address(email_address='bar')]),
+ User(
+ name='edward',
+ addresses=[
+ Address(email_address='foo'),
+ Address(email_address='bar')]),
User(name='jack', addresses=[Address(email_address='bat')])
]
)
s.commit()
- eq_(s.query(User).order_by(User.id).all(),
+ eq_(
+ s.query(User).order_by(User.id).all(),
[
- User(name='edward', addresses=[Address(email_address='foo'), Address(email_address='bar')]),
+ User(
+ name='edward',
+ addresses=[
+ Address(email_address='foo'),
+ Address(email_address='bar')]),
User(name='jack', addresses=[Address(email_address='bat')])
]
)
sess = self.session()
sess.begin_nested()
- u2= User(name='newuser')
+ u2 = User(name='newuser')
sess.add(u2)
assert u2 in sess
sess.rollback()
sess.add(u1)
sess.commit()
- testing.db.execute(users.update(users.c.name=='ed').values(name='edward'))
+ testing.db.execute(
+ users.update(users.c.name == 'ed').values(name='edward'))
assert u1.name == 'ed'
sess.expire_all()
u1.name = 'edwardo'
sess.rollback()
- testing.db.execute(users.update(users.c.name=='ed').values(name='edward'))
+ testing.db.execute(
+ users.update(users.c.name == 'ed').values(name='edward'))
assert u1.name == 'edwardo'
sess.expire_all()
u1.name = 'edwardo'
sess.rollback()
- testing.db.execute(users.update(users.c.name=='ed').values(name='edward'))
+ testing.db.execute(
+ users.update(users.c.name == 'ed').values(name='edward'))
assert u1.name == 'edwardo'
sess.commit()
- assert testing.db.execute(select([users.c.name])).fetchall() == [('edwardo',)]
+ assert testing.db.execute(select([users.c.name])).fetchall() == \
+ [('edwardo',)]
assert u1.name == 'edwardo'
sess.delete(u1)
def test_preflush_no_accounting(self):
User, users = self.classes.User, self.tables.users
- sess = Session(_enable_transaction_accounting=False,
- autocommit=True, autoflush=False)
+ sess = Session(
+ _enable_transaction_accounting=False, autocommit=True,
+ autoflush=False)
u1 = User(name='ed')
sess.add(u1)
sess.flush()
sess.rollback()
sess.begin()
- assert testing.db.execute(select([users.c.name])).fetchall() == [('ed',)]
+ assert testing.db.execute(select([users.c.name])).fetchall() == \
+ [('ed',)]
class AutoCommitTest(_LocalFixture):
sess = create_session(autocommit=True)
fail = False
+
def fail_fn(*arg, **kw):
if fail:
raise Exception("commit fails")
sess = create_session(autocommit=True)
fail = False
+
def fail_fn(*arg, **kw):
if fail:
raise Exception("commit fails")
assert 'id' not in u1.__dict__
eq_(u1.id, 3)
+
class NaturalPKRollbackTest(fixtures.MappedTest):
__backend__ = True
@classmethod
def define_tables(cls, metadata):
- Table('users', metadata,
- Column('name', String(50), primary_key=True)
- )
+ Table('users', metadata, Column('name', String(50), primary_key=True))
@classmethod
def setup_classes(cls):
session = sessionmaker()()
- u1, u2, u3= \
- User(name='u1'),\
- User(name='u2'),\
- User(name='u3')
+ u1, u2, u3 = User(name='u1'), User(name='u2'), User(name='u3')
session.add_all([u1, u2, u3])
assert u2 not in s
assert s.identity_map[(User, ('u1',))] is u1
-
-
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
