From: Mike Bayer Date: Fri, 21 Dec 2012 18:29:48 +0000 (-0500) Subject: - cleanup on these tests X-Git-Tag: rel_0_8_0~38^2~3 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=cf50ac219e03886940e4e7fef8dadbb2a9ec5c32;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git - cleanup on these tests - add "extend()" to AppenderQuery --- diff --git a/lib/sqlalchemy/orm/dynamic.py b/lib/sqlalchemy/orm/dynamic.py index c1dea95f1c..64353cfafa 100644 --- a/lib/sqlalchemy/orm/dynamic.py +++ b/lib/sqlalchemy/orm/dynamic.py @@ -286,6 +286,12 @@ class AppenderMixin(object): return query + def extend(self, iterator): + for item in iterator: + self.attr.append( + attributes.instance_state(self.instance), + attributes.instance_dict(self.instance), item, None) + def append(self, item): self.attr.append( attributes.instance_state(self.instance), diff --git a/test/orm/test_dynamic.py b/test/orm/test_dynamic.py index 3c4ad4b113..a356a562be 100644 --- a/test/orm/test_dynamic.py +++ b/test/orm/test_dynamic.py @@ -1,69 +1,60 @@ -from sqlalchemy.testing import eq_, ne_ -import operator -from sqlalchemy.orm import dynamic_loader, backref, configure_mappers +from sqlalchemy.testing import eq_ +from sqlalchemy.orm import backref, configure_mappers from sqlalchemy import testing -from sqlalchemy import Integer, String, ForeignKey, desc, select, func, exc -from sqlalchemy.testing.schema import Table, Column -from sqlalchemy.orm import mapper, relationship, create_session, Query, attributes, exc as orm_exc +from sqlalchemy import desc, select, func, exc +from sqlalchemy.orm import mapper, relationship, create_session, Query, \ + attributes, exc as orm_exc from sqlalchemy.orm.dynamic import AppenderMixin -from sqlalchemy.testing import eq_, AssertsCompiledSQL, assert_raises_message, assert_raises -from sqlalchemy.testing import fixtures +from sqlalchemy.testing import AssertsCompiledSQL, \ + assert_raises_message, assert_raises from test.orm import _fixtures -class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL): - def test_basic(self): +class _DynamicFixture(object): + def _user_address_fixture(self, addresses_args={}): users, Address, addresses, User = (self.tables.users, self.classes.Address, self.tables.addresses, self.classes.User) mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses)) + 'addresses': relationship(Address, lazy="dynamic", + **addresses_args) }) + mapper(Address, addresses) + return User, Address + +class DynamicTest(_DynamicFixture, _fixtures.FixtureTest, AssertsCompiledSQL): + + def test_basic(self): + User, Address = self._user_address_fixture() sess = create_session() q = sess.query(User) - u = q.filter(User.id==7).first() - eq_([User(id=7, addresses=[Address(id=1, email_address='jack@bean.com')])], - q.filter(User.id==7).all()) + q.filter(User.id == 7).all()) eq_(self.static.user_address_result, q.all()) def test_statement(self): """test that the .statement accessor returns the actual statement that would render, without any _clones called.""" - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses)) - }) + User, Address = self._user_address_fixture() sess = create_session() q = sess.query(User) - u = q.filter(User.id==7).first() + u = q.filter(User.id == 7).first() self.assert_compile( u.addresses.statement, - "SELECT addresses.id, addresses.user_id, addresses.email_address FROM " + "SELECT addresses.id, addresses.user_id, addresses.email_address " + "FROM " "addresses WHERE :param_1 = addresses.user_id", use_default_dialect=True ) def test_detached_raise(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties={ - 'addresses': dynamic_loader(mapper(Address, addresses)) - }) + User, Address = self._user_address_fixture() sess = create_session() u = sess.query(User).get(8) sess.expunge(u) @@ -74,14 +65,8 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL): ) def test_no_uselist_false(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - mapper(Address, addresses) - mapper(User, users, properties={ - "addresses": relationship(Address, lazy='dynamic', uselist=False) - }) + User, Address = self._user_address_fixture( + addresses_args={"uselist": False}) assert_raises_message( exc.InvalidRequestError, "On relationship User.addresses, 'dynamic' loaders cannot be " @@ -108,68 +93,71 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL): ) def test_order_by(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses)) - }) + User, Address = self._user_address_fixture() sess = create_session() u = sess.query(User).get(8) eq_( list(u.addresses.order_by(desc(Address.email_address))), - [Address(email_address=u'ed@wood.com'), Address(email_address=u'ed@lala.com'), - Address(email_address=u'ed@bettyboop.com')] - ) + [ + Address(email_address=u'ed@wood.com'), + Address(email_address=u'ed@lala.com'), + Address(email_address=u'ed@bettyboop.com') + ] + ) def test_configured_order_by(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) + addresses = self.tables.addresses + User, Address = self._user_address_fixture( + addresses_args={ + "order_by": + addresses.c.email_address.desc()}) - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses), order_by=desc(Address.email_address)) - }) sess = create_session() u = sess.query(User).get(8) - eq_(list(u.addresses), [Address(email_address=u'ed@wood.com'), Address(email_address=u'ed@lala.com'), Address(email_address=u'ed@bettyboop.com')]) + eq_( + list(u.addresses), + [ + Address(email_address=u'ed@wood.com'), + Address(email_address=u'ed@lala.com'), + Address(email_address=u'ed@bettyboop.com') + ] + ) # test cancellation of None, replacement with something else eq_( list(u.addresses.order_by(None).order_by(Address.email_address)), - [Address(email_address=u'ed@bettyboop.com'), Address(email_address=u'ed@lala.com'), Address(email_address=u'ed@wood.com')] + [ + Address(email_address=u'ed@bettyboop.com'), + Address(email_address=u'ed@lala.com'), + Address(email_address=u'ed@wood.com') + ] ) # test cancellation of None, replacement with nothing eq_( set(u.addresses.order_by(None)), - set([Address(email_address=u'ed@bettyboop.com'), Address(email_address=u'ed@lala.com'), Address(email_address=u'ed@wood.com')]) + set([ + Address(email_address=u'ed@bettyboop.com'), + Address(email_address=u'ed@lala.com'), + Address(email_address=u'ed@wood.com') + ]) ) def test_count(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses)) - }) + User, Address = self._user_address_fixture() sess = create_session() u = sess.query(User).first() eq_(u.addresses.count(), 1) - def test_backref(self): + def test_dynamic_on_backref(self): users, Address, addresses, User = (self.tables.users, self.classes.Address, self.tables.addresses, self.classes.User) mapper(Address, addresses, properties={ - 'user':relationship(User, backref=backref('addresses', lazy='dynamic')) + 'user': relationship(User, + backref=backref('addresses', lazy='dynamic')) }) mapper(User, users) @@ -183,14 +171,7 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL): assert ad not in u.addresses def test_no_count(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses)) - }) + User, Address = self._user_address_fixture() sess = create_session() q = sess.query(User) @@ -198,21 +179,19 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL): # returns a live database result), else additional count() queries are # issued when evaluating in a list context def go(): - eq_([User(id=7, - addresses=[Address(id=1, - email_address='jack@bean.com')])], - q.filter(User.id==7).all()) + eq_( + q.filter(User.id == 7).all(), + [ + User(id=7, + addresses=[ + Address(id=1, email_address='jack@bean.com') + ]) + ] + ) self.assert_sql_count(testing.db, go, 2) def test_no_populate(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses)) - }) + User, Address = self._user_address_fixture() u1 = User() assert_raises_message( NotImplementedError, @@ -228,8 +207,11 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL): self.classes.Item) mapper(Order, orders, properties={ - 'items':relationship(Item, secondary=order_items, lazy="dynamic", - backref=backref('orders', lazy="dynamic")) + 'items': relationship(Item, + secondary=order_items, + lazy="dynamic", + backref=backref('orders', lazy="dynamic") + ) }) mapper(Item, items) @@ -244,7 +226,7 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL): assert i1 in o1.items.all() @testing.exclude('mysql', 'between', - ((5, 1,49), (5, 1, 52)), + ((5, 1, 49), (5, 1, 52)), 'https://bugs.launchpad.net/ubuntu/+source/mysql-5.1/+bug/706988') def test_association_nonaliased(self): items, Order, orders, order_items, Item = (self.tables.items, @@ -254,7 +236,8 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL): self.classes.Item) mapper(Order, orders, properties={ - 'items':relationship(Item, secondary=order_items, + 'items': relationship(Item, + secondary=order_items, lazy="dynamic", order_by=order_items.c.item_id) }) @@ -276,39 +259,28 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL): # filter criterion against the secondary table # works eq_( - o.items.filter(order_items.c.item_id==2).all(), + o.items.filter(order_items.c.item_id == 2).all(), [Item(id=2)] ) - - def test_transient_detached(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses)) - }) - sess = create_session() + def test_transient_count(self): + User, Address = self._user_address_fixture() u1 = User() u1.addresses.append(Address()) eq_(u1.addresses.count(), 1) + + def test_transient_access(self): + User, Address = self._user_address_fixture() + u1 = User() + u1.addresses.append(Address()) eq_(u1.addresses[0], Address()) def test_custom_query(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - class MyQuery(Query): pass + User, Address = self._user_address_fixture( + addresses_args={"query_class": MyQuery}) - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses), - query_class=MyQuery) - }) sess = create_session() u = User() sess.add(u) @@ -326,11 +298,6 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL): eq_(type(q).__name__, 'MyQuery') def test_custom_query_with_custom_mixin(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - class MyAppenderMixin(AppenderMixin): def add(self, items): if isinstance(items, list): @@ -345,10 +312,11 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL): class MyAppenderQuery(MyAppenderMixin, MyQuery): query_class = MyQuery - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses), - query_class=MyAppenderQuery) - }) + User, Address = self._user_address_fixture( + addresses_args={ + "query_class": MyAppenderQuery}) + + sess = create_session() u = User() sess.add(u) @@ -368,58 +336,71 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL): eq_(type(q).__name__, 'MyQuery') -class SessionTest(_fixtures.FixtureTest): +class UOWTest(_DynamicFixture, _fixtures.FixtureTest): run_inserts = None - def test_events(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) + def test_persistence(self): + addresses = self.tables.addresses + User, Address = self._user_address_fixture() - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses)) - }) sess = create_session() u1 = User(name='jack') a1 = Address(email_address='foo') sess.add_all([u1, a1]) sess.flush() - eq_(testing.db.scalar(select([func.count(1)]).where(addresses.c.user_id!=None)), 0) + eq_( + testing.db.scalar( + select([func.count(1)]).where(addresses.c.user_id != None) + ), + 0 + ) u1 = sess.query(User).get(u1.id) u1.addresses.append(a1) sess.flush() - eq_(testing.db.execute(select([addresses]).where(addresses.c.user_id!=None)).fetchall(), - [(a1.id, u1.id, 'foo')]) + eq_( + testing.db.execute( + select([addresses]).where(addresses.c.user_id != None) + ).fetchall(), + [(a1.id, u1.id, 'foo')] + ) u1.addresses.remove(a1) sess.flush() - eq_(testing.db.scalar(select([func.count(1)]).where(addresses.c.user_id!=None)), 0) + eq_( + testing.db.scalar( + select([func.count(1)]).where(addresses.c.user_id != None) + ), + 0 + ) u1.addresses.append(a1) sess.flush() - eq_(testing.db.execute(select([addresses]).where(addresses.c.user_id!=None)).fetchall(), - [(a1.id, u1.id, 'foo')]) + eq_( + testing.db.execute( + select([addresses]).where(addresses.c.user_id != None) + ).fetchall(), + [(a1.id, u1.id, 'foo')] + ) a2 = Address(email_address='bar') u1.addresses.remove(a1) u1.addresses.append(a2) sess.flush() - eq_(testing.db.execute(select([addresses]).where(addresses.c.user_id!=None)).fetchall(), - [(a2.id, u1.id, 'bar')]) + eq_( + testing.db.execute( + select([addresses]).where(addresses.c.user_id != None) + ).fetchall(), + [(a2.id, u1.id, 'bar')] + ) def test_merge(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses), order_by=addresses.c.email_address) - }) + addresses = self.tables.addresses + User, Address = self._user_address_fixture( + addresses_args={ + "order_by": addresses.c.email_address}) sess = create_session() u1 = User(name='jack') a1 = Address(email_address='a1') @@ -449,64 +430,20 @@ class SessionTest(_fixtures.FixtureTest): [a1, a3] ) - def test_flush(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses)) - }) - sess = create_session() - u1 = User(name='jack') - u2 = User(name='ed') - u2.addresses.append(Address(email_address='foo@bar.com')) - u1.addresses.append(Address(email_address='lala@hoho.com')) - sess.add_all((u1, u2)) - sess.flush() - - from sqlalchemy.orm import attributes - eq_(attributes.get_history(u1, 'addresses'), ([], [Address(email_address='lala@hoho.com')], [])) - - sess.expunge_all() - - # test the test fixture a little bit - ne_(User(name='jack', addresses=[Address(email_address='wrong')]), - sess.query(User).first()) - eq_(User(name='jack', addresses=[Address(email_address='lala@hoho.com')]), - sess.query(User).first()) - - eq_([ - User(name='jack', addresses=[Address(email_address='lala@hoho.com')]), - User(name='ed', addresses=[Address(email_address='foo@bar.com')]) - ], - sess.query(User).all()) - def test_hasattr(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) + User, Address = self._user_address_fixture() - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses)) - }) u1 = User(name='jack') - assert 'addresses' not in u1.__dict__.keys() + assert 'addresses' not in u1.__dict__ u1.addresses = [Address(email_address='test')] - assert 'addresses' in dir(u1) + assert 'addresses' in u1.__dict__ def test_collection_set(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses), order_by=addresses.c.email_address) - }) + addresses = self.tables.addresses + User, Address = self._user_address_fixture( + addresses_args={ + "order_by": addresses.c.email_address}) sess = create_session(autoflush=True, autocommit=False) u1 = User(name='jack') a1 = Address(email_address='a1') @@ -525,161 +462,95 @@ class SessionTest(_fixtures.FixtureTest): eq_(list(u1.addresses), []) def test_rollback(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses)) - }) - sess = create_session(expire_on_commit=False, autocommit=False, autoflush=True) + User, Address = self._user_address_fixture() + sess = create_session( + expire_on_commit=False, autocommit=False, autoflush=True) u1 = User(name='jack') u1.addresses.append(Address(email_address='lala@hoho.com')) sess.add(u1) sess.flush() sess.commit() u1.addresses.append(Address(email_address='foo@bar.com')) - eq_(u1.addresses.order_by(Address.id).all(), - [Address(email_address='lala@hoho.com'), Address(email_address='foo@bar.com')]) + eq_( + u1.addresses.order_by(Address.id).all(), + [ + Address(email_address='lala@hoho.com'), + Address(email_address='foo@bar.com') + ] + ) sess.rollback() - eq_(u1.addresses.all(), [Address(email_address='lala@hoho.com')]) + eq_( + u1.addresses.all(), + [Address(email_address='lala@hoho.com')] + ) - @testing.fails_on('maxdb', 'FIXME: unknown') - def test_delete_nocascade(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) + def _test_delete_cascade(self, expected): + addresses = self.tables.addresses + User, Address = self._user_address_fixture(addresses_args={ + "order_by": addresses.c.id, + "backref": "user", + "cascade": "save-update" if expected \ + else "all, delete" + }) - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses), order_by=Address.id, - backref='user') - }) sess = create_session(autoflush=True, autocommit=False) u = User(name='ed') - u.addresses.append(Address(email_address='a')) - u.addresses.append(Address(email_address='b')) - u.addresses.append(Address(email_address='c')) - u.addresses.append(Address(email_address='d')) - u.addresses.append(Address(email_address='e')) - u.addresses.append(Address(email_address='f')) + u.addresses.extend( + [Address(email_address=letter) for letter in 'abcdef'] + ) sess.add(u) - - eq_(Address(email_address='c'), u.addresses[2]) - sess.delete(u.addresses[2]) - sess.delete(u.addresses[4]) - sess.delete(u.addresses[3]) - eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='d')], - list(u.addresses)) - - sess.expunge_all() - u = sess.query(User).get(u.id) + sess.commit() + eq_(testing.db.scalar( + addresses.count(addresses.c.user_id == None)), 0) + eq_(testing.db.scalar( + addresses.count(addresses.c.user_id != None)), 6) sess.delete(u) - # u.addresses relationship will have to force the load - # of all addresses so that they can be updated - sess.flush() - sess.close() - - eq_(testing.db.scalar(addresses.count(addresses.c.user_id != None)), 0) - - @testing.fails_on('maxdb', 'FIXME: unknown') - def test_delete_cascade(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses), order_by=Address.id, - backref='user', cascade="all, delete-orphan") - }) - sess = create_session(autoflush=True, autocommit=False) - u = User(name='ed') - u.addresses.append(Address(email_address='a')) - u.addresses.append(Address(email_address='b')) - u.addresses.append(Address(email_address='c')) - u.addresses.append(Address(email_address='d')) - u.addresses.append(Address(email_address='e')) - u.addresses.append(Address(email_address='f')) - sess.add(u) - - eq_(Address(email_address='c'), u.addresses[2]) - sess.delete(u.addresses[2]) - sess.delete(u.addresses[4]) - sess.delete(u.addresses[3]) - eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='d')], - list(u.addresses)) - - sess.expunge_all() - u = sess.query(User).get(u.id) + sess.commit() - sess.delete(u) + if expected: + eq_(testing.db.scalar( + addresses.count(addresses.c.user_id == None)), 6) + eq_(testing.db.scalar( + addresses.count(addresses.c.user_id != None)), 0) + else: + eq_(testing.db.scalar(addresses.count()), 0) - # u.addresses relationship will have to force the load - # of all addresses so that they can be updated - sess.flush() - sess.close() + def test_delete_nocascade(self): + self._test_delete_cascade(True) - eq_(testing.db.scalar(addresses.count()), 0) + def test_delete_cascade(self): + self._test_delete_cascade(False) - @testing.fails_on('maxdb', 'FIXME: unknown') def test_remove_orphans(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) + addresses = self.tables.addresses + User, Address = self._user_address_fixture(addresses_args={ + "order_by": addresses.c.id, + "backref": "user", + "cascade": "all, delete-orphan" + }) - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses), order_by=Address.id, - cascade="all, delete-orphan", backref='user') - }) sess = create_session(autoflush=True, autocommit=False) u = User(name='ed') - u.addresses.append(Address(email_address='a')) - u.addresses.append(Address(email_address='b')) - u.addresses.append(Address(email_address='c')) - u.addresses.append(Address(email_address='d')) - u.addresses.append(Address(email_address='e')) - u.addresses.append(Address(email_address='f')) - sess.add(u) - - eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='c'), - Address(email_address='d'), Address(email_address='e'), Address(email_address='f')], - sess.query(Address).all()) - - eq_(Address(email_address='c'), u.addresses[2]) - - def go(): - del u.addresses[3] - assert_raises( - TypeError, - go + u.addresses.extend( + [Address(email_address=letter) for letter in 'abcdef'] ) + sess.add(u) - for a in u.addresses.filter(Address.email_address.in_(['c', 'e', 'f'])): + for a in u.addresses.filter( + Address.email_address.in_(['c', 'e', 'f'])): u.addresses.remove(a) - eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='d')], - list(u.addresses)) - - eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='d')], - sess.query(Address).all()) - - sess.delete(u) - sess.close() + eq_( + set(ad for ad, in sess.query(Address.email_address)), + set(['a', 'b', 'd']) + ) def _backref_test(self, autoflush, saveuser): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties={ - 'addresses':dynamic_loader(mapper(Address, addresses), backref='user') - }) + User, Address = self._user_address_fixture(addresses_args={ + "backref": "user", + }) sess = create_session(autoflush=autoflush, autocommit=False) u = User(name='buffy') @@ -720,37 +591,10 @@ class SessionTest(_fixtures.FixtureTest): def test_backref_savead(self): self._backref_test(False, False) -class DontDereferenceTest(fixtures.MappedTest): - @classmethod - def define_tables(cls, metadata): - Table('users', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), - Column('name', String(40)), - Column('fullname', String(100)), - Column('password', String(15))) - - Table('addresses', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), - Column('email_address', String(100), nullable=False), - Column('user_id', Integer, ForeignKey('users.id'))) - - @classmethod - def setup_mappers(cls): - users, addresses = cls.tables.users, cls.tables.addresses - - class User(cls.Comparable): - pass - - class Address(cls.Comparable): - pass - - mapper(User, users, properties={ - 'addresses': relationship(Address, backref='user', lazy='dynamic') - }) - mapper(Address, addresses) - def test_no_deref(self): - User, Address = self.classes.User, self.classes.Address + User, Address = self._user_address_fixture(addresses_args={ + "backref": "user", + }) session = create_session() user = User()