diff options
| author | Jason Kirtland <jek@discorporate.us> | 2008-05-09 20:26:09 +0000 |
|---|---|---|
| committer | Jason Kirtland <jek@discorporate.us> | 2008-05-09 20:26:09 +0000 |
| commit | e41c0f4107a132b2feac83ba07a25a336e7eae0b (patch) | |
| tree | 09c785fd5ef9557c3fc926afd7e0a78702dd8023 /test/orm/eager_relations.py | |
| parent | a2122a89f6d4f2d3ccc4ba7665cd588c2b0b93b0 (diff) | |
| download | sqlalchemy-e41c0f4107a132b2feac83ba07a25a336e7eae0b.tar.gz | |
Test suite modernization in progress. Big changes:
- @unsupported now only accepts a single target and demands a reason
for not running the test.
- @exclude also demands an exclusion reason
- Greatly expanded @testing.requires.<feature>, eliminating many
decorators in the suite and signficantly easing integration of
multi-driver support.
- New ORM test base class, and a featureful base for mapped tests
- Usage of 'global' for shared setup going away, * imports as well
Diffstat (limited to 'test/orm/eager_relations.py')
| -rw-r--r-- | test/orm/eager_relations.py | 598 |
1 files changed, 352 insertions, 246 deletions
diff --git a/test/orm/eager_relations.py b/test/orm/eager_relations.py index 544e505f4..b78ba12e9 100644 --- a/test/orm/eager_relations.py +++ b/test/orm/eager_relations.py @@ -1,17 +1,18 @@ """basic tests of eager loaded attributes""" import testenv; testenv.configure_for_tests() -from sqlalchemy import * -from sqlalchemy.orm import * -from testlib import * -from testlib.fixtures import * -from query import QueryTest -from sqlalchemy.orm import attributes - -class EagerTest(FixtureTest): - keep_mappers = False - keep_data = True - +from testlib import sa, testing +from sqlalchemy.orm import eagerload, deferred, undefer +from testlib.sa import Table, Column, Integer, String, ForeignKey +from testlib.sa.orm import mapper, relation, create_session +from testlib.testing import eq_ +from orm import _base, _fixtures + +class EagerTest(_fixtures.FixtureTest): + run_inserts = 'once' + run_deletes = None + + @testing.resolve_artifact_names def test_basic(self): mapper(User, users, properties={ 'addresses':relation(mapper(Address, addresses), lazy=False) @@ -20,11 +21,11 @@ class EagerTest(FixtureTest): q = sess.query(User) assert [User(id=7, addresses=[Address(id=1, email_address='jack@bean.com')])] == q.filter(User.id==7).all() - assert fixtures.user_address_result == q.all() + assert self.static.user_address_result == q.all() + @testing.resolve_artifact_names def test_no_orphan(self): - """test that an eagerly loaded child object is not marked as an orphan""" - + """An eagerly loaded child object is not marked as an orphan""" mapper(User, users, properties={ 'addresses':relation(Address, cascade="all,delete-orphan", lazy=False) }) @@ -32,9 +33,10 @@ class EagerTest(FixtureTest): sess = create_session() user = sess.query(User).get(7) - assert getattr(User, 'addresses').hasparent(attributes.instance_state(user.addresses[0]), optimistic=True) - assert not class_mapper(Address)._is_orphan(attributes.instance_state(user.addresses[0])) + assert getattr(User, 'addresses').hasparent(sa.orm.attributes.instance_state(user.addresses[0]), optimistic=True) + assert not sa.orm.class_mapper(Address)._is_orphan(sa.orm.attributes.instance_state(user.addresses[0])) + @testing.resolve_artifact_names def test_orderby(self): mapper(User, users, properties = { 'addresses':relation(mapper(Address, addresses), lazy=False, order_by=addresses.c.email_address), @@ -55,6 +57,7 @@ class EagerTest(FixtureTest): User(id=10, addresses=[]) ] == q.all() + @testing.resolve_artifact_names def test_orderby_multi(self): mapper(User, users, properties = { 'addresses':relation(mapper(Address, addresses), lazy=False, order_by=[addresses.c.email_address, addresses.c.id]), @@ -75,11 +78,10 @@ class EagerTest(FixtureTest): User(id=10, addresses=[]) ] == q.all() + @testing.resolve_artifact_names def test_orderby_related(self): - """tests that a regular mapper select on a single table can order by a relation to a second table""" - + """A regular mapper select on a single table can order by a relation to a second table""" mapper(Address, addresses) - mapper(User, users, properties = dict( addresses = relation(Address, lazy=False), )) @@ -101,11 +103,12 @@ class EagerTest(FixtureTest): ]), ] == l + @testing.resolve_artifact_names def test_orderby_desc(self): mapper(Address, addresses) - mapper(User, users, properties = dict( - addresses = relation(Address, lazy=False, order_by=[desc(addresses.c.email_address)]), + addresses = relation(Address, lazy=False, + order_by=[sa.desc(addresses.c.email_address)]), )) sess = create_session() assert [ @@ -123,7 +126,13 @@ class EagerTest(FixtureTest): User(id=10, addresses=[]) ] == sess.query(User).all() + @testing.resolve_artifact_names def test_deferred_fk_col(self): + User, Address, Dingaling = self.classes.get_all( + 'User', 'Address', 'Dingaling') + users, addresses, dingalings = self.tables.get_all( + 'users', 'addresses', 'dingalings') + mapper(Address, addresses, properties={ 'user_id':deferred(addresses.c.user_id), 'user':relation(User, lazy=False) @@ -131,39 +140,43 @@ class EagerTest(FixtureTest): mapper(User, users) sess = create_session() - + for q in [ sess.query(Address).filter(Address.id.in_([1, 4, 5])), sess.query(Address).filter(Address.id.in_([1, 4, 5])).limit(3) ]: sess.clear() - self.assertEquals(q.all(), - [Address(id=1, user=User(id=7)), Address(id=4, user=User(id=8)), Address(id=5, user=User(id=9))] + eq_(q.all(), + [Address(id=1, user=User(id=7)), + Address(id=4, user=User(id=8)), + Address(id=5, user=User(id=9))] ) a = sess.query(Address).filter(Address.id==1).first() def go(): - assert a.user_id==7 - # assert that the eager loader added 'user_id' to the row - # and deferred loading of that col was disabled + eq_(a.user_id, 7) + # assert that the eager loader added 'user_id' to the row and deferred + # loading of that col was disabled self.assert_sql_count(testing.db, go, 0) # do the mapping in reverse - # (we would have just used an "addresses" backref but the test fixtures then require the whole - # backref to be set up, lazy loaders trigger, etc.) - clear_mappers() + # (we would have just used an "addresses" backref but the test + # fixtures then require the whole backref to be set up, lazy loaders + # trigger, etc.) + sa.orm.clear_mappers() mapper(Address, addresses, properties={ 'user_id':deferred(addresses.c.user_id), }) - mapper(User, users, properties={'addresses':relation(Address, lazy=False)}) - + mapper(User, users, properties={ + 'addresses':relation(Address, lazy=False)}) + for q in [ sess.query(User).filter(User.id==7), sess.query(User).filter(User.id==7).limit(1) ]: sess.clear() - self.assertEquals(q.all(), + eq_(q.all(), [User(id=7, addresses=[Address(id=1)])] ) @@ -175,81 +188,115 @@ class EagerTest(FixtureTest): # and that its still deferred self.assert_sql_count(testing.db, go, 1) - clear_mappers() + sa.orm.clear_mappers() - mapper(User, users, properties={'addresses':relation(Address, lazy=False)}) + mapper(User, users, properties={ + 'addresses':relation(Address, lazy=False)}) mapper(Address, addresses, properties={ 'user_id':deferred(addresses.c.user_id), - 'dingalings':relation(Dingaling, lazy=False) - }) + 'dingalings':relation(Dingaling, lazy=False)}) mapper(Dingaling, dingalings, properties={ - 'address_id':deferred(dingalings.c.address_id) - }) + 'address_id':deferred(dingalings.c.address_id)}) sess.clear() def go(): u = sess.query(User).get(8) - assert User(id=8, addresses=[Address(id=2, dingalings=[Dingaling(id=1)]), Address(id=3), Address(id=4)]) == u + eq_(User(id=8, + addresses=[Address(id=2, dingalings=[Dingaling(id=1)]), + Address(id=3), + Address(id=4)]), + u) self.assert_sql_count(testing.db, go, 1) + @testing.resolve_artifact_names def test_many_to_many(self): + Keyword, Item = self.Keyword, self.Item + keywords, item_keywords, items = self.tables.get_all( + 'keywords', 'item_keywords', 'items') mapper(Keyword, keywords) mapper(Item, items, properties = dict( - keywords = relation(Keyword, secondary=item_keywords, lazy=False, order_by=keywords.c.id), - )) + keywords = relation(Keyword, secondary=item_keywords, + lazy=False, order_by=keywords.c.id))) q = create_session().query(Item) def go(): - assert fixtures.item_keyword_result == q.all() + assert self.static.item_keyword_result == q.all() self.assert_sql_count(testing.db, go, 1) def go(): - assert fixtures.item_keyword_result[0:2] == q.join('keywords').filter(Keyword.name == 'red').all() + eq_(self.static.item_keyword_result[0:2], + q.join('keywords').filter(Keyword.name == 'red').all()) self.assert_sql_count(testing.db, go, 1) def go(): - assert fixtures.item_keyword_result[0:2] == q.join('keywords', aliased=True).filter(Keyword.name == 'red').all() + eq_(self.static.item_keyword_result[0:2], + (q.join('keywords', aliased=True). + filter(Keyword.name == 'red')).all()) self.assert_sql_count(testing.db, go, 1) - + @testing.resolve_artifact_names def test_eager_option(self): + Keyword, Item = self.Keyword, self.Item + keywords, item_keywords, items = self.tables.get_all( + 'keywords', 'item_keywords', 'items') + mapper(Keyword, keywords) mapper(Item, items, properties = dict( - keywords = relation(Keyword, secondary=item_keywords, lazy=True, order_by=keywords.c.id), - )) + keywords = relation(Keyword, secondary=item_keywords, lazy=True, + order_by=keywords.c.id))) q = create_session().query(Item) def go(): - assert fixtures.item_keyword_result[0:2] == q.options(eagerload('keywords')).join('keywords').filter(keywords.c.name == 'red').all() + eq_(self.static.item_keyword_result[0:2], + (q.options(eagerload('keywords')). + join('keywords').filter(keywords.c.name == 'red')).all()) self.assert_sql_count(testing.db, go, 1) + @testing.resolve_artifact_names def test_cyclical(self): - """test that a circular eager relationship breaks the cycle with a lazy loader""" + """A circular eager relationship breaks the cycle with a lazy loader""" + User, Address = self.User, self.Address + users, addresses = self.tables.get_all('users', 'addresses') mapper(Address, addresses) mapper(User, users, properties = dict( - addresses = relation(Address, lazy=False, backref=backref('user', lazy=False)) + addresses = relation(Address, lazy=False, + backref=sa.orm.backref('user', lazy=False)) )) - assert class_mapper(User).get_property('addresses').lazy is False - assert class_mapper(Address).get_property('user').lazy is False + assert sa.orm.class_mapper(User).get_property('addresses').lazy is False + assert sa.orm.class_mapper(Address).get_property('user').lazy is False sess = create_session() - assert fixtures.user_address_result == sess.query(User).all() + assert self.static.user_address_result == sess.query(User).all() + @testing.resolve_artifact_names def test_double(self): - """tests eager loading with two relations simulatneously, from the same table, using aliases. """ - openorders = alias(orders, 'openorders') - closedorders = alias(orders, 'closedorders') + """Eager loading with two relations simultaneously, from the same table, using aliases.""" + User, Address, Order = self.classes.get_all( + 'User', 'Address', 'Order') + users, addresses, orders = self.tables.get_all( + 'users', 'addresses', 'orders') + + openorders = sa.alias(orders, 'openorders') + closedorders = sa.alias(orders, 'closedorders') mapper(Address, addresses) mapper(User, users, properties = dict( addresses = relation(Address, lazy=False), - open_orders = relation(mapper(Order, openorders, entity_name='open'), primaryjoin = and_(openorders.c.isopen == 1, users.c.id==openorders.c.user_id), lazy=False), - closed_orders = relation(mapper(Order, closedorders,entity_name='closed'), primaryjoin = and_(closedorders.c.isopen == 0, users.c.id==closedorders.c.user_id), lazy=False) - )) + open_orders = relation( + mapper(Order, openorders, entity_name='open'), + primaryjoin=sa.and_(openorders.c.isopen == 1, + users.c.id==openorders.c.user_id), + lazy=False), + closed_orders = relation( + mapper(Order, closedorders,entity_name='closed'), + primaryjoin=sa.and_(closedorders.c.isopen == 0, + users.c.id==closedorders.c.user_id), + lazy=False))) + q = create_session().query(User) def go(): @@ -277,48 +324,83 @@ class EagerTest(FixtureTest): ] == q.all() self.assert_sql_count(testing.db, go, 1) + @testing.resolve_artifact_names def test_double_same_mappers(self): - """tests eager loading with two relations simulatneously, from the same table, using aliases. """ + """Eager loading with two relations simulatneously, from the same table, using aliases.""" + User, Address, Order = self.classes.get_all( + 'User', 'Address', 'Order') + users, addresses, orders = self.tables.get_all( + 'users', 'addresses', 'orders') mapper(Address, addresses) mapper(Order, orders, properties={ - 'items':relation(Item, secondary=order_items, lazy=False, order_by=items.c.id), - }) + 'items': relation(Item, secondary=order_items, lazy=False, + order_by=items.c.id)}) mapper(Item, items) - mapper(User, users, properties = dict( - addresses = relation(Address, lazy=False), - open_orders = relation(Order, primaryjoin = and_(orders.c.isopen == 1, users.c.id==orders.c.user_id), lazy=False), - closed_orders = relation(Order, primaryjoin = and_(orders.c.isopen == 0, users.c.id==orders.c.user_id), lazy=False) - )) + mapper(User, users, properties=dict( + addresses=relation(Address, lazy=False), + open_orders=relation( + Order, + primaryjoin=sa.and_(orders.c.isopen == 1, + users.c.id==orders.c.user_id), + lazy=False), + closed_orders=relation( + Order, + primaryjoin=sa.and_(orders.c.isopen == 0, + users.c.id==orders.c.user_id), + lazy=False))) q = create_session().query(User) def go(): assert [ - User( - id=7, - addresses=[Address(id=1)], - open_orders = [Order(id=3, items=[Item(id=3), Item(id=4), Item(id=5)])], - closed_orders = [Order(id=1, items=[Item(id=1), Item(id=2), Item(id=3)]), Order(id=5, items=[Item(id=5)])] - ), - User( - id=8, - addresses=[Address(id=2), Address(id=3), Address(id=4)], - open_orders = [], - closed_orders = [] - ), - User( - id=9, - addresses=[Address(id=5)], - open_orders = [Order(id=4, items=[Item(id=1), Item(id=5)])], - closed_orders = [Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])] - ), + User(id=7, + addresses=[ + Address(id=1)], + open_orders=[Order(id=3, + items=[ + Item(id=3), + Item(id=4), + Item(id=5)])], + closed_orders=[Order(id=1, + items=[ + Item(id=1), + Item(id=2), + Item(id=3)]), + Order(id=5, + items=[ + Item(id=5)])]), + User(id=8, + addresses=[ + Address(id=2), + Address(id=3), + Address(id=4)], + open_orders = [], + closed_orders = []), + User(id=9, + addresses=[ + Address(id=5)], + open_orders=[ + Order(id=4, + items=[ + Item(id=1), + Item(id=5)])], + closed_orders=[ + Order(id=2, + items=[ + Item(id=1), + Item(id=2), + Item(id=3)])]), User(id=10) - ] == q.all() self.assert_sql_count(testing.db, go, 1) + @testing.resolve_artifact_names def test_no_false_hits(self): - """test that eager loaders don't interpret main table columns as part of their eager load.""" + """Eager loaders don't interpret main table columns as part of their eager load.""" + User, Address, Order = self.classes.get_all( + 'User', 'Address', 'Order') + users, addresses, orders = self.tables.get_all( + 'users', 'addresses', 'orders') mapper(User, users, properties={ 'addresses':relation(Address, lazy=False), @@ -329,16 +411,22 @@ class EagerTest(FixtureTest): allusers = create_session().query(User).all() - # using a textual select, the columns will be 'id' and 'name'. - # the eager loaders have aliases which should not hit on those columns, they should - # be required to locate only their aliased/fully table qualified column name. + # using a textual select, the columns will be 'id' and 'name'. the + # eager loaders have aliases which should not hit on those columns, + # they should be required to locate only their aliased/fully table + # qualified column name. noeagers = create_session().query(User).from_statement("select * from users").all() assert 'orders' not in noeagers[0].__dict__ assert 'addresses' not in noeagers[0].__dict__ @testing.fails_on('maxdb') + @testing.resolve_artifact_names def test_limit(self): - """test limit operations combined with lazy-load relationships.""" + """Limit operations combined with lazy-load relationships.""" + User, Item, Address, Order = self.classes.get_all( + 'User', 'Item', 'Address', 'Order') + users, items, order_items, orders, addresses = self.tables.get_all( + 'users', 'items', 'order_items', 'orders', 'addresses') mapper(Item, items) mapper(Order, orders, properties={ @@ -354,19 +442,20 @@ class EagerTest(FixtureTest): if testing.against('mysql'): l = q.limit(2).all() - assert fixtures.user_all_result[:2] == l + assert self.static.user_all_result[:2] == l else: l = q.limit(2).offset(1).order_by(User.id).all() - print fixtures.user_all_result[1:3] + print self.static.user_all_result[1:3] print l - assert fixtures.user_all_result[1:3] == l + assert self.static.user_all_result[1:3] == l + @testing.resolve_artifact_names def test_distinct(self): # this is an involved 3x union of the users table to get a lot of rows. # then see if the "distinct" works its way out. you actually get the same # result with or without the distinct, just via less or more rows. u2 = users.alias('u2') - s = union_all(u2.select(use_labels=True), u2.select(use_labels=True), u2.select(use_labels=True)).alias('u') + s = sa.union_all(u2.select(use_labels=True), u2.select(use_labels=True), u2.select(use_labels=True)).alias('u') mapper(User, users, properties={ 'addresses':relation(mapper(Address, addresses), lazy=False), @@ -377,10 +466,11 @@ class EagerTest(FixtureTest): def go(): l = q.filter(s.c.u2_id==User.id).distinct().all() - assert fixtures.user_address_result == l + assert self.static.user_address_result == l self.assert_sql_count(testing.db, go, 1) @testing.fails_on('maxdb') + @testing.resolve_artifact_names def test_limit_2(self): mapper(Keyword, keywords) mapper(Item, items, properties = dict( @@ -392,9 +482,10 @@ class EagerTest(FixtureTest): l = q.filter((Item.description=='item 2') | (Item.description=='item 5') | (Item.description=='item 3')).\ order_by(Item.id).limit(2).all() - assert fixtures.item_keyword_result[1:3] == l + assert self.static.item_keyword_result[1:3] == l @testing.fails_on('maxdb') + @testing.resolve_artifact_names def test_limit_3(self): """test that the ORDER BY is propigated from the inner select to the outer select, when using the 'wrapped' select statement resulting from the combination of eager loading and limit/offset clauses.""" @@ -434,16 +525,17 @@ class EagerTest(FixtureTest): ) ] == l.all() + @testing.resolve_artifact_names def test_limit_4(self): # tests the LIMIT/OFFSET aliasing on a mapper against a select. original issue from ticket #904 - sel = select([users, addresses.c.email_address], users.c.id==addresses.c.user_id).alias('useralias') + sel = sa.select([users, addresses.c.email_address], users.c.id==addresses.c.user_id).alias('useralias') mapper(User, sel, properties={ 'orders':relation(Order, primaryjoin=sel.c.id==orders.c.user_id, lazy=False) }) mapper(Order, orders) sess = create_session() - self.assertEquals(sess.query(User).first(), + eq_(sess.query(User).first(), User(name=u'jack',orders=[ Order(address_id=1,description=u'order 1',isopen=0,user_id=7,id=1), Order(address_id=1,description=u'order 3',isopen=1,user_id=7,id=3), @@ -451,6 +543,7 @@ class EagerTest(FixtureTest): email_address=u'jack@bean.com',id=7) ) + @testing.resolve_artifact_names def test_one_to_many_scalar(self): mapper(User, users, properties = dict( address = relation(mapper(Address, addresses), lazy=False, uselist=False) @@ -463,6 +556,7 @@ class EagerTest(FixtureTest): self.assert_sql_count(testing.db, go, 1) @testing.fails_on('maxdb') + @testing.resolve_artifact_names def test_many_to_one(self): mapper(Address, addresses, properties = dict( user = relation(mapper(User, users), lazy=False) @@ -477,7 +571,7 @@ class EagerTest(FixtureTest): assert a.user is u1 self.assert_sql_count(testing.db, go, 1) - + @testing.resolve_artifact_names def test_one_and_many(self): """tests eager load for a parent object with a child object that contains a many-to-many relationship to a third object.""" @@ -495,12 +589,12 @@ class EagerTest(FixtureTest): l = q.filter("users.id in (7, 8, 9)") def go(): - assert fixtures.user_order_result[0:3] == l.all() + assert self.static.user_order_result[0:3] == l.all() self.assert_sql_count(testing.db, go, 1) + @testing.resolve_artifact_names def test_double_with_aggregate(self): - - max_orders_by_user = select([func.max(orders.c.id).label('order_id')], group_by=[orders.c.user_id]).alias('max_orders_by_user') + max_orders_by_user = sa.select([sa.func.max(orders.c.id).label('order_id')], group_by=[orders.c.user_id]).alias('max_orders_by_user') max_orders = orders.select(orders.c.id==max_orders_by_user.c.order_id).alias('max_orders') @@ -528,6 +622,7 @@ class EagerTest(FixtureTest): ] == q.all() self.assert_sql_count(testing.db, go, 1) + @testing.resolve_artifact_names def test_wide(self): mapper(Order, orders, properties={'items':relation(Item, secondary=order_items, lazy=False, order_by=items.c.id)}) mapper(Item, items) @@ -537,12 +632,13 @@ class EagerTest(FixtureTest): )) q = create_session().query(User) l = q.all() - assert fixtures.user_all_result == q.all() + assert self.static.user_all_result == q.all() + @testing.resolve_artifact_names def test_against_select(self): """test eager loading of a mapper which is against a select""" - s = select([orders], orders.c.isopen==1).alias('openorders') + s = sa.select([orders], orders.c.isopen==1).alias('openorders') mapper(Order, s, properties={ 'user':relation(User, lazy=False) @@ -561,6 +657,7 @@ class EagerTest(FixtureTest): Order(id=3, user=User(id=7)), ] == q.all() + @testing.resolve_artifact_names def test_aliasing(self): """test that eager loading uses aliases to insulate the eager load from regular criterion against those tables.""" @@ -569,12 +666,13 @@ class EagerTest(FixtureTest): )) q = create_session().query(User) l = q.filter(addresses.c.email_address == 'ed@lala.com').filter(Address.user_id==User.id) - assert fixtures.user_address_result[1:2] == l.all() + assert self.static.user_address_result[1:2] == l.all() -class AddEntityTest(FixtureTest): - keep_mappers = False - keep_data = True +class AddEntityTest(_fixtures.FixtureTest): + run_inserts = 'once' + run_deletes = None + @testing.resolve_artifact_names def _assert_result(self): return [ ( @@ -619,6 +717,7 @@ class AddEntityTest(FixtureTest): ) ] + @testing.resolve_artifact_names def test_mapper_configured(self): mapper(User, users, properties={ 'addresses':relation(Address, lazy=False), @@ -632,12 +731,13 @@ class AddEntityTest(FixtureTest): sess = create_session() - oalias = aliased(Order) + oalias = sa.orm.aliased(Order) def go(): ret = sess.query(User, oalias).join(('orders', oalias)).order_by(User.id, oalias.id).all() - self.assertEquals(ret, self._assert_result()) + eq_(ret, self._assert_result()) self.assert_sql_count(testing.db, go, 1) + @testing.resolve_artifact_names def test_options(self): mapper(User, users, properties={ 'addresses':relation(Address), @@ -651,82 +751,78 @@ class AddEntityTest(FixtureTest): sess = create_session() - oalias = aliased(Order) + oalias = sa.orm.aliased(Order) def go(): ret = sess.query(User, oalias).options(eagerload('addresses')).join(('orders', oalias)).order_by(User.id, oalias.id).all() - self.assertEquals(ret, self._assert_result()) + eq_(ret, self._assert_result()) self.assert_sql_count(testing.db, go, 6) sess.clear() def go(): ret = sess.query(User, oalias).options(eagerload('addresses'), eagerload(oalias.items)).join(('orders', oalias)).order_by(User.id, oalias.id).all() - self.assertEquals(ret, self._assert_result()) + eq_(ret, self._assert_result()) self.assert_sql_count(testing.db, go, 1) -class OrderBySecondaryTest(ORMTest): +class OrderBySecondaryTest(_base.MappedTest): def define_tables(self, metadata): - global a, b, m2m - m2m = Table('mtom', metadata, - Column('id', Integer, primary_key=True), - Column('aid', Integer, ForeignKey('a.id')), - Column('bid', Integer, ForeignKey('b.id')), - ) - - a = Table('a', metadata, - Column('id', Integer, primary_key=True), - Column('data', String(50)), - ) - b = Table('b', metadata, - Column('id', Integer, primary_key=True), - Column('data', String(50)), - ) - - def insert_data(self): - a.insert().execute([ - {'id':1, 'data':'a1'}, - {'id':2, 'data':'a2'} - ]) - - b.insert().execute([ - {'id':1, 'data':'b1'}, - {'id':2, 'data':'b2'}, - {'id':3, 'data':'b3'}, - {'id':4, 'data':'b4'}, - ]) - - m2m.insert().execute([ - {'id':2, 'aid':1, 'bid':1}, - {'id':4, 'aid':2, 'bid':4}, - {'id':1, 'aid':1, 'bid':3}, - {'id':6, 'aid':2, 'bid':2}, - {'id':3, 'aid':1, 'bid':2}, - {'id':5, 'aid':2, 'bid':3}, - ]) - + Table('m2m', metadata, + Column('id', Integer, primary_key=True), + Column('aid', Integer, ForeignKey('a.id')), + Column('bid', Integer, ForeignKey('b.id'))) + + Table('a', metadata, + Column('id', Integer, primary_key=True), + Column('data', String(50))) + Table('b', metadata, + Column('id', Integer, primary_key=True), + Column('data', String(50))) + + def fixtures(self): + return dict( + a=(('id', 'data'), + (1, 'a1'), + (2, 'a2')), + + b=(('id', 'data'), + (1, 'b1'), + (2, 'b2'), + (3, 'b3'), + (4, 'b4')), + + m2m=(('id', 'aid', 'bid'), + (2, 1, 1), + (4, 2, 4), + (1, 1, 3), + (6, 2, 2), + (3, 1, 2), + (5, 2, 3))) + + @testing.resolve_artifact_names def test_ordering(self): - class A(Base):pass - class B(Base):pass - + class A(_base.ComparableEntity):pass + class B(_base.ComparableEntity):pass + mapper(A, a, properties={ 'bs':relation(B, secondary=m2m, lazy=False, order_by=m2m.c.id) }) mapper(B, b) - + sess = create_session() - self.assertEquals(sess.query(A).all(), [A(data='a1', bs=[B(data='b3'), B(data='b1'), B(data='b2')]), A(bs=[B(data='b4'), B(data='b3'), B(data='b2')])]) - - -class SelfReferentialEagerTest(ORMTest): + eq_(sess.query(A).all(), [A(data='a1', bs=[B(data='b3'), B(data='b1'), B(data='b2')]), A(bs=[B(data='b4'), B(data='b3'), B(data='b2')])]) + + +class SelfReferentialEagerTest(_base.MappedTest): def define_tables(self, metadata): - global nodes - nodes = Table('nodes', metadata, - Column('id', Integer, Sequence('node_id_seq', optional=True), primary_key=True), + Table('nodes', metadata, + Column('id', Integer, sa.Sequence('node_id_seq', optional=True), + primary_key=True), Column('parent_id', Integer, ForeignKey('nodes.id')), Column('data', String(30))) @testing.fails_on('maxdb') + @testing.resolve_artifact_names def test_basic(self): - class Node(Base): + class Node(_base.ComparableEntity): def append(self, node): self.children.append(node) @@ -758,8 +854,9 @@ class SelfReferentialEagerTest(ORMTest): self.assert_sql_count(testing.db, go, 1) + @testing.resolve_artifact_names def test_lazy_fallback_doesnt_affect_eager(self): - class Node(Base): + class Node(_base.ComparableEntity): def append(self, node): self.children.append(node) @@ -778,14 +875,14 @@ class SelfReferentialEagerTest(ORMTest): sess.flush() sess.clear() - # eager load with join depth 1. when eager load of 'n1' - # hits the children of 'n12', no columns are present, eager loader - # degrades to lazy loader; fine. but then, 'n12' is *also* in the - # first level of columns since we're loading the whole table. - # when those rows arrive, now we *can* eager load its children and an - # eager collection should be initialized. essentially the 'n12' instance - # is present in not just two different rows but two distinct sets of columns - # in this result set. + # eager load with join depth 1. when eager load of 'n1' hits the + # children of 'n12', no columns are present, eager loader degrades to + # lazy loader; fine. but then, 'n12' is *also* in the first level of + # columns since we're loading the whole table. when those rows + # arrive, now we *can* eager load its children and an eager collection + # should be initialized. essentially the 'n12' instance is present in + # not just two different rows but two distinct sets of columns in this + # result set. def go(): allnodes = sess.query(Node).order_by(Node.data).all() n12 = allnodes[2] @@ -799,8 +896,9 @@ class SelfReferentialEagerTest(ORMTest): ] == list(n12.children) self.assert_sql_count(testing.db, go, 1) + @testing.resolve_artifact_names def test_with_deferred(self): - class Node(Base): + class Node(_base.ComparableEntity): def append(self, node): self.children.append(node) @@ -833,9 +931,9 @@ class SelfReferentialEagerTest(ORMTest): self.assert_sql_count(testing.db, go, 1) - + @testing.resolve_artifact_names def test_options(self): - class Node(Base): + class Node(_base.ComparableEntity): def append(self, node): self.children.append(node) @@ -881,8 +979,9 @@ class SelfReferentialEagerTest(ORMTest): ]) @testing.fails_on('maxdb') + @testing.resolve_artifact_names def test_no_depth(self): - class Node(Base): + class Node(_base.ComparableEntity): def append(self, node): self.children.append(node) @@ -913,22 +1012,22 @@ class SelfReferentialEagerTest(ORMTest): ]) == d self.assert_sql_count(testing.db, go, 3) -class SelfReferentialM2MEagerTest(ORMTest): +class SelfReferentialM2MEagerTest(_base.MappedTest): def define_tables(self, metadata): - global widget, widget_rel - - widget = Table('widget', metadata, + Table('widget', metadata, Column('id', Integer, primary_key=True), - Column('name', Unicode(40), nullable=False, unique=True), + Column('name', sa.Unicode(40), nullable=False, unique=True), ) - widget_rel = Table('widget_rel', metadata, + Table('widget_rel', metadata, Column('parent_id', Integer, ForeignKey('widget.id')), Column('child_id', Integer, ForeignKey('widget.id')), - UniqueConstraint('parent_id', 'child_id'), + sa.UniqueConstraint('parent_id', 'child_id'), ) + + @testing.resolve_artifact_names def test_basic(self): - class Widget(Base): + class Widget(_base.ComparableEntity): pass mapper(Widget, widget, properties={ @@ -949,10 +1048,12 @@ class SelfReferentialM2MEagerTest(ORMTest): assert [Widget(name='w1', children=[Widget(name='w2')])] == sess.query(Widget).filter(Widget.name==u'w1').all() -class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL): - keep_mappers = True - keep_data = True - +class MixedEntitiesTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): + run_setup_mappers = 'once' + run_inserts = 'once' + run_deletes = None + + @testing.resolve_artifact_names def setup_mappers(self): mapper(User, users, properties={ 'addresses':relation(Address, backref='user'), @@ -966,13 +1067,14 @@ class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL): 'keywords':relation(Keyword, secondary=item_keywords) #m2m }) mapper(Keyword, keywords) - + + @testing.resolve_artifact_names def test_two_entities(self): sess = create_session() # two FROM clauses def go(): - self.assertEquals( + eq_( [ (User(id=9, addresses=[Address(id=5)]), Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])), (User(id=9, addresses=[Address(id=5)]), Order(id=4, items=[Item(id=1), Item(id=5)])), @@ -985,7 +1087,7 @@ class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL): # one FROM clause def go(): - self.assertEquals( + eq_( [ (User(id=9, addresses=[Address(id=5)]), Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])), (User(id=9, addresses=[Address(id=5)]), Order(id=4, items=[Item(id=1), Item(id=5)])), @@ -994,15 +1096,16 @@ class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL): order_by(User.id, Order.id).all(), ) self.assert_sql_count(testing.db, go, 1) - + + @testing.resolve_artifact_names def test_aliased_entity(self): sess = create_session() - - oalias = aliased(Order) - + + oalias = sa.orm.aliased(Order) + # two FROM clauses def go(): - self.assertEquals( + eq_( [ (User(id=9, addresses=[Address(id=5)]), Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])), (User(id=9, addresses=[Address(id=5)]), Order(id=4, items=[Item(id=1), Item(id=5)])), @@ -1015,7 +1118,7 @@ class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL): # one FROM clause def go(): - self.assertEquals( + eq_( [ (User(id=9, addresses=[Address(id=5)]), Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])), (User(id=9, addresses=[Address(id=5)]), Order(id=4, items=[Item(id=1), Item(id=5)])), @@ -1024,12 +1127,13 @@ class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL): order_by(User.id, oalias.id).all(), ) self.assert_sql_count(testing.db, go, 1) - + from sqlalchemy.engine.default import DefaultDialect - - # improper setup: oalias in the columns clause but join to usual orders alias. - # this should create two FROM clauses even though the query has a from_clause set up via the join - self.assert_compile(sess.query(User, oalias).join(User.orders).options(eagerload(oalias.items)).with_labels().statement, + + # improper setup: oalias in the columns clause but join to usual + # orders alias. this should create two FROM clauses even though the + # query has a from_clause set up via the join + self.assert_compile(sess.query(User, oalias).join(User.orders).options(eagerload(oalias.items)).with_labels().statement, "SELECT users.id AS users_id, users.name AS users_name, orders_1.id AS orders_1_id, "\ "orders_1.user_id AS orders_1_user_id, orders_1.address_id AS orders_1_address_id, "\ "orders_1.description AS orders_1_description, orders_1.isopen AS orders_1_isopen, items_1.id AS items_1_id, "\ @@ -1038,23 +1142,23 @@ class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL): "LEFT OUTER JOIN items AS items_1 ON items_1.id = order_items_1.item_id ORDER BY users.id, items_1.id", dialect=DefaultDialect() ) - -class CyclicalInheritingEagerTest(ORMTest): + +class CyclicalInheritingEagerTest(_base.MappedTest): def define_tables(self, metadata): - global t1, t2 - t1 = Table('t1', metadata, + Table('t1', metadata, Column('c1', Integer, primary_key=True), Column('c2', String(30)), Column('type', String(30)) ) - t2 = Table('t2', metadata, + Table('t2', metadata, Column('c1', Integer, primary_key=True), Column('c2', String(30)), Column('type', String(30)), Column('t1.id', Integer, ForeignKey('t1.c1'))) + @testing.resolve_artifact_names def test_basic(self): class T(object): pass @@ -1070,7 +1174,7 @@ class CyclicalInheritingEagerTest(ORMTest): mapper(T, t1, polymorphic_on=t1.c.type, polymorphic_identity='t1') mapper(SubT, None, inherits=T, polymorphic_identity='subt1', properties={ - 't2s':relation(SubT2, lazy=False, backref=backref('subt', lazy=False)) + 't2s':relation(SubT2, lazy=False, backref=sa.orm.backref('subt', lazy=False)) }) mapper(T2, t2, polymorphic_on=t2.c.type, polymorphic_identity='t2') mapper(SubT2, None, inherits=T2, polymorphic_identity='subt2') @@ -1078,65 +1182,66 @@ class CyclicalInheritingEagerTest(ORMTest): # testing a particular endless loop condition in eager join setup create_session().query(SubT).all() -class SubqueryTest(ORMTest): +class SubqueryTest(_base.MappedTest): def define_tables(self, metadata): - global users_table, tags_table - - users_table = Table('users', metadata, + Table('users_table', metadata, Column('id', Integer, primary_key=True), Column('name', String(16)) ) - tags_table = Table('tags', metadata, + Table('tags_table', metadata, Column('id', Integer, primary_key=True), - Column('user_id', Integer, ForeignKey("users.id")), - Column('score1', Float), - Column('score2', Float), + Column('user_id', Integer, ForeignKey("users_table.id")), + Column('score1', sa.Float), + Column('score2', sa.Float), ) + @testing.resolve_artifact_names def test_label_anonymizing(self): - """test that eager loading works with subqueries with labels, - even if an explicit labelname which conflicts with a label on the parent. - - There's not much reason a column_property() would ever need to have a label - of a specific name (and they don't even need labels these days), - unless you'd like the name to line up with a name - that you may be using for a straight textual statement used for loading - instances of that type. - + """Eager loading works with subqueries with labels, + + Even if an explicit labelname which conflicts with a label on the + parent. + + There's not much reason a column_property() would ever need to have a + label of a specific name (and they don't even need labels these days), + unless you'd like the name to line up with a name that you may be + using for a straight textual statement used for loading instances of + that type. + """ - class User(Base): + class User(_base.ComparableEntity): @property def prop_score(self): return sum([tag.prop_score for tag in self.tags]) - class Tag(Base): + class Tag(_base.ComparableEntity): @property def prop_score(self): return self.score1 * self.score2 - + for labeled, labelname in [(True, 'score'), (True, None), (False, None)]: - clear_mappers() - + sa.orm.clear_mappers() + tag_score = (tags_table.c.score1 * tags_table.c.score2) - user_score = select([func.sum(tags_table.c.score1 * - tags_table.c.score2)], - tags_table.c.user_id == users_table.c.id) - + user_score = sa.select([sa.func.sum(tags_table.c.score1 * + tags_table.c.score2)], + tags_table.c.user_id == users_table.c.id) + if labeled: tag_score = tag_score.label(labelname) user_score = user_score.label(labelname) else: user_score = user_score.as_scalar() - + mapper(Tag, tags_table, properties={ - 'query_score': column_property(tag_score), + 'query_score': sa.orm.column_property(tag_score), }) mapper(User, users_table, properties={ - 'tags': relation(Tag, backref='user', lazy=False), - 'query_score': column_property(user_score), + 'tags': relation(Tag, backref='user', lazy=False), + 'query_score': sa.orm.column_property(user_score), }) session = create_session() @@ -1144,17 +1249,18 @@ class SubqueryTest(ORMTest): session.save(User(name='bar', tags=[Tag(score1=5.0, score2=4.0), Tag(score1=50.0, score2=1.0), Tag(score1=15.0, score2=2.0)])) session.flush() session.clear() - + for user in session.query(User).all(): - self.assertEquals(user.query_score, user.prop_score) + eq_(user.query_score, user.prop_score) def go(): u = session.query(User).filter_by(name='joe').one() - self.assertEquals(u.query_score, u.prop_score) + eq_(u.query_score, u.prop_score) self.assert_sql_count(testing.db, go, 1) - + for t in (tags_table, users_table): t.delete().execute() - + + if __name__ == '__main__': testenv.main() |
